333 lines
12 KiB
Python
333 lines
12 KiB
Python
"""
|
|
app/utils/photo_stamp.py
|
|
------------------------
|
|
Burn a capture-time + geolocation overlay into uploaded evidence photos.
|
|
|
|
Applied at UPLOAD time (``app/api/photos.py``) rather than on
|
|
``PATCH /issues/<id>/photos``. At upload the raw bytes and any camera EXIF are
|
|
in hand, so nothing has to be read back out of R2, and each upload writes
|
|
exactly one already-stamped object. That also keeps stamping clear of the
|
|
retry/double-burn hazard the PATCH endpoint would have — it is deliberately
|
|
idempotent and re-runnable (CLAUDE.md rule 45), so burning there could stack a
|
|
second bar onto an already-stamped image.
|
|
|
|
Metadata resolution order
|
|
-------------------------
|
|
1. Client-supplied ``captured_at`` / ``latitude`` / ``longitude`` — most
|
|
reliable for an offline-first app: the iPad knows when and where the shot
|
|
was taken even if it syncs hours later.
|
|
2. The image's own EXIF ``DateTimeOriginal`` / ``GPSInfo``.
|
|
3. Server receipt time (last resort; no geo).
|
|
|
|
FAILURE POLICY
|
|
--------------
|
|
Stamping must never cost us the photo. Every failure path falls back to
|
|
storing the original bytes unmodified — an unstamped photo beats a lost one.
|
|
"""
|
|
|
|
import io
|
|
import logging
|
|
from datetime import datetime
|
|
|
|
import pytz
|
|
|
|
from app.utils.time_utils import EASTERN, now_eastern
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Only these are stamped. GIF (possibly animated) and anything exotic passes
|
|
# through untouched rather than risking a broken re-encode.
|
|
_STAMPABLE_FORMATS = {'JPEG', 'PNG'}
|
|
|
|
# Candidate TrueType fonts, in preference order. Pillow does not reliably ship
|
|
# a TTF, and the bitmap default is unreadably small on a 4000px photo, so we
|
|
# probe the usual Linux locations and degrade gracefully.
|
|
_FONT_CANDIDATES = (
|
|
'/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf',
|
|
'/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf',
|
|
'/usr/share/fonts/truetype/liberation/LiberationSans-Bold.ttf',
|
|
'/usr/share/fonts/truetype/freefont/FreeSansBold.ttf',
|
|
'C:/Windows/Fonts/arialbd.ttf',
|
|
'C:/Windows/Fonts/arial.ttf',
|
|
)
|
|
|
|
_EXIF_DATETIME_ORIGINAL = 36867 # 0x9003
|
|
_EXIF_DATETIME_DIGITIZED = 36868 # 0x9004
|
|
_EXIF_DATETIME = 306 # 0x0132
|
|
_EXIF_GPS_IFD = 34853 # 0x8825
|
|
|
|
|
|
# ── Font ──────────────────────────────────────────────────────────────────────
|
|
|
|
def _load_font(size):
|
|
"""Return a TrueType font at *size*, or Pillow's bitmap default."""
|
|
from PIL import ImageFont
|
|
for path in _FONT_CANDIDATES:
|
|
try:
|
|
return ImageFont.truetype(path, size)
|
|
except Exception:
|
|
continue
|
|
try:
|
|
# Pillow >= 9.2 can scale the default font.
|
|
return ImageFont.load_default(size=size)
|
|
except Exception:
|
|
return ImageFont.load_default()
|
|
|
|
|
|
# ── Metadata extraction ───────────────────────────────────────────────────────
|
|
|
|
def _parse_client_datetime(value):
|
|
"""Parse a client ISO-8601 timestamp into naive Eastern, or None.
|
|
|
|
Accepts offsets and a trailing 'Z'. An offset-aware value is converted to
|
|
Eastern; a naive value is taken as already-Eastern wall time.
|
|
"""
|
|
if not value:
|
|
return None
|
|
text = str(value).strip()
|
|
if not text:
|
|
return None
|
|
if text.endswith(('Z', 'z')):
|
|
text = text[:-1] + '+00:00'
|
|
try:
|
|
dt = datetime.fromisoformat(text)
|
|
except ValueError:
|
|
return None
|
|
if dt.tzinfo is not None:
|
|
dt = dt.astimezone(EASTERN).replace(tzinfo=None)
|
|
return dt
|
|
|
|
|
|
def _parse_exif_datetime(raw):
|
|
"""Parse an EXIF 'YYYY:MM:DD HH:MM:SS' string into a naive datetime."""
|
|
if not raw:
|
|
return None
|
|
try:
|
|
return datetime.strptime(str(raw).strip(), '%Y:%m:%d %H:%M:%S')
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
|
|
def _exif_datetime(exif):
|
|
"""Best available capture time from EXIF, or None."""
|
|
if not exif:
|
|
return None
|
|
for tag in (_EXIF_DATETIME_ORIGINAL, _EXIF_DATETIME_DIGITIZED, _EXIF_DATETIME):
|
|
dt = _parse_exif_datetime(exif.get(tag))
|
|
if dt:
|
|
return dt
|
|
return None
|
|
|
|
|
|
def _dms_to_decimal(dms, ref):
|
|
"""Convert EXIF degrees/minutes/seconds rationals to signed decimal."""
|
|
try:
|
|
deg, minutes, seconds = (float(x) for x in dms)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
value = deg + minutes / 60.0 + seconds / 3600.0
|
|
if str(ref).upper().strip() in ('S', 'W'):
|
|
value = -value
|
|
return value
|
|
|
|
|
|
def _exif_gps(exif):
|
|
"""Return (lat, lng) decimal degrees from EXIF GPSInfo, or (None, None)."""
|
|
if not exif:
|
|
return (None, None)
|
|
try:
|
|
gps = exif.get_ifd(_EXIF_GPS_IFD)
|
|
except Exception:
|
|
gps = None
|
|
if not gps:
|
|
return (None, None)
|
|
# 1/2 = LatitudeRef/Latitude, 3/4 = LongitudeRef/Longitude
|
|
lat = _dms_to_decimal(gps.get(2), gps.get(1)) if gps.get(2) and gps.get(1) else None
|
|
lng = _dms_to_decimal(gps.get(4), gps.get(3)) if gps.get(4) and gps.get(3) else None
|
|
return (lat, lng)
|
|
|
|
|
|
def _coerce_coord(value):
|
|
"""Parse a coordinate to float, rejecting out-of-range/garbage values."""
|
|
if value is None or value == '':
|
|
return None
|
|
try:
|
|
num = float(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
if num != num or abs(num) > 180: # NaN or impossible
|
|
return None
|
|
return num
|
|
|
|
|
|
def resolve_metadata(exif, captured_at=None, latitude=None, longitude=None):
|
|
"""Resolve (capture_dt, lat, lng, source) from client fields then EXIF.
|
|
|
|
``source`` is one of 'client', 'exif', or 'server' and describes where the
|
|
*timestamp* came from — useful for logging and for judging trust later.
|
|
"""
|
|
dt = _parse_client_datetime(captured_at)
|
|
source = 'client' if dt else None
|
|
|
|
lat = _coerce_coord(latitude)
|
|
lng = _coerce_coord(longitude)
|
|
|
|
if dt is None:
|
|
dt = _exif_datetime(exif)
|
|
source = 'exif' if dt else None
|
|
|
|
if lat is None or lng is None:
|
|
ex_lat, ex_lng = _exif_gps(exif)
|
|
lat = lat if lat is not None else ex_lat
|
|
lng = lng if lng is not None else ex_lng
|
|
|
|
if dt is None:
|
|
dt = now_eastern()
|
|
source = 'server'
|
|
|
|
return dt, lat, lng, source
|
|
|
|
|
|
# ── Overlay rendering ─────────────────────────────────────────────────────────
|
|
|
|
def _tz_abbrev(dt):
|
|
"""EDT/EST label for a naive Eastern datetime."""
|
|
try:
|
|
return EASTERN.localize(dt).strftime('%Z')
|
|
except Exception:
|
|
return 'ET'
|
|
|
|
|
|
def _overlay_lines(dt, lat, lng):
|
|
"""Text lines for the overlay bar."""
|
|
lines = [f"{dt.strftime('%Y-%m-%d %H:%M:%S')} {_tz_abbrev(dt)}"]
|
|
if lat is not None and lng is not None:
|
|
lines.append(f'{lat:.5f}, {lng:.5f}')
|
|
return lines
|
|
|
|
|
|
def _draw_overlay(img, lines):
|
|
"""Draw a translucent bar with *lines* across the bottom of *img*."""
|
|
from PIL import Image, ImageDraw
|
|
|
|
if img.mode not in ('RGB', 'RGBA'):
|
|
img = img.convert('RGB')
|
|
|
|
width, height = img.size
|
|
# Scale everything off the short edge so portrait and landscape match.
|
|
base = min(width, height)
|
|
font_size = max(14, int(base * 0.035))
|
|
pad = max(8, int(base * 0.018))
|
|
font = _load_font(font_size)
|
|
|
|
draw = ImageDraw.Draw(img)
|
|
|
|
# Measure the block.
|
|
heights, widths = [], []
|
|
for line in lines:
|
|
box = draw.textbbox((0, 0), line, font=font)
|
|
widths.append(box[2] - box[0])
|
|
heights.append(box[3] - box[1])
|
|
line_gap = max(2, int(font_size * 0.25))
|
|
text_h = sum(heights) + line_gap * (len(lines) - 1)
|
|
bar_h = text_h + pad * 2
|
|
|
|
# Translucent black bar, composited so it works on RGB too.
|
|
bar = Image.new('RGBA', (width, bar_h), (0, 0, 0, 150))
|
|
if img.mode == 'RGBA':
|
|
img.alpha_composite(bar, (0, height - bar_h))
|
|
else:
|
|
img.paste(Image.alpha_composite(
|
|
img.crop((0, height - bar_h, width, height)).convert('RGBA'), bar
|
|
).convert('RGB'), (0, height - bar_h))
|
|
|
|
draw = ImageDraw.Draw(img)
|
|
y = height - bar_h + pad
|
|
for line, h in zip(lines, heights):
|
|
# Thin dark outline keeps the text legible over a bright photo.
|
|
for dx, dy in ((-1, 0), (1, 0), (0, -1), (0, 1)):
|
|
draw.text((pad + dx, y + dy), line, font=font, fill=(0, 0, 0))
|
|
draw.text((pad, y), line, font=font, fill=(255, 255, 255))
|
|
y += h + line_gap
|
|
|
|
return img
|
|
|
|
|
|
# ── Public API ────────────────────────────────────────────────────────────────
|
|
|
|
def stamp_image_bytes(data, captured_at=None, latitude=None, longitude=None):
|
|
"""Burn the capture-time/geo overlay into *data*.
|
|
|
|
Returns ``(out_bytes, meta)``. On any failure — unsupported format, corrupt
|
|
image, missing Pillow — returns the ORIGINAL bytes with ``meta['stamped']``
|
|
False rather than raising, so an upload is never lost to a stamping bug.
|
|
"""
|
|
meta = {'stamped': False, 'captured_at': None, 'latitude': None,
|
|
'longitude': None, 'source': None}
|
|
try:
|
|
from PIL import Image, ImageOps
|
|
|
|
img = Image.open(io.BytesIO(data))
|
|
fmt = (img.format or '').upper()
|
|
|
|
try:
|
|
exif = img.getexif()
|
|
except Exception:
|
|
exif = None
|
|
|
|
dt, lat, lng, source = resolve_metadata(exif, captured_at, latitude, longitude)
|
|
meta.update({'captured_at': dt, 'latitude': lat,
|
|
'longitude': lng, 'source': source})
|
|
|
|
if fmt not in _STAMPABLE_FORMATS:
|
|
logger.info('PHOTO STAMP | skipped unsupported format=%s', fmt or '?')
|
|
return data, meta
|
|
|
|
# Honour the camera's EXIF orientation BEFORE drawing, otherwise the
|
|
# bar lands on a rotated edge and the re-encode (which drops EXIF)
|
|
# would leave the photo visibly rotated versus the original.
|
|
img = ImageOps.exif_transpose(img)
|
|
|
|
img = _draw_overlay(img, _overlay_lines(dt, lat, lng))
|
|
|
|
out = io.BytesIO()
|
|
if fmt == 'JPEG':
|
|
if img.mode != 'RGB':
|
|
img = img.convert('RGB')
|
|
img.save(out, format='JPEG', quality=88, optimize=True)
|
|
else:
|
|
img.save(out, format='PNG', optimize=True)
|
|
|
|
meta['stamped'] = True
|
|
return out.getvalue(), meta
|
|
|
|
except Exception as exc: # never lose a photo
|
|
logger.warning('PHOTO STAMP | failed, storing original: %s', exc)
|
|
return data, meta
|
|
|
|
|
|
def stamp_file_storage(file_obj, captured_at=None, latitude=None, longitude=None):
|
|
"""Return a FileStorage of the stamped image, plus the resolved metadata.
|
|
|
|
The result is a drop-in replacement for the incoming upload: it keeps the
|
|
original ``filename``/``content_type``, so ``storage.save()`` derives the
|
|
same key and works unchanged on both the local and s3 backends.
|
|
"""
|
|
from werkzeug.datastructures import FileStorage
|
|
|
|
file_obj.stream.seek(0)
|
|
original = file_obj.stream.read()
|
|
|
|
out_bytes, meta = stamp_image_bytes(
|
|
original, captured_at=captured_at, latitude=latitude, longitude=longitude
|
|
)
|
|
if not meta['stamped']:
|
|
file_obj.stream.seek(0) # hand back the untouched upload
|
|
return file_obj, meta
|
|
|
|
return FileStorage(
|
|
stream=io.BytesIO(out_bytes),
|
|
filename=file_obj.filename,
|
|
content_type=file_obj.content_type,
|
|
), meta
|