diff --git a/.env-sample b/.env-sample index dfd0045..bce1610 100644 --- a/.env-sample +++ b/.env-sample @@ -59,6 +59,8 @@ BACKGROUND_SCRAPER_CONCURRENT_WORKERS=1 # Number of concurrent workers for scrap BACKGROUND_SCRAPER_INTERVAL=3600 # Interval between scraping cycles in seconds BACKGROUND_SCRAPER_MAX_MOVIES_PER_RUN=100 # Maximum number of movies to scrape per run BACKGROUND_SCRAPER_MAX_SERIES_PER_RUN=100 # Maximum number of series to scrape per run +ANIME_MAPPING_SOURCE=remote # Options: remote, database - remote downloads on startup; database reads cached table +ANIME_MAPPING_REFRESH_INTERVAL=86400 # Seconds between background anime mapping refreshes when using database cache (<=0 disables) # ============================== # # Proxy Configuration # diff --git a/CHANGELOG.md b/CHANGELOG.md index 01e96a9..715cada 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ ### Features * add optional PostgreSQL read replica routing with transparent primary fallback +* add optional database-backed anime mapping cache with configurable refresh interval * add `DATABASE_STARTUP_CLEANUP_INTERVAL` to throttle heavy startup cleanup sweeps across workers ## [2.31.0](https://github.com/g0ldyy/comet/compare/v2.30.0...v2.31.0) (2025-12-08) diff --git a/comet/core/database.py b/comet/core/database.py index 874237e..b2bb828 100644 --- a/comet/core/database.py +++ b/comet/core/database.py @@ -341,6 +341,26 @@ async def setup_database(): """ ) + await database.execute( + """ + CREATE TABLE IF NOT EXISTS anime_mapping_cache ( + kitsu_id TEXT PRIMARY KEY, + imdb_id TEXT, + is_anime BOOLEAN, + updated_at INTEGER + ) + """ + ) + + await database.execute( + """ + CREATE TABLE IF NOT EXISTS anime_mapping_state ( + id INTEGER PRIMARY KEY CHECK (id = 1), + refreshed_at INTEGER + ) + """ + ) + # ============================================================================= # TORRENTS TABLE INDEXES - Most critical for performance # ============================================================================= @@ -590,6 +610,13 @@ async def setup_database(): """ ) + await database.execute( + """ + CREATE INDEX IF NOT EXISTS idx_anime_mapping_imdb + ON anime_mapping_cache (imdb_id) + """ + ) + if settings.DATABASE_TYPE == "sqlite": await database.execute("PRAGMA busy_timeout=30000") # 30 seconds timeout await database.execute("PRAGMA journal_mode=WAL") diff --git a/comet/core/models.py b/comet/core/models.py index 25fd2a8..d092704 100644 --- a/comet/core/models.py +++ b/comet/core/models.py @@ -105,6 +105,8 @@ class AppSettings(BaseSettings): BACKGROUND_SCRAPER_INTERVAL: Optional[int] = 3600 BACKGROUND_SCRAPER_MAX_MOVIES_PER_RUN: Optional[int] = 100 BACKGROUND_SCRAPER_MAX_SERIES_PER_RUN: Optional[int] = 100 + ANIME_MAPPING_SOURCE: Optional[str] = "remote" + ANIME_MAPPING_REFRESH_INTERVAL: Optional[int] = 86400 @field_validator("INDEXER_MANAGER_TYPE") def set_indexer_manager_type(cls, v, values): @@ -112,6 +114,15 @@ class AppSettings(BaseSettings): return None return v + @field_validator("ANIME_MAPPING_SOURCE") + def normalize_anime_mapping_source(cls, v): + if not v: + return "remote" + normalized = v.strip().lower() + if normalized not in {"remote", "database"}: + raise ValueError("ANIME_MAPPING_SOURCE must be 'remote' or 'database'") + return normalized + @field_validator("DATABASE_TYPE", mode="before") def normalize_database_type(cls, v): if v is None: diff --git a/comet/services/anime.py b/comet/services/anime.py index 51eee93..28e1dc1 100644 --- a/comet/services/anime.py +++ b/comet/services/anime.py @@ -1,7 +1,12 @@ +import asyncio +import time +from collections.abc import Mapping + import aiohttp import orjson from comet.core.logger import logger +from comet.core.models import database, settings class AnimeMapper: @@ -10,44 +15,28 @@ class AnimeMapper: self.imdb_to_kitsu = {} self.anime_imdb_ids = set() self.loaded = False + self._refresh_lock = asyncio.Lock() + self._background_task = None - async def load_anime_mapping(self, session: aiohttp.ClientSession): - try: - url = "https://raw.githubusercontent.com/Fribb/anime-lists/refs/heads/master/anime-list-full.json" - response = await session.get(url) - - if response.status != 200: - logger.error(f"Failed to load anime mapping: HTTP {response.status}") - return False - - text = await response.text() - data = orjson.loads(text) - - kitsu_count = 0 - imdb_count = 0 - - for entry in data: - kitsu_id = entry.get("kitsu_id") - imdb_id = entry.get("imdb_id") - - if kitsu_id and imdb_id: - self.kitsu_to_imdb[kitsu_id] = imdb_id - self.imdb_to_kitsu[imdb_id] = kitsu_id - self.anime_imdb_ids.add(imdb_id) - imdb_count += 1 - - if kitsu_id: - kitsu_count += 1 - - self.loaded = True + async def load_anime_mapping(self, session: aiohttp.ClientSession | None = None): + if self.loaded: logger.log( "COMET", - f"✅ Anime mapping loaded: {kitsu_count} Kitsu entries, {imdb_count} with IMDB IDs", + "Anime mapping already loaded in this process; skipping reload", ) return True - except Exception as e: - logger.error(f"Exception while loading anime mapping: {e}") - return False + + source = (settings.ANIME_MAPPING_SOURCE or "remote").lower() + + if source == "database": + loaded = await self._load_from_database() + if loaded: + if await self._is_cache_stale(): + asyncio.create_task(self._refresh_from_remote(background=True)) + self._ensure_periodic_refresh() + return True + + return await self._refresh_from_remote(session) def get_imdb_from_kitsu(self, kitsu_id: int): return self.kitsu_to_imdb.get(kitsu_id) @@ -70,5 +59,179 @@ class AnimeMapper: def is_loaded(self): return self.loaded + async def _load_from_database(self): + try: + rows = await database.fetch_all( + "SELECT kitsu_id, imdb_id FROM anime_mapping_cache" + ) + + if not rows: + return False + + self._populate_from_rows(rows) + logger.log( + "COMET", + f"✅ Anime mapping loaded from database: {len(rows)} cached entries", + ) + return True + except Exception as exc: + logger.error(f"Failed to load anime mapping from database: {exc}") + return False + + async def _is_cache_stale(self): + interval = settings.ANIME_MAPPING_REFRESH_INTERVAL or 0 + if interval <= 0: + return False + + row = await database.fetch_one( + "SELECT refreshed_at FROM anime_mapping_state WHERE id = 1", + force_primary=True, + ) + + if not row or row.get("refreshed_at") is None: + return True + + last_refresh = float(row["refreshed_at"]) + return (time.time() - last_refresh) >= interval + + def _ensure_periodic_refresh(self): + interval = settings.ANIME_MAPPING_REFRESH_INTERVAL or 0 + if interval <= 0: + return + + if self._background_task and not self._background_task.done(): + return + + self._background_task = asyncio.create_task( + self._refresh_loop(interval) + ) + + async def _refresh_from_remote( + self, + session: aiohttp.ClientSession | None = None, + *, + background: bool = False, + ): + async with self._refresh_lock: + if self.loaded and background: + return True + + own_session = False + if session is None: + own_session = True + session = aiohttp.ClientSession() + + try: + url = "https://raw.githubusercontent.com/Fribb/anime-lists/refs/heads/master/anime-list-full.json" + response = await session.get(url) + + if response.status != 200: + logger.error( + f"Failed to load anime mapping: HTTP {response.status}" + ) + return False + + text = await response.text() + data = orjson.loads(text) + + self._populate_from_rows(data) + logger.log( + "COMET", + f"✅ Anime mapping loaded: {len(self.kitsu_to_imdb)} Kitsu entries, {len(self.imdb_to_kitsu)} with IMDB IDs", + ) + + if settings.ANIME_MAPPING_SOURCE == "database": + await self._persist_mapping(data) + self._ensure_periodic_refresh() + + return True + except Exception as exc: + log_fn = logger.warning if background else logger.error + log_fn(f"Exception while loading anime mapping: {exc}") + return False + finally: + if own_session and session: + await session.close() + + def _populate_from_rows(self, rows): + self.kitsu_to_imdb.clear() + self.imdb_to_kitsu.clear() + self.anime_imdb_ids.clear() + + for entry in rows: + kitsu_id = self._entry_value(entry, "kitsu_id") + imdb_id = self._entry_value(entry, "imdb_id") + + if kitsu_id and imdb_id: + kitsu_id_str = str(kitsu_id) + self.kitsu_to_imdb[kitsu_id_str] = imdb_id + self.imdb_to_kitsu[imdb_id] = kitsu_id_str + self.anime_imdb_ids.add(imdb_id) + + self.loaded = True + + async def _persist_mapping(self, rows): + timestamp = time.time() + params = [] + for entry in rows: + kitsu_id = self._entry_value(entry, "kitsu_id") + if not kitsu_id: + continue + + params.append( + { + "kitsu_id": str(kitsu_id), + "imdb_id": self._entry_value(entry, "imdb_id"), + "is_anime": True, + "updated_at": timestamp, + } + ) + + insert_query = ( + "INSERT INTO anime_mapping_cache (kitsu_id, imdb_id, is_anime, updated_at) " + "VALUES (:kitsu_id, :imdb_id, :is_anime, :updated_at)" + ) + + chunk_size = 500 + + try: + async with database.transaction(): + await database.execute("DELETE FROM anime_mapping_cache") + for idx in range(0, len(params), chunk_size): + await database.execute_many( + insert_query, + params[idx : idx + chunk_size], + ) + await database.execute( + """ + INSERT INTO anime_mapping_state (id, refreshed_at) + VALUES (1, :timestamp) + ON CONFLICT (id) DO UPDATE SET refreshed_at = :timestamp + """, + {"timestamp": timestamp}, + ) + logger.log( + "DATABASE", + f"Anime mapping cache updated ({len(params)} rows)", + ) + except Exception as exc: + logger.error(f"Failed to persist anime mapping cache: {exc}") + + @staticmethod + def _entry_value(entry, key): + if isinstance(entry, Mapping): + return entry.get(key) + return entry[key] + + async def _refresh_loop(self, interval: int): + try: + while True: + await asyncio.sleep(interval) + await self._refresh_from_remote(background=True) + except asyncio.CancelledError: + raise + except Exception as exc: + logger.warning(f"Anime mapping refresh loop encountered an error: {exc}") + anime_mapper = AnimeMapper()