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