rippy/docker/api/clients/tmdb.py

"""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:*")