Trixy7/plugins/tv_program/enrichment/tmdb.py

# -*- coding: utf-8 -*-
"""
TMDB (The Movie Database) API Client fuer Film- und Serien-Metadaten.

TMDB bietet eine kostenlose API-Stufe. Ein API-Key wird benoetigt
und ueber die Plugin-Konfiguration uebergeben.

Features:
    - Film-Suche mit optionalem Erscheinungsjahr
    - Serien-Suche
    - Watch-Provider fuer DE-Region (Streaming-Dienste)

API-Dokumentation: https://developer.themoviedb.org/docs

Abhaengigkeiten:
    - aiohttp: Fuer HTTP-Requests
"""

from __future__ import annotations

import asyncio
from datetime import datetime, timezone
from typing import TYPE_CHECKING

from trixy_core.utils.debug import pdebug, perror, pwarn

_AIOHTTP_AVAILABLE = False
try:
    import aiohttp
    _AIOHTTP_AVAILABLE = True
except ImportError:
    pass


class TMDBClient:
    """API-Client fuer The Movie Database (TMDB)."""

    API_BASE = "https://api.themoviedb.org/3"

    # Timeout fuer einzelne Requests
    _REQUEST_TIMEOUT: float = 15.0

    def __init__(self, api_key: str, language: str = "de-DE"):
        """
        Initialisiert den TMDB-Client.

        Args:
            api_key: TMDB API v3 Key
            language: Sprache fuer Ergebnisse (Standard: de-DE)
        """
        self._api_key = api_key
        self._language = language

    async def search_movie(self, title: str, year: int | None = None) -> dict | None:
        """
        Sucht einen Film auf TMDB.

        Args:
            title: Film-Titel
            year: Optionales Erscheinungsjahr fuer genauere Suche

        Returns:
            Enrichment-Dict mit kompakten Feldnamen oder None.
            Keine URLs oder Bilder — nur Metadaten fuer den
            Sprachassistenten.
        """
        params: dict[str, str | int] = {"query": title.strip()}
        if year:
            params["year"] = year

        data = await self._get("/search/movie", params)
        if not data:
            return None

        results = data.get("results", [])
        if not results:
            pdebug(f"[TMDB] Kein Film gefunden fuer '{title}'")
            return None

        # Bestes Ergebnis nehmen (erstes, nach Popularitaet sortiert)
        movie = results[0]
        tmdb_id = movie.get("id")

        # Genre-IDs in Namen aufloesen
        genres = await self._resolve_movie_genres(movie.get("genre_ids", []))

        # Beschreibung kuerzen
        description = (movie.get("overview") or "").strip()
        if len(description) > 500:
            description = description[:497] + "..."

        # Rating: vote_average ist 0-10
        rating = movie.get("vote_average")
        if rating is not None:
            rating = round(float(rating), 1)

        # Watch-Provider holen
        streaming: list[str] = []
        if tmdb_id:
            streaming = await self.get_watch_providers("movie", tmdb_id)

        enrichment = {
            "t": movie.get("title") or title,
            "ty": "film",
            "g": genres,
            "tags": [],
            "d": description,
            "rt": rating,
            "str": streaming,
            "sim": [],
            "src": "tmdb",
            "at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        }

        pdebug(f"[TMDB] Film-Enrichment: '{title}' -> '{enrichment['t']}' ({len(genres)} Genres)")
        return enrichment

    async def search_tv(self, title: str) -> dict | None:
        """
        Sucht eine Serie auf TMDB.

        Args:
            title: Serien-Titel

        Returns:
            Enrichment-Dict mit kompakten Feldnamen oder None.
        """
        data = await self._get("/search/tv", {"query": title.strip()})
        if not data:
            return None

        results = data.get("results", [])
        if not results:
            pdebug(f"[TMDB] Keine Serie gefunden fuer '{title}'")
            return None

        show = results[0]
        tmdb_id = show.get("id")

        # Genre-IDs in Namen aufloesen
        genres = await self._resolve_tv_genres(show.get("genre_ids", []))

        # Beschreibung kuerzen
        description = (show.get("overview") or "").strip()
        if len(description) > 500:
            description = description[:497] + "..."

        # Rating
        rating = show.get("vote_average")
        if rating is not None:
            rating = round(float(rating), 1)

        # Watch-Provider holen
        streaming: list[str] = []
        if tmdb_id:
            streaming = await self.get_watch_providers("tv", tmdb_id)

        enrichment = {
            "t": show.get("name") or title,
            "ty": "serie",
            "g": genres,
            "tags": [],
            "d": description,
            "rt": rating,
            "str": streaming,
            "sim": [],
            "src": "tmdb",
            "at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        }

        pdebug(f"[TMDB] Serien-Enrichment: '{title}' -> '{enrichment['t']}' ({len(genres)} Genres)")
        return enrichment

    async def get_watch_providers(self, media_type: str, tmdb_id: int) -> list[str]:
        """
        Holt Streaming-Provider fuer DE-Region.

        Gibt nur Provider-Namen zurueck, keine URLs — ist ein
        Sprachassistent, der die Info vorliest.

        Args:
            media_type: "movie" oder "tv"
            tmdb_id: TMDB-ID des Films/der Serie

        Returns:
            Liste von Provider-Namen, z.B. ["Netflix", "Amazon Prime Video"]
        """
        data = await self._get(f"/{media_type}/{tmdb_id}/watch/providers")
        if not data:
            return []

        # DE-Region filtern
        de_data = data.get("results", {}).get("DE")
        if not de_data:
            pdebug(f"[TMDB] Keine Watch-Provider fuer DE (ID: {tmdb_id})")
            return []

        providers: list[str] = []
        seen: set[str] = set()

        # Flatrate (Abo-Streaming) bevorzugt
        for provider in de_data.get("flatrate", []):
            name = provider.get("provider_name", "")
            if name and name not in seen:
                providers.append(name)
                seen.add(name)

        # Auch "free" Provider einschliessen (z.B. Joyn, RTL+)
        for provider in de_data.get("free", []):
            name = provider.get("provider_name", "")
            if name and name not in seen:
                providers.append(name)
                seen.add(name)

        return providers

    # --- Genre-Aufloesung ---
    # TMDB liefert bei der Suche nur Genre-IDs.
    # Statt die Genre-Liste jedes Mal abzufragen, verwenden wir einen
    # statischen Cache der gaengigsten Genres.

    _MOVIE_GENRES: dict[int, str] = {
        28: "Action", 12: "Abenteuer", 16: "Animation",
        35: "Komoedie", 80: "Krimi", 99: "Dokumentarfilm",
        18: "Drama", 10751: "Familie", 14: "Fantasy",
        36: "Historie", 27: "Horror", 10402: "Musik",
        9648: "Mystery", 10749: "Liebesfilm", 878: "Science-Fiction",
        10770: "TV-Film", 53: "Thriller", 10752: "Kriegsfilm",
        37: "Western",
    }

    _TV_GENRES: dict[int, str] = {
        10759: "Action & Abenteuer", 16: "Animation",
        35: "Komoedie", 80: "Krimi", 99: "Dokumentarfilm",
        18: "Drama", 10751: "Familie", 10762: "Kinder",
        9648: "Mystery", 10763: "News", 10764: "Reality",
        10765: "Sci-Fi & Fantasy", 10766: "Soap",
        10767: "Talk", 10768: "War & Politics", 37: "Western",
    }

    async def _resolve_movie_genres(self, genre_ids: list[int]) -> list[str]:
        """Loest Film-Genre-IDs in deutsche Genre-Namen auf."""
        return [
            self._MOVIE_GENRES[gid]
            for gid in genre_ids
            if gid in self._MOVIE_GENRES
        ]

    async def _resolve_tv_genres(self, genre_ids: list[int]) -> list[str]:
        """Loest Serien-Genre-IDs in deutsche Genre-Namen auf."""
        return [
            self._TV_GENRES[gid]
            for gid in genre_ids
            if gid in self._TV_GENRES
        ]

    # --- HTTP-Hilfsmethode ---

    async def _get(self, endpoint: str, params: dict | None = None) -> dict | None:
        """
        GET-Request mit API-Key und Fehlerbehandlung.

        Args:
            endpoint: API-Endpunkt (z.B. "/search/movie")
            params: Zusaetzliche Query-Parameter

        Returns:
            JSON-Antwort als Dict oder None bei Fehler
        """
        if not _AIOHTTP_AVAILABLE:
            perror("[TMDB] aiohttp nicht installiert — Enrichment deaktiviert")
            return None

        url = f"{self.API_BASE}{endpoint}"
        query_params: dict[str, str | int] = {
            "api_key": self._api_key,
            "language": self._language,
        }
        if params:
            query_params.update(params)

        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(
                    url,
                    params=query_params,
                    timeout=aiohttp.ClientTimeout(total=self._REQUEST_TIMEOUT),
                ) as resp:
                    if resp.status == 401:
                        perror("[TMDB] Ungueltiger API-Key (401)")
                        return None
                    if resp.status == 429:
                        retry_after = float(resp.headers.get("Retry-After", "2"))
                        pwarn(f"[TMDB] Rate-Limit (429), warte {retry_after:.1f}s")
                        await asyncio.sleep(retry_after)
                        # Einmaliger Retry
                        return await self._get(endpoint, params)
                    if resp.status != 200:
                        perror(f"[TMDB] HTTP {resp.status} fuer {endpoint}")
                        return None

                    return await resp.json()

        except asyncio.TimeoutError:
            perror(f"[TMDB] Timeout fuer {endpoint}")
            return None
        except aiohttp.ClientError as exc:
            perror(f"[TMDB] Verbindungsfehler: {exc}")
            return None
        except Exception as exc:
            perror(f"[TMDB] Unerwarteter Fehler: {exc}")
            return None