560 lines
18 KiB
Python
560 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import json
|
|
import threading
|
|
from typing import Callable, Dict, List, Optional, Tuple
|
|
from urllib.parse import urlencode
|
|
|
|
try: # pragma: no cover - optional dependency
|
|
import requests
|
|
except ImportError: # pragma: no cover
|
|
requests = None
|
|
|
|
|
|
TMDB_API_BASE = "https://api.themoviedb.org/3"
|
|
TMDB_IMAGE_BASE = "https://image.tmdb.org/t/p"
|
|
MAX_CAST_MEMBERS = 30
|
|
_TMDB_THREAD_LOCAL = threading.local()
|
|
|
|
|
|
def _get_tmdb_session() -> "requests.Session | None":
|
|
"""Returns a per-thread shared requests Session.
|
|
|
|
We use thread-local storage because ViewIt prefetches TMDB metadata using threads.
|
|
`requests.Session` is not guaranteed to be thread-safe, but reusing a session within
|
|
the same thread keeps connections warm.
|
|
"""
|
|
|
|
if requests is None:
|
|
return None
|
|
sess = getattr(_TMDB_THREAD_LOCAL, "session", None)
|
|
if sess is None:
|
|
sess = requests.Session()
|
|
setattr(_TMDB_THREAD_LOCAL, "session", sess)
|
|
return sess
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TmdbCastMember:
|
|
name: str
|
|
role: str
|
|
thumb: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TmdbShowMeta:
|
|
tmdb_id: int
|
|
plot: str
|
|
poster: str
|
|
fanart: str
|
|
rating: float
|
|
votes: int
|
|
cast: List[TmdbCastMember]
|
|
|
|
|
|
def _image_url(path: str, *, size: str) -> str:
|
|
path = (path or "").strip()
|
|
if not path:
|
|
return ""
|
|
return f"{TMDB_IMAGE_BASE}/{size}{path}"
|
|
|
|
|
|
def _fetch_credits(
|
|
*,
|
|
kind: str,
|
|
tmdb_id: int,
|
|
api_key: str,
|
|
language: str,
|
|
timeout: int,
|
|
log: Callable[[str], None] | None,
|
|
log_responses: bool,
|
|
) -> List[TmdbCastMember]:
|
|
if requests is None or not tmdb_id:
|
|
return []
|
|
params = {"api_key": api_key, "language": (language or "de-DE").strip()}
|
|
url = f"{TMDB_API_BASE}/{kind}/{tmdb_id}/credits?{urlencode(params)}"
|
|
status, payload, body_text = _tmdb_get_json(url=url, timeout=timeout, log=log, log_responses=log_responses)
|
|
if callable(log):
|
|
log(f"TMDB RESPONSE /{kind}/{{id}}/credits status={status}")
|
|
if log_responses and payload is None and body_text:
|
|
log(f"TMDB RESPONSE_BODY /{kind}/{{id}}/credits body={body_text[:2000]}")
|
|
if status != 200 or not isinstance(payload, dict):
|
|
return []
|
|
cast_payload = payload.get("cast") or []
|
|
if callable(log):
|
|
log(f"TMDB CREDITS /{kind}/{{id}}/credits cast={len(cast_payload)}")
|
|
return _parse_cast_payload(cast_payload)
|
|
|
|
|
|
def _parse_cast_payload(cast_payload: object) -> List[TmdbCastMember]:
|
|
if not isinstance(cast_payload, list):
|
|
return []
|
|
with_images: List[TmdbCastMember] = []
|
|
without_images: List[TmdbCastMember] = []
|
|
for entry in cast_payload:
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
name = (entry.get("name") or "").strip()
|
|
role = (entry.get("character") or "").strip()
|
|
thumb = _image_url(entry.get("profile_path") or "", size="w185")
|
|
if not name:
|
|
continue
|
|
member = TmdbCastMember(name=name, role=role, thumb=thumb)
|
|
if thumb:
|
|
with_images.append(member)
|
|
else:
|
|
without_images.append(member)
|
|
if with_images:
|
|
return with_images[:MAX_CAST_MEMBERS]
|
|
return without_images[:MAX_CAST_MEMBERS]
|
|
|
|
|
|
def _tmdb_get_json(
|
|
*,
|
|
url: str,
|
|
timeout: int,
|
|
log: Callable[[str], None] | None,
|
|
log_responses: bool,
|
|
session: "requests.Session | None" = None,
|
|
) -> Tuple[int | None, object | None, str]:
|
|
"""Fetches TMDB JSON with optional shared session.
|
|
|
|
Returns: (status_code, payload_or_none, body_text_or_empty)
|
|
"""
|
|
|
|
if requests is None:
|
|
return None, None, ""
|
|
if callable(log):
|
|
log(f"TMDB GET {url}")
|
|
sess = session or _get_tmdb_session() or requests.Session()
|
|
response = None
|
|
try:
|
|
response = sess.get(url, timeout=timeout)
|
|
status = getattr(response, "status_code", None)
|
|
payload: object | None = None
|
|
body_text = ""
|
|
try:
|
|
payload = response.json()
|
|
except Exception:
|
|
try:
|
|
body_text = (response.text or "").strip()
|
|
except Exception:
|
|
body_text = ""
|
|
except Exception as exc: # pragma: no cover
|
|
if callable(log):
|
|
log(f"TMDB ERROR request_failed url={url} error={exc!r}")
|
|
return None, None, ""
|
|
finally:
|
|
if response is not None:
|
|
try:
|
|
response.close()
|
|
except Exception:
|
|
pass
|
|
|
|
if callable(log):
|
|
log(f"TMDB RESPONSE status={status} url={url}")
|
|
if log_responses:
|
|
if payload is not None:
|
|
try:
|
|
dumped = json.dumps(payload, ensure_ascii=False)
|
|
except Exception:
|
|
dumped = str(payload)
|
|
log(f"TMDB RESPONSE_BODY url={url} body={dumped[:2000]}")
|
|
elif body_text:
|
|
log(f"TMDB RESPONSE_BODY url={url} body={body_text[:2000]}")
|
|
return status, payload, body_text
|
|
|
|
|
|
def fetch_tv_episode_credits(
|
|
*,
|
|
tmdb_id: int,
|
|
season_number: int,
|
|
episode_number: int,
|
|
api_key: str,
|
|
language: str = "de-DE",
|
|
timeout: int = 15,
|
|
log: Callable[[str], None] | None = None,
|
|
log_responses: bool = False,
|
|
) -> List[TmdbCastMember]:
|
|
"""Lädt Cast für eine konkrete Episode (/tv/{id}/season/{n}/episode/{e}/credits)."""
|
|
if requests is None:
|
|
return []
|
|
api_key = (api_key or "").strip()
|
|
if not api_key or not tmdb_id:
|
|
return []
|
|
params = {"api_key": api_key, "language": (language or "de-DE").strip()}
|
|
url = f"{TMDB_API_BASE}/tv/{tmdb_id}/season/{season_number}/episode/{episode_number}/credits?{urlencode(params)}"
|
|
status, payload, body_text = _tmdb_get_json(url=url, timeout=timeout, log=log, log_responses=log_responses)
|
|
if callable(log):
|
|
log(f"TMDB RESPONSE /tv/{{id}}/season/{{n}}/episode/{{e}}/credits status={status}")
|
|
if log_responses and payload is None and body_text:
|
|
log(f"TMDB RESPONSE_BODY /tv/{{id}}/season/{{n}}/episode/{{e}}/credits body={body_text[:2000]}")
|
|
if status != 200 or not isinstance(payload, dict):
|
|
return []
|
|
cast_payload = payload.get("cast") or []
|
|
if callable(log):
|
|
log(f"TMDB CREDITS /tv/{{id}}/season/{{n}}/episode/{{e}}/credits cast={len(cast_payload)}")
|
|
return _parse_cast_payload(cast_payload)
|
|
|
|
|
|
def lookup_tv_show(
|
|
*,
|
|
title: str,
|
|
api_key: str,
|
|
language: str = "de-DE",
|
|
timeout: int = 15,
|
|
log: Callable[[str], None] | None = None,
|
|
log_responses: bool = False,
|
|
include_cast: bool = False,
|
|
) -> Optional[TmdbShowMeta]:
|
|
"""Sucht eine TV-Show bei TMDB und liefert Plot + Poster-URL (wenn vorhanden)."""
|
|
if requests is None:
|
|
return None
|
|
api_key = (api_key or "").strip()
|
|
if not api_key:
|
|
return None
|
|
query = (title or "").strip()
|
|
if not query:
|
|
return None
|
|
|
|
params = {
|
|
"api_key": api_key,
|
|
"language": (language or "de-DE").strip(),
|
|
"query": query,
|
|
"include_adult": "false",
|
|
"page": "1",
|
|
}
|
|
url = f"{TMDB_API_BASE}/search/tv?{urlencode(params)}"
|
|
status, payload, body_text = _tmdb_get_json(
|
|
url=url,
|
|
timeout=timeout,
|
|
log=log,
|
|
log_responses=log_responses,
|
|
)
|
|
results = (payload or {}).get("results") if isinstance(payload, dict) else []
|
|
results = results or []
|
|
if callable(log):
|
|
log(f"TMDB RESPONSE /search/tv status={status} results={len(results)}")
|
|
if log_responses and payload is None and body_text:
|
|
log(f"TMDB RESPONSE_BODY /search/tv body={body_text[:2000]}")
|
|
|
|
if status != 200:
|
|
return None
|
|
if not results:
|
|
return None
|
|
|
|
normalized_query = query.casefold()
|
|
best = None
|
|
for candidate in results:
|
|
name = (candidate.get("name") or "").casefold()
|
|
original_name = (candidate.get("original_name") or "").casefold()
|
|
if name == normalized_query or original_name == normalized_query:
|
|
best = candidate
|
|
break
|
|
if best is None:
|
|
best = results[0]
|
|
|
|
tmdb_id = int(best.get("id") or 0)
|
|
plot = (best.get("overview") or "").strip()
|
|
poster = _image_url(best.get("poster_path") or "", size="w342")
|
|
fanart = _image_url(best.get("backdrop_path") or "", size="w780")
|
|
try:
|
|
rating = float(best.get("vote_average") or 0.0)
|
|
except Exception:
|
|
rating = 0.0
|
|
try:
|
|
votes = int(best.get("vote_count") or 0)
|
|
except Exception:
|
|
votes = 0
|
|
if not tmdb_id:
|
|
return None
|
|
cast: List[TmdbCastMember] = []
|
|
if include_cast and tmdb_id:
|
|
detail_params = {
|
|
"api_key": api_key,
|
|
"language": (language or "de-DE").strip(),
|
|
"append_to_response": "credits",
|
|
}
|
|
detail_url = f"{TMDB_API_BASE}/tv/{tmdb_id}?{urlencode(detail_params)}"
|
|
d_status, d_payload, d_body = _tmdb_get_json(
|
|
url=detail_url,
|
|
timeout=timeout,
|
|
log=log,
|
|
log_responses=log_responses,
|
|
)
|
|
if callable(log):
|
|
log(f"TMDB RESPONSE /tv/{{id}} status={d_status}")
|
|
if log_responses and d_payload is None and d_body:
|
|
log(f"TMDB RESPONSE_BODY /tv/{{id}} body={d_body[:2000]}")
|
|
if d_status == 200 and isinstance(d_payload, dict):
|
|
credits = d_payload.get("credits") or {}
|
|
cast = _parse_cast_payload((credits or {}).get("cast"))
|
|
if not plot and not poster and not fanart and not rating and not votes and not cast:
|
|
return None
|
|
return TmdbShowMeta(
|
|
tmdb_id=tmdb_id,
|
|
plot=plot,
|
|
poster=poster,
|
|
fanart=fanart,
|
|
rating=rating,
|
|
votes=votes,
|
|
cast=cast,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TmdbMovieMeta:
|
|
tmdb_id: int
|
|
plot: str
|
|
poster: str
|
|
fanart: str
|
|
runtime_minutes: int
|
|
rating: float
|
|
votes: int
|
|
cast: List[TmdbCastMember]
|
|
|
|
|
|
def _fetch_movie_details(
|
|
*,
|
|
tmdb_id: int,
|
|
api_key: str,
|
|
language: str,
|
|
timeout: int,
|
|
log: Callable[[str], None] | None,
|
|
log_responses: bool,
|
|
include_cast: bool,
|
|
) -> Tuple[int, List[TmdbCastMember]]:
|
|
"""Fetches /movie/{id} and (optionally) bundles credits via append_to_response=credits."""
|
|
if requests is None or not tmdb_id:
|
|
return 0, []
|
|
api_key = (api_key or "").strip()
|
|
if not api_key:
|
|
return 0, []
|
|
params: Dict[str, str] = {
|
|
"api_key": api_key,
|
|
"language": (language or "de-DE").strip(),
|
|
}
|
|
if include_cast:
|
|
params["append_to_response"] = "credits"
|
|
url = f"{TMDB_API_BASE}/movie/{tmdb_id}?{urlencode(params)}"
|
|
status, payload, body_text = _tmdb_get_json(url=url, timeout=timeout, log=log, log_responses=log_responses)
|
|
if callable(log):
|
|
log(f"TMDB RESPONSE /movie/{{id}} status={status}")
|
|
if log_responses and payload is None and body_text:
|
|
log(f"TMDB RESPONSE_BODY /movie/{{id}} body={body_text[:2000]}")
|
|
if status != 200 or not isinstance(payload, dict):
|
|
return 0, []
|
|
try:
|
|
runtime = int(payload.get("runtime") or 0)
|
|
except Exception:
|
|
runtime = 0
|
|
cast: List[TmdbCastMember] = []
|
|
if include_cast:
|
|
credits = payload.get("credits") or {}
|
|
cast = _parse_cast_payload((credits or {}).get("cast"))
|
|
return runtime, cast
|
|
|
|
|
|
def lookup_movie(
|
|
*,
|
|
title: str,
|
|
api_key: str,
|
|
language: str = "de-DE",
|
|
timeout: int = 15,
|
|
log: Callable[[str], None] | None = None,
|
|
log_responses: bool = False,
|
|
include_cast: bool = False,
|
|
) -> Optional[TmdbMovieMeta]:
|
|
"""Sucht einen Film bei TMDB und liefert Plot + Poster-URL (wenn vorhanden)."""
|
|
if requests is None:
|
|
return None
|
|
api_key = (api_key or "").strip()
|
|
if not api_key:
|
|
return None
|
|
query = (title or "").strip()
|
|
if not query:
|
|
return None
|
|
|
|
params = {
|
|
"api_key": api_key,
|
|
"language": (language or "de-DE").strip(),
|
|
"query": query,
|
|
"include_adult": "false",
|
|
"page": "1",
|
|
}
|
|
url = f"{TMDB_API_BASE}/search/movie?{urlencode(params)}"
|
|
status, payload, body_text = _tmdb_get_json(
|
|
url=url,
|
|
timeout=timeout,
|
|
log=log,
|
|
log_responses=log_responses,
|
|
)
|
|
results = (payload or {}).get("results") if isinstance(payload, dict) else []
|
|
results = results or []
|
|
if callable(log):
|
|
log(f"TMDB RESPONSE /search/movie status={status} results={len(results)}")
|
|
if log_responses and payload is None and body_text:
|
|
log(f"TMDB RESPONSE_BODY /search/movie body={body_text[:2000]}")
|
|
|
|
if status != 200:
|
|
return None
|
|
if not results:
|
|
return None
|
|
|
|
normalized_query = query.casefold()
|
|
best = None
|
|
for candidate in results:
|
|
name = (candidate.get("title") or "").casefold()
|
|
original_name = (candidate.get("original_title") or "").casefold()
|
|
if name == normalized_query or original_name == normalized_query:
|
|
best = candidate
|
|
break
|
|
if best is None:
|
|
best = results[0]
|
|
|
|
tmdb_id = int(best.get("id") or 0)
|
|
plot = (best.get("overview") or "").strip()
|
|
poster = _image_url(best.get("poster_path") or "", size="w342")
|
|
fanart = _image_url(best.get("backdrop_path") or "", size="w780")
|
|
runtime_minutes = 0
|
|
try:
|
|
rating = float(best.get("vote_average") or 0.0)
|
|
except Exception:
|
|
rating = 0.0
|
|
try:
|
|
votes = int(best.get("vote_count") or 0)
|
|
except Exception:
|
|
votes = 0
|
|
if not tmdb_id:
|
|
return None
|
|
cast: List[TmdbCastMember] = []
|
|
runtime_minutes, cast = _fetch_movie_details(
|
|
tmdb_id=tmdb_id,
|
|
api_key=api_key,
|
|
language=language,
|
|
timeout=timeout,
|
|
log=log,
|
|
log_responses=log_responses,
|
|
include_cast=include_cast,
|
|
)
|
|
if not plot and not poster and not fanart and not rating and not votes and not cast:
|
|
return None
|
|
return TmdbMovieMeta(
|
|
tmdb_id=tmdb_id,
|
|
plot=plot,
|
|
poster=poster,
|
|
fanart=fanart,
|
|
runtime_minutes=runtime_minutes,
|
|
rating=rating,
|
|
votes=votes,
|
|
cast=cast,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TmdbEpisodeMeta:
|
|
plot: str
|
|
thumb: str
|
|
runtime_minutes: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TmdbSeasonMeta:
|
|
plot: str
|
|
poster: str
|
|
|
|
|
|
def lookup_tv_season_summary(
|
|
*,
|
|
tmdb_id: int,
|
|
season_number: int,
|
|
api_key: str,
|
|
language: str = "de-DE",
|
|
timeout: int = 15,
|
|
log: Callable[[str], None] | None = None,
|
|
log_responses: bool = False,
|
|
) -> Optional[TmdbSeasonMeta]:
|
|
"""Lädt Staffel-Meta (Plot + Poster)."""
|
|
if requests is None:
|
|
return None
|
|
|
|
api_key = (api_key or "").strip()
|
|
if not api_key or not tmdb_id:
|
|
return None
|
|
|
|
params = {"api_key": api_key, "language": (language or "de-DE").strip()}
|
|
url = f"{TMDB_API_BASE}/tv/{tmdb_id}/season/{season_number}?{urlencode(params)}"
|
|
status, payload, body_text = _tmdb_get_json(url=url, timeout=timeout, log=log, log_responses=log_responses)
|
|
if callable(log):
|
|
log(f"TMDB RESPONSE /tv/{{id}}/season/{{n}} status={status}")
|
|
if log_responses and payload is None and body_text:
|
|
log(f"TMDB RESPONSE_BODY /tv/{{id}}/season/{{n}} body={body_text[:2000]}")
|
|
if status != 200 or not isinstance(payload, dict):
|
|
return None
|
|
|
|
plot = (payload.get("overview") or "").strip()
|
|
poster_path = (payload.get("poster_path") or "").strip()
|
|
poster = f"{TMDB_IMAGE_BASE}/w342{poster_path}" if poster_path else ""
|
|
if not plot and not poster:
|
|
return None
|
|
return TmdbSeasonMeta(plot=plot, poster=poster)
|
|
|
|
|
|
def lookup_tv_season(
|
|
*,
|
|
tmdb_id: int,
|
|
season_number: int,
|
|
api_key: str,
|
|
language: str = "de-DE",
|
|
timeout: int = 15,
|
|
log: Callable[[str], None] | None = None,
|
|
log_responses: bool = False,
|
|
) -> Optional[Dict[int, TmdbEpisodeMeta]]:
|
|
"""Lädt Episoden-Meta für eine Staffel: episode_number -> (plot, thumb)."""
|
|
if requests is None:
|
|
return None
|
|
api_key = (api_key or "").strip()
|
|
if not api_key or not tmdb_id or season_number is None:
|
|
return None
|
|
params = {"api_key": api_key, "language": (language or "de-DE").strip()}
|
|
url = f"{TMDB_API_BASE}/tv/{tmdb_id}/season/{season_number}?{urlencode(params)}"
|
|
status, payload, body_text = _tmdb_get_json(url=url, timeout=timeout, log=log, log_responses=log_responses)
|
|
episodes = (payload or {}).get("episodes") if isinstance(payload, dict) else []
|
|
episodes = episodes or []
|
|
if callable(log):
|
|
log(f"TMDB RESPONSE /tv/{{id}}/season/{{n}} status={status} episodes={len(episodes)}")
|
|
if log_responses:
|
|
if payload is not None:
|
|
try:
|
|
dumped = json.dumps(payload, ensure_ascii=False)
|
|
except Exception:
|
|
dumped = str(payload)
|
|
log(f"TMDB RESPONSE_BODY /tv/{{id}}/season/{{n}} body={dumped[:2000]}")
|
|
elif body_text:
|
|
log(f"TMDB RESPONSE_BODY /tv/{{id}}/season/{{n}} body={body_text[:2000]}")
|
|
|
|
if status != 200 or not episodes:
|
|
return None
|
|
|
|
result: Dict[int, TmdbEpisodeMeta] = {}
|
|
for entry in episodes:
|
|
try:
|
|
ep_number = int(entry.get("episode_number") or 0)
|
|
except Exception:
|
|
continue
|
|
if not ep_number:
|
|
continue
|
|
plot = (entry.get("overview") or "").strip()
|
|
runtime_minutes = 0
|
|
try:
|
|
runtime_minutes = int(entry.get("runtime") or 0)
|
|
except Exception:
|
|
runtime_minutes = 0
|
|
still_path = (entry.get("still_path") or "").strip()
|
|
thumb = f"{TMDB_IMAGE_BASE}/w300{still_path}" if still_path else ""
|
|
if not plot and not thumb and not runtime_minutes:
|
|
continue
|
|
result[ep_number] = TmdbEpisodeMeta(plot=plot, thumb=thumb, runtime_minutes=runtime_minutes)
|
|
return result or None
|