Bulk RenamerandCursor
64d7a30328
Emit plain tuples across thread boundaries, prevent overlapping match workers, scope episode comparisons by season, and cap API pagination. Co-authored-by: Cursor <[email protected]>
265 lines
8.2 KiB
Python
265 lines
8.2 KiB
Python
"""
|
|
TheTVDB API v4 client (read-only). Uses stdlib urllib — no extra dependencies.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from dataclasses import dataclass
|
|
from typing import Any, Optional
|
|
|
|
BASE_URL = "https://api4.thetvdb.com/v4"
|
|
TVDB_API_KEY = "78384b24-0f0b-461e-9415-69cce3da35e4"
|
|
|
|
# Labels and API path segments for /series/{id}/episodes/{season-type}
|
|
SEASON_TYPE_CHOICES: list[tuple[str, str]] = [
|
|
("Default (TheTVDB default for show)", "default"),
|
|
("Official / aired order", "official"),
|
|
("DVD order", "dvd"),
|
|
("Absolute order", "absolute"),
|
|
("Combined order (multi-episode files)", "alternate"),
|
|
("Regional order", "regional"),
|
|
]
|
|
|
|
# Language path segment for /series/{id}/episodes/{season-type}/{language}
|
|
LANGUAGE_CHOICES: list[tuple[str, str]] = [
|
|
("Default (show language)", ""),
|
|
("English", "eng"),
|
|
]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TvdbSeries:
|
|
id: int
|
|
name: str
|
|
year: Optional[str] = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TvdbEpisode:
|
|
number: int
|
|
season_number: int
|
|
name: str
|
|
|
|
|
|
class TvdbError(Exception):
|
|
pass
|
|
|
|
|
|
def _parse_series_id(item: dict[str, Any]) -> Optional[int]:
|
|
"""Extract numeric series id from a TheTVDB v4 search hit."""
|
|
tvdb_id = item.get("tvdb_id")
|
|
if tvdb_id is not None:
|
|
try:
|
|
return int(tvdb_id)
|
|
except (TypeError, ValueError):
|
|
pass
|
|
|
|
raw_id = item.get("id")
|
|
if raw_id is None:
|
|
return None
|
|
if isinstance(raw_id, int):
|
|
return raw_id
|
|
if isinstance(raw_id, str):
|
|
if raw_id.isdigit():
|
|
return int(raw_id)
|
|
m = re.match(r"^series-(\d+)$", raw_id, re.IGNORECASE)
|
|
if m:
|
|
return int(m.group(1))
|
|
return None
|
|
|
|
|
|
class TvdbClient:
|
|
def __init__(self, api_key: str, pin: str = ""):
|
|
self.api_key = api_key.strip()
|
|
self.pin = pin.strip()
|
|
self._token: Optional[str] = None
|
|
|
|
def _request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
params: Optional[dict[str, Any]] = None,
|
|
body: Optional[dict[str, Any]] = None,
|
|
) -> dict[str, Any]:
|
|
if not self._token and path != "/login":
|
|
self.login()
|
|
|
|
url = BASE_URL + path
|
|
if params:
|
|
query = urllib.parse.urlencode({k: v for k, v in params.items() if v is not None})
|
|
if query:
|
|
url = f"{url}?{query}"
|
|
|
|
data_bytes: Optional[bytes] = None
|
|
headers = {"Accept": "application/json"}
|
|
if body is not None:
|
|
data_bytes = json.dumps(body).encode("utf-8")
|
|
headers["Content-Type"] = "application/json"
|
|
if self._token:
|
|
headers["Authorization"] = f"Bearer {self._token}"
|
|
|
|
req = urllib.request.Request(url, data=data_bytes, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
payload = json.loads(resp.read().decode("utf-8"))
|
|
except urllib.error.HTTPError as e:
|
|
detail = e.read().decode("utf-8", errors="replace")
|
|
raise TvdbError(f"TheTVDB HTTP {e.code}: {detail}") from e
|
|
except urllib.error.URLError as e:
|
|
raise TvdbError(f"TheTVDB network error: {e.reason}") from e
|
|
except json.JSONDecodeError as e:
|
|
raise TvdbError("TheTVDB returned invalid JSON") from e
|
|
|
|
if payload.get("status") == "failure":
|
|
msg = payload.get("message") or "Unknown TheTVDB error"
|
|
raise TvdbError(msg)
|
|
return payload
|
|
|
|
def login(self) -> None:
|
|
if not self.api_key:
|
|
raise TvdbError("TheTVDB API key is required")
|
|
body: dict[str, str] = {"apikey": self.api_key}
|
|
if self.pin:
|
|
body["pin"] = self.pin
|
|
payload = self._request("POST", "/login", body=body)
|
|
token = payload.get("data", {}).get("token")
|
|
if not token:
|
|
raise TvdbError("TheTVDB login did not return a token")
|
|
self._token = token
|
|
|
|
def search_series(self, query: str, limit: int = 25) -> list[TvdbSeries]:
|
|
query = query.strip()
|
|
if not query:
|
|
return []
|
|
payload = self._request(
|
|
"GET",
|
|
"/search",
|
|
params={"query": query, "type": "series", "limit": limit},
|
|
)
|
|
results: list[TvdbSeries] = []
|
|
for item in payload.get("data") or []:
|
|
item_type = item.get("type") or item.get("primary_type")
|
|
if item_type and item_type != "series":
|
|
continue
|
|
series_id = _parse_series_id(item)
|
|
if series_id is None:
|
|
continue
|
|
year = item.get("year")
|
|
if year is not None:
|
|
year = str(year)
|
|
results.append(
|
|
TvdbSeries(
|
|
id=series_id,
|
|
name=item.get("name") or item.get("title") or f"Series {series_id}",
|
|
year=year,
|
|
)
|
|
)
|
|
return results
|
|
|
|
def _episodes_path(self, series_id: int, season_type: str, language: str = "") -> str:
|
|
base = f"/series/{series_id}/episodes/{season_type}"
|
|
if language:
|
|
return f"{base}/{language}"
|
|
return base
|
|
|
|
def _parse_episode_batch(
|
|
self,
|
|
batch: list[dict[str, Any]],
|
|
*,
|
|
season: Optional[int] = None,
|
|
skip_specials: bool = False,
|
|
) -> list[TvdbEpisode]:
|
|
episodes: list[TvdbEpisode] = []
|
|
for ep in batch:
|
|
ep_season = ep.get("seasonNumber")
|
|
number = ep.get("number")
|
|
name = ep.get("name")
|
|
if number is None or ep_season is None or not name:
|
|
continue
|
|
ep_season = int(ep_season)
|
|
if skip_specials and ep_season == 0:
|
|
continue
|
|
if season is not None and ep_season != season:
|
|
continue
|
|
episodes.append(
|
|
TvdbEpisode(
|
|
number=int(number),
|
|
season_number=ep_season,
|
|
name=str(name),
|
|
)
|
|
)
|
|
return episodes
|
|
|
|
def _fetch_episodes(
|
|
self,
|
|
series_id: int,
|
|
season_type: str,
|
|
*,
|
|
season: Optional[int] = None,
|
|
language: str = "",
|
|
skip_specials: bool = False,
|
|
) -> list[TvdbEpisode]:
|
|
"""Fetch episodes; optional translated titles via language code (e.g. eng)."""
|
|
episodes: list[TvdbEpisode] = []
|
|
page = 0
|
|
max_pages = 50
|
|
use_api_season = season is not None and not language
|
|
while page < max_pages:
|
|
params: dict[str, Any] = {"page": page}
|
|
if use_api_season:
|
|
params["season"] = season
|
|
payload = self._request(
|
|
"GET",
|
|
self._episodes_path(series_id, season_type, language),
|
|
params=params,
|
|
)
|
|
batch = (payload.get("data") or {}).get("episodes") or []
|
|
if not batch:
|
|
break
|
|
filter_season = season if language else None
|
|
episodes.extend(
|
|
self._parse_episode_batch(
|
|
batch,
|
|
season=filter_season,
|
|
skip_specials=skip_specials,
|
|
)
|
|
)
|
|
links = payload.get("links") or {}
|
|
if not links.get("next"):
|
|
break
|
|
page += 1
|
|
episodes.sort(key=lambda e: (e.season_number, e.number))
|
|
return episodes
|
|
|
|
def get_season_episodes(
|
|
self,
|
|
series_id: int,
|
|
season: int,
|
|
season_type: str = "default",
|
|
language: str = "",
|
|
) -> list[TvdbEpisode]:
|
|
return self._fetch_episodes(
|
|
series_id,
|
|
season_type,
|
|
season=season,
|
|
language=language,
|
|
)
|
|
|
|
def get_all_episodes(
|
|
self,
|
|
series_id: int,
|
|
season_type: str = "default",
|
|
language: str = "",
|
|
) -> list[TvdbEpisode]:
|
|
"""Fetch every episode for a series (all seasons), paginated."""
|
|
return self._fetch_episodes(
|
|
series_id,
|
|
season_type,
|
|
language=language,
|
|
skip_specials=True,
|
|
)
|