Creepso/stream-fusion-old
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fastFix RD limiter for torrents stream.

Browse source
This commit is contained in:
LimeDrive 2024-07-13 13:51:26 +02:00
parent a34302182f
commit 44f91ad681
5 changed files with 106 additions and 27 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

35
poetry.lock generated
View file

@ -1404,6 +1404,21 @@ files = [
[package.extras]
windows-terminal = ["colorama (>=0.4.6)"]
[[package]]
name = "pyrate-limiter"
version = "2.10.0"
description = "Python Rate-Limiter using Leaky-Bucket Algorithm"
optional = false
python-versions = ">=3.7,<4.0"
files = [
{file = "pyrate_limiter-2.10.0-py3-none-any.whl", hash = "sha256:a99e52159f5ed5eb58118bed8c645e30818e7c0e0d127a0585c8277c776b0f7f"},
{file = "pyrate_limiter-2.10.0.tar.gz", hash = "sha256:98cc52cdbe058458e945ae87d4fd5a73186497ffa545ee6e98372f8599a5bd34"},
]
[package.extras]
all = ["filelock (>=3.0)", "redis (>=3.3,<4.0)", "redis-py-cluster (>=2.1.3,<3.0.0)"]
docs = ["furo (>=2022.3.4,<2023.0.0)", "myst-parser (>=0.17)", "sphinx (>=4.3.0,<5.0.0)", "sphinx-autodoc-typehints (>=1.17,<2.0)", "sphinx-copybutton (>=0.5)", "sphinxcontrib-apidoc (>=0.3,<0.4)"]
[[package]]
name = "python-dateutil"
version = "2.9.0.post0"
@ -1780,6 +1795,24 @@ urllib3 = ">=1.21.1,<3"
socks = ["PySocks (>=1.5.6,!=1.5.7)"]
use-chardet-on-py3 = ["chardet (>=3.0.2,<6)"]
[[package]]
name = "requests-ratelimiter"
version = "0.7.0"
description = "Rate-limiting for the requests library"
optional = false
python-versions = "<4.0,>=3.7"
files = [
{file = "requests_ratelimiter-0.7.0-py3-none-any.whl", hash = "sha256:1a7ef2faaa790272722db8539728690046237766fcc479f85b9591e5356a8185"},
{file = "requests_ratelimiter-0.7.0.tar.gz", hash = "sha256:a070c8a359a6f3a001b0ccb08f17228b7ae0a6e21d8df5b6f6bd58389cddde45"},
]
[package.dependencies]
pyrate-limiter = "<3.0"
requests = ">=2.20"
[package.extras]
docs = ["furo (>=2023.3,<2024.0)", "myst-parser (>=1.0)", "sphinx (>=5.2,<6.0)", "sphinx-autodoc-typehints (>=1.22,<2.0)", "sphinx-copybutton (>=0.5)"]
[[package]]
name = "rich"
version = "13.7.1"
@ -2485,4 +2518,4 @@ multidict = ">=4.0"
[metadata]
lock-version = "2.0"
python-versions = "^3.11"
content-hash = "d47e260deaed0289ddb0fd8a0d1259877fe78c61775f3424be110bb955cfa0d6"
content-hash = "0b5b911bb0176eadd997449036af61ee87b9bd39798792adbc60afd512d97c21"

1
pyproject.toml
View file

@ -30,6 +30,7 @@ jsonpickle = "^3.2.2"
fastapi-simple-rate-limiter = "^0.0.4"
itsdangerous = "^2.2.0"
email-validator = "^2.2.0"
requests-ratelimiter = "^0.7.0"
[build-system]

79
stream_fusion/utils/debrid/base_debrid.py
View file

@ -1,3 +1,4 @@
from collections import deque
import time
import requests
@ -10,29 +11,71 @@ class BaseDebrid:
self.config = config
self.logger = logger
self.__session = requests.Session()
# Limiteurs de débit
self.global_limit = 250
self.global_period = 60
self.torrent_limit = 1
self.torrent_period = 1
self.global_requests = deque()
self.torrent_requests = deque()
def _rate_limit(self, requests_queue, limit, period):
current_time = time.time()
while requests_queue and requests_queue[0] <= current_time - period:
requests_queue.popleft()
if len(requests_queue) >= limit:
sleep_time = requests_queue[0] - (current_time - period)
if sleep_time > 0:
time.sleep(sleep_time)
requests_queue.append(time.time())
def _global_rate_limit(self):
self._rate_limit(self.global_requests, self.global_limit, self.global_period)
def _torrent_rate_limit(self):
self._rate_limit(self.torrent_requests, self.torrent_limit, self.torrent_period)
def get_json_response(self, url, method='get', data=None, headers=None, files=None):
if method == 'get':
response = self.__session.get(url, headers=headers)
elif method == 'post':
response = self.__session.post(url, data=data, headers=headers, files=files)
elif method == 'put':
response = self.__session.put(url, data=data, headers=headers)
elif method == 'delete':
response = self.__session.delete(url, headers=headers)
else:
raise ValueError(f"Unsupported HTTP method: {method}")
# Check if the request was successful
if response.ok:
self._global_rate_limit()
if 'torrents' in url:
self._torrent_rate_limit()
max_attempts = 5
for attempt in range(max_attempts):
try:
if method == 'get':
response = self.__session.get(url, headers=headers)
elif method == 'post':
response = self.__session.post(url, data=data, headers=headers, files=files)
elif method == 'put':
response = self.__session.put(url, data=data, headers=headers)
elif method == 'delete':
response = self.__session.delete(url, headers=headers)
else:
raise ValueError(f"Unsupported HTTP method: {method}")
response.raise_for_status()
return response.json()
except ValueError:
self.logger.error(f"Failed to parse response as JSON: {response.text}")
except requests.exceptions.HTTPError as e:
if e.response.status_code == 429:
wait_time = 2 ** attempt + 1
self.logger.warning(f"Rate limit exceeded. Attempt {attempt + 1}/{max_attempts}. Waiting for {wait_time} seconds.")
time.sleep(wait_time)
else:
self.logger.error(f"HTTP error occurred: {e}")
return None
except Exception as e:
self.logger.error(f"An error occurred: {e}")
return None
else:
self.logger.error(f"Request failed with status code {response.status_code}")
return None
self.logger.error("Max attempts reached. Unable to complete request.")
return None
def wait_for_ready_status(self, check_status_func, timeout=30, interval=5):
self.logger.info(f"Waiting for {timeout} seconds to cache.")

7
stream_fusion/utils/debrid/realdebrid.py
View file

@ -37,14 +37,13 @@ class RealDebrid(BaseDebrid):
logger.info(f"Getting torrent info for: {torrent_id}")
url = f"{self.base_url}/rest/1.0/torrents/info/{torrent_id}"
torrent_info = self.get_json_response(url, headers=self.headers)
if not torrent_info or 'files' not in torrent_info:
return None
return torrent_info
def select_files(self, torrent_id, file_id):
logger.info(f"Selecting file(s): {file_id}")
self._torrent_rate_limit()
url = f"{self.base_url}/rest/1.0/torrents/selectFiles/{torrent_id}"
data = {"files": str(file_id)}
requests.post(url, headers=self.headers, data=data)
@ -63,7 +62,7 @@ class RealDebrid(BaseDebrid):
return torrent['id']
return False
def wait_for_link(self, torrent_id, timeout=30, interval=5):
def wait_for_link(self, torrent_id, timeout=60, interval=5):
start_time = time.time()
while time.time() - start_time < timeout:
torrent_info = self.get_torrent_info(torrent_id)
@ -74,6 +73,7 @@ class RealDebrid(BaseDebrid):
return None
def get_availability_bulk(self, hashes_or_magnets, ip=None):
self._torrent_rate_limit()
if len(hashes_or_magnets) == 0:
logger.info("No hashes to be sent to Real-Debrid.")
return dict()
@ -145,6 +145,7 @@ class RealDebrid(BaseDebrid):
return unrestrict_response['download']
def __get_cached_torrent_ids(self, info_hash):
self._torrent_rate_limit()
url = f"{self.base_url}/rest/1.0/torrents"
torrents = self.get_json_response(url, headers=self.headers)

11
stream_fusion/web/playback/stream/views.py
View file

@ -16,6 +16,7 @@ from stream_fusion.utils.debrid.get_debrid_service import get_debrid_service
from stream_fusion.utils.parse_config import parse_config
from stream_fusion.utils.string_encoding import decodeb64
from stream_fusion.utils.security import check_api_key
from stream_fusion.constants import NO_CACHE_VIDEO_URL
from stream_fusion.web.playback.stream.schemas import (
ErrorResponse,
HeadResponse,
@ -96,9 +97,9 @@ def get_stream_link(
debrid_service = get_debrid_service(config)
link = debrid_service.get_stream_link(decoded_query, config, ip)
redis_cache.set(cache_key, link, expiration=3600) # Cache for 1 hour
logger.info(f"Stream link generated and cached: {link}")
if link != NO_CACHE_VIDEO_URL:
redis_cache.set(cache_key, link, expiration=3600) # Cache for 1 hour
logger.info(f"Stream link generated and cached: {link}")
return link
@ -129,13 +130,13 @@ async def get_playback(
ip = request.client.host
lock_key = f"lock:stream:{decoded_query}_{ip}"
lock = redis_client.lock(lock_key, timeout=10) # Timeout de 10 secondes
lock = redis_client.lock(lock_key, timeout=60)
try:
if lock.acquire(blocking=False):
link = get_stream_link(decoded_query, config, ip, redis_cache)
else:
await asyncio.sleep(2)
await asyncio.sleep(10)
cache_key = f"stream_link:{decoded_query}_{ip}"
cached_link = redis_cache.get(cache_key)
if cached_link: