ViewIT/addon/tmdb.py

from __future__ import annotations

from dataclasses import dataclass
import json
import threading
from typing import Callable, Dict, List, Optional, Tuple
from urllib.parse import urlencode

try:  # pragma: no cover - optional dependency
    import requests
except ImportError:  # pragma: no cover
    requests = None


TMDB_API_BASE = "https://api.themoviedb.org/3"
TMDB_IMAGE_BASE = "https://image.tmdb.org/t/p"
MAX_CAST_MEMBERS = 30
_TMDB_THREAD_LOCAL = threading.local()


def _get_tmdb_session() -> "requests.Session | None":
    """Returns a per-thread shared requests Session.

    We use thread-local storage because ViewIt prefetches TMDB metadata using threads.
    `requests.Session` is not guaranteed to be thread-safe, but reusing a session within
    the same thread keeps connections warm.
    """

    if requests is None:
        return None
    sess = getattr(_TMDB_THREAD_LOCAL, "session", None)
    if sess is None:
        sess = requests.Session()
        setattr(_TMDB_THREAD_LOCAL, "session", sess)
    return sess


@dataclass(frozen=True)
class TmdbCastMember:
    name: str
    role: str
    thumb: str


@dataclass(frozen=True)
class TmdbShowMeta:
    tmdb_id: int
    plot: str
    poster: str
    fanart: str
    rating: float
    votes: int
    cast: List[TmdbCastMember]


def _image_url(path: str, *, size: str) -> str:
    path = (path or "").strip()
    if not path:
        return ""
    return f"{TMDB_IMAGE_BASE}/{size}{path}"


def _fetch_credits(
    *,
    kind: str,
    tmdb_id: int,
    api_key: str,
    language: str,
    timeout: int,
    log: Callable[[str], None] | None,
    log_responses: bool,
) -> List[TmdbCastMember]:
    if requests is None or not tmdb_id:
        return []
    params = {"api_key": api_key, "language": (language or "de-DE").strip()}
    url = f"{TMDB_API_BASE}/{kind}/{tmdb_id}/credits?{urlencode(params)}"
    status, payload, body_text = _tmdb_get_json(url=url, timeout=timeout, log=log, log_responses=log_responses)
    if callable(log):
        log(f"TMDB RESPONSE /{kind}/{{id}}/credits status={status}")
        if log_responses and payload is None and body_text:
            log(f"TMDB RESPONSE_BODY /{kind}/{{id}}/credits body={body_text[:2000]}")
    if status != 200 or not isinstance(payload, dict):
        return []
    cast_payload = payload.get("cast") or []
    if callable(log):
        log(f"TMDB CREDITS /{kind}/{{id}}/credits cast={len(cast_payload)}")
    return _parse_cast_payload(cast_payload)


def _parse_cast_payload(cast_payload: object) -> List[TmdbCastMember]:
    if not isinstance(cast_payload, list):
        return []
    with_images: List[TmdbCastMember] = []
    without_images: List[TmdbCastMember] = []
    for entry in cast_payload:
        if not isinstance(entry, dict):
            continue
        name = (entry.get("name") or "").strip()
        role = (entry.get("character") or "").strip()
        thumb = _image_url(entry.get("profile_path") or "", size="w185")
        if not name:
            continue
        member = TmdbCastMember(name=name, role=role, thumb=thumb)
        if thumb:
            with_images.append(member)
        else:
            without_images.append(member)
    if with_images:
        return with_images[:MAX_CAST_MEMBERS]
    return without_images[:MAX_CAST_MEMBERS]


def _tmdb_get_json(
    *,
    url: str,
    timeout: int,
    log: Callable[[str], None] | None,
    log_responses: bool,
    session: "requests.Session | None" = None,
) -> Tuple[int | None, object | None, str]:
    """Fetches TMDB JSON with optional shared session.

    Returns: (status_code, payload_or_none, body_text_or_empty)
    """

    if requests is None:
        return None, None, ""
    if callable(log):
        log(f"TMDB GET {url}")
    sess = session or _get_tmdb_session() or requests.Session()
    response = None
    try:
        response = sess.get(url, timeout=timeout)
        status = getattr(response, "status_code", None)
        payload: object | None = None
        body_text = ""
        try:
            payload = response.json()
        except Exception:
            try:
                body_text = (response.text or "").strip()
            except Exception:
                body_text = ""
    except Exception as exc:  # pragma: no cover
        if callable(log):
            log(f"TMDB ERROR request_failed url={url} error={exc!r}")
        return None, None, ""
    finally:
        if response is not None:
            try:
                response.close()
            except Exception:
                pass

    if callable(log):
        log(f"TMDB RESPONSE status={status} url={url}")
        if log_responses:
            if payload is not None:
                try:
                    dumped = json.dumps(payload, ensure_ascii=False)
                except Exception:
                    dumped = str(payload)
                log(f"TMDB RESPONSE_BODY url={url} body={dumped[:2000]}")
            elif body_text:
                log(f"TMDB RESPONSE_BODY url={url} body={body_text[:2000]}")
    return status, payload, body_text


