Viren070/MediaFusion
1
0
Fork 0
mirror of https://github.com/Viren070/MediaFusion.git synced 2025-12-01 23:21:11 +01:00
Code Issues Packages Projects Releases Wiki Activity

Migrate streaming providers functions to async & improve performance (#331)

Browse Source 
* Migrate streaming providers functions to async & improve performance

* Fix streaming provider file selection & validate file extension as video type

* Improve RD torrent handling

* Find file based on filename or the largest file or episode based and remove index base file finding due to not trustable

* Add support for updating torrent matadata with offcloud

* Fix RD background task to update torrent metadata

* Fix OC file size callback function

* Fix AD, DL & PM

* Fix PikPak httpx client leak & use medialink for speed & set 5 min for token cache

* Migrate to AIOSeedrcc & fix season pack file selecting & refactorings

* Qbittorrent: Add support for season pack file finding & disable dht, pex, lsd on private indexer torrent

* Migrate to aiohttp for stability on resource fetching & bugfixes

* Migrated to aioseedrcc & update libs

* pass indexer type & remove Upper info hash finding
This commit is contained in:
Mohamed Zumair
2024-11-01 15:58:39 +05:30
committed by GitHub
parent 95f57a9da8
commit 48b3f5e85f
24 changed files with 2325 additions and 1815 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Pipfile
+1 -1
View File
@@ -20,7 +20,6 @@ python-dateutil = "*"
bencodepy = "*"
pydantic-settings = "*"
pycryptodome = "*"
seedrcc = "*"
pillow = "*"
httpx = "*"
uvloop = { markers = "sys_platform != 'win32'" }
@@ -49,6 +48,7 @@ ratelimit = "*"
qrcode = "*"
scrapeops-scrapy = "*"
scrapeops-python-requests = "*"
aioseedrcc = "*"
[dev-packages]
pysocks = "*"
Pipfile.lock
Generated
+714 -695
View File
File diff suppressed because it is too large Load Diff
streaming_providers/alldebrid/client.py
+38 -31
View File
@@ -1,4 +1,4 @@
from typing import Any
from typing import Any, Optional
from streaming_providers.debrid_client import DebridClient
from streaming_providers.exceptions import ProviderException
@@ -8,35 +8,37 @@ class AllDebrid(DebridClient):
BASE_URL = "https://api.alldebrid.com/v4"
AGENT = "mediafusion"
def __init__(self, token: str, user_ip: str | None = None):
def __init__(self, token: str, user_ip: Optional[str] = None):
self.user_ip = user_ip
super().__init__(token)
def initialize_headers(self):
async def initialize_headers(self):
self.headers = {"Authorization": f"Bearer {self.token}"}
def __del__(self):
async def disable_access_token(self):
pass
def _handle_service_specific_errors(self, error):
async def _handle_service_specific_errors(self, error_data: dict, status_code: int):
pass
def _make_request(
async def _make_request(
self,
method: str,
url: str,
data=None,
params=None,
is_return_none=False,
is_expected_to_fail=False,
data: Optional[dict] = None,
json: Optional[dict] = None,
params: Optional[dict] = None,
is_return_none: bool = False,
is_expected_to_fail: bool = False,
retry_count: int = 0,
) -> dict:
params = params or {}
params["agent"] = self.AGENT
if self.user_ip:
params["ip"] = self.user_ip
url = self.BASE_URL + url
return super()._make_request(
method, url, data, params, is_return_none, is_expected_to_fail
return await super()._make_request(
method, url, data, json, params, is_return_none, is_expected_to_fail
)
@staticmethod
@@ -46,7 +48,7 @@ class AllDebrid(DebridClient):
match error_code:
case "AUTH_BAD_APIKEY":
raise ProviderException(
f"Invalid AllDebrid API key", "invalid_token.mp4"
"Invalid AllDebrid API key", "invalid_token.mp4"
)
case "NO_SERVER":
raise ProviderException(
@@ -55,15 +57,15 @@ class AllDebrid(DebridClient):
)
case "AUTH_BLOCKED":
raise ProviderException(
f"API got blocked on AllDebrid", "alldebrid_api_blocked.mp4"
"API got blocked on AllDebrid", "alldebrid_api_blocked.mp4"
)
case "MAGNET_MUST_BE_PREMIUM":
raise ProviderException(
f"Torrent must be premium on AllDebrid", "need_premium.mp4"
"Torrent must be premium on AllDebrid", "need_premium.mp4"
)
case "MAGNET_TOO_MANY_ACTIVE":
raise ProviderException(
f"Too many active torrents on AllDebrid", "torrent_limit.mp4"
"Too many active torrents on AllDebrid", "torrent_limit.mp4"
)
case _:
raise ProviderException(
@@ -71,37 +73,40 @@ class AllDebrid(DebridClient):
"transfer_error.mp4",
)
def add_magnet_link(self, magnet_link):
response_data = self._make_request(
"POST", f"/magnet/upload", data={"magnets[]": magnet_link}
async def add_magnet_link(self, magnet_link):
response_data = await self._make_request(
"POST", "/magnet/upload", data={"magnets[]": magnet_link}
)
self._validate_error_response(response_data)
return response_data
def get_user_torrent_list(self):
return self._make_request("GET", "/magnet/status")
async def get_user_torrent_list(self):
return await self._make_request("GET", "/magnet/status")
def get_torrent_info(self, magnet_id):
response = self._make_request("GET", "/magnet/status", params={"id": magnet_id})
async def get_torrent_info(self, magnet_id):
response = await self._make_request(
"GET", "/magnet/status", params={"id": magnet_id}
)
return response.get("data", {}).get("magnets")
def get_torrent_instant_availability(self, magnet_links: list[str]):
response = self._make_request(
async def get_torrent_instant_availability(self, magnet_links: list[str]):
response = await self._make_request(
"POST", "/magnet/instant", data={"magnets[]": magnet_links}
)
return response.get("data", {}).get("magnets", [])
def get_available_torrent(self, info_hash) -> dict[str, Any] | None:
available_torrents = self.get_user_torrent_list()
async def get_available_torrent(self, info_hash) -> Optional[dict[str, Any]]:
available_torrents = await self.get_user_torrent_list()
self._validate_error_response(available_torrents)
if not available_torrents.get("data"):
return None
for torrent in available_torrents["data"]["magnets"]:
if torrent["hash"] == info_hash:
return torrent
return None
def create_download_link(self, link):
response = self._make_request(
async def create_download_link(self, link):
response = await self._make_request(
"GET",
"/link/unlock",
params={"link": link},
@@ -114,5 +119,7 @@ class AllDebrid(DebridClient):
"transfer_error.mp4",
)
def delete_torrent(self, magnet_id):
return self._make_request("GET", "/magnet/delete", params={"id": magnet_id})
async def delete_torrent(self, magnet_id):
return await self._make_request(
"GET", "/magnet/delete", params={"id": magnet_id}
)
streaming_providers/alldebrid/utils.py
+103 -54
View File
@@ -1,116 +1,165 @@
import asyncio
from typing import Optional
from fastapi import BackgroundTasks
from db.models import TorrentStreams
from db.schemas import UserData
from streaming_providers.alldebrid.client import AllDebrid
from streaming_providers.exceptions import ProviderException
from streaming_providers.parser import select_file_index_from_torrent
from streaming_providers.parser import (
select_file_index_from_torrent,
update_torrent_streams_metadata,
)
def get_torrent_info(ad_client, info_hash):
torrent_info = ad_client.get_available_torrent(info_hash)
async def get_torrent_info(ad_client, info_hash):
torrent_info = await ad_client.get_available_torrent(info_hash)
if torrent_info and torrent_info["status"] == "Ready":
return torrent_info
elif torrent_info and torrent_info["statusCode"] == 7:
ad_client.delete_torrent(torrent_info.get("id"))
await ad_client.delete_torrent(torrent_info.get("id"))
raise ProviderException(
"Not enough seeders available for parse magnet link",
"Not enough seeders available to parse magnet link",
"transfer_error.mp4",
)
return None
def add_new_torrent(ad_client, magnet_link):
response_data = ad_client.add_magnet_link(magnet_link)
async def add_new_torrent(ad_client, magnet_link):
response_data = await ad_client.add_magnet_link(magnet_link)
return response_data["data"]["magnets"][0]["id"]
def wait_for_download_and_get_link(
ad_client, torrent_id, filename, file_index, episode, max_retries, retry_interval
async def wait_for_download_and_get_link(
ad_client,
torrent_id,
stream,
filename,
season,
episode,
max_retries,
retry_interval,
background_tasks,
):
torrent_info = ad_client.wait_for_status(
torrent_info = await ad_client.wait_for_status(
torrent_id, "Ready", max_retries, retry_interval
)
file_index = select_file_index_from_torrent(
file_index = await select_file_index_from_torrent(
torrent_info,
filename,
file_index,
episode,
file_key="links",
name_key="filename",
)
response = ad_client.create_download_link(torrent_info["links"][file_index]["link"])
if filename is None:
background_tasks.add_task(
update_torrent_streams_metadata,
torrent_stream=stream,
torrent_info=torrent_info,
file_index=file_index,
season=season,
file_key="links",
name_key="filename",
is_index_trustable=False,
)
response = await ad_client.create_download_link(
torrent_info["links"][file_index]["link"]
)
return response["data"]["link"]
def get_video_url_from_alldebrid(
async def get_video_url_from_alldebrid(
info_hash: str,
magnet_link: str,
user_data: UserData,
filename: str,
stream: TorrentStreams,
filename: Optional[str],
user_ip: str,
file_index: int,
episode: int = None,
background_tasks: BackgroundTasks,
season: Optional[int] = None,
episode: Optional[int] = None,
max_retries=5,
retry_interval=5,
**kwargs,
) -> str:
ad_client = AllDebrid(token=user_data.streaming_provider.token, user_ip=user_ip)
async with AllDebrid(
token=user_data.streaming_provider.token, user_ip=user_ip
) as ad_client:
torrent_info = await get_torrent_info(ad_client, info_hash)
if not torrent_info:
torrent_id = await add_new_torrent(ad_client, magnet_link)
else:
torrent_id = torrent_info.get("id")
torrent_info = get_torrent_info(ad_client, info_hash)
if not torrent_info:
torrent_id = add_new_torrent(ad_client, magnet_link)
else:
torrent_id = torrent_info.get("id")
return wait_for_download_and_get_link(
ad_client,
torrent_id,
filename,
file_index,
episode,
max_retries,
retry_interval,
)
return await wait_for_download_and_get_link(
ad_client,
torrent_id,
stream,
filename,
season,
episode,
max_retries,
retry_interval,
background_tasks,
)
def update_ad_cache_status(
async def update_ad_cache_status(
streams: list[TorrentStreams], user_data: UserData, user_ip: str, **kwargs
):
"""Updates the cache status of streams based on AllDebrid's instant availability."""
try:
ad_client = AllDebrid(token=user_data.streaming_provider.token, user_ip=user_ip)
instant_availability_data = ad_client.get_torrent_instant_availability(
[stream.id for stream in streams]
)
for stream in streams:
stream.cached = any(
torrent["instant"]
for torrent in instant_availability_data
if torrent["hash"] == stream.id
async with AllDebrid(
token=user_data.streaming_provider.token, user_ip=user_ip
) as ad_client:
instant_availability_data = (
await ad_client.get_torrent_instant_availability(
[stream.id for stream in streams]
)
)
for stream in streams:
stream.cached = any(
torrent.get("instant", False)
for torrent in instant_availability_data
if torrent.get("hash") == stream.id
)
except ProviderException:
pass
def fetch_downloaded_info_hashes_from_ad(
async def fetch_downloaded_info_hashes_from_ad(
user_data: UserData, user_ip: str, **kwargs
) -> list[str]:
"""Fetches the info_hashes of all torrents downloaded in the AllDebrid account."""
try:
ad_client = AllDebrid(token=user_data.streaming_provider.token, user_ip=user_ip)
available_torrents = ad_client.get_user_torrent_list()
if not available_torrents.get("data"):
return []
return [torrent["hash"] for torrent in available_torrents["data"]["magnets"]]
async with AllDebrid(
token=user_data.streaming_provider.token, user_ip=user_ip
) as ad_client:
available_torrents = await ad_client.get_user_torrent_list()
if not available_torrents.get("data"):
return []
return [
torrent["hash"] for torrent in available_torrents["data"]["magnets"]
]
except ProviderException:
return []
def delete_all_torrents_from_ad(user_data: UserData, user_ip: str, **kwargs):
async def delete_all_torrents_from_ad(user_data: UserData, user_ip: str, **kwargs):
"""Deletes all torrents from the AllDebrid account."""
ad_client = AllDebrid(token=user_data.streaming_provider.token, user_ip=user_ip)
torrents = ad_client.get_user_torrent_list()
for torrent in torrents["data"]["magnets"]:
ad_client.delete_torrent(torrent["id"])
async with AllDebrid(
token=user_data.streaming_provider.token, user_ip=user_ip
) as ad_client:
torrents = await ad_client.get_user_torrent_list()
await asyncio.gather(
*[
ad_client.delete_torrent(torrent["id"])
for torrent in torrents["data"]["magnets"]
]
)
streaming_providers/debrid_client.py
+132 -59
View File
@@ -1,121 +1,194 @@
import time
import asyncio
import traceback
from abc import abstractmethod
from base64 import b64encode, b64decode
from typing import Optional, Dict, Union
import requests
from requests import RequestException, JSONDecodeError
import aiohttp
from aiohttp import ClientResponse, ClientTimeout, ContentTypeError
from streaming_providers.exceptions import ProviderException
from utils import const
class DebridClient:
def __init__(self, token=None):
def __init__(self, token: Optional[str] = None):
self.token = token
self.headers = {}
self.initialize_headers()
self.is_private_token = False
self.headers: Dict[str, str] = {}
self._session: Optional[aiohttp.ClientSession] = None
self._timeout = ClientTimeout(total=15) # Stremio timeout is 20s
def __del__(self):
if self.token:
@property
def session(self) -> aiohttp.ClientSession:
if self._session is None:
self._session = aiohttp.ClientSession(
timeout=self._timeout,
connector=aiohttp.TCPConnector(ttl_dns_cache=300),
)
return self._session
async def __aenter__(self):
await self.initialize_headers()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.token and not self.is_private_token:
try:
self.disable_access_token()
await self.disable_access_token()
except ProviderException:
pass
def _make_request(
if self._session:
await self._session.close()
self._session = None
async def _make_request(
self,
method: str,
url: str,
data=None,
params=None,
is_return_none=False,
is_expected_to_fail=False,
) -> dict:
response = self._perform_request(method, url, data, params)
self._handle_errors(response, is_expected_to_fail)
return self._parse_response(response, is_return_none)
def _perform_request(self, method, url, data, params):
data: Optional[dict | str] = None,
json: Optional[dict] = None,
params: Optional[dict] = None,
is_return_none: bool = False,
is_expected_to_fail: bool = False,
retry_count: int = 0,
) -> dict | list | str:
try:
return requests.request(
method,
url,
params=params,
data=data,
headers=self.headers,
timeout=const.DEBRID_SERVER_TIMEOUT,
)
except requests.exceptions.Timeout:
raise ProviderException("Request timed out.", "torrent_not_downloaded.mp4")
except requests.exceptions.ConnectionError:
raise ProviderException(
"Failed to connect to Debrid service.", "debrid_service_down_error.mp4"
)
async with self.session.request(
method, url, data=data, json=json, params=params, headers=self.headers
) as response:
await self._check_response_status(response, is_expected_to_fail)
return await self._parse_response(
response, is_return_none, is_expected_to_fail
)
def _handle_errors(self, response, is_expected_to_fail):
except aiohttp.ClientConnectorError as error:
if retry_count < 1: # Try one more time
return await self._make_request(
method,
url,
data=data,
json=json,
params=params,
is_return_none=is_return_none,
is_expected_to_fail=is_expected_to_fail,
retry_count=retry_count + 1,
)
await self._handle_request_error(error)
except aiohttp.ClientError as error:
await self._handle_request_error(error)
except Exception as error:
await self._handle_request_error(error)
async def _check_response_status(
self, response: ClientResponse, is_expected_to_fail: bool
):
"""Check response status and handle HTTP errors."""
try:
response.raise_for_status()
except RequestException as error:
if error.response.status_code in [502, 503, 504]:
except aiohttp.ClientResponseError as error:
if error.status in [502, 503, 504]:
raise ProviderException(
"Debrid service is down.", "debrid_service_down_error.mp4"
) from error
)
if is_expected_to_fail:
return
self._handle_service_specific_errors(error)
if error.response.status_code == 401:
if response.headers.get("Content-Type") == "application/json":
error_content = await response.json()
await self._handle_service_specific_errors(error_content, error.status)
else:
error_content = await response.text()
if error.status == 401:
raise ProviderException("Invalid token", "invalid_token.mp4")
formatted_traceback = "".join(traceback.format_exception(error))
raise ProviderException(
f"API Error {error.response.text} \n{formatted_traceback}",
f"API Error {error_content} \n{formatted_traceback}",
"api_error.mp4",
)
def _handle_service_specific_errors(self, error):
@staticmethod
async def _handle_request_error(error: Exception):
if isinstance(error, asyncio.TimeoutError):
raise ProviderException("Request timed out.", "torrent_not_downloaded.mp4")
elif isinstance(error, aiohttp.ClientConnectorError):
raise ProviderException(
"Failed to connect to Debrid service.", "debrid_service_down_error.mp4"
)
raise ProviderException(f"Request error: {str(error)}", "api_error.mp4")
@abstractmethod
async def _handle_service_specific_errors(self, error_data: dict, status_code: int):
"""
Service specific errors on api requests.
"""
raise NotImplementedError
@staticmethod
def _parse_response(response, is_return_none):
async def _parse_response(
response: ClientResponse, is_return_none: bool, is_expected_to_fail: bool
) -> Union[dict, list, str]:
if is_return_none:
return {}
try:
return response.json()
except JSONDecodeError as error:
return await response.json()
except (ValueError, ContentTypeError) as error:
response_text = await response.text()
if is_expected_to_fail:
return response_text
raise ProviderException(
f"Failed to parse response error: {error}. \nresponse: {response.text}",
f"Failed to parse response error: {error}. \nresponse: {response_text}",
"api_error.mp4",
)
def initialize_headers(self):
@abstractmethod
async def initialize_headers(self):
raise NotImplementedError
def disable_access_token(self):
@abstractmethod
async def disable_access_token(self):
raise NotImplementedError
def wait_for_status(
async def wait_for_status(
self,
torrent_id: str,
target_status: str | int,
target_status: Union[str, int],
max_retries: int,
retry_interval: int,
):
torrent_info: Optional[dict] = None,
) -> dict:
"""Wait for the torrent to reach a particular status."""
retries = 0
while retries < max_retries:
torrent_info = self.get_torrent_info(torrent_id)
# if torrent_info is available, check the status from it
if torrent_info:
if torrent_info["status"] == target_status:
return torrent_info
time.sleep(retry_interval)
retries += 1
for _ in range(max_retries):
torrent_info = await self.get_torrent_info(torrent_id)
if torrent_info["status"] == target_status:
return torrent_info
await asyncio.sleep(retry_interval)
raise ProviderException(
f"Torrent did not reach {target_status} status.",
"torrent_not_downloaded.mp4",
)
def get_torrent_info(self, torrent_id):
@abstractmethod
async def get_torrent_info(self, torrent_id: str) -> dict:
raise NotImplementedError
@staticmethod
def encode_token_data(code: str, *args, **kwargs) -> str:
token = f"code:{code}"
return b64encode(token.encode()).decode()
@staticmethod
def decode_token_str(token: str) -> Optional[str]:
try:
_, code = b64decode(token).decode().split(":")
except (ValueError, UnicodeDecodeError):
# Assume as private token
return None
return code
streaming_providers/debridlink/api.py
+9 -7
View File
@@ -10,14 +10,16 @@ router = APIRouter()
@router.get("/get-device-code")
async def get_device_code():
dl_client = DebridLink()
return JSONResponse(
content=dl_client.get_device_code(), headers=const.NO_CACHE_HEADERS
)
async with DebridLink() as dl_client:
return JSONResponse(
content=await dl_client.get_device_code(), headers=const.NO_CACHE_HEADERS
)
@router.post("/authorize")
async def authorize(data: AuthorizeData):
dl_client = DebridLink()
response = dl_client.authorize(data.device_code)
return JSONResponse(content=response, headers=const.NO_CACHE_HEADERS)
async with DebridLink() as dl_client:
return JSONResponse(
content=await dl_client.authorize(data.device_code),
headers=const.NO_CACHE_HEADERS,
)
streaming_providers/debridlink/client.py
+67 -57
View File
@@ -1,5 +1,4 @@
from base64 import b64encode, b64decode
from typing import Any
from typing import Any, Optional
from streaming_providers.debrid_client import DebridClient
from streaming_providers.exceptions import ProviderException
@@ -10,40 +9,49 @@ class DebridLink(DebridClient):
OAUTH_URL = "https://debrid-link.com/api/oauth"
OPENSOURCE_CLIENT_ID = "RyrV22FOg30DsxjYPziRKA"
def _handle_service_specific_errors(self, error):
if (
error.response.status_code == 400
and error.response.json().get("error") == "freeServerOverload"
):
raise ProviderException(
"Debrid-Link free servers are overloaded", "need_premium.mp4"
)
@staticmethod
def _handle_error_message(error_message):
match error_message:
case "freeServerOverload":
raise ProviderException(
"Debrid-Link free servers are overloaded", "need_premium.mp4"
)
case "badToken" | "expired_token":
raise ProviderException("Invalid token", "invalid_token.mp4")
case "server_error" | "notDebrid":
raise ProviderException(
"Debrid-Link server error", "debrid_service_down_error.mp4"
)
case "maxLink" | "maxLinkHost" | "maxData" | "maxDataHost" | "maxTorrent":
raise ProviderException(
"Debrid-Link daily limit reached", "daily_download_limit.mp4"
)
case "disabledServerHost":
raise ProviderException(
"Debrid-Link Server / VPN are not allowed on this host",
"ip_not_allowed.mp4",
)
def initialize_headers(self):
async def _handle_service_specific_errors(self, error_data: dict, status_code: int):
self._handle_error_message(error_data.get("error"))
async def initialize_headers(self):
if self.token:
token_data = self.decode_token_str(self.token)
access_token_data = self.refresh_token(
token_data["client_id"], token_data["code"]
)
token_code = self.decode_token_str(self.token)
if token_code:
access_token_data = await self.refresh_token(
self.OPENSOURCE_CLIENT_ID, token_code
)
auth_token = access_token_data.get("access_token")
else:
auth_token = self.token
self.is_private_token = True
self.headers = {
"Authorization": f"Bearer {access_token_data['access_token']}"
"Authorization": f"Bearer {auth_token}",
}
@staticmethod
def encode_token_data(client_id: str, code: str):
token = f"{client_id}:{code}"
return b64encode(str(token).encode()).decode()
@staticmethod
def decode_token_str(token: str) -> dict[str, str]:
try:
client_id, code = b64decode(token).decode().split(":")
except ValueError:
raise ProviderException("Invalid token", "invalid_token.mp4")
return {"client_id": client_id, "code": code}
def get_device_code(self):
return self._make_request(
async def get_device_code(self) -> dict[str, Any]:
return await self._make_request(
"POST",
f"{self.OAUTH_URL}/device/code",
data={
@@ -52,8 +60,8 @@ class DebridLink(DebridClient):
},
)
def get_token(self, client_id, device_code):
return self._make_request(
async def get_token(self, client_id, device_code):
return await self._make_request(
"POST",
f"{self.OAUTH_URL}/token",
data={
@@ -64,8 +72,8 @@ class DebridLink(DebridClient):
is_expected_to_fail=True,
)
def refresh_token(self, client_id, refresh_token):
return self._make_request(
async def refresh_token(self, client_id, refresh_token):
return await self._make_request(
"POST",
f"{self.OAUTH_URL}/token",
data={
@@ -75,39 +83,38 @@ class DebridLink(DebridClient):
},
)
def authorize(self, device_code):
token_data = self.get_token(self.OPENSOURCE_CLIENT_ID, device_code)
async def authorize(self, device_code):
token_data = await self.get_token(self.OPENSOURCE_CLIENT_ID, device_code)
if "error" in token_data:
return token_data
if "access_token" in token_data:
token = self.encode_token_data(
self.OPENSOURCE_CLIENT_ID, token_data["refresh_token"]
)
token = self.encode_token_data(token_data["refresh_token"])
return {"token": token}
else:
return token_data
def add_magnet_link(self, magnet_link):
response = self._make_request(
async def add_magnet_link(self, magnet_link):
response = await self._make_request(
"POST",
f"{self.BASE_URL}/seedbox/add",
data={"url": magnet_link},
is_expected_to_fail=True,
)
if response.get("success") is False or response.get("error"):
if response.get("error"):
self._handle_error_message(response.get("error"))
raise ProviderException(
f"Failed to add magnet link to Debrid-Link: {response.get('error')}",
"transfer_error.mp4",
)
return response.get("value", {})
def get_user_torrent_list(self):
return self._make_request("GET", f"{self.BASE_URL}/seedbox/list")
async def get_user_torrent_list(self) -> dict[str, Any]:
return await self._make_request("GET", f"{self.BASE_URL}/seedbox/list")
def get_torrent_info(self, torrent_id):
response = self._make_request(
async def get_torrent_info(self, torrent_id) -> dict[str, Any]:
response = await self._make_request(
"GET", f"{self.BASE_URL}/seedbox/list", params={"ids": torrent_id}
)
if response.get("value"):
@@ -116,29 +123,31 @@ class DebridLink(DebridClient):
"Failed to get torrent info from Debrid-Link", "transfer_error.mp4"
)
def get_torrent_files_list(self, torrent_id):
return self._make_request("GET", f"{self.BASE_URL}/files/{torrent_id}/list")
async def get_torrent_files_list(self, torrent_id) -> dict[str, Any]:
return await self._make_request(
"GET", f"{self.BASE_URL}/files/{torrent_id}/list"
)
def get_torrent_instant_availability(self, torrent_hash):
return self._make_request(
async def get_torrent_instant_availability(self, torrent_hash) -> dict[str, Any]:
return await self._make_request(
"GET", f"{self.BASE_URL}/seedbox/cached", params={"url": torrent_hash}
)
def delete_torrent(self, torrent_id):
return self._make_request(
async def delete_torrent(self, torrent_id) -> dict[str, Any]:
return await self._make_request(
"DELETE", f"{self.BASE_URL}/seedbox/{torrent_id}/delete"
)
def disable_access_token(self):
return self._make_request(
async def disable_access_token(self) -> Optional[dict[str, Any]]:
return await self._make_request(
"GET",
f"{self.OAUTH_URL}/revoke",
is_return_none=True,
is_expected_to_fail=True,
)
def get_available_torrent(self, info_hash: str) -> dict[str, Any] | None:
torrent_list_response = self.get_user_torrent_list()
async def get_available_torrent(self, info_hash: str) -> Optional[dict[str, Any]]:
torrent_list_response = await self.get_user_torrent_list()
if "error" in torrent_list_response:
raise ProviderException(
"Failed to get torrent info from Debrid-Link", "transfer_error.mp4"
@@ -148,3 +157,4 @@ class DebridLink(DebridClient):
for torrent in available_torrents:
if torrent["hashString"] == info_hash:
return torrent
return None
streaming_providers/debridlink/utils.py
+91 -49
View File
@@ -1,95 +1,137 @@
import asyncio
from typing import Optional
from fastapi import BackgroundTasks
from db.models import TorrentStreams
from db.schemas import UserData
from streaming_providers.debridlink.client import DebridLink
from streaming_providers.exceptions import ProviderException
from streaming_providers.parser import select_file_index_from_torrent
from streaming_providers.parser import (
select_file_index_from_torrent,
update_torrent_streams_metadata,
)
def get_download_link(
torrent_info: dict, filename: str, file_index: int, episode: int | None
async def get_download_link(
torrent_info: dict,
stream: TorrentStreams,
background_tasks,
filename: Optional[str],
file_index: Optional[int],
episode: Optional[int],
season: Optional[int],
) -> str:
file_index = select_file_index_from_torrent(
torrent_info, filename, file_index, episode
selected_file_index = await select_file_index_from_torrent(
torrent_info, filename, episode
)
if torrent_info["files"][file_index]["downloadPercent"] != 100:
if filename is None or file_index is None:
background_tasks.add_task(
update_torrent_streams_metadata,
torrent_stream=stream,
torrent_info=torrent_info,
file_index=selected_file_index,
season=season,
is_index_trustable=True,
)
if torrent_info["files"][selected_file_index]["downloadPercent"] != 100:
raise ProviderException(
"Torrent not downloaded yet.", "torrent_not_downloaded.mp4"
)
return torrent_info["files"][file_index]["downloadUrl"]
return torrent_info["files"][selected_file_index]["downloadUrl"]
def get_video_url_from_debridlink(
async def get_video_url_from_debridlink(
info_hash: str,
magnet_link: str,
user_data: UserData,
filename: str,
file_index: int,
episode: int | None,
filename: Optional[str],
file_index: Optional[int],
stream: TorrentStreams,
background_tasks: BackgroundTasks,
episode: Optional[int],
season: Optional[int] = None,
max_retries=5,
retry_interval=5,
**kwargs,
) -> str:
dl_client = DebridLink(token=user_data.streaming_provider.token)
async with DebridLink(token=user_data.streaming_provider.token) as dl_client:
torrent_info = await dl_client.get_available_torrent(info_hash)
if not torrent_info:
torrent_info = await dl_client.add_magnet_link(magnet_link)
torrent_info = dl_client.get_available_torrent(info_hash)
if not torrent_info:
torrent_id = dl_client.add_magnet_link(magnet_link).get("id")
torrent_info = dl_client.get_torrent_info(torrent_id)
else:
torrent_id = torrent_info.get("id")
if not torrent_id:
raise ProviderException(
"Failed to add magnet link to DebridLink", "transfer_error.mp4"
)
if not torrent_id:
raise ProviderException(
"Failed to add magnet link to DebridLink", "transfer_error.mp4"
if torrent_info.get("error"):
await dl_client.delete_torrent(torrent_id)
raise ProviderException(
f"Torrent cannot be downloaded due to error: {torrent_info.get('errorString')}",
"transfer_error.mp4",
)
torrent_info = await dl_client.wait_for_status(
torrent_id, 100, max_retries, retry_interval, torrent_info
)
if torrent_info.get("error"):
dl_client.delete_torrent(torrent_id)
raise ProviderException(
f"Torrent cannot be downloaded due to error: {torrent_info.get('errorString')}",
"transfer_error.mp4",
return await get_download_link(
torrent_info,
stream,
background_tasks,
filename,
file_index,
episode,
season,
)
torrent_info = dl_client.wait_for_status(
torrent_id, 100, max_retries, retry_interval
)
return get_download_link(torrent_info, filename, file_index, episode)
def update_dl_cache_status(
async def update_dl_cache_status(
streams: list[TorrentStreams], user_data: UserData, **kwargs
):
"""Updates the cache status of streams based on DebridLink's instant availability."""
try:
dl_client = DebridLink(token=user_data.streaming_provider.token)
instant_availability_response = dl_client.get_torrent_instant_availability(
",".join([stream.id for stream in streams])
)
for stream in streams:
stream.cached = bool(stream.id in instant_availability_response["value"])
async with DebridLink(token=user_data.streaming_provider.token) as dl_client:
instant_availability_response = (
await dl_client.get_torrent_instant_availability(
",".join([stream.id for stream in streams])
)
)
if "error" in instant_availability_response:
return
for stream in streams:
stream.cached = bool(
stream.id in instant_availability_response["value"]
)
except ProviderException:
pass
def fetch_downloaded_info_hashes_from_dl(user_data: UserData, **kwargs) -> list[str]:
async def fetch_downloaded_info_hashes_from_dl(
user_data: UserData, **kwargs
) -> list[str]:
"""Fetches the info_hashes of all torrents downloaded in the DebridLink account."""
try:
dl_client = DebridLink(token=user_data.streaming_provider.token)
available_torrents = dl_client.get_user_torrent_list()
if "error" in available_torrents:
return []
return [torrent["hashString"] for torrent in available_torrents["value"]]
async with DebridLink(token=user_data.streaming_provider.token) as dl_client:
available_torrents = await dl_client.get_user_torrent_list()
if "error" in available_torrents:
return []
return [torrent["hashString"] for torrent in available_torrents["value"]]
except ProviderException:
return []
def delete_all_torrents_from_dl(user_data: UserData, **kwargs):
async def delete_all_torrents_from_dl(user_data: UserData, **kwargs):
"""Deletes all torrents from the DebridLink account."""
dl_client = DebridLink(token=user_data.streaming_provider.token)
torrents = dl_client.get_user_torrent_list()
for torrent in torrents["value"]:
dl_client.delete_torrent(torrent["id"])
async with DebridLink(token=user_data.streaming_provider.token) as dl_client:
torrents = await dl_client.get_user_torrent_list()
await asyncio.gather(
*[dl_client.delete_torrent(torrent["id"]) for torrent in torrents["value"]]
)
streaming_providers/offcloud/client.py
+124 -67
View File
@@ -1,50 +1,60 @@
from typing import Any
import asyncio
from os import path
from typing import Optional, List
import PTT
import aiohttp
from thefuzz import fuzz
from urllib.request import urlopen
from db.models import TorrentStreams
from streaming_providers.debrid_client import DebridClient
from streaming_providers.exceptions import ProviderException
from utils.validation_helper import is_video_file
from streaming_providers.parser import (
select_file_index_from_torrent,
update_torrent_streams_metadata,
)
class OffCloud(DebridClient):
BASE_URL = "https://offcloud.com/api"
DELETE_URL = "https://offcloud.com"
def initialize_headers(self):
async def initialize_headers(self):
pass
def __del__(self):
async def disable_access_token(self):
pass
def _handle_service_specific_errors(self, error):
pass
async def _handle_service_specific_errors(self, error_data: dict, status_code: int):
if status_code == 403:
raise ProviderException("Invalid OffCloud API key", "invalid_token.mp4")
if status_code == 429:
raise ProviderException(
"OffCloud rate limit exceeded", "too_many_requests.mp4"
)
def _make_request(
async def _make_request(
self,
method: str,
url: str,
data=None,
params=None,
is_return_none=False,
is_expected_to_fail=False,
delete=False,
) -> dict:
data: Optional[dict | str] = None,
json: Optional[dict] = None,
params: Optional[dict] = None,
is_return_none: bool = False,
is_expected_to_fail: bool = False,
retry_count: int = 0,
delete: bool = False,
) -> dict | list:
params = params or {}
params["key"] = self.token
if delete:
url = self.DELETE_URL + url
else:
url = self.BASE_URL + url
return super()._make_request(
method, url, data, params, is_return_none, is_expected_to_fail
full_url = (self.DELETE_URL if delete else self.BASE_URL) + url
return await super()._make_request(
method, full_url, data, json, params, is_return_none, is_expected_to_fail
)
def add_magnet_link(self, magnet_link):
response_data = self._make_request("POST", "/cloud", data={"url": magnet_link})
async def add_magnet_link(self, magnet_link: str) -> dict:
response_data = await self._make_request(
"POST", "/cloud", data={"url": magnet_link}
)
if "requestId" not in response_data:
if "not_available" in response_data:
@@ -58,60 +68,107 @@ class OffCloud(DebridClient):
)
return response_data
def get_user_torrent_list(self):
return self._make_request("GET", "/cloud/history")
async def get_user_torrent_list(self) -> List[dict]:
return await self._make_request("GET", "/cloud/history")
def get_torrent_info(self, request_id):
response = self._make_request(
async def get_torrent_info(self, request_id: str) -> dict:
response = await self._make_request(
"POST", "/cloud/status", data={"requestIds": [request_id]}
)
return response.get("requests")[0] if response.get("requests") else {}
return response.get("requests", [{}])[0]
def get_torrent_instant_availability(self, magnet_links: list[str]):
response = self._make_request("POST", "/cache", data={"hashes": magnet_links})
async def get_torrent_instant_availability(self, magnet_links: List[str]) -> dict:
response = await self._make_request(
"POST", "/cache", data={"hashes": magnet_links}
)
return response.get("cachedItems", {})
def get_available_torrent(self, info_hash) -> dict[str, Any] | None:
available_torrents = self.get_user_torrent_list()
for torrent in available_torrents:
if info_hash.casefold() in torrent["originalLink"].casefold():
return torrent
async def get_available_torrent(self, info_hash: str) -> Optional[dict]:
available_torrents = await self.get_user_torrent_list()
return next(
(
torrent
for torrent in available_torrents
if info_hash.casefold() in torrent.get("originalLink", "").casefold()
),
None,
)
def explore_folder_links(self, request_id):
return self._make_request("GET", f"/cloud/explore/{request_id}")
async def explore_folder_links(self, request_id: str) -> List[str]:
return await self._make_request("GET", f"/cloud/explore/{request_id}")
def create_download_link(self, request_id, torrent_info, filename, episode):
if torrent_info["isDirectory"] is False:
return f"https://{torrent_info.get('server')}.offcloud.com/cloud/download/{request_id}/{torrent_info.get('fileName')}"
async def update_file_sizes(self, files_data: list[dict]):
"""
Update file sizes for a list of files by making HEAD requests.
links = self.explore_folder_links(request_id)
Args:
files_data (list[dict]): List of file data dictionaries containing 'link' keys
if filename:
exact_match = next((link for link in links if filename in link), None)
if exact_match:
return exact_match
Note:
This method modifies the input files_data list in-place, adding 'size' keys
where the HEAD request was successful.
"""
# Fuzzy matching as a fallback
for link in links:
link["fuzzy_ratio"] = fuzz.ratio(filename, link)
selected_file = max(links, key=lambda x: x["fuzzy_ratio"])
async def get_file_size(file_data: dict) -> tuple[dict, Optional[int]]:
"""Helper function to get file size for a single file."""
try:
async with self.session.head(
file_data["link"],
timeout=aiohttp.ClientTimeout(total=5),
allow_redirects=True,
) as response:
if response.status == 200:
return file_data, int(response.headers.get("Content-Length", 0))
except (aiohttp.ClientError, asyncio.TimeoutError):
pass
return file_data, 0
# If the fuzzy ratio is less than 50, then select the largest file
if selected_file["fuzzy_ratio"] < 50:
selected_file = max(
links, key=lambda x: int(urlopen(x).info()["Content-Length"])
# Gather all HEAD requests with proper concurrency
tasks = [get_file_size(file_data) for file_data in files_data]
results = await asyncio.gather(*tasks, return_exceptions=False)
# Update file sizes in the original data
for file_data, size in results:
file_data["size"] = size
async def create_download_link(
self,
request_id: str,
torrent_info: dict,
stream: TorrentStreams,
filename: Optional[str],
season: Optional[int],
episode: Optional[int],
background_tasks,
) -> str:
if not torrent_info["isDirectory"]:
return f"https://{torrent_info['server']}.offcloud.com/cloud/download/{request_id}/{torrent_info['fileName']}"
links = await self.explore_folder_links(request_id)
files_data = [{"name": path.basename(link), "link": link} for link in links]
file_index = await select_file_index_from_torrent(
{"files": files_data},
filename=filename,
episode=episode,
file_size_callback=self.update_file_sizes,
)
if filename is None:
if "size" not in files_data[0]:
await self.update_file_sizes(files_data)
background_tasks.add_task(
update_torrent_streams_metadata,
torrent_stream=stream,
torrent_info={"files": files_data},
file_index=file_index,
season=season,
is_index_trustable=False,
)
selected_file_url = links[file_index]
return selected_file_url
if episode:
# Select the file with the matching episode number
for link in links:
if episode in PTT.parse_title(link).get("episodes", []):
return link
if is_video_file(selected_file):
raise ProviderException(
"No matching file available for this torrent", "no_matching_file.mp4"
)
def delete_torrent(self, request_id):
return self._make_request("GET", f"/cloud/remove/{request_id}", delete=True)
async def delete_torrent(self, request_id: str) -> dict:
return await self._make_request(
"GET", f"/cloud/remove/{request_id}", delete=True
)
streaming_providers/offcloud/utils.py
+77 -54
View File
@@ -1,83 +1,106 @@
import asyncio
from typing import List, Optional
from fastapi import BackgroundTasks
from db.models import TorrentStreams
from db.schemas import UserData
from streaming_providers.exceptions import ProviderException
from streaming_providers.offcloud.client import OffCloud
def get_video_url_from_offcloud(
async def get_video_url_from_offcloud(
info_hash: str,
magnet_link: str,
user_data: UserData,
filename: str,
max_retries=5,
retry_interval=5,
episode: int = None,
stream: TorrentStreams,
background_tasks: BackgroundTasks,
filename: Optional[str] = None,
season: Optional[int] = None,
episode: Optional[int] = None,
max_retries: int = 5,
retry_interval: int = 5,
**kwargs,
) -> str:
oc_client = OffCloud(token=user_data.streaming_provider.token)
async with OffCloud(token=user_data.streaming_provider.token) as oc_client:
# Check if the torrent already exists
torrent_info = await oc_client.get_available_torrent(info_hash)
if torrent_info:
request_id = torrent_info.get("requestId")
torrent_info = await oc_client.get_torrent_info(request_id)
if torrent_info["status"] == "downloaded":
return await oc_client.create_download_link(
request_id,
torrent_info,
stream,
filename,
season,
episode,
background_tasks,
)
if torrent_info["status"] == "error":
raise ProviderException(
f"Error transferring magnet link to OffCloud. {torrent_info['errorMessage']}",
"transfer_error.mp4",
)
else:
# If torrent doesn't exist, add it
response_data = await oc_client.add_magnet_link(magnet_link)
request_id = response_data["requestId"]
# Check if the torrent already exists
torrent_info = oc_client.get_available_torrent(info_hash)
if torrent_info:
request_id = torrent_info.get("requestId")
torrent_info = oc_client.get_torrent_info(request_id)
if torrent_info["status"] == "downloaded":
return oc_client.create_download_link(
request_id, torrent_info, filename, episode
)
if torrent_info["status"] == "error":
raise ProviderException(
f"Error transferring magnet link to OffCloud. {torrent_info['errorMessage']}",
"transfer_error.mp4",
)
else:
# If torrent doesn't exist, add it
response_data = oc_client.add_magnet_link(magnet_link)
request_id = response_data["requestId"]
# Wait for download completion and get the direct link
torrent_info = oc_client.wait_for_status(
request_id, "downloaded", max_retries, retry_interval
)
return oc_client.create_download_link(request_id, torrent_info, filename, episode)
# Wait for download completion and get the direct link
torrent_info = await oc_client.wait_for_status(
request_id, "downloaded", max_retries, retry_interval
)
return await oc_client.create_download_link(
request_id,
torrent_info,
stream,
filename,
season,
episode,
background_tasks,
)
def update_oc_cache_status(
streams: list[TorrentStreams], user_data: UserData, **kwargs
async def update_oc_cache_status(
streams: List[TorrentStreams], user_data: UserData, **kwargs
):
"""Updates the cache status of streams based on OffCloud's instant availability."""
try:
oc_client = OffCloud(token=user_data.streaming_provider.token)
instant_availability_data = oc_client.get_torrent_instant_availability(
[stream.id for stream in streams]
)
for stream in streams:
stream.cached = any(
torrent == stream.id for torrent in instant_availability_data
async with OffCloud(token=user_data.streaming_provider.token) as oc_client:
instant_availability_data = (
await oc_client.get_torrent_instant_availability(
[stream.id for stream in streams]
)
)
for stream in streams:
stream.cached = stream.id in instant_availability_data
except ProviderException:
pass
def fetch_downloaded_info_hashes_from_oc(user_data: UserData, **kwargs) -> list[str]:
async def fetch_downloaded_info_hashes_from_oc(
user_data: UserData, **kwargs
) -> List[str]:
"""Fetches the info_hashes of all torrents downloaded in the OffCloud account."""
try:
oc_client = OffCloud(token=user_data.streaming_provider.token)
available_torrents = oc_client.get_user_torrent_list()
magnet_links = [torrent["originalLink"] for torrent in available_torrents]
return [
magnet_link.split("btih:")[1].split("&")[0] for magnet_link in magnet_links
]
async with OffCloud(token=user_data.streaming_provider.token) as oc_client:
available_torrents = await oc_client.get_user_torrent_list()
return [
torrent["originalLink"].split("btih:")[1].split("&")[0]
for torrent in available_torrents
if "btih:" in torrent["originalLink"]
]
except ProviderException:
return []
def delete_all_torrents_from_oc(user_data: UserData, **kwargs):
async def delete_all_torrents_from_oc(user_data: UserData, **kwargs):
"""Deletes all torrents from the Offcloud account."""
oc_client = OffCloud(token=user_data.streaming_provider.token)
torrents = oc_client.get_user_torrent_list()
for torrent in torrents:
oc_client.delete_torrent(torrent.get("requestId"))
async with OffCloud(token=user_data.streaming_provider.token) as oc_client:
torrents = await oc_client.get_user_torrent_list()
await asyncio.gather(
*[oc_client.delete_torrent(torrent["requestId"]) for torrent in torrents],
return_exceptions=True,
)
streaming_providers/parser.py
+90 -13
View File
@@ -1,40 +1,117 @@
import logging
from datetime import datetime, timezone
from typing import Any, Optional
import PTT
from db.models import TorrentStreams, Episode, Season
from streaming_providers.exceptions import ProviderException
from utils.validation_helper import is_video_file
def select_file_index_from_torrent(
async def select_file_index_from_torrent(
torrent_info: dict[str, Any],
filename: str,
file_index: int,
filename: Optional[str],
episode: Optional[int] = None,
file_key: str = "files",
name_key: str = "name",
size_key: str = "size",
selected_key: Optional[str] = None,
add_leading_slash: bool = False,
file_size_callback: Optional[callable] = None,
) -> int:
"""Select the file index from the torrent info."""
files = torrent_info[file_key]
if selected_key:
files = [file for file in files if file[selected_key] == 1]
if file_index is not None and file_index < len(files):
return file_index
if filename:
if add_leading_slash:
filename = "/" + filename
for index, file in enumerate(files):
if file[name_key] == filename:
if file[name_key] == filename and is_video_file(file[name_key]):
return index
if episode:
# Select the file with the matching episode number
for index, file in enumerate(files):
if episode in PTT.parse_title(file[name_key]).get("episodes", []):
if episode in PTT.parse_title(file[name_key]).get(
"episodes", []
) and is_video_file(file[name_key]):
return index
raise ProviderException(
"No matching file available for this torrent", "no_matching_file.mp4"
)
if file_size_callback:
# Get the file sizes
await file_size_callback(files)
# If no file index is provided, select the largest file
largest_file = max(files, key=lambda file: file[size_key])
return files.index(largest_file)
index = files.index(largest_file)
if is_video_file(largest_file[name_key]):
return index
raise ProviderException(
"No matching file available for this torrent", "no_matching_file.mp4"
)
async def update_torrent_streams_metadata(
torrent_stream: TorrentStreams,
torrent_info: dict,
file_index: int,
season: Optional[int] = None,
file_key: str = "files",
name_key: str = "name",
size_key: str = "size",
remove_leading_slash: bool = False,
is_index_trustable: bool = False,
):
files_data = torrent_info[file_key]
if season is None:
torrent_stream.filename = (
files_data[file_index][name_key].lstrip("/")
if remove_leading_slash
else files_data[file_index][name_key]
)
if file_index is not None and is_index_trustable:
torrent_stream.file_index = file_index
torrent_stream.updated_at = datetime.now(timezone.utc)
await torrent_stream.save()
logging.info(f"Updated {torrent_stream.id} metadata")
else:
parsed_data = [
{
"filename": (
file[name_key].lstrip("/")
if remove_leading_slash
else file[name_key]
),
"size": file.get(size_key),
"index": idx,
**PTT.parse_title(file[name_key]),
}
for idx, file in enumerate(files_data)
]
episodes = [
Episode(
episode_number=file["episodes"][0],
filename=file["filename"],
size=file["size"],
file_index=file["index"] if is_index_trustable else None,
)
for file in parsed_data
if season in file["seasons"] and file["episodes"]
]
if not episodes:
return
torrent_stream.season = Season(
season_number=season,
episodes=episodes,
)
total_size = sum(file["size"] for file in parsed_data if file["size"])
if total_size > torrent_stream.size:
torrent_stream.size = total_size
torrent_stream.updated_at = datetime.now()
await torrent_stream.save()
logging.info(f"Updated {torrent_stream.id} episode data")
streaming_providers/pikpak/utils.py
+79 -87
View File
@@ -1,5 +1,6 @@
import asyncio
import logging
from contextlib import asynccontextmanager
from datetime import datetime
import aiohttp
@@ -122,7 +123,6 @@ async def find_file_in_folder_tree(
my_pack_folder_id: str,
info_hash: str,
filename: str,
file_index: int,
episode: int | None,
) -> dict | None:
torrent_file = await get_torrent_file_by_info_hash(
@@ -136,12 +136,13 @@ async def find_file_in_folder_tree(
else:
files = await get_files_from_folder(pikpak, torrent_file["id"])
file_index = select_file_index_from_torrent(
{"files": files}, filename, file_index, episode
file_index = await select_file_index_from_torrent(
{"files": files}, filename, episode
)
return files[file_index]
@asynccontextmanager
async def initialize_pikpak(user_data: UserData):
cache_key = f"pikpak:{crypto.get_text_hash(user_data.streaming_provider.email + user_data.streaming_provider.password, full_hash=True)}"
if pikpak_encrypted_token := await REDIS_ASYNC_CLIENT.get(cache_key):
@@ -157,39 +158,41 @@ async def initialize_pikpak(user_data: UserData):
token_refresh_callback=store_pikpak_token_in_cache,
token_refresh_callback_kwargs={"user_data": user_data},
)
return pikpak
else:
pikpak = PikPakApi(
username=user_data.streaming_provider.email,
password=user_data.streaming_provider.password,
httpx_client_args={
"transport": httpx.AsyncHTTPTransport(retries=3),
"timeout": 10,
},
token_refresh_callback=store_pikpak_token_in_cache,
token_refresh_callback_kwargs={"user_data": user_data},
)
pikpak = PikPakApi(
username=user_data.streaming_provider.email,
password=user_data.streaming_provider.password,
httpx_client_args={
"transport": httpx.AsyncHTTPTransport(retries=3),
"timeout": 10,
},
token_refresh_callback=store_pikpak_token_in_cache,
token_refresh_callback_kwargs={"user_data": user_data},
)
try:
await pikpak.login()
except PikpakException as error:
if "Invalid username or password" == str(error):
raise ProviderException(
"Invalid PikPak credentials", "invalid_credentials.mp4"
)
logging.error(f"Failed to connect to PikPak: {error}")
raise ProviderException(
"Failed to connect to PikPak. Please try again later.",
"debrid_service_down_error.mp4",
)
except (aiohttp.ClientError, httpx.ReadTimeout):
raise ProviderException(
"Failed to connect to PikPak. Please try again later.",
"debrid_service_down_error.mp4",
)
await store_pikpak_token_in_cache(pikpak, user_data)
try:
await pikpak.login()
except PikpakException as error:
if "Invalid username or password" == str(error):
raise ProviderException(
"Invalid PikPak credentials", "invalid_credentials.mp4"
)
logging.error(f"Failed to connect to PikPak: {error}")
raise ProviderException(
"Failed to connect to PikPak. Please try again later.",
"debrid_service_down_error.mp4",
)
except (aiohttp.ClientError, httpx.ReadTimeout):
raise ProviderException(
"Failed to connect to PikPak. Please try again later.",
"debrid_service_down_error.mp4",
)
await store_pikpak_token_in_cache(pikpak, user_data)
return pikpak
yield pikpak
finally:
await pikpak.httpx_client.aclose()
async def store_pikpak_token_in_cache(pikpak: PikPakApi, user_data: UserData):
@@ -199,7 +202,7 @@ async def store_pikpak_token_in_cache(pikpak: PikPakApi, user_data: UserData):
crypto.encrypt_text(
pikpak.encoded_token, user_data.streaming_provider.password
),
ex=7 * 24 * 60 * 60, # Store for 7 days
ex=5 * 60, # 5 minutes
)
@@ -266,7 +269,7 @@ async def retrieve_or_download_file(
retry_interval: int,
):
selected_file = await find_file_in_folder_tree(
pikpak, my_pack_folder_id, info_hash, filename, stream.file_index, episode
pikpak, my_pack_folder_id, info_hash, filename, episode
)
if not selected_file:
await free_up_space(pikpak, stream.size)
@@ -275,7 +278,7 @@ async def retrieve_or_download_file(
pikpak, info_hash, max_retries, retry_interval
)
selected_file = await find_file_in_folder_tree(
pikpak, my_pack_folder_id, info_hash, filename, stream.file_index, episode
pikpak, my_pack_folder_id, info_hash, filename, episode
)
if selected_file is None:
raise ProviderException(
@@ -328,76 +331,65 @@ async def get_video_url_from_pikpak(
retry_interval=0,
**kwargs,
) -> str:
pikpak = await initialize_pikpak(user_data)
await handle_torrent_status(pikpak, info_hash, max_retries, retry_interval)
async with initialize_pikpak(user_data) as pikpak:
await handle_torrent_status(pikpak, info_hash, max_retries, retry_interval)
my_pack_folder_id = await get_my_pack_folder_id(pikpak)
selected_file = await retrieve_or_download_file(
pikpak,
my_pack_folder_id,
filename,
magnet_link,
info_hash,
stream,
episode,
max_retries,
retry_interval,
)
my_pack_folder_id = await get_my_pack_folder_id(pikpak)
selected_file = await retrieve_or_download_file(
pikpak,
my_pack_folder_id,
filename,
magnet_link,
info_hash,
stream,
episode,
max_retries,
retry_interval,
)
file_data = await pikpak.get_download_url(selected_file["id"])
file_data = await pikpak.get_download_url(selected_file["id"])
# if file_data.get("medias"):
# return file_data["medias"][0]["link"]["url"]
if file_data.get("medias"):
return file_data["medias"][0]["link"]["url"]
return file_data["web_content_link"]
return file_data["web_content_link"]
async def update_pikpak_cache_status(
streams: list[TorrentStreams], user_data: UserData, **kwargs
):
"""Updates the cache status of streams based on PikPak's instant availability."""
try:
pikpak = await initialize_pikpak(user_data)
except ProviderException:
return
tasks = await pikpak.offline_list(phase=["PHASE_TYPE_COMPLETE"])
for stream in streams:
stream.cached = any(
stream.id in task.get("params", {}).get("url", "")
for task in tasks["tasks"]
)
async with initialize_pikpak(user_data) as pikpak:
tasks = await pikpak.offline_list(phase=["PHASE_TYPE_COMPLETE"])
for stream in streams:
stream.cached = any(
stream.id in task.get("params", {}).get("url", "")
for task in tasks["tasks"]
)
async def fetch_downloaded_info_hashes_from_pikpak(
user_data: UserData, **kwargs
) -> list[str]:
"""Fetches the info_hashes of all torrents downloaded in the PikPak account."""
try:
pikpak = await initialize_pikpak(user_data)
except ProviderException:
return []
async with initialize_pikpak(user_data) as pikpak:
my_pack_folder_id = await get_my_pack_folder_id(pikpak)
file_list_content = await pikpak.file_list(parent_id=my_pack_folder_id)
my_pack_folder_id = await get_my_pack_folder_id(pikpak)
file_list_content = await pikpak.file_list(parent_id=my_pack_folder_id)
def _parse_info_hash_from_magnet(magnet_link: str) -> str:
return magnet_link.split(":")[-1]
def _parse_info_hash_from_magnet(magnet_link: str) -> str:
return magnet_link.split(":")[-1]
return [
_parse_info_hash_from_magnet(file.get("params", {}).get("url"))
for file in file_list_content["files"]
if file.get("params", {}).get("url", "").startswith("magnet:")
]
return [
_parse_info_hash_from_magnet(file.get("params", {}).get("url"))
for file in file_list_content["files"]
if file.get("params", {}).get("url", "").startswith("magnet:")
]
async def delete_all_torrents_from_pikpak(user_data: UserData, **kwargs):
"""Deletes all torrents from the PikPak account."""
try:
pikpak = await initialize_pikpak(user_data)
except ProviderException:
return
my_pack_folder_id = await get_my_pack_folder_id(pikpak)
file_list_content = await pikpak.file_list(parent_id=my_pack_folder_id)
file_ids = [file["id"] for file in file_list_content["files"]]
await pikpak.delete_forever(file_ids)
async with initialize_pikpak(user_data) as pikpak:
my_pack_folder_id = await get_my_pack_folder_id(pikpak)
file_list_content = await pikpak.file_list(parent_id=my_pack_folder_id)
file_ids = [file["id"] for file in file_list_content["files"]]
await pikpak.delete_forever(file_ids)
streaming_providers/premiumize/api.py
+16 -14
View File
@@ -13,21 +13,23 @@ async def authorize():
if not Premiumize.OAUTH_CLIENT_ID or not Premiumize.OAUTH_CLIENT_SECRET:
return {"error": "Premiumize OAuth not configured"}
premiumize_client = Premiumize()
return RedirectResponse(
premiumize_client.get_authorization_url(), headers=const.NO_CACHE_HEADERS
)
async with Premiumize() as pm_client:
return RedirectResponse(
pm_client.get_authorization_url(), headers=const.NO_CACHE_HEADERS
)
@router.get("/oauth2_redirect")
async def oauth2_redirect(code: str):
premiumize_client = Premiumize()
token_data = premiumize_client.get_token(code)
token = premiumize_client.encode_token_data(token_data["access_token"])
user_data = schemas.UserData(
streaming_provider=schemas.StreamingProvider(service="premiumize", token=token)
)
encrypted_str = crypto.encrypt_user_data(user_data)
return RedirectResponse(
f"/{encrypted_str}/configure", headers=const.NO_CACHE_HEADERS
)
async with Premiumize() as pm_client:
token_data = await pm_client.get_token(code)
token = pm_client.encode_token_data(token_data["access_token"])
user_data = schemas.UserData(
streaming_provider=schemas.StreamingProvider(
service="premiumize", token=token
)
)
encrypted_str = crypto.encrypt_user_data(user_data)
return RedirectResponse(
f"/{encrypted_str}/configure", headers=const.NO_CACHE_HEADERS
)
streaming_providers/premiumize/client.py
+58 -51
View File
@@ -1,5 +1,4 @@
from base64 import b64encode, b64decode
from typing import Any
from typing import Any, Optional
from urllib.parse import quote_plus
from uuid import uuid4
@@ -17,35 +16,41 @@ class Premiumize(DebridClient):
OAUTH_CLIENT_ID = settings.premiumize_oauth_client_id
OAUTH_CLIENT_SECRET = settings.premiumize_oauth_client_secret
def _handle_service_specific_errors(self, error):
async def _handle_service_specific_errors(self, error_data: dict, status_code: int):
pass
def initialize_headers(self):
async def _make_request(
self,
method: str,
url: str,
data: Optional[dict | str] = None,
json: Optional[dict] = None,
params: Optional[dict] = None,
is_return_none: bool = False,
is_expected_to_fail: bool = False,
retry_count: int = 0,
) -> dict | list:
params = params or {}
if self.is_private_token:
params["apikey"] = self.token
return await super()._make_request(
method, url, data, json, params, is_return_none, is_expected_to_fail
)
async def initialize_headers(self):
if self.token:
token_data = self.decode_token_str(self.token)
self.headers = {"Authorization": f"Bearer {token_data['access_token']}"}
@staticmethod
def encode_token_data(access_token: str):
"""
Premiumize support grant_type device_code which has 10years of token expiration.
"""
return b64encode(str(access_token).encode()).decode()
@staticmethod
def decode_token_str(token: str) -> dict[str, str]:
try:
access_token = b64decode(token).decode()
except ValueError:
raise ProviderException("Invalid token", "invalid_token.mp4")
return {"access_token": access_token}
access_token = self.decode_token_str(self.token)
if access_token:
self.headers = {"Authorization": f"Bearer {access_token}"}
else:
self.is_private_token = True
def get_authorization_url(self) -> str:
state = uuid4().hex
return f"{self.OAUTH_URL}?client_id={self.OAUTH_CLIENT_ID}&response_type=code&redirect_uri={quote_plus(self.REDIRECT_URI)}&state={state}"
def get_token(self, code):
return self._make_request(
async def get_token(self, code):
return await self._make_request(
"POST",
self.OAUTH_TOKEN_URL,
data={
@@ -57,25 +62,25 @@ class Premiumize(DebridClient):
},
)
def add_magnet_link(self, magnet_link: str, folder_id: str = None):
return self._make_request(
async def add_magnet_link(self, magnet_link: str, folder_id: str = None):
return await self._make_request(
"POST",
f"{self.BASE_URL}/transfer/create",
data={"src": magnet_link, "folder_id": folder_id},
)
def create_folder(self, name, parent_id=None):
return self._make_request(
async def create_folder(self, name, parent_id=None):
return await self._make_request(
"POST",
f"{self.BASE_URL}/folder/create",
data={"name": name, "parent_id": parent_id},
)
def get_transfer_list(self):
return self._make_request("GET", f"{self.BASE_URL}/transfer/list")
async def get_transfer_list(self):
return await self._make_request("GET", f"{self.BASE_URL}/transfer/list")
def get_torrent_info(self, torrent_id):
transfer_list = self.get_transfer_list()
async def get_torrent_info(self, torrent_id):
transfer_list = await self.get_transfer_list()
torrent_info = next(
(
torrent
@@ -86,25 +91,25 @@ class Premiumize(DebridClient):
)
return torrent_info
def get_folder_list(self, folder_id: str = None):
return self._make_request(
async def get_folder_list(self, folder_id: str = None):
return await self._make_request(
"GET",
f"{self.BASE_URL}/folder/list",
params={"id": folder_id} if folder_id else None,
)
def delete_folder(self, folder_id: str):
return self._make_request(
async def delete_folder(self, folder_id: str):
return await self._make_request(
"POST", f"{self.BASE_URL}/folder/delete", data={"id": folder_id}
)
def delete_torrent(self, torrent_id):
return self._make_request(
async def delete_torrent(self, torrent_id):
return await self._make_request(
"POST", f"{self.BASE_URL}/transfer/delete", data={"id": torrent_id}
)
def get_torrent_instant_availability(self, torrent_hashes: list[str]):
results = self._make_request(
async def get_torrent_instant_availability(self, torrent_hashes: list[str]):
results = await self._make_request(
"GET", f"{self.BASE_URL}/cache/check", params={"items[]": torrent_hashes}
)
if results.get("status") != "success":
@@ -114,13 +119,11 @@ class Premiumize(DebridClient):
)
return results
def disable_access_token(self):
async def disable_access_token(self):
pass
def get_available_torrent(
self, info_hash: str, torrent_name
) -> dict[str, Any] | None:
torrent_list_response = self.get_transfer_list()
async def get_available_torrent(self, info_hash: str) -> dict[str, Any] | None:
torrent_list_response = await self.get_transfer_list()
if torrent_list_response.get("status") != "success":
if torrent_list_response.get("message") == "Not logged in.":
raise ProviderException(
@@ -132,10 +135,14 @@ class Premiumize(DebridClient):
available_torrents = torrent_list_response["transfers"]
for torrent in available_torrents:
src = torrent.get("src")
if (
(src and info_hash in src)
or info_hash == torrent["name"]
or torrent_name == torrent["name"]
):
return torrent
src = torrent["src"]
async with self.session.head(src) as response:
if response.status != 200:
continue
magnet_link_content = response.headers.get("Content-Disposition")
if info_hash in magnet_link_content:
return torrent
async def get_account_info(self):
return await self._make_request("GET", f"{self.BASE_URL}/account/info")
streaming_providers/premiumize/utils.py
+85 -84
View File
@@ -1,20 +1,20 @@
from typing import Any
from thefuzz import fuzz
import asyncio
from typing import Any, Optional
from db.models import TorrentStreams
from db.schemas import UserData
from streaming_providers.exceptions import ProviderException
from streaming_providers.parser import select_file_index_from_torrent
from streaming_providers.premiumize.client import Premiumize
def create_or_get_folder_id(pm_client: Premiumize, info_hash: str):
folder_data = pm_client.get_folder_list()
for folder in folder_data["content"]:
async def create_or_get_folder_id(pm_client: Premiumize, info_hash: str):
folder_data = await pm_client.get_folder_list()
for folder in folder_data.get("content", []):
if folder["name"] == info_hash:
return folder["id"]
folder_data = pm_client.create_folder(info_hash)
folder_data = await pm_client.create_folder(info_hash)
if folder_data.get("status") != "success":
raise ProviderException(
"Folder already created in meanwhile", "torrent_not_downloaded.mp4"
@@ -22,119 +22,120 @@ def create_or_get_folder_id(pm_client: Premiumize, info_hash: str):
return folder_data.get("id")
def get_video_url_from_premiumize(
async def get_video_url_from_premiumize(
info_hash: str,
magnet_link: str,
user_data: UserData,
torrent_name: str,
filename: str,
filename: Optional[str],
episode: Optional[int],
max_retries=5,
retry_interval=5,
**kwargs,
) -> str:
pm_client = Premiumize(token=user_data.streaming_provider.token)
async with Premiumize(token=user_data.streaming_provider.token) as pm_client:
# Check if the torrent already exists
torrent_info = await pm_client.get_available_torrent(info_hash)
if torrent_info:
torrent_id = torrent_info.get("id")
if torrent_info["status"] == "error":
await pm_client.delete_torrent(torrent_id)
raise ProviderException(
"Not enough seeders available for parse magnet link",
"transfer_error.mp4",
)
else:
# If torrent doesn't exist, add it
folder_id = await create_or_get_folder_id(pm_client, info_hash)
response_data = await pm_client.add_magnet_link(magnet_link, folder_id)
if "id" not in response_data:
raise ProviderException(
"Failed to add magnet link to Premiumize", "transfer_error.mp4"
)
torrent_id = response_data["id"]
# Check if the torrent already exists
torrent_info = pm_client.get_available_torrent(info_hash, torrent_name)
if torrent_info:
torrent_id = torrent_info.get("id")
if torrent_info["status"] == "error":
pm_client.delete_torrent(torrent_id)
raise ProviderException(
"Not enough seeders available for parse magnet link",
"transfer_error.mp4",
)
else:
# If torrent doesn't exist, add it
folder_id = create_or_get_folder_id(pm_client, info_hash)
response_data = pm_client.add_magnet_link(magnet_link, folder_id)
if "id" not in response_data:
raise ProviderException(
"Failed to add magnet link to Real-Debrid", "transfer_error.mp4"
)
torrent_id = response_data["id"]
# Wait for file selection and then start torrent download
torrent_info = pm_client.wait_for_status(
torrent_id, "finished", max_retries, retry_interval
)
return get_stream_link(pm_client, torrent_info, filename, info_hash)
# Wait for file selection and then start torrent download
torrent_info = await pm_client.wait_for_status(
torrent_id, "finished", max_retries, retry_interval
)
return await get_stream_link(
pm_client, torrent_info, filename, info_hash, episode
)
def get_stream_link(
pm_client: Premiumize, torrent_info: dict[str, Any], filename: str, info_hash: str
async def get_stream_link(
pm_client: Premiumize,
torrent_info: dict[str, Any],
filename: Optional[str],
info_hash: str,
episode: Optional[int] = None,
) -> str:
"""Get the stream link from the torrent info."""
if torrent_info["folder_id"] is None:
torrent_folder_data = pm_client.get_folder_list(
create_or_get_folder_id(pm_client, info_hash)
torrent_folder_data = await pm_client.get_folder_list(
await create_or_get_folder_id(pm_client, info_hash)
)
else:
torrent_folder_data = pm_client.get_folder_list(torrent_info["folder_id"])
exact_match = next(
(f for f in torrent_folder_data["content"] if f["name"] == filename), None
torrent_folder_data = await pm_client.get_folder_list(torrent_info["folder_id"])
torrent_info = {
"files": file_data
for file_data in torrent_folder_data["content"]
if "video" in file_data["mime_type"]
}
selected_file_index = select_file_index_from_torrent(
torrent_info,
filename,
episode,
)
if exact_match:
return exact_match["link"]
# Fuzzy matching as a fallback
for file in torrent_folder_data["content"]:
file["fuzzy_ratio"] = fuzz.ratio(filename, file["name"])
selected_file = max(torrent_folder_data["content"], key=lambda x: x["fuzzy_ratio"])
# If the fuzzy ratio is less than 50, then select the largest file
if selected_file["fuzzy_ratio"] < 50:
selected_file = max(
torrent_folder_data["content"], key=lambda x: x.get("size", 0)
)
if "video" not in selected_file["mime_type"]:
raise ProviderException(
"No matching file available for this torrent", "no_matching_file.mp4"
)
selected_file = torrent_info["files"][selected_file_index]
return selected_file["link"]
def update_pm_cache_status(
async def update_pm_cache_status(
streams: list[TorrentStreams], user_data: UserData, **kwargs
):
"""Updates the cache status of streams based on Premiumize's instant availability."""
try:
pm_client = Premiumize(token=user_data.streaming_provider.token)
instant_availability_data = pm_client.get_torrent_instant_availability(
[stream.id for stream in streams]
)
for stream, cached_status in zip(
streams, instant_availability_data.get("response")
):
stream.cached = cached_status
async with Premiumize(token=user_data.streaming_provider.token) as pm_client:
instant_availability_data = (
await pm_client.get_torrent_instant_availability(
[stream.id for stream in streams]
)
)
for stream, cached_status in zip(
streams, instant_availability_data.get("response")
):
stream.cached = cached_status
except ProviderException:
pass
def fetch_downloaded_info_hashes_from_premiumize(
async def fetch_downloaded_info_hashes_from_premiumize(
user_data: UserData, **kwargs
) -> list[str]:
"""Fetches the info_hashes of all torrents downloaded in the Premiumize account."""
try:
pm_client = Premiumize(token=user_data.streaming_provider.token)
available_folders = pm_client.get_folder_list()
return [
folder["name"]
for folder in available_folders["content"]
if folder["name"] and folder["type"] == "folder"
]
async with Premiumize(token=user_data.streaming_provider.token) as pm_client:
available_folders = await pm_client.get_folder_list()
if "content" not in available_folders:
return []
return [
folder["name"]
for folder in available_folders["content"]
if folder["name"] and len(folder["name"]) in (40, 32)
]
except ProviderException:
return []
def delete_all_torrents_from_pm(user_data: UserData, **kwargs):
async def delete_all_torrents_from_pm(user_data: UserData, **kwargs):
"""Deletes all torrents from the Premiumize account."""
pm_client = Premiumize(token=user_data.streaming_provider.token)
folders = pm_client.get_folder_list()
for folder in folders["content"]:
pm_client.delete_folder(folder["id"])
async with Premiumize(token=user_data.streaming_provider.token) as pm_client:
folders = await pm_client.get_folder_list()
delete_tasks = [
pm_client.delete_folder(folder["id"]) for folder in folders["content"]
]
await asyncio.gather(*delete_tasks)
streaming_providers/qbittorrent/utils.py
+38 -31
View File
@@ -2,6 +2,7 @@ import asyncio
from contextlib import asynccontextmanager
from datetime import timedelta
from os import path
from typing import Optional
from urllib.parse import urljoin, urlparse, quote
from aiohttp import ClientConnectorError, ClientSession, CookieJar
@@ -10,11 +11,11 @@ from aioqbt.client import create_client, APIClient
from aioqbt.exc import LoginError, AddTorrentError, NotFoundError
from aiowebdav.client import Client as WebDavClient
from aiowebdav.exceptions import RemoteResourceNotFound, NoConnection
from thefuzz import fuzz
from db.models import TorrentStreams
from db.schemas import UserData
from streaming_providers.exceptions import ProviderException
from streaming_providers.parser import select_file_index_from_torrent
async def check_torrent_status(
@@ -104,7 +105,11 @@ async def get_files_from_folder(
async def find_file_in_folder_tree(
webdav: WebDavClient, user_data: UserData, info_hash: str, filename: str
webdav: WebDavClient,
user_data: UserData,
info_hash: str,
filename: Optional[str],
episode: Optional[int],
) -> dict | None:
base_url_path = urlparse(
user_data.streaming_provider.qbittorrent_config.webdav_url
@@ -119,30 +124,16 @@ async def find_file_in_folder_tree(
if not files:
return None
exact_match = next((file for file in files if file["name"] == filename), None)
if exact_match:
return exact_match
# Fuzzy matching as a fallback
for file in files:
file["fuzzy_ratio"] = fuzz.ratio(filename, file["name"])
selected_file = max(files, key=lambda x: x["fuzzy_ratio"])
# If the fuzzy ratio is less than 50, then select the largest file
if selected_file["fuzzy_ratio"] < 50:
selected_file = max(files, key=lambda x: x["size"])
if "video" not in selected_file["content_type"]:
raise ProviderException(
"No matching file available for this torrent", "no_matching_file.mp4"
)
selected_file_index = await select_file_index_from_torrent(
{"files": files}, filename, episode
)
selected_file = files[selected_file_index]
return selected_file
@asynccontextmanager
async def initialize_qbittorrent(user_data: UserData) -> APIClient:
async with ClientSession(cookie_jar=CookieJar(unsafe=True)) as session:
async with ClientSession(cookie_jar=CookieJar(unsafe=True), timeout=18) as session:
try:
qbittorrent = await create_client(
user_data.streaming_provider.qbittorrent_config.qbittorrent_url.rstrip(
@@ -215,32 +206,45 @@ async def handle_torrent_status(
return torrent
async def set_qbittorrent_preferences(qbittorrent, indexer_type):
if indexer_type == "private":
await qbittorrent.app.set_preferences(
{"dht": False, "pex": False, "lsd": False}
)
# else:
# # Enable DHT, PEX, and LSD for public trackers
# await qbittorrent.app.set_preferences({"dht": True, "pex": True, "lsd": True})
async def retrieve_or_download_file(
qbittorrent: APIClient,
webdav: WebDavClient,
user_data: UserData,
play_video_after: int,
filename: str,
magnet_link: str,
info_hash: str,
indexer_type: str,
filename: Optional[str],
episode: Optional[int],
max_retries: int,
retry_interval: int,
):
selected_file = await find_file_in_folder_tree(
webdav, user_data, info_hash, filename
webdav, user_data, info_hash, filename, episode
)
if not selected_file:
await set_qbittorrent_preferences(qbittorrent, indexer_type)
await add_magnet(qbittorrent, magnet_link, info_hash, user_data)
await wait_for_torrent_to_complete(
qbittorrent, info_hash, play_video_after, max_retries, retry_interval
)
selected_file = await find_file_in_folder_tree(
webdav, user_data, info_hash, filename
)
if selected_file is None:
raise ProviderException(
"No matching file available for this torrent", "no_matching_file.mp4"
webdav, user_data, info_hash, filename, episode
)
if not selected_file:
raise ProviderException(
"No matching file available for this torrent", "no_matching_file.mp4"
)
return selected_file
@@ -270,8 +274,9 @@ async def get_video_url_from_qbittorrent(
info_hash: str,
magnet_link: str,
user_data: UserData,
stream: TorrentStreams,
filename: str,
indexer_type: str,
filename: Optional[str],
episode: Optional[int],
max_retries=5,
retry_interval=5,
**kwargs,
@@ -294,9 +299,11 @@ async def get_video_url_from_qbittorrent(
webdav,
user_data,
play_video_after,
filename or stream.torrent_name,
magnet_link,
info_hash,
indexer_type,
filename,
episode,
max_retries,
retry_interval,
)
streaming_providers/realdebrid/api.py
+7 -7
View File
@@ -10,14 +10,14 @@ router = APIRouter()
@router.get("/get-device-code")
async def get_device_code():
rd_client = RealDebrid()
return JSONResponse(
content=rd_client.get_device_code(), headers=const.NO_CACHE_HEADERS
)
async with RealDebrid() as rd_client:
return JSONResponse(
content=await rd_client.get_device_code(), headers=const.NO_CACHE_HEADERS
)
@router.post("/authorize")
async def authorize(data: AuthorizeData):
rd_client = RealDebrid()
response = rd_client.authorize(data.device_code)
return JSONResponse(content=response, headers=const.NO_CACHE_HEADERS)
async with RealDebrid() as rd_client:
response = await rd_client.authorize(data.device_code)
return JSONResponse(content=response, headers=const.NO_CACHE_HEADERS)
streaming_providers/realdebrid/client.py
+64 -66
View File
@@ -1,5 +1,5 @@
from base64 import b64encode, b64decode
from typing import Any
from typing import Any, Optional, override, overload
from binascii import Error as BinasciiError
@@ -12,50 +12,46 @@ class RealDebrid(DebridClient):
OAUTH_URL = "https://api.real-debrid.com/oauth/v2"
OPENSOURCE_CLIENT_ID = "X245A4XAIBGVM"
def __init__(self, token: str | None = None, user_ip: str | None = None):
def __init__(self, token: Optional[str] = None, user_ip: Optional[str] = None):
self.user_ip = user_ip
self.is_private_token = False
super().__init__(token)
def _handle_service_specific_errors(self, error):
if error.response.status_code == 403:
error_code = error.response.json().get("error_code")
match error_code:
case 9:
raise ProviderException(
"Real-Debrid Permission denied for free account",
"need_premium.mp4",
)
case 22:
raise ProviderException(
"IP address not allowed", "ip_not_allowed.mp4"
)
case 34:
raise ProviderException(
"Too many requests", "too_many_requests.mp4"
)
case 35:
raise ProviderException(
"Content marked as infringing", "content_infringing.mp4"
)
async def _handle_service_specific_errors(self, error_data: dict, status_code: int):
error_code = error_data.get("error_code")
match error_code:
case 9:
raise ProviderException(
"Real-Debrid Permission denied for free account",
"need_premium.mp4",
)
case 22:
raise ProviderException("IP address not allowed", "ip_not_allowed.mp4")
case 34:
raise ProviderException("Too many requests", "too_many_requests.mp4")
case 35:
raise ProviderException(
"Content marked as infringing", "content_infringing.mp4"
)
def _make_request(
async def _make_request(
self,
method: str,
url: str,
data=None,
params=None,
is_return_none=False,
is_expected_to_fail=False,
data: Optional[dict] = None,
json: Optional[dict] = None,
params: Optional[dict] = None,
is_return_none: bool = False,
is_expected_to_fail: bool = False,
retry_count: int = 0,
) -> dict:
if method == "POST" and self.user_ip:
data = data or {}
data["ip"] = self.user_ip
return super()._make_request(
method, url, data, params, is_return_none, is_expected_to_fail
return await super()._make_request(
method, url, data, json, params, is_return_none, is_expected_to_fail
)
def initialize_headers(self):
async def initialize_headers(self):
if self.token:
token_data = self.decode_token_str(self.token)
if "private_token" in token_data:
@@ -64,7 +60,7 @@ class RealDebrid(DebridClient):
}
self.is_private_token = True
else:
access_token_data = self.get_token(
access_token_data = await self.get_token(
token_data["client_id"],
token_data["client_secret"],
token_data["code"],
@@ -74,7 +70,9 @@ class RealDebrid(DebridClient):
}
@staticmethod
def encode_token_data(client_id: str, client_secret: str, code: str):
def encode_token_data(
code: str, client_id: str = None, client_secret: str = None, *args, **kwargs
):
token = f"{client_id}:{client_secret}:{code}"
return b64encode(str(token).encode()).decode()
@@ -86,15 +84,15 @@ class RealDebrid(DebridClient):
return {"private_token": token}
return {"client_id": client_id, "client_secret": client_secret, "code": code}
def get_device_code(self):
return self._make_request(
async def get_device_code(self):
return await self._make_request(
"GET",
f"{self.OAUTH_URL}/device/code",
params={"client_id": self.OPENSOURCE_CLIENT_ID, "new_credentials": "yes"},
)
def get_token(self, client_id, client_secret, device_code):
return self._make_request(
async def get_token(self, client_id, client_secret, device_code):
return await self._make_request(
"POST",
f"{self.OAUTH_URL}/token",
data={
@@ -105,8 +103,8 @@ class RealDebrid(DebridClient):
},
)
def authorize(self, device_code):
response_data = self._make_request(
async def authorize(self, device_code):
response_data = await self._make_request(
"GET",
f"{self.OAUTH_URL}/device/credentials",
params={"client_id": self.OPENSOURCE_CLIENT_ID, "code": device_code},
@@ -116,7 +114,7 @@ class RealDebrid(DebridClient):
if "client_secret" not in response_data:
return response_data
token_data = self.get_token(
token_data = await self.get_token(
response_data["client_id"], response_data["client_secret"], device_code
)
@@ -130,56 +128,56 @@ class RealDebrid(DebridClient):
else:
return token_data
def add_magnet_link(self, magnet_link):
return self._make_request(
async def add_magnet_link(self, magnet_link):
return await self._make_request(
"POST", f"{self.BASE_URL}/torrents/addMagnet", data={"magnet": magnet_link}
)
def get_active_torrents(self):
return self._make_request("GET", f"{self.BASE_URL}/torrents/activeCount")
async def get_active_torrents(self):
return await self._make_request("GET", f"{self.BASE_URL}/torrents/activeCount")
def get_user_torrent_list(self):
return self._make_request("GET", f"{self.BASE_URL}/torrents")
async def get_user_torrent_list(self):
return await self._make_request("GET", f"{self.BASE_URL}/torrents")
def get_user_downloads(self):
return self._make_request("GET", f"{self.BASE_URL}/downloads")
async def get_user_downloads(self):
return await self._make_request("GET", f"{self.BASE_URL}/downloads")
def get_torrent_info(self, torrent_id):
return self._make_request("GET", f"{self.BASE_URL}/torrents/info/{torrent_id}")
async def get_torrent_info(self, torrent_id):
return await self._make_request(
"GET", f"{self.BASE_URL}/torrents/info/{torrent_id}"
)
def get_torrent_instant_availability(self, torrent_hashes: list[str]):
return self._make_request(
async def get_torrent_instant_availability(self, torrent_hashes: list[str]):
return await self._make_request(
"GET",
f"{self.BASE_URL}/torrents/instantAvailability/{'/'.join(torrent_hashes)}",
)
def disable_access_token(self):
if self.is_private_token:
return
return self._make_request(
async def disable_access_token(self):
return await self._make_request(
"GET",
f"{self.BASE_URL}/disable_access_token",
is_return_none=True,
is_expected_to_fail=True,
)
def start_torrent_download(self, torrent_id, file_ids="all"):
return self._make_request(
async def start_torrent_download(self, torrent_id, file_ids="all"):
return await self._make_request(
"POST",
f"{self.BASE_URL}/torrents/selectFiles/{torrent_id}",
data={"files": file_ids},
is_return_none=True,
)
def get_available_torrent(self, info_hash) -> dict[str, Any] | None:
available_torrents = self.get_user_torrent_list()
async def get_available_torrent(self, info_hash) -> Optional[dict[str, Any]]:
available_torrents = await self.get_user_torrent_list()
for torrent in available_torrents:
if torrent["hash"] == info_hash:
return torrent
return None
def create_download_link(self, link):
response = self._make_request(
async def create_download_link(self, link):
response = await self._make_request(
"POST",
f"{self.BASE_URL}/unrestrict/link",
data={"link": link},
@@ -197,8 +195,8 @@ class RealDebrid(DebridClient):
f"Failed to create download link. response: {response}", "api_error.mp4"
)
def delete_torrent(self, torrent_id):
return self._make_request(
async def delete_torrent(self, torrent_id):
return await self._make_request(
"DELETE",
f"{self.BASE_URL}/torrents/delete/{torrent_id}",
is_return_none=True,
streaming_providers/realdebrid/utils.py
+169 -67
View File
@@ -1,122 +1,224 @@
import asyncio
from typing import Optional
from fastapi import BackgroundTasks
from db.models import TorrentStreams
from db.schemas import UserData
from streaming_providers.exceptions import ProviderException
from streaming_providers.parser import (
select_file_index_from_torrent,
update_torrent_streams_metadata,
)
from streaming_providers.realdebrid.client import RealDebrid
from streaming_providers.parser import select_file_index_from_torrent
def create_download_link(rd_client, torrent_info, filename, file_index, episode):
file_index = select_file_index_from_torrent(
async def create_download_link(
rd_client: RealDebrid,
magnet_link: str,
torrent_info: dict,
filename: Optional[str],
file_index: Optional[int],
episode: Optional[int],
season: Optional[int],
stream: TorrentStreams,
background_tasks,
max_retries: int,
retry_interval: int,
) -> str:
selected_file_index = await select_file_index_from_torrent(
torrent_info,
filename,
file_index,
episode,
"files",
"path",
"bytes",
"selected",
True,
)
try:
response = rd_client.create_download_link(torrent_info["links"][file_index])
except IndexError:
raise ProviderException(
"No matching file available for this torrent", "no_matching_file.mp4"
if filename is None or file_index is None:
background_tasks.add_task(
update_torrent_streams_metadata,
torrent_stream=stream,
torrent_info=torrent_info,
file_index=selected_file_index,
season=season,
file_key="files",
name_key="path",
size_key="bytes",
remove_leading_slash=True,
is_index_trustable=True,
)
relevant_file = torrent_info["files"][selected_file_index]
selected_files = [file for file in torrent_info["files"] if file["selected"] == 1]
if relevant_file not in selected_files or len(selected_files) != len(
torrent_info["links"]
):
await rd_client.delete_torrent(torrent_info["id"])
torrent_id = (await rd_client.add_magnet_link(magnet_link)).get("id")
torrent_info = await rd_client.wait_for_status(
torrent_id, "waiting_files_selection", max_retries, retry_interval
)
await rd_client.start_torrent_download(
torrent_info["id"],
file_ids=torrent_info["files"][selected_file_index]["id"],
)
torrent_info = await rd_client.wait_for_status(
torrent_id, "downloaded", max_retries, retry_interval
)
link_index = 0
else:
link_index = selected_files.index(relevant_file)
response = await rd_client.create_download_link(torrent_info["links"][link_index])
if not response.get("mimeType").startswith("video"):
await rd_client.delete_torrent(torrent_info["id"])
raise ProviderException(
f"Requested file is not a video file, deleting torrent and retrying. {response['mimeType']}",
"torrent_not_downloaded.mp4",
)
return response.get("download")
def get_video_url_from_realdebrid(
async def get_video_url_from_realdebrid(
info_hash: str,
magnet_link: str,
user_data: UserData,
filename: str,
file_index: int,
user_ip: str,
filename: Optional[str],
file_index: Optional[int],
stream: TorrentStreams,
background_tasks: BackgroundTasks,
max_retries=5,
retry_interval=5,
user_ip: str = None,
episode: int = None,
episode: Optional[int] = None,
season: Optional[int] = None,
**kwargs,
) -> str:
rd_client = RealDebrid(token=user_data.streaming_provider.token, user_ip=user_ip)
torrent_info = rd_client.get_available_torrent(info_hash)
if not torrent_info:
response = rd_client.get_active_torrents()
if response["limit"] == response["nb"]:
async with RealDebrid(
token=user_data.streaming_provider.token, user_ip=user_ip
) as rd_client:
torrent_info = await rd_client.get_available_torrent(info_hash)
if not torrent_info:
torrent_info = await add_new_torrent(rd_client, magnet_link, info_hash)
torrent_id = torrent_info["id"]
status = torrent_info["status"]
if status in ["magnet_error", "error", "virus", "dead"]:
await rd_client.delete_torrent(torrent_id)
raise ProviderException(
"Torrent limit reached. Please try again later.", "torrent_limit.mp4"
f"Torrent cannot be downloaded due to status: {status}",
"transfer_error.mp4",
)
if info_hash in response["list"]:
raise ProviderException(
"Torrent is already being downloading", "torrent_not_downloaded.mp4"
if status not in ["queued", "downloading", "downloaded"]:
torrent_info = await rd_client.wait_for_status(
torrent_id,
"waiting_files_selection",
max_retries,
retry_interval,
torrent_info,
)
try:
await rd_client.start_torrent_download(
torrent_info["id"],
file_ids="all",
)
except ProviderException as error:
await rd_client.delete_torrent(torrent_id)
raise ProviderException(
f"Failed to start torrent download, {error}", "transfer_error.mp4"
)
torrent_id = rd_client.add_magnet_link(magnet_link).get("id")
if not torrent_id:
raise ProviderException(
"Failed to add magnet link to Real-Debrid", "transfer_error.mp4"
)
torrent_info = rd_client.get_torrent_info(torrent_id)
else:
torrent_id = torrent_info.get("id")
torrent_info = await rd_client.wait_for_status(
torrent_id, "downloaded", max_retries, retry_interval
)
status = torrent_info["status"]
if status in ["magnet_error", "error", "virus", "dead"]:
rd_client.delete_torrent(torrent_id)
return await create_download_link(
rd_client,
magnet_link,
torrent_info,
filename,
file_index,
episode,
season,
stream,
background_tasks,
max_retries,
retry_interval,
)
async def add_new_torrent(rd_client, magnet_link, info_hash):
response = await rd_client.get_active_torrents()
if response["limit"] == response["nb"]:
raise ProviderException(
f"Torrent cannot be downloaded due to status: {status}",
"transfer_error.mp4",
"Torrent limit reached. Please try again later.", "torrent_limit.mp4"
)
elif status in ["queued", "downloading", "downloaded"]:
pass # No action needed, proceed to create download link
else:
# "waiting_files_selection", "magnet_conversion", "compressing", "uploading"
rd_client.wait_for_status(
torrent_id, "waiting_files_selection", max_retries, retry_interval
if info_hash in response["list"]:
raise ProviderException(
"Torrent is already being downloading", "torrent_not_downloaded.mp4"
)
rd_client.start_torrent_download(torrent_id)
torrent_info = rd_client.wait_for_status(
torrent_id, "downloaded", max_retries, retry_interval
)
torrent_id = (await rd_client.add_magnet_link(magnet_link)).get("id")
if not torrent_id:
raise ProviderException(
"Failed to add magnet link to Real-Debrid", "transfer_error.mp4"
)
return create_download_link(rd_client, torrent_info, filename, file_index, episode)
return await rd_client.get_torrent_info(torrent_id)
def update_rd_cache_status(
streams: list[TorrentStreams], user_data: UserData, **kwargs
async def update_rd_cache_status(
streams: list[TorrentStreams], user_data: UserData, user_ip: str, **kwargs
):
"""Updates the cache status of streams based on RealDebrid's instant availability."""
try:
rd_client = RealDebrid(token=user_data.streaming_provider.token)
instant_availability_data = rd_client.get_torrent_instant_availability(
[stream.id for stream in streams]
)
if not instant_availability_data:
return
for stream in streams:
stream.cached = bool(instant_availability_data.get(stream.id, False))
async with RealDebrid(
token=user_data.streaming_provider.token, user_ip=user_ip
) as rd_client:
instant_availability_data = (
await rd_client.get_torrent_instant_availability(
[stream.id for stream in streams]
)
)
if not instant_availability_data:
return
for stream in streams:
stream.cached = bool(instant_availability_data.get(stream.id, False))
except ProviderException:
pass
def fetch_downloaded_info_hashes_from_rd(user_data: UserData, **kwargs) -> list[str]:
async def fetch_downloaded_info_hashes_from_rd(
user_data: UserData, user_ip: str, **kwargs
) -> list[str]:
"""Fetches the info_hashes of all torrents downloaded in the RealDebrid account."""
try:
rd_client = RealDebrid(token=user_data.streaming_provider.token)
available_torrents = rd_client.get_user_torrent_list()
return [torrent["hash"] for torrent in available_torrents]
async with RealDebrid(
token=user_data.streaming_provider.token, user_ip=user_ip
) as rd_client:
available_torrents = await rd_client.get_user_torrent_list()
return [torrent["hash"] for torrent in available_torrents]
except ProviderException:
return []
def delete_all_watchlist_rd(user_data: UserData, **kwargs):
async def delete_all_watchlist_rd(user_data: UserData, user_ip: str, **kwargs):
"""Deletes all torrents from the RealDebrid watchlist."""
rd_client = RealDebrid(token=user_data.streaming_provider.token)
torrents = rd_client.get_user_torrent_list()
for torrent in torrents:
rd_client.delete_torrent(torrent["id"])
async with RealDebrid(
token=user_data.streaming_provider.token, user_ip=user_ip
) as rd_client:
torrents = await rd_client.get_user_torrent_list()
await asyncio.gather(
*[rd_client.delete_torrent(torrent["id"]) for torrent in torrents]
)
streaming_providers/routes.py
+13 -9
View File
@@ -1,6 +1,7 @@
import asyncio
import logging
from os import path
from typing import Annotated
from fastapi import (
Request,
@@ -8,6 +9,7 @@ from fastapi import (
HTTPException,
APIRouter,
Depends,
BackgroundTasks,
)
from fastapi.responses import RedirectResponse
@@ -75,16 +77,11 @@ async def fetch_stream_or_404(info_hash):
if stream:
return stream
# TODO: added for backwards compatibility, remove in the future
stream = await crud.get_stream_by_info_hash(info_hash.upper())
if stream:
return stream
raise HTTPException(status_code=400, detail="Stream not found.")
async def get_or_create_video_url(
stream, user_data, info_hash, season, episode, user_ip
stream, user_data, info_hash, season, episode, user_ip, background_tasks
):
"""
Retrieves or generates the video URL based on stream data and user info.
@@ -94,6 +91,7 @@ async def get_or_create_video_url(
)
episode_data = stream.get_episode(season, episode)
filename = episode_data.filename if episode_data else stream.filename
file_index = episode_data.file_index if episode_data else stream.file_index
get_video_url = mapper.GET_VIDEO_URL_FUNCTIONS.get(
user_data.streaming_provider.service
@@ -103,13 +101,18 @@ async def get_or_create_video_url(
magnet_link=magnet_link,
user_data=user_data,
filename=filename,
file_index=stream.file_index,
file_index=file_index,
user_ip=user_ip,
season=season,
episode=episode,
max_retries=1,
retry_interval=0,
stream=stream,
torrent_name=stream.torrent_name,
background_tasks=background_tasks,
indexer_type=(
"public" if stream.indexer_flags in [[], ["freeleech"]] else "private"
),
)
if asyncio.iscoroutinefunction(get_video_url):
@@ -187,9 +190,10 @@ async def streaming_provider_endpoint(
info_hash: str,
response: Response,
request: Request,
user_data: Annotated[schemas.UserData, Depends(get_user_data)],
background_tasks: BackgroundTasks,
season: int = None,
episode: int = None,
user_data: schemas.UserData = Depends(get_user_data),
):
"""
Handles streaming provider requests, using caching for performance and
@@ -227,7 +231,7 @@ async def streaming_provider_endpoint(
try:
video_url = await get_or_create_video_url(
stream, user_data, info_hash, season, episode, user_ip
stream, user_data, info_hash, season, episode, user_ip, background_tasks
)
await cache_stream_url(cached_stream_url_key, video_url)
video_url = apply_mediaflow_proxy_if_needed(video_url, user_data)
streaming_providers/seedr/api.py
+12 -13
View File
@@ -1,6 +1,6 @@
from fastapi import APIRouter
from fastapi.responses import JSONResponse
from seedrcc import Login
from aioseedrcc import Login
from db.schemas import AuthorizeData
from utils import const
@@ -10,19 +10,18 @@ router = APIRouter()
@router.get("/get-device-code")
async def get_device_code():
seedr = Login()
device_code = seedr.getDeviceCode()
return JSONResponse(content=device_code, headers=const.NO_CACHE_HEADERS)
async with Login() as seedr_login:
device_code = await seedr_login.get_device_code()
return JSONResponse(content=device_code, headers=const.NO_CACHE_HEADERS)
@router.post("/authorize")
async def authorize(data: AuthorizeData):
seedr = Login()
response = seedr.authorize(data.device_code)
if "access_token" in response:
return JSONResponse(
content={"token": seedr.token}, headers=const.NO_CACHE_HEADERS
)
else:
return JSONResponse(content=response, headers=const.NO_CACHE_HEADERS)
async with Login() as seedr_login:
response = await seedr_login.authorize(data.device_code)
if "access_token" in response:
return JSONResponse(
content={"token": seedr_login.token}, headers=const.NO_CACHE_HEADERS
)
else:
return JSONResponse(content=response, headers=const.NO_CACHE_HEADERS)
streaming_providers/seedr/utils.py
+209 -190
View File
@@ -1,186 +1,89 @@
import math
import asyncio
import re
import time
from contextlib import asynccontextmanager
from datetime import datetime
from typing import Optional, Dict, List, Any, Tuple
from enum import Enum
import math
import logging
from seedrcc import Seedr
from thefuzz import fuzz
from aioseedrcc import Seedr
from db.models import TorrentStreams
from db.schemas import UserData
from streaming_providers.exceptions import ProviderException
from streaming_providers.parser import select_file_index_from_torrent
def get_info_hash_folder_id(seedr, info_hash: str):
"""Gets the folder_id of a folder with a given info_hash."""
folder_content = seedr.listContents()
info_hash_folder = next(
(f for f in folder_content["folders"] if f["name"] == info_hash), None
)
if info_hash_folder:
return info_hash_folder["id"]
class TorrentStatus(Enum):
DOWNLOADING = "downloading"
COMPLETED = "completed"
NOT_FOUND = "not_found"
def get_folder_content(seedr, info_hash: str, sub_content_key: str):
"""Gets the folder content based on the info_hash and sub_content_key."""
if info_hash_folder_id := get_info_hash_folder_id(seedr, info_hash):
sub_folder_content = seedr.listContents(info_hash_folder_id)
if sub_folder_content[sub_content_key]:
return sub_folder_content[sub_content_key][0]
def clean_filename(name: Optional[str], replace: str = "") -> str:
"""Clean filename of special characters."""
if not name:
return ""
return re.sub(r"[^a-zA-Z0-9 .,;:_~\-()]", replace, name)
def check_torrent_status(seedr, info_hash: str):
"""Checks if a torrent with a given info_hash is currently downloading."""
return get_folder_content(seedr, info_hash, "torrents")
def check_folder_status(seedr, info_hash: str) -> dict | None:
"""Checks if a torrent with a given folder_name has completed downloading."""
return get_folder_content(seedr, info_hash, "folders")
def add_magnet_and_get_torrent(seedr, magnet_link: str, info_hash: str) -> None:
"""Adds a magnet link to Seedr and returns the corresponding torrent."""
seedr.addFolder(info_hash)
info_hash_folder_id = get_info_hash_folder_id(seedr, info_hash)
transfer = seedr.addTorrent(magnet_link, folderId=info_hash_folder_id)
if transfer["result"] is True:
return
elif transfer["result"] in (
"not_enough_space_added_to_wishlist",
"not_enough_space_wishlist_full",
):
raise ProviderException(
"Not enough space in Seedr account to add this torrent",
"not_enough_space.mp4",
)
elif transfer["result"] == "queue_full_added_to_wishlist":
raise ProviderException(
"Seedr queue is full, remove queued torrents and try again",
"queue_full.mp4",
)
raise ProviderException(
"Error transferring magnet link to Seedr", "transfer_error.mp4"
)
def wait_for_torrent_to_complete(
seedr, info_hash: str, max_retries: int, retry_interval: int
):
"""Waits for a torrent with the given info_hash to complete downloading."""
retries = 0
while retries < max_retries:
torrent = check_torrent_status(seedr, info_hash)
if torrent is None:
return # Torrent was already downloaded
if torrent and torrent.get("progress") == "100":
return
time.sleep(retry_interval)
retries += 1
raise ProviderException("Torrent not downloaded yet.", "torrent_not_downloaded.mp4")
def get_file_details_from_folder(seedr, folder_id: int, filename: str):
"""Gets the details of the file in a given folder."""
folder_content = seedr.listContents(folder_id)
exact_match = [f for f in folder_content["files"] if f["name"] == filename]
if exact_match:
return exact_match[0]
# If the file is not found with exact match, try to find it by fuzzy ratio
for file in folder_content["files"]:
file["fuzzy_ratio"] = fuzz.ratio(filename, file["name"])
selected_file = max(folder_content["files"], key=lambda x: x["fuzzy_ratio"])
# If the fuzzy ratio is less than 50, then select the largest file
if selected_file["fuzzy_ratio"] < 50:
selected_file = max(folder_content["files"], key=lambda x: x["size"])
if selected_file["play_video"] is False:
raise ProviderException(
"No matching file available for this torrent", "no_matching_file.mp4"
)
return selected_file
def seedr_clean_name(name: str, replace: str = "") -> str:
# Only allow alphanumeric characters, spaces, and `.,;:_~-()`
cleaned_name = re.sub(r"[^a-zA-Z0-9 .,;:_~\-()]", replace, name)
return cleaned_name
def get_seedr_client(user_data: UserData) -> Seedr:
"""Returns a Seedr client with the user's token."""
@asynccontextmanager
async def get_seedr_client(user_data: UserData) -> Seedr:
"""Context manager that provides a Seedr client instance."""
try:
seedr = Seedr(token=user_data.streaming_provider.token)
except Exception:
raise ProviderException("Invalid Seedr token", "invalid_token.mp4")
response = seedr.testToken()
if "error" in response:
raise ProviderException("Invalid Seedr token", "invalid_token.mp4")
return seedr
async with Seedr(token=user_data.streaming_provider.token) as seedr:
response = await seedr.test_token()
if "error" in response:
raise ProviderException("Invalid Seedr token", "invalid_token.mp4")
yield seedr
except ProviderException as error:
raise error
except Exception as error:
logging.exception(error)
raise error
def rename_seedr_files(seedr, folder_id):
"""rename filenames to fix stremio android client bug."""
folder_content = seedr.listContents(folder_id)
for file in folder_content["files"]:
# rename file to remove any special characters
if file["name"] != seedr_clean_name(file["name"]):
seedr.renameFile(file["folder_file_id"], seedr_clean_name(file["name"]))
async def get_folder_by_info_hash(
seedr: Seedr, info_hash: str
) -> Optional[Dict[str, Any]]:
"""Find a folder by info_hash."""
contents = await seedr.list_contents()
return next((f for f in contents["folders"] if f["name"] == info_hash), None)
async def get_video_url_from_seedr(
info_hash: str,
magnet_link: str,
user_data: UserData,
stream: TorrentStreams,
filename: str,
max_retries=5,
retry_interval=5,
**kwargs,
) -> str:
"""Gets a direct download link from Seedr using a magnet link and token."""
seedr = get_seedr_client(user_data)
# Check for existing torrent or folder
torrent = check_torrent_status(seedr, info_hash)
if torrent:
wait_for_torrent_to_complete(seedr, info_hash, max_retries, retry_interval)
folder = check_folder_status(seedr, info_hash)
# Handle the torrent based on its status or if it's already in a folder
async def check_torrent_status(
seedr: Seedr, info_hash: str
) -> Tuple[TorrentStatus, Optional[Dict]]:
"""Check the current status of a torrent."""
folder = await get_folder_by_info_hash(seedr, info_hash)
if not folder:
free_up_space(seedr, stream.size)
add_magnet_and_get_torrent(seedr, magnet_link, info_hash)
wait_for_torrent_to_complete(seedr, info_hash, max_retries, retry_interval)
folder = check_folder_status(seedr, info_hash)
folder_id = folder["id"]
return TorrentStatus.NOT_FOUND, None
# rename filenames to fix stremio android client bug.
rename_seedr_files(seedr, folder_id)
folder_content = await seedr.list_contents(folder["id"])
selected_file = get_file_details_from_folder(
seedr,
folder_id,
seedr_clean_name(filename or stream.torrent_name),
)
video_link = seedr.fetchFile(selected_file["folder_file_id"])["url"]
if folder_content["torrents"]:
return TorrentStatus.DOWNLOADING, folder_content["torrents"][0]
elif folder_content["folders"]:
return TorrentStatus.COMPLETED, folder_content["folders"][0]
return video_link
return TorrentStatus.NOT_FOUND, None
def free_up_space(seedr, required_space):
"""Frees up space in the Seedr account by deleting folders until the required space is available."""
contents = seedr.listContents()
async def ensure_space_available(seedr: Seedr, required_space: int | float) -> None:
"""Ensure enough space is available, deleting old content if necessary."""
contents = await seedr.list_contents()
if required_space != math.inf and required_space > contents["space_max"]:
raise ProviderException(
"Not enough space in Seedr account", "not_enough_space.mp4"
)
available_space = contents["space_max"] - contents["space_used"]
if available_space >= required_space:
return # There's enough space, no need to delete anything
return
# Sort folders by size (descending) and last update time
folders = sorted(
contents["folders"],
key=lambda x: (
@@ -192,51 +95,167 @@ def free_up_space(seedr, required_space):
for folder in folders:
if available_space >= required_space:
break
# delete sub folder torrents and folders
sub_folder_content = seedr.listContents(folder["id"])
for sub_folder_torrent in sub_folder_content["torrents"]:
seedr.deleteTorrent(sub_folder_torrent["id"])
for sub_folder in sub_folder_content["folders"]:
seedr.deleteFolder(sub_folder["id"])
seedr.deleteFolder(folder["id"])
# Delete folder contents first
sub_content = await seedr.list_contents(folder["id"])
for torrent in sub_content["torrents"]:
await seedr.delete_item(torrent["id"], "torrent")
for subfolder in sub_content["folders"]:
await seedr.delete_item(subfolder["id"], "folder")
await seedr.delete_item(folder["id"], "folder")
available_space += folder["size"]
def update_seedr_cache_status(
streams: list[TorrentStreams], user_data: UserData, **kwargs
):
"""Updates the cache status of the streams based on the user's Seedr account."""
async def add_torrent(seedr: Seedr, magnet_link: str, info_hash: str) -> None:
"""Add a new torrent to Seedr."""
await seedr.add_folder(info_hash)
folder = await get_folder_by_info_hash(seedr, info_hash)
if not folder:
raise ProviderException("Failed to create folder", "folder_creation_error.mp4")
transfer = await seedr.add_torrent(magnet_link=magnet_link, folder_id=folder["id"])
if transfer["result"] is True:
return
error_messages = {
"not_enough_space_added_to_wishlist": (
"Not enough space in Seedr account",
"not_enough_space.mp4",
),
"not_enough_space_wishlist_full": (
"Not enough space in Seedr account",
"not_enough_space.mp4",
),
"queue_full_added_to_wishlist": ("Seedr queue is full", "queue_full.mp4"),
}
if transfer["result"] in error_messages:
msg, video = error_messages[transfer["result"]]
raise ProviderException(msg, video)
raise ProviderException(
"Error transferring magnet link to Seedr", "transfer_error.mp4"
)
async def wait_for_completion(
seedr: Seedr, info_hash: str, max_retries: int = 1, retry_interval: int = 1
) -> None:
"""Wait for torrent to complete downloading."""
for _ in range(max_retries):
status, data = await check_torrent_status(seedr, info_hash)
if status == TorrentStatus.COMPLETED:
return
elif status == TorrentStatus.DOWNLOADING and data.get("progress") == "100":
return
await asyncio.sleep(retry_interval)
raise ProviderException("Torrent not downloaded yet.", "torrent_not_downloaded.mp4")
async def clean_names(seedr: Seedr, folder_id: str) -> None:
"""Clean special characters from all file and folder names."""
content = await seedr.list_contents(folder_id)
for file in content["files"]:
clean_name = clean_filename(file["name"])
if file["name"] != clean_name:
await seedr.rename_item(file["folder_file_id"], clean_name, "file")
async def get_video_url_from_seedr(
info_hash: str,
magnet_link: str,
user_data: UserData,
stream: TorrentStreams,
filename: Optional[str] = None,
episode: Optional[int] = None,
**kwargs
) -> str:
"""Main function to get video URL from Seedr."""
async with get_seedr_client(user_data) as seedr:
# Check existing torrent status
status, data = await check_torrent_status(seedr, info_hash)
if status == TorrentStatus.NOT_FOUND:
await ensure_space_available(seedr, stream.size)
await add_torrent(seedr, magnet_link, info_hash)
await wait_for_completion(seedr, info_hash)
status, data = await check_torrent_status(seedr, info_hash)
if status != TorrentStatus.COMPLETED or not data:
raise ProviderException(
"Failed to get completed torrent", "torrent_error.mp4"
)
# Clean filenames for compatibility
await clean_names(seedr, data["id"])
# Get file details
folder_content = await seedr.list_contents(data["id"])
file_index = await select_file_index_from_torrent(
folder_content, clean_filename(filename), episode
)
selected_file = folder_content["files"][file_index]
if not selected_file["play_video"]:
raise ProviderException(
"No matching file available", "no_matching_file.mp4"
)
video_data = await seedr.fetch_file(selected_file["folder_file_id"])
return video_data["url"]
async def update_seedr_cache_status(
streams: List[TorrentStreams], user_data: UserData, **kwargs
) -> None:
"""Update cache status for multiple streams."""
try:
seedr = get_seedr_client(user_data)
async with get_seedr_client(user_data) as seedr:
contents = await seedr.list_contents()
# Create lookup of valid info hash folders
folder_map = {
folder["name"]: folder["id"]
for folder in contents["folders"]
if len(folder["name"]) in (40, 32)
}
if not folder_map:
return
# Update stream cache status
for stream in streams:
if stream.id in folder_map:
folder_content = await seedr.list_contents(folder_map[stream.id])
if folder_content["folders"]:
stream.cached = True
except ProviderException:
return
folder_content = {
folder["name"]: folder["id"] for folder in seedr.listContents()["folders"]
}
for stream in streams:
if stream.id in folder_content:
# check if folder is not empty
sub_folder_content = seedr.listContents(folder_content[stream.id])
if sub_folder_content["folders"]:
stream.cached = True
continue
stream.cached = False
def fetch_downloaded_info_hashes_from_seedr(user_data: UserData, **kwargs) -> list[str]:
"""Fetches the info_hashes of all the torrents downloaded in the user's Seedr account."""
async def fetch_downloaded_info_hashes_from_seedr(
user_data: UserData, **kwargs
) -> List[str]:
"""Fetch the info_hashes of all downloaded torrents in the user's account."""
try:
seedr = get_seedr_client(user_data)
async with get_seedr_client(user_data) as seedr:
contents = await seedr.list_contents()
return [
folder["name"]
for folder in contents["folders"]
if len(folder["name"]) in (40, 32)
]
except ProviderException:
return []
return [folder["name"] for folder in seedr.listContents()["folders"]]
def delete_all_torrents_from_seedr(user_data: UserData, **kwargs):
"""Deletes all torrents from the user's Seedr account."""
seedr = get_seedr_client(user_data)
free_up_space(seedr, math.inf)
async def delete_all_torrents_from_seedr(user_data: UserData, **kwargs) -> None:
"""Delete all torrents from the user's account."""
async with get_seedr_client(user_data) as seedr:
await ensure_space_available(seedr, math.inf)
streaming_providers/torbox/client.py
+41 -35
View File
@@ -1,6 +1,4 @@
import json
from typing import Any
from typing import Any, Optional
from streaming_providers.debrid_client import DebridClient
from streaming_providers.exceptions import ProviderException
@@ -9,36 +7,41 @@ from streaming_providers.exceptions import ProviderException
class Torbox(DebridClient):
BASE_URL = "https://api.torbox.app/v1/api"
def initialize_headers(self):
async def initialize_headers(self):
self.headers = {"Authorization": f"Bearer {self.token}"}
def __del__(self):
async def disable_access_token(self):
pass
def _handle_service_specific_errors(self, error):
async def __aexit__(self, exc_type, exc_val, exc_tb):
await super().__aexit__(exc_type, exc_val, exc_tb)
async def _handle_service_specific_errors(self, error_data: dict, status_code: int):
pass
def _make_request(
self,
method: str,
url: str,
data=None,
params=None,
is_return_none=False,
is_expected_to_fail=False,
async def _make_request(
self,
method: str,
url: str,
data: Optional[dict | str] = None,
json: Optional[dict] = None,
params: Optional[dict] = None,
is_return_none: bool = False,
is_expected_to_fail: bool = False,
retry_count: int = 0,
) -> dict:
params = params or {}
url = self.BASE_URL + url
return super()._make_request(
method, url, data, params, is_return_none, is_expected_to_fail
return await super()._make_request(
method, url, data, json, params, is_return_none, is_expected_to_fail
)
def add_magnet_link(self, magnet_link):
response_data = self._make_request(
async def add_magnet_link(self, magnet_link):
response_data = await self._make_request(
"POST",
"/torrents/createtorrent",
data={"magnet": magnet_link},
is_expected_to_fail=True
is_expected_to_fail=True,
)
if response_data.get("detail") is False:
@@ -48,33 +51,37 @@ class Torbox(DebridClient):
)
return response_data
def get_user_torrent_list(self):
return self._make_request("GET", "/torrents/mylist", params={"bypass_cache": "true"})
async def get_user_torrent_list(self):
return await self._make_request(
"GET", "/torrents/mylist", params={"bypass_cache": "true"}
)
def get_torrent_info(self, magnet_id):
response = self.get_user_torrent_list()
async def get_torrent_info(self, magnet_id):
response = await self.get_user_torrent_list()
torrent_list = response.get("data", [])
for torrent in torrent_list:
if torrent.get("magnet", "") == magnet_id:
return torrent
return {}
def get_torrent_instant_availability(self, torrent_hashes: list[str]):
response = self._make_request(
"GET", "/torrents/checkcached", params={"hash": torrent_hashes, "format": "object"}
async def get_torrent_instant_availability(self, torrent_hashes: list[str]):
response = await self._make_request(
"GET",
"/torrents/checkcached",
params={"hash": torrent_hashes, "format": "object"},
)
return response.get("data", {})
return response.get("data", [])
def get_available_torrent(self, info_hash) -> dict[str, Any] | None:
response = self.get_user_torrent_list()
async def get_available_torrent(self, info_hash) -> dict[str, Any] | None:
response = await self.get_user_torrent_list()
torrent_list = response.get("data", [])
for torrent in torrent_list:
if torrent.get("hash", "") == info_hash:
return torrent
return {}
def create_download_link(self, torrent_id, filename):
response = self._make_request(
async def create_download_link(self, torrent_id, filename):
response = await self._make_request(
"GET",
"/torrents/requestdl",
params={"token": self.token, "torrent_id": torrent_id, "file_id": filename},
@@ -87,10 +94,9 @@ class Torbox(DebridClient):
"transfer_error.mp4",
)
def delete_torrent(self, torrent_id):
return self._make_request(
async def delete_torrent(self, torrent_id):
return await self._make_request(
"POST",
"/torrents/controltorrent",
data=json.dumps({"torrent_id": torrent_id, "operation": "delete"})
json={"torrent_id": torrent_id, "operation": "delete"},
)
streaming_providers/torbox/utils.py
+88 -74
View File
@@ -1,116 +1,130 @@
import asyncio
import logging
from typing import Any
from typing import Any, Dict, List, Optional, Iterator
from db.models import TorrentStreams
from db.schemas import UserData
from streaming_providers.exceptions import ProviderException
from streaming_providers.torbox.client import Torbox
from streaming_providers.parser import select_file_index_from_torrent
from streaming_providers.torbox.client import Torbox
def get_video_url_from_torbox(
async def get_video_url_from_torbox(
info_hash: str,
magnet_link: str,
user_data: UserData,
filename: str,
episode: int = None,
**kwargs,
episode: Optional[int] = None,
**kwargs: Any,
) -> str:
torbox_client = Torbox(token=user_data.streaming_provider.token)
# Check if the torrent already exists
torrent_info = torbox_client.get_available_torrent(info_hash)
if torrent_info:
if (
torrent_info["download_finished"] is True
and torrent_info["download_present"] is True
):
file_id = select_file_id_from_torrent(torrent_info, filename, episode)
response = torbox_client.create_download_link(
torrent_info.get("id"),
file_id,
)
return response["data"]
else:
# If torrent doesn't exist, add it
response = torbox_client.add_magnet_link(magnet_link)
# Response detail has "Found Cached Torrent. Using Cached Torrent." if it's a cached torrent,
# create download link from it directly in the same call.
if "Found Cached" in response.get("detail"):
torrent_info = torbox_client.get_available_torrent(info_hash)
if torrent_info:
file_id = select_file_id_from_torrent(torrent_info, filename, episode)
response = torbox_client.create_download_link(
torrent_info.get("id"),
async with Torbox(token=user_data.streaming_provider.token) as torbox_client:
# Check if the torrent already exists
torrent_info = await torbox_client.get_available_torrent(info_hash)
if torrent_info:
if (
torrent_info["download_finished"] is True
and torrent_info["download_present"] is True
):
file_id = await select_file_id_from_torrent(
torrent_info, filename, episode
)
response = await torbox_client.create_download_link(
torrent_info.get("id", ""),
file_id,
)
return response["data"]
else:
# If torrent doesn't exist, add it
response = await torbox_client.add_magnet_link(magnet_link)
# Response detail has "Found Cached Torrent" If it's a cached torrent,
# create a download link from it directly in the same call.
if "Found Cached" in response.get("detail", ""):
torrent_info = await torbox_client.get_available_torrent(info_hash)
if torrent_info:
file_id = await select_file_id_from_torrent(
torrent_info, filename, episode
)
response = await torbox_client.create_download_link(
torrent_info.get("id", ""),
file_id,
)
return response["data"]
raise ProviderException(
f"Torrent did not reach downloaded status.",
"Torrent did not reach downloaded status.",
"torrent_not_downloaded.mp4",
)
# Yield successive n-sized
# chunks from l.
def divide_chunks(lst, n):
# looping till length lst
def divide_chunks(lst: List[Any], n: int) -> Iterator[List[Any]]:
"""Yield successive n-sized chunks from lst."""
for i in range(0, len(lst), n):
yield lst[i:i + n]
yield lst[i : i + n]
def update_torbox_cache_status(
streams: list[TorrentStreams], user_data: UserData, **kwargs
):
async def update_chunk_cache_status(
torbox_client: Torbox, streams_chunk: List[TorrentStreams]
) -> None:
"""Update cache status for a chunk of streams."""
try:
instant_availability_data = (
await torbox_client.get_torrent_instant_availability(
[stream.id for stream in streams_chunk]
)
or []
)
for stream in streams_chunk:
stream.cached = bool(stream.id in instant_availability_data)
except ProviderException as e:
logging.error(f"Failed to get cached status from torbox for a chunk: {e}")
async def update_torbox_cache_status(
streams: List[TorrentStreams], user_data: UserData, **kwargs: Any
) -> None:
"""Updates the cache status of streams based on Torbox's instant availability."""
torbox_client = Torbox(token=user_data.streaming_provider.token)
# Torbox allows only 100 torrents to be passed for cache status, send 80 at a time.
streams_divided_list = list(divide_chunks(streams, 80))
for streams_list in streams_divided_list:
try:
instant_availability_data = torbox_client.get_torrent_instant_availability(
[stream.id for stream in streams_list]
) or []
for stream in streams_list:
stream.cached = bool(stream.id in instant_availability_data)
except ProviderException as e:
logging.error(f"Failed to get cached status from torbox {e}")
pass
async with Torbox(token=user_data.streaming_provider.token) as torbox_client:
# Torbox allows only 100 torrents to be passed for cache status, send 80 at a time.
chunks = list(divide_chunks(streams, 80))
update_tasks = [
update_chunk_cache_status(torbox_client, chunk) for chunk in chunks
]
await asyncio.gather(*update_tasks)
def fetch_downloaded_info_hashes_from_torbox(
user_data: UserData, **kwargs
) -> list[str]:
async def fetch_downloaded_info_hashes_from_torbox(
user_data: UserData, **kwargs: Any
) -> List[str]:
"""Fetches the info_hashes of all torrents downloaded in the Torbox account."""
try:
torbox_client = Torbox(token=user_data.streaming_provider.token)
available_torrents = torbox_client.get_user_torrent_list()
if not available_torrents.get("data"):
return []
return [torrent["hash"] for torrent in available_torrents["data"]]
async with Torbox(token=user_data.streaming_provider.token) as torbox_client:
available_torrents = await torbox_client.get_user_torrent_list()
if not available_torrents.get("data"):
return []
return [torrent["hash"] for torrent in available_torrents["data"]]
except ProviderException:
return []
def select_file_id_from_torrent(
torrent_info: dict[str, Any], filename: str, episode: int
async def select_file_id_from_torrent(
torrent_info: Dict[str, Any], filename: str, episode: Optional[int]
) -> int:
"""Select the file id from the torrent info."""
file_index = select_file_index_from_torrent(
torrent_info, filename, None, episode, name_key="short_name",
file_index = await select_file_index_from_torrent(
torrent_info,
filename,
episode,
name_key="short_name",
)
return torrent_info["files"][file_index]["id"]
def delete_all_torrents_from_torbox(user_data: UserData, **kwargs):
async def delete_all_torrents_from_torbox(user_data: UserData, **kwargs: Any) -> None:
"""Deletes all torrents from the Torbox account."""
torbox_client = Torbox(token=user_data.streaming_provider.token)
torrents = torbox_client.get_user_torrent_list().get("data")
if not torrents:
return
for torrent in torrents:
torbox_client.delete_torrent(torrent.get("id"))
async with Torbox(token=user_data.streaming_provider.token) as torbox_client:
torrents = (await torbox_client.get_user_torrent_list()).get("data")
if not torrents:
return
for torrent in torrents:
await torbox_client.delete_torrent(torrent.get("id", ""))