Hitonabi
ba2429612a
- Doppelte Arcane-Einträge behoben (rippy statt Rippy) - Import-Fixes: relative → absolute imports in main.py, cache/__init__.py, prescan/__init__.py, clients/__init__.py - Neue Dateien: cache.py, prescan.py, Settings.tsx - API-Port 8000 in docker-compose.yml gemappt - UI mit Sidebar, Dark Mode, Einstellungen-Tabs Fixes: #2978 (doppelte Einträge), #2888 (Import-Fehler)
110 lines
3.5 KiB
Python
110 lines
3.5 KiB
Python
"""TheTVDB API Client mit Caching."""
|
|
|
|
import requests
|
|
from typing import Dict, List, Optional
|
|
|
|
from config import settings
|
|
from cache import get, set, delete
|
|
|
|
|
|
THETVDB_BASE_URL = "https://api.thetvdb.com"
|
|
|
|
|
|
class TheTVDBClient:
|
|
def __init__(self):
|
|
self.api_key = settings.thetvdb_api_key
|
|
self.base_url = THETVDB_BASE_URL
|
|
self.session = requests.Session()
|
|
self.session.headers.update({
|
|
"Authorization": f"Bearer {self.api_key}",
|
|
"Content-Type": "application/json"
|
|
})
|
|
self._cached_token = None
|
|
|
|
def _get_token(self) -> str:
|
|
"""Hole API-Token (gecachted)."""
|
|
if self._cached_token:
|
|
return self._cached_token
|
|
|
|
try:
|
|
response = self.session.post(
|
|
f"{self.base_url}/login",
|
|
json={"apikey": self.api_key},
|
|
timeout=10
|
|
)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
if "token" in data:
|
|
self._cached_token = data["token"]
|
|
return self._cached_token
|
|
except Exception as e:
|
|
print(f"TheTVDB Login Error: {e}")
|
|
|
|
return ""
|
|
|
|
def _request(self, endpoint: str, params: Dict = None) -> Optional[Dict]:
|
|
"""Mache API-Request mit Caching."""
|
|
cache_key = f"thetvdb:{endpoint}:{params or {}}"
|
|
|
|
cached = get(cache_key)
|
|
if cached:
|
|
return cached
|
|
|
|
try:
|
|
self.session.headers.update({
|
|
"Authorization": f"Bearer {self._get_token()}"
|
|
})
|
|
|
|
response = self.session.get(
|
|
f"{self.base_url}/{endpoint}",
|
|
params=params,
|
|
timeout=10
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
set(cache_key, data)
|
|
return data
|
|
except Exception as e:
|
|
print(f"TheTVDB API Error: {e}")
|
|
return None
|
|
|
|
def search_series(self, name: str) -> List[Dict]:
|
|
"""Suche Serien."""
|
|
params = {"name": name}
|
|
result = self._request("search/series", params)
|
|
|
|
if result and "data" in result:
|
|
return result["data"]
|
|
return []
|
|
|
|
def get_series_details(self, series_id: int) -> Optional[Dict]:
|
|
"""Hole Serien-Details."""
|
|
return self._request(f"series/{series_id}")
|
|
|
|
def get_episode_details(self, series_id: int, episode_id: int) -> Optional[Dict]:
|
|
"""Hole Episode-Details."""
|
|
return self._request(f"series/{series_id}/episodes/{episode_id}")
|
|
|
|
def get_series_images(self, series_id: int) -> Dict[str, str]:
|
|
"""Hole Serien-Poster/Fanart URLs."""
|
|
params = {"type": "poster"}
|
|
result = self._request(f"series/{series_id}/images/query", params)
|
|
|
|
if result and "data" in result:
|
|
images = result["data"]
|
|
# Finde Poster und Fanart
|
|
poster = next((img for img in images if img.get("subKey") == "poster"), None)
|
|
fanart = next((img for img in images if img.get("subKey") == "fanart"), None)
|
|
|
|
return {
|
|
"poster": f"{self.base_url}/images/series/{series_id}_{poster['fileName']}" if poster else "",
|
|
"fanart": f"{self.base_url}/images/series/{series_id}_{fanart['fileName']}" if fanart else ""
|
|
}
|
|
return {}
|
|
|
|
def clear_cache(self) -> None:
|
|
"""Leere TheTVDB-Cache."""
|
|
delete("thetvdb:*")
|