560 lines
20 KiB
Python
560 lines
20 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Annotated, Any
|
|
|
|
from fastapi import FastAPI, File, Form, HTTPException, UploadFile
|
|
from fastapi.responses import FileResponse, Response
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
APP_NAME = os.getenv("APP_NAME", "Photo Date Editor")
|
|
APP_URL = os.getenv("APP_URL", "http://localhost:8080")
|
|
APP_VERSION = "0.4.0"
|
|
MAX_UPLOAD_MB = int(os.getenv("MAX_UPLOAD_MB", "150"))
|
|
MAX_UPLOAD_BYTES = MAX_UPLOAD_MB * 1024 * 1024
|
|
LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper()
|
|
|
|
MAP_TILE_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
|
|
|
|
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"
|
|
|
|
app = FastAPI(title=APP_NAME, 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, str | int | float]:
|
|
return {
|
|
"appName": APP_NAME,
|
|
"appUrl": APP_URL,
|
|
"version": APP_VERSION,
|
|
"maxUploadMb": MAX_UPLOAD_MB,
|
|
"mapTileUrl": MAP_TILE_URL,
|
|
"mapAttribution": MAP_ATTRIBUTION,
|
|
"defaultMapLat": DEFAULT_MAP_LAT,
|
|
"defaultMapLon": DEFAULT_MAP_LON,
|
|
"defaultMapZoom": DEFAULT_MAP_ZOOM,
|
|
}
|
|
|
|
|
|
|
|
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]
|
|
|
|
|
|
async def _save_upload(upload: UploadFile, destination: Path) -> 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)
|
|
|
|
with destination.open("rb") as source:
|
|
signature = source.read(3)
|
|
if signature[:2] != b"\xff\xd8":
|
|
raise HTTPException(status_code=415, detail="The uploaded file is not a valid JPEG image.")
|
|
|
|
|
|
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
|
|
|
|
|
|
@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 {".jpg", ".jpeg"}:
|
|
raise HTTPException(status_code=415, detail="This version supports JPG and JPEG files only.")
|
|
|
|
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)
|
|
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 {".jpg", ".jpeg"}:
|
|
raise HTTPException(status_code=415, detail="This version supports JPG and JPEG files only.")
|
|
|
|
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)
|
|
|
|
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=",
|
|
"-IPTC:Caption-Abstract=",
|
|
"-XMP-dc:Subject=",
|
|
"-IPTC:Keywords=",
|
|
]
|
|
|
|
clean_description = description.strip()
|
|
if clean_description:
|
|
command.extend(
|
|
[
|
|
f"-EXIF:ImageDescription={clean_description}",
|
|
f"-XMP-dc:Description={clean_description}",
|
|
f"-IPTC:Caption-Abstract={clean_description}",
|
|
]
|
|
)
|
|
|
|
for keyword in keywords:
|
|
command.append(f"-XMP-dc:Subject+={keyword}")
|
|
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="image/jpeg",
|
|
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")
|