diff --git a/.env-sample b/.env-sample index add83db..83aa61f 100644 --- a/.env-sample +++ b/.env-sample @@ -92,8 +92,10 @@ CATALOG_TIMEOUT=30 # Max time to fetch catalog pages (seconds) # ============================== # # Anime Mapping Configuration # # ============================== # -ANIME_MAPPING_SOURCE=database # 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) +# Anime Mapping is used to match IMDb/Kitsu IDs to Anime specific IDs (AniList, MAL, Kitsu, etc.) to retrieve title aliases. +# This greatly improves title matching performances for anime content by using accurate alternative titles. +ANIME_MAPPING_ENABLED=True +ANIME_MAPPING_REFRESH_INTERVAL=432000 # Seconds between background anime mapping refreshes when using database cache (<=0 disables) # ============================== # # Networking & Proxy Configuration # @@ -140,7 +142,7 @@ MAGNET_RESOLVE_TIMEOUT=60 # Max time to resolve a magnet link (seconds) # Multi-Instance Scraping Support: # - Single URL: Use a simple string for one instance (default behavior) # - Multiple URLs: Use JSON array format for multiple instances -# - Example single: COMET_URL=https://comet.elfhosted.com +# - Example single: COMET_URL=https://comet.feels.legal # - Example multi: COMET_URL='["https://comet1.example.com", "https://comet2.example.com"]' # # Scraper Context Modes: @@ -159,7 +161,7 @@ MAGNET_RESOLVE_TIMEOUT=60 # Max time to resolve a magnet link (seconds) # SCRAPE_PROWLARR=background # Prowlarr for background scraping only SCRAPE_COMET=False -COMET_URL=https://comet.elfhosted.com +COMET_URL=https://comet.feels.legal # Multi-instance example: # COMET_URL='["https://comet1.example.com", "https://comet2.example.com"]' @@ -242,7 +244,7 @@ PROXY_DEBRID_STREAM_INACTIVITY_THRESHOLD=300 # Seconds to wait before cleaning u DISABLE_TORRENT_STREAMS=False # When true, torrent-only requests return a friendly message instead of magnets TORRENT_DISABLED_STREAM_NAME=[INFO] Comet # Stremio stream name shown when torrents are disabled TORRENT_DISABLED_STREAM_DESCRIPTION=Direct torrent playback is disabled on this server. # Description shown to users in Stremio -TORRENT_DISABLED_STREAM_URL=https://comet.looks.legal # Optional URL included in the placeholder stream response +TORRENT_DISABLED_STREAM_URL=https://comet.feels.legal # Optional URL included in the placeholder stream response # ============================== # # Content Filtering # @@ -261,3 +263,30 @@ CUSTOM_HEADER_HTML=None # StremThru Integration # # ============================== # STREMTHRU_URL=https://stremthru.13377001.xyz # needed to use debrid services + +# ============================== # +# HTTP Cache Settings # +# ============================== # +# These settings control HTTP-level caching. +# Comet sends proper Cache-Control headers that can be respected by proxies. +# +# Key cacheable endpoints: +# - /stream/{type}/{id}.json (PUBLIC - shared cache, no user config) +# - /{config}/stream/... (PRIVATE - user-specific, browser only) +# - /configure (PUBLIC/PRIVATE - depends on CUSTOM_HEADER_HTML) + +HTTP_CACHE_ENABLED=False # Master switch for HTTP cache headers + +# PUBLIC streams TTL (for /stream/movie/tt1234567.json without user config) +# These can be cached at edge and shared between all users. +# Higher = less origin traffic, lower = fresher results +HTTP_CACHE_PUBLIC_STREAMS_TTL=300 # 5 minutes (recommended: 120-600) + +# PRIVATE streams TTL (for /{config}/stream/... with user config) +# Only cached in user's browser (private cache). +# These contain user-specific debrid availability info. +HTTP_CACHE_PRIVATE_STREAMS_TTL=60 # 1 minute (recommended: 30-120) + +# Stale-While-Revalidate: serve stale content while fetching fresh in background +# Improves perceived performance - users get instant response with cached data +HTTP_CACHE_STALE_WHILE_REVALIDATE=60 # 1 minute diff --git a/comet/api/app.py b/comet/api/app.py index 1e885c6..bc4b245 100644 --- a/comet/api/app.py +++ b/comet/api/app.py @@ -104,7 +104,6 @@ async def lifespan(app: FastAPI): if settings.PROXY_DEBRID_STREAM: await bandwidth_monitor.shutdown() - await anime_mapper.stop() await add_torrent_queue.stop() await torrent_update_queue.stop() diff --git a/comet/api/endpoints/chilllink.py b/comet/api/endpoints/chilllink.py index 443f7e3..bd0c852 100644 --- a/comet/api/endpoints/chilllink.py +++ b/comet/api/endpoints/chilllink.py @@ -70,7 +70,7 @@ async def chilllink_streams( { "id": "comet.fast", "title": "You need to configure a debrid service to use Comet in Chillio.", - "url": "https://comet.looks.legal", + "url": "https://comet.feels.legal", "metadata": [], } ] diff --git a/comet/api/endpoints/config.py b/comet/api/endpoints/config.py index 97525de..3eba004 100644 --- a/comet/api/endpoints/config.py +++ b/comet/api/endpoints/config.py @@ -2,6 +2,7 @@ from fastapi import APIRouter, Request from fastapi.templating import Jinja2Templates from comet.core.models import settings, web_config +from comet.utils.cache import CachePolicies router = APIRouter() templates = Jinja2Templates("comet/templates") @@ -20,7 +21,7 @@ templates = Jinja2Templates("comet/templates") description="Renders the configuration page with existing configuration.", ) async def configure(request: Request): - return templates.TemplateResponse( + response = templates.TemplateResponse( "index.html", { "request": request, @@ -32,3 +33,9 @@ async def configure(request: Request): "disableTorrentStreams": settings.DISABLE_TORRENT_STREAMS, }, ) + + if settings.HTTP_CACHE_ENABLED: + response.headers["Cache-Control"] = CachePolicies.configure_page().build() + response.headers["Vary"] = "Accept, Accept-Encoding" + + return response diff --git a/comet/api/endpoints/manifest.py b/comet/api/endpoints/manifest.py index d669a32..5a0e3b6 100644 --- a/comet/api/endpoints/manifest.py +++ b/comet/api/endpoints/manifest.py @@ -6,6 +6,9 @@ from fastapi import APIRouter, Request from comet.core.config_validation import config_check from comet.core.models import settings from comet.debrid.manager import get_debrid_extension +from comet.utils.cache import (CachedJSONResponse, CachePolicies, + check_etag_match, generate_etag, + not_modified_response) router = APIRouter() @@ -54,4 +57,18 @@ async def manifest(request: Request, b64config: str = None): f"{settings.ADDON_NAME}{(' | ' + debrid_extension) if debrid_extension != 'TORRENT' else ''}" ) + if settings.HTTP_CACHE_ENABLED: + base_manifest_etag_data = base_manifest.copy() + base_manifest_etag_data.pop("id", None) + etag = generate_etag(base_manifest_etag_data) + if check_etag_match(request, etag): + return not_modified_response(etag) + + return CachedJSONResponse( + content=base_manifest, + cache_control=CachePolicies.manifest(), + etag=etag, + vary=["Accept", "Accept-Encoding"], + ) + return base_manifest diff --git a/comet/api/endpoints/playback.py b/comet/api/endpoints/playback.py index 020b031..72b7d80 100644 --- a/comet/api/endpoints/playback.py +++ b/comet/api/endpoints/playback.py @@ -11,7 +11,8 @@ from comet.core.models import database, settings from comet.debrid.manager import get_debrid from comet.metadata.manager import MetadataScraper from comet.services.streaming.manager import custom_handle_stream_request -from comet.utils.network import NO_CACHE_HEADERS, get_client_ip +from comet.utils.cache import NO_CACHE_HEADERS +from comet.utils.network import get_client_ip from comet.utils.parsing import parse_optional_int router = APIRouter() diff --git a/comet/api/endpoints/stream.py b/comet/api/endpoints/stream.py index 8a36dfe..eaef725 100644 --- a/comet/api/endpoints/stream.py +++ b/comet/api/endpoints/stream.py @@ -12,9 +12,13 @@ from comet.debrid.exceptions import DebridAuthError from comet.debrid.manager import get_debrid_extension from comet.metadata.filter import release_filter from comet.metadata.manager import MetadataScraper +from comet.services.anime import anime_mapper from comet.services.debrid import DebridService from comet.services.lock import DistributedLock, is_scrape_in_progress from comet.services.orchestration import TorrentManager +from comet.utils.cache import (CachedJSONResponse, CachePolicies, + check_etag_match, generate_etag, + not_modified_response) from comet.utils.formatting import (format_chilllink, format_title, get_formatted_components) from comet.utils.network import get_client_ip @@ -23,6 +27,38 @@ from comet.utils.parsing import parse_media_id streams = APIRouter() +def _build_stream_response( + request: Request, + content: dict, + is_public: bool = False, + vary_headers: list = None, +): + if not settings.HTTP_CACHE_ENABLED: + return content + + etag = generate_etag(content) + + if check_etag_match(request, etag): + return not_modified_response(etag) + + if is_public: + cache_policy = CachePolicies.public_torrents() + vary = ["Accept", "Accept-Encoding"] + else: + cache_policy = CachePolicies.private_streams() + vary = ["Accept", "Accept-Encoding", "Authorization"] + + if vary_headers: + vary.extend(vary_headers) + + return CachedJSONResponse( + content=content, + cache_control=cache_policy, + etag=etag, + vary=list(set(vary)), + ) + + async def is_first_search(media_id: str): params = {"media_id": media_id, "timestamp": time.time()} @@ -137,27 +173,35 @@ async def stream( b64config: str = None, chilllink: bool = False, ): + is_public_request = b64config is None + if media_type not in ["movie", "series"]: - return {"streams": []} + return _build_stream_response( + request, {"streams": []}, is_public=is_public_request + ) if "tmdb:" in media_id: - return {"streams": []} + return _build_stream_response( + request, {"streams": []}, is_public=is_public_request + ) media_id = media_id.replace("imdb_id:", "") config = config_check(b64config) if not config: - return { + error_response = { "streams": [ { "name": "[❌] Comet", "description": f"⚠️ OBSOLETE CONFIGURATION, PLEASE RE-CONFIGURE ON {request.url.scheme}://{request.url.netloc} ⚠️", - "url": "https://comet.looks.legal", + "url": "https://comet.feels.legal", } ] } + return error_response - if settings.DISABLE_TORRENT_STREAMS and config["debridService"] == "torrent": + is_torrent = config["debridService"] == "torrent" + if settings.DISABLE_TORRENT_STREAMS and is_torrent: placeholder_stream = { "name": settings.TORRENT_DISABLED_STREAM_NAME, "description": settings.TORRENT_DISABLED_STREAM_DESCRIPTION, @@ -165,7 +209,9 @@ async def stream( if settings.TORRENT_DISABLED_STREAM_URL: placeholder_stream["url"] = settings.TORRENT_DISABLED_STREAM_URL - return {"streams": [placeholder_stream]} + return _build_stream_response( + request, {"streams": [placeholder_stream]}, is_public=is_public_request + ) connector = aiohttp.TCPConnector(limit=0) async with aiohttp.ClientSession(connector=connector) as session: @@ -181,19 +227,23 @@ async def stream( if not is_released: logger.log("FILTER", f"🚫 {media_id} is not released yet. Skipping.") - return { - "streams": [ - { - "name": "[🚫] Comet", - "description": "Content not digitally released yet.", - "url": "https://comet.looks.legal", - } - ] - } + return _build_stream_response( + request, + { + "streams": [ + { + "name": "[🚫] Comet", + "description": "Content not digitally released yet.", + "url": "https://comet.feels.legal", + } + ] + }, + is_public=is_public_request, + ) # Check if metadata is already cached - cached_metadata = await metadata_scraper.get_cached( - id, season if "kitsu" not in media_id else 1, episode + cached_metadata = await metadata_scraper.get_from_cache_by_media_id( + media_id, id, season, episode ) # Quick check for "fresh" cached torrents to decide if we need to re-scrape. @@ -239,7 +289,7 @@ async def stream( cache_is_stale = fresh_cached_count == 0 # If both metadata and fresh torrents are cached, skip lock entirely - if cached_metadata is not None and fresh_cached_count > 0: + if cached_metadata is not None and not cache_is_stale: logger.log("SCRAPER", f"🚀 Fast path: using cached data for {media_id}") metadata, aliases = cached_metadata[0], cached_metadata[1] # Variables for fast path @@ -264,14 +314,14 @@ async def stream( waited_for_other_scrape = True # After waiting, re-check cached metadata - cached_metadata = await metadata_scraper.get_cached( - id, season if "kitsu" not in media_id else 1, episode + cached_metadata = await metadata_scraper.get_from_cache_by_media_id( + media_id, id, season, episode ) if lock_acquired: # We have the lock, scrape metadata normally metadata, aliases = await metadata_scraper.fetch_metadata_and_aliases( - media_type, media_id + media_type, media_id, id, season, episode ) elif cached_metadata is not None: # Use cached metadata after waiting @@ -279,7 +329,7 @@ async def stream( else: # No cached metadata available, fallback to scraping metadata, aliases = await metadata_scraper.fetch_metadata_and_aliases( - media_type, media_id + media_type, media_id, id, season, episode ) if metadata is None: if lock_acquired and scrape_lock: @@ -290,7 +340,7 @@ async def stream( { "name": "[⚠️] Comet", "description": "Unable to get metadata.", - "url": "https://comet.looks.legal", + "url": "https://comet.feels.legal", } ] } @@ -307,7 +357,6 @@ async def stream( logger.log("SCRAPER", f"🔍 Starting search for {log_title}") - id, season, episode = parse_media_id(media_type, media_id) media_only_id = id debrid_service = config["debridService"] @@ -318,6 +367,28 @@ async def stream( get_client_ip(request), ) + is_kitsu = media_id.startswith("kitsu:") + search_episode = episode + search_season = season + + if is_kitsu and episode is not None: + kitsu_mapping = anime_mapper.get_kitsu_episode_mapping(id) + if kitsu_mapping: + from_episode = kitsu_mapping.get("from_episode") + from_season = kitsu_mapping.get("from_season") + if from_episode: + new_episode = from_episode + episode - 1 + if new_episode != episode: + search_episode = new_episode + + if from_season and from_season != season: + search_season = from_season + if search_season != season or search_episode != episode: + logger.log( + "SCRAPER", + f"📺 Multi-part anime detected (kitsu:{id}): searching for S{search_season:02d}E{search_episode:02d} instead of S{season:02d}E{episode:02d}", + ) + torrent_manager = TorrentManager( debrid_service, config["debridApiKey"], @@ -332,6 +403,9 @@ async def stream( episode, aliases, settings.REMOVE_ADULT_CONTENT and config["removeTrash"], + is_kitsu=is_kitsu, + search_episode=search_episode, + search_season=search_season, ) await torrent_manager.get_cached_torrents() @@ -342,7 +416,7 @@ async def stream( is_first = await is_first_search(media_id) has_cached_results = len(torrent_manager.torrents) > 0 - sort_mixed = config["sortCachedUncachedTogether"] + sort_mixed = is_torrent or config["sortCachedUncachedTogether"] cached_results = [] non_cached_results = [] @@ -380,7 +454,7 @@ async def stream( { "name": "[🔄] Comet", "description": "Scraping in progress by another instance, please try again in a few seconds...", - "url": "https://comet.looks.legal", + "url": "https://comet.feels.legal", } ] } @@ -396,7 +470,7 @@ async def stream( { "name": "[🔄] Comet", "description": "First search for this media - More results will be available in a few seconds...", - "url": "https://comet.looks.legal", + "url": "https://comet.feels.legal", } ) else: @@ -458,7 +532,7 @@ async def stream( { "name": "[❌] Comet", "description": e.display_message, - "url": "https://comet.looks.legal", + "url": "https://comet.feels.legal", } ] } @@ -500,7 +574,7 @@ async def stream( { "name": "[⚠️] Comet", "description": "Debrid Stream Proxy Password incorrect.\nStreams will not be proxied.", - "url": "https://comet.looks.legal", + "url": "https://comet.feels.legal", } ) @@ -517,11 +591,7 @@ async def stream( torrent = torrents[info_hash] rtn_data = torrent["parsed"] - debrid_emoji = ( - "🧲" - if debrid_service == "torrent" - else ("⚡" if torrent["cached"] else "⬇️") - ) + debrid_emoji = "🧲" if is_torrent else ("⚡" if torrent["cached"] else "⬇️") torrent_title = torrent["title"] formatted_components = get_formatted_components( @@ -548,7 +618,7 @@ async def stream( formatted_components, torrent["cached"] ) - if debrid_service == "torrent": + if is_torrent: the_stream["infoHash"] = info_hash if torrent["fileIndex"] is not None: @@ -571,6 +641,14 @@ async def stream( non_cached_results.append(the_stream) if sort_mixed: - return {"streams": cached_results} + return _build_stream_response( + request, + {"streams": cached_results}, + is_public=is_public_request, + ) - return {"streams": cached_results + non_cached_results} + return _build_stream_response( + request, + {"streams": cached_results + non_cached_results}, + is_public=is_public_request, + ) diff --git a/comet/background_scraper/worker.py b/comet/background_scraper/worker.py index 26c6268..7210668 100644 --- a/comet/background_scraper/worker.py +++ b/comet/background_scraper/worker.py @@ -328,7 +328,7 @@ class BackgroundScraperWorker: async def _scrape_movie(self, media_id: str, title: str, year: int): metadata, aliases = await self.metadata_scraper.fetch_aliases_with_metadata( - "movie", media_id, title, year + "movie", media_id, title, year, id=media_id ) manager = TorrentManager( @@ -359,7 +359,7 @@ class BackgroundScraperWorker: series_media_id = f"{media_id}:1:1" metadata, aliases = await self.metadata_scraper.fetch_aliases_with_metadata( - "series", series_media_id, title, year, year_end + "series", series_media_id, title, year, year_end, id=media_id ) for episode in episodes: diff --git a/comet/core/database.py b/comet/core/database.py index 1417a8c..306969e 100644 --- a/comet/core/database.py +++ b/comet/core/database.py @@ -336,15 +336,31 @@ async def setup_database(): await database.execute( """ - CREATE TABLE IF NOT EXISTS anime_mapping_cache ( - kitsu_id INTEGER PRIMARY KEY, - imdb_id TEXT, - is_anime BOOLEAN, - updated_at INTEGER + CREATE TABLE IF NOT EXISTS anime_entries ( + id INTEGER PRIMARY KEY, + data TEXT ) """ ) + await database.execute( + """ + CREATE TABLE IF NOT EXISTS anime_ids ( + provider TEXT, + provider_id TEXT, + entry_id INTEGER, + PRIMARY KEY (provider, provider_id) + ) + """ + ) + + await database.execute( + """ + CREATE INDEX IF NOT EXISTS idx_anime_ids_entry_provider + ON anime_ids (entry_id, provider, provider_id) + """ + ) + await database.execute( """ CREATE TABLE IF NOT EXISTS anime_mapping_state ( @@ -354,6 +370,18 @@ async def setup_database(): """ ) + await database.execute( + """ + CREATE TABLE IF NOT EXISTS kitsu_imdb_mapping ( + kitsu_id TEXT PRIMARY KEY, + imdb_id TEXT, + title TEXT, + from_season INTEGER, + from_episode INTEGER + ) + """ + ) + await database.execute( """ CREATE TABLE IF NOT EXISTS digital_release_cache ( @@ -408,6 +436,24 @@ async def setup_database(): """ ) + # Index for torrent mode: queries without debrid_service filter + # Covers: get_cached_availability when debrid_service == "torrent" + await database.execute( + """ + CREATE INDEX IF NOT EXISTS idx_debrid_info_hash + ON debrid_availability (info_hash) + """ + ) + + # Composite index for full filter path with season/episode + # Covers: get_cached_availability with season/episode filters + await database.execute( + """ + CREATE INDEX IF NOT EXISTS idx_debrid_hash_season_episode + ON debrid_availability (info_hash, season, episode, timestamp) + """ + ) + # ============================================================================= # DOWNLOAD_LINKS_CACHE TABLE INDEXES # ============================================================================= @@ -518,13 +564,7 @@ async def setup_database(): await database.execute( """ - CREATE INDEX IF NOT EXISTS idx_anime_mapping_imdb - ON anime_mapping_cache (imdb_id) - """ - ) - await database.execute( - """ CREATE INDEX IF NOT EXISTS idx_digital_release_timestamp ON digital_release_cache (timestamp) """ @@ -650,7 +690,7 @@ async def _run_startup_cleanup(): logger.error(f"Error executing startup cleanup: {e}") -async def _should_run_startup_cleanup(current_time: float, interval: int) -> bool: +async def _should_run_startup_cleanup(current_time: float, interval: int): row = await database.fetch_one( "SELECT last_startup_cleanup FROM db_maintenance WHERE id = 1", force_primary=True, @@ -721,6 +761,7 @@ async def _migrate_indexes(): "torrents_seeders_idx", "idx_first_searches_cleanup", "idx_metadata_title_search", + "idx_anime_ids_entry_id", ] dropped_count = 0 diff --git a/comet/core/db_router.py b/comet/core/db_router.py index 9a1b4f7..3a867c9 100644 --- a/comet/core/db_router.py +++ b/comet/core/db_router.py @@ -26,11 +26,11 @@ class ReplicaAwareDatabase: ) @property - def has_replicas(self) -> bool: + def has_replicas(self): return bool(self._active_replicas) @property - def is_connected(self) -> bool: + def is_connected(self): return self._primary.is_connected async def connect(self): @@ -86,7 +86,7 @@ class ReplicaAwareDatabase: ): return await self._run_read("fetch_val", force_primary, query, values, column) - def _should_use_primary(self, explicit_force: bool) -> bool: + def _should_use_primary(self, explicit_force: bool): if explicit_force or self._force_primary_context.get(): return True @@ -98,7 +98,7 @@ class ReplicaAwareDatabase: return False - def _next_replica(self) -> Database: + def _next_replica(self): replica = self._active_replicas[ self._replica_index % len(self._active_replicas) ] diff --git a/comet/core/log_levels.py b/comet/core/log_levels.py index 9fe25dc..a7802ac 100644 --- a/comet/core/log_levels.py +++ b/comet/core/log_levels.py @@ -68,6 +68,12 @@ CUSTOM_LOG_LEVELS = { "loguru_color": "", "no": 35, }, + "PLAYBACK": { + "color": "#00FA9A", + "icon": "▶️", + "loguru_color": "", + "no": 35, + }, } ALL_LOG_LEVELS = {**STANDARD_LOG_LEVELS, **CUSTOM_LOG_LEVELS} diff --git a/comet/core/logger.py b/comet/core/logger.py index 2f83db2..30be3d4 100644 --- a/comet/core/logger.py +++ b/comet/core/logger.py @@ -8,6 +8,7 @@ from loguru import logger from comet.core.execution import max_workers from comet.core.log_levels import (CUSTOM_LOG_LEVELS, STANDARD_LOG_LEVELS, get_level_info) +from comet.utils.parsing import associate_urls_credentials logging.getLogger("demagnetize").setLevel( logging.CRITICAL @@ -157,8 +158,6 @@ def log_scraper_error( def log_startup_info(settings): - from comet.utils.parsing import associate_urls_credentials - def get_urls_with_passwords(urls, passwords): url_credentials_pairs = associate_urls_credentials(urls, passwords) @@ -215,9 +214,14 @@ def log_startup_info(settings): "Use PostgreSQL for reliable background scraping." ) + anime_mapping_refresh = ( + f" - Refresh Interval: {settings.ANIME_MAPPING_REFRESH_INTERVAL}s" + if settings.ANIME_MAPPING_ENABLED + else "" + ) logger.log( "COMET", - f"Anime Mapping: source={settings.ANIME_MAPPING_SOURCE} - refresh_interval={settings.ANIME_MAPPING_REFRESH_INTERVAL}s", + f"Anime Mapping: {settings.ANIME_MAPPING_ENABLED}{anime_mapping_refresh}", ) logger.log( @@ -431,6 +435,16 @@ def log_startup_info(settings): ) logger.log("COMET", f"Custom Header HTML: {bool(settings.CUSTOM_HEADER_HTML)}") + http_cache_info = ( + f" - Public Streams TTL: {settings.HTTP_CACHE_PUBLIC_STREAMS_TTL}s - Private Streams TTL: {settings.HTTP_CACHE_PRIVATE_STREAMS_TTL}s - Stale While Revalidate: {settings.HTTP_CACHE_STALE_WHILE_REVALIDATE}s" + if settings.HTTP_CACHE_ENABLED + else "" + ) + logger.log( + "COMET", + f"HTTP Cache: {bool(settings.HTTP_CACHE_ENABLED)}{http_cache_info}", + ) + background_scraper_display = ( f" - Workers: {settings.BACKGROUND_SCRAPER_CONCURRENT_WORKERS} - Interval: {settings.BACKGROUND_SCRAPER_INTERVAL}s - Max Movies/Run: {settings.BACKGROUND_SCRAPER_MAX_MOVIES_PER_RUN} - Max Series/Run: {settings.BACKGROUND_SCRAPER_MAX_SERIES_PER_RUN}" if settings.BACKGROUND_SCRAPER_ENABLED diff --git a/comet/core/models.py b/comet/core/models.py index 41098e5..25abd80 100644 --- a/comet/core/models.py +++ b/comet/core/models.py @@ -70,7 +70,7 @@ class AppSettings(BaseSettings): CATALOG_TIMEOUT: Optional[int] = 30 DOWNLOAD_TORRENT_FILES: Optional[bool] = False SCRAPE_COMET: Union[bool, str] = False - COMET_URL: Union[str, List[str]] = "https://comet.elfhosted.com" + COMET_URL: Union[str, List[str]] = "https://comet.feels.legal" SCRAPE_NYAA: Union[bool, str] = False NYAA_ANIME_ONLY: Optional[bool] = True NYAA_MAX_CONCURRENT_PAGES: Optional[int] = 5 @@ -120,7 +120,7 @@ class AppSettings(BaseSettings): TORRENT_DISABLED_STREAM_DESCRIPTION: Optional[str] = ( "Direct torrent playback is disabled on this server." ) - TORRENT_DISABLED_STREAM_URL: Optional[str] = "https://comet.looks.legal" + TORRENT_DISABLED_STREAM_URL: Optional[str] = "https://comet.feels.legal" PUBLIC_BASE_URL: Optional[str] = None REMOVE_ADULT_CONTENT: Optional[bool] = False BACKGROUND_SCRAPER_ENABLED: Optional[bool] = False @@ -128,8 +128,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] = "database" - ANIME_MAPPING_REFRESH_INTERVAL: Optional[int] = 86400 + ANIME_MAPPING_ENABLED: Optional[bool] = True + ANIME_MAPPING_REFRESH_INTERVAL: Optional[int] = 432000 DIGITAL_RELEASE_FILTER: Optional[bool] = False TMDB_READ_ACCESS_TOKEN: Optional[str] = None GLOBAL_PROXY_URL: Optional[str] = None @@ -137,6 +137,10 @@ class AppSettings(BaseSettings): RATELIMIT_MAX_RETRIES: Optional[int] = 3 RATELIMIT_RETRY_BASE_DELAY: Optional[float] = 1.0 RTN_FILTER_DEBUG: Optional[bool] = False + HTTP_CACHE_ENABLED: Optional[bool] = False + HTTP_CACHE_PUBLIC_STREAMS_TTL: Optional[int] = 300 + HTTP_CACHE_PRIVATE_STREAMS_TTL: Optional[int] = 60 + HTTP_CACHE_STALE_WHILE_REVALIDATE: Optional[int] = 60 @field_validator("INDEXER_MANAGER_TYPE") def set_indexer_manager_type(cls, v, values): @@ -144,15 +148,6 @@ 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: @@ -760,7 +755,7 @@ web_config = { } -def _build_database_instance(raw_url: str) -> Database: +def _build_database_instance(raw_url: str): driver = "sqlite" if settings.DATABASE_TYPE == "sqlite" else "postgresql+asyncpg" prefix = "/" if settings.DATABASE_TYPE == "sqlite" else "" return Database(f"{driver}://{prefix}{raw_url}") diff --git a/comet/debrid/stremthru.py b/comet/debrid/stremthru.py index 561fac2..780ae18 100644 --- a/comet/debrid/stremthru.py +++ b/comet/debrid/stremthru.py @@ -2,13 +2,14 @@ import asyncio from urllib.parse import quote, unquote import aiohttp -from RTN import parse, title_match +from RTN import normalize_title, parse, title_match from comet.core.execution import get_executor from comet.core.logger import logger from comet.core.models import settings from comet.debrid.exceptions import DebridAuthError from comet.services.debrid_cache import cache_availability +from comet.services.filtering import quick_alias_match from comet.services.torrent_manager import torrent_update_queue from comet.utils.parsing import is_video @@ -217,6 +218,18 @@ class StremThru: sources: list = None, aliases: dict = None, ): + """ + Smart file selection algorithm with scoring system. + + Priority order (highest to lowest): + 1. Exact season + episode match with single episode file (+1000) + 2. Exact season + episode match with multi-episode file (+500) + 3. Episode match without season info (+200) + 4. Exact filename match with requested torrent_name (+100) + 5. Title alias match (+50) + 6. Index match from original selection (+25) + 7. Fallback: largest video file (+file_size as tiebreaker) + """ try: magnet_uri = f"magnet:?xt=urn:btih:{hash}&dn={quote(torrent_name)}" @@ -236,82 +249,170 @@ class StremThru: name = unquote(name) torrent_name = unquote(torrent_name) - name_parsed = parse(name) - target_file = None + ez_aliases = aliases.get("ez", []) + if ez_aliases: + ez_aliases_normalized = [normalize_title(a) for a in ez_aliases] debrid_files = magnet["data"]["files"] - debrid_files_parsed = [] - files = [] + + # Filter to video files only, excluding samples + video_files = [] + filenames_to_parse = [] for file in debrid_files: filename = file["name"] + filename_lower = filename.lower() - if "sample" in filename.lower(): + if "sample" in filename_lower: continue - if not is_video(filename): continue - filename_parsed = parse(filename) + video_files.append(file) + filenames_to_parse.append(filename) - file_season = ( - filename_parsed.seasons[0] if filename_parsed.seasons else None - ) - file_episode = ( - filename_parsed.episodes[0] if filename_parsed.episodes else None - ) - file_index = file["index"] if file["index"] != -1 else None - file_size = file["size"] if file["size"] != -1 else None + if not video_files: + logger.warning(f"No video files found in torrent {hash}") + return - file = { - "index": file_index, - "title": filename, - "size": file_size, - "season": file_season, - "episode": file_episode, - "link": file.get("link", None), - } + loop = asyncio.get_running_loop() + parsed_results = await loop.run_in_executor( + get_executor(), batch_parse, filenames_to_parse + ) - debrid_files_parsed.append(file) + scored_files = [] - if not filename_parsed.parsed_title or not title_match( - name_parsed.parsed_title, - filename_parsed.parsed_title, - aliases=aliases, - ): + for file, filename, parsed in zip( + video_files, filenames_to_parse, parsed_results + ): + file_index = file["index"] if file.get("index", -1) != -1 else None + file_size = file["size"] if file.get("size", -1) != -1 else 0 + file_link = file.get("link") + + if not file_link: continue - file["info_hash"] = hash - file["parsed"] = filename_parsed - files.append(file) + file_season = parsed.seasons[0] if parsed.seasons else None + file_episode = parsed.episodes[0] if parsed.episodes else None - for file in debrid_files_parsed: - if file["title"] == torrent_name: - target_file = file - break + # Calculate score + score = 0 + match_reason = [] - if season == file["season"] and episode == file["episode"]: - target_file = file - break + # Season + Episode matching (highest priority) + if season is not None and episode is not None: + season_matches = (not parsed.seasons) or (season in parsed.seasons) + episode_matches = parsed.episodes and episode in parsed.episodes - if not target_file: - for file in files: - if str(file["index"]) == index: - target_file = file - break + if season_matches and episode_matches: + if len(parsed.episodes) == 1: + score += 1000 # Perfect single episode match + match_reason.append("exact_episode") + else: + score += 500 # Multi-episode file containing our episode + match_reason.append("multi_episode") + elif episode_matches: + score += 200 # Episode matches but season doesn't + match_reason.append("episode_only") - if len(files) > 0: - asyncio.create_task(cache_availability(self.real_debrid_name, files)) + # Exact filename match + if filename == torrent_name: + score += 100 + match_reason.append("exact_name") - if not target_file and len(debrid_files) > 0: - files_with_link = [ - file for file in debrid_files if file.get("link", None) - ] - if len(files_with_link) > 0: - target_file = max(files_with_link, key=lambda x: x["size"]) + # Title/alias matching + if parsed.parsed_title: + # Quick alias match first + if ez_aliases and quick_alias_match( + normalize_title(filename), ez_aliases_normalized + ): + score += 50 + match_reason.append("alias") + elif title_match(name, parsed.parsed_title, aliases=aliases): + score += 50 + match_reason.append("title") - if not target_file: + # Index match from original selection + if file_index is not None and str(file_index) == str(index): + score += 25 + match_reason.append("index") + + # Use file size as tiebreaker (larger files preferred) + # Normalize to 0-10 range to not overwhelm other scores + size_score = min( + file_size / (10 * 1024 * 1024 * 1024), 10 + ) # Cap at 10GB + score += size_score + + enriched_file = { + "index": file_index, + "title": filename, + "size": file_size if file_size > 0 else None, + "season": file_season, + "episode": file_episode, + "link": file_link, + "parsed": parsed, + "score": score, + "match_reason": match_reason, + } + + scored_files.append(enriched_file) + + if not scored_files: + logger.log( + "PLAYBACK", + f"No valid video files with links found in torrent {hash}", + ) return + # Sort by score descending + scored_files.sort(key=lambda x: x["score"], reverse=True) + + # Select best file + target_file = scored_files[0] + + logger.log( + "PLAYBACK", + f"File selection for {hash}: selected '{target_file['title']}' " + f"(score={target_file['score']:.1f}, reasons={target_file['match_reason']}) " + f"from {len(scored_files)} candidates", + ) + + all_files_for_cache = [] + + for f in scored_files: + if f["season"] is not None or f["episode"] is not None: + all_files_for_cache.append( + { + "info_hash": hash, + "index": f["index"], + "title": f["title"], + "size": f["size"], + "season": f["season"] or season, + "episode": f["episode"], + "parsed": f["parsed"], + } + ) + + # Also ensure the selected file is cached with the REQUESTED season/episode + # This handles cases where filename doesn't contain S/E info but user requested it + if season is not None or episode is not None: + all_files_for_cache.append( + { + "info_hash": hash, + "index": target_file["index"], + "title": target_file["title"], + "size": target_file["size"], + "season": season, + "episode": episode, + "parsed": target_file["parsed"], + } + ) + + if all_files_for_cache: + asyncio.create_task( + cache_availability(self.real_debrid_name, all_files_for_cache) + ) + link = await self.session.post( f"{self.base_url}/link/generate?client_ip={self.client_ip}", json={"link": target_file["link"]}, diff --git a/comet/metadata/filter.py b/comet/metadata/filter.py index 4a7f15c..5e3ffce 100644 --- a/comet/metadata/filter.py +++ b/comet/metadata/filter.py @@ -15,6 +15,9 @@ class DigitalReleaseFilter: season: int = None, episode: int = None, ): + if not media_id.startswith("tt"): + return True + try: cached_date = await database.fetch_val( """ @@ -36,12 +39,8 @@ class DigitalReleaseFilter: tmdb = TMDBApi(session) - if media_id.startswith("tt"): - imdb_id = media_id.split(":")[0] - tmdb_id = await tmdb.get_tmdb_id_from_imdb(imdb_id) - else: - # Other formats (e.g. kitsu) are not supported - return True + imdb_id = media_id.partition(":")[0] + tmdb_id = await tmdb.get_tmdb_id_from_imdb(imdb_id) if not tmdb_id: logger.warning( diff --git a/comet/metadata/imdb.py b/comet/metadata/imdb.py index 8edcac5..07ecf9d 100644 --- a/comet/metadata/imdb.py +++ b/comet/metadata/imdb.py @@ -5,15 +5,23 @@ from comet.core.logger import logger async def get_imdb_metadata(session: aiohttp.ClientSession, id: str): try: - response = await session.get( - f"https://v3.sg.media-imdb.com/suggestion/a/{id}.json" - ) - metadata = await response.json() + async with session.get( + f"https://v3.sg.media-imdb.com/suggestion/a/{id}.json", + ) as response: + metadata = await response.json() + for element in metadata["d"]: if "/" not in element["id"]: title = element["l"] year = element.get("y") - year_end = int(element["yr"].split("-")[1]) if "yr" in element else None + + year_end = None + yr = element.get("yr") + if yr: + _, _, end_part = yr.partition("-") + if end_part: + year_end = int(end_part) + return title, year, year_end except Exception as e: additional_info = "" diff --git a/comet/metadata/kitsu.py b/comet/metadata/kitsu.py index 3252357..4aef0f5 100644 --- a/comet/metadata/kitsu.py +++ b/comet/metadata/kitsu.py @@ -5,42 +5,39 @@ from comet.core.logger import logger async def get_kitsu_metadata(session: aiohttp.ClientSession, id: str): try: - response = await session.get(f"https://kitsu.io/api/edge/anime/{id}") - metadata = await response.json() + async with session.get( + f"https://kitsu.io/api/edge/anime/{id}", + ) as response: + metadata = await response.json() - attributes = metadata["data"]["attributes"] - year = int(attributes["createdAt"].split("-")[0]) - year_end = int(attributes["updatedAt"].split("-")[0]) + attributes = metadata.get("data", {}).get("attributes") - return attributes["canonicalTitle"], year, year_end + title = attributes.get("canonicalTitle") + if not title: + titles = attributes.get("titles") or {} + title = titles.get("en") or titles.get("en_jp") or titles.get("ja_jp") + + year = None + start_date = attributes.get("startDate") + if start_date and len(start_date) >= 4: + year = int(start_date[:4]) + + year_end = None + end_date = attributes.get("endDate") + if end_date and len(end_date) >= 4: + year_end = int(end_date[:4]) + + # if year is None: + # created_at = attributes.get("createdAt") + # if created_at and len(created_at) >= 4: + # year = int(created_at[:4]) + + # if year_end is None: + # updated_at = attributes.get("updatedAt") + # if updated_at and len(updated_at) >= 4: + # year_end = int(updated_at[:4]) + + return title, year, year_end except Exception as e: logger.warning(f"Exception while getting Kitsu metadata for {id}: {e}") return None, None, None - - -async def get_kitsu_aliases(session: aiohttp.ClientSession, id: str): - aliases = {} - try: - response = await session.get( - f"https://find-my-anime.dtimur.de/api?id={id}&provider=Kitsu" - ) - data = await response.json() - - aliases["ez"] = [] - aliases["ez"].append(data[0]["title"]) - for synonym in data[0]["synonyms"]: - aliases["ez"].append(synonym) - - total_aliases = len(aliases["ez"]) - if total_aliases > 0: - logger.log( - "SCRAPER", - f"📜 Found {total_aliases} Kitsu title aliases for {id}", - ) - return aliases - except Exception: - pass - - logger.log("SCRAPER", f"📜 No Kitsu title aliases found for {id}") - - return {} diff --git a/comet/metadata/manager.py b/comet/metadata/manager.py index c9ba36d..4d704d7 100644 --- a/comet/metadata/manager.py +++ b/comet/metadata/manager.py @@ -4,74 +4,121 @@ import time import aiohttp import orjson +from comet.core.logger import logger from comet.core.models import database, settings from comet.services.anime import anime_mapper from comet.utils.parsing import parse_media_id from .imdb import get_imdb_metadata -from .kitsu import get_kitsu_aliases, get_kitsu_metadata +from .kitsu import get_kitsu_metadata from .trakt import get_trakt_aliases +_CACHE_SELECT_QUERY = """ + SELECT title, year, year_end, aliases + FROM metadata_cache + WHERE media_id = :media_id + AND timestamp >= :min_timestamp +""" + +_CACHE_INSERT_SQLITE = """ + INSERT OR IGNORE INTO metadata_cache + VALUES (:media_id, :title, :year, :year_end, :aliases, :timestamp) +""" + +_CACHE_INSERT_POSTGRESQL = """ + INSERT INTO metadata_cache + VALUES (:media_id, :title, :year, :year_end, :aliases, :timestamp) + ON CONFLICT DO NOTHING +""" + class MetadataScraper: def __init__(self, session: aiohttp.ClientSession): self.session = session - - async def fetch_metadata_and_aliases(self, media_type: str, media_id: str): - id, season, episode = parse_media_id(media_type, media_id) - - get_cached = await self.get_cached( - id, season if "kitsu" not in media_id else 1, episode + self._cache_insert_query = ( + _CACHE_INSERT_SQLITE + if settings.DATABASE_TYPE == "sqlite" + else _CACHE_INSERT_POSTGRESQL ) + + async def get_from_cache_by_media_id( + self, media_id: str, id: str, season: int | None, episode: int | None + ): + provider = self._extract_provider(media_id) + cache_id = f"{provider}:{id}" if provider else id + cache_season = 1 if provider == "kitsu" else season + + return await self.get_cached(cache_id, cache_season, episode) + + async def fetch_metadata_and_aliases( + self, + media_type: str, + media_id: str, + id: str | None = None, + season: int | None = None, + episode: int | None = None, + ): + if id is None: + id, season, episode = parse_media_id(media_type, media_id) + + provider = self._extract_provider(media_id) + cache_id = f"{provider}:{id}" if provider else id + cache_season = 1 if provider == "kitsu" else season + + get_cached = await self.get_cached(cache_id, cache_season, episode) if get_cached is not None: return get_cached[0], get_cached[1] - is_kitsu = "kitsu" in media_id + is_kitsu = provider == "kitsu" + metadata_task = asyncio.create_task( self.get_metadata(id, season, episode, is_kitsu) ) - aliases_task = asyncio.create_task(self.get_aliases(media_type, id, is_kitsu)) + aliases_task = asyncio.create_task(self.get_aliases(media_type, id, provider)) metadata, aliases = await asyncio.gather(metadata_task, aliases_task) if metadata is not None: - await self.cache_metadata(id, metadata, aliases) + await self.cache_metadata(cache_id, metadata, aliases) return metadata, aliases + @staticmethod + def _extract_provider(media_id: str): + if media_id.startswith("tt"): + return "imdb" + + first_part, sep, _ = media_id.partition(":") + + if sep: + return first_part.lower() + + return None + async def get_cached(self, media_id: str, season: int, episode: int): row = await database.fetch_one( - """ - SELECT title, year, year_end, aliases - FROM metadata_cache - WHERE media_id = :media_id - AND timestamp + :cache_ttl >= :current_time - """, + _CACHE_SELECT_QUERY, { "media_id": media_id, - "cache_ttl": settings.METADATA_CACHE_TTL, - "current_time": time.time(), + "min_timestamp": time.time() - settings.METADATA_CACHE_TTL, }, ) - if row is not None: - metadata = { + if row is None: + return None + + return ( + { "title": row["title"], "year": row["year"], "year_end": row["year_end"], "season": season, "episode": episode, - } - return metadata, orjson.loads(row["aliases"]) - - return None + }, + orjson.loads(row["aliases"]), + ) async def cache_metadata(self, media_id: str, metadata: dict, aliases: dict): await database.execute( - f""" - INSERT {"OR IGNORE " if settings.DATABASE_TYPE == "sqlite" else ""} - INTO metadata_cache - VALUES (:media_id, :title, :year, :year_end, :aliases, :timestamp) - {" ON CONFLICT DO NOTHING" if settings.DATABASE_TYPE == "postgresql" else ""} - """, + self._cache_insert_query, { "media_id": media_id, "title": metadata["title"], @@ -114,14 +161,19 @@ class MetadataScraper: title: str, year: int, year_end: int = None, + id: str | None = None, ): """ Fetch only aliases for media when we already have the metadata from another source. This method will cache the provided metadata along with the scraped aliases. """ - id, _, _ = parse_media_id(media_type, media_id) + if id is None: + id, _, _ = parse_media_id(media_type, media_id) - get_cached = await self.get_cached(id, 1, 1) + provider = self._extract_provider(media_id) + cache_id = f"{provider}:{id}" if provider else id + + get_cached = await self.get_cached(cache_id, 1, 1) if get_cached is not None: return get_cached[0], get_cached[1] @@ -131,64 +183,41 @@ class MetadataScraper: "year_end": year_end, } - is_kitsu = "kitsu" in media_id - aliases = await self.get_aliases(media_type, id, is_kitsu) + aliases = await self.get_aliases(media_type, id, provider) - await self.cache_metadata(id, metadata, aliases) + await self.cache_metadata(cache_id, metadata, aliases) return metadata, aliases - def combine_aliases(self, kitsu_aliases: dict, trakt_aliases: dict): - combined = {"ez": []} + async def get_aliases( + self, + media_type: str, + media_id: str, + provider: str | None = None, + ): + if anime_mapper.is_loaded(): + full_media_id = f"{provider}:{media_id}" - # Add Kitsu aliases - if kitsu_aliases and "ez" in kitsu_aliases: - combined["ez"].extend(kitsu_aliases["ez"]) + if anime_mapper.is_anime_content(full_media_id, media_id): + aliases = await anime_mapper.get_aliases(full_media_id) + logger.log( + "SCRAPER", + f"📜 Found {len(aliases.get('ez', []))} Anime title aliases for {media_id}", + ) + if aliases: + return aliases - # Add Trakt aliases - if trakt_aliases and "ez" in trakt_aliases: - combined["ez"].extend(trakt_aliases["ez"]) + if provider == "kitsu": + logger.log("SCRAPER", f"📜 No Anime title aliases found for {media_id}") + return {} - # Case-insensitive deduplication - combined["ez"] = list( - {alias.lower(): alias for alias in combined["ez"]}.values() - ) - - return combined if combined["ez"] else {} - - async def get_aliases(self, media_type: str, media_id: str, is_kitsu: bool): - if not anime_mapper.is_loaded(): - # Fallback to original behavior if mapping not loaded - if is_kitsu: - return await get_kitsu_aliases(self.session, media_id) - return await get_trakt_aliases(self.session, media_type, media_id) - - kitsu_aliases = {} - trakt_aliases = {} - - if is_kitsu: - # Get Kitsu aliases - kitsu_aliases = await get_kitsu_aliases(self.session, media_id) - - # Try to convert Kitsu ID to IMDB ID for Trakt aliases - try: - kitsu_id = int(media_id) - imdb_id = anime_mapper.get_imdb_from_kitsu(kitsu_id) - if imdb_id: - # We have an IMDB ID, get Trakt aliases too - trakt_aliases = await get_trakt_aliases( - self.session, media_type, imdb_id - ) - except Exception: - pass + trakt_aliases = await get_trakt_aliases(self.session, media_type, media_id) + if trakt_aliases: + logger.log( + "SCRAPER", + f"📜 Found {len(trakt_aliases['ez'])} Trakt title aliases for {media_id}", + ) else: - # Get Trakt aliases for IMDB ID - trakt_aliases = await get_trakt_aliases(self.session, media_type, media_id) + logger.log("SCRAPER", f"📜 No Trakt title aliases found for {media_id}") - # Check if this IMDB ID has a Kitsu equivalent for additional aliases - kitsu_id = anime_mapper.get_kitsu_from_imdb(media_id) - if kitsu_id: - kitsu_aliases = await get_kitsu_aliases(self.session, kitsu_id) - - # Combine the aliases from both sources - return self.combine_aliases(kitsu_aliases, trakt_aliases) + return trakt_aliases diff --git a/comet/metadata/tmdb.py b/comet/metadata/tmdb.py index 4757c5e..1fbd4e1 100644 --- a/comet/metadata/tmdb.py +++ b/comet/metadata/tmdb.py @@ -24,18 +24,16 @@ class TMDBApi: data = await response.json() - release_dates = [] - for result in data.get("results", []): - for release in result.get("release_dates", []): - if release.get("type") in [4, 5]: # Digital or Physical - date_str = release.get("release_date", "").split("T")[0] - if date_str: - release_dates.append(date_str) + release_dates = [] + for result in data.get("results", []): + for release in result.get("release_dates", []): + # Type 4 = Digital, Type 5 = Physical + if release.get("type") in (4, 5): + date_str = release.get("release_date", "").split("T")[0] + if date_str: + release_dates.append(date_str) - if release_dates: - return min(release_dates) - - return None + return min(release_dates) if release_dates else None except Exception as e: logger.error(f"TMDB: Error getting movie release date for {tmdb_id}: {e}") return None @@ -68,12 +66,15 @@ class TMDBApi: data = await response.json() - if data.get("movie_results"): - return str(data["movie_results"][0]["id"]) - if data.get("tv_results"): - return str(data["tv_results"][0]["id"]) + movie_results = data.get("movie_results") + if movie_results: + return str(movie_results[0]["id"]) - return None + tv_results = data.get("tv_results") + if tv_results: + return str(tv_results[0]["id"]) + + return None except Exception as e: logger.error(f"TMDB: Error converting IMDB ID {imdb_id}: {e}") return None diff --git a/comet/metadata/trakt.py b/comet/metadata/trakt.py index 1897101..71c8079 100644 --- a/comet/metadata/trakt.py +++ b/comet/metadata/trakt.py @@ -1,31 +1,25 @@ import aiohttp -from comet.core.logger import logger - async def get_trakt_aliases( session: aiohttp.ClientSession, media_type: str, media_id: str ): - aliases = set() try: - response = await session.get( + async with session.get( f"https://api.trakt.tv/{'movies' if media_type == 'movie' else 'shows'}/{media_id}/aliases" - ) - data = await response.json() + ) as response: + data = await response.json() - for aliase in data: - aliases.add(aliase["title"]) + seen = {} + for alias_entry in data: + title = alias_entry.get("title") + if title and title not in seen: + seen[title] = None - total_aliases = len(aliases) - if total_aliases > 0: - logger.log( - "SCRAPER", - f"📜 Found {total_aliases} Trakt title aliases for {media_id}", - ) - return {"ez": list(aliases)} + if seen: + aliases_list = list(seen.keys()) + return {"ez": aliases_list} except Exception: pass - logger.log("SCRAPER", f"📜 No Trakt title aliases found for {media_id}") - return {} diff --git a/comet/scrapers/aiostreams.py b/comet/scrapers/aiostreams.py index 59ea615..b3c2f4b 100644 --- a/comet/scrapers/aiostreams.py +++ b/comet/scrapers/aiostreams.py @@ -37,10 +37,14 @@ class AiostreamsScraper(BaseScraper): if "indexer" in torrent: tracker += f"|{torrent['indexer']}" + infoHash = torrent["infoHash"] + if infoHash is None: + continue + torrents.append( { "title": torrent["filename"], - "infoHash": torrent["infoHash"], + "infoHash": infoHash, "fileIndex": torrent.get("fileIdx", None), "seeders": torrent.get("seeders", None), "size": torrent["size"], diff --git a/comet/scrapers/comet.py b/comet/scrapers/comet.py index 2784fa4..a2e55ba 100644 --- a/comet/scrapers/comet.py +++ b/comet/scrapers/comet.py @@ -11,7 +11,7 @@ class CometScraper(BaseScraper): torrents = [] try: async with self.session.get( - f"{self.url}/e30=/stream/{request.media_type}/{request.media_id}.json", + f"{self.url}/stream/{request.media_type}/{request.media_id}.json", ) as response: results = await response.json() diff --git a/comet/scrapers/stremthru.py b/comet/scrapers/stremthru.py index cd80e7a..f71f64e 100644 --- a/comet/scrapers/stremthru.py +++ b/comet/scrapers/stremthru.py @@ -16,7 +16,7 @@ class StremthruScraper(BaseScraper): try: media_id = request.media_only_id if "kitsu" in request.media_id: - imdb_id = anime_mapper.get_imdb_from_kitsu(int(media_id)) + imdb_id = await anime_mapper.get_imdb_from_kitsu(int(media_id)) if imdb_id: media_id = imdb_id diff --git a/comet/services/anime.py b/comet/services/anime.py index f3bd8e6..ede6e0a 100644 --- a/comet/services/anime.py +++ b/comet/services/anime.py @@ -1,6 +1,8 @@ import asyncio +import ctypes +import gc +import sys import time -from collections.abc import Mapping import aiohttp import orjson @@ -8,82 +10,174 @@ import orjson from comet.core.logger import logger from comet.core.models import database, settings +_PROVIDER_URL_PATTERNS = ( + ("anilist.co/anime/", "anilist"), + ("myanimelist.net/anime/", "myanimelist"), + ("kitsu.app/anime/", "kitsu"), + ("kitsu.io/anime/", "kitsu"), + ("anidb.net/anime/", "anidb"), + ("anime-planet.com/anime/", "anime-planet"), + ("anisearch.com/anime/", "anisearch"), + ("livechart.me/anime/", "livechart"), + ("animecountdown.com/", "animecountdown"), + ("simkl.com/anime/", "simkl"), +) + +_FRIBB_PROVIDER_ORDER = ( + ("anilist", "anilist_id"), + ("myanimelist", "mal_id"), + ("kitsu", "kitsu_id"), + ("anidb", "anidb_id"), + ("anime-planet", "anime-planet_id"), + ("anisearch", "anisearch_id"), + ("livechart", "livechart_id"), + ("animecountdown", "animecountdown_id"), + ("simkl", "simkl_id"), +) + +_DB_CHUNK_SIZE = 10000 + class AnimeMapper: def __init__(self): - self.kitsu_to_imdb = {} - self.imdb_to_kitsu = {} - self.anime_imdb_ids = set() self.loaded = False self._refresh_lock = asyncio.Lock() - self._background_task = None + self._refresh_task = None + + self.anime_imdb_ids = set() + self._kitsu_mapping_cache = {} + + self._aod_url = "https://github.com/manami-project/anime-offline-database/releases/latest/download/anime-offline-database-minified.json" + self._fribb_url = "https://raw.githubusercontent.com/Fribb/anime-lists/refs/heads/master/anime-list-full.json" + self._kitsu_imdb_url = "https://raw.githubusercontent.com/TheBeastLT/stremio-kitsu-anime/master/static/data/imdb_mapping.json" async def load_anime_mapping(self, session: aiohttp.ClientSession | None = None): + if not settings.ANIME_MAPPING_ENABLED: + return True + if self.loaded: + return True + + count = await database.fetch_val("SELECT COUNT(*) FROM anime_entries") + if count and count > 0: + await self._load_provider_ids() + await self._load_kitsu_mapping_cache() + + kitsu_count = await database.fetch_val( + "SELECT COUNT(*) FROM kitsu_imdb_mapping" + ) + needs_kitsu_refresh = ( + kitsu_count == 0 or len(self._kitsu_mapping_cache) == 0 + ) + + if await self._is_cache_stale() or needs_kitsu_refresh: + self._refresh_task = asyncio.create_task( + self._refresh_from_remote(background=True) + ) + + self.loaded = True logger.log( "COMET", - "Anime mapping already loaded in this process; skipping reload", + f"✅ Anime mapping loaded from database: {count} entries, {len(self._kitsu_mapping_cache)} Kitsu-IMDB mappings", ) return True - 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) - - def get_kitsu_from_imdb(self, imdb_id: str): - return self.imdb_to_kitsu.get(imdb_id) - - def is_anime(self, imdb_id: str): - return imdb_id in self.anime_imdb_ids - def is_anime_content(self, media_id: str, media_only_id: str): - if "kitsu" in media_id: + if not settings.ANIME_MAPPING_ENABLED: + return False + + if not self.loaded: # to prevent blocking anime-only scrapers return True + provider, provider_id = self._parse_media_id(media_id) + + if provider == "kitsu": + return True + + if provider == "imdb": + return provider_id in self.anime_imdb_ids + + return media_only_id in self.anime_imdb_ids + + async def _get_entry_data(self, media_id: str): + provider, provider_id = self._parse_media_id(media_id) + if provider is None: + return None + + row = await database.fetch_one( + """ + SELECT e.data + FROM anime_entries e + INNER JOIN anime_ids i ON e.id = i.entry_id + WHERE i.provider = :provider AND i.provider_id = :provider_id + LIMIT 1 + """, + {"provider": provider, "provider_id": provider_id}, + ) + if not row: + return None + + return orjson.loads(row["data"]) + + async def get_aliases(self, media_id: str): if not self.loaded: - return True + return {} - return self.is_anime(media_only_id) + data = await self._get_entry_data(media_id) + if not data: + return {} + + title = data.get("title") + synonyms = data.get("synonyms") + + if not title and not synonyms: + return {} + + if title and synonyms: + return {"ez": [title, *synonyms]} + elif title: + return {"ez": [title]} + else: + return {"ez": list(synonyms)} + + async def get_imdb_from_kitsu(self, kitsu_id: str | int): + if not self.loaded: + return None + + val = await database.fetch_val( + """ + SELECT i2.provider_id + FROM anime_ids i1 + JOIN anime_ids i2 ON i1.entry_id = i2.entry_id + WHERE i1.provider = 'kitsu' AND i1.provider_id = :kitsu_id + AND i2.provider = 'imdb' + LIMIT 1 + """, + {"kitsu_id": str(kitsu_id)}, + ) + + return val + + def get_kitsu_episode_mapping(self, kitsu_id: str | int): + if not self.loaded: + return None + + return self._kitsu_mapping_cache.get(str(kitsu_id)) 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: - logger.log( - "DATABASE", - "Anime mapping cache empty; falling back to remote source", - ) - 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 + @staticmethod + def _parse_media_id(media_id: str): + provider, sep, provider_id = media_id.partition(":") + if not sep: + return None, None + return provider, provider_id async def _is_cache_stale(self): - interval = settings.ANIME_MAPPING_REFRESH_INTERVAL or 0 + interval = settings.ANIME_MAPPING_REFRESH_INTERVAL if interval <= 0: return False @@ -98,18 +192,41 @@ class AnimeMapper: if last_refresh is None: return True - last_refresh = float(last_refresh) - return (time.time() - last_refresh) >= interval + return (time.time() - float(last_refresh)) >= interval - def _ensure_periodic_refresh(self): - interval = settings.ANIME_MAPPING_REFRESH_INTERVAL or 0 - if interval <= 0: - return + async def _load_provider_ids(self): + try: + query = "SELECT provider_id FROM anime_ids WHERE provider = 'imdb'" + rows = await database.fetch_all(query) - if self._background_task and not self._background_task.done(): - return + self.anime_imdb_ids = { + row[0] if isinstance(row, tuple) else row["provider_id"] for row in rows + } + except Exception as e: + logger.error(f"Failed to load anime provider IDs: {e}") - self._background_task = asyncio.create_task(self._refresh_loop(interval)) + async def _load_kitsu_mapping_cache(self): + try: + rows = await database.fetch_all( + """ + SELECT kitsu_id, imdb_id, from_season, from_episode + FROM kitsu_imdb_mapping + WHERE (from_episode IS NOT NULL AND from_episode > 1) + OR from_season IS NOT NULL + """ + ) + + self._kitsu_mapping_cache.clear() + for row in rows: + kitsu_id = row["kitsu_id"] + + self._kitsu_mapping_cache[str(kitsu_id)] = { + "imdb_id": row["imdb_id"], + "from_season": row["from_season"], + "from_episode": row["from_episode"], + } + except Exception as e: + logger.warning(f"Failed to load Kitsu-IMDB mapping cache: {e}") async def _refresh_from_remote( self, @@ -118,7 +235,7 @@ class AnimeMapper: background: bool = False, ): async with self._refresh_lock: - if self.loaded and background: + if self.loaded and not background: return True own_session = False @@ -127,27 +244,72 @@ class AnimeMapper: 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", + "Downloading anime mapping (Source 1/3: Anime Offline Database)...", ) + async with session.get(self._aod_url) as response_aod: + if response_aod.status != 200: + logger.error(f"Failed to load AOD: HTTP {response_aod.status}") + return False + data_aod = orjson.loads(await response_aod.read()) - if settings.ANIME_MAPPING_SOURCE == "database": - await self._persist_mapping(data) - self._ensure_periodic_refresh() + logger.log( + "COMET", + "Downloading anime mapping (Source 2/3: Fribb Anime List)...", + ) + async with session.get(self._fribb_url) as response_fribb: + if response_fribb.status != 200: + logger.error( + f"Failed to load Fribb List: HTTP {response_fribb.status}" + ) + return False + data_fribb = orjson.loads(await response_fribb.read()) + + logger.log( + "COMET", + "Downloading anime mapping (Source 3/3: Kitsu-IMDB Mapping)...", + ) + async with session.get(self._kitsu_imdb_url) as response_kitsu: + if response_kitsu.status != 200: + logger.warning( + f"Failed to load Kitsu-IMDB mapping: HTTP {response_kitsu.status}" + ) + return False + data_kitsu_imdb = orjson.loads(await response_kitsu.read()) + + anime_list = data_aod.get("data", []) + total_entries = await self._persist_mapping(anime_list, data_fribb) + + await self._persist_kitsu_imdb_mapping(data_kitsu_imdb) + + del data_aod + del data_fribb + del data_kitsu_imdb + del anime_list + gc.collect() + + if sys.platform == "linux": + try: + ctypes.CDLL("libc.so.6").malloc_trim(0) + except Exception: + pass + elif sys.platform == "win32": + try: + ctypes.windll.psapi.EmptyWorkingSet( + ctypes.windll.kernel32.GetCurrentProcess() + ) + except Exception: + pass + + await self._load_provider_ids() + await self._load_kitsu_mapping_cache() + + self.loaded = True + logger.log( + "COMET", + f"✅ Anime mapping loaded: {total_entries} entries, {len(self._kitsu_mapping_cache)} Kitsu-IMDB mappings cached", + ) return True except Exception as exc: @@ -158,54 +320,113 @@ class AnimeMapper: 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: - self.kitsu_to_imdb[kitsu_id] = imdb_id - self.imdb_to_kitsu[imdb_id] = kitsu_id - self.anime_imdb_ids.add(imdb_id) - - self.loaded = True - - async def _persist_mapping(self, rows): + async def _persist_mapping(self, anime_list: list, fribb_list: list): 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": kitsu_id, - "imdb_id": self._entry_value(entry, "imdb_id"), - "is_anime": True, - "updated_at": timestamp, - } - ) + entries_batch = [] + ids_batch = [] + lookup_map = {} + total_entries = 0 - 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 + entries_query = "INSERT INTO anime_entries (id, data) VALUES (:id, :data)" + ids_query = """ + INSERT INTO anime_ids (provider, provider_id, entry_id) + VALUES (:provider, :provider_id, :entry_id) + ON CONFLICT (provider, provider_id) DO NOTHING + """ 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("DELETE FROM anime_entries") + await database.execute("DELETE FROM anime_ids") + + for idx, entry in enumerate(anime_list): + entry_id = idx + 1 + entries_batch.append( + {"id": entry_id, "data": orjson.dumps(entry).decode("utf-8")} ) + + sources = entry.get("sources") + if sources: + for source in sources: + for url_part, provider in _PROVIDER_URL_PATTERNS: + if url_part in source: + try: + if "id=" in source: + provider_id = source.split("id=", 1)[ + 1 + ].split("&", 1)[0] + else: + provider_id = source.rstrip("/").rsplit( + "/", 1 + )[-1] + + ids_batch.append( + { + "provider": provider, + "provider_id": provider_id, + "entry_id": entry_id, + } + ) + lookup_map[f"{provider}:{provider_id}"] = ( + entry_id + ) + except (IndexError, ValueError): + pass + break + + if len(entries_batch) >= _DB_CHUNK_SIZE: + await database.execute_many(entries_query, entries_batch) + total_entries += len(entries_batch) + entries_batch.clear() + + if len(ids_batch) >= _DB_CHUNK_SIZE: + await database.execute_many(ids_query, ids_batch) + ids_batch.clear() + + if entries_batch: + await database.execute_many(entries_query, entries_batch) + total_entries += len(entries_batch) + entries_batch.clear() + + if ids_batch: + await database.execute_many(ids_query, ids_batch) + ids_batch.clear() + + del entries_batch + del ids_batch + + fribb_batch = [] + for entry in fribb_list: + imdb_id = entry.get("imdb_id") + if not imdb_id: + continue + + for provider, key in _FRIBB_PROVIDER_ORDER: + val = entry.get(key) + if val: + found_entry_id = lookup_map.get(f"{provider}:{val}") + if found_entry_id is not None: + fribb_batch.append( + { + "provider": "imdb", + "provider_id": imdb_id, + "entry_id": found_entry_id, + } + ) + break + + if len(fribb_batch) >= _DB_CHUNK_SIZE: + await database.execute_many(ids_query, fribb_batch) + fribb_batch.clear() + + if fribb_batch: + await database.execute_many(ids_query, fribb_batch) + fribb_batch.clear() + + del fribb_batch + del lookup_map + await database.execute( """ INSERT INTO anime_mapping_state (id, refreshed_at) @@ -214,38 +435,67 @@ class AnimeMapper: """, {"timestamp": timestamp}, ) - logger.log( - "DATABASE", - f"Anime mapping cache updated ({len(params)} rows)", - ) + + return total_entries except Exception as exc: logger.error(f"Failed to persist anime mapping cache: {exc}") + return 0 - @staticmethod - def _entry_value(entry, key): - if isinstance(entry, Mapping): - return entry.get(key) - return entry[key] + async def _persist_kitsu_imdb_mapping(self, kitsu_imdb_data: dict): + total_count = 0 + batch = [] + batch_size = 1000 - async def _refresh_loop(self, interval: int): - while True: - try: - 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}" - ) + try: + async with database.transaction(): + await database.execute("DELETE FROM kitsu_imdb_mapping") - async def stop(self): - if self._background_task: - self._background_task.cancel() - try: - await self._background_task - except asyncio.CancelledError: - pass + insert_query = """ + INSERT INTO kitsu_imdb_mapping + (kitsu_id, imdb_id, title, from_season, from_episode) + VALUES (:kitsu_id, :imdb_id, :title, :from_season, :from_episode) + ON CONFLICT (kitsu_id) DO UPDATE SET + imdb_id = :imdb_id, + title = :title, + from_season = :from_season, + from_episode = :from_episode + """ + + for kitsu_id, entry in kitsu_imdb_data.items(): + if not isinstance(entry, dict): + continue + + imdb_id = entry.get("imdb_id") + if not imdb_id: + continue + + from_season = entry.get("fromSeason") + from_episode = entry.get("fromEpisode") + + batch.append( + { + "kitsu_id": str(kitsu_id), + "imdb_id": imdb_id, + "title": entry.get("title"), + "from_season": from_season, + "from_episode": from_episode, + } + ) + + if len(batch) >= batch_size: + await database.execute_many(insert_query, batch) + total_count += len(batch) + batch.clear() + + if batch: + await database.execute_many(insert_query, batch) + total_count += len(batch) + batch.clear() + + return total_count + except Exception as exc: + logger.error(f"Failed to persist Kitsu-IMDB mapping: {exc}") + return 0 anime_mapper = AnimeMapper() diff --git a/comet/services/debrid.py b/comet/services/debrid.py index 7c04810..bac8ad8 100644 --- a/comet/services/debrid.py +++ b/comet/services/debrid.py @@ -54,6 +54,8 @@ class DebridService: continue info_hash = file["info_hash"] + if info_hash not in torrents: + continue torrents[info_hash]["cached"] = True debrid_parsed = file["parsed"] @@ -80,7 +82,7 @@ class DebridService: for hash in info_hashes: torrents[hash]["cached"] = False - if self.debrid_service == "torrent" or len(torrents) == 0: + if len(torrents) == 0: return rows = await get_cached_availability( @@ -91,13 +93,24 @@ class DebridService: info_hash = row["info_hash"] torrents[info_hash]["cached"] = True - if row["parsed"] is not None: - torrents[info_hash]["parsed"] = ParsedData( - **orjson.loads(row["parsed"]) - ) if row["file_index"] is not None: - torrents[info_hash]["fileIndex"] = row["file_index"] - if row["title"] is not None: - torrents[info_hash]["title"] = row["title"] + try: + torrents[info_hash]["fileIndex"] = int(row["file_index"]) + except ValueError: + pass + if row["size"] is not None: torrents[info_hash]["size"] = row["size"] + + # Only update title/parsed if the cached file has resolution info + # Otherwise keep the original torrent info which may have better quality data + # E.g. torrent "[Group] Show S01 1080p" vs file "Show - 02.mkv" + if row["parsed"] is not None: + cached_parsed = ParsedData(**orjson.loads(row["parsed"])) + if ( + cached_parsed.resolution != "unknown" + or torrents[info_hash]["parsed"].resolution == "unknown" + ): + torrents[info_hash]["parsed"] = cached_parsed + if row["title"] is not None: + torrents[info_hash]["title"] = row["title"] diff --git a/comet/services/debrid_cache.py b/comet/services/debrid_cache.py index 34fa235..1796cb6 100644 --- a/comet/services/debrid_cache.py +++ b/comet/services/debrid_cache.py @@ -116,31 +116,43 @@ async def cache_availability(debrid_service: str, availability: list): async def get_cached_availability( debrid_service: str, info_hashes: list, season: int = None, episode: int = None ): - base_query = f""" - SELECT info_hash, file_index, title, size, parsed + select_clause = "SELECT info_hash, file_index, title, size, parsed" + if debrid_service == "torrent" and settings.DATABASE_TYPE == "postgresql": + select_clause = ( + "SELECT DISTINCT ON (info_hash) info_hash, file_index, title, size, parsed" + ) + + base_from_where = f""" FROM debrid_availability WHERE info_hash IN (SELECT CAST(value as TEXT) FROM {"json_array_elements_text" if settings.DATABASE_TYPE == "postgresql" else "json_each"}(:info_hashes)) - AND debrid_service = :debrid_service AND timestamp + :cache_ttl >= :current_time """ params = { "info_hashes": orjson.dumps(info_hashes).decode("utf-8"), - "debrid_service": debrid_service, "cache_ttl": settings.DEBRID_CACHE_TTL, "current_time": time.time(), "season": season, "episode": episode, } + if debrid_service != "torrent": + base_from_where += " AND debrid_service = :debrid_service" + params["debrid_service"] = debrid_service + + group_by_clause = "" + if debrid_service == "torrent" and settings.DATABASE_TYPE == "sqlite": + group_by_clause = " GROUP BY info_hash" + if debrid_service == "offcloud": - query = ( - base_query - + """ + season_episode_filter = """ AND ((CAST(:season as INTEGER) IS NULL AND season IS NULL) OR season = CAST(:season as INTEGER)) AND ((CAST(:episode as INTEGER) IS NULL AND episode IS NULL) OR episode = CAST(:episode as INTEGER)) """ + query = ( + select_clause + base_from_where + season_episode_filter + group_by_clause ) + results = await database.fetch_all(query, params) found_hashes = {r["info_hash"] for r in results} @@ -149,20 +161,26 @@ async def get_cached_availability( if remaining_hashes: null_title_params = { "info_hashes": orjson.dumps(remaining_hashes).decode("utf-8"), - "debrid_service": debrid_service, "cache_ttl": settings.DEBRID_CACHE_TTL, "current_time": time.time(), } - null_title_query = base_query + " AND title IS NULL" + if debrid_service != "torrent": + null_title_params["debrid_service"] = debrid_service + + null_title_query = ( + select_clause + base_from_where + " AND title IS NULL" + group_by_clause + ) null_results = await database.fetch_all(null_title_query, null_title_params) results.extend(null_results) else: query = ( - base_query + select_clause + + base_from_where + """ AND ((CAST(:season as INTEGER) IS NULL AND season IS NULL) OR season = CAST(:season as INTEGER)) AND ((CAST(:episode as INTEGER) IS NULL AND episode IS NULL) OR episode = CAST(:episode as INTEGER)) """ + + group_by_clause ) results = await database.fetch_all(query, params) diff --git a/comet/services/filtering.py b/comet/services/filtering.py index 95f0737..7aca56c 100644 --- a/comet/services/filtering.py +++ b/comet/services/filtering.py @@ -1,4 +1,4 @@ -from RTN import parse, title_match +from RTN import normalize_title, parse, title_match from comet.core.logger import logger from comet.core.models import settings @@ -13,12 +13,22 @@ else: pass +def quick_alias_match(text_normalized: str, ez_aliases_normalized: list[str]): + return any(alias in text_normalized for alias in ez_aliases_normalized) + + def filter_worker(torrents, title, year, year_end, aliases, remove_adult_content): results = [] + ez_aliases = aliases.get("ez", []) + if ez_aliases: + ez_aliases_normalized = [normalize_title(a) for a in ez_aliases] + for torrent in torrents: torrent_title = torrent["title"] - if "sample" in torrent_title.lower() or torrent_title == "": + torrent_title_lower = torrent_title.lower() + + if "sample" in torrent_title_lower or torrent_title == "": _log_exclusion(f"🚫 Rejected (Sample/Empty) | {torrent_title}") continue @@ -28,14 +38,20 @@ def filter_worker(torrents, title, year, year_end, aliases, remove_adult_content _log_exclusion(f"🔞 Rejected (Adult) | {torrent_title}") continue - if not parsed.parsed_title or not title_match( - title, parsed.parsed_title, aliases=aliases - ): - _log_exclusion( - f"❌ Rejected (Title Mismatch) | {torrent_title} | Parsed: {parsed.parsed_title} | Expected: {title}" - ) + if not parsed.parsed_title: + _log_exclusion(f"❌ Rejected (No Parsed Title) | {torrent_title}") continue + alias_matched = ez_aliases and quick_alias_match( + normalize_title(torrent_title), ez_aliases_normalized + ) + if not alias_matched: + if not title_match(title, parsed.parsed_title, aliases=aliases): + _log_exclusion( + f"❌ Rejected (Title Mismatch) | {torrent_title} | Parsed: {parsed.parsed_title} | Expected: {title}" + ) + continue + if year and parsed.year: if year_end is not None: if not (year <= parsed.year <= year_end): diff --git a/comet/services/orchestration.py b/comet/services/orchestration.py index 3516d27..6c4bfe2 100644 --- a/comet/services/orchestration.py +++ b/comet/services/orchestration.py @@ -28,7 +28,10 @@ class TorrentManager: episode: int, aliases: dict, remove_adult_content: bool, - context: str = "live", # "live" or "background" + is_kitsu: bool = False, + context: str = "live", + search_episode: int | None = None, + search_season: int | None = None, ): self.debrid_service = debrid_service self.debrid_api_key = debrid_api_key @@ -41,8 +44,11 @@ class TorrentManager: self.year_end = year_end self.season = season self.episode = episode + self.search_episode = search_episode if search_episode is not None else episode + self.search_season = search_season if search_season is not None else season self.aliases = aliases self.remove_adult_content = remove_adult_content + self.is_kitsu = is_kitsu self.context = context self.seen_hashes = set() @@ -62,8 +68,8 @@ class TorrentManager: title=self.title, year=self.year, year_end=self.year_end, - season=self.season, - episode=self.episode, + season=self.search_season, + episode=self.search_episode, context=self.context, ) @@ -78,10 +84,14 @@ class TorrentManager: torrent["parsed"].episodes[0] if torrent["parsed"].episodes else None ) - if (season is not None and season != self.season) or ( - episode is not None and episode != self.episode - ): - continue + if self.is_kitsu: + if episode is not None and episode != self.search_episode: + continue + else: + if (season is not None and season != self.season) or ( + episode is not None and episode != self.episode + ): + continue info_hash = torrent["infoHash"] self.torrents[info_hash] = { @@ -95,22 +105,45 @@ class TorrentManager: } async def get_cached_torrents(self): - rows = await database.fetch_all( - """ - SELECT info_hash, file_index, title, seeders, size, tracker, sources, parsed - FROM torrents - WHERE media_id = :media_id - AND ((season IS NOT NULL AND season = CAST(:season as INTEGER)) OR (season IS NULL AND CAST(:season as INTEGER) IS NULL)) - AND (episode IS NULL OR episode = CAST(:episode as INTEGER)) - """, - { - "media_id": self.media_only_id, - "season": self.season, - "episode": self.episode, - }, - ) + if self.is_kitsu: + rows = await database.fetch_all( + """ + SELECT info_hash, file_index, title, seeders, size, tracker, sources, parsed, episode + FROM torrents + WHERE media_id = :media_id + AND (episode IS NULL OR episode = CAST(:episode as INTEGER)) + """, + { + "media_id": self.media_only_id, + "episode": self.search_episode, + }, + ) + else: + rows = await database.fetch_all( + """ + SELECT info_hash, file_index, title, seeders, size, tracker, sources, parsed, episode + FROM torrents + WHERE media_id = :media_id + AND ((season IS NOT NULL AND season = CAST(:season as INTEGER)) OR (season IS NULL AND CAST(:season as INTEGER) IS NULL)) + AND (episode IS NULL OR episode = CAST(:episode as INTEGER)) + """, + { + "media_id": self.media_only_id, + "season": self.season, + "episode": self.episode, + }, + ) + + rows = sorted(rows, key=lambda r: (r["episode"] is not None, r["episode"])) for row in rows: + parsed_data = ParsedData(**orjson.loads(row["parsed"])) + + if row["episode"] is None and parsed_data.episodes: + target_episode = self.search_episode if self.is_kitsu else self.episode + if target_episode not in parsed_data.episodes: + continue + info_hash = row["info_hash"] self.torrents[info_hash] = { "fileIndex": row["file_index"], @@ -119,28 +152,45 @@ class TorrentManager: "size": row["size"], "tracker": row["tracker"], "sources": orjson.loads(row["sources"]), - "parsed": ParsedData(**orjson.loads(row["parsed"])), + "parsed": parsed_data, } async def cache_torrents(self): for torrent in self.ready_to_cache: - file_info = { - "info_hash": torrent["infoHash"], - "index": torrent["fileIndex"], - "title": torrent["title"], - "size": torrent["size"], - "season": torrent["parsed"].seasons[0] - if torrent["parsed"].seasons - else self.season, - "episode": torrent["parsed"].episodes[0] - if torrent["parsed"].episodes - else None, - "parsed": torrent["parsed"], - "seeders": torrent["seeders"], - "tracker": torrent["tracker"], - "sources": torrent["sources"], - } - await torrent_update_queue.add_torrent_info(file_info, self.media_only_id) + if self.is_kitsu: + cache_seasons = [self.season] + else: + cache_seasons = ( + torrent["parsed"].seasons + if torrent["parsed"].seasons + else [self.season] + ) + + parsed_episodes = ( + torrent["parsed"].episodes if torrent["parsed"].episodes else [None] + ) + + if len(parsed_episodes) > 1: + episode = None + else: + episode = parsed_episodes[0] + + for season in cache_seasons: + file_info = { + "info_hash": torrent["infoHash"], + "index": torrent["fileIndex"], + "title": torrent["title"], + "size": torrent["size"], + "season": season, + "episode": episode, + "parsed": torrent["parsed"], + "seeders": torrent["seeders"], + "tracker": torrent["tracker"], + "sources": torrent["sources"], + } + await torrent_update_queue.add_torrent_info( + file_info, self.media_only_id + ) async def filter_manager(self, scraper_name: str, torrents: list): if len(torrents) == 0: diff --git a/comet/services/streaming/manager.py b/comet/services/streaming/manager.py index 56b6a28..75cb2dc 100644 --- a/comet/services/streaming/manager.py +++ b/comet/services/streaming/manager.py @@ -10,7 +10,7 @@ from comet.core.logger import logger from comet.core.models import database, settings from comet.services.bandwidth import bandwidth_monitor from comet.services.streaming.wrapper import monitored_handle_stream_request -from comet.utils.network import NO_CACHE_HEADERS +from comet.utils.cache import NO_CACHE_HEADERS async def on_stream_end(connection_id: str, ip: str): diff --git a/comet/services/torrent_manager.py b/comet/services/torrent_manager.py index f1a02e5..7b1bcc8 100644 --- a/comet/services/torrent_manager.py +++ b/comet/services/torrent_manager.py @@ -3,12 +3,14 @@ import base64 import hashlib import re import time +from asyncio import QueueEmpty from collections import defaultdict from urllib.parse import unquote import anyio import bencodepy import orjson +import xxhash from demagnetize.core import Demagnetizer from RTN import ParsedData, parse from torf import Magnet @@ -125,48 +127,34 @@ async def add_torrent( parsed: ParsedData, ): try: - parsed_season = parsed.seasons[0] if parsed.seasons else search_season - parsed_episode = parsed.episodes[0] if parsed.episodes else None + seasons_to_process = parsed.seasons if parsed.seasons else [search_season] + parsed_episodes = parsed.episodes if parsed.episodes else [None] - if parsed_episode is not None: - await database.execute( - """ - DELETE FROM torrents - WHERE info_hash = :info_hash - AND season = :season - AND episode IS NULL - """, + episode_to_insert = parsed_episodes[0] if len(parsed_episodes) == 1 else None + + for season in seasons_to_process: + await _upsert_torrent_record( { + "media_id": media_id, "info_hash": info_hash, - "season": parsed_season, - }, + "file_index": file_index, + "season": season, + "episode": episode_to_insert, + "title": title, + "seeders": seeders, + "size": size, + "tracker": tracker, + "sources": orjson.dumps(sources).decode("utf-8"), + "parsed": orjson.dumps(parsed, default_dump).decode("utf-8"), + "timestamp": time.time(), + } ) - logger.log( - "SCRAPER", - f"Deleted season-only entry for S{parsed_season:02d} of {info_hash}", - ) - - await _upsert_torrent_record( - { - "media_id": media_id, - "info_hash": info_hash, - "file_index": file_index, - "season": parsed_season, - "episode": parsed_episode, - "title": title, - "seeders": seeders, - "size": size, - "tracker": tracker, - "sources": orjson.dumps(sources).decode("utf-8"), - "parsed": orjson.dumps(parsed, default_dump).decode("utf-8"), - "timestamp": time.time(), - } - ) additional = "" - if parsed_season: - additional += f" - S{parsed_season:02d}" - additional += f"E{parsed_episode:02d}" if parsed_episode else "" + if seasons_to_process: + additional += f" - S{seasons_to_process[0]:02d}" + if parsed_episodes and parsed_episodes[0] is not None: + additional += f"E{parsed_episodes[0]:02d}" logger.log("SCRAPER", f"Added torrent for {media_id} - {title}{additional}") except Exception as e: @@ -249,151 +237,164 @@ UPDATE_INTERVAL = ( class TorrentUpdateQueue: + __slots__ = ( + "queue", + "batch_size", + "flush_interval", + "is_running", + "_lock", + "_event", + "upserts", + "_is_postgresql", + "_grouped_upserts", + ) + def __init__(self, batch_size: int = 1000, flush_interval: float = 5.0): self.queue = asyncio.Queue() self.batch_size = batch_size self.flush_interval = flush_interval self.is_running = False - self.batches = {"to_delete": set(), "upserts": {}} + self._lock = asyncio.Lock() + self._event = asyncio.Event() + self.upserts = {} + self._is_postgresql = settings.DATABASE_TYPE == "postgresql" + self._grouped_upserts = defaultdict(list) async def add_torrent_info(self, file_info: dict, media_id: str = None): await self.queue.put((file_info, media_id)) + self._event.set() + if not self.is_running: - self.is_running = True - asyncio.create_task(self._process_queue()) + async with self._lock: + if not self.is_running: + self.is_running = True + asyncio.create_task(self._process_queue()) async def _process_queue(self): last_flush_time = time.time() - while self.is_running: - try: - while not self.queue.empty(): + try: + while True: + try: + await asyncio.wait_for( + self._event.wait(), timeout=self.flush_interval + ) + except asyncio.TimeoutError: + pass + + self._event.clear() + + batch_time = time.time() + items_processed = 0 + while True: try: file_info, media_id = self.queue.get_nowait() - await self._process_file_info(file_info, media_id) - except asyncio.QueueEmpty: + self._process_file_info(file_info, media_id, batch_time) + self.queue.task_done() + items_processed += 1 + + if len(self.upserts) >= self.batch_size: + await self._flush_batch() + last_flush_time = time.time() + batch_time = last_flush_time + except QueueEmpty: break current_time = time.time() - if ( + if self.upserts and ( current_time - last_flush_time >= self.flush_interval - or self.queue.empty() - ) and any(len(batch) > 0 for batch in self.batches.values()): + ): await self._flush_batch() last_flush_time = current_time - if self.queue.empty() and not any( - len(batch) > 0 for batch in self.batches.values() - ): - self.is_running = False + if self.queue.empty() and not self.upserts: break - await asyncio.sleep(0.1) - - except asyncio.CancelledError: - break - except Exception as e: - logger.warning(f"Error in _process_queue: {e}") - self._reset_batches() - - if any(len(batch) > 0 for batch in self.batches.values()): - await self._flush_batch() - - self.is_running = False + except asyncio.CancelledError: + pass + except Exception as e: + logger.warning(f"Error in _process_queue: {e}") + finally: + if self.upserts: + await self._flush_batch() + if self.is_running: + async with self._lock: + self.is_running = False async def stop(self): self.is_running = False + self._event.set() - # Process remaining items in queue - while not self.queue.empty(): + shutdown_time = time.time() + while True: try: file_info, media_id = self.queue.get_nowait() - await self._process_file_info(file_info, media_id) + self._process_file_info(file_info, media_id, shutdown_time) + except QueueEmpty: + break except Exception as e: logger.warning( f"Error processing remaining queue items during shutdown: {e}" ) break - # Flush any remaining batches - if any(len(batch) > 0 for batch in self.batches.values()): + if self.upserts: await self._flush_batch() - def _reset_batches(self): - for key, batch in self.batches.items(): - if len(batch) > 0: - logger.warning( - f"Ignoring {len(batch)} items in problematic '{key}' batch" - ) - batch.clear() - async def _flush_batch(self): + if not self.upserts: + return + + upserts_to_flush = self.upserts + self.upserts = {} + try: - if self.batches["to_delete"]: - delete_items = list(self.batches["to_delete"]) - sub_batch_size = 100 - for i in range(0, len(delete_items), sub_batch_size): - try: - sub_batch = delete_items[i : i + sub_batch_size] + grouped = self._grouped_upserts + for params in upserts_to_flush.values(): + key = _determine_conflict_key(params["season"], params["episode"]) + grouped[key].append(params) - placeholders = [] - params = {} - for idx, item in enumerate(sub_batch): - info_hash, season = item - key_suffix = f"_{idx}" - placeholders.append( - f"(CAST(:info_hash{key_suffix} AS TEXT), CAST(:season{key_suffix} AS INTEGER))" - ) - params[f"info_hash{key_suffix}"] = info_hash - params[f"season{key_suffix}"] = season - - async with database.transaction(): - delete_query = f""" - DELETE FROM torrents - WHERE (info_hash, season) IN ( - {",".join(placeholders)} - ) - AND episode IS NULL - """ - await database.execute(delete_query, params) - except Exception as e: - logger.warning(f"Error processing delete batch: {e}") - - self.batches["to_delete"].clear() - - if self.batches["upserts"]: - grouped: dict[str, list[dict]] = defaultdict(list) - for params in self.batches["upserts"].values(): - key = _determine_conflict_key(params["season"], params["episode"]) - grouped[key].append(params) - - for key, rows in grouped.items(): - query = _get_torrent_upsert_query(key) - try: - await _execute_batched_upsert(query, rows) - except Exception as e: - logger.warning(f"Error processing upsert batch: {e}") - - total_upserts = len(self.batches["upserts"]) - if total_upserts > 0: - logger.log( - "SCRAPER", - f"Upserted {total_upserts} torrents in batch", - ) - self.batches["upserts"].clear() + for key, rows in grouped.items(): + query = _get_torrent_upsert_query(key) + try: + await _execute_batched_upsert(query, rows) + except Exception as e: + logger.warning(f"Error processing upsert batch: {e}") + total_upserts = len(upserts_to_flush) + if total_upserts > 0: + logger.log( + "SCRAPER", + f"Upserted {total_upserts} torrents in batch", + ) except Exception as e: logger.warning(f"Error in flush_batch: {e}") - self._reset_batches() + finally: + grouped.clear() - async def _process_file_info(self, file_info: dict, media_id: str = None): + def _process_file_info( + self, file_info: dict, media_id: str = None, current_time: float = None + ): try: + info_hash = file_info["info_hash"] + season = file_info["season"] + episode = file_info["episode"] + + if current_time is None: + current_time = time.time() + + upsert_key = (media_id, info_hash, season, episode) + + existing = self.upserts.get(upsert_key) + if existing and existing["timestamp"] >= current_time: + return + params = { - "info_hash": file_info["info_hash"], + "info_hash": info_hash, "file_index": file_info["index"], - "season": file_info["season"], - "episode": file_info["episode"], + "season": season, + "episode": episode, "title": file_info["title"], "seeders": file_info["seeders"], "size": file_info["size"], @@ -402,46 +403,21 @@ class TorrentUpdateQueue: "parsed": orjson.dumps( file_info["parsed"], default=default_dump ).decode("utf-8"), - "timestamp": time.time(), + "timestamp": current_time, "media_id": media_id, } - if settings.DATABASE_TYPE == "postgresql": + if self._is_postgresql: params["update_interval"] = UPDATE_INTERVAL params["lock_key"] = _compute_advisory_lock_key( - media_id, - file_info["info_hash"], - file_info["season"], - file_info["episode"], + media_id, info_hash, season, episode ) - upsert_key = _build_upsert_key( - file_info["info_hash"], - file_info["season"], - file_info["episode"], - media_id, - ) + self.upserts[upsert_key] = params - # In-memory deduplication: keep the freshest timestamp - existing = self.batches["upserts"].get(upsert_key) - if not existing or params["timestamp"] > existing["timestamp"]: - self.batches["upserts"][upsert_key] = params - - if file_info["episode"] is not None: - self.batches["to_delete"].add( - (file_info["info_hash"], file_info["season"]) - ) - - await self._check_batch_size() except Exception as e: logger.warning(f"Error processing file info: {e}") - finally: - self.queue.task_done() - - async def _check_batch_size(self): - if any(len(batch) >= self.batch_size for batch in self.batches.values()): - await self._flush_batch() TORRENT_INSERT_TEMPLATE = """ @@ -513,97 +489,70 @@ POSTGRES_CONFLICT_TARGETS = { _POSTGRES_UPSERT_CACHE: dict[str, str] = {} -def _determine_conflict_key(season, episode) -> str: - if season is not None and episode is not None: - return "series" +def _determine_conflict_key(season, episode): if season is not None: - return "season_only" - if episode is not None: - return "episode_only" - return "none" + return "series" if episode is not None else "season_only" + return "episode_only" if episode is not None else "none" -def _build_upsert_key(info_hash, season, episode, media_id): - return (media_id, info_hash, season, episode) +def _compute_advisory_lock_key(media_id, info_hash, season, episode): + payload = f"{media_id}|{info_hash}|{season}|{episode}" + return xxhash.xxh64_intdigest(payload, seed=0) - (1 << 63) -def _compute_advisory_lock_key(media_id, info_hash, season, episode) -> int: - payload = f"{media_id}|{info_hash}|{season}|{episode}".encode("utf-8") - digest = hashlib.sha1(payload).digest() - return int.from_bytes(digest[:8], byteorder="big", signed=True) +_SQLITE_CHECK_COLS = frozenset( + ["file_index", "title", "seeders", "size", "tracker", "sources", "parsed"] +) async def _execute_sqlite_batched_upsert(rows: list[dict]): if not rows: return - # Fetch relevant existing records to compare against - media_ids = list({row["media_id"] for row in rows if row.get("media_id")}) - if not media_ids: - async with database.transaction(): - await _execute_standard_sqlite_insert(rows) + info_hashes = {row["info_hash"] for row in rows} + + if not info_hashes: + await _execute_standard_sqlite_insert(rows) return - placeholders = ",".join(f":mid{i}" for i in range(len(media_ids))) - params = {f"mid{i}": mid for i, mid in enumerate(media_ids)} + info_hashes_list = list(info_hashes) + chunk_size = 900 + existing_rows = [] - existing_rows = await database.fetch_all( - f"SELECT * FROM torrents WHERE media_id IN ({placeholders})", params - ) + for i in range(0, len(info_hashes_list), chunk_size): + chunk = info_hashes_list[i : i + chunk_size] + placeholders = ",".join(f":ih{j}" for j in range(len(chunk))) + params = {f"ih{j}": ih for j, ih in enumerate(chunk)} - # Build lookup map: (media_id, info_hash, season, episode) -> row - existing_map = {} - for row in existing_rows: - key = ( - row["media_id"], - row["info_hash"], - row["season"], - row["episode"], + chunk_rows = await database.fetch_all( + f"SELECT media_id, info_hash, season, episode, file_index, title, seeders, size, tracker, sources, parsed, timestamp FROM torrents WHERE info_hash IN ({placeholders})", + params, ) - existing_map[key] = row + existing_rows.extend(chunk_rows) + + existing_map = { + (row["media_id"], row["info_hash"], row["season"], row["episode"]): row + for row in existing_rows + } + + if not existing_map: + await _execute_standard_sqlite_insert(rows) + return - # Filter in Python to_insert = [] - # Columns to check for changes (everything except timestamp and lock_key/update_interval) - check_cols = [ - "file_index", - "title", - "seeders", - "size", - "tracker", - "sources", - "parsed", - ] - for row in rows: - key = ( - row["media_id"], - row["info_hash"], - row["season"], - row["episode"], - ) + key = (row["media_id"], row["info_hash"], row["season"], row["episode"]) existing = existing_map.get(key) if not existing: - # New record to_insert.append(row) continue - # Check for data changes - changed = False - for col in check_cols: - # Simple equality check. - if row.get(col) != existing[col]: - changed = True - break - - if changed: + if any(row.get(col) != existing[col] for col in _SQLITE_CHECK_COLS): to_insert.append(row) continue - # Check TTL - existing_ts = existing["timestamp"] - if existing_ts < (row["timestamp"] - UPDATE_INTERVAL): + if existing["timestamp"] < (row["timestamp"] - UPDATE_INTERVAL): to_insert.append(row) if to_insert: @@ -620,7 +569,6 @@ async def _execute_standard_sqlite_insert(rows: list[dict]): VALUES ({", ".join(f":{col}" for col in columns)}) """ - # Retry logic for busy database for attempt in range(5): try: async with database.transaction(): @@ -652,15 +600,12 @@ async def _execute_batched_upsert(query: str, rows): rows_to_insert.append(row) continue - # Use transaction-level non-blocking lock - # This automatically releases at the end of the transaction acquired = await database.fetch_val( "SELECT pg_try_advisory_xact_lock(CAST(:lock_key AS BIGINT))", {"lock_key": lock_key}, ) if acquired: rows_to_insert.append(row) - # If not acquired, skip this row - another replica is handling it if rows_to_insert: sanitized_rows = [ @@ -674,7 +619,7 @@ async def _execute_batched_upsert(query: str, rows): logger.warning(f"Error executing batched upsert: {e}") -def _get_torrent_upsert_query(conflict_key: str) -> str: +def _get_torrent_upsert_query(conflict_key: str): if settings.DATABASE_TYPE == "sqlite": return SQLITE_UPSERT_QUERY diff --git a/comet/templates/index.html b/comet/templates/index.html index ab11d28..d5c04ae 100644 --- a/comet/templates/index.html +++ b/comet/templates/index.html @@ -6,7 +6,7 @@ - + @@ -422,8 +422,8 @@ {% if not disableTorrentStreams %} Torrent {% endif %} - Real-Debrid TorBox + Real-Debrid All-Debrid Debrid-Link Premiumize diff --git a/comet/utils/cache.py b/comet/utils/cache.py new file mode 100644 index 0000000..2db9160 --- /dev/null +++ b/comet/utils/cache.py @@ -0,0 +1,212 @@ +import hashlib +from typing import Any, Optional + +import orjson +from fastapi import Request, Response + +from comet.core.models import settings + +NO_CACHE_HEADERS = { + "Cache-Control": "no-store, no-cache, must-revalidate, max-age=0", + "Pragma": "no-cache", + "Expires": "0", +} + + +class CacheControl: + def __init__(self): + self._directives = [] + self._max_age = None + self._s_maxage = None + self._stale_while_revalidate = None + self._stale_if_error = None + + def public(self): + """Response can be cached by any cache.""" + self._directives.append("public") + return self + + def private(self): + """Response is intended for a single user.""" + self._directives.append("private") + return self + + def no_cache(self): + """Cache must revalidate with origin before using cached copy.""" + self._directives.append("no-cache") + return self + + def no_store(self): + """Response must not be stored in any cache.""" + self._directives.append("no-store") + return self + + def must_revalidate(self): + """Cache must revalidate stale responses.""" + self._directives.append("must-revalidate") + return self + + def immutable(self): + """Response will not change during its freshness lifetime.""" + self._directives.append("immutable") + return self + + def max_age(self, seconds: int): + """Maximum time response is considered fresh (browser cache).""" + self._max_age = seconds + return self + + def s_maxage(self, seconds: int): + """Maximum time response is fresh for shared caches (CDN/proxy).""" + self._s_maxage = seconds + return self + + def stale_while_revalidate(self, seconds: int): + """Serve stale while revalidating in background.""" + self._stale_while_revalidate = seconds + return self + + def stale_if_error(self, seconds: int): + """Serve stale if origin returns error.""" + self._stale_if_error = seconds + return self + + def build(self): + """Build the Cache-Control header value.""" + parts = list(self._directives) + + if self._max_age is not None: + parts.append(f"max-age={self._max_age}") + if self._s_maxage is not None: + parts.append(f"s-maxage={self._s_maxage}") + if self._stale_while_revalidate is not None: + parts.append(f"stale-while-revalidate={self._stale_while_revalidate}") + if self._stale_if_error is not None: + parts.append(f"stale-if-error={self._stale_if_error}") + + return ", ".join(parts) + + +def generate_etag(data: Any): + if isinstance(data, bytes): + content = data + elif isinstance(data, str): + content = data.encode("utf-8") + else: + content = orjson.dumps(data, option=orjson.OPT_SORT_KEYS) + + hash_digest = hashlib.md5(content, usedforsecurity=False).hexdigest()[:16] + return f'W/"{hash_digest}"' + + +def check_etag_match(request: Request, etag: str): + if_none_match = request.headers.get("If-None-Match") + if not if_none_match: + return False + + client_etags = [e.strip() for e in if_none_match.split(",")] + + normalized_etag = etag.replace('W/"', '"') + for client_etag in client_etags: + normalized_client = client_etag.replace('W/"', '"') + if normalized_client == normalized_etag or client_etag == "*": + return True + + return False + + +class CachedJSONResponse(Response): + def __init__( + self, + content: Any, + status_code: int = 200, + cache_control: Optional[CacheControl] = None, + etag: Optional[str] = None, + vary: Optional[list[str]] = None, + **kwargs, + ): + body = orjson.dumps(content) + super().__init__( + content=body, + status_code=status_code, + media_type="application/json", + **kwargs, + ) + + if cache_control: + self.headers["Cache-Control"] = cache_control.build() + + self.headers["ETag"] = etag or generate_etag(body) + + if vary: + self.headers["Vary"] = ", ".join(vary) + + +def not_modified_response(etag: str): + return Response( + status_code=304, + headers={ + "ETag": etag, + "Cache-Control": "must-revalidate", + }, + ) + + +class CachePolicies: + @staticmethod + def public_torrents(): + """ + For public torrent lists (without user config). + Cache for a short time at CDN, revalidate often. + """ + + ttl = settings.HTTP_CACHE_PUBLIC_STREAMS_TTL + swr = settings.HTTP_CACHE_STALE_WHILE_REVALIDATE + + return ( + CacheControl() + .public() + .max_age(ttl // 2) # Browser cache shorter + .s_maxage(ttl) # CDN/proxy cache longer + .stale_while_revalidate(swr) + .stale_if_error(300) + ) + + @staticmethod + def private_streams(): + """ + For user-specific stream results (with b64config). + Private cache only, short TTL. + """ + + ttl = settings.HTTP_CACHE_PRIVATE_STREAMS_TTL + + return CacheControl().private().max_age(ttl).must_revalidate() + + @staticmethod + def manifest(): + """ + For manifest.json responses. + Very short cache as it can change based on config. + """ + return CacheControl().private().max_age(60).must_revalidate() + + @staticmethod + def configure_page(): + """ + For the /configure page. + Cacheable if no custom HTML, otherwise private. + """ + + if settings.CUSTOM_HEADER_HTML: + return CacheControl().private().max_age(300) + + return CacheControl().public().max_age(300).s_maxage(3600) + + @staticmethod + def no_cache(): + """ + For responses that should never be cached. + Used for playback redirects, errors, etc. + """ + return CacheControl().private().no_store().no_cache().max_age(0) diff --git a/comet/utils/network.py b/comet/utils/network.py index 0a58c22..b409c40 100644 --- a/comet/utils/network.py +++ b/comet/utils/network.py @@ -2,12 +2,6 @@ import ipaddress from fastapi import Request -NO_CACHE_HEADERS = { - "Cache-Control": "no-store, no-cache, must-revalidate, max-age=0", - "Pragma": "no-cache", - "Expires": "0", -} - IP_REQUEST_HEADERS = [ "X-Client-Ip", "Cf-Connecting-Ip", diff --git a/comet/utils/network_manager.py b/comet/utils/network_manager.py index 176d551..b80960d 100644 --- a/comet/utils/network_manager.py +++ b/comet/utils/network_manager.py @@ -11,7 +11,7 @@ from comet.core.logger import logger from comet.core.models import settings -def resolve_proxy_url(proxy_url: Optional[str]) -> Optional[str]: +def resolve_proxy_url(proxy_url: Optional[str]): """ Resolve proxy hostname to IP address. @@ -255,18 +255,18 @@ class AsyncClientWrapper: self._aiohttp_session: Optional[aiohttp.ClientSession] = None self._curl_session: Optional[CurlSession] = None - async def _get_aiohttp_session(self) -> aiohttp.ClientSession: + async def _get_aiohttp_session(self): if not self._aiohttp_session or self._aiohttp_session.closed: self._aiohttp_session = aiohttp.ClientSession( headers=self.headers, timeout=aiohttp.ClientTimeout(total=self.timeout) ) return self._aiohttp_session - async def _get_curl_session(self) -> CurlSession: + async def _get_curl_session(self): if not self._curl_session: self._curl_session = CurlSession( headers=self.headers, - impersonate=self.impersonate or "chrome", + impersonate=self.impersonate, timeout=self.timeout, ) return self._curl_session @@ -277,13 +277,13 @@ class AsyncClientWrapper: if self._curl_session: await self._curl_session.close() - def request(self, method: str, url: str, **kwargs) -> _RequestContextManager: + def request(self, method: str, url: str, **kwargs): return _RequestContextManager(self, method, url, **kwargs) - def get(self, url: str, **kwargs) -> _RequestContextManager: + def get(self, url: str, **kwargs): return self.request("GET", url, **kwargs) - def post(self, url: str, **kwargs) -> _RequestContextManager: + def post(self, url: str, **kwargs): return self.request("POST", url, **kwargs) @@ -301,7 +301,7 @@ class NetworkManager: scraper_name: str, impersonate: Optional[str] = None, headers: Optional[dict] = None, - ) -> AsyncClientWrapper: + ): # Unique key for client configuration key = f"{scraper_name}|{impersonate}" diff --git a/comet/utils/parsing.py b/comet/utils/parsing.py index 0e39898..ce69306 100644 --- a/comet/utils/parsing.py +++ b/comet/utils/parsing.py @@ -59,20 +59,20 @@ def parse_optional_int(value: str | None): def parse_media_id(media_type: str, media_id: str): - if "kitsu" in media_id: - info = media_id.split(":") - - if len(info) > 2: - return info[1], 1, parse_optional_int(info[2]) - else: - return info[1], 1, None - + if media_id.startswith("kitsu:"): + _, _, rest = media_id.partition(":") + kitsu_id, _, episode_str = rest.partition(":") + return kitsu_id, 1, parse_optional_int(episode_str) if episode_str else None if media_type == "series": - info = media_id.split(":") - series_id = info[0] - season = parse_optional_int(info[1]) if len(info) > 1 else None - episode = parse_optional_int(info[2]) if len(info) > 2 else None - return series_id, season, episode + series_id, sep1, rest1 = media_id.partition(":") + if not sep1: + return series_id, None, None + season_str, sep2, episode_str = rest1.partition(":") + return ( + series_id, + parse_optional_int(season_str), + parse_optional_int(episode_str) if sep2 else None, + ) return media_id, None, None diff --git a/pyproject.toml b/pyproject.toml index 2443313..1c7534b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,4 +25,5 @@ dependencies = [ "python-multipart", "rank-torrent-name", "uvicorn", + "xxhash", ] diff --git a/scripts/uptime_monitor.py b/scripts/uptime_monitor.py index 2a975fc..d9c8a5d 100644 --- a/scripts/uptime_monitor.py +++ b/scripts/uptime_monitor.py @@ -66,7 +66,7 @@ async def check_instance(session: aiohttp.ClientSession, url: str) -> InstanceSt if manifest_ok: try: - async with session.get(f"{url}/e30=/stream/movie/{IMDB_ID}.json") as resp: + async with session.get(f"{url}/stream/movie/{IMDB_ID}.json") as resp: if resp.status == 200: data = await resp.json() streams = data.get("streams", [])