g0ldyy/comet
1
0
Fork 0
mirror of https://github.com/g0ldyy/comet.git synced 2026-01-12 01:16:12 +01:00
Code Issues Packages Projects Releases Wiki Activity

add option to source anime mapping from DB instead of remote fetching

Browse Source 
Signed-off-by: David Young <davidy@funkypenguin.co.nz>
This commit is contained in:
David Young
2025-12-10 12:42:11 +13:00
parent 69117996d2
commit 1b2258a374
5 changed files with 237 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env-sample
+2
View File
@@ -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 #
CHANGELOG.md
+1
View File
@@ -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)
comet/core/database.py
+27
View File
@@ -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")
comet/core/models.py
+11
View File
@@ -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:
comet/services/anime.py
+196 -33
View File
@@ -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()