diff --git a/Pipfile b/Pipfile index 2cb111d..c9dc765 100644 --- a/Pipfile +++ b/Pipfile @@ -36,7 +36,6 @@ aiowebdav = "*" m3u-ipytv = "*" python-multipart = "*" prometheus-client = "*" -curl-cffi = "*" pyasynctracker = "*" humanize = {git = "git+https://github.com/python-humanize/humanize.git"} scrapy-playwright = "*" @@ -48,6 +47,8 @@ ratelimit = "*" qrcode = "*" aioseedrcc = "*" pytz = "*" +aiohttp-socks = "*" +socksio = "*" [dev-packages] pysocks = "*" diff --git a/api/main.py b/api/main.py index 71a1a54..069b698 100644 --- a/api/main.py +++ b/api/main.py @@ -6,6 +6,7 @@ from io import BytesIO from typing import Literal, Annotated import aiohttp +import httpx from apscheduler.schedulers.asyncio import AsyncIOScheduler from fastapi import ( Depends, @@ -140,7 +141,7 @@ async def get_home(request: Request): @wrappers.exclude_rate_limit async def health(): start_time = asyncio.get_event_loop().time() - async with aiohttp.ClientSession() as session: + async with httpx.AsyncClient(proxy=settings.requests_proxy_url) as session: try: async with session.head("https://www.google.com", timeout=10) as response: return { diff --git a/db/config.py b/db/config.py index 1f123d6..2b0a9fd 100644 --- a/db/config.py +++ b/db/config.py @@ -47,7 +47,7 @@ class Settings(BaseSettings): redis_max_connections: int = 100 # External Service URLs - scraper_proxy_url: str | None = None + requests_proxy_url: str | None = None playwright_cdp_url: str = "ws://browserless:3000?blockAds=true&stealth=true" flaresolverr_url: str = "http://flaresolverr:8191/v1" diff --git a/docs/configuration.md b/docs/configuration.md index b7eaa5a..99a7974 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -33,7 +33,7 @@ These settings control the database and caching behavior of MediaFusion. These URLs define the locations of various external services used by MediaFusion. - **poster_host_url** (default: Use the Host URL value): The URL where poster images are served from. Use the same value as `host_url` if posters are served from the same location. -- **scraper_proxy_url**: The proxy URL for the scraper, if any. +- **requests_proxy_url**: The URL for forwarding requests through a proxy. This setting is optional. - **torrentio_url** (default: `"https://torrentio.strem.fun"`): The Torrentio / KightCrawler URL. - **mediafusion_url** (default: `"https://mediafusion.elfhosted.com"`): The Mediafusion URL. - **playwright_cdp_url** (default: `"ws://browserless:3000?blockAds=true&stealth=true"`): The URL for the Playwright CDP (Chrome DevTools Protocol) service. diff --git a/scrapers/helpers.py b/scrapers/helpers.py index c8252d6..f186473 100644 --- a/scrapers/helpers.py +++ b/scrapers/helpers.py @@ -4,7 +4,7 @@ from datetime import datetime import dramatiq from bs4 import BeautifulSoup -from curl_cffi.requests import AsyncSession +import httpx from db.config import settings from db.models import TorrentStreams, Episode, Season @@ -14,16 +14,6 @@ from utils.torrent import info_hashes_to_torrent_metadata logging.getLogger("httpx").setLevel(logging.WARNING) -PROXIES = ( - { - "http": settings.scraper_proxy_url, - "https": settings.scraper_proxy_url, - } - if settings.scraper_proxy_url - else None -) - - @dramatiq.actor(time_limit=30 * 60 * 1000, priority=10) async def update_torrent_movie_streams_metadata( info_hashes: list[str], tracker: list[str] = None @@ -101,8 +91,8 @@ def get_country_name(country_code): async def get_page_bs4(url: str): try: - async with AsyncSession(proxies=PROXIES) as session: - response = await session.get(url, impersonate="chrome", timeout=30) + async with httpx.AsyncClient(proxy=settings.requests_proxy_url) as session: + response = await session.get(url, timeout=10) if response.status_code != 200: return None return BeautifulSoup(response.text, "html.parser") diff --git a/scrapers/mediafusion.py b/scrapers/mediafusion.py index e976423..0442772 100644 --- a/scrapers/mediafusion.py +++ b/scrapers/mediafusion.py @@ -2,6 +2,7 @@ from datetime import timedelta from typing import Dict, Any, Optional, List import PTT +import httpx from db.config import settings from db.models import MediaFusionMetaData, TorrentStreams @@ -11,12 +12,17 @@ from utils.runtime_const import MEDIAFUSION_SEARCH_TTL class MediafusionScraper(StremioScraper): + cache_key_prefix = "mediafusion" + def __init__(self): super().__init__( - cache_key_prefix="mediafusion", + cache_key_prefix=self.cache_key_prefix, base_url=settings.mediafusion_url, logger_name=__name__, ) + self.http_client = httpx.AsyncClient( + timeout=30, proxy=settings.requests_proxy_url + ) @StremioScraper.cache(ttl=MEDIAFUSION_SEARCH_TTL) @StremioScraper.rate_limit(calls=5, period=timedelta(seconds=1)) diff --git a/scrapers/prowlarr.py b/scrapers/prowlarr.py index 0c0401c..953f6f6 100644 --- a/scrapers/prowlarr.py +++ b/scrapers/prowlarr.py @@ -74,6 +74,7 @@ class ProwlarrScraper(BaseScraper): super().__init__(cache_key_prefix=self.cache_key_prefix, logger_name=__name__) self.indexer_status = {} self.indexer_circuit_breakers = {} + self.http_client = httpx.AsyncClient(timeout=30, headers=self.headers) async def get_healthy_indexers(self) -> List[int]: """Fetch and return list of healthy indexer IDs with detailed health checks""" @@ -439,21 +440,20 @@ class ProwlarrScraper(BaseScraper): if circuit_breaker.is_closed(): try: search_params = {**params, "indexerIds": [indexer_id]} - async with httpx.AsyncClient(headers=self.headers) as client: - response = await client.get( - self.search_url, - params=search_params, - timeout=timeout, - ) - response.raise_for_status() - indexer_results = response.json() + response = await self.http_client.get( + self.search_url, + params=search_params, + timeout=timeout, + ) + response.raise_for_status() + indexer_results = response.json() - # Record success - circuit_breaker.record_success() - self.metrics.record_indexer_success( - indexer_name, len(indexer_results) - ) - results.extend(indexer_results) + # Record success + 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)}" @@ -842,12 +842,11 @@ class ProwlarrScraper(BaseScraper): async def fetch_indexers(self): try: - async with httpx.AsyncClient(headers=self.headers) as client: - response = await client.get( - self.base_url + "/api/v1/indexer", timeout=10 - ) - response.raise_for_status() - return response.json() + response = await self.http_client.get( + self.base_url + "/api/v1/indexer", timeout=10 + ) + response.raise_for_status() + return response.json() except Exception as e: self.logger.exception(f"Failed to fetch indexers: {e}") return [] @@ -855,22 +854,21 @@ class ProwlarrScraper(BaseScraper): async def fetch_indexer_statuses(self) -> Dict[int, Dict]: """Fetch current status information for all indexers""" try: - async with httpx.AsyncClient(headers=self.headers) as client: - response = await client.get( - f"{self.base_url}/api/v1/indexerstatus", timeout=10 - ) - response.raise_for_status() + response = await self.http_client.get( + f"{self.base_url}/api/v1/indexerstatus", timeout=10 + ) + response.raise_for_status() - # Create a mapping of indexerId to status info - status_data = {} - for status in response.json(): - indexer_id = status.get("indexerId") - if indexer_id: - status_data[indexer_id] = { - "disabled_till": status.get("disabledTill"), - "most_recent_failure": status.get("mostRecentFailure"), - "initial_failure": status.get("initialFailure"), - } + # Create a mapping of indexerId to status info + status_data = {} + for status in response.json(): + indexer_id = status.get("indexerId") + if indexer_id: + status_data[indexer_id] = { + "disabled_till": status.get("disabledTill"), + "most_recent_failure": status.get("mostRecentFailure"), + "initial_failure": status.get("initialFailure"), + } return status_data except Exception as e: @@ -892,22 +890,21 @@ class ProwlarrScraper(BaseScraper): return {}, False return {"info_hash": magnet.infohash, "announce_list": magnet.tr}, False - async with httpx.AsyncClient() as client: - response = await client.get( - download_url, - follow_redirects=False, - timeout=settings.prowlarr_search_query_timeout, + response = await self.http_client.get( + download_url, + follow_redirects=False, + timeout=settings.prowlarr_search_query_timeout, + ) + if response.status_code in [301, 302, 303, 307, 308]: + redirect_url = response.headers.get("Location") + return await self.get_torrent_data(redirect_url, indexer) + response.raise_for_status() + if response.headers.get("Content-Type") == "application/x-bittorrent": + return ( + extract_torrent_metadata(response.content, is_parse_ptt=False), + True, ) - if response.status_code in [301, 302, 303, 307, 308]: - redirect_url = response.headers.get("Location") - return await self.get_torrent_data(redirect_url, indexer) - response.raise_for_status() - if response.headers.get("Content-Type") == "application/x-bittorrent": - return ( - extract_torrent_metadata(response.content, is_parse_ptt=False), - True, - ) - return {}, False + return {}, False @staticmethod def parse_title_data(title: str) -> dict: diff --git a/scrapers/rpdb.py b/scrapers/rpdb.py index 1708c7e..7e055c5 100644 --- a/scrapers/rpdb.py +++ b/scrapers/rpdb.py @@ -5,6 +5,7 @@ import time import httpx from db import schemas +from db.config import settings from db.redis_database import REDIS_ASYNC_CLIENT RPDB_SUPPORTED_SET = "rpdb_supported_ids" @@ -14,7 +15,9 @@ RPDB_UNSUPPORTED_EXPIRY = 60 * 60 * 24 * 7 # 7 days in seconds async def check_rpdb_poster_availability(rpdb_poster_url: str) -> bool: try: - async with httpx.AsyncClient(timeout=10.0) as client: + async with httpx.AsyncClient( + timeout=10.0, proxy=settings.requests_proxy_url + ) as client: response = await client.head(rpdb_poster_url) return response.status_code == 200 except httpx.HTTPError as exc: diff --git a/scrapers/torrentio.py b/scrapers/torrentio.py index d47f271..f624e8b 100644 --- a/scrapers/torrentio.py +++ b/scrapers/torrentio.py @@ -2,6 +2,7 @@ from datetime import timedelta from typing import Dict, Any, List import PTT +import httpx from db.config import settings from db.models import MediaFusionMetaData, TorrentStreams @@ -13,12 +14,17 @@ from utils.runtime_const import TORRENTIO_SEARCH_TTL class TorrentioScraper(StremioScraper): + cache_key_prefix = "torrentio" + def __init__(self): super().__init__( - cache_key_prefix="torrentio", + cache_key_prefix=self.cache_key_prefix, base_url=settings.torrentio_url, logger_name=__name__, ) + self.http_client = httpx.AsyncClient( + timeout=30, proxy=settings.requests_proxy_url + ) @StremioScraper.cache(ttl=TORRENTIO_SEARCH_TTL) @StremioScraper.rate_limit(calls=5, period=timedelta(seconds=1)) diff --git a/scrapers/tv.py b/scrapers/tv.py index dc7bd4c..d6c4ad9 100644 --- a/scrapers/tv.py +++ b/scrapers/tv.py @@ -11,6 +11,7 @@ from ipytv import playlist from ipytv.channel import IPTVAttr from db import schemas, crud +from db.config import settings from db.models import TVStreams, MediaFusionTVMetaData from utils import validation_helper from utils.parser import is_contain_18_plus_keywords @@ -185,7 +186,7 @@ async def update_tv_posters_in_db(*args, **kwargs): logging.info("No TV posters to update.") return - async with httpx.AsyncClient() as client: + async with httpx.AsyncClient(proxy=settings.requests_proxy_url) as client: response = await client.get( "https://iptv-org.github.io/api/channels.json", timeout=30 ) diff --git a/streaming_providers/debrid_client.py b/streaming_providers/debrid_client.py index 7f17951..ae8cfcf 100644 --- a/streaming_providers/debrid_client.py +++ b/streaming_providers/debrid_client.py @@ -7,7 +7,9 @@ from typing import Optional, Dict, Union import aiohttp from aiohttp import ClientResponse, ClientTimeout, ContentTypeError +from aiohttp_socks import ProxyConnector +from db.config import settings from streaming_providers.exceptions import ProviderException @@ -22,9 +24,11 @@ class DebridClient(AsyncContextDecorator): @property def session(self) -> aiohttp.ClientSession: if self._session is None: + connector = aiohttp.TCPConnector(ttl_dns_cache=300) + if settings.requests_proxy_url: + connector = ProxyConnector.from_url(settings.requests_proxy_url) self._session = aiohttp.ClientSession( - timeout=self._timeout, - connector=aiohttp.TCPConnector(ttl_dns_cache=300), + timeout=self._timeout, connector=connector ) return self._session diff --git a/streaming_providers/pikpak/utils.py b/streaming_providers/pikpak/utils.py index c99fc80..3812b62 100644 --- a/streaming_providers/pikpak/utils.py +++ b/streaming_providers/pikpak/utils.py @@ -3,10 +3,10 @@ import logging from contextlib import asynccontextmanager from datetime import datetime -import aiohttp import httpx from pikpakapi import PikPakApi, PikpakException +from db.config import settings from db.models import TorrentStreams from db.schemas import UserData from streaming_providers.exceptions import ProviderException @@ -154,6 +154,7 @@ async def initialize_pikpak(user_data: UserData): httpx_client_args={ "transport": httpx.AsyncHTTPTransport(retries=3), "timeout": 10, + "proxy": settings.requests_proxy_url, }, token_refresh_callback=store_pikpak_token_in_cache, token_refresh_callback_kwargs={"user_data": user_data}, @@ -165,6 +166,7 @@ async def initialize_pikpak(user_data: UserData): httpx_client_args={ "transport": httpx.AsyncHTTPTransport(retries=3), "timeout": 10, + "proxy": settings.requests_proxy_url, }, token_refresh_callback=store_pikpak_token_in_cache, token_refresh_callback_kwargs={"user_data": user_data}, @@ -182,7 +184,7 @@ async def initialize_pikpak(user_data: UserData): "Failed to connect to PikPak. Please try again later.", "debrid_service_down_error.mp4", ) - except (aiohttp.ClientError, httpx.ReadTimeout): + except httpx.ReadTimeout: raise ProviderException( "Failed to connect to PikPak. Please try again later.", "debrid_service_down_error.mp4", diff --git a/utils/network.py b/utils/network.py index bf5bed3..6ce6242 100644 --- a/utils/network.py +++ b/utils/network.py @@ -7,6 +7,7 @@ from urllib.parse import urlencode, urlparse import httpx from fastapi.requests import Request +from db.config import settings from db.schemas import UserData from utils import crypto from utils.crypto import encrypt_data @@ -217,7 +218,7 @@ async def get_redirector_url(url: str, headers: dict) -> str | None: Get the final URL after following all redirects. """ try: - async with httpx.AsyncClient() as client: + async with httpx.AsyncClient(proxy=settings.requests_proxy_url) as client: response = await client.head(url, headers=headers, follow_redirects=True) return str(response.url) except httpx.HTTPError as e: diff --git a/utils/poster.py b/utils/poster.py index a52d2b0..3ac4979 100644 --- a/utils/poster.py +++ b/utils/poster.py @@ -5,14 +5,14 @@ from io import BytesIO import aiohttp from PIL import Image, ImageDraw, ImageFont, UnidentifiedImageError, ImageStat -from imdb import Cinemagoer +from aiohttp_socks import ProxyConnector +from db.config import settings from db.models import MediaFusionMetaData from scrapers.imdb_data import get_imdb_rating from utils import const from db.redis_database import REDIS_ASYNC_CLIENT -ia = Cinemagoer() font_cache = {} executor = ThreadPoolExecutor(max_workers=4) @@ -24,7 +24,10 @@ async def fetch_poster_image(url: str) -> bytes: logging.info(f"Using cached image for URL: {url}") return cached_image - async with aiohttp.ClientSession() as session: + connector = aiohttp.TCPConnector() + if settings.requests_proxy_url: + connector = ProxyConnector.from_url(settings.requests_proxy_url) + async with aiohttp.ClientSession(connector=connector) as session: async with session.get(url, timeout=10, headers=const.UA_HEADER) as response: response.raise_for_status() if not response.headers["Content-Type"].lower().startswith("image/"): diff --git a/utils/torrent.py b/utils/torrent.py index 125b8ef..633c3b3 100644 --- a/utils/torrent.py +++ b/utils/torrent.py @@ -18,6 +18,7 @@ from demagnetize.core import Demagnetizer from torf import Magnet, MagnetError import utils.runtime_const +from db.config import settings from utils.parser import is_contain_18_plus_keywords from utils.runtime_const import TRACKERS from utils.validation_helper import is_video_file @@ -172,7 +173,7 @@ async def init_best_trackers(): # get the best trackers from https://raw.githubusercontent.com/ngosang/trackerslist/master/trackers_best.txt try: - async with httpx.AsyncClient() as client: + async with httpx.AsyncClient(proxy=settings.requests_proxy_url) as client: response = await client.get( "https://raw.githubusercontent.com/ngosang/trackerslist/master/trackers_best.txt", timeout=30, diff --git a/utils/validation_helper.py b/utils/validation_helper.py index 7e81de2..cff26a4 100644 --- a/utils/validation_helper.py +++ b/utils/validation_helper.py @@ -6,6 +6,7 @@ from urllib.parse import urlparse, urljoin import httpx from db import schemas +from db.config import settings from utils import const from utils.runtime_const import PRIVATE_CIDR from db.redis_database import REDIS_ASYNC_CLIENT @@ -17,7 +18,7 @@ def is_valid_url(url: str) -> bool: async def does_url_exist(url: str) -> bool: - async with httpx.AsyncClient() as client: + async with httpx.AsyncClient(proxy=settings.requests_proxy_url) as client: try: response = await client.head( url, timeout=10, headers=const.UA_HEADER, follow_redirects=True @@ -43,7 +44,7 @@ async def validate_live_stream_url( return False headers = behaviour_hint.get("proxyHeaders", {}).get("request", {}) - async with httpx.AsyncClient() as client: + async with httpx.AsyncClient(proxy=settings.requests_proxy_url) as client: try: response = await client.head( url, timeout=10, headers=headers, follow_redirects=True