import copy
import threading
import time
from datetime import datetime
from typing import Any, Dict, List, Optional, cast
from lib.api.debrid.base import ProviderException
from lib.api.debrid.realdebrid import RealDebrid
from lib.clients.debrid.common import (
ensure_direct_playable_file_for_provider,
get_file_name,
)
from lib.domain.torrent import TorrentStream
from lib.utils.general.utils import (
DebridType,
IndexerType,
debrid_dialog_update,
filter_debrid_episode,
get_cached,
info_hash_to_magnet,
set_cached,
supported_video_extensions,
)
from lib.utils.kodi.logging import kodilog
from lib.utils.kodi.utils import (
dialog_text,
get_setting,
translation,
)
class LinkNotFoundError(Exception):
pass
class RealDebridHelper:
def __init__(self) -> None:
self.client = RealDebrid(token=str(get_setting("real_debrid_token", "")))
def check_cached(
self,
results: List[TorrentStream],
cached_results: List[TorrentStream],
uncached_results: List[TorrentStream],
total: int,
dialog: object,
lock: threading.Lock,
) -> None:
# Checks if torrents are cached in Real-Debrid.
torr_available = self.client.get_user_torrent_list()
torr_available_hashes = [
t.get("hash") for t in torr_available if isinstance(t, dict) and t.get("hash")
]
for res in copy.deepcopy(results):
debrid_dialog_update("RD", total, dialog, lock)
res.type = IndexerType.DEBRID
res.debridType = DebridType.RD
with lock:
if res.infoHash in torr_available_hashes:
res.isCached = True
cached_results.append(res)
elif res.isCached:
cached_results.append(res)
else:
res.isCached = False
uncached_results.append(res)
if get_setting("show_uncached"):
cached_results.extend(uncached_results)
def add_magnet(self, info_hash: str, is_pack: bool = False):
"""Adds a magnet link to Real-Debrid and returns the torrent ID."""
try:
kodilog(
f"RealDebridHelper.add_magnet: checking existing torrent for hash={str(info_hash).lower()[:12]!r}"
)
torrent_info = self.client.get_available_torrent(info_hash)
if not torrent_info:
self.check_max_active_count()
kodilog("RealDebridHelper.add_magnet: calling Real-Debrid magnet upload API")
response = self.client.add_magnet_link(info_hash_to_magnet(info_hash))
torrent_id = response.get("id")
if not torrent_id:
raise ProviderException("Failed to add magnet link to Real-Debrid")
torrent_info = self.client.get_torrent_info(torrent_id)
self._handle_torrent_status(torrent_info, is_pack)
return torrent_info.get("id")
except Exception as e:
raise ProviderException(str(e)) from e
def add_torrent_file(
self,
torrent_data: bytes,
torrent_name: str = "torrent.torrent",
is_pack: bool = False,
) -> Optional[str]:
"""Uploads a torrent file to Real-Debrid and returns the torrent ID."""
try:
self.check_max_active_count()
kodilog(
"RealDebridHelper.add_torrent_file: calling Real-Debrid torrent file upload API with torrent_name={!r}, size={} bytes".format(
torrent_name,
len(torrent_data or b""),
)
)
response = self.client.add_torrent_file(torrent_data, torrent_name=torrent_name)
torrent_id = response.get("id")
if not torrent_id:
raise ProviderException("Failed to upload torrent file to Real-Debrid")
torrent_info = self.client.get_torrent_info(torrent_id)
self._handle_torrent_status(torrent_info, is_pack)
return torrent_info.get("id")
except Exception as e:
raise ProviderException(str(e)) from e
def _handle_torrent_status(self, torrent_info: Dict, is_pack: bool = False) -> Optional[str]:
"""Processes torrent_info status and handles errors or file selection."""
status = torrent_info["status"]
torrent_id = torrent_info.get("id", "unknown")
if status in ["magnet_error", "error", "virus", "dead"]:
self.client.delete_torrent(torrent_info["id"])
raise ProviderException(f"Torrent cannot be downloaded: {status}")
elif status in ["queued", "downloading", "magnet_conversion"]:
kodilog(
f"RealDebridHelper: Torrent {torrent_id} is still being processed (status: {status}). It has been added to your cloud but is not ready yet."
)
return torrent_id
elif status == "waiting_files_selection":
if torrent_info.get("files"):
self.handle_file_selection(torrent_info, is_pack)
else:
raise ProviderException("No files available for this torrent yet.")
def handle_file_selection(self, torrent_info: Dict, is_pack: bool) -> None:
"""Handles file selection for Real-Debrid torrents."""
files = torrent_info["files"]
extensions = supported_video_extensions()[:-1]
video_files = [
item for item in files if any(item["path"].lower().endswith(ext) for ext in extensions)
]
if video_files:
torrents_ids = (
[str(i["id"]) for i in video_files]
if is_pack or len(video_files) > 1
else [str(video_files[0]["id"])]
)
if torrents_ids:
self.client.select_files(torrent_info["id"], ",".join(torrents_ids))
def get_link(self, info_hash: str, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""Gets a direct download link for a Real-Debrid torrent."""
torrent_id = self.add_magnet(info_hash)
torrent_info = self.client.get_torrent_info(torrent_id)
links = torrent_info.get("links", [])
files = torrent_info.get("files", [])
if not links:
raise ProviderException("No links found in torrent info.")
def create_download_for_link(link: str) -> str:
response = self.client.create_download_link(link)
url = response.get("download")
if not url:
raise ProviderException("File not cached!")
return url
# --- Single-file torrent ---
if len(links) == 1:
single_file = files[0] if files else {}
ensure_direct_playable_file_for_provider(get_file_name(single_file), "Real-Debrid")
data["url"] = create_download_for_link(links[0])
return data
# --- Multi-file torrent ---
tv_data = data.get("tv_data")
if tv_data:
season = tv_data.get("season")
episode = tv_data.get("episode")
possible_matches = filter_debrid_episode(files, episode_num=episode, season_num=season)
if not possible_matches:
raise ProviderException("No matching episode found in torrent.")
match_file = next((f for f in possible_matches if f.get("selected") == 1), None)
if not match_file:
raise ValueError("File is not cached")
ensure_direct_playable_file_for_provider(get_file_name(match_file), "Real-Debrid")
selected_files = [f for f in files if f.get("selected") == 1]
try:
file_index = selected_files.index(match_file)
except ValueError as exc:
raise ProviderException("Could not map episode to Real-Debrid link.") from exc
data["url"] = create_download_for_link(links[file_index])
return data
selected_files = [f for f in files if f.get("selected") == 1]
if selected_files:
largest_file = max(selected_files, key=lambda f: f.get("bytes", 0))
ensure_direct_playable_file_for_provider(get_file_name(largest_file), "Real-Debrid")
try:
file_index = selected_files.index(largest_file)
except ValueError:
file_index = -1
if 0 <= file_index < len(links):
data["url"] = create_download_for_link(links[file_index])
return data
data["is_pack"] = True
return data
def get_pack_link(self, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""Gets a direct download link for a file inside a Real-Debrid torrent pack."""
pack_info = cast(Dict[str, Any], data.get("pack_info", {}) or {})
file_position = pack_info.get("file_position", "")
torrent_id = pack_info.get("torrent_id", "")
torrent_info = self.client.get_torrent_info(torrent_id)
links = torrent_info.get("links", [])
response = self.client.create_download_link(links[file_position])
url = response.get("download")
if not url:
raise ProviderException("Failed to retrieve download link")
data["url"] = url
return data
def get_pack_info(self, info_hash: str) -> Optional[Dict[str, Any]]:
"""Retrieves information about a torrent pack, including file names."""
cached_info = get_cached(info_hash)
if cached_info:
return cast(Dict[str, Any], cached_info)
torrent_id = self.add_magnet(info_hash, is_pack=True)
if not torrent_id:
return None
torr_info = self.client.get_torrent_info(torrent_id)
torrent_files = torr_info["files"]
if len(torrent_files) <= 1:
raise ProviderException("No files on the current source")
torr_items = [item for item in torrent_files if item["selected"] == 1]
files = [(item["id"], item["path"].split("/", 1)[1]) for item in torr_items]
info = {"torrent_id": torr_info["id"], "files": files}
set_cached(info, info_hash)
return info
def get_info(self) -> None:
"""Fetches Real-Debrid account details and displays them."""
user = self.client.get_user()
expiration = user["expiration"]
try:
expires = datetime.strptime(expiration, "%Y-%m-%dT%H:%M:%S.%fZ")
except ValueError:
expires = datetime(*(time.strptime(expiration, "%Y-%m-%dT%H:%M:%S.%fZ")[0:6]))
days_remaining = (expires - datetime.today()).days
body = [
f"[B]Account:[/B] {user['email']}",
f"[B]Username:[/B] {user['username']}",
f"[B]Status:[/B] {user['type'].capitalize()}",
f"[B]Expires:[/B] {expires}",
f"[B]Days Remaining:[/B] {days_remaining}",
f"[B]Fidelity Points:[/B] {user['points']}",
]
dialog_text(translation(90659), "\n".join(body))
def check_max_active_count(self) -> None:
"""Ensures Real-Debrid does not exceed active torrent limit."""
active_count = self.client.get_torrent_active_count()
if active_count.get("nb", 0) >= active_count.get("limit", 0):
hashes = active_count.get("list", [])
if hashes:
torrents = self.client.get_user_torrent_list()
torrent_info = next(
(
item
for item in torrents
if isinstance(item, dict) and item.get("hash", "") == hashes[0]
),
None,
)
if torrent_info:
self.client.delete_torrent(torrent_info.get("id"))