from __future__ import annotations import asyncio import io import json import logging import os import re import shutil import subprocess import tempfile import threading import time import uuid import urllib.error import urllib.parse import urllib.request from collections import OrderedDict from datetime import datetime from pathlib import Path from typing import Annotated, Any from fastapi import FastAPI, File, Form, HTTPException, Request, UploadFile from pydantic import BaseModel, Field from fastapi.responses import FileResponse, Response, StreamingResponse from fastapi.staticfiles import StaticFiles from PIL import Image, ImageOps, UnidentifiedImageError from PIL import ImageSequence, JpegImagePlugin FORMAT_GROUPS: dict[str, tuple[str, ...]] = { "jpg": (".jpg", ".jpeg"), "png": (".png",), "tiff": (".tif", ".tiff"), "heif": (".heic", ".heif", ".hif"), "dng": (".dng",), "canon": (".cr2", ".cr3"), "nikon": (".nef", ".nrw"), "sony": (".arw", ".arq"), } FORMAT_LABELS = { "jpg": "JPG", "png": "PNG", "tiff": "TIFF", "heif": "HEIF", "dng": "DNG", "canon": "Canon RAW", "nikon": "Nikon RAW", "sony": "Sony RAW", } def _enabled_format_groups() -> tuple[str, ...]: configured = os.getenv("ENABLED_FORMATS", ",".join(FORMAT_GROUPS)) requested = tuple(dict.fromkeys( value.strip().casefold() for value in configured.split(",") if value.strip() )) unknown = sorted(set(requested) - set(FORMAT_GROUPS)) if unknown: raise RuntimeError( "Unknown ENABLED_FORMATS value(s): " f"{', '.join(unknown)}. Available groups: {', '.join(FORMAT_GROUPS)}." ) if not requested: raise RuntimeError("ENABLED_FORMATS must contain at least one format group.") return tuple(group for group in FORMAT_GROUPS if group in requested) def _format_list_text(groups: tuple[str, ...]) -> str: labels = [FORMAT_LABELS[group] for group in groups] if len(labels) == 1: return labels[0] if len(labels) == 2: return " and ".join(labels) return f"{', '.join(labels[:-1])} and {labels[-1]}" ENABLED_FORMAT_GROUPS = _enabled_format_groups() SUPPORTED_EXTENSIONS = { extension for group in ENABLED_FORMAT_GROUPS for extension in FORMAT_GROUPS[group] } ENABLED_FORMAT_LABEL = _format_list_text(ENABLED_FORMAT_GROUPS) APP_NAME = os.getenv("APP_NAME", "Photo Date Editor") APP_TITLE = os.getenv("APP_TITLE") or APP_NAME APP_SUBTITLE = ( os.getenv("APP_SUBTITLE") or f"Local folder ยท in-place {ENABLED_FORMAT_LABEL} metadata" ) APP_URL = os.getenv("APP_URL", "http://localhost:8080") APP_VERSION = "0.9.2" MAX_UPLOAD_MB = int(os.getenv("MAX_UPLOAD_MB", "150")) MAX_UPLOAD_BYTES = MAX_UPLOAD_MB * 1024 * 1024 PREVIEW_MAX_EDGE = int(os.getenv("PREVIEW_MAX_EDGE", "2400")) RAW_PREVIEW_MAX_EDGE = int(os.getenv("RAW_PREVIEW_MAX_EDGE", "1600")) PREVIEW_MAX_PIXELS = int(os.getenv("PREVIEW_MAX_PIXELS", "300000000")) Image.MAX_IMAGE_PIXELS = PREVIEW_MAX_PIXELS TIFF_EXTENSIONS = {".tif", ".tiff"} PIXEL_ROTATION_EXTENSIONS = {".jpg", ".jpeg", ".png", ".tif", ".tiff"} HEIF_EXTENSIONS = {".heic", ".heif", ".hif"} DNG_EXTENSIONS = {".dng"} CANON_EXTENSIONS = {".cr2", ".cr3"} NIKON_EXTENSIONS = {".nef", ".nrw"} SONY_EXTENSIONS = {".arw", ".arq"} RAW_EXTENSIONS = DNG_EXTENSIONS | CANON_EXTENSIONS | NIKON_EXTENSIONS | SONY_EXTENSIONS IPTC_EXTENSIONS = { ".jpg", ".jpeg", ".tif", ".tiff", ".dng", ".cr2", ".nef", ".nrw", ".arw", ".arq", } SERVER_PREVIEW_EXTENSIONS = ( TIFF_EXTENSIONS | HEIF_EXTENSIONS | RAW_EXTENSIONS ) & SUPPORTED_EXTENSIONS MEDIA_TYPES = { ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png", ".tif": "image/tiff", ".tiff": "image/tiff", ".heic": "image/heic", ".heif": "image/heif", ".hif": "image/heif", ".dng": "image/x-adobe-dng", ".cr2": "image/x-canon-cr2", ".cr3": "image/x-canon-cr3", ".nef": "image/x-nikon-nef", ".nrw": "image/x-nikon-nrw", ".arw": "image/x-sony-arw", ".arq": "image/x-sony-arq", } LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper() MAP_TILE_SOURCE_URL = os.getenv( "MAP_TILE_SOURCE_URL", os.getenv("MAP_TILE_URL", "https://tile.openstreetmap.org/{z}/{x}/{y}.png"), ) MAP_ATTRIBUTION = os.getenv( "MAP_ATTRIBUTION", '© OpenStreetMap contributors', ) DEFAULT_MAP_LAT = float(os.getenv("DEFAULT_MAP_LAT", "64.5")) DEFAULT_MAP_LON = float(os.getenv("DEFAULT_MAP_LON", "11.0")) DEFAULT_MAP_ZOOM = int(os.getenv("DEFAULT_MAP_ZOOM", "5")) GEOCODER_URL = os.getenv("GEOCODER_URL", "https://nominatim.openstreetmap.org/search") GEOCODER_USER_AGENT = os.getenv( "GEOCODER_USER_AGENT", f"{APP_NAME.replace(' ', '')}/{APP_VERSION} (+{APP_URL})", ) _geocode_lock = asyncio.Lock() _geocode_cache: dict[str, list[dict[str, Any]]] = {} _last_geocode_request = 0.0 _tile_cache: OrderedDict[str, tuple[bytes, str]] = OrderedDict() _tile_cache_limit = 512 _tile_lock = asyncio.Lock() _decoder_lock = threading.Lock() _heif_decoder_registered = False _rawpy_module: Any | None = None logging.basicConfig( level=getattr(logging, LOG_LEVEL, logging.INFO), format="%(asctime)s %(levelname)s %(name)s: %(message)s", ) logger = logging.getLogger("photo-date-editor") BASE_DIR = Path(__file__).resolve().parent STATIC_DIR = BASE_DIR / "static" DATA_DIR = Path(os.getenv("DATA_DIR", "/data")) FAVORITES_FILE = DATA_DIR / "favorite-locations.json" _favorites_lock = threading.Lock() class FavoriteLocationInput(BaseModel): name: str = Field(min_length=1, max_length=120) latitude: float = Field(ge=-90, le=90) longitude: float = Field(ge=-180, le=180) app = FastAPI(title=APP_TITLE, docs_url=None, redoc_url=None) @app.middleware("http") async def request_timing(request: Request, call_next: Any) -> Response: started = time.perf_counter() response = await call_next(request) elapsed_seconds = time.perf_counter() - started if request.url.path in {"/api/inspect", "/api/process"}: response.headers["X-Request-Duration-Ms"] = f"{elapsed_seconds * 1000:.1f}" logger.info("%s completed in %.2fs end-to-end on the server", request.url.path, elapsed_seconds) return response @app.get("/api/health") def health() -> dict[str, str]: return {"status": "ok", "version": APP_VERSION} @app.get("/api/config") def config() -> dict[str, Any]: return { "appName": APP_NAME, "appTitle": APP_TITLE, "appSubtitle": APP_SUBTITLE, "appUrl": APP_URL, "version": APP_VERSION, "maxUploadMb": MAX_UPLOAD_MB, "enabledFormats": list(ENABLED_FORMAT_GROUPS), "enabledExtensions": sorted(SUPPORTED_EXTENSIONS), "enabledFormatLabel": ENABLED_FORMAT_LABEL, "mapTileUrl": "/api/map/tiles/{z}/{x}/{y}.png", "mapAttribution": MAP_ATTRIBUTION, "defaultMapLat": DEFAULT_MAP_LAT, "defaultMapLon": DEFAULT_MAP_LON, "defaultMapZoom": DEFAULT_MAP_ZOOM, } def _load_favorites_unlocked() -> list[dict[str, Any]]: if not FAVORITES_FILE.exists(): return [] try: payload = json.loads(FAVORITES_FILE.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: logger.warning("Could not read favorite locations: %s", exc) return [] raw = payload.get("favorites", []) if isinstance(payload, dict) else [] favorites: list[dict[str, Any]] = [] for item in raw if isinstance(raw, list) else []: if not isinstance(item, dict): continue try: latitude = float(item["latitude"]) longitude = float(item["longitude"]) name = str(item["name"]).strip() favorite_id = str(item["id"]).strip() except (KeyError, TypeError, ValueError): continue if not name or not favorite_id or not -90 <= latitude <= 90 or not -180 <= longitude <= 180: continue favorites.append({ "id": favorite_id, "name": name, "latitude": latitude, "longitude": longitude, }) return sorted(favorites, key=lambda item: item["name"].casefold()) def _write_favorites_unlocked(favorites: list[dict[str, Any]]) -> None: DATA_DIR.mkdir(parents=True, exist_ok=True) payload = {"favorites": sorted(favorites, key=lambda item: item["name"].casefold())} temporary = FAVORITES_FILE.with_suffix(".tmp") temporary.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") temporary.replace(FAVORITES_FILE) @app.get("/api/favorites") def list_favorites() -> dict[str, list[dict[str, Any]]]: with _favorites_lock: return {"favorites": _load_favorites_unlocked()} @app.post("/api/favorites", status_code=201) def create_favorite(value: FavoriteLocationInput) -> dict[str, dict[str, Any]]: name = " ".join(value.name.split()) if not name: raise HTTPException(status_code=422, detail="Favorite name cannot be empty.") with _favorites_lock: favorites = _load_favorites_unlocked() if any(item["name"].casefold() == name.casefold() for item in favorites): raise HTTPException(status_code=409, detail="A favorite with that name already exists.") favorite = { "id": uuid.uuid4().hex, "name": name, "latitude": round(value.latitude, 6), "longitude": round(value.longitude, 6), } favorites.append(favorite) _write_favorites_unlocked(favorites) return {"favorite": favorite} @app.delete("/api/favorites/{favorite_id}", status_code=204) def delete_favorite(favorite_id: str) -> Response: with _favorites_lock: favorites = _load_favorites_unlocked() remaining = [item for item in favorites if item["id"] != favorite_id] if len(remaining) == len(favorites): raise HTTPException(status_code=404, detail="Favorite location not found.") _write_favorites_unlocked(remaining) return Response(status_code=204) def _fetch_map_tile(z: int, x: int, y: int) -> tuple[bytes, str]: tile_url = ( MAP_TILE_SOURCE_URL .replace("{z}", str(z)) .replace("{x}", str(x)) .replace("{y}", str(y)) ) request = urllib.request.Request( tile_url, headers={ "User-Agent": GEOCODER_USER_AGENT, "Accept": "image/avif,image/webp,image/png,image/*,*/*;q=0.8", }, ) with urllib.request.urlopen(request, timeout=20) as response: content_type = response.headers.get_content_type() or "image/png" return response.read(), content_type @app.get("/api/map/tiles/{z}/{x}/{y}.png", include_in_schema=False) async def map_tile(z: int, x: int, y: int) -> Response: if not 0 <= z <= 19: raise HTTPException(status_code=404, detail="Unsupported map zoom level.") limit = 1 << z if not 0 <= x < limit or not 0 <= y < limit: raise HTTPException(status_code=404, detail="Invalid map tile coordinates.") cache_key = f"{z}/{x}/{y}" cached = _tile_cache.get(cache_key) if cached is not None: _tile_cache.move_to_end(cache_key) payload, content_type = cached return Response(payload, media_type=content_type, headers={"Cache-Control": "public, max-age=86400"}) async with _tile_lock: cached = _tile_cache.get(cache_key) if cached is None: try: cached = await asyncio.to_thread(_fetch_map_tile, z, x, y) except (urllib.error.URLError, TimeoutError) as exc: logger.warning("Map tile request failed for %s: %s", cache_key, exc) raise HTTPException(status_code=502, detail="The map tile service could not be reached.") from exc _tile_cache[cache_key] = cached if len(_tile_cache) > _tile_cache_limit: _tile_cache.popitem(last=False) else: _tile_cache.move_to_end(cache_key) payload, content_type = cached return Response(payload, media_type=content_type, headers={"Cache-Control": "public, max-age=86400"}) def _fetch_geocode_results(query: str) -> list[dict[str, Any]]: parameters = urllib.parse.urlencode( { "q": query, "format": "jsonv2", "addressdetails": "1", "limit": "5", } ) request = urllib.request.Request( f"{GEOCODER_URL}?{parameters}", headers={ "User-Agent": GEOCODER_USER_AGENT, "Accept": "application/json", }, ) with urllib.request.urlopen(request, timeout=20) as response: raw = json.loads(response.read().decode("utf-8")) results: list[dict[str, Any]] = [] for item in raw if isinstance(raw, list) else []: try: latitude = float(item["lat"]) longitude = float(item["lon"]) except (KeyError, TypeError, ValueError): continue address = item.get("address") if isinstance(item.get("address"), dict) else {} results.append( { "label": str(item.get("display_name") or "").strip(), "latitude": latitude, "longitude": longitude, "type": str(item.get("type") or item.get("category") or "place"), "address": { key: str(address[key]) for key in ( "house_number", "road", "neighbourhood", "suburb", "city", "town", "village", "municipality", "county", "state", "postcode", "country", "country_code", ) if address.get(key) }, } ) return results @app.get("/api/geocode") async def geocode(q: str) -> dict[str, list[dict[str, Any]]]: global _last_geocode_request query = " ".join(q.split()) if len(query) < 3: raise HTTPException(status_code=422, detail="Enter at least three characters to search.") if len(query) > 200: raise HTTPException(status_code=422, detail="The address search is too long.") cache_key = query.casefold() if cache_key in _geocode_cache: return {"results": _geocode_cache[cache_key]} async with _geocode_lock: if cache_key in _geocode_cache: return {"results": _geocode_cache[cache_key]} delay = 1.0 - (time.monotonic() - _last_geocode_request) if delay > 0: await asyncio.sleep(delay) try: results = await asyncio.to_thread(_fetch_geocode_results, query) except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc: logger.warning("Geocoding request failed: %s", exc) raise HTTPException(status_code=502, detail="The address search service could not be reached.") from exc finally: _last_geocode_request = time.monotonic() if len(_geocode_cache) >= 200: _geocode_cache.pop(next(iter(_geocode_cache))) _geocode_cache[cache_key] = results return {"results": results} def _normalise_datetime( precision: str, year: int, month: int | None, day: int | None, time_value: str | None, ) -> tuple[str, str, str]: """Return EXIF datetime, XMP reduced-precision date, and a readable precision label.""" if not 1800 <= year <= 2200: raise HTTPException(status_code=422, detail="Year must be between 1800 and 2200.") precision_labels = { "exact": "Exact date", "month": "Month and year", "year": "Year only", "approximate": "Approximate year", } if precision not in precision_labels: raise HTTPException(status_code=422, detail="Unsupported date precision.") if precision == "exact": if month is None or day is None: raise HTTPException(status_code=422, detail="Exact dates require month and day.") effective_month = month effective_day = day xmp_date = f"{year:04d}-{month:02d}-{day:02d}" elif precision == "month": if month is None: raise HTTPException(status_code=422, detail="Month-and-year dates require a month.") effective_month = month effective_day = 1 xmp_date = f"{year:04d}-{month:02d}" else: effective_month = 1 effective_day = 1 xmp_date = f"{year:04d}" effective_time = time_value.strip() if time_value else "12:00:00" try: parsed = datetime.strptime( f"{year:04d}-{effective_month:02d}-{effective_day:02d} {effective_time}", "%Y-%m-%d %H:%M:%S", ) except ValueError as exc: raise HTTPException(status_code=422, detail=f"Invalid date or time: {exc}") from exc exif_datetime = parsed.strftime("%Y:%m:%d %H:%M:%S") return exif_datetime, xmp_date, precision_labels[precision] def _format_name(suffix: str) -> str: if suffix in {".jpg", ".jpeg"}: return "JPEG" if suffix == ".png": return "PNG" if suffix in TIFF_EXTENSIONS: return "TIFF" if suffix in HEIF_EXTENSIONS: return "HEIF" if suffix in DNG_EXTENSIONS: return "DNG" if suffix == ".cr2": return "CR2" if suffix == ".cr3": return "CR3" if suffix == ".nef": return "NEF" if suffix == ".nrw": return "NRW" if suffix == ".arw": return "ARW" if suffix == ".arq": return "ARQ" return suffix.lstrip(".").upper() or "image" def _validate_image_signature(path: Path, suffix: str) -> None: with path.open("rb") as source: signature = source.read(64) valid = False if suffix in {".jpg", ".jpeg"}: valid = signature.startswith(b"\xff\xd8\xff") elif suffix == ".png": valid = signature.startswith(b"\x89PNG\r\n\x1a\n") elif suffix in TIFF_EXTENSIONS: valid = signature.startswith((b"II*\x00", b"MM\x00*", b"II+\x00", b"MM\x00+")) elif suffix in HEIF_EXTENSIONS: hevc_brands = { b"heic", b"heix", b"hevc", b"hevx", b"heim", b"heis", b"hevm", b"hevs", } valid = len(signature) >= 16 and signature[4:8] == b"ftyp" and any( signature[offset:offset + 4] in hevc_brands for offset in range(8, len(signature) - 3, 4) ) elif suffix in DNG_EXTENSIONS: # DNG is TIFF-based. ExifTool performs the deeper DNG structure # validation when reading, extracting a preview, or writing metadata. valid = signature.startswith((b"II*\x00", b"MM\x00*")) elif suffix == ".cr2": valid = ( signature.startswith((b"II*\x00", b"MM\x00*")) and signature[8:12] == b"CR\x02\x00" ) elif suffix == ".cr3": valid = ( len(signature) >= 16 and signature[4:8] == b"ftyp" and any( signature[offset:offset + 4] == b"crx " for offset in range(8, len(signature) - 3, 4) ) ) elif suffix in NIKON_EXTENSIONS | SONY_EXTENSIONS: # NEF, NRW, ARW, and ARQ are TIFF-based. ExifTool performs deeper # camera-format validation when the file is inspected or written. valid = signature.startswith((b"II*\x00", b"MM\x00*")) if not valid: raise HTTPException( status_code=415, detail=f"The uploaded file is not a valid {_format_name(suffix)} image.", ) async def _save_upload(upload: UploadFile, destination: Path, suffix: str) -> None: total = 0 with destination.open("wb") as output: while chunk := await upload.read(1024 * 1024): total += len(chunk) if total > MAX_UPLOAD_BYTES: raise HTTPException( status_code=413, detail=f"Image is larger than the configured {MAX_UPLOAD_MB} MB limit.", ) output.write(chunk) _validate_image_signature(destination, suffix) def _first_value(metadata: dict[str, Any], *keys: str) -> Any: for key in keys: value = metadata.get(key) if value not in (None, "", []): return value return None def _as_text(value: Any) -> str: if isinstance(value, list): return str(value[0]).strip() if value else "" if isinstance(value, dict): for candidate in ("x-default", "en", "en-US"): if candidate in value: return str(value[candidate]).strip() return str(next(iter(value.values()), "")).strip() return str(value or "").strip() def _as_keywords(*values: Any) -> list[str]: output: list[str] = [] seen: set[str] = set() for value in values: items = value if isinstance(value, list) else [value] for item in items: text = _as_text(item) if not text or text.casefold() in seen: continue seen.add(text.casefold()) output.append(text) return output def _signed_coordinate(value: Any, reference: Any, negative_letter: str) -> float | None: try: coordinate = float(value) except (TypeError, ValueError): return None if coordinate < 0: return coordinate reference_text = _as_text(reference).upper() if reference_text == negative_letter or reference_text == "1": return -coordinate return coordinate def _metadata_values(metadata: dict[str, Any]) -> tuple[bool, bool, dict[str, Any]]: date_text = _as_text( _first_value( metadata, "DateTimeOriginal", "CreateDate", "DateCreated", "ModifyDate", ) ) instructions = _as_text(_first_value(metadata, "Instructions")) description = _as_text( _first_value( metadata, "ImageDescription", "Description", "Caption-Abstract", ) ) keywords = _as_keywords(metadata.get("Subject"), metadata.get("Keywords")) latitude = _signed_coordinate(metadata.get("GPSLatitude"), metadata.get("GPSLatitudeRef"), "S") longitude = _signed_coordinate(metadata.get("GPSLongitude"), metadata.get("GPSLongitudeRef"), "W") location_name = _as_text( _first_value(metadata, "Location", "LocationShownLocationName", "City") ) precision = "exact" time_mode = "known" marker = re.search( r"Photo Date Editor precision:\s*(Exact date|Month and year|Year only|Approximate year)" r"(?:;\s*time:\s*(Known|Unknown))?", instructions, flags=re.IGNORECASE, ) if marker: precision = { "exact date": "exact", "month and year": "month", "year only": "year", "approximate year": "approximate", }[marker.group(1).casefold()] if marker.group(2): time_mode = "known" if marker.group(2).casefold() == "known" else "unknown" parsed_date: datetime | None = None for pattern in ("%Y:%m:%d %H:%M:%S", "%Y-%m-%d %H:%M:%S", "%Y:%m:%d", "%Y-%m-%d"): try: parsed_date = datetime.strptime(date_text, pattern) break except ValueError: continue if marker and not marker.group(2) and parsed_date and parsed_date.strftime("%H:%M:%S") == "12:00:00": # Backward compatibility with v0.1, which used noon for unknown time but did not store a marker. time_mode = "unknown" values = { "precision": precision, "year": str(parsed_date.year) if parsed_date else "", "month": str(parsed_date.month) if parsed_date else "1", "day": str(parsed_date.day) if parsed_date else "1", "timeMode": time_mode, "time": parsed_date.strftime("%H:%M:%S") if parsed_date else "12:00:00", "description": description, "keywords": ", ".join(keywords), "latitude": f"{latitude:.8f}" if latitude is not None else "", "longitude": f"{longitude:.8f}" if longitude is not None else "", "locationName": location_name, "locationDirty": False, } has_metadata = bool(parsed_date or description or keywords or marker or latitude is not None or longitude is not None or location_name) edited_by_app = bool(marker) return has_metadata, edited_by_app, values def _ensure_heif_decoder() -> None: """Register pillow-heif once, on the first HEIF preview request.""" global _heif_decoder_registered if _heif_decoder_registered: return with _decoder_lock: if _heif_decoder_registered: return from pillow_heif import register_heif_opener register_heif_opener() _heif_decoder_registered = True logger.info("HEIF preview decoder loaded.") def _get_rawpy() -> Any: """Import rawpy once, on the first RAW file that needs a rendered fallback.""" global _rawpy_module if _rawpy_module is not None: return _rawpy_module with _decoder_lock: if _rawpy_module is None: import rawpy _rawpy_module = rawpy logger.info("LibRaw preview decoder loaded.") return _rawpy_module def _raw_preview_tags(suffix: str) -> tuple[str, ...]: if suffix == ".cr2": return ("PreviewImage", "JpgFromRaw", "ThumbnailImage") if suffix == ".cr3": return ("JpgFromRaw", "PreviewImage", "ThumbnailImage") if suffix in NIKON_EXTENSIONS: return ("JpgFromRaw", "PreviewImage", "OtherImage", "ThumbnailImage") if suffix in SONY_EXTENSIONS: return ("JpgFromRaw", "PreviewImage", "ThumbnailImage") return ("JpgFromRaw", "PreviewImage", "ThumbnailImage") def _embedded_raw_preview(path: Path, format_name: str, suffix: str) -> Image.Image | None: """Return the first usable embedded JPEG, ordered for the RAW container.""" for tag in _raw_preview_tags(suffix): try: completed = subprocess.run( ["exiftool", "-b", f"-{tag}", "--", str(path)], capture_output=True, timeout=45, check=False, ) except subprocess.TimeoutExpired: logger.warning("Timed out extracting %s %s data.", format_name, tag) continue if completed.returncode != 0 or not completed.stdout: continue try: with Image.open(io.BytesIO(completed.stdout)) as source: source.load() preview = source.copy() logger.info("Using %s for %s preview.", tag, format_name) return preview except (UnidentifiedImageError, OSError, ValueError): logger.debug("%s %s data was not a usable image preview.", format_name, tag) return None def _render_raw(path: Path, format_name: str) -> Image.Image: """Render a half-size RGB fallback when RAW has no embedded JPEG preview.""" rawpy = _get_rawpy() try: with rawpy.imread(str(path)) as raw: width = int(raw.sizes.width) height = int(raw.sizes.height) if width <= 0 or height <= 0 or width * height > PREVIEW_MAX_PIXELS: raise ValueError( f"{format_name} dimensions exceed the configured preview pixel limit." ) rgb = raw.postprocess( use_camera_wb=True, use_auto_wb=False, no_auto_bright=False, half_size=True, output_bps=8, ) except rawpy.LibRawError as exc: raise ValueError(f"LibRaw could not decode this {format_name} file.") from exc return Image.fromarray(rgb) def _transpose_for_clockwise_rotation(image: Image.Image, rotation: int) -> Image.Image: operation = { 90: Image.Transpose.ROTATE_270, 180: Image.Transpose.ROTATE_180, 270: Image.Transpose.ROTATE_90, }[rotation] return image.transpose(operation) def _copy_metadata(source: Path, destination: Path) -> None: completed = subprocess.run( [ "exiftool", "-overwrite_original", "-m", "-TagsFromFile", str(source), "-all:all", str(destination), ], capture_output=True, text=True, timeout=90, check=False, ) if completed.returncode != 0: logger.error("ExifTool metadata restore after rotation failed: %s", completed.stderr.strip()) raise ValueError("Metadata could not be preserved after rotating the image.") def _rotate_jpeg(source: Path, destination: Path, rotation: int) -> str: with Image.open(source) as image: orientation = int(image.getexif().get(274, 1) or 1) if orientation == 1: completed = subprocess.run( [ "jpegtran", "-copy", "all", "-perfect", "-rotate", str(rotation), "-outfile", str(destination), str(source), ], capture_output=True, timeout=90, check=False, ) if completed.returncode == 0 and destination.exists(): return "jpeg-lossless" with Image.open(source) as image: image.load() corrected = ImageOps.exif_transpose(image) rotated = _transpose_for_clockwise_rotation(corrected, rotation) try: sampling = JpegImagePlugin.get_sampling(image) save_options: dict[str, Any] = { "format": "JPEG", "quality": 95, "optimize": True, "progressive": bool(image.info.get("progressive") or image.info.get("progression")), } if sampling >= 0: save_options["subsampling"] = sampling if image.info.get("icc_profile"): save_options["icc_profile"] = image.info["icc_profile"] rotated.save(destination, **save_options) finally: if corrected is not image: corrected.close() rotated.close() _copy_metadata(source, destination) return "jpeg-reencoded" def _rotate_png_or_tiff(source: Path, destination: Path, suffix: str, rotation: int) -> str: with Image.open(source) as image: image.seek(0) if suffix == ".png" and int(getattr(image, "n_frames", 1)) > 1: raise ValueError("Animated PNG rotation is not supported.") frames: list[Image.Image] = [] for frame in ImageSequence.Iterator(image): base = frame.copy() corrected = ImageOps.exif_transpose(base) if corrected is not base: base.close() rotated = _transpose_for_clockwise_rotation(corrected, rotation) corrected.close() frames.append(rotated) if not frames: raise ValueError("The image contains no rotatable frames.") try: if suffix == ".png": options: dict[str, Any] = {"format": "PNG", "optimize": True} if image.info.get("icc_profile"): options["icc_profile"] = image.info["icc_profile"] if image.info.get("dpi"): options["dpi"] = image.info["dpi"] frames[0].save(destination, **options) mode = "png-lossless" else: options = { "format": "TIFF", "save_all": len(frames) > 1, "append_images": frames[1:], "compression": image.info.get("compression", "tiff_deflate"), } if image.info.get("icc_profile"): options["icc_profile"] = image.info["icc_profile"] if image.info.get("dpi"): options["dpi"] = image.info["dpi"] frames[0].save(destination, **options) mode = "tiff-lossless-reencode" finally: for frame in frames: frame.close() _copy_metadata(source, destination) return mode def _rotate_pixels(path: Path, suffix: str, rotation: int) -> str: source = path.with_name(f"rotation-source{suffix}") destination = path.with_name(f"rotation-output{suffix}") shutil.copy2(path, source) try: if suffix in {".jpg", ".jpeg"}: mode = _rotate_jpeg(source, destination, rotation) else: mode = _rotate_png_or_tiff(source, destination, suffix, rotation) destination.replace(path) return mode except (Image.DecompressionBombError, UnidentifiedImageError, OSError, ValueError) as exc: logger.warning("Pixel rotation failed for %s: %s", path.name, exc) raise HTTPException(status_code=422, detail=f"The image could not be rotated safely: {exc}") from exc def _preview_bytes(temp_path: Path, suffix: str, filename: str) -> bytes: try: if suffix in RAW_EXTENSIONS: format_name = _format_name(suffix) preview = _embedded_raw_preview(temp_path, format_name, suffix) if preview is None: preview = _render_raw(temp_path, format_name) preview = ImageOps.exif_transpose(preview) else: if suffix in HEIF_EXTENSIONS: _ensure_heif_decoder() with Image.open(temp_path) as source: source.seek(0) preview = ImageOps.exif_transpose(source) preview.load() try: max_edge = RAW_PREVIEW_MAX_EDGE if suffix in RAW_EXTENSIONS else PREVIEW_MAX_EDGE preview.thumbnail((max_edge, max_edge), Image.Resampling.LANCZOS) if preview.mode in {"RGBA", "LA"} or (preview.mode == "P" and "transparency" in preview.info): rgba = preview.convert("RGBA") background = Image.new("RGBA", rgba.size, (255, 255, 255, 255)) background.alpha_composite(rgba) preview = background.convert("RGB") elif preview.mode != "RGB": preview = preview.convert("RGB") output = io.BytesIO() preview.save(output, format="JPEG", quality=86, optimize=True) return output.getvalue() finally: preview.close() except (Image.DecompressionBombError, UnidentifiedImageError, OSError, ValueError) as exc: format_name = _format_name(suffix) logger.warning("%s preview generation failed for %s: %s", format_name, filename, exc) raise HTTPException( status_code=422, detail=f"The {format_name} image could not be rendered for preview.", ) from exc def _read_metadata_path(temp_path: Path) -> dict[str, bool | dict[str, Any]]: completed = subprocess.run( [ "exiftool", "-json", "-s", "-n", "-DateTimeOriginal", "-CreateDate", "-ModifyDate", "-DateCreated", "-Instructions", "-ImageDescription", "-Description", "-Caption-Abstract", "-Subject", "-Keywords", "-GPSLatitude", "-GPSLatitudeRef", "-GPSLongitude", "-GPSLongitudeRef", "-Location", "-LocationShownLocationName", "-City", str(temp_path), ], capture_output=True, text=True, timeout=45, check=False, ) if completed.returncode != 0: logger.error("ExifTool metadata read failed: %s", completed.stderr.strip()) raise HTTPException(status_code=500, detail="ExifTool could not read this image's metadata.") try: output = json.loads(completed.stdout) metadata = output[0] if isinstance(output, list) and output else {} except json.JSONDecodeError as exc: raise HTTPException(status_code=500, detail="ExifTool returned invalid metadata output.") from exc has_metadata, edited_by_app, values = _metadata_values(metadata) return {"hasMetadata": has_metadata, "editedByApp": edited_by_app, "values": values} @app.post("/api/preview") async def create_preview(file: Annotated[UploadFile, File(...)]) -> Response: """Generate a browser-friendly preview for server-preview formats.""" filename = file.filename or "photo.tiff" suffix = Path(filename).suffix.lower() if suffix not in SUPPORTED_EXTENSIONS or suffix not in SERVER_PREVIEW_EXTENSIONS: raise HTTPException(status_code=415, detail="This format is disabled or does not require a server preview.") with tempfile.TemporaryDirectory(prefix="photo-date-editor-preview-") as temp_dir: temp_path = Path(temp_dir) / f"source{suffix}" await _save_upload(file, temp_path, suffix) preview_bytes = _preview_bytes(temp_path, suffix, filename) return Response(content=preview_bytes, media_type="image/jpeg", headers={"Cache-Control": "no-store"}) @app.post("/api/inspect") async def inspect_photo(file: Annotated[UploadFile, File(...)]) -> Response: """Return length-prefixed JSON metadata followed by binary JPEG preview data.""" filename = file.filename or "photo.tiff" suffix = Path(filename).suffix.lower() if suffix not in SUPPORTED_EXTENSIONS or suffix not in SERVER_PREVIEW_EXTENSIONS: raise HTTPException(status_code=415, detail="This format is disabled or does not require server inspection.") with tempfile.TemporaryDirectory(prefix="photo-date-editor-inspect-") as temp_dir: temp_path = Path(temp_dir) / f"source{suffix}" copy_started = time.perf_counter() await _save_upload(file, temp_path, suffix) copy_seconds = time.perf_counter() - copy_started preview_started = time.perf_counter() preview_bytes = _preview_bytes(temp_path, suffix, filename) preview_seconds = time.perf_counter() - preview_started metadata_started = time.perf_counter() metadata = _read_metadata_path(temp_path) metadata_seconds = time.perf_counter() - metadata_started metadata_bytes = json.dumps(metadata, ensure_ascii=False, separators=(",", ":")).encode("utf-8") body = len(metadata_bytes).to_bytes(4, "big") + metadata_bytes + preview_bytes logger.info( "Inspected %s: temporary copy %.2fs, preview %.2fs, metadata %.2fs", filename, copy_seconds, preview_seconds, metadata_seconds, ) return Response( content=body, media_type="application/vnd.photo-date-editor.inspect", headers={ "Cache-Control": "no-store", "Server-Timing": ( f'copy;dur={copy_seconds * 1000:.1f}, ' f'preview;dur={preview_seconds * 1000:.1f}, ' f'metadata;dur={metadata_seconds * 1000:.1f}' ), }, ) @app.post("/api/metadata") async def read_metadata(file: Annotated[UploadFile, File(...)]) -> dict[str, bool | dict[str, Any]]: filename = file.filename or "photo.jpg" suffix = Path(filename).suffix.lower() if suffix not in SUPPORTED_EXTENSIONS: raise HTTPException( status_code=415, detail=f"This format is disabled. Enabled format groups: {ENABLED_FORMAT_LABEL}.", ) with tempfile.TemporaryDirectory(prefix="photo-date-editor-read-") as temp_dir: temp_path = Path(temp_dir) / f"source{suffix}" await _save_upload(file, temp_path, suffix) result = _read_metadata_path(temp_path) return result @app.post("/api/process") async def process_photo( file: Annotated[UploadFile, File(...)], precision: Annotated[str, Form(...)], year: Annotated[int, Form(...)], month: Annotated[int | None, Form()] = None, day: Annotated[int | None, Form()] = None, time_value: Annotated[str | None, Form()] = None, description: Annotated[str, Form()] = "", keywords_json: Annotated[str, Form()] = "[]", gps_action: Annotated[str, Form()] = "preserve", latitude: Annotated[float | None, Form()] = None, longitude: Annotated[float | None, Form()] = None, location_name: Annotated[str, Form()] = "", rotation: Annotated[int, Form()] = 0, ) -> Response: filename = file.filename or "photo.jpg" suffix = Path(filename).suffix.lower() if suffix not in SUPPORTED_EXTENSIONS: raise HTTPException( status_code=415, detail=f"This format is disabled. Enabled format groups: {ENABLED_FORMAT_LABEL}.", ) if rotation not in {0, 90, 180, 270}: raise HTTPException(status_code=422, detail="Rotation must be 0, 90, 180, or 270 degrees.") if rotation and suffix not in PIXEL_ROTATION_EXTENSIONS: raise HTTPException( status_code=422, detail=f"Pixel rotation is not supported for {_format_name(suffix)} files.", ) exif_datetime, xmp_date, precision_label = _normalise_datetime( precision=precision, year=year, month=month, day=day, time_value=time_value, ) time_label = "Known" if time_value else "Unknown" try: raw_keywords = json.loads(keywords_json) if not isinstance(raw_keywords, list): raise ValueError keywords = [str(value).strip() for value in raw_keywords if str(value).strip()] except (json.JSONDecodeError, ValueError) as exc: raise HTTPException(status_code=422, detail="Keywords must be a JSON array.") from exc if gps_action not in {"preserve", "set", "clear"}: raise HTTPException(status_code=422, detail="Unsupported GPS action.") if gps_action == "set": if latitude is None or longitude is None: raise HTTPException(status_code=422, detail="Both latitude and longitude are required.") if not -90 <= latitude <= 90 or not -180 <= longitude <= 180: raise HTTPException(status_code=422, detail="The GPS coordinates are outside the valid range.") with tempfile.TemporaryDirectory(prefix="photo-date-editor-") as temp_dir: temp_path = Path(temp_dir) / f"working{suffix}" upload_started = time.perf_counter() await _save_upload(file, temp_path, suffix) upload_seconds = time.perf_counter() - upload_started rotation_mode = "none" rotation_seconds = 0.0 if rotation: rotation_started = time.perf_counter() rotation_mode = _rotate_pixels(temp_path, suffix, rotation) rotation_seconds = time.perf_counter() - rotation_started supports_iptc = suffix in IPTC_EXTENSIONS command = [ "exiftool", "-overwrite_original", "-m", "-n", f"-EXIF:DateTimeOriginal={exif_datetime}", f"-EXIF:CreateDate={exif_datetime}", f"-EXIF:ModifyDate={exif_datetime}", f"-XMP-exif:DateTimeOriginal={exif_datetime}", f"-XMP-xmp:CreateDate={exif_datetime}", f"-XMP-xmp:ModifyDate={exif_datetime}", f"-XMP-photoshop:DateCreated={xmp_date}", f"-XMP-photoshop:Instructions=Photo Date Editor precision: {precision_label}; time: {time_label}", "-EXIF:ImageDescription=", "-XMP-dc:Description=", "-XMP-dc:Subject=", ] if supports_iptc: command.extend(["-IPTC:Caption-Abstract=", "-IPTC:Keywords="]) if rotation: command.extend(["-EXIF:Orientation=1", "-XMP-tiff:Orientation=1"]) clean_description = description.strip() if clean_description: command.extend( [ f"-EXIF:ImageDescription={clean_description}", f"-XMP-dc:Description={clean_description}", ] ) if supports_iptc: command.append(f"-IPTC:Caption-Abstract={clean_description}") for keyword in keywords: command.append(f"-XMP-dc:Subject+={keyword}") if supports_iptc: command.append(f"-IPTC:Keywords+={keyword}") if gps_action == "clear": command.extend( [ "-EXIF:GPSLatitude=", "-EXIF:GPSLatitudeRef=", "-EXIF:GPSLongitude=", "-EXIF:GPSLongitudeRef=", "-XMP-exif:GPSLatitude=", "-XMP-exif:GPSLongitude=", "-XMP-iptcCore:Location=", ] ) elif gps_action == "set" and latitude is not None and longitude is not None: latitude_ref = "N" if latitude >= 0 else "S" longitude_ref = "E" if longitude >= 0 else "W" command.extend( [ f"-EXIF:GPSLatitude={abs(latitude):.8f}", f"-EXIF:GPSLatitudeRef={latitude_ref}", f"-EXIF:GPSLongitude={abs(longitude):.8f}", f"-EXIF:GPSLongitudeRef={longitude_ref}", f"-XMP-exif:GPSLatitude={latitude:.8f}", f"-XMP-exif:GPSLongitude={longitude:.8f}", "-XMP-iptcCore:Location=", ] ) clean_location_name = location_name.strip() if clean_location_name: command.append(f"-XMP-iptcCore:Location={clean_location_name}") command.append(str(temp_path)) logger.info("Processing %s with precision %s", filename, precision) exiftool_started = time.perf_counter() completed = subprocess.run( command, capture_output=True, text=True, timeout=180 if suffix in RAW_EXTENSIONS else 90, check=False, ) exiftool_seconds = time.perf_counter() - exiftool_started if completed.returncode != 0: logger.error("ExifTool failed: %s", completed.stderr.strip()) raise HTTPException(status_code=500, detail="ExifTool could not update this image.") response_started = time.perf_counter() updated_bytes = temp_path.read_bytes() response_read_seconds = time.perf_counter() - response_started logger.info( "Processed %s: upload %.2fs, rotation %.2fs (%s), ExifTool %.2fs, response read %.2fs", filename, upload_seconds, rotation_seconds, rotation_mode, exiftool_seconds, response_read_seconds, ) def response_chunks() -> Any: chunk_size = 256 * 1024 for offset in range(0, len(updated_bytes), chunk_size): yield updated_bytes[offset:offset + chunk_size] return StreamingResponse( response_chunks(), media_type=MEDIA_TYPES[suffix], headers={ "Content-Disposition": f'inline; filename="{Path(filename).name}"', "Content-Length": str(len(updated_bytes)), "Cache-Control": "no-store", "X-Date-Precision": precision_label, "X-Pixel-Rotation": rotation_mode, "Server-Timing": ( f'upload;dur={upload_seconds * 1000:.1f}, ' f'rotation;dur={rotation_seconds * 1000:.1f}, ' f'exiftool;dur={exiftool_seconds * 1000:.1f}, ' f'read;dur={response_read_seconds * 1000:.1f}' ), }, ) @app.get("/service-worker.js", include_in_schema=False) def service_worker() -> FileResponse: return FileResponse(STATIC_DIR / "service-worker.js", media_type="application/javascript") app.mount("/", StaticFiles(directory=STATIC_DIR, html=True), name="static")