import datetime
import requests
from requests.adapters import HTTPAdapter
from lib.api.debrid.base import DebridClient, ProviderException
from lib.services.debrid.auth import run_torbox_auth
from lib.utils.kodi.utils import (
dialog_ok,
kodilog,
notification,
set_setting,
translation,
)
class Torbox(DebridClient):
def __init__(self, token, timeout=15):
session = requests.Session()
adapter = HTTPAdapter(max_retries=1)
session.mount("https://", adapter)
super().__init__(token, timeout, session)
BASE_URL = "https://api.torbox.app/v1/api"
def initialize_headers(self):
self.headers = {
"User-Agent": "Jacktook/1.0",
"Accept": "application/json",
}
if self.token:
self.headers["Authorization"] = f"Bearer {self.token}"
def disable_access_token(self):
pass
def _handle_service_specific_errors(self, error_data: dict, status_code: int):
error_code = error_data.get("error")
if error_code in {"BAD_TOKEN", "AUTH_ERROR", "OAUTH_VERIFICATION_ERROR"}:
raise ProviderException("Invalid Torbox token")
elif error_code == "DOWNLOAD_TOO_LARGE":
raise ProviderException("Download size too large for the user plan")
elif error_code in {"ACTIVE_LIMIT", "MONTHLY_LIMIT"}:
raise ProviderException("Download limit exceeded")
elif error_code in {"DOWNLOAD_SERVER_ERROR", "DATABASE_ERROR"}:
raise ProviderException("Torbox server error")
def _make_request(
self,
method,
url,
data=None,
params=None,
json=None,
is_return_none=False,
is_expected_to_fail=False,
):
params = params or {}
url = self.BASE_URL + url
response = self._perform_request(method, url, data, params, json)
validated_response = self._validate_error_response(response)
if validated_response is not None:
return validated_response
self._handle_errors(response, is_expected_to_fail)
return self._parse_response(response, is_return_none)
def _validate_error_response(self, response):
if response.ok:
return None
content_type = response.headers.get("Content-Type", "")
if not content_type.startswith("application/json"):
return None
try:
error_data = response.json()
except ValueError:
return None
if not isinstance(error_data, dict):
return None
error_code = error_data.get("error", "")
error_detail = str(error_data.get("detail", "")).lower()
if (
response.status_code == 400
and error_code == "DIFF_ISSUE"
and "already queued" in error_detail
):
kodilog(
"Torbox._validate_error_response: treating duplicate queued torrent as recoverable state"
)
normalized = dict(error_data)
normalized["success"] = True
normalized["queued"] = True
return normalized
return None
def add_magnet_link(self, magnet_link):
return self._make_request(
"POST",
"/torrents/createtorrent",
data={"magnet": magnet_link, "seed": 3, "allow_zip": "false"},
is_expected_to_fail=False,
)
def add_torrent_file(self, torrent_data, torrent_name="torrent.torrent"):
response = self.session.request(
method="POST",
url=f"{self.BASE_URL}/torrents/createtorrent",
data={
"seed": 3,
"allow_zip": "false",
"name": torrent_name or "torrent.torrent",
},
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_user_torrent_list(self):
try:
return self._make_request(
"GET",
"/torrents/mylist",
params={"bypass_cache": "true"},
)
except ProviderException as exc:
if "Request timed out" in str(exc):
raise ProviderException(
"Torbox timed out while checking your cloud torrents. Please try again."
) from exc
raise
def get_torrent_info(self, magnet_id):
response = self.get_user_torrent_list()
torrent_list = response.get("data", {})
for torrent in torrent_list:
if torrent.get("magnet", "") == magnet_id:
return torrent
def get_available_torrent(self, info_hash):
try:
response = self.get_user_torrent_list()
except ProviderException:
raise
torrent_list = response.get("data", {})
for torrent in torrent_list:
if torrent.get("hash", "") == info_hash:
return torrent
def get_torrent_instant_availability(self, torrent_hashes):
return self._make_request(
"POST",
"/torrents/checkcached",
params={"format": "list"},
json={"hashes": torrent_hashes},
)
def create_download_link(self, torrent_id, filename, user_ip=None):
params = {
"token": self.token,
"torrent_id": torrent_id,
"file_id": filename,
}
if user_ip:
params["user_ip"] = user_ip
response = self._make_request(
"GET",
"/torrents/requestdl",
params=params,
is_expected_to_fail=True,
)
if response.get("success"):
return response
else:
detail = response.get("detail", "Unknown error")
notification(f"Failed to create download link: {detail}")
return None
def delete_torrent(self, torrent_id):
self._make_request(
"POST",
"/torrents/controltorrent",
data={"torrent_id": torrent_id, "operation": "Delete"},
is_expected_to_fail=False,
)
def download(self, magnet_link):
pass
def get_device_code(self):
return self._make_request(
"GET",
"/user/auth/device/start",
params={"app": "Jacktook"},
)
def get_token(self, device_code, is_expected_to_fail=False):
return self._make_request(
"POST",
"/user/auth/device/token",
json={"device_code": device_code},
is_expected_to_fail=is_expected_to_fail,
)
def authorize(self, device_code):
response_data = self.get_token(device_code, is_expected_to_fail=True)
if response_data.get("success"):
token_data = response_data.get("data", {})
if isinstance(token_data, dict):
return {"token": token_data.get("access_token")}
return {"token": token_data}
return response_data
def remove_auth(self):
self.token = ""
set_setting("torbox_token", "")
set_setting("torbox_user", "")
set_setting("torbox_enabled", "false")
dialog_ok(translation(90544), translation(90561))
def auth(self):
run_torbox_auth(self)
def get_user(self):
return self._make_request("GET", "/user/me")
def days_remaining(self):
try:
account_info = self.get_user()
if not account_info.get("success"):
return None
data = account_info.get("data", {})
expires_at = data.get("premium_expires_at")
if not expires_at:
return None
# Try common ISO formats
for fmt in (
"%Y-%m-%dT%H:%M:%S.%fZ",
"%Y-%m-%dT%H:%M:%SZ",
"%Y-%m-%dT%H:%M:%S",
):
try:
expires = datetime.datetime.strptime(expires_at, fmt)
break
except ValueError:
continue
else:
return None
days = (expires - datetime.datetime.utcnow()).days
return days
except Exception as e:
kodilog(f"Error calculating days remaining: {e}")
return None