Hitonabi
518155051f
- SQLite-Cache für API-Rate-Limits (LRU, 10k Einträge) - TMDB/MusicBrainz/TheTVDB Clients - Pre-Scan-Modul für TOC-Lesung ohne Ripping - Metadaten-Preview UI - Jellyfin-Formatierung (NFO + Images) - API Endpoints für Lookup, Confirm, Format
116 lines
3.8 KiB
Python
116 lines
3.8 KiB
Python
"""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"
|
|
|
|
|
|
class TMDBClient:
|
|
def __init__(self):
|
|
self.api_key = settings.tmdb_api_key
|
|
self.session = requests.Session()
|
|
self.session.headers.update({
|
|
"Authorization": f"Bearer {self.api_key}",
|
|
"Content-Type": "application/json"
|
|
})
|
|
|
|
def _request(self, endpoint: str, params: Dict = None) -> Optional[Dict]:
|
|
"""Mache API-Request mit Caching."""
|
|
cache_key = f"tmdb:{endpoint}:{params or {}}"
|
|
|
|
cached = get(cache_key)
|
|
if cached:
|
|
return cached
|
|
|
|
try:
|
|
response = self.session.get(
|
|
f"{TMDB_BASE_URL}/{endpoint}",
|
|
params=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:*")
|