refactor(scripts): extract shared sae_* helpers and dedupe gallery builders

Consolidate copy-pasted logic across the gallery/run scripts into shared
modules:
  - sae_env.py     — zero-dependency .env loader (populates os.environ)
  - sae_tmdb.py    — TMDB API helpers (tmdb_get, person images, id lookups)
  - sae_jellyfin.py— Jellyfin API helpers (jf_get, id/URL normalisation)
  - sae_gallery.py — image download + gallery.json writing

make_gallery, make_jellyfin_gallery and filter_gallery now import these
instead of carrying their own near-identical copies.
This commit is contained in:
2026-07-04 18:55:37 +02:00
parent aacaefb3dc
commit 152c34b1f4
7 changed files with 431 additions and 258 deletions
+105
View File
@@ -0,0 +1,105 @@
"""Shared image-download and gallery-writing helpers.
Consolidates the three near-identical download loops (make_gallery.download_images,
make_jellyfin_gallery.download_urls + download_person_images) and the duplicated
"write gallery.json + .missing_images.json" tail from both builders.
"""
import io
import json
import sys
from pathlib import Path
import requests
from PIL import Image, UnidentifiedImageError
# Errors that mean "this one image couldn't be fetched/decoded" rather than a bug:
# network/HTTP failures, filesystem errors, and undecodable image bytes. We catch
# exactly these around a single download so a coding error isn't swallowed as a
# spurious "download failed".
DOWNLOAD_ERRORS = (requests.RequestException, OSError, UnidentifiedImageError)
# Wikidata SPARQL: map an IMDB person id (P345) to their image (P18) on
# Wikimedia Commons. Used as a free, CC-licensed headshot fallback when a
# provider (TMDB/Jellyfin) has no usable image. Wikidata requires a
# descriptive User-Agent.
WIKIDATA_SPARQL = "https://query.wikidata.org/sparql"
WIKIDATA_HEADERS = {
"Accept": "application/sparql-results+json",
"User-Agent": "scene-actor-extraction/1.0 (https://github.com/; gallery builder)",
}
def wikidata_image_urls(imdb_person_id: str) -> list[str]:
"""Return Commons image URL(s) for a person via their IMDB id (P345 -> P18)."""
if not imdb_person_id:
return []
query = (
"SELECT ?image WHERE { "
f'?person wdt:P345 "{imdb_person_id}" . '
"?person wdt:P18 ?image . "
"}"
)
try:
r = requests.get(WIKIDATA_SPARQL, params={"query": query, "format": "json"},
headers=WIKIDATA_HEADERS, timeout=15)
r.raise_for_status()
bindings = r.json().get("results", {}).get("bindings", [])
return [b["image"]["value"] for b in bindings if "image" in b]
except (requests.RequestException, ValueError) as e:
print(f" [warn] wikidata lookup failed for {imdb_person_id}: {e}", file=sys.stderr)
return []
def download_image(url: str, out_path: Path, *,
headers: dict | None = None,
params: dict | None = None) -> Path | None:
"""Download one image to out_path as JPEG, returning the path or None on failure.
Skips the download if out_path already exists and is larger than 1 KiB
(a previously cached, non-truncated image).
"""
if out_path.exists() and out_path.stat().st_size > 1024:
return out_path
try:
r = requests.get(url, headers=headers, params=params, timeout=15)
r.raise_for_status()
ctype = r.headers.get("Content-Type", "")
if ctype and not ctype.startswith("image/"):
print(f" [warn] unexpected response for {url}: "
f"status={r.status_code} content-type={ctype!r} len={len(r.content)}",
file=sys.stderr)
return None
out_path.parent.mkdir(parents=True, exist_ok=True)
Image.open(io.BytesIO(r.content)).convert("RGB").save(out_path, "JPEG")
return out_path
except DOWNLOAD_ERRORS as e:
print(f" [warn] image download failed: {url}: {e}", file=sys.stderr)
return None
def download_images(urls: list[str], dest_dir: Path, n: int,
start_index: int = 0, *,
headers: dict | None = None,
params: dict | None = None) -> list[Path]:
"""Download up to n images from urls into dest_dir, numbered from start_index."""
dest_dir.mkdir(parents=True, exist_ok=True)
paths = []
for i, url in enumerate(urls[:n]):
out = dest_dir / f"{start_index + i:02d}.jpg"
path = download_image(url, out, headers=headers, params=params)
if path is not None:
paths.append(path)
return paths
def save_gallery(gallery: dict, missing: list[dict], output: Path) -> None:
"""Write gallery.json and, if any actors lack images, a .missing_images.json sidecar."""
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(json.dumps(gallery, indent=2) + "\n")
print(f"Saved: {output}", file=sys.stderr)
if missing:
missing_path = output.with_name(output.stem + ".missing_images.json")
missing_path.write_text(json.dumps(missing, indent=2) + "\n")
print(f"{len(missing)} actor(s) need images — see {missing_path}", file=sys.stderr)