Add a basic xem provider

scanner/providers/implementations/themoviedatabase.py

@@ -3,6 +3,7 @@ import logging
 from aiohttp import ClientSession
 from datetime import datetime
 from typing import Awaitable, Callable, Dict, List, Optional, Any, TypeVar
+from providers.implementations.thexem import TheXem
 
 from providers.utils import ProviderError
 
@@ -18,9 +19,10 @@ from ..types.collection import Collection, CollectionTranslation
 
 
 class TheMovieDatabase(Provider):
-	def __init__(self, client: ClientSession, api_key: str) -> None:
+	def __init__(self, client: ClientSession, api_key: str, xem: TheXem) -> None:
 		super().__init__()
 		self._client = client
+		self._xem = xem
 		self.base = "https://api.themoviedb.org/3"
 		self.api_key = api_key
 		self.genre_map = {
@@ -382,6 +384,9 @@ class TheMovieDatabase(Provider):
 		if search["original_language"] not in language:
 			language.append(search["original_language"])
 
+		if season is None:
+			season = await self._xem.get_season_override("tvdb", tvdbid, name)
+
 		if not show_id in self.absolute_episode_cache:
 			await self.get_absolute_order(show_id)
 

scanner/providers/implementations/thexem.py

@@ -0,0 +1,67 @@
+import asyncio
+import logging
+from typing import Dict, List, Literal, Tuple
+from aiohttp import ClientSession
+from datetime import timedelta, datetime
+from functools import wraps
+
+from providers.types.episode import Episode
+from providers.utils import ProviderError
+
+
+def cache(ttl: timedelta):
+	def wrap(func):
+		time, value = None, None
+
+		@wraps(func)
+		async def wrapped(*args, **kw):
+			nonlocal time
+			nonlocal value
+			now = datetime.now()
+			if not time or now - time > ttl:
+				value = await func(*args, **kw)
+				time = now
+			return value
+
+		return wrapped
+
+	return wrap
+
+
+class TheXem:
+	def __init__(self, client: ClientSession) -> None:
+		self._client = client
+		self.base = "https://thexem.info"
+
+	# TODO: make the cache support different providers and handle concurrent calls to the function.
+	@cache(ttl=timedelta(days=1))
+	async def get_map(
+		self, provider: Literal["tvdb"] | Literal["anidb"]
+	) -> Dict[str, List[Dict[str, int]]]:
+		logging.info("Fetching data from thexem for %s", provider)
+		async with self._client.get(
+			f"{self.base}/map/allNames",
+			params={
+				"origin": provider,
+				"seasonNumbers": True,
+			},
+		) as r:
+			r.raise_for_status()
+			ret = (await r.json())
+			if "data" not in ret or ret["result"] == "failure":
+				logging.error("Could not fetch xem metadata. Error: %s", ret["message"])
+				raise ProviderError("Could not fetch xem metadata")
+			return ret["data"]
+
+	async def get_season_override(
+		self, provider: Literal["tvdb"] | Literal["anidb"], id: str, show_name: str
+	):
+		map = await self.get_map(provider)
+		if id not in map:
+			return None
+		for x in map[id]:
+			[(name, season)] = x.items()
+			# TODO: replace .lower() with something a bit smarter
+			if show_name.lower() == name.lower():
+				return season
+		return None

scanner/providers/provider.py

@@ -17,13 +17,15 @@ Self = TypeVar("Self", bound="Provider")
 class Provider:
 	@classmethod
 	def get_all(cls: type[Self], client: ClientSession) -> list[Self]:
-		from providers.implementations.themoviedatabase import TheMovieDatabase
-
 		providers = []
 
+		from providers.implementations.thexem import TheXem
+		xem = TheXem(client)
+
+		from providers.implementations.themoviedatabase import TheMovieDatabase
 		tmdb = os.environ.get("THEMOVIEDB_APIKEY")
 		if tmdb:
-			providers.append(TheMovieDatabase(client, tmdb))
+			providers.append(TheMovieDatabase(client, tmdb, xem))
 
 		if not any(providers):
 			raise ProviderError(