Jul 21 - Update uploaded photos location/timestamp
This commit is contained in:
@@ -0,0 +1,332 @@
|
||||
"""
|
||||
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
|
||||
Reference in New Issue
Block a user