From 4ac34069ecad81977b4ee532aeb75c2a56982d65 Mon Sep 17 00:00:00 2001 From: imSp4rky Date: Sun, 2 Aug 2026 10:48:32 -0600 Subject: [PATCH] feat(providers): add TheTVDB v4 as a metadata provider TVDB covers TV far better than the existing sources, so add it as a provider and let users decide where it sits in the lookup chain. - new TVDB v4 provider: search, get_by_id, get_external_ids and find_by_imdb_id, with a process-level token cache and 401 re-auth - metadata_providers config both orders and filters the providers used, so TVDB can be primary rather than a TMDB fallback. Cross-validation keeps its own fixed trust ranking, unchanged by search order - --tvdb-order renumbers episodes into a TVDB season order. A service does not always use aired order, so the source order is detected rather than assumed, and the renumber is refused when an order that omits episodes would give two of them the same filename - --tvdb sets the series ID directly, skipping the lookup - the Matroska TVDB2 tag now populates on every lookup path, including --imdb, which previously produced no ID tags when IMDxAPI was off Search matches on year as well as title so an identically named remake cannot win on title alone. --- tests/core/test_tvdb.py | 173 +++++++++++++++ unshackle/commands/dl.py | 104 ++++++++- unshackle/core/api/download_manager.py | 4 + unshackle/core/api/handlers.py | 2 + unshackle/core/api/routes.py | 7 + unshackle/core/config.py | 4 + unshackle/core/providers/__init__.py | 83 +++++-- unshackle/core/providers/tvdb.py | 286 +++++++++++++++++++++++++ unshackle/core/utils/tags.py | 30 ++- 9 files changed, 666 insertions(+), 27 deletions(-) create mode 100644 tests/core/test_tvdb.py create mode 100644 unshackle/core/providers/tvdb.py diff --git a/tests/core/test_tvdb.py b/tests/core/test_tvdb.py new file mode 100644 index 00000000..03f82573 --- /dev/null +++ b/tests/core/test_tvdb.py @@ -0,0 +1,173 @@ +"""TVDB provider ordering and season-order remapping.""" + +from __future__ import annotations + +import pytest + +from unshackle.core import providers +from unshackle.core.config import config +from unshackle.core.providers.tvdb import TVDBProvider, _ids_from_remote, _pick_match +from unshackle.core.utils.tags import _build_tags_from_ids + +REMOTE_IDS = [ + {"id": "tt0149460", "type": 2, "sourceName": "IMDB"}, + {"id": "http://example.com", "type": 4, "sourceName": "Official Website"}, + {"id": "615", "type": 12, "sourceName": "TheMovieDB.com"}, +] + +# Futurama: aired S1 stops at E9, the four held-back episodes open aired S2. +# DVD order folds them back into S1 as E10-E13. +AIRED = [ + {"id": 1, "seasonNumber": 1, "number": 9, "name": "Hell Is Other Robots"}, + {"id": 2, "seasonNumber": 2, "number": 1, "name": "A Flight to Remember"}, + {"id": 3, "seasonNumber": 2, "number": 2, "name": "Mars University"}, + {"id": 99, "seasonNumber": 2, "number": 3, "name": "Only In Aired"}, +] +DVD = [ + {"id": 1, "seasonNumber": 1, "number": 9, "name": "Hell Is Other Robots"}, + {"id": 2, "seasonNumber": 1, "number": 10, "name": "A Flight to Remember"}, + {"id": 3, "seasonNumber": 1, "number": 11, "name": "Mars University"}, +] + + +def test_ids_from_remote() -> None: + ext = _ids_from_remote(REMOTE_IDS, 73871) + assert ext.imdb_id == "tt0149460" + assert ext.tmdb_id == 615 + assert ext.tvdb_id == 73871 + + +def _stub(p: TVDBProvider, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(p, "get_episodes", lambda _id, order: {"official": AIRED, "dvd": DVD}.get(order, [])) + + +def test_order_map_joins_on_episode_id(monkeypatch: pytest.MonkeyPatch) -> None: + p = TVDBProvider() + _stub(p, monkeypatch) + + mapping = p.get_order_map(73871, "dvd") + assert mapping[(1, 9)] == (1, 9, "Hell Is Other Robots") + assert mapping[(2, 1)] == (1, 10, "A Flight to Remember") + assert mapping[(2, 2)] == (1, 11, "Mars University") + # an episode absent from the target order is left out, so it keeps its own numbering + assert (2, 3) not in mapping + + +def test_order_map_reverses_with_source_order(monkeypatch: pytest.MonkeyPatch) -> None: + p = TVDBProvider() + _stub(p, monkeypatch) + + mapping = p.get_order_map(73871, "official", source_order="dvd") + assert mapping[(1, 10)] == (2, 1, "A Flight to Remember") + + +def test_order_map_is_noop_when_orders_match() -> None: + assert TVDBProvider().get_order_map(73871, "official") == {} + + +def test_order_map_empty_when_order_missing(monkeypatch: pytest.MonkeyPatch) -> None: + p = TVDBProvider() + _stub(p, monkeypatch) + assert p.get_order_map(73871, "alternate") == {} + + +def test_detect_order_picks_the_order_the_numbering_matches(monkeypatch: pytest.MonkeyPatch) -> None: + p = TVDBProvider() + _stub(p, monkeypatch) + # a service may number a series in a non-aired order, so detection must not assume aired + assert p.detect_order(73871, [(1, 10), (1, 11)]) == "dvd" + assert p.detect_order(73871, [(2, 1), (2, 2), (2, 3)]) == "official" + + +def test_detect_order_falls_back_to_aired(monkeypatch: pytest.MonkeyPatch) -> None: + p = TVDBProvider() + monkeypatch.setattr(p, "get_episodes", lambda _id, order: []) + assert p.detect_order(73871, [(1, 1)]) == "official" + + +# TVDB holds both Battlestar Galactica series; the 2003 revival is filed under year 2005 +BSG = [ + {"name": "Battlestar Galactica", "year": "1978", "tvdb_id": "71173"}, + {"name": "Battlestar Galactica (2003)", "year": "2005", "tvdb_id": "73545"}, +] + + +def test_pick_match_prefers_the_right_era() -> None: + # exact title match on the 1978 series must not beat the era the caller asked for + best, _, _ = _pick_match(BSG, "Battlestar Galactica", 2004) + assert best["tvdb_id"] == "73545" + best, _, _ = _pick_match(BSG, "Battlestar Galactica", 1978) + assert best["tvdb_id"] == "71173" + + +def test_pick_match_without_a_year_scores_on_title_alone() -> None: + best, _, _ = _pick_match(BSG, "Battlestar Galactica", None) + assert best["tvdb_id"] == "71173" + + +def test_pick_match_returns_nothing_when_no_year_fits() -> None: + assert _pick_match(BSG, "Battlestar Galactica", 1995)[0] is None + + +def test_tvdb2_tag_follows_the_matroska_spec() -> None: + # TVDB2 IDs are prefixed by entity type: matroska.org/technical/tagging.html + from unshackle.core.providers import ExternalIds + + assert _build_tags_from_ids(ExternalIds(tvdb_id=73871), "tv")["TVDB2"] == "series/73871" + assert _build_tags_from_ids(ExternalIds(tvdb_id=113), "movie")["TVDB2"] == "movies/113" + + +def test_provider_order_defaults(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(config, "metadata_providers", []) + assert [c.NAME for c in providers.provider_order()] == list(providers.DEFAULT_ORDER) + + +def test_provider_order_is_configurable(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(config, "metadata_providers", ["TVDB", "tmdb", "nope", "tvdb"]) + # unknown entries dropped, duplicates collapsed, case ignored, order kept + assert [c.NAME for c in providers.provider_order()] == ["tvdb", "tmdb"] + + +def test_enrichment_authority_is_fixed(monkeypatch: pytest.MonkeyPatch) -> None: + # search order must not demote TMDB below SIMKL for cross-validation tie-breaks + monkeypatch.setattr(config, "metadata_providers", []) + assert providers._enrichment_providers() == ["tmdb", "simkl", "tvdb"] + # `metadata_providers` filters the set but does not re-rank the trust order + monkeypatch.setattr(config, "metadata_providers", ["tvdb", "simkl"]) + assert providers._enrichment_providers() == ["simkl", "tvdb"] + + +def test_cached_tvdb_result_keeps_enriched_ids() -> None: + cached = { + "tvdb_id": "73871", + "name": "Futurama", + "year": "1999", + "_enriched_ids": {"imdb_id": "tt0149460", "tmdb_id": 615}, + } + result = providers._cached_to_result(cached, "tvdb", "tv") + assert result is not None + assert result.external_ids.tvdb_id == 73871 + assert result.external_ids.imdb_id == "tt0149460" + assert result.external_ids.tmdb_id == 615 + # remote_ids stay authoritative over the enriched copies + cached["remote_ids"] = REMOTE_IDS + cached["_enriched_ids"] = {"imdb_id": "tt9999999", "tmdb_id": 1} + result = providers._cached_to_result(cached, "tvdb", "tv") + assert result.external_ids.imdb_id == "tt0149460" + assert result.external_ids.tmdb_id == 615 + + +def test_get_episodes_fails_closed_on_a_truncated_listing(monkeypatch: pytest.MonkeyPatch) -> None: + p = TVDBProvider() + pages: dict[int, object] = {0: {"episodes": [{"id": n} for n in range(500)]}, 1: None} + monkeypatch.setattr(p, "_get", lambda _path, params: pages[params["page"]]) + # page 1 fails mid-listing: a partial list would renumber wrongly + assert p.get_episodes(73871, "official") == [] + assert p._episodes == {} + + +def test_get_episodes_does_not_cache_an_empty_listing(monkeypatch: pytest.MonkeyPatch) -> None: + p = TVDBProvider() + monkeypatch.setattr(p, "_get", lambda _path, params: None) + assert p.get_episodes(73871, "official") == [] + assert p._episodes == {} diff --git a/unshackle/commands/dl.py b/unshackle/commands/dl.py index 8ef9832e..685ac411 100644 --- a/unshackle/commands/dl.py +++ b/unshackle/commands/dl.py @@ -11,6 +11,7 @@ import shutil import subprocess import sys import time +from collections import Counter from collections.abc import Iterable from concurrent import futures from concurrent.futures import ThreadPoolExecutor @@ -58,6 +59,7 @@ from unshackle.core.music import ( write_music_manifest, write_music_metadata, ) +from unshackle.core.providers.tvdb import SEASON_TYPES from unshackle.core.proxies import Basic, ExpressVPN, Gluetun, Hola, NordVPN, ProtonVPN, SurfsharkVPN, WindscribeVPN from unshackle.core.service import Service from unshackle.core.services import Services @@ -587,6 +589,20 @@ class dl: default=None, help="Use this TMDB ID for tagging instead of automatic lookup.", ) + @click.option( + "--tvdb", + "tvdb_id", + type=int, + default=None, + help="Use this TVDB ID for --tvdb-order instead of looking the series up automatically.", + ) + @click.option( + "--tvdb-order", + type=click.Choice(SEASON_TYPES, case_sensitive=False), + default=None, + help="Renumber episodes to a TVDB season order. Services normally use 'official' (aired) order; " + "a series' Streaming Order is 'alternate'. Defaults to the tvdb_order config option.", + ) @click.option( "--animeapi", "animeapi_id", @@ -760,6 +776,8 @@ class dl: tag: Optional[str] = None, tmdb_id: Optional[int] = None, imdb_id: Optional[str] = None, + tvdb_id: Optional[int] = None, + tvdb_order: Optional[str] = None, animeapi_id: Optional[str] = None, enrich: bool = False, output_dir: Optional[Path] = None, @@ -798,6 +816,8 @@ class dl: tag = ctx.params.get("tag", tag) tmdb_id = ctx.params.get("tmdb_id", tmdb_id) imdb_id = ctx.params.get("imdb_id", imdb_id) + tvdb_id = ctx.params.get("tvdb_id", tvdb_id) + tvdb_order = ctx.params.get("tvdb_order", tvdb_order) animeapi_id = ctx.params.get("animeapi_id", animeapi_id) enrich = ctx.params.get("enrich", enrich) output_dir = ctx.params.get("output_dir", output_dir) @@ -806,6 +826,10 @@ class dl: self.proxy_requested = bool(proxy) self.tmdb_id = tmdb_id self.imdb_id = imdb_id + self.tvdb_id = tvdb_id + self.tvdb_order = (tvdb_order or config.tvdb_order or "").lower() + if self.tvdb_order and self.tvdb_order not in SEASON_TYPES: + raise click.UsageError(f"tvdb_order must be one of {', '.join(SEASON_TYPES)}, not {self.tvdb_order!r}") self.enrich = enrich self.animeapi_title: Optional[str] = None self.output_dir = output_dir @@ -819,6 +843,8 @@ class dl: self.tmdb_id = anime_ids.tmdb_id if not self.imdb_id and anime_ids.imdb_id: self.imdb_id = anime_ids.imdb_id + if not self.tvdb_id and anime_ids.tvdb_id: + self.tvdb_id = anime_ids.tvdb_id if self.enrich and not (self.tmdb_id or self.imdb_id or self.animeapi_title): raise click.UsageError("--enrich requires --tmdb, --imdb, or --animeapi to provide a metadata source.") @@ -1501,6 +1527,9 @@ class dl: if enrich_year and not titles.year: titles.year = enrich_year + if self.tvdb_order and isinstance(titles, Series): + titles = self.apply_tvdb_order(titles, title_cacher, cache_title_id, cache_region, cache_account_hash) + music_mode = isinstance(titles, Music) music_collection_mode = ( isinstance(titles, list) and bool(titles) and all(isinstance(title, Music) for title in titles) @@ -3622,7 +3651,7 @@ class dl: shutil.move(muxed_path, final_path) used_final_paths.add(final_path) self.completed_files.append(final_path) - tags.tag_file(final_path, title, self.tmdb_id, self.imdb_id) + tags.tag_file(final_path, title, self.tmdb_id, self.imdb_id, self.tvdb_id) title_dl_time = time_elapsed_since(dl_start_time) downloaded_label = "Track" if music_mode and isinstance(title, Song) else "Title" @@ -3648,6 +3677,79 @@ class dl: console.print(Padding(f"Processed all titles in [progress.elapsed]{dl_time}", (0, 5, 1, 5))) + def apply_tvdb_order( + self, + titles: Series, + title_cacher: Any = None, + cache_title_id: Optional[str] = None, + cache_region: Optional[str] = None, + cache_account_hash: Optional[str] = None, + ) -> Series: + """Renumber episodes from the service's own numbering into a TVDB season order.""" + tvdb = providers.get_provider("tvdb") + if not tvdb: + self.log.warning("--tvdb-order needs a tvdb_api_key in your config, skipping renumber.") + return titles + + tvdb_id = self.tvdb_id + if not tvdb_id: + show = titles[0] + result = providers.search_metadata( + show.title, show.year, "tv", title_cacher, cache_title_id, cache_region, cache_account_hash + ) + tvdb_id = result.external_ids.tvdb_id if result else None + if not tvdb_id: + self.log.warning("Could not resolve a TVDB ID for %r, skipping %s renumber.", str(titles), self.tvdb_order) + return titles + self.tvdb_id = tvdb_id # keep the mux tag on the series the renumbering used + + keys = [(t.season, t.number) for t in titles] + source_order = tvdb.detect_order(tvdb_id, keys) # type: ignore[attr-defined] + if source_order == self.tvdb_order: + self.log.info("%s already lists this series in TVDB %s order.", self.service, self.tvdb_order) + return titles + + mapping = tvdb.get_order_map(tvdb_id, self.tvdb_order, source_order) # type: ignore[attr-defined] + if not mapping: + return titles + + # an unmapped episode keeps its own numbering and can land on a remapped episode's slot + renumbered = [mapping.get(key, (*key, None))[:2] for key in keys] + unmapped = sum(1 for key in keys if key not in mapping) + collisions = [slot for slot, count in Counter(renumbered).items() if count > 1] + if collisions: + self.log.error( + "TVDB's %s order does not cover %d of this series' episodes, and renumbering the rest " + "would give %d season/episode slot(s) two episodes each. Keeping the %s numbering " + "%s uses. Pick an order that covers the whole series.", + self.tvdb_order, + unmapped, + len(collisions), + source_order, + self.service, + ) + return titles + + for title in titles: + entry = mapping.get((title.season, title.number)) + if not entry: + continue + title.season, title.number, tvdb_name = entry + if tvdb_name and not title.name: + title.name = tvdb_name + + if unmapped: + self.log.warning( + "%d episode(s) are not in TVDB's %s order and keep their %s numbering.", + unmapped, + self.tvdb_order, + source_order, + ) + self.log.info("Renumbered episodes from TVDB %s to %s order (ID %s).", source_order, self.tvdb_order, tvdb_id) + + # Series is a SortedKeyList; renumbering in place does not re-sort it + return type(titles)(list(titles)) + @staticmethod def title_to_meta(title: Title_T) -> dict[str, Any]: """Capture the title fields needed to rebuild a Title for output naming on import.""" diff --git a/unshackle/core/api/download_manager.py b/unshackle/core/api/download_manager.py index 3e7d3bdf..180ed7e5 100644 --- a/unshackle/core/api/download_manager.py +++ b/unshackle/core/api/download_manager.py @@ -417,6 +417,8 @@ def _perform_download( "tag": params.get("tag"), "tmdb_id": params.get("tmdb_id"), "imdb_id": params.get("imdb_id"), + "tvdb_id": params.get("tvdb_id"), + "tvdb_order": params.get("tvdb_order"), "animeapi_id": params.get("animeapi_id"), "enrich": params.get("enrich", False), "output_dir": Path(params["output_dir"]) if params.get("output_dir") else None, @@ -447,6 +449,8 @@ def _perform_download( tag=params.get("tag"), tmdb_id=params.get("tmdb_id"), imdb_id=params.get("imdb_id"), + tvdb_id=params.get("tvdb_id"), + tvdb_order=params.get("tvdb_order"), animeapi_id=params.get("animeapi_id"), enrich=params.get("enrich", False), output_dir=Path(params["output_dir"]) if params.get("output_dir") else None, diff --git a/unshackle/core/api/handlers.py b/unshackle/core/api/handlers.py index 47afc898..f22d9065 100644 --- a/unshackle/core/api/handlers.py +++ b/unshackle/core/api/handlers.py @@ -74,6 +74,8 @@ DEFAULT_DOWNLOAD_PARAMS = { "tag": None, "tmdb_id": None, "imdb_id": None, + "tvdb_id": None, + "tvdb_order": None, "animeapi_id": None, "enrich": False, "output_dir": None, diff --git a/unshackle/core/api/routes.py b/unshackle/core/api/routes.py index 8f49811d..21fffadd 100644 --- a/unshackle/core/api/routes.py +++ b/unshackle/core/api/routes.py @@ -787,6 +787,13 @@ async def download(request: web.Request) -> web.Response: imdb_id: type: string description: Use this IMDB ID (e.g. tt1375666) for tagging (default - None) + tvdb_id: + type: integer + description: Use this TVDB ID for tagging and episode ordering (default - None) + tvdb_order: + type: string + enum: [official, dvd, absolute, alternate, regional] + description: Renumber episodes to a TVDB season order (default - the tvdb_order config option) output_dir: type: string description: Override the output directory for this download (default - None) diff --git a/unshackle/core/config.py b/unshackle/core/config.py index 92ee1b97..b543a1e4 100644 --- a/unshackle/core/config.py +++ b/unshackle/core/config.py @@ -117,6 +117,10 @@ class Config: self.imdb_api_enabled: bool = kwargs.get("imdb_api_enabled", False) self.omdb_api_key: str = kwargs.get("omdb_api_key") or "" self.tmdb_api_key: str = kwargs.get("tmdb_api_key") or "" + self.tvdb_api_key: str = kwargs.get("tvdb_api_key") or "" + self.tvdb_pin: str = kwargs.get("tvdb_pin") or "" + self.tvdb_order: str = (kwargs.get("tvdb_order") or "").lower() + self.metadata_providers: list = kwargs.get("metadata_providers") or [] self.simkl_client_id: str = kwargs.get("simkl_client_id") or "" self.decrypt_labs_api_key: str = kwargs.get("decrypt_labs_api_key") or "" self.ipinfo_api_key: str = kwargs.get("ipinfo_api_key") or "" diff --git a/unshackle/core/providers/__init__.py b/unshackle/core/providers/__init__.py index 85bdc667..157196bb 100644 --- a/unshackle/core/providers/__init__.py +++ b/unshackle/core/providers/__init__.py @@ -9,26 +9,42 @@ from unshackle.core.providers.imdbapi import IMDBApiProvider from unshackle.core.providers.omdb import OMDBProvider from unshackle.core.providers.simkl import SimklProvider from unshackle.core.providers.tmdb import TMDBProvider +from unshackle.core.providers.tvdb import TVDBProvider if TYPE_CHECKING: from unshackle.core.title_cacher import TitleCacher -# Ordered by priority: IMDBApi (free), OMDb, SIMKL, TMDB -ALL_PROVIDERS: list[type[MetadataProvider]] = [IMDBApiProvider, OMDBProvider, SimklProvider, TMDBProvider] +REGISTRY: dict[str, type[MetadataProvider]] = { + cls.NAME: cls for cls in (IMDBApiProvider, OMDBProvider, SimklProvider, TMDBProvider, TVDBProvider) +} + +# used when `metadata_providers` is unset +DEFAULT_ORDER: tuple[str, ...] = ("imdbapi", "omdb", "simkl", "tmdb", "tvdb") + + +def provider_order() -> list[type[MetadataProvider]]: + """Provider classes in the order configured by `metadata_providers`.""" + from unshackle.core.config import config + + names = [str(n).lower() for n in (config.metadata_providers or DEFAULT_ORDER)] + unknown = [n for n in names if n not in REGISTRY] + if unknown: + log.warning("Ignoring unknown metadata_providers entries: %s", ", ".join(unknown)) + return [REGISTRY[n] for n in dict.fromkeys(names) if n in REGISTRY] def get_available_providers() -> list[MetadataProvider]: """Return instantiated providers that have valid credentials.""" - return [cls() for cls in ALL_PROVIDERS if cls().is_available()] + return [cls() for cls in provider_order() if cls().is_available()] def get_provider(name: str) -> Optional[MetadataProvider]: """Get a specific provider by name.""" - for cls in ALL_PROVIDERS: - if cls.NAME == name: - p = cls() - return p if p.is_available() else None - return None + cls = REGISTRY.get(name) + if not cls: + return None + p = cls() + return p if p.is_available() else None # -- Public API (replaces tags.py functions) -- @@ -44,9 +60,11 @@ def search_metadata( cache_account_hash: Optional[str] = None, ) -> Optional[MetadataResult]: """Search all available providers for metadata. Returns best match.""" + ordered = provider_order() + # Check cache first if title_cacher and cache_title_id: - for cls in ALL_PROVIDERS: + for cls in ordered: p = cls() if not p.is_available(): continue @@ -58,7 +76,7 @@ def search_metadata( return result # Search providers in priority order - for cls in ALL_PROVIDERS: + for cls in ordered: p = cls() if not p.is_available(): continue @@ -230,9 +248,15 @@ def fetch_external_ids( # -- Internal helpers -- -# Provider authority ranking for tie-breaking (lower index = more authoritative) -_ENRICHMENT_PROVIDERS = ("tmdb", "simkl") -_ENRICHMENT_AUTHORITY: dict[str, int] = {name: i for i, name in enumerate(_ENRICHMENT_PROVIDERS)} +# trust ranking for cross-validating enrichments; `metadata_providers` filters this +# set but sets search order only, not the ranking +_ENRICHMENT_AUTHORITY: tuple[str, ...] = ("tmdb", "simkl", "tvdb") + + +def _enrichment_providers() -> list[str]: + """Names of configured providers that can resolve an IMDB ID, most trusted first.""" + configured = {cls.NAME for cls in provider_order() if hasattr(cls, "find_by_imdb_id")} + return [name for name in _ENRICHMENT_AUTHORITY if name in configured] def enrich_ids(result: MetadataResult) -> None: @@ -251,13 +275,14 @@ def enrich_ids(result: MetadataResult) -> None: kind = result.kind or "movie" # Step 1: Collect enrichment results from all available providers + authority = {name: i for i, name in enumerate(_enrichment_providers())} enrichments: list[tuple[str, ExternalIds]] = [] - for provider_name in _ENRICHMENT_PROVIDERS: + for provider_name in authority: p = get_provider(provider_name) if not p: continue try: - enriched = p.find_by_imdb_id(ids.imdb_id, kind) # type: ignore[union-attr] + enriched = p.find_by_imdb_id(ids.imdb_id, kind) # type: ignore[attr-defined] except Exception as exc: log.debug("Enrichment via %s failed: %s", provider_name, exc) continue @@ -268,7 +293,7 @@ def enrich_ids(result: MetadataResult) -> None: return # Step 2: Cross-validate using tmdb_id as anchor — drop providers that disagree - validated = _validate_enrichments(enrichments) + validated = _validate_enrichments(enrichments, authority) # Step 3: Merge validated data (fill gaps only) for _provider_name, ext in validated: @@ -281,6 +306,7 @@ def enrich_ids(result: MetadataResult) -> None: def _validate_enrichments( enrichments: list[tuple[str, ExternalIds]], + authority: dict[str, int], ) -> list[tuple[str, ExternalIds]]: """Drop providers whose tmdb_id conflicts with the authoritative value. @@ -309,7 +335,7 @@ def _validate_enrichments( # No majority — pick the most authoritative provider best_provider = min( tmdb_votes.keys(), - key=lambda name: _ENRICHMENT_AUTHORITY.get(name, 99), + key=lambda name: authority.get(name, 99), ) anchor_tmdb_id = tmdb_votes[best_provider] @@ -409,6 +435,25 @@ def _cached_to_result(cached: dict, provider_name: str, kind: str) -> Optional[M source="omdb", raw=cached, ) + elif provider_name == "tvdb": + from unshackle.core.providers.tvdb import _ids_from_remote, _parse_int + + tvdb_id = _parse_int(cached.get("tvdb_id") or cached.get("id")) + ext = _ids_from_remote(cached.get("remote_ids") or cached.get("remoteIds"), tvdb_id) + # restore IDs that enrichment filled in beyond the raw remote_ids + enriched = cached.get("_enriched_ids", {}) + ext.imdb_id = ext.imdb_id or enriched.get("imdb_id") + ext.tmdb_id = ext.tmdb_id or enriched.get("tmdb_id") + if ext.tmdb_id: + ext.tmdb_kind = kind + return MetadataResult( + title=cached.get("name"), + year=_parse_int(cached.get("year")), + kind=kind, + external_ids=ext, + source="tvdb", + raw=cached, + ) elif provider_name == "imdbapi": title = cached.get("primaryTitle") or cached.get("originalTitle") year = cached.get("startYear") @@ -432,7 +477,8 @@ def _cached_to_result(cached: dict, provider_name: str, kind: str) -> Optional[M __all__ = [ - "ALL_PROVIDERS", + "DEFAULT_ORDER", + "REGISTRY", "ExternalIds", "MetadataProvider", "MetadataResult", @@ -443,5 +489,6 @@ __all__ = [ "get_provider", "get_title_by_id", "get_year_by_id", + "provider_order", "search_metadata", ] diff --git a/unshackle/core/providers/tvdb.py b/unshackle/core/providers/tvdb.py new file mode 100644 index 00000000..8fa7f097 --- /dev/null +++ b/unshackle/core/providers/tvdb.py @@ -0,0 +1,286 @@ +from __future__ import annotations + +from collections.abc import Collection +from difflib import SequenceMatcher +from typing import Any, Optional, Union + +import requests + +from unshackle.core.config import config +from unshackle.core.providers._base import ExternalIds, MetadataProvider, MetadataResult, _clean, _strip_year + +KIND_TO_TYPE: dict[str, str] = {"movie": "movie", "tv": "series"} +KIND_TO_PATH: dict[str, str] = {"movie": "movies", "tv": "series"} + +# a series' Streaming Order is published as "alternate" +SEASON_TYPES: tuple[str, ...] = ("official", "dvd", "absolute", "alternate", "regional") +AIRED_ORDER = "official" + +# cached per-process; providers are re-instantiated per lookup +_token: Optional[str] = None + + +def _parse_int(value: Optional[str]) -> Optional[int]: + try: + return int(value) # type: ignore[arg-type] + except (TypeError, ValueError): + return None + + +def _ids_from_remote(remote_ids: Optional[list], tvdb_id: Optional[int]) -> ExternalIds: + """Build ExternalIds from a TVDB remote_ids/remoteIds list.""" + ext = ExternalIds(tvdb_id=tvdb_id) + for entry in remote_ids or []: + source = (entry.get("sourceName") or "").lower() + value = entry.get("id") + if source == "imdb": + ext.imdb_id = value + elif source.startswith("themoviedb"): + ext.tmdb_id = _parse_int(value) + return ext + + +def _pick_match( + results: list[dict], search_title: str, year: Optional[int] +) -> tuple[Optional[dict], Optional[str], float]: + """Best title match among results, ignoring any whose year is more than a year out.""" + best: Optional[dict] = None + best_title: Optional[str] = None + best_ratio = 0.0 + for result in results: + result_year = _parse_int(result.get("year")) + if year and result_year and abs(result_year - year) > 1: + continue + for candidate in [result.get("name"), *(result.get("aliases") or [])]: + if not candidate: + continue + ratio = SequenceMatcher(None, _clean(search_title), _clean(candidate)).ratio() + if ratio > best_ratio: + best, best_title, best_ratio = result, candidate, ratio + return best, best_title, best_ratio + + +class TVDBProvider(MetadataProvider): + """TheTVDB v4 metadata provider. Native IDs are TVDB IDs.""" + + NAME = "tvdb" + REQUIRES_KEY = True + BASE_URL = "https://api4.thetvdb.com/v4" + + def __init__(self) -> None: + super().__init__() + self._episodes: dict[tuple[int, str], list[dict]] = {} + + def is_available(self) -> bool: + return bool(config.tvdb_api_key) + + def _login(self) -> Optional[str]: + global _token + payload: dict[str, str] = {"apikey": config.tvdb_api_key} + if config.tvdb_pin: + payload["pin"] = config.tvdb_pin + try: + r = self.session.post(f"{self.BASE_URL}/login", json=payload, timeout=30) + r.raise_for_status() + _token = r.json()["data"]["token"] + except (requests.RequestException, ValueError, KeyError) as exc: + self.log.warning("Failed to authenticate with TVDB: %s", exc) + _token = None + return _token + + def _get(self, path: str, params: Optional[dict] = None) -> Any: + """The `data` payload of a TVDB response: a dict for entities, a list for searches.""" + global _token + token = _token or self._login() + if not token: + return None + for _ in range(2): + try: + r = self.session.get( + f"{self.BASE_URL}{path}", + params=params, + headers={"Authorization": f"Bearer {token}"}, + timeout=30, + ) + if r.status_code == 401: + _token = None + token = self._login() + if not token: + return None + continue + r.raise_for_status() + return r.json().get("data") + except (requests.RequestException, ValueError) as exc: + self.log.debug("TVDB request %s failed: %s", path, exc) + return None + return None + + def search(self, title: str, year: Optional[int], kind: str) -> Optional[MetadataResult]: + search_title = _strip_year(title) + self.log.debug("Searching TVDB for %r (%s, %s)", search_title, kind, year) + + params: dict[str, str | int] = {"query": search_title, "limit": 10} + entity_type = KIND_TO_TYPE.get(kind) + if entity_type: + params["type"] = entity_type + if year is not None: + params["year"] = year + + results = self._get("/search", params) + if not results and year is not None: + del params["year"] + results = self._get("/search", params) + if not results: + self.log.debug("No TVDB results for %r", search_title) + return None + + # year first, so an identically named remake cannot win on title alone + best, best_title, best_ratio = _pick_match(results, search_title, year) + if best is None: + best, best_title, best_ratio = _pick_match(results, search_title, None) + if best is None: + best, best_title = results[0], results[0].get("name") + + tvdb_id = _parse_int(best.get("tvdb_id")) + ext = _ids_from_remote(best.get("remote_ids"), tvdb_id) + if ext.tmdb_id: + ext.tmdb_kind = kind + + self.log.debug("TVDB -> %r (ID %s, ratio %.2f)", best_title, tvdb_id, best_ratio) + + return MetadataResult( + title=best_title or best.get("name"), + year=_parse_int(best.get("year")), + kind=kind, + external_ids=ext, + source="tvdb", + raw=best, + ) + + def _fetch_extended(self, tvdb_id: int, kind: str) -> Optional[dict]: + path = KIND_TO_PATH.get(kind, "series") + return self._get(f"/{path}/{tvdb_id}/extended", {"short": "true"}) + + def get_by_id(self, provider_id: Union[int, str], kind: str) -> Optional[MetadataResult]: + tvdb_id = int(provider_id) + detail = self._fetch_extended(tvdb_id, kind) + if not detail: + return None + + ext = _ids_from_remote(detail.get("remoteIds"), tvdb_id) + if ext.tmdb_id: + ext.tmdb_kind = kind + + return MetadataResult( + title=detail.get("name"), + year=_parse_int(detail.get("year")), + kind=kind, + external_ids=ext, + source="tvdb", + raw=detail, + ) + + def get_external_ids(self, provider_id: Union[int, str], kind: str) -> ExternalIds: + tvdb_id = int(provider_id) + detail = self._fetch_extended(tvdb_id, kind) + if not detail: + return ExternalIds(tvdb_id=tvdb_id) + ext = _ids_from_remote(detail.get("remoteIds"), tvdb_id) + if ext.tmdb_id: + ext.tmdb_kind = kind + return ext + + def get_episodes(self, tvdb_id: int, order: str) -> list[dict]: + """All episodes of a series in the given season order.""" + cached = self._episodes.get((tvdb_id, order)) + if cached is not None: + return cached + + episodes: list[dict] = [] + for page in range(20): # ponytail: hard page cap, 500 eps/page covers any real series + data = self._get(f"/series/{tvdb_id}/episodes/{order}", {"page": page}) + if data is None and page: + # a partial listing would renumber episodes wrongly + self.log.warning("TVDB %s order listing for series %s failed at page %d", order, tvdb_id, page) + return [] + batch = (data or {}).get("episodes") or [] + episodes.extend(batch) + if len(batch) < 500: + break + + if episodes: # an empty listing may be a transient failure + self._episodes[(tvdb_id, order)] = episodes + return episodes + + def detect_order(self, tvdb_id: int, keys: Collection[tuple[int, int]]) -> str: + """Guess which TVDB season order a service's (season, episode) numbering follows. + + A service does not always number a series in aired order, so the numbering is + scored against every order rather than assumed. + """ + best, best_score = AIRED_ORDER, -1 + for order in SEASON_TYPES: + episodes = self.get_episodes(tvdb_id, order) + if not episodes: + continue + slots = {(_parse_int(e.get("seasonNumber")), _parse_int(e.get("number"))) for e in episodes} + score = sum(1 for key in keys if key in slots) + self.log.debug("TVDB order %s matches %d/%d of the service's episodes", order, score, len(keys)) + if score > best_score: + best, best_score = order, score + if order == AIRED_ORDER and score == len(keys): + break # aired order already accounts for every episode + return best + + def get_order_map( + self, tvdb_id: int, order: str, source_order: str = AIRED_ORDER + ) -> dict[tuple[int, int], tuple[int, int, Optional[str]]]: + """Map (season, episode) in `source_order` to (season, episode, name) in `order`. + + Every order lists the same episodes under the same TVDB episode ID, which is what + the two listings join on. + """ + if order == source_order: + return {} + + source = self.get_episodes(tvdb_id, source_order) + target = self.get_episodes(tvdb_id, order) + if not source or not target: + self.log.warning("TVDB has no %s order for series %s", order, tvdb_id) + return {} + + by_id = {ep.get("id"): ep for ep in target} + mapping: dict[tuple[int, int], tuple[int, int, Optional[str]]] = {} + for ep in source: + match = by_id.get(ep.get("id")) + if not match: + continue + src_season, src_number = _parse_int(ep.get("seasonNumber")), _parse_int(ep.get("number")) + dst_season, dst_number = _parse_int(match.get("seasonNumber")), _parse_int(match.get("number")) + if src_season is None or src_number is None or dst_season is None or dst_number is None: + continue + mapping[(src_season, src_number)] = (dst_season, dst_number, match.get("name")) + + self.log.debug("TVDB %s -> %s order: mapped %d/%d episodes", source_order, order, len(mapping), len(source)) + return mapping + + def find_by_imdb_id(self, imdb_id: str, kind: str) -> Optional[ExternalIds]: + """Resolve a TVDB ID from an IMDb ID via /search/remoteid.""" + results = self._get(f"/search/remoteid/{imdb_id}") + if not results: + self.log.debug("No TVDB results for IMDB ID %s", imdb_id) + return None + + # each hit wraps a single entity, e.g. {"series": {...}} or {"movie": {...}} + preferred = KIND_TO_TYPE.get(kind, "series") + entity = None + for hit in results: + entity = hit.get(preferred) or entity or next(iter(hit.values()), None) + if hit.get(preferred): + break + if not entity: + return None + + tvdb_id = _parse_int(entity.get("id")) + self.log.debug("TVDB find -> TVDB %s for IMDB %s", tvdb_id, imdb_id) + return ExternalIds(imdb_id=imdb_id, tvdb_id=tvdb_id) diff --git a/unshackle/core/utils/tags.py b/unshackle/core/utils/tags.py index c77d0597..1b92e2d5 100644 --- a/unshackle/core/utils/tags.py +++ b/unshackle/core/utils/tags.py @@ -86,6 +86,7 @@ def tag_file( title: Title, tmdb_id: Optional[int] = None, imdb_id: Optional[str] = None, + tvdb_id: Optional[int] = None, ) -> None: log.debug("Tagging file %s with title %r", path, title) custom_tags: dict[str, str] = {} @@ -118,7 +119,7 @@ def tag_file( if config.tag_imdb_tmdb: try: providers = get_available_providers() - if not providers: + if not providers and tvdb_id is None: log.debug("No metadata providers available; skipping tag lookup") apply_tags(path, custom_tags) return @@ -130,9 +131,11 @@ def tag_file( imdbapi = get_provider("imdbapi") if imdbapi: result = imdbapi.get_by_id(imdb_id, kind) - if result: - result.external_ids.imdb_id = imdb_id - enrich_ids(result) + if not result: + # IMDxAPI is off or had no answer; the ID alone still resolves the others + result = MetadataResult(title=name, year=year, kind=kind) + result.external_ids.imdb_id = imdb_id + enrich_ids(result) elif tmdb_id is not None: tmdb = get_provider("tmdb") if tmdb: @@ -140,17 +143,28 @@ def tag_file( if result: ext = tmdb.get_external_ids(tmdb_id, kind) result.external_ids = ext + enrich_ids(result) else: # Search across providers in priority order result = search_metadata(name, year, kind) - # If we got a TMDB ID from search but no full external IDs, fetch them - if result and result.external_ids.tmdb_id and not result.external_ids.imdb_id: + # If we got a TMDB ID from search but not the other IDs, fetch them + if ( + result + and result.external_ids.tmdb_id + and not (result.external_ids.imdb_id and result.external_ids.tvdb_id) + ): ext = fetch_external_ids(result.external_ids.tmdb_id, kind) - if ext.imdb_id: + if ext.imdb_id and not result.external_ids.imdb_id: result.external_ids.imdb_id = ext.imdb_id - if ext.tvdb_id: + if ext.tvdb_id and not result.external_ids.tvdb_id: result.external_ids.tvdb_id = ext.tvdb_id + enrich_ids(result) + + if tvdb_id is not None: + if result is None: + result = MetadataResult(kind=kind) + result.external_ids.tvdb_id = tvdb_id if result and result.external_ids: standard_tags = _build_tags_from_ids(result.external_ids, kind)