Add Jackett scraper and enhance Prowlarr scraper integration & Refactorings common codes

Introduced a new Jackett scraper to support torrent indexers and refactored Prowlarr scraper for consistency. Enhanced support for advanced individual indexer searching, searching capability with category validation, and healthy indexer management across both scrapers.

db/config.py

@@ -56,12 +56,13 @@ class Settings(BaseSettings):
     tmdb_api_key: str | None = None
 
     # Prowlarr Settings
+    is_scrap_from_prowlarr: bool = True
     prowlarr_url: str = "http://prowlarr-service:9696"
     prowlarr_api_key: str | None = None
-    prowlarr_live_title_search: bool = False
+    prowlarr_live_title_search: bool = True
     prowlarr_background_title_search: bool = True
-    prowlarr_search_query_timeout: int = 15
-    prowlarr_search_interval_hour: int = 24
+    prowlarr_search_query_timeout: int = 30
+    prowlarr_search_interval_hour: int = 72
     prowlarr_immediate_max_process: int = 10
     prowlarr_immediate_max_process_time: int = 15
     prowlarr_feed_scrape_interval_hour: int = 3
@@ -88,6 +89,23 @@ class Settings(BaseSettings):
     bt4g_search_timeout: int = 15
     bt4g_immediate_max_process: int = 30
     bt4g_immediate_max_process_time: int = 150000
+    bt4g_search_interval_hour: int = 72
+    bt4g_search_timeout: int = 10
+    bt4g_immediate_max_process: int = 15
+    bt4g_immediate_max_process_time: int = 15
+
+    # Jackett Settings
+    is_scrap_from_jackett: bool = True
+    jackett_url: str = "http://jackett-service:9117"
+    jackett_api_key: str | None = None
+    jackett_search_interval_hour: int = 72
+    jackett_search_query_timeout: int = 30
+    jackett_immediate_max_process: int = 10
+    jackett_immediate_max_process_time: int = 15
+    jackett_live_title_search: bool = True
+    jackett_background_title_search: bool = True
+    jackett_feed_scrape_interval_hour: int = 3
+
 
     # Premiumize Settings
     premiumize_oauth_client_id: str | None = None

db/crud.py

@@ -28,7 +28,7 @@ from db.models import (
 )
 from db.schemas import Stream, TorrentStreamsList
 from scrapers.tmdb_data import search_tmdb
-from scrapers.utils import run_scrapers
+from scrapers.scraper_tasks import run_scrapers
 from scrapers.imdb_data import get_imdb_title, search_imdb
 from streaming_providers.cache_helpers import store_cached_info_hashes
 from utils import crypto

scrapers/base_scraper.py

@@ -1,4 +1,6 @@
 import abc
+import asyncio
+import json
 import logging
 import time
 from collections import Counter
@@ -6,17 +8,31 @@ from dataclasses import dataclass, field
 from datetime import datetime
 from datetime import timedelta
 from functools import wraps
-from typing import Any
-from typing import Dict, List, Optional
+from typing import Any, Literal, AsyncGenerator, AsyncIterable
+from typing import Dict
+from typing import List
+from typing import Optional
 
+import PTT
 import httpx
 from ratelimit import limits, sleep_and_retry
 from tenacity import retry, stop_after_attempt, wait_exponential
+from torf import Magnet, MagnetError
 
 from db.config import settings
-from db.models import TorrentStreams, MediaFusionMetaData
-from utils.parser import calculate_max_similarity_ratio
+from db.models import MediaFusionMovieMetaData, MediaFusionSeriesMetaData
+from db.models import (
+    TorrentStreams,
+    MediaFusionMetaData,
+    Episode,
+    Season,
+)
 from db.redis_database import REDIS_ASYNC_CLIENT
+from scrapers import torrent_info
+from scrapers.imdb_data import get_episode_by_date
+from utils.network import batch_process_with_circuit_breaker, CircuitBreaker
+from utils.parser import calculate_max_similarity_ratio, is_contain_18_plus_keywords
+from utils.torrent import extract_torrent_metadata, info_hashes_to_torrent_metadata
 
 
 @dataclass
@@ -280,6 +296,103 @@ class BaseScraper(abc.ABC):
             self.metrics.stop()
             self.metrics.log_summary(self.logger)
 
