feat: add default item limits and concurrency settings for recommendations

Author: TimilsinaBimal

app/core/constants.py

@@ -1 +1,5 @@
 RECOMMENDATIONS_CATALOG_NAME: str = "Top Picks For You"
+DEFAULT_MIN_ITEMS: int = 20
+DEFAULT_MAX_ITEMS: int = 32
+
+DEFAULT_CONCURRENCY_LIMIT: int = 30

app/services/recommendation/catalog_service.py

@@ -1,16 +1,12 @@
-"""
-Catalog Service - Facade for catalog generation.
-
-Encapsulates all catalog logic: auth, profile building, routing, and recommendations.
-"""
-
 import re
 from typing import Any
 
+from fastapi import HTTPException
 from loguru import logger
 
 from app.api.endpoints.manifest import get_config_id
 from app.core.config import settings
+from app.core.constants import DEFAULT_MAX_ITEMS, DEFAULT_MIN_ITEMS
 from app.core.settings import UserSettings, get_default_settings
 from app.models.taste_profile import TasteProfile
 from app.services.catalog_updater import catalog_updater
@@ -24,19 +20,9 @@
 from app.services.tmdb.service import get_tmdb_service
 from app.services.token_store import token_store
 
-DEFAULT_MIN_ITEMS = 20
-DEFAULT_MAX_ITEMS = 32
-
 
 class CatalogService:
-    """
-    Facade for catalog generation.
-
-    Handles all catalog logic: validation, auth, profile building, routing, and recommendations.
-    """
-
     def __init__(self):
