Implement ViewIt Plugin System Documentation and Update Project Notes
- Added comprehensive documentation for the ViewIt Plugin System, detailing the plugin loading process, required methods, optional features, and community extension workflow. - Updated project notes to reflect the current structure, build process, search logic, and known issues. - Introduced new build scripts for installing the add-on and creating ZIP packages. - Added test scripts for TMDB API integration, including argument parsing and logging functionality. - Enhanced existing plugins with improved search logic and error handling.
This commit is contained in:
652
addon/tmdb.py
Normal file
652
addon/tmdb.py
Normal file
@@ -0,0 +1,652 @@
|
||||
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"
|
||||
_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)}"
|
||||
if callable(log):
|
||||
log(f"TMDB GET {url}")
|
||||
try:
|
||||
response = requests.get(url, timeout=timeout)
|
||||
except Exception as exc: # pragma: no cover
|
||||
if callable(log):
|
||||
log(f"TMDB ERROR /{kind}/{{id}}/credits request_failed error={exc!r}")
|
||||
return []
|
||||
status = getattr(response, "status_code", None)
|
||||
if callable(log):
|
||||
log(f"TMDB RESPONSE /{kind}/{{id}}/credits status={status}")
|
||||
if status != 200:
|
||||
return []
|
||||
try:
|
||||
payload = response.json() or {}
|
||||
except Exception:
|
||||
return []
|
||||
if callable(log) and log_responses:
|
||||
try:
|
||||
dumped = json.dumps(payload, ensure_ascii=False)
|
||||
except Exception:
|
||||
dumped = str(payload)
|
||||
log(f"TMDB RESPONSE_BODY /{kind}/{{id}}/credits body={dumped[:2000]}")
|
||||
|
||||
cast_payload = payload.get("cast") or []
|
||||
if callable(log):
|
||||
log(f"TMDB CREDITS /{kind}/{{id}}/credits cast={len(cast_payload)}")
|
||||
with_images: List[TmdbCastMember] = []
|
||||
without_images: List[TmdbCastMember] = []
|
||||
for entry in cast_payload:
|
||||
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)
|
||||
|
||||
# Viele Kodi-Skins zeigen bei fehlendem Thumbnail Platzhalter-Köpfe.
|
||||
# Bevorzugt daher Cast-Einträge mit Bild; nur wenn gar keine Bilder existieren,
|
||||
# geben wir Namen ohne Bild zurück.
|
||||
if with_images:
|
||||
return with_images[:30]
|
||||
return without_images[:30]
|
||||
|
||||
|
||||
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[:30]
|
||||
return without_images[:30]
|
||||
|
||||
|
||||
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()
|
||||
try:
|
||||
response = sess.get(url, timeout=timeout)
|
||||
except Exception as exc: # pragma: no cover
|
||||
if callable(log):
|
||||
log(f"TMDB ERROR request_failed url={url} error={exc!r}")
|
||||
return None, None, ""
|
||||
|
||||
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 = ""
|
||||
|
||||
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)}"
|
||||
if callable(log):
|
||||
log(f"TMDB GET {url}")
|
||||
try:
|
||||
response = requests.get(url, timeout=timeout)
|
||||
except Exception as exc: # pragma: no cover
|
||||
if callable(log):
|
||||
log(f"TMDB ERROR /tv/{{id}}/season/{{n}}/episode/{{e}}/credits request_failed error={exc!r}")
|
||||
return []
|
||||
status = getattr(response, "status_code", None)
|
||||
if callable(log):
|
||||
log(f"TMDB RESPONSE /tv/{{id}}/season/{{n}}/episode/{{e}}/credits status={status}")
|
||||
if status != 200:
|
||||
return []
|
||||
try:
|
||||
payload = response.json() or {}
|
||||
except Exception:
|
||||
return []
|
||||
if callable(log) and log_responses:
|
||||
try:
|
||||
dumped = json.dumps(payload, ensure_ascii=False)
|
||||
except Exception:
|
||||
dumped = str(payload)
|
||||
log(f"TMDB RESPONSE_BODY /tv/{{id}}/season/{{n}}/episode/{{e}}/credits body={dumped[:2000]}")
|
||||
|
||||
cast_payload = payload.get("cast") or []
|
||||
if callable(log):
|
||||
log(f"TMDB CREDITS /tv/{{id}}/season/{{n}}/episode/{{e}}/credits cast={len(cast_payload)}")
|
||||
with_images: List[TmdbCastMember] = []
|
||||
without_images: List[TmdbCastMember] = []
|
||||
for entry in cast_payload:
|
||||
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[:30]
|
||||
return without_images[:30]
|
||||
|
||||
|
||||
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)}"
|
||||
if callable(log):
|
||||
log(f"TMDB GET {url}")
|
||||
try:
|
||||
response = requests.get(url, timeout=timeout)
|
||||
except Exception:
|
||||
return None
|
||||
status = getattr(response, "status_code", None)
|
||||
if callable(log):
|
||||
log(f"TMDB RESPONSE /tv/{{id}}/season/{{n}} status={status}")
|
||||
if status != 200:
|
||||
return None
|
||||
try:
|
||||
payload = response.json() or {}
|
||||
except Exception:
|
||||
return None
|
||||
if callable(log) and log_responses:
|
||||
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]}")
|
||||
|
||||
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)}"
|
||||
if callable(log):
|
||||
log(f"TMDB GET {url}")
|
||||
try:
|
||||
response = requests.get(url, timeout=timeout)
|
||||
except Exception as exc: # pragma: no cover
|
||||
if callable(log):
|
||||
log(f"TMDB ERROR /tv/{{id}}/season/{{n}} request_failed error={exc!r}")
|
||||
return None
|
||||
|
||||
status = getattr(response, "status_code", None)
|
||||
payload = None
|
||||
body_text = ""
|
||||
try:
|
||||
payload = response.json() or {}
|
||||
except Exception:
|
||||
try:
|
||||
body_text = (response.text or "").strip()
|
||||
except Exception:
|
||||
body_text = ""
|
||||
|
||||
episodes = (payload or {}).get("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
|
||||
Reference in New Issue
Block a user