+    async def process_streams(
+        self,
+        *stream_generators: AsyncGenerator[TorrentStreams, None],
+        max_process: int = None,
+        max_process_time: int = None,
+        catalog_type: str = None,
+        season: int = None,
+        episode: int = None,
+    ) -> AsyncGenerator[TorrentStreams, None]:
+        """
+        Process streams from multiple generators and yield them as they become available.
+        """
+        queue = asyncio.Queue()
+        streams_processed = 0
+        active_generators = len(stream_generators)
+        processed_info_hashes = set()
+
+        async def producer(gen: AsyncIterable, generator_id: int):
+            try:
+                async for stream_item in gen:
+                    await queue.put((stream_item, generator_id))
+                    self.logger.debug("Generator % produced a stream", generator_id)
+            except Exception as err:
+                self.logger.exception(f"Error in generator {generator_id}: {err}")
+            finally:
+                await queue.put(("DONE", generator_id))
+                self.logger.debug("Generator %s finished", generator_id)
+
+        async def queue_processor():
+            nonlocal active_generators, streams_processed
+            while active_generators > 0:
+                item, gen_id = await queue.get()
+
+                if item == "DONE":
+                    active_generators -= 1
+                    self.logger.debug(
+                        "Generator %s completed. %s generators remaining",
+                        gen_id,
+                        active_generators,
+                    )
+                elif (
+                    isinstance(item, TorrentStreams)
+                    and item.id not in processed_info_hashes
+                ):
+                    processed_info_hashes.add(item.id)
+                    if (
+                        catalog_type != "series"
+                        or item.get_episode(season, episode) is not None
+                    ):
+                        streams_processed += 1
+                    self.logger.debug(
+                        f"Processed stream from generator {gen_id}. Total streams processed: {streams_processed}"
+                    )
+                    yield item
+
+                if max_process and streams_processed >= max_process:
+                    self.logger.info(f"Reached max process limit of {max_process}")
+                    raise MaxProcessLimitReached("Max process limit reached")
+
+        try:
+            async with asyncio.timeout(max_process_time):
+                async with asyncio.TaskGroup() as tg:
+                    # Create tasks for each stream generator
+                    [
+                        tg.create_task(producer(gen, i))
+                        for i, gen in enumerate(stream_generators)
+                    ]
+
+                    # Yield items as they become available
+                    async for stream in queue_processor():
+                        yield stream
+        except asyncio.TimeoutError:
+            self.logger.warning(
+                f"Stream processing timed out after {max_process_time} seconds. "
+                f"Processed {streams_processed} streams"
+            )
+            self.metrics.record_skip("Max process time")
+        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}"
+                    )
+                    self.metrics.record_skip("Max process limit")
+                else:
+                    self.logger.exception(
+                        f"An error occurred during stream processing: {e}"
+                    )
+                    self.metrics.record_error(f"unexpected_stream_processing_error {e}")
+        except Exception as e:
+            self.logger.exception(f"An error occurred during stream processing: {e}")
+            self.metrics.record_error(f"unexpected_stream_processing_error {e}")
+        self.logger.info(
+            f"Finished processing {streams_processed} streams from "
+            f"{len(stream_generators)} generators"
+        )
+
     @abc.abstractmethod
     async def _scrape_and_parse(self, *args, **kwargs) -> List[TorrentStreams]:
         """
@@ -412,7 +525,7 @@ class BaseScraper(abc.ABC):
     def validate_title_and_year(
         self,
         parsed_data: dict,
-        metadata: MediaFusionMetaData,
+        metadata: MediaFusionMovieMetaData | MediaFusionSeriesMetaData,
         catalog_type: str,
         torrent_title: str,
         expected_ratio: int = 87,

scrapers/jackett.py

@@ -0,0 +1,322 @@
+from datetime import timedelta
+from typing import List, Dict, Any, Literal, Optional
+from xml.etree import ElementTree
+
+import httpx
+
+from db.config import settings
+from db.models import (
+    TorrentStreams,
+    MediaFusionMetaData,
+)
+from scrapers.base_scraper import IndexerBaseScraper
+from utils.network import CircuitBreaker
+from utils.runtime_const import JACKETT_SEARCH_TTL
+
+
+class JackettScraper(IndexerBaseScraper):
+    cache_key_prefix = "jackett"
+    search_url = "/api/v2.0/indexers/all/results"
+
+    def __init__(self):
+        super().__init__(
+            cache_key_prefix=self.cache_key_prefix,
+            base_url=settings.jackett_url,
+        )
+
+    @property
+    def live_title_search_enabled(self) -> bool:
+        return settings.jackett_live_title_search
+
+    @property
+    def background_title_search_enabled(self) -> bool:
+        return settings.jackett_background_title_search
+
+    @property
+    def immediate_max_process(self) -> int:
+        return settings.jackett_immediate_max_process
+
+    @property
+    def immediate_max_process_time(self) -> int:
+        return settings.jackett_immediate_max_process_time
+
+    @property
+    def search_query_timeout(self) -> int:
+        return settings.jackett_search_query_timeout
+
+    def get_info_hash(self, item: dict) -> str:
+        return item.get("InfoHash")
+
+    def get_guid(self, item: dict) -> str:
+        return item.get("Guid")
+
+    def get_title(self, item: dict) -> str:
+        return item.get("Title")
+
+    def get_imdb_id(self, item: dict) -> str:
+        return item.get("Imdb")
+
+    def get_category_ids(self, item: dict) -> List[int]:
+        return item["Category"]
+
+    def get_magent_link(self, item: dict) -> str:
+        return item.get("MagnetUri")
+
+    def get_download_link(self, item: dict) -> str:
+        return item.get("Link")
+
+    def get_info_url(self, item: dict) -> str:
+        return item.get("Details")
+
+    def get_indexer(self, item: dict) -> str:
+        return item.get("Tracker")
+
+    @IndexerBaseScraper.cache(ttl=JACKETT_SEARCH_TTL)
+    @IndexerBaseScraper.rate_limit(calls=5, period=timedelta(seconds=1))
+    async def _scrape_and_parse(
+        self,
+        metadata: MediaFusionMetaData,
+        catalog_type: str,
+        season: int = None,
+        episode: int = None,
+    ) -> List[TorrentStreams]:
+        """Scrape and parse Jackett indexers for torrent streams"""
+        return await super()._scrape_and_parse(metadata, catalog_type, season, episode)
+
+    async def get_healthy_indexers(self) -> List[dict]:
+        """Fetch and return list of healthy Jackett indexers with their capabilities"""
+        try:
+            response = await self.http_client.get(
+                f"{self.base_url}/api/v2.0/indexers/!status:failing/results/torznab/api",
+                params={
+                    "apikey": settings.jackett_api_key,
+                    "t": "indexers",
+                    "configured": "true",
+                },
+                timeout=15,
+            )
+            response.raise_for_status()
+
+            # Parse XML response
+
+            root = ElementTree.fromstring(response.text)
+
+            healthy_indexers = []
+            for indexer in root.findall("indexer"):
+                indexer_id = indexer.get("id")
+                if not indexer_id:
+                    continue
+
+                # Get basic indexer info
+                title = indexer.find("title").text
+                description = (
+                    indexer.find("description").text
+                    if indexer.find("description") is not None
+                    else ""
+                )
+
+                # Get searching capabilities
+                caps = indexer.find("caps")
+                if caps is None:
+                    continue
+
+                searching = caps.find("searching")
+                if searching is None:
+                    continue
+
+                # Get category information
+                categories = []
+                cats_elem = caps.find("categories")
+                if cats_elem is not None:
+                    for cat in cats_elem.findall(".//category"):
+                        cat_id = int(cat.get("id"))
+                        categories.append(cat_id)
+                        # Add subcategories
+                        for subcat in cat.findall("subcat"):
+                            categories.append(int(subcat.get("id")))
+
+                # Parse search capabilities
+                search_caps = {}
+                for search_type in ["search", "tv-search", "movie-search"]:
+                    search_elem = searching.find(search_type)
+                    if (
+                        search_elem is not None
+                        and search_elem.get("available") == "yes"
+                    ):
+                        supported_params = search_elem.get("supportedParams", "").split(
+                            ","
+                        )
+                        search_caps[search_type] = supported_params
+
+                indexer_info = {
+                    "id": indexer_id,
+                    "name": title,
+                    "description": description,
+                    "categories": categories,
+                    "search_capabilities": search_caps,
+                }
+
+                # Initialize circuit breaker
+                self.indexer_circuit_breakers[indexer_id] = CircuitBreaker(
+                    failure_threshold=3,
+                    recovery_timeout=300,
+                    half_open_attempts=1,
+                )
+
+                # Store indexer status
+                self.indexer_status[indexer_id] = {
+                    "is_healthy": True,  # If it's in the list, it's configured and healthy
+                    "name": title,
+                    "description": description,
+                }
+
+                healthy_indexers.append(indexer_info)
+
+            self.logger.info(f"Found {len(healthy_indexers)} healthy indexers")
+            return healthy_indexers
+
+        except Exception as e:
+            self.logger.error(f"Failed to determine healthy indexers: {e}")
+            return []
+
+    async def fetch_search_results(
+        self,
+        params: dict,
+        indexer_ids: List[int],
+        timeout: Optional[int] = None,
+    ) -> List[Dict[str, Any]]:
+        """Fetch search results from Jackett indexers with circuit breaker handling"""
+        results = []
+        timeout = timeout or self.search_query_timeout
+
+        for indexer_id in indexer_ids:
+            indexer_status = self.indexer_status.get(indexer_id, {})
+            if not indexer_status.get("is_healthy", False):
+                continue
+
+            circuit_breaker = self.indexer_circuit_breakers.get(indexer_id)
+            if not circuit_breaker:
+                continue
+
+            indexer_name = indexer_status.get("name", f"ID:{indexer_id}")
+
+            if circuit_breaker.is_closed():
+                try:
+                    search_params = {
+                        **params,
+                        "Tracker[]": [indexer_id],
+                        "apikey": settings.jackett_api_key,
+                    }
+                    response = await self.http_client.get(
+                        f"{self.base_url}{self.search_url}",
+                        params=search_params,
+                        timeout=timeout,
+                    )
+                    response.raise_for_status()
+                    indexer_results = response.json().get("Results", [])
+
+                    circuit_breaker.record_success()
+                    self.metrics.record_indexer_success(
+                        indexer_name, len(indexer_results)
+                    )
+                    results.extend(indexer_results)
+
+                except Exception as e:
+                    error_msg = f"Error searching indexer {indexer_name}: {str(e)}"
+                    self.logger.error(error_msg)
+
+                    circuit_breaker.record_failure()
+                    self.metrics.record_indexer_error(indexer_name, str(e))
+
+                    if not circuit_breaker.is_closed():
+                        self.logger.warning(
+                            f"Circuit breaker opened for indexer {indexer_name}. "
+                            f"Status: {circuit_breaker.get_status()}"
+                        )
+                        indexer_status["is_healthy"] = False
+                        self.indexer_status[indexer_id] = indexer_status
+            else:
+                self.logger.debug(
+                    f"Skipping indexer {indexer_name} - circuit breaker is {circuit_breaker.state}"
+                )
+                self.metrics.record_indexer_error(
+                    indexer_name, f"Circuit breaker {circuit_breaker.state}"
+                )
+
+        return results
+
+    async def build_search_params(
+        self,
+        video_id: str,
+        search_type: Literal["search", "tvsearch", "movie"],
+        categories: list[int],
+        search_query: str = None,
+    ) -> dict:
+        """Build search parameters for Jackett API"""
+        params = {
+            "Category[]": categories,
+        }
+
+        if search_type in ["movie", "tvsearch"]:
+            params["imdbid"] = video_id
+        else:
+            params["Query"] = search_query
+
+        return params
+
+    async def parse_indexer_data(
+        self, indexer_data: dict, catalog_type: str, parsed_data: dict
+    ) -> dict | None:
+        """Parse Jackett-specific indexer data"""
+        if not self.validate_category_with_title(indexer_data):
+            return None
+
+        download_url = indexer_data.get("MagnetUri") or indexer_data.get("Link")
+        if not download_url:
+            return None
+        download_url = await self.get_download_url(indexer_data)
+        if not download_url:
+            return None
+
+        try:
+            torrent_data, is_torrent_downloaded = await self.get_torrent_data(
+                download_url, indexer_data.get("Tracker"), parsed_data
+            )
+        except httpx.HTTPStatusError as error:
+            if error.response.status_code in [429, 500]:
+                raise error
+            self.logger.error(
+                f"HTTP Error getting torrent data: {download_url}, status code: {error.response.status_code}"
+            )
+            return None
+        except httpx.TimeoutException as error:
+            self.logger.warning("Timeout while getting torrent data")
+            raise error
+        except httpx.RequestError as error:
+            self.logger.error(f"Request error getting torrent data: {error}")
+            raise error
+        except Exception as e:
+            self.logger.exception(f"Error getting torrent data: {e}")
+            return None
+
+        info_hash = torrent_data.get("info_hash", "").lower()
+        if not info_hash:
+            return None
+
+        torrent_data.update(
+            {
+                "info_hash": info_hash,
+                "seeders": indexer_data.get("Seeders"),
+                "created_at": indexer_data.get("PublishDate"),
+                "source": indexer_data.get("Tracker"),
+                "catalog": [
+                    "jackett_streams",
+                    f"jackett_{catalog_type.rstrip('s')}s",
+                ],
+                "total_size": torrent_data.get("total_size")
+                or indexer_data.get("Size"),
+                **parsed_data,
+            }
+        )
+
+        return torrent_data

utils/runtime_const.py

@@ -50,3 +50,6 @@ YTS_SEARCH_TTL = 259200  # 3 days in seconds
 BT4G_SEARCH_TTL = int(
     timedelta(hours=settings.bt4g_search_interval_hour).total_seconds()
 )
+JACKETT_SEARCH_TTL = int(
+    timedelta(hours=settings.jackett_search_interval_hour).total_seconds()
+)