-        """Initialize catalog service."""
         pass
 
     async def get_catalog(
@@ -61,13 +47,18 @@ async def get_catalog(
         # Get credentials
         credentials = await token_store.get_user_data(token)
         if not credentials:
-            from fastapi import HTTPException
-
+            logger.error("No credentials found for token")
             raise HTTPException(status_code=401, detail="Invalid or expired token. Please reconfigure the addon.")
 
         # Trigger lazy update if needed
         if settings.AUTO_UPDATE_CATALOGS:
-            await catalog_updater.trigger_update(token, credentials)
+            logger.info(f"[{token[:8]}...] Triggering auto update for token")
+            try:
+                await catalog_updater.trigger_update(token, credentials)
+            except Exception as e:
+                logger.error(f"[{token[:8]}...] Failed to trigger auto update: {e}")
+                # continue with the request even if the auto update fails
+                pass
 
         bundle = StremioBundle()
         try:
@@ -82,11 +73,13 @@ async def get_catalog(
             # Initialize services
             services = self._initialize_services(language, user_settings)
 
+            integration_service: ProfileIntegration = services["integration"]
+
             # Build profile and watched sets (once, reused)
-            profile, watched_tmdb, watched_imdb = await services["integration"].build_profile_from_library(
+            profile, watched_tmdb, watched_imdb = await integration_service.build_profile_from_library(
                 library_items, content_type, bundle, auth_key
             )
-            whitelist = await services["integration"].get_genre_whitelist(profile, content_type) if profile else set()
+            whitelist = await integration_service.get_genre_whitelist(profile, content_type) if profile else set()
 
             # Get catalog limits
             min_items, max_items = self._get_catalog_limits(catalog_id, user_settings)
@@ -127,9 +120,6 @@ async def get_catalog(
             await bundle.close()
 
     def _validate_inputs(self, token: str, content_type: str, catalog_id: str) -> None:
-        """Validate input parameters."""
-        from fastapi import HTTPException
-
         if not token:
             raise HTTPException(
                 status_code=400,
@@ -144,26 +134,18 @@ def _validate_inputs(self, token: str, content_type: str, catalog_id: str) -> No
         if catalog_id not in ["watchly.rec", "watchly.creators"] and not any(
             catalog_id.startswith(p)
             for p in (
-                "tt",
                 "watchly.theme.",
-                "watchly.item.",
                 "watchly.loved.",
                 "watchly.watched.",
             )
         ):
             logger.warning(f"Invalid id: {catalog_id}")
             raise HTTPException(
                 status_code=400,
-                detail=(
-                    "Invalid id. Supported: 'watchly.rec', 'watchly.creators', 'watchly.theme.<params>',"
-                    "'watchly.item.<id>', or specific item IDs."
-                ),
+                detail=("Invalid id. Supported: 'watchly.rec', 'watchly.creators', 'watchly.theme.<params>'"),
             )
 
     async def _resolve_auth(self, bundle: StremioBundle, credentials: dict, token: str) -> str:
-        """Resolve and validate auth key."""
-        from fastapi import HTTPException
-
         auth_key = credentials.get("authKey")
         email = credentials.get("email")
         password = credentials.get("password")
@@ -174,7 +156,8 @@ async def _resolve_auth(self, bundle: StremioBundle, credentials: dict, token: s
             try:
                 await bundle.auth.get_user_info(auth_key)
                 is_valid = True
-            except Exception:
+            except Exception as e:
+                logger.error(f"Failed to validate auth key during catalog fetch: {e}")
                 pass
 
         # Try to refresh if invalid
@@ -188,31 +171,27 @@ async def _resolve_auth(self, bundle: StremioBundle, credentials: dict, token: s
                 logger.error(f"Failed to refresh auth key during catalog fetch: {e}")
 
         if not auth_key:
+            logger.error("No auth key found during catalog fetch")
             raise HTTPException(status_code=401, detail="Stremio session expired. Please reconfigure.")
 
         return auth_key
 
     def _extract_settings(self, credentials: dict) -> UserSettings:
-        """Extract user settings from credentials."""
         settings_dict = credentials.get("settings", {})
         return UserSettings(**settings_dict) if settings_dict else get_default_settings()
 
     def _initialize_services(self, language: str, user_settings: UserSettings) -> dict[str, Any]:
-        """Initialize all recommendation services."""
         tmdb_service = get_tmdb_service(language=language)
-        integration = ProfileIntegration(language=language)
-
         return {
             "tmdb": tmdb_service,
-            "integration": integration,
+            "integration": ProfileIntegration(language=language),
             "item": ItemBasedService(tmdb_service, user_settings),
             "theme": ThemeBasedService(tmdb_service, user_settings),
             "top_picks": TopPicksService(tmdb_service, user_settings),
             "creators": CreatorsService(tmdb_service, user_settings),
         }
 
     def _get_catalog_limits(self, catalog_id: str, user_settings: UserSettings) -> tuple[int, int]:
-        """Get min/max items for catalog."""
         try:
             cfg_id = get_config_id({"id": catalog_id})
         except Exception:
@@ -255,21 +234,19 @@ async def _get_recommendations(
     ) -> list[dict[str, Any]]:
         """Route to appropriate recommendation service based on catalog ID."""
         # Item-based recommendations
-        if catalog_id.startswith("tt") or any(
+        if any(
             catalog_id.startswith(p)
             for p in (
-                "watchly.item.",
                 "watchly.loved.",
                 "watchly.watched.",
             )
         ):
             # Extract item ID
-            if catalog_id.startswith("tt"):
-                item_id = catalog_id
-            else:
-                item_id = re.sub(r"^watchly\.(item|loved|watched)\.", "", catalog_id)
+            item_id = re.sub(r"^watchly\.(loved|watched)\.", "", catalog_id)
 
-            recommendations = await services["item"].get_recommendations_for_item(
+            item_service: ItemBasedService = services["item"]
+
+            recommendations = await item_service.get_recommendations_for_item(
                 item_id=item_id,
                 content_type=content_type,
                 watched_tmdb=watched_tmdb,
@@ -281,7 +258,9 @@ async def _get_recommendations(
 
         # Theme-based recommendations
         elif catalog_id.startswith("watchly.theme."):
-            recommendations = await services["theme"].get_recommendations_for_theme(
+            theme_service: ThemeBasedService = services["theme"]
+
+            recommendations = await theme_service.get_recommendations_for_theme(
                 theme_id=catalog_id,
                 content_type=content_type,
                 profile=profile,
@@ -294,8 +273,10 @@ async def _get_recommendations(
 
         # Creators-based recommendations
         elif catalog_id == "watchly.creators":
+            creators_service: CreatorsService = services["creators"]
+
             if profile:
-                recommendations = await services["creators"].get_recommendations_from_creators(
+                recommendations = await creators_service.get_recommendations_from_creators(
                     profile=profile,
                     content_type=content_type,
                     library_items=library_items,
@@ -311,7 +292,9 @@ async def _get_recommendations(
         # Top picks
         elif catalog_id == "watchly.rec":
             if profile:
-                recommendations = await services["top_picks"].get_top_picks(
+                top_picks_service: TopPicksService = services["top_picks"]
+
+                recommendations = await top_picks_service.get_top_picks(
                     profile=profile,
                     content_type=content_type,
                     library_items=library_items,

app/services/recommendation/creators.py

@@ -157,8 +157,6 @@ async def get_recommendations_from_creators(
 
             # Genre whitelist check
             genre_ids = item.get("genre_ids", [])
-            if not RecommendationFiltering.passes_top_genre_whitelist(genre_ids, whitelist):
-                continue
 
             # Excluded genres check
             if excluded_ids and any(gid in excluded_ids for gid in genre_ids):
@@ -168,7 +166,7 @@ async def get_recommendations_from_creators(
 
         # Enrich metadata
         enriched = await RecommendationMetadata.fetch_batch(
-            self.tmdb_service, filtered, content_type, target_count=limit, user_settings=self.user_settings
+            self.tmdb_service, filtered, content_type, user_settings=self.user_settings
         )
 
         # Final filter (remove watched by IMDB ID)

app/services/recommendation/filtering.py

@@ -162,7 +162,4 @@ def passes_top_genre_whitelist(genre_ids: list[int] | None, whitelist: set[int])
         gids = set(genre_ids or [])
         if not gids:
             return True
-        # If it's animation and not in whitelist, we still block it to prevent 'Anime Takeover'
-        if 16 in gids and 16 not in whitelist:
-            return False
         return True

app/services/recommendation/item_based.py

@@ -71,13 +71,13 @@ async def get_recommendations_for_item(
 
         # Enrich metadata
         enriched = await RecommendationMetadata.fetch_batch(
-            self.tmdb_service, filtered, content_type, target_count=limit, user_settings=self.user_settings
+            self.tmdb_service, filtered, content_type, user_settings=self.user_settings
         )
 
         # Final filter (remove watched by IMDB ID)
         final = filter_watched_by_imdb(enriched, watched_imdb or set())
 
-        return final[:limit]
+        return final
 
     async def _fetch_candidates(self, tmdb_id: int, mtype: str) -> list[dict[str, Any]]:
         """

app/services/recommendation/metadata.py

@@ -1,6 +1,9 @@
 import asyncio
 from typing import Any
 
+from loguru import logger
+
+from app.core.constants import DEFAULT_CONCURRENCY_LIMIT
 from app.services.rpdb import RPDBService
 
 
@@ -117,14 +120,13 @@ async def fetch_batch(
         tmdb_service: Any,
         items: list[dict[str, Any]],
         media_type: str,
-        target_count: int,
         user_settings: Any = None,
     ) -> list[dict[str, Any]]:
         """Fetch details for a batch of items in parallel with target-based short-circuiting."""
         final_results = []
         valid_items = [it for it in items if it.get("id")]
         query_type = "movie" if media_type == "movie" else "tv"
-        sem = asyncio.Semaphore(30)
+        sem = asyncio.Semaphore(DEFAULT_CONCURRENCY_LIMIT)
 
         async def _fetch_one(tid: int):
             async with sem:
@@ -135,25 +137,20 @@ async def _fetch_one(tid: int):
                 except Exception:
                     return None
 
-        # Process in chunks to allow early exit once target_count is reached
-        batch_size = 20
-        for i in range(0, len(valid_items), batch_size):
-            if len(final_results) >= target_count:
-                break
-
-            chunk = valid_items[i : i + batch_size]  # noqa
-            tasks = [_fetch_one(it["id"]) for it in chunk]
-            details_list = await asyncio.gather(*tasks)
+        tasks = [_fetch_one(it.get("id")) for it in valid_items]
+        details_list = await asyncio.gather(*tasks)
 
-            for details in details_list:
-                if not details:
-                    continue
+        format_task = [
+            cls.format_for_stremio(details, media_type, user_settings) for details in details_list if details
+        ]
 
-                formatted = await cls.format_for_stremio(details, media_type, user_settings)
-                if formatted:
-                    final_results.append(formatted)
+        formatted_list = await asyncio.gather(*format_task, return_exceptions=True)
 
-                if len(final_results) >= target_count:
-                    break
+        for formatted in formatted_list:
+            if isinstance(formatted, Exception):
+                logger.warning(f"Error formatting metadata: {formatted}")
+                continue
+            if formatted:
+                final_results.append(formatted)
 
         return final_results

app/services/recommendation/theme_based.py

@@ -145,7 +145,7 @@ async def get_recommendations_for_theme(
 
         # Enrich metadata
         enriched = await RecommendationMetadata.fetch_batch(
-            self.tmdb_service, filtered, content_type, target_count=limit, user_settings=self.user_settings
+            self.tmdb_service, filtered, content_type, user_settings=self.user_settings
         )
 
         # Final filter (remove watched by IMDB ID)

app/services/recommendation/top_picks.py

@@ -135,7 +135,7 @@ async def get_top_picks(
 
         # 8. Enrich metadata
         enriched = await RecommendationMetadata.fetch_batch(
-            self.tmdb_service, result, content_type, target_count=limit, user_settings=self.user_settings
+            self.tmdb_service, result, content_type, user_settings=self.user_settings
         )
 
         # 9. Apply creator cap (after enrichment, we have full metadata)

app/services/recommendation/utils.py

@@ -1,7 +1,3 @@
-"""
-Utility functions for recommendations.
-"""
-
 from typing import Any
 
 from loguru import logger
@@ -11,15 +7,6 @@
 
 
 def content_type_to_mtype(content_type: str) -> str:
-    """
-    Convert content_type to TMDB media type.
-
-    Args:
-        content_type: Content type (movie/series/tv)
-
-    Returns:
-        TMDB media type (movie/tv)
-    """
     return "tv" if content_type in ("tv", "series") else "movie"
 
 
@@ -103,10 +90,6 @@ def filter_by_genres(
 
         genre_ids = item.get("genre_ids", [])
 
-        # Genre whitelist check
-        if not RecommendationFiltering.passes_top_genre_whitelist(genre_ids, whitelist):
-            continue
-
         # Excluded genres check
         if excluded_ids and any(gid in excluded_ids for gid in genre_ids):
             continue
@@ -198,7 +181,6 @@ async def pad_to_min(
         tmdb_service,
         list(dedup.values()),
         content_type,
-        target_count=need * 2,
         user_settings=user_settings,
     )