HS-Rename/engine/tvdb_client.py

"""
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"),
    ("Alternate order", "alternate"),
    ("Regional order", "regional"),
]


@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 get_season_episodes(
        self,
        series_id: int,
        season: int,
        season_type: str = "default",
    ) -> list[TvdbEpisode]:
        episodes: list[TvdbEpisode] = []
        page = 0
        while True:
            payload = self._request(
                "GET",
                f"/series/{series_id}/episodes/{season_type}",
                params={"page": page, "season": season},
            )
            batch = (payload.get("data") or {}).get("episodes") or []
            if not batch:
                break
            for ep in batch:
                if ep.get("seasonNumber") != season:
                    continue
                number = ep.get("number")
                name = ep.get("name")
                if number is None or not name:
                    continue
                episodes.append(
                    TvdbEpisode(
                        number=int(number),
                        season_number=int(season),
                        name=str(name),
                    )
                )
            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_all_episodes(
        self,
        series_id: int,
        season_type: str = "default",
    ) -> list[TvdbEpisode]:
        """Fetch every episode for a series (all seasons), paginated."""
        episodes: list[TvdbEpisode] = []
        page = 0
        while True:
            payload = self._request(
                "GET",
                f"/series/{series_id}/episodes/{season_type}",
                params={"page": page},
            )
            batch = (payload.get("data") or {}).get("episodes") or []
            if not batch:
                break
            for ep in batch:
                number = ep.get("number")
                season = ep.get("seasonNumber")
                name = ep.get("name")
                if number is None or season is None or not name:
                    continue
                if int(season) == 0:
                    continue
                episodes.append(
                    TvdbEpisode(
                        number=int(number),
                        season_number=int(season),
                        name=str(name),
                    )
                )
            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