from base64 import b64decode, b64encode
from lib.api.debrid.base import DebridClient, ProviderException
from lib.services.debrid.auth import run_realdebrid_auth
from lib.services.debrid.download import run_realdebrid_download
from lib.utils.kodi.utils import (
dialog_ok,
kodilog,
set_setting,
translation,
)
class RealDebrid(DebridClient):
BASE_URL = "https://api.real-debrid.com/rest/1.0"
OAUTH_URL = "https://api.real-debrid.com/oauth/v2"
CLIENT_ID = "X245A4XAIBGVM"
def initialize_headers(self):
self.headers = {"User-Agent": "Jacktook/1.0"}
if not self.token or not isinstance(self.token, str):
kodilog("Invalid token")
return
token_data = self.decode_token_str(self.token)
if "private_token" in token_data:
self.headers["Authorization"] = f"Bearer {token_data['private_token']}"
else:
# Exchange client_id/secret/code for an access token
access_token_data = self.get_token(
token_data["client_id"],
token_data["client_secret"],
token_data["code"],
)
self.headers["Authorization"] = f"Bearer {access_token_data['access_token']}"
def _handle_service_specific_errors(self, error_data: dict, status_code: int):
error_code = error_data.get("error_code")
messages = {
-1: "Internal error",
1: "Missing parameter",
2: "Bad parameter value",
3: "Unknown method",
4: "Method not allowed",
5: "Slow down",
6: "Resource unreachable",
7: "Resource not found",
8: "Bad token",
9: "Permission denied",
10: "Two-Factor authentication needed",
11: "Two-Factor authentication pending",
12: "Invalid login",
13: "Invalid password",
14: "Account locked",
15: "Account not activated",
16: "Unsupported hoster",
17: "Hoster in maintenance",
18: "Hoster limit reached",
19: "Hoster temporarily unavailable",
20: "Hoster not available for free users",
21: "Too many active downloads",
22: "IP address not allowed",
23: "Traffic exhausted",
24: "File unavailable",
25: "Service unavailable",
26: "Upload too big",
27: "Upload error",
28: "File not allowed",
29: "Torrent too big",
30: "Torrent file invalid",
31: "Action already done",
32: "Image resolution error",
33: "Torrent already active",
34: "Too many requests",
35: "Infringing file",
36: "Fair Usage Limit",
37: "Disabled endpoint",
}
if error_code in messages:
raise ProviderException(messages[error_code])
err = error_data.get("error") or error_data.get("message")
if err:
raise ProviderException(f"{err} (code {error_code})")
raise ProviderException(f"Real-Debrid error (code {error_code})")
def _make_request(
self,
method,
url,
data=None,
params=None,
json=None,
is_return_none=False,
is_expected_to_fail=False,
):
params = params or {}
return super()._make_request(
method,
url,
data=data,
params=params,
json=json,
is_return_none=is_return_none,
is_expected_to_fail=is_expected_to_fail,
)
@staticmethod
def encode_token_data(client_id, client_secret, code):
token = f"{client_id}:{client_secret}:{code}"
return b64encode(str(token).encode()).decode()
@staticmethod
def decode_token_str(token):
try:
decoded = b64decode(token)
decoded_str = decoded.decode()
parts = decoded_str.split(":")
if len(parts) != 3:
raise ProviderException("Invalid token format")
client_id, client_secret, code = parts
return {
"client_id": client_id,
"client_secret": client_secret,
"code": code,
}
except Exception as e:
raise ProviderException(f"Invalid token {e}") from e
def get_device_code(self):
return self._make_request(
"GET",
f"{self.OAUTH_URL}/device/code",
params={"client_id": self.CLIENT_ID, "new_credentials": "yes"},
)
def get_token(self, client_id, client_secret, device_code):
return self._make_request(
"POST",
f"{self.OAUTH_URL}/token",
data={
"client_id": client_id,
"client_secret": client_secret,
"code": device_code,
"grant_type": "http://oauth.net/grant_type/device/1.0",
},
)
def authorize(self, device_code):
response_data = self._make_request(
"GET",
f"{self.OAUTH_URL}/device/credentials",
params={"client_id": self.CLIENT_ID, "code": device_code},
is_expected_to_fail=True,
)
if "client_secret" not in response_data:
return response_data
token_data = self.get_token(
response_data["client_id"], response_data["client_secret"], device_code
)
if "access_token" in token_data:
token = self.encode_token_data(
response_data["client_id"],
response_data["client_secret"],
token_data["refresh_token"],
)
return {"token": token}
else:
return token_data
def remove_auth(self):
self.token = ""
set_setting("real_debrid_token", "")
set_setting("real_debid_authorized", "false")
set_setting("real_debrid_user", "")
dialog_ok(translation(90544), translation(90561))
def auth(self):
run_realdebrid_auth(self)
def download(self, magnet_url, pack=False):
run_realdebrid_download(self, magnet_url, pack=pack)
def get_hosts(self):
return self._make_request("GET", f"{self.BASE_URL}/hosts")
def add_magnet_link(self, magnet_link):
return self._make_request(
"POST", f"{self.BASE_URL}/torrents/addMagnet", data={"magnet": magnet_link}
)
def get_available_torrent_hosts(self):
return self._make_request("GET", f"{self.BASE_URL}/torrents/availableHosts")
def add_torrent_file(self, torrent_data, torrent_name="torrent.torrent", host=""):
selected_host = host or self._get_default_torrent_host()
if not selected_host:
raise ProviderException("No available Real-Debrid torrent host found")
response = self.session.request(
method="PUT",
url=f"{self.BASE_URL}/torrents/addTorrent",
params={"host": selected_host},
files={
"file": (
torrent_name or "torrent.torrent",
torrent_data,
"application/x-bittorrent",
)
},
headers=self.headers,
timeout=self.timeout,
)
self._handle_errors(response, is_expected_to_fail=False)
return self._parse_response(response, is_return_none=False)
def _get_default_torrent_host(self):
hosts = self.get_available_torrent_hosts()
for host in self._iter_host_candidates(hosts):
if host:
return host
return ""
def _iter_host_candidates(self, payload):
if isinstance(payload, str):
yield payload
return
if isinstance(payload, dict):
preferred_host = payload.get("host") or payload.get("domain")
if isinstance(preferred_host, str) and preferred_host:
yield preferred_host
for key, value in payload.items():
if isinstance(key, str) and "." in key:
yield key
yield from self._iter_host_candidates(value)
return
if isinstance(payload, list):
for item in payload:
yield from self._iter_host_candidates(item)
def get_user_torrent_list(self):
return self._make_request("GET", f"{self.BASE_URL}/torrents")
def get_user_downloads_list(self, page=1):
return self._make_request("GET", f"{self.BASE_URL}/downloads", params={"page": page})
def get_user(self):
return self._make_request("GET", f"{self.BASE_URL}/user")
def days_remaining(self):
try:
user = self.get_user()
if not user or "expiration" not in user:
return None
expiration = user["expiration"]
if not expiration:
return None
import datetime
try:
expires = datetime.datetime.strptime(expiration, "%Y-%m-%dT%H:%M:%S.%fZ")
except ValueError:
expires = datetime.datetime.strptime(expiration, "%Y-%m-%dT%H:%M:%SZ")
days = (expires - datetime.datetime.utcnow()).days
return days
except Exception as e:
kodilog(f"Error calculating RealDebrid days remaining: {e}")
return None
def get_torrent_active_count(self):
return self._make_request("GET", f"{self.BASE_URL}/torrents/activeCount")
def get_torrent_info(self, torrent_id):
return self._make_request("GET", f"{self.BASE_URL}/torrents/info/{torrent_id}")
def get_torrent_instant_availability(self, torrent_hash):
return self._make_request(
"GET", f"{self.BASE_URL}/torrents/instantAvailability/{torrent_hash}"
)
def disable_access_token(self):
pass
def select_files(self, torrent_id, file_ids="all"):
return 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):
available_torrents = self.get_user_torrent_list()
for torrent in available_torrents:
if isinstance(torrent, dict) and torrent.get("hash") == info_hash:
return torrent
def create_download_link(self, link):
response = self._make_request(
"POST",
f"{self.BASE_URL}/unrestrict/link",
data={"link": link},
)
if "download" in response:
return response
if "error_code" in response and response["error_code"] == 23:
raise ProviderException("Exceed remote traffic limit")
raise ProviderException(f"Failed to create download link. response: {response}")
def delete_torrent(self, torrent_id):
return self._make_request(
"DELETE",
f"{self.BASE_URL}/torrents/delete/{torrent_id}",
is_return_none=True,
)