def fetch_tv_episode_credits(
    *,
    tmdb_id: int,
    season_number: int,
    episode_number: int,
    api_key: str,
    language: str = "de-DE",
    timeout: int = 15,
    log: Callable[[str], None] | None = None,
    log_responses: bool = False,
) -> List[TmdbCastMember]:
    """Lädt Cast für eine konkrete Episode (/tv/{id}/season/{n}/episode/{e}/credits)."""
    if requests is None:
        return []
    api_key = (api_key or "").strip()
    if not api_key or not tmdb_id:
        return []
    params = {"api_key": api_key, "language": (language or "de-DE").strip()}
    url = f"{TMDB_API_BASE}/tv/{tmdb_id}/season/{season_number}/episode/{episode_number}/credits?{urlencode(params)}"
    status, payload, body_text = _tmdb_get_json(url=url, timeout=timeout, log=log, log_responses=log_responses)
    if callable(log):
        log(f"TMDB RESPONSE /tv/{{id}}/season/{{n}}/episode/{{e}}/credits status={status}")
        if log_responses and payload is None and body_text:
            log(f"TMDB RESPONSE_BODY /tv/{{id}}/season/{{n}}/episode/{{e}}/credits body={body_text[:2000]}")
    if status != 200 or not isinstance(payload, dict):
        return []
    cast_payload = payload.get("cast") or []
    if callable(log):
        log(f"TMDB CREDITS /tv/{{id}}/season/{{n}}/episode/{{e}}/credits cast={len(cast_payload)}")
    return _parse_cast_payload(cast_payload)


def lookup_tv_show(
    *,
    title: str,
    api_key: str,
    language: str = "de-DE",
    timeout: int = 15,
    log: Callable[[str], None] | None = None,
    log_responses: bool = False,
    include_cast: bool = False,
) -> Optional[TmdbShowMeta]:
    """Sucht eine TV-Show bei TMDB und liefert Plot + Poster-URL (wenn vorhanden)."""
    if requests is None:
        return None
    api_key = (api_key or "").strip()
    if not api_key:
        return None
    query = (title or "").strip()
    if not query:
        return None

    params = {
        "api_key": api_key,
        "language": (language or "de-DE").strip(),
        "query": query,
        "include_adult": "false",
        "page": "1",
    }
    url = f"{TMDB_API_BASE}/search/tv?{urlencode(params)}"
    status, payload, body_text = _tmdb_get_json(
        url=url,
        timeout=timeout,
        log=log,
        log_responses=log_responses,
    )
    results = (payload or {}).get("results") if isinstance(payload, dict) else []
    results = results or []
    if callable(log):
        log(f"TMDB RESPONSE /search/tv status={status} results={len(results)}")
        if log_responses and payload is None and body_text:
            log(f"TMDB RESPONSE_BODY /search/tv body={body_text[:2000]}")

    if status != 200:
        return None
    if not results:
        return None

    normalized_query = query.casefold()
    best = None
    for candidate in results:
        name = (candidate.get("name") or "").casefold()
        original_name = (candidate.get("original_name") or "").casefold()
        if name == normalized_query or original_name == normalized_query:
            best = candidate
            break
    if best is None:
        best = results[0]

    tmdb_id = int(best.get("id") or 0)
    plot = (best.get("overview") or "").strip()
    poster = _image_url(best.get("poster_path") or "", size="w342")
    fanart = _image_url(best.get("backdrop_path") or "", size="w780")
    try:
        rating = float(best.get("vote_average") or 0.0)
    except Exception:
        rating = 0.0
    try:
        votes = int(best.get("vote_count") or 0)
    except Exception:
        votes = 0
    if not tmdb_id:
        return None
    cast: List[TmdbCastMember] = []
    if include_cast and tmdb_id:
        detail_params = {
            "api_key": api_key,
            "language": (language or "de-DE").strip(),
            "append_to_response": "credits",
        }
        detail_url = f"{TMDB_API_BASE}/tv/{tmdb_id}?{urlencode(detail_params)}"
        d_status, d_payload, d_body = _tmdb_get_json(
            url=detail_url,
            timeout=timeout,
            log=log,
            log_responses=log_responses,
        )
        if callable(log):
            log(f"TMDB RESPONSE /tv/{{id}} status={d_status}")
            if log_responses and d_payload is None and d_body:
                log(f"TMDB RESPONSE_BODY /tv/{{id}} body={d_body[:2000]}")
        if d_status == 200 and isinstance(d_payload, dict):
            credits = d_payload.get("credits") or {}
            cast = _parse_cast_payload((credits or {}).get("cast"))
    if not plot and not poster and not fanart and not rating and not votes and not cast:
        return None
    return TmdbShowMeta(
        tmdb_id=tmdb_id,
        plot=plot,
        poster=poster,
        fanart=fanart,
        rating=rating,
        votes=votes,
        cast=cast,
    )


