"""TMDB API Client mit Caching.""" import requests from typing import Dict, List, Optional from config import settings from cache import get, set, delete TMDB_BASE_URL = "https://api.themoviedb.org/3" TMDB_IMAGE_BASE_URL = "https://image.tmdb.org/t/p" def ist_v4_token(key: str) -> bool: """Pure Funktion (testbar): TMDB hat ZWEI Key-Arten — der v4 Read Access Token ist ein langes JWT ('eyJ…', Bearer-Header), der klassische v3-Key 32 Hex-Zeichen (api_key-Query-Parameter). Befund 24.07.: der Client konnte NUR v4 — mit dem üblichen v3-Key aus dem TMDB-Konto war jede Anfrage 401 und die Suche lieferte still nur OMDb-Treffer.""" return bool(key) and key.startswith("eyJ") class TMDBClient: def __init__(self): # DB-Einstellung (Settings-UI/Wizard) gewinnt gegen die Env-Variable — # vorher war das Settings-Feld reine Dekoration (Fix 23.07.). from db import get_settings self.api_key = get_settings().get("tmdbApiKey") or settings.tmdb_api_key self.session = requests.Session() self.session.headers.update({"Content-Type": "application/json"}) # Beide Key-Arten unterstützen (developer.themoviedb.org: v3 als # api_key-Parameter, v4-Token als Bearer-Header) self._key_params = {} if ist_v4_token(self.api_key): self.session.headers.update({"Authorization": f"Bearer {self.api_key}"}) elif self.api_key: self._key_params = {"api_key": self.api_key} def _request(self, endpoint: str, params: Dict = None) -> Optional[Dict]: """Mache API-Request mit Caching.""" from cache.keys import generate_tmdb_key # Cache-Key OHNE den api_key — der gehört nicht in den Cache cache_key = generate_tmdb_key(endpoint, params) cached = get(cache_key) if cached: return cached try: response = self.session.get( f"{TMDB_BASE_URL}/{endpoint}", params={**(params or {}), **self._key_params}, timeout=10 ) response.raise_for_status() data = response.json() # Cache nur für erfolgreiche Requests if response.status_code == 200: set(cache_key, data) return data except Exception as e: print(f"TMDB API Error: {e}") return None def search_movie(self, title: str, year: int = None) -> List[Dict]: """Suche Filme.""" params = {"query": title, "language": "de-DE", "include_adult": False} if year: params["year"] = year result = self._request("search/movie", params) if result and "results" in result: return result["results"] return [] def search_tv(self, title: str, year: int = None) -> List[Dict]: """Suche Serien.""" params = {"query": title, "language": "de-DE", "include_adult": False} if year: params["first_air_date_year"] = year result = self._request("search/tv", params) if result and "results" in result: return result["results"] return [] def get_movie_details(self, movie_id: int) -> Optional[Dict]: """Hole Film-Details.""" params = {"language": "de-DE"} return self._request(f"movie/{movie_id}", params) def get_tv_details(self, tv_id: int) -> Optional[Dict]: """Hole Serien-Details.""" params = {"language": "de-DE"} return self._request(f"tv/{tv_id}", params) def get_tv_season(self, tv_id: int, season: int) -> Optional[Dict]: """Staffel-Details inkl. Episoden-Laufzeiten (Basis fürs Episoden-Matching; developer.themoviedb.org/reference/tv-season-details).""" return self._request(f"tv/{tv_id}/season/{season}", {"language": "de-DE"}) def find_by_imdb(self, imdb_id: str) -> Optional[Dict]: """IMDb-ID → deutscher TMDB-Eintrag (/find, external_source=imdb_id; developer.themoviedb.org/reference/find-by-id). Zweck: OMDb liefert nur englische Texte — über die IMDb-ID holt sich Rippy Titel/Beschreibung/Poster auf Deutsch nach (Commander-Wunsch). """ result = self._request( f"find/{imdb_id}", {"external_source": "imdb_id", "language": "de-DE"}, ) if not result: return None for schluessel, typ, titel_feld, datum_feld in ( ("movie_results", "movie", "title", "release_date"), ("tv_results", "tv", "name", "first_air_date"), ): eintraege = result.get(schluessel) or [] if eintraege: e = eintraege[0] datum = e.get(datum_feld) or "" return { "type": typ, "tmdb_id": e.get("id"), "title": e.get(titel_feld) or "", "overview": e.get("overview") or "", "poster_path": e.get("poster_path") or "", "year": int(datum[:4]) if datum[:4].isdigit() else None, } return None def get_movie_images(self, movie_id: int) -> Dict[str, str]: """Hole Poster/Fanart URLs.""" params = {"language": "de-DE", "include_image_language": "de,en,null"} result = self._request(f"movie/{movie_id}/images", params) if not result or "posters" not in result: return {} posters = result["posters"] fanart = result.get("backdrops", []) return { "poster": f"{TMDB_IMAGE_BASE_URL}/w500{posters[0]['file_path']}" if posters else "", "fanart": f"{TMDB_IMAGE_BASE_URL}/w1920{fanart[0]['file_path']}" if fanart else "" } def get_tv_images(self, tv_id: int) -> Dict[str, str]: """Hole Serien-Poster/Fanart URLs.""" params = {"language": "de-DE", "include_image_language": "de,en,null"} result = self._request(f"tv/{tv_id}/images", params) if not result or "posters" not in result: return {} posters = result["posters"] fanart = result.get("backdrops", []) return { "poster": f"{TMDB_IMAGE_BASE_URL}/w500{posters[0]['file_path']}" if posters else "", "fanart": f"{TMDB_IMAGE_BASE_URL}/w1920{fanart[0]['file_path']}" if fanart else "" } def clear_cache(self) -> None: """Leere TMDB-Cache.""" delete("tmdb:*")