""" metadata.py — Poster + metadata fetching. - Poster: IMDB public suggestion API (no key required) - Alt titles (FR/EN): Wikidata SPARQL (free, no registration) - Local poster detection (poster.jpg / folder.jpg alongside video) """ import json import logging import urllib.error import urllib.parse import urllib.request from dataclasses import dataclass, field from pathlib import Path from typing import Optional logger = logging.getLogger(__name__) _REQUEST_TIMEOUT = 4 _WIKIDATA_TIMEOUT = 8 _LOCAL_POSTER_NAMES = ["poster.jpg", "poster.png", "folder.jpg", "folder.png", "cover.jpg", "cover.png", "movie.jpg"] _ALT_TITLE_LANGS = ("fr", "en", "de", "es") @dataclass class MediaMetadata: poster_url: Optional[str] = None imdb_id: Optional[str] = None # ─── IMDB suggestion API ───────────────────────────────────────────────────── def _imdb_suggest(query: str) -> list[dict]: query_clean = query.lower().strip() if not query_clean: return [] first_char = query_clean[0] if query_clean[0].isalpha() else "a" encoded = urllib.parse.quote(query_clean) url = f"https://sg.media-imdb.com/suggests/{first_char}/{encoded}.json" try: req = urllib.request.Request(url, headers={ "User-Agent": "Mozilla/5.0", "Accept": "application/json, text/javascript", }) with urllib.request.urlopen(req, timeout=_REQUEST_TIMEOUT) as resp: raw = resp.read().decode("utf-8") start = raw.index('(') end = raw.rindex(')') data = json.loads(raw[start + 1:end]) return data.get("d", []) except (ValueError, KeyError): pass except urllib.error.HTTPError as exc: if exc.code != 404: logger.debug("IMDB suggest HTTP %d for %r", exc.code, query) except Exception as exc: logger.debug("IMDB suggest error for %r: %s", query, exc) return [] def _extract_poster(item: dict) -> Optional[str]: img = item.get("i") if isinstance(img, list) and img: return str(img[0]) if isinstance(img, dict): return img.get("imageUrl") return None def _best_match(items: list[dict], title: str, year: Optional[int], qid_filter: str) -> Optional[dict]: """Return the best matching IMDB item dict, or None.""" title_lower = title.lower() candidates = [it for it in items if it.get("qid") == qid_filter] if not candidates: return None if year: for it in candidates: if it.get("l", "").lower() == title_lower and it.get("y") == year: return it for it in candidates: if it.get("l", "").lower() == title_lower: return it # First candidate that has a poster for it in candidates: if _extract_poster(it): return it return candidates[0] if candidates else None def _get_metadata(title: str, year: Optional[int], qid_filter: str) -> MediaMetadata: items = _imdb_suggest(title) item = _best_match(items, title, year, qid_filter) if not item: return MediaMetadata() return MediaMetadata( poster_url=_extract_poster(item), imdb_id=item.get("id"), ) def get_movie_metadata(title: str, year: Optional[int]) -> MediaMetadata: return _get_metadata(title, year, "movie") def get_series_metadata(title: str, year: Optional[int]) -> MediaMetadata: return _get_metadata(title, year, "tvSeries") # ─── Wikidata alt titles ───────────────────────────────────────────────────── def get_alt_titles_batch(imdb_ids: list[str], retries: int = 3) -> dict[str, list[tuple[str, str]]]: """ Fetch FR/EN/DE/ES titles from Wikidata for a batch of IMDB IDs. Returns {imdb_id: [(lang, title), ...]} Retries up to `retries` times on 429/503 with exponential backoff. """ import time as _time if not imdb_ids: return {} values = " ".join(f'"{iid}"' for iid in imdb_ids) langs = " ".join(f'"{l}"' for l in _ALT_TITLE_LANGS) query = f""" SELECT ?imdbId ?label WHERE {{ VALUES ?imdbId {{ {values} }} ?item wdt:P345 ?imdbId . ?item rdfs:label ?label . FILTER(LANG(?label) IN ({langs})) }} """ url = "https://query.wikidata.org/sparql?" + urllib.parse.urlencode({ "query": query, "format": "json", }) for attempt in range(retries): try: req = urllib.request.Request(url, headers={ "User-Agent": "MediaTracker/1.0", "Accept": "application/sparql-results+json", }) with urllib.request.urlopen(req, timeout=_WIKIDATA_TIMEOUT) as resp: data = json.loads(resp.read().decode("utf-8")) result: dict[str, list[tuple[str, str]]] = {} for row in data.get("results", {}).get("bindings", []): iid = row["imdbId"]["value"] label = row["label"]["value"] lang = row["label"].get("xml:lang", "") result.setdefault(iid, []).append((lang, label)) return result except urllib.error.HTTPError as exc: if exc.code in (429, 503) and attempt < retries - 1: wait = 5 * (2 ** attempt) logger.info("Wikidata rate-limited — retrying in %ds (attempt %d/%d)", wait, attempt + 1, retries) _time.sleep(wait) else: logger.warning("Wikidata batch failed (%d ids): HTTP %d", len(imdb_ids), exc.code) return {} except Exception as exc: logger.warning("Wikidata batch failed (%d ids): %s", len(imdb_ids), exc) return {} return {} # ─── Local poster detection ─────────────────────────────────────────────────── def find_local_poster(video_path: Path) -> Optional[Path]: directory = video_path.parent for name in _LOCAL_POSTER_NAMES: candidate = directory / name if candidate.is_file(): return candidate return None