# -*- coding: utf-8 -*- """ TVDB v4 API Client (Fallback fuer Serien-Metadaten). TVDB braucht einen kostenlosen API-Key fuer Bearer-Token-Auth. Wird nur als Fallback verwendet, wenn TMDB kein Ergebnis liefert. API-Dokumentation: https://thetvdb.github.io/v4-api/ Abhaengigkeiten: - aiohttp: Fuer HTTP-Requests """ from __future__ import annotations import asyncio import time 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 TVDBClient: """API-Client fuer TheTVDB v4 (Serien-Fallback).""" API_BASE = "https://api4.thetvdb.com/v4" # Token-Lebensdauer: 24 Stunden laut TVDB-Doku, wir erneuern nach 23h _TOKEN_LIFETIME: float = 23 * 3600 _REQUEST_TIMEOUT: float = 15.0 def __init__(self, api_key: str): """ Initialisiert den TVDB-Client. Args: api_key: TVDB v4 API-Key """ self._api_key = api_key self._token: str = "" self._token_expires: float = 0 self._login_failed: bool = False # Nach 401 nicht endlos wiederholen @property def _token_valid(self) -> bool: """Prueft ob der aktuelle Token noch gueltig ist.""" return bool(self._token) and time.monotonic() < self._token_expires async def authenticate(self) -> bool: """ Holt einen Bearer-Token via POST /login. Returns: True bei erfolgreicher Authentifizierung """ if not _AIOHTTP_AVAILABLE: perror("[TVDB] aiohttp nicht installiert — Enrichment deaktiviert") return False try: async with aiohttp.ClientSession() as session: async with session.post( f"{self.API_BASE}/login", json={"apikey": self._api_key}, headers={"Content-Type": "application/json"}, timeout=aiohttp.ClientTimeout(total=self._REQUEST_TIMEOUT), ) as resp: if resp.status != 200: perror(f"[TVDB] Login fehlgeschlagen: HTTP {resp.status} — TVDB deaktiviert fuer diese Session") self._login_failed = True return False data = await resp.json() token = data.get("data", {}).get("token", "") if not token: perror("[TVDB] Kein Token in Login-Antwort") return False self._token = token self._token_expires = time.monotonic() + self._TOKEN_LIFETIME pdebug("[TVDB] Authentifizierung erfolgreich") return True except asyncio.TimeoutError: perror("[TVDB] Timeout bei Login") return False except aiohttp.ClientError as exc: perror(f"[TVDB] Verbindungsfehler bei Login: {exc}") return False except Exception as exc: perror(f"[TVDB] Unerwarteter Fehler bei Login: {exc}") return False async def _ensure_authenticated(self) -> bool: """Stellt sicher, dass ein gueltiger Token vorhanden ist.""" if self._token_valid: return True if self._login_failed: return False # Nicht endlos wiederholen nach 401 return await self.authenticate() async def search(self, title: str) -> dict | None: """ Sucht eine Serie auf TVDB. Fuehrt zuerst eine Suche durch, dann holt Details fuer das beste Ergebnis. Args: title: Serien-Titel Returns: Enrichment-Dict mit kompakten Feldnamen oder None """ if not title or not title.strip(): return None if not await self._ensure_authenticated(): return None # Suche ausfuehren search_data = await self._get(f"/search?query={title.strip()}&type=series") if not search_data: return None results = search_data.get("data", []) if not results: pdebug(f"[TVDB] Kein Ergebnis fuer '{title}'") return None # Bestes Ergebnis nehmen best = results[0] tvdb_id = best.get("tvdb_id") or best.get("id") # Details holen fuer Genres etc. genres: list[str] = [] description = "" if tvdb_id: details = await self._get(f"/series/{tvdb_id}/extended") if details: series_data = details.get("data", {}) # Genres extrahieren for genre_entry in series_data.get("genres", []): genre_name = genre_entry.get("name", "") if genre_name: genres.append(genre_name) # Beschreibung (deutsch bevorzugt, Fallback englisch) description = series_data.get("overview") or "" # Fallback-Beschreibung aus Suchergebnis if not description: description = best.get("overview") or best.get("overview_deu") or "" if len(description) > 500: description = description[:497] + "..." enrichment = { "t": best.get("name") or best.get("translated_name") or title, "ty": "serie", "g": genres, "tags": [], "d": description, "rt": None, "str": [], "sim": [], "src": "tvdb", "at": datetime.now(timezone.utc).isoformat(timespec="seconds"), } pdebug(f"[TVDB] Enrichment: '{title}' -> '{enrichment['t']}' ({len(genres)} Genres)") return enrichment async def _get(self, endpoint: str) -> dict | None: """ GET-Request mit Bearer-Token-Auth und Fehlerbehandlung. Args: endpoint: API-Endpunkt (z.B. "/search?query=...") Returns: JSON-Antwort als Dict oder None bei Fehler """ if not _AIOHTTP_AVAILABLE: perror("[TVDB] aiohttp nicht installiert") return None if not self._token: perror("[TVDB] Nicht authentifiziert — bitte zuerst authenticate() aufrufen") return None url = f"{self.API_BASE}{endpoint}" headers = { "Authorization": f"Bearer {self._token}", "Accept": "application/json", } try: async with aiohttp.ClientSession() as session: async with session.get( url, headers=headers, timeout=aiohttp.ClientTimeout(total=self._REQUEST_TIMEOUT), ) as resp: if resp.status == 401: # Token abgelaufen — einmal erneuern und Retry pwarn("[TVDB] Token abgelaufen, erneuere...") self._token = "" self._token_expires = 0 if await self.authenticate(): return await self._get(endpoint) return None if resp.status == 429: retry_after = float(resp.headers.get("Retry-After", "2")) pwarn(f"[TVDB] Rate-Limit (429), warte {retry_after:.1f}s") await asyncio.sleep(retry_after) return await self._get(endpoint) if resp.status != 200: perror(f"[TVDB] HTTP {resp.status} fuer {endpoint}") return None return await resp.json() except asyncio.TimeoutError: perror(f"[TVDB] Timeout fuer {endpoint}") return None except aiohttp.ClientError as exc: perror(f"[TVDB] Verbindungsfehler: {exc}") return None except Exception as exc: perror(f"[TVDB] Unerwarteter Fehler: {exc}") return None