1021 lines
36 KiB
Python
1021 lines
36 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import io
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
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, UploadFile
|
|
from pydantic import BaseModel, Field
|
|
from fastapi.responses import FileResponse, Response
|
|
from fastapi.staticfiles import StaticFiles
|
|
from PIL import Image, ImageOps, UnidentifiedImageError
|
|
|
|
FORMAT_GROUPS: dict[str, tuple[str, ...]] = {
|
|
"jpg": (".jpg", ".jpeg"),
|
|
"png": (".png",),
|
|
"tiff": (".tif", ".tiff"),
|
|
"heif": (".heic", ".heif", ".hif"),
|
|
"dng": (".dng",),
|
|
"canon": (".cr2", ".cr3"),
|
|
}
|
|
FORMAT_LABELS = {
|
|
"jpg": "JPG",
|
|
"png": "PNG",
|
|
"tiff": "TIFF",
|
|
"heif": "HEIF",
|
|
"dng": "DNG",
|
|
"canon": "Canon 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.8.0"
|
|
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"))
|
|
PREVIEW_MAX_PIXELS = int(os.getenv("PREVIEW_MAX_PIXELS", "300000000"))
|
|
Image.MAX_IMAGE_PIXELS = PREVIEW_MAX_PIXELS
|
|
|
|
TIFF_EXTENSIONS = {".tif", ".tiff"}
|
|
HEIF_EXTENSIONS = {".heic", ".heif", ".hif"}
|
|
DNG_EXTENSIONS = {".dng"}
|
|
CANON_EXTENSIONS = {".cr2", ".cr3"}
|
|
IPTC_EXTENSIONS = {".jpg", ".jpeg", ".tif", ".tiff", ".dng", ".cr2"}
|
|
SERVER_PREVIEW_EXTENSIONS = (
|
|
TIFF_EXTENSIONS | HEIF_EXTENSIONS | DNG_EXTENSIONS | CANON_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",
|
|
}
|
|
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",
|
|
'© <a href="https://www.openstreetmap.org/copyright">OpenStreetMap contributors</a>',
|
|
)
|
|
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.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"
|
|
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)
|
|
)
|
|
)
|
|
|
|
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 _embedded_raw_preview(path: Path, format_name: str) -> Image.Image | None:
|
|
"""Extract the largest usable JPEG from the common RAW preview tags."""
|
|
best: Image.Image | None = None
|
|
best_area = 0
|
|
for tag in ("JpgFromRaw", "PreviewImage", "ThumbnailImage"):
|
|
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()
|
|
candidate = source.copy()
|
|
area = candidate.width * candidate.height
|
|
if area > best_area:
|
|
if best is not None:
|
|
best.close()
|
|
best = candidate
|
|
best_area = area
|
|
else:
|
|
candidate.close()
|
|
except (UnidentifiedImageError, OSError, ValueError):
|
|
logger.debug("%s %s data was not a usable image preview.", format_name, tag)
|
|
return best
|
|
|
|
|
|
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)
|
|
|
|
|
|
@app.post("/api/preview")
|
|
async def create_preview(file: Annotated[UploadFile, File(...)]) -> Response:
|
|
"""Generate a browser-friendly preview for TIFF, HEIF, DNG, CR2, and CR3."""
|
|
filename = file.filename or "photo.tiff"
|
|
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 suffix not in SERVER_PREVIEW_EXTENSIONS:
|
|
raise HTTPException(
|
|
status_code=415,
|
|
detail="Server-generated previews are only required for TIFF, HEIF, DNG, CR2, and CR3 files.",
|
|
)
|
|
|
|
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)
|
|
try:
|
|
if suffix in DNG_EXTENSIONS | CANON_EXTENSIONS:
|
|
format_name = _format_name(suffix)
|
|
preview = _embedded_raw_preview(temp_path, format_name)
|
|
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:
|
|
preview.thumbnail((PREVIEW_MAX_EDGE, PREVIEW_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=88, optimize=True)
|
|
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
|
|
|
|
return Response(
|
|
content=output.getvalue(),
|
|
media_type="image/jpeg",
|
|
headers={"Cache-Control": "no-store"},
|
|
)
|
|
|
|
|
|
@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)
|
|
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/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()] = "",
|
|
) -> 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}.",
|
|
)
|
|
|
|
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}"
|
|
await _save_upload(file, temp_path, suffix)
|
|
|
|
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="])
|
|
|
|
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)
|
|
completed = subprocess.run(
|
|
command,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=90,
|
|
check=False,
|
|
)
|
|
if completed.returncode != 0:
|
|
logger.error("ExifTool failed: %s", completed.stderr.strip())
|
|
raise HTTPException(status_code=500, detail="ExifTool could not update this image.")
|
|
|
|
updated_bytes = temp_path.read_bytes()
|
|
|
|
return Response(
|
|
content=updated_bytes,
|
|
media_type=MEDIA_TYPES[suffix],
|
|
headers={
|
|
"Content-Disposition": f'inline; filename="{Path(filename).name}"',
|
|
"Cache-Control": "no-store",
|
|
"X-Date-Precision": precision_label,
|
|
},
|
|
)
|
|
|
|
|
|
@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")
|