import copy
import time
from typing import Any, Dict, List, Optional
from lib.api.debrid.torbox import Torbox
from lib.clients.debrid.common import get_file_name, get_packed_release_message
from lib.domain.torrent import TorrentStream
from lib.utils.general.utils import (
DebridType,
IndexerType,
debrid_dialog_update,
get_cached,
get_public_ip,
info_hash_to_magnet,
set_cached,
supported_video_extensions,
)
from lib.utils.kodi.utils import (
dialog_text,
get_setting,
kodilog,
notification,
translation,
)
EXTENSIONS = supported_video_extensions()[:-1]
class TorboxException(Exception):
def __init__(self, message):
self.message = message
super().__init__(self.message)
class TorboxHelper:
def __init__(self):
self.client = Torbox(token=get_setting("torbox_token"))
def check_cached(
self,
results: List[TorrentStream],
cached_results: List[TorrentStream],
uncached_results: List[TorrentStream],
total: int,
dialog: Any,
lock: Any,
) -> None:
hashes = [res.infoHash for res in results if res.infoHash]
cached_response = set()
if hashes:
kodilog(
f"TorboxHelper.check_cached: checking {len(hashes)} hashes against TorBox cache"
)
response = self.client.get_torrent_instant_availability(hashes)
raw_data = response.get("data", [])
for item in raw_data:
if isinstance(item, dict):
if item.get("hash"):
cached_response.add(item.get("hash"))
else:
cached_response.add(item)
missing_hashes = len(results) - len(hashes)
if missing_hashes:
kodilog(
f"TorboxHelper.check_cached: skipping {missing_hashes} results without info_hash"
)
for res in copy.deepcopy(results):
debrid_dialog_update("TB", total, dialog, lock)
res.type = IndexerType.DEBRID
res.debridType = DebridType.TB
with lock:
if res.infoHash in cached_response:
res.isCached = True
cached_results.append(res)
else:
res.isCached = False
uncached_results.append(res)
def add_torbox_torrent(self, info_hash):
kodilog(
f"TorboxHelper.add_torbox_torrent: checking existing torrent for hash={str(info_hash).lower()[:12]!r}"
)
torrent_info = self.client.get_available_torrent(info_hash)
if (
torrent_info
and torrent_info.get("download_finished")
and torrent_info.get("download_present")
):
return torrent_info
magnet = info_hash_to_magnet(info_hash)
kodilog("TorboxHelper.add_torbox_torrent: calling TorBox magnet upload API")
response = self.client.add_magnet_link(magnet)
if not response.get("success"):
raise TorboxException(f"Failed to add magnet link to Torbox: {response}")
if "Found Cached" in response.get("detail", ""):
return self.client.get_available_torrent(info_hash)
kodilog(
f"TorboxHelper: Magnet added successfully, waiting for download. Detail: {response.get('detail', 'N/A')}"
)
return response
def add_torrent_file(self, torrent_data: bytes, torrent_name: str = "torrent.torrent"):
kodilog(
"TorboxHelper.add_torrent_file: calling TorBox 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)
if not response.get("success"):
raise TorboxException(f"Failed to add torrent file to Torbox: {response}")
data = response.get("data", {}) or {}
torrent_hash = data.get("hash", "")
if torrent_hash:
torrent_info = self.client.get_available_torrent(torrent_hash)
if torrent_info:
return torrent_info
return data
def get_cloud_downloads(self) -> List[Dict[str, Any]]:
response = self.client.get_user_torrent_list()
torrents = response.get("data", {}) or []
downloads = []
skipped_not_dict = 0
skipped_not_present = 0
skipped_no_playable = 0
for torrent in torrents:
if not isinstance(torrent, dict):
skipped_not_dict += 1
continue
if not torrent.get("download_present"):
skipped_not_present += 1
continue
files = torrent.get("files", []) or []
video_files = [
item
for item in files
if any(get_file_name(item).lower().endswith(ext) for ext in EXTENSIONS)
]
if not video_files:
skipped_no_playable += 1
continue
file_item = max(video_files, key=lambda item: item.get("size", 0))
downloads.append(
{
"name": file_item.get("name") or torrent.get("name") or "Unknown",
"torrent_id": torrent.get("id"),
"file_id": file_item.get("id"),
"info_hash": torrent.get("hash", ""),
"created_at": torrent.get("created_at", ""),
"updated_at": torrent.get("updated_at", ""),
}
)
kodilog(
f"Torbox cloud listing: total={len(torrents)}, playable={len(downloads)}, skipped_not_dict={skipped_not_dict}, skipped_not_present={skipped_not_present}, skipped_no_playable={skipped_no_playable}"
)
return downloads
def get_link(self, info_hash, data) -> Optional[Dict[str, Any]]:
torrent_id = data.get("torrent_id", "")
file_id = data.get("file_id", "")
if torrent_id and file_id:
response_data = self.client.create_download_link(torrent_id, file_id, get_public_ip())
if response_data:
data["url"] = response_data.get("data", {})
return data
kodilog(
f"Torbox cloud playback failed to create link for torrent_id={torrent_id!r}, file_id={file_id!r}"
)
return None
torrent_info = self.add_torbox_torrent(info_hash)
if not torrent_info:
return None
torrent_files = torrent_info.get("files", []) or []
if not torrent_files and info_hash:
kodilog(
f"TorboxHelper.get_link: torrent has no files yet, retrying metadata fetch for hash={str(info_hash).lower()[:12]!r}"
)
for _ in range(3):
time.sleep(1)
refreshed_info = self.client.get_available_torrent(info_hash)
if refreshed_info:
torrent_info = refreshed_info
torrent_files = refreshed_info.get("files", []) or []
if torrent_files:
break
if not torrent_files:
kodilog(
f"TorboxHelper.get_link: no playable files available yet for hash={str(info_hash).lower()[:12]!r}"
)
notification("Torbox is still processing this torrent. Please try again in a moment.")
return None
video_files = [
item
for item in torrent_files
if any(get_file_name(item).lower().endswith(ext) for ext in EXTENSIONS)
]
if not video_files:
raise TorboxException(get_packed_release_message("Torbox"))
file = max(video_files, key=lambda x: x.get("size", 0))
response_data = self.client.create_download_link(
torrent_info.get("id"), file.get("id"), get_public_ip()
)
if response_data:
data["url"] = response_data.get("data", {})
return data
def get_pack_link(self, data) -> Optional[Dict[str, Any]]:
pack_info = data.get("pack_info", {})
file_id = pack_info.get("file_id", "")
torrent_id = pack_info.get("torrent_id", "")
response_data = self.client.create_download_link(torrent_id, file_id)
if response_data:
data["url"] = response_data.get("data", {})
return data
def get_pack_info(self, info_hash):
info = get_cached(info_hash)
if info:
return info
torrent_info = self.add_torbox_torrent(info_hash)
if not torrent_info:
return None
info = {
"id": torrent_info["id"],
"torrent_id": torrent_info["id"],
"files": [],
}
torrent_files = torrent_info.get("files", [])
if not torrent_files:
notification("No files on the current source")
return None
files = [
(item.get("id"), item["name"])
for id, item in enumerate(torrent_files)
if any(get_file_name(item).lower().endswith(ext) for ext in EXTENSIONS)
]
info["files"] = files
set_cached(info, info_hash)
return info
def get_info(self) -> None:
"""Fetches Torbox account details and displays them."""
response = self.client.get_user()
if not response.get("success"):
notification("Failed to fetch Torbox user info")
return
user = response.get("data", {})
customer_email = user.get("customer_email", "")
plan = user.get("plan", 0)
# Plan mapping (based on general knowledge of Torbox plans, might need adjustment)
plans = {0: "Free", 1: "Essential", 2: "Pro", 3: "Standard"}
plan_name = plans.get(plan, f"Plan {plan}")
body = [
f"[B]Account:[/B] {customer_email}",
f"[B]Plan:[/B] {plan_name}",
f"[B]User ID:[/B] {user.get('auth_id')}",
f"[B]Cooldown Until:[/B] {user.get('cooldown_until') or 'None'}",
]
days = self.client.days_remaining()
if days is not None:
body.append(f"[B]Days Remaining:[/B] {days}")
dialog_text(translation(90656), "\n".join(body))