@dataclass(frozen=True)
class TmdbMovieMeta:
    tmdb_id: int
    plot: str
    poster: str
    fanart: str
    runtime_minutes: int
    rating: float
    votes: int
    cast: List[TmdbCastMember]


def _fetch_movie_details(
    *,
    tmdb_id: int,
    api_key: str,
    language: str,
    timeout: int,
    log: Callable[[str], None] | None,
    log_responses: bool,
    include_cast: bool,
) -> Tuple[int, List[TmdbCastMember]]:
    """Fetches /movie/{id} and (optionally) bundles credits via append_to_response=credits."""
    if requests is None or not tmdb_id:
        return 0, []
    api_key = (api_key or "").strip()
    if not api_key:
        return 0, []
    params: Dict[str, str] = {
        "api_key": api_key,
        "language": (language or "de-DE").strip(),
    }
    if include_cast:
        params["append_to_response"] = "credits"
    url = f"{TMDB_API_BASE}/movie/{tmdb_id}?{urlencode(params)}"
    status, payload, body_text = _tmdb_get_json(url=url, timeout=timeout, log=log, log_responses=log_responses)
    if callable(log):
        log(f"TMDB RESPONSE /movie/{{id}} status={status}")
        if log_responses and payload is None and body_text:
            log(f"TMDB RESPONSE_BODY /movie/{{id}} body={body_text[:2000]}")
    if status != 200 or not isinstance(payload, dict):
        return 0, []
    try:
        runtime = int(payload.get("runtime") or 0)
    except Exception:
        runtime = 0
    cast: List[TmdbCastMember] = []
    if include_cast:
        credits = payload.get("credits") or {}
        cast = _parse_cast_payload((credits or {}).get("cast"))
    return runtime, cast


def lookup_movie(
    *,
    title: str,
    api_key: str,
    language: str = "de-DE",
    timeout: int = 15,
    log: Callable[[str], None] | None = None,
    log_responses: bool = False,
    include_cast: bool = False,
) -> Optional[TmdbMovieMeta]:
    """Sucht einen Film bei TMDB und liefert Plot + Poster-URL (wenn vorhanden)."""
    if requests is None:
        return None
    api_key = (api_key or "").strip()
    if not api_key:
        return None
    query = (title or "").strip()
    if not query:
        return None

    params = {
        "api_key": api_key,
        "language": (language or "de-DE").strip(),
        "query": query,
        "include_adult": "false",
        "page": "1",
    }
    url = f"{TMDB_API_BASE}/search/movie?{urlencode(params)}"
    status, payload, body_text = _tmdb_get_json(
        url=url,
        timeout=timeout,
        log=log,
        log_responses=log_responses,
    )
    results = (payload or {}).get("results") if isinstance(payload, dict) else []
    results = results or []
    if callable(log):
        log(f"TMDB RESPONSE /search/movie status={status} results={len(results)}")
        if log_responses and payload is None and body_text:
            log(f"TMDB RESPONSE_BODY /search/movie body={body_text[:2000]}")

    if status != 200:
        return None
    if not results:
        return None

    normalized_query = query.casefold()
    best = None
    for candidate in results:
        name = (candidate.get("title") or "").casefold()
        original_name = (candidate.get("original_title") or "").casefold()
        if name == normalized_query or original_name == normalized_query:
            best = candidate
            break
    if best is None:
        best = results[0]

    tmdb_id = int(best.get("id") or 0)
    plot = (best.get("overview") or "").strip()
    poster = _image_url(best.get("poster_path") or "", size="w342")
    fanart = _image_url(best.get("backdrop_path") or "", size="w780")
    runtime_minutes = 0
    try:
        rating = float(best.get("vote_average") or 0.0)
    except Exception:
        rating = 0.0
    try:
        votes = int(best.get("vote_count") or 0)
    except Exception:
        votes = 0
    if not tmdb_id:
        return None
    cast: List[TmdbCastMember] = []
    runtime_minutes, cast = _fetch_movie_details(
        tmdb_id=tmdb_id,
        api_key=api_key,
        language=language,
        timeout=timeout,
        log=log,
        log_responses=log_responses,
        include_cast=include_cast,
    )
    if not plot and not poster and not fanart and not rating and not votes and not cast:
        return None
    return TmdbMovieMeta(
        tmdb_id=tmdb_id,
        plot=plot,
        poster=poster,
        fanart=fanart,
        runtime_minutes=runtime_minutes,
        rating=rating,
        votes=votes,
        cast=cast,
    )


