Mediatracker/backend/metadata.py

"""
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