Map anidb/tvdb season/episode numbers

docker-compose.dev.yml

@@ -140,7 +140,7 @@ services:
           path: ./scanner
           target: /app
         - action: rebuild
-          path: ./scanner/pyproject.toml
+          path: ./scanner/uv.lock
 
   transcoder:
     <<: *transcoder-base

scanner/scanner/client.py

@@ -54,6 +54,8 @@ class KyooClient(metaclass=Singleton):
 			return VideoInfo(**await r.json())
 
 	async def create_videos(self, videos: list[Video]) -> list[VideoCreated]:
+		if len(videos) == 0:
+			return []
 		async with self._client.post(
 			"videos",
 			data=TypeAdapter(list[Video]).dump_json(videos, by_alias=True),

scanner/scanner/identifiers/anilist.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import re
 import unicodedata
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from datetime import datetime, timedelta
 from functools import cached_property
 from logging import getLogger
@@ -10,10 +10,10 @@ from typing import Literal
 
 from aiohttp import ClientSession
 from pydantic import field_validator
-from pydantic_xml import BaseXmlModel, attr, element
+from pydantic_xml import BaseXmlModel, attr, element, wrapped
 
 from ..cache import cache
-from ..models.metadataid import EpisodeId, MetadataId
+from ..models.metadataid import EpisodeId
 from ..models.videos import Guess
 from ..providers.names import ProviderName
 
@@ -53,53 +53,56 @@ class AnimeListDb(BaseXmlModel, tag="anime-list"):
 		tvdbid: str | None = attr(default=None)
 		defaulttvdbseason: int | Literal["a"] | None = attr(default=None)
 		episodeoffset: int = attr(default=0)
+		tmdbtv: str | None = attr(default=None)
 		tmdbid: str | None = attr(default=None)
 		imdbid: str | None = attr(default=None)
 		name: str | None = element(default=None)
-		mapping_list: MappingList | None = element(default=[])
+		mappings: list[EpisodeMapping] = wrapped(
+			"mapping-list/mappings", element(default=[])
+		)
 
-		@field_validator("tmdbid", "imdbid")
+		@field_validator("tvdbid", "tmdbtv", "tmdbid", "imdbid", "defaulttvdbseason")
 		@classmethod
 		def _empty_to_none(cls, v: str | None) -> str | None:
+			# pornographic titles have this id.
+			if v == "hentai":
+				return None
 			return v or None
 
-		class MappingList(BaseXmlModel, tag="mapping-list"):
-			mappings: list[EpisodeMapping] = element(default=[])
+		class EpisodeMapping(BaseXmlModel):
+			anidbseason: int = attr()
+			tvdbseason: int | None = attr(default=None)
+			start: int | None = attr(default=None)
+			end: int | None = attr(default=None)
+			offset: int = attr(default=0)
+			text: str | None = None
 
-			class EpisodeMapping(BaseXmlModel):
-				anidbseason: int = attr()
-				tvdbseason: int | None = attr(default=None)
-				start: int | None = attr(default=None)
-				end: int | None = attr(default=None)
-				offset: int = attr(default=0)
-				text: str | None = None
-
-				@cached_property
-				def tvdb_mappings(self) -> dict[int, list[int]]:
-					if self.tvdbseason is None or not self.text:
-						return {}
-					ret = {}
-					for map in self.text.split(";"):
-						map = map.strip()
-						if not map or "-" not in map:
-							continue
-						[aid, tvdbids] = map.split("-", 1)
-						try:
-							ret[int(aid.strip())] = [
-								int(x.strip()) for x in tvdbids.split("+")
-							]
-						except ValueError:
-							continue
-					return ret
+			@cached_property
+			def tvdb_mappings(self) -> dict[int, list[int]]:
+				if self.tvdbseason is None or not self.text:
+					return {}
+				ret = {}
+				for map in self.text.split(";"):
+					map = map.strip()
+					if not map or "-" not in map:
+						continue
+					[aid, tvdbids] = map.split("-", 1)
+					try:
+						ret[int(aid.strip())] = [
+							int(x.strip()) for x in tvdbids.split("+")
+						]
+					except ValueError:
+						continue
+				return ret
 
 
 @dataclass
 class AnimeListData:
 	fetched_at: datetime
 	# normalized title -> anidbid
-	titles: dict[str, str] = {}
+	titles: dict[str, str] = field(default_factory=dict)
 	# anidbid -> AnimeEntry
-	animes: dict[str, AnimeListDb.AnimeEntry] = {}
+	animes: dict[str, AnimeListDb.AnimeEntry] = field(default_factory=dict)
 
 
 @cache(ttl=timedelta(days=30))
@@ -136,6 +139,62 @@ def normalize_title(title: str) -> str:
 	return title
 
 
+def anidb_to_tvdb(
+	anime: AnimeListDb.AnimeEntry,
+	anidb_ep: int,
+) -> tuple[int | None, list[int]]:
+	for map in anime.mappings:
+		if map.anidbseason != 1 or map.tvdbseason is None:
+			continue
+
+		# Handle mapping overrides (;anidb-tvdb; format)
+		if anidb_ep in map.tvdb_mappings:
+			tvdb_eps = map.tvdb_mappings[anidb_ep]
+			# Mapped to 0 means no TVDB equivalent
+			if tvdb_eps[0] == 0:
+				return (None, [])
+			return (map.tvdbseason, tvdb_eps)
+
+		# Check start/end range with offset
+		if (
+			map.start is not None
+			and map.end is not None
+			and map.start <= anidb_ep <= map.end
+		):
+			return (map.tvdbseason, [anidb_ep + map.offset])
+
+	if anime.defaulttvdbseason == "a":
+		return (None, [anidb_ep])
+	return (anime.defaulttvdbseason, [anidb_ep + anime.episodeoffset])
+
+
+def tvdb_to_anidb(
+	anime: AnimeListDb.AnimeEntry,
+	tvdb_season: int | None,
+	tvdb_ep: int,
+) -> list[int]:
+	for map in anime.mappings:
+		if map.anidbseason != 1 or map.tvdbseason != tvdb_season:
+			continue
+
+		# Handle mapping overrides (;anidb-tvdb; format)
+		overrides = [
+			anidb_num
+			for anidb_num, tvdb_nums in map.tvdb_mappings.items()
+			if tvdb_ep in tvdb_nums
+		]
+		if len(overrides):
+			return overrides
+
+		# Reverse the start/end range offset
+		if map.start is not None and map.end is not None:
+			candidate = tvdb_ep - map.offset
+			if map.start <= candidate <= map.end:
+				return [candidate]
+
+	return [tvdb_ep - anime.episodeoffset]
+
+
 async def anilist(_path: str, guess: Guess) -> Guess:
 	data = await get_data()
 
@@ -144,7 +203,6 @@ async def anilist(_path: str, guess: Guess) -> Guess:
 		return guess
 	anime = data.animes.get(aid)
 	if anime is None:
-		logger.warning("AniDB id %s found in titles but not in anime-list.xml", aid)
 		return guess
 
 	logger.info(
@@ -159,15 +217,64 @@ async def anilist(_path: str, guess: Guess) -> Guess:
 	new_external_id[ProviderName.ANIDB] = aid
 	if anime.tvdbid:
 		new_external_id[ProviderName.TVDB] = anime.tvdbid
-	if anime.tmdbid:
+	# tmdbtv is for TV series, tmdbid is for standalone movies
+	if anime.tmdbtv:
+		new_external_id[ProviderName.TMDB] = anime.tmdbtv
+	elif anime.tmdbid:
 		new_external_id[ProviderName.TMDB] = anime.tmdbid
 	if anime.imdbid:
 		new_external_id[ProviderName.IMDB] = anime.imdbid
 
 	new_episodes: list[Guess.Episode] = []
 	for ep in guess.episodes:
-		# TODO: implement this
-		...
+		if anime.defaulttvdbseason is None or anime.tvdbid is None:
+			new_episodes.append(
+				Guess.Episode(
+					season=ep.season,
+					episode=ep.episode,
+					external_id={
+						ProviderName.ANIDB: EpisodeId(
+							serie_id=aid,
+							season=None,
+							episode=ep.episode,
+						),
+					},
+				)
+			)
+			continue
+
+		# guess numbers are anidb-relative if defaulttvdbseason != 1 because
+		# the title already contains season information.
+		tvdb_season, tvdb_eps = (
+			(ep.season if ep.season is not None else 1, [ep.episode])
+			if anime.defaulttvdbseason == 1
+			else anidb_to_tvdb(anime, ep.episode)
+		)
+		anidb_eps = (
+			tvdb_to_anidb(anime, tvdb_season, ep.episode)
+			if anime.defaulttvdbseason == 1
+			else [ep.episode]
+		)
+
+		new_episodes += [
+			Guess.Episode(
+				season=tvdb_season,
+				episode=tvdb_ep,
+				external_id={
+					ProviderName.TVDB: EpisodeId(
+						serie_id=anime.tvdbid,
+						season=tvdb_season,
+						episode=tvdb_ep,
+					),
+					ProviderName.ANIDB: EpisodeId(
+						serie_id=aid,
+						season=None,
+						episode=anidb_ep,
+					),
+				},
+			)
+			for tvdb_ep, anidb_ep in zip(tvdb_eps, anidb_eps)
+		]
 
 	return Guess(
 		title=guess.title,