@dataclass(frozen=True)
class TmdbEpisodeMeta:
    plot: str
    thumb: str
    runtime_minutes: int


@dataclass(frozen=True)
class TmdbSeasonMeta:
    plot: str
    poster: str


def lookup_tv_season_summary(
    *,
    tmdb_id: int,
    season_number: int,
    api_key: str,
    language: str = "de-DE",
    timeout: int = 15,
    log: Callable[[str], None] | None = None,
    log_responses: bool = False,
) -> Optional[TmdbSeasonMeta]:
    """Lädt Staffel-Meta (Plot + Poster)."""
    if requests is None:
        return None

    api_key = (api_key or "").strip()
    if not api_key or not tmdb_id:
        return None

    params = {"api_key": api_key, "language": (language or "de-DE").strip()}
    url = f"{TMDB_API_BASE}/tv/{tmdb_id}/season/{season_number}?{urlencode(params)}"
    status, payload, body_text = _tmdb_get_json(url=url, timeout=timeout, log=log, log_responses=log_responses)
    if callable(log):
        log(f"TMDB RESPONSE /tv/{{id}}/season/{{n}} status={status}")
        if log_responses and payload is None and body_text:
            log(f"TMDB RESPONSE_BODY /tv/{{id}}/season/{{n}} body={body_text[:2000]}")
    if status != 200 or not isinstance(payload, dict):
        return None

    plot = (payload.get("overview") or "").strip()
    poster_path = (payload.get("poster_path") or "").strip()
    poster = f"{TMDB_IMAGE_BASE}/w342{poster_path}" if poster_path else ""
    if not plot and not poster:
        return None
    return TmdbSeasonMeta(plot=plot, poster=poster)


def lookup_tv_season(
    *,
    tmdb_id: int,
    season_number: int,
    api_key: str,
    language: str = "de-DE",
    timeout: int = 15,
    log: Callable[[str], None] | None = None,
    log_responses: bool = False,
) -> Optional[Dict[int, TmdbEpisodeMeta]]:
    """Lädt Episoden-Meta für eine Staffel: episode_number -> (plot, thumb)."""
    if requests is None:
        return None
    api_key = (api_key or "").strip()
    if not api_key or not tmdb_id or season_number is None:
        return None
    params = {"api_key": api_key, "language": (language or "de-DE").strip()}
    url = f"{TMDB_API_BASE}/tv/{tmdb_id}/season/{season_number}?{urlencode(params)}"
    status, payload, body_text = _tmdb_get_json(url=url, timeout=timeout, log=log, log_responses=log_responses)
    episodes = (payload or {}).get("episodes") if isinstance(payload, dict) else []
    episodes = episodes or []
    if callable(log):
        log(f"TMDB RESPONSE /tv/{{id}}/season/{{n}} status={status} episodes={len(episodes)}")
        if log_responses:
            if payload is not None:
                try:
                    dumped = json.dumps(payload, ensure_ascii=False)
                except Exception:
                    dumped = str(payload)
                log(f"TMDB RESPONSE_BODY /tv/{{id}}/season/{{n}} body={dumped[:2000]}")
            elif body_text:
                log(f"TMDB RESPONSE_BODY /tv/{{id}}/season/{{n}} body={body_text[:2000]}")

    if status != 200 or not episodes:
        return None

    result: Dict[int, TmdbEpisodeMeta] = {}
    for entry in episodes:
        try:
            ep_number = int(entry.get("episode_number") or 0)
        except Exception:
            continue
        if not ep_number:
            continue
        plot = (entry.get("overview") or "").strip()
        runtime_minutes = 0
        try:
            runtime_minutes = int(entry.get("runtime") or 0)
        except Exception:
            runtime_minutes = 0
        still_path = (entry.get("still_path") or "").strip()
        thumb = f"{TMDB_IMAGE_BASE}/w300{still_path}" if still_path else ""
        if not plot and not thumb and not runtime_minutes:
            continue
        result[ep_number] = TmdbEpisodeMeta(plot=plot, thumb=thumb, runtime_minutes=runtime_minutes)
    return result or None