# -*- 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