378 lines
15 KiB
Python
378 lines
15 KiB
Python
"""TMDB API client: rate limiting, disk cache, episode metadata."""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import threading
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from collections import deque
|
|
from pathlib import Path
|
|
from urllib.parse import urlencode
|
|
|
|
from .config import (
|
|
TMDB_API_BASE,
|
|
TMDB_LANGUAGE,
|
|
TMDB_CACHE_TTL_SEC,
|
|
TMDB_RATE_LIMIT_REQUESTS,
|
|
TMDB_RATE_LIMIT_WINDOW_SEC,
|
|
TMDB_CACHE_DIR,
|
|
TMDB_IMAGE_BASE,
|
|
TMDB_IMAGE_STILL_SIZE,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_EPISODE_PATTERN = re.compile(r"(?i)s(\d+)e(\d+)")
|
|
|
|
|
|
def parse_season_episode(stem: str) -> tuple[int, int] | None:
|
|
"""Parse S01E01-style from filename stem. Returns (season, episode) or None."""
|
|
m = _EPISODE_PATTERN.search(stem)
|
|
if m:
|
|
return int(m.group(1)), int(m.group(2))
|
|
return None
|
|
|
|
|
|
def _sanitize_key(name: str) -> str:
|
|
"""Sanitize a string for use as a cache filename."""
|
|
s = re.sub(r"[^\w\-.\s]", "", str(name))
|
|
return re.sub(r"\s+", "_", s).strip("_") or "unknown"
|
|
|
|
|
|
class TMDBClient:
|
|
"""TMDB API client with rate limiting and disk cache."""
|
|
|
|
def __init__(self):
|
|
self._series_cache: dict[str, tuple[int, str, float]] = {}
|
|
self._series_details_cache: dict[int, tuple[dict, float]] = {}
|
|
self._episode_cache: dict[tuple[int, int, int], tuple[dict, float]] = {}
|
|
self._episode_credits_cache: dict[tuple[int, int, int], tuple[dict, float]] = {}
|
|
self._request_times: deque = deque(maxlen=TMDB_RATE_LIMIT_REQUESTS + 10)
|
|
self._rate_limit_lock = threading.Lock()
|
|
self._ttl_sec = TMDB_CACHE_TTL_SEC
|
|
|
|
def _cache_path(self, subdir: str, key: str) -> Path:
|
|
"""Path to a cache file."""
|
|
TMDB_CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
|
(TMDB_CACHE_DIR / subdir).mkdir(exist_ok=True)
|
|
return TMDB_CACHE_DIR / subdir / f"{key}.json"
|
|
|
|
def _read_disk_cache(self, subdir: str, key: str) -> dict | None:
|
|
path = self._cache_path(subdir, key)
|
|
if not path.exists():
|
|
return None
|
|
try:
|
|
with open(path, encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except (OSError, json.JSONDecodeError):
|
|
return None
|
|
|
|
def _write_disk_cache(self, subdir: str, key: str, data: dict) -> None:
|
|
path = self._cache_path(subdir, key)
|
|
try:
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, ensure_ascii=False, indent=0)
|
|
except OSError:
|
|
pass
|
|
|
|
def _wait_rate_limit(self) -> None:
|
|
with self._rate_limit_lock:
|
|
while True:
|
|
now = time.monotonic()
|
|
while self._request_times and now - self._request_times[0] >= TMDB_RATE_LIMIT_WINDOW_SEC:
|
|
self._request_times.popleft()
|
|
if len(self._request_times) < TMDB_RATE_LIMIT_REQUESTS:
|
|
break
|
|
wait = TMDB_RATE_LIMIT_WINDOW_SEC - (now - self._request_times[0])
|
|
if wait > 0:
|
|
logger.debug(
|
|
"TMDB rate limit: waiting %.1fs (%s/40 in window)",
|
|
wait,
|
|
len(self._request_times),
|
|
)
|
|
time.sleep(min(0.5, wait))
|
|
|
|
def _record_request(self) -> None:
|
|
with self._rate_limit_lock:
|
|
self._request_times.append(time.monotonic())
|
|
|
|
def _request(self, path: str, params: dict | None = None) -> dict | None:
|
|
api_key = os.environ.get("TMDB_API_KEY", "").strip()
|
|
if not api_key:
|
|
return None
|
|
self._wait_rate_limit()
|
|
q = {"api_key": api_key, "language": TMDB_LANGUAGE}
|
|
if params:
|
|
q.update(params)
|
|
url = f"{TMDB_API_BASE}{path}?{urlencode(q)}"
|
|
req = urllib.request.Request(url)
|
|
logger.debug("TMDB request: %s", path)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=15) as resp:
|
|
data = json.load(resp)
|
|
self._record_request()
|
|
return data
|
|
except (
|
|
urllib.error.HTTPError,
|
|
urllib.error.URLError,
|
|
OSError,
|
|
json.JSONDecodeError,
|
|
) as e:
|
|
logger.debug("TMDB request failed %s: %s", path, e)
|
|
return None
|
|
|
|
def search_series(self, name: str) -> tuple[int, str] | None:
|
|
"""Search for a TV series by name. Returns (series_id, series_name_nl) or None."""
|
|
now = time.time()
|
|
if name in self._series_cache:
|
|
sid, sname, ts = self._series_cache[name]
|
|
if now - ts < self._ttl_sec:
|
|
logger.debug("TMDB series_search %s: memory hit -> %s", name, sid)
|
|
return (sid, sname)
|
|
key = _sanitize_key(name)
|
|
cached = self._read_disk_cache("series_search", key)
|
|
if cached is not None:
|
|
sid, sname = cached.get("series_id"), cached.get("series_name_nl")
|
|
if sid is not None and sname is not None:
|
|
self._series_cache[name] = (int(sid), sname, now)
|
|
logger.debug("TMDB series_search %s: disk hit -> %s", name, sid)
|
|
return (int(sid), sname)
|
|
logger.debug("TMDB series_search %s: API request", name)
|
|
data = self._request("/search/tv", {"query": name})
|
|
if not data:
|
|
return None
|
|
results = data.get("results") or []
|
|
if not results:
|
|
return None
|
|
first = results[0]
|
|
sid = first.get("id")
|
|
sname = (first.get("name") or name).strip() or name
|
|
if sid is not None:
|
|
self._series_cache[name] = (int(sid), sname, now)
|
|
self._write_disk_cache("series_search", key, {"series_id": sid, "series_name_nl": sname})
|
|
return (int(sid), sname)
|
|
return None
|
|
|
|
def get_series_details(self, series_id: int) -> dict | None:
|
|
"""Fetch series details in Dutch. Returns {genres, country, name} or None."""
|
|
now = time.time()
|
|
if series_id in self._series_details_cache:
|
|
details, ts = self._series_details_cache[series_id]
|
|
if now - ts < self._ttl_sec:
|
|
logger.debug("TMDB series_details %s: memory hit", series_id)
|
|
return details
|
|
key = str(series_id)
|
|
cached = self._read_disk_cache("series_details", key)
|
|
if cached is not None:
|
|
self._series_details_cache[series_id] = (cached, now)
|
|
logger.debug("TMDB series_details %s: disk hit", series_id)
|
|
return cached
|
|
logger.debug("TMDB series_details %s: API request", series_id)
|
|
data = self._request(f"/tv/{series_id}")
|
|
if not data:
|
|
return None
|
|
genres = [g.get("name") for g in (data.get("genres") or []) if g.get("name")]
|
|
countries = data.get("production_countries") or []
|
|
country = (countries[0].get("iso_3166_1") or "").upper() if countries else ""
|
|
name = (data.get("name") or "").strip() or ""
|
|
details = {"genres": genres, "country": country, "name": name}
|
|
self._series_details_cache[series_id] = (details, now)
|
|
self._write_disk_cache("series_details", key, details)
|
|
return details
|
|
|
|
def get_episode(self, series_id: int, season: int, episode: int) -> dict | None:
|
|
"""Fetch one episode in Dutch."""
|
|
now = time.time()
|
|
key_tuple = (series_id, season, episode)
|
|
if key_tuple in self._episode_cache:
|
|
meta, ts = self._episode_cache[key_tuple]
|
|
if now - ts < self._ttl_sec:
|
|
logger.debug("TMDB episode %s s%s e%s: memory hit", series_id, season, episode)
|
|
return meta
|
|
key = f"{series_id}_{season}_{episode}"
|
|
cached = self._read_disk_cache("episode", key)
|
|
if cached is not None:
|
|
self._episode_cache[key_tuple] = (cached, now)
|
|
logger.debug("TMDB episode %s s%s e%s: disk hit", series_id, season, episode)
|
|
return cached
|
|
logger.debug("TMDB episode %s s%s e%s: API request", series_id, season, episode)
|
|
data = self._request(f"/tv/{series_id}/season/{season}/episode/{episode}")
|
|
if not data:
|
|
return None
|
|
name = (data.get("name") or "").strip()
|
|
overview = (data.get("overview") or "").strip()
|
|
air_date = (data.get("air_date") or "").strip()
|
|
still_path = (data.get("still_path") or "").strip()
|
|
if still_path and not still_path.startswith("/"):
|
|
still_path = "/" + still_path
|
|
meta = {
|
|
"name": name,
|
|
"overview": overview,
|
|
"air_date": air_date,
|
|
"still_path": still_path or None,
|
|
}
|
|
self._episode_cache[key_tuple] = (meta, now)
|
|
self._write_disk_cache("episode", key, meta)
|
|
return meta
|
|
|
|
def get_episode_credits(self, series_id: int, season: int, episode: int) -> dict | None:
|
|
"""Fetch episode credits."""
|
|
now = time.time()
|
|
key_tuple = (series_id, season, episode)
|
|
if key_tuple in self._episode_credits_cache:
|
|
cred, ts = self._episode_credits_cache[key_tuple]
|
|
if now - ts < self._ttl_sec:
|
|
logger.debug("TMDB episode_credits %s s%s e%s: memory hit", series_id, season, episode)
|
|
return cred
|
|
key = f"{series_id}_{season}_{episode}"
|
|
cached = self._read_disk_cache("episode_credits", key)
|
|
if cached is not None:
|
|
self._episode_credits_cache[key_tuple] = (cached, now)
|
|
logger.debug("TMDB episode_credits %s s%s e%s: disk hit", series_id, season, episode)
|
|
return cached
|
|
logger.debug("TMDB episode_credits %s s%s e%s: API request", series_id, season, episode)
|
|
data = self._request(f"/tv/{series_id}/season/{season}/episode/{episode}/credits")
|
|
if not data:
|
|
return None
|
|
directors = []
|
|
producers = []
|
|
for c in data.get("crew") or []:
|
|
job = (c.get("job") or "").strip()
|
|
name = (c.get("name") or "").strip()
|
|
if not name:
|
|
continue
|
|
if job == "Director":
|
|
directors.append(name)
|
|
elif job in (
|
|
"Producer",
|
|
"Executive Producer",
|
|
"Co-Executive Producer",
|
|
"Supervising Producer",
|
|
"Co-Producer",
|
|
):
|
|
producers.append(name)
|
|
actors = [
|
|
(c.get("name") or "").strip()
|
|
for c in (data.get("cast") or [])
|
|
if (c.get("name") or "").strip()
|
|
]
|
|
directors = list(dict.fromkeys(directors))
|
|
producers = list(dict.fromkeys(producers))
|
|
cred = {"directors": directors, "actors": actors, "producers": producers}
|
|
self._episode_credits_cache[key_tuple] = (cred, now)
|
|
self._write_disk_cache("episode_credits", key, cred)
|
|
return cred
|
|
|
|
def has_episode_metadata_cached(
|
|
self,
|
|
programme_name: str,
|
|
path: Path,
|
|
*,
|
|
tmdb_search: str | None = None,
|
|
tmdb_id: int | None = None,
|
|
series_root: Path,
|
|
) -> bool:
|
|
"""
|
|
Check if episode metadata is already on disk (no API calls).
|
|
Used by prefetcher to skip episodes that don't need fetching.
|
|
"""
|
|
if series_root not in path.parents:
|
|
return True
|
|
se = parse_season_episode(path.stem)
|
|
if not se:
|
|
return True
|
|
_, episode_num = se
|
|
|
|
if tmdb_id is not None:
|
|
series_id = tmdb_id
|
|
else:
|
|
search_term = (tmdb_search or programme_name).strip() or programme_name
|
|
key = _sanitize_key(search_term)
|
|
cached = self._read_disk_cache("series_search", key)
|
|
if cached is None:
|
|
return False
|
|
sid = cached.get("series_id")
|
|
if sid is None:
|
|
return False
|
|
series_id = int(sid)
|
|
|
|
ep_key = f"{series_id}_{se[0]}_{episode_num}"
|
|
ep_path = self._cache_path("episode", ep_key)
|
|
return ep_path.exists()
|
|
|
|
def get_episode_metadata(
|
|
self,
|
|
programme_name: str,
|
|
path: Path,
|
|
*,
|
|
display_name: str | None = None,
|
|
tmdb_search: str | None = None,
|
|
tmdb_id: int | None = None,
|
|
series_root: Path,
|
|
) -> dict | None:
|
|
"""
|
|
Return TMDB metadata for this series episode in Dutch.
|
|
Only for series under series_root; parses S01E01 from path.stem.
|
|
"""
|
|
if series_root not in path.parents:
|
|
return None
|
|
if not os.environ.get("TMDB_API_KEY", "").strip():
|
|
return None
|
|
se = parse_season_episode(path.stem)
|
|
if not se:
|
|
return None
|
|
season_num, episode_num = se
|
|
|
|
if tmdb_id is not None:
|
|
series_id = tmdb_id
|
|
details = self.get_series_details(series_id)
|
|
series_name_nl = (details.get("name") or programme_name).strip() or programme_name
|
|
else:
|
|
search_term = (tmdb_search or programme_name).strip() or programme_name
|
|
hit = self.search_series(search_term)
|
|
if not hit:
|
|
return None
|
|
series_id, series_name_nl = hit
|
|
|
|
ep = self.get_episode(series_id, season_num, episode_num)
|
|
if not ep:
|
|
return None
|
|
|
|
sub_title = ep.get("name") or path.stem
|
|
overview = (ep.get("overview") or "").strip() or None
|
|
air_date = ep.get("air_date") or ""
|
|
date_str = (air_date.replace("-", "")[:8]) if air_date and len(air_date) >= 10 else "20090101"
|
|
still_path = ep.get("still_path") or ""
|
|
icon_url = f"{TMDB_IMAGE_BASE}/{TMDB_IMAGE_STILL_SIZE}{still_path}" if still_path and still_path.startswith("/") else None
|
|
|
|
details = self.get_series_details(series_id)
|
|
genres = (details.get("genres") or []) if details else []
|
|
country = (details.get("country") or "") if details else ""
|
|
|
|
credits = self.get_episode_credits(series_id, season_num, episode_num)
|
|
directors = (credits.get("directors") or []) if credits else []
|
|
actors = (credits.get("actors") or []) if credits else []
|
|
producers = (credits.get("producers") or []) if credits else []
|
|
|
|
xmltv_ns = f"{season_num}.{episode_num}.0/1"
|
|
onscreen = f"S{season_num:02d}E{episode_num:02d}"
|
|
epg_title = (display_name or series_name_nl).strip() or series_name_nl
|
|
|
|
return {
|
|
"title": epg_title,
|
|
"sub_title": sub_title,
|
|
"desc": overview,
|
|
"lang": TMDB_LANGUAGE,
|
|
"icon": icon_url,
|
|
"directors": directors,
|
|
"actors": actors,
|
|
"producers": producers,
|
|
"date": date_str,
|
|
"categories": genres,
|
|
"country": country,
|
|
"episode_nums": {"xmltv_ns": xmltv_ns, "onscreen": onscreen},
|
|
}
|