Added filter for only show cached stream, Redis lock for prevent concurrent live scrape, bug fixes & improvements (#325)

* cleanup lifespan vars

* Show API process time in logs

* store the cache validation of full URL for live sports event

* Add support for filter only show cached streams

* Update default configs

* properly handle prowlarr exception on MaxProcessLimitReached

* Add redis lock mechanism to prevent concurrent live scrapers from multiple instances of stremio addon installation users

api/main.py

@@ -58,32 +58,31 @@ async def lifespan(fastapi_app: FastAPI):
     # Startup logic
     await database.init()
     await torrent.init_best_trackers()
+    scheduler = None
+    scheduler_lock = None
 
     if not settings.disable_all_scheduler:
-        acquired, lock = await acquire_scheduler_lock()
+        acquired, scheduler_lock = await acquire_scheduler_lock()
         if acquired:
             try:
                 scheduler = AsyncIOScheduler()
                 setup_scheduler(scheduler)
                 scheduler.start()
-                fastapi_app.state.scheduler = scheduler
-                fastapi_app.state.scheduler_lock = lock
                 await asyncio.create_task(maintain_heartbeat())
             except Exception as e:
-                await release_scheduler_lock(lock)
+                await release_scheduler_lock(scheduler_lock)
                 raise e
 
     yield
 
     # Shutdown logic
-    if hasattr(fastapi_app.state, "scheduler"):
-        fastapi_app.state.scheduler.shutdown(wait=False)
-
-    if (
-        hasattr(fastapi_app.state, "scheduler_lock")
-        and fastapi_app.state.scheduler_lock
-    ):
-        await release_scheduler_lock(fastapi_app.state.scheduler_lock)
+    if scheduler:
+        try:
+            scheduler.shutdown(wait=False)
+        except Exception as e:
+            logging.exception("Error shutting down scheduler")
+        finally:
+            await release_scheduler_lock(scheduler_lock)
 
     await REDIS_ASYNC_CLIENT.aclose()
 

api/middleware.py

@@ -42,12 +42,13 @@ class SecureLoggingMiddleware(BaseHTTPMiddleware):
     async def custom_log(request: Request, response: Response):
         ip = get_client_ip(request)
         url_path = str(request.url)
+        process_time = response.headers.get("X-Process-Time", "")
         if request.path_params.get("secret_str"):
             url_path = url_path.replace(
-                request.path_params.get("secret_str"), "***MASKED***"
+                request.path_params.get("secret_str"), "*MASKED*"
             )
         logging.info(
-            f'{ip} - "{request.method} {url_path} HTTP/1.1" {response.status_code}'
+            f'{ip} - "{request.method} {url_path} HTTP/1.1" {response.status_code} {process_time}'
         )
 
 

db/crud.py

@@ -30,6 +30,7 @@ from db.schemas import Stream, TorrentStreamsList
 from scrapers.utils import run_scrapers
 from scrapers.imdb_data import get_imdb_movie_data, search_imdb
 from utils import crypto
+from utils.lock import acquire_redis_lock, release_redis_lock
 from utils.parser import (
     fetch_downloaded_info_hashes,
     parse_stream_data,
@@ -319,6 +320,13 @@ async def get_movie_streams(
         return []
 
     cache_key = f"torrent_streams:{video_id}"
+    lock_key = f"{cache_key}_lock" if video_id.startswith("tt") else None
+    redis_lock = None
+
+    # Acquire the lock to prevent multiple scrapers from running at the same time
+    if lock_key:
+        _, redis_lock = await acquire_redis_lock(lock_key, timeout=60, block=True)
+
     cached_streams = await get_cached_torrent_streams(cache_key, video_id)
 
     if video_id.startswith("tt"):
@@ -331,7 +339,9 @@ async def get_movie_streams(
         if new_streams:
             # reset the cache
             await REDIS_ASYNC_CLIENT.delete(cache_key)
-            background_tasks.add_task(store_new_torrent_streams, new_streams)
+            background_tasks.add_task(
+                store_new_torrent_streams, new_streams, redis_lock=redis_lock
+            )
     else:
         all_streams = cached_streams
 
@@ -354,6 +364,13 @@ async def get_series_streams(
         return []
 
     cache_key = f"torrent_streams:{video_id}:{season}:{episode}"
+    lock_key = f"{cache_key}_lock" if video_id.startswith("tt") else None
+    redis_lock = None
+
+    # Acquire the lock to prevent multiple scrapers from running at the same time
+    if lock_key:
+        _, redis_lock = await acquire_redis_lock(lock_key, timeout=60, block=True)
+
     cached_streams = await get_cached_torrent_streams(
         cache_key, video_id, season, episode
     )
@@ -370,7 +387,9 @@ async def get_series_streams(
         if new_streams:
             # reset the cache
             await REDIS_ASYNC_CLIENT.delete(cache_key)
-            background_tasks.add_task(store_new_torrent_streams, new_streams)
+            background_tasks.add_task(
+                store_new_torrent_streams, new_streams, redis_lock
+            )
     else:
         all_streams = cached_streams
 
@@ -386,7 +405,7 @@ async def get_series_streams(
 
 
 async def store_new_torrent_streams(
-    streams: list[TorrentStreams] | set[TorrentStreams],
+    streams: list[TorrentStreams] | set[TorrentStreams], redis_lock=None
 ):
     if not streams:
         return
@@ -439,6 +458,8 @@ async def store_new_torrent_streams(
             )
 
     await bulk_writer.commit()
+    if redis_lock:
+        await release_redis_lock(redis_lock)
 
 
 async def get_tv_streams(video_id: str, namespace: str, user_data) -> list[Stream]:

db/schemas.py

@@ -144,6 +144,7 @@ class StreamingProvider(BaseModel):
     enable_watchlist_catalogs: bool = Field(default=True, alias="ewc")
     qbittorrent_config: QBittorrentConfig | None = Field(default=None, alias="qbc")
     download_via_browser: bool = Field(default=False, alias="dvb")
+    only_show_cached_streams: bool = Field(default=False, alias="oscs")
 
     @model_validator(mode="after")
     def validate_token_or_username_password(self) -> "StreamingProvider":
@@ -167,7 +168,13 @@ class StreamingProvider(BaseModel):
 class UserData(BaseModel):
     streaming_provider: StreamingProvider | None = Field(default=None, alias="sp")
     selected_catalogs: list[str] = Field(
-        default=["prowlarr_streams", "torrentio_streams", "zilean_dmm_streams"],
+        default=[
+            "prowlarr_streams",
+            "torrentio_streams",
+            "zilean_dmm_streams",
+            "english_hdrip",
+            "english_series",
+        ],
         alias="sc",
     )
     selected_resolutions: list[str | None] = Field(
@@ -176,7 +183,7 @@ class UserData(BaseModel):
     enable_catalogs: bool = Field(default=True, alias="ec")
     enable_imdb_metadata: bool = Field(default=True, alias="eim")
     max_size: int | str | float = Field(default=math.inf, alias="ms")
-    max_streams_per_resolution: int = Field(default=3, alias="mspr")
+    max_streams_per_resolution: int = Field(default=10, alias="mspr")
     show_full_torrent_name: bool = Field(default=True, alias="sftn")
     torrent_sorting_priority: list[str] = Field(
         default=const.TORRENT_SORTING_PRIORITY, alias="tsp"

resources/html/configure.html

@@ -227,6 +227,13 @@
                                 </label>
                             </div>
                         {% endif %}
+                        <div class="form-check mb-2">
+                            <input class="form-check-input" type="checkbox" name="only_show_cached_streams"
+                                   id="only_show_cached_streams" {% if user_data.streaming_provider and user_data.streaming_provider.only_show_cached_streams %}checked{% endif %}>
+                            <label class="form-check-label" for="only_show_cached_streams">
+                                Only Show Cached Streams
+                            </label>
+                        </div>
                     </div>
 
                     <!-- Enable Catalogs Checkbox -->

resources/js/config_script.js

@@ -303,6 +303,7 @@ function getUserData() {
         if (document.getElementById('download_via_browser')) {
             streamingProviderData.download_via_browser = document.getElementById('download_via_browser').checked;
         }
+        streamingProviderData.only_show_cached_streams = document.getElementById('only_show_cached_streams').checked;
     } else {
         streamingProviderData = null;
     }

scrapers/prowlarr.py

@@ -303,12 +303,18 @@ class ProwlarrScraper(BaseScraper):
                 f"Stream processing timed out after {max_process_time} seconds. "
                 f"Processed {streams_processed} streams"
             )
-        except MaxProcessLimitReached:
-            self.logger.info(
-                f"Stream processing cancelled after reaching max process limit of {max_process}"
-            )
+        except ExceptionGroup as eg:
+            for e in eg.exceptions:
+                if isinstance(e, MaxProcessLimitReached):
+                    self.logger.info(
+                        f"Stream processing cancelled after reaching max process limit of {max_process}"
+                    )
+                else:
+                    self.logger.exception(
+                        f"An error occurred during stream processing: {e}"
+                    )
         except Exception as e:
-            self.logger.error(f"An error occurred during stream processing: {e}")
+            self.logger.exception(f"An error occurred during stream processing: {e}")
         self.logger.info(
             f"Finished processing {streams_processed} streams from "
             f"{len(stream_generators)} generators"

utils/parser.py

@@ -95,6 +95,9 @@ async def filter_and_sort_streams(
                     f"Failed to update cache status for {user_data.streaming_provider.service}: {error}"
                 )
 
+        if user_data.streaming_provider.only_show_cached_streams:
+            filtered_streams = [stream for stream in filtered_streams if stream.cached]
+
     # Step 3: Dynamically sort streams based on user preferences
     def dynamic_sort_key(stream: TorrentStreams) -> tuple:
         def key_value(key: str) -> Any:

utils/validation_helper.py

@@ -1,7 +1,6 @@
 import asyncio
 import json
 import logging
-from urllib import parse
 from urllib.parse import urlparse
 
 import httpx
@@ -70,13 +69,13 @@ async def validate_live_stream_url(
 
 async def validate_m3u8_or_mpd_url_with_cache(url: str, behaviour_hint: dict):
     try:
-        cache_key = f"m3u8_url:{parse.urlparse(url).netloc}"
+        cache_key = f"m3u8_url:{url}"
         cache_data = await REDIS_ASYNC_CLIENT.get(cache_key)
         if cache_data:
             return json.loads(cache_data)
 
         is_valid = await validate_live_stream_url(url, behaviour_hint)
-        await REDIS_ASYNC_CLIENT.set(cache_key, json.dumps(is_valid), ex=180)
+        await REDIS_ASYNC_CLIENT.set(cache_key, json.dumps(is_valid), ex=300)
         return is_valid
     except Exception as e:
         logging.exception(e)