import threading from datetime import datetime from typing import Any, Dict, List, Optional from lib.api.debrid.alldebrid import AllDebrid from lib.api.debrid.base import ProviderException 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, filter_debrid_episode, info_hash_to_magnet, supported_video_extensions, ) from lib.utils.kodi.utils import ( dialog_text, get_setting, kodilog, notification, translation, ) class AllDebridHelper: def __init__(self): self.client = AllDebrid(token=str(get_setting("alldebrid_token", ""))) def check_cached( self, results: List[TorrentStream], cached_results: List[TorrentStream], uncached_results: List[TorrentStream], total: int, dialog: Any, lock: threading.Lock, ) -> None: # Checks if torrents are cached in AllDebrid. torr_available = self.client.get_user_torrent_list() magnets = ( torr_available.get("data", {}).get("magnets") or torr_available.get("magnets") or [] ) torr_available_hashes = [ m.get("hash") for m in magnets if isinstance(m, dict) and m.get("hash") ] for res in results: debrid_dialog_update(DebridType.AD, total, dialog, lock) res.type = IndexerType.DEBRID res.debridType = DebridType.AD 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 get_link(self, info_hash: str, data: Dict[str, Any]) -> Optional[Dict[str, Any]]: result = self.client.add_magnet(info_hash_to_magnet(info_hash)) magnet = result.get("data", {}).get("magnets", [])[0] if not magnet.get("ready"): raise ProviderException("Torrent is not yet cached!") torrent_id = magnet.get("id") # --- Get the files (and their direct links) --- torr_info = self.client.get_files_and_links(torrent_id) files = torr_info[0].get("files", []) if not files: raise ProviderException("No files found in torrent info.") # Flatten file list flat_files = self.flatten_files(files) # --- Single-file torrents --- if not data.get("tv_data"): # Filter only supported video files all_links = [ f.get("l") for f in flat_files if any( get_file_name(f).lower().endswith(ext) for ext in supported_video_extensions()[:-1] ) ] if not all_links: raise ProviderException("No cached links available!") data["url"] = self.create_download_for_link(all_links[0]) return data # --- Multi-file torrents (TV shows) --- season = data["tv_data"].get("season", "") episode = data["tv_data"].get("episode", "") matched = filter_debrid_episode(flat_files, episode_num=episode, season_num=season) if not matched: raise ProviderException("No matching episode found in torrent.") matched_link = matched[0]["l"] direct_link = self.create_download_for_link(matched_link) data["url"] = direct_link return data def add_torrent_file( self, torrent_data: bytes, torrent_name: str = "torrent.torrent" ) -> Optional[Dict[str, Any]]: kodilog( "AllDebridHelper.add_torrent_file: calling AllDebrid 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) files = response.get("data", {}).get("files", []) or [] if not files: raise ProviderException("Failed to upload torrent file to AllDebrid") file_info = files[0] if file_info.get("error"): error = file_info.get("error", {}) raise ProviderException(error.get("message") or "Invalid torrent file") return file_info def flatten_files(self, files: list) -> list: flat_list = [] for f in files: if "e" in f: flat_list.extend(self.flatten_files(f["e"])) else: if "l" in f: flat_list.append(f) return flat_list 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 = data.get("pack_info", {}) link = pack_info.get("url", "") data["url"] = self.create_download_for_link(link) return data def get_pack_info(self, info_hash: str) -> Optional[Dict]: result = self.client.add_magnet(info_hash_to_magnet(info_hash)) magnet = result.get("data", {}).get("magnets", [])[0] if not magnet.get("ready"): raise ProviderException("Torrent is not yet cached!") torrent_id = magnet.get("id") # --- Get the files (and their direct links) --- torr_info = self.client.get_files_and_links(torrent_id) files = torr_info[0].get("files", []) if not files: raise ProviderException("No files found in torrent info.") # Flatten file list flat_files = self.flatten_files(files) # Filter only supported video files files = [ (f.get("l"), get_file_name(f)) for f in flat_files if any( get_file_name(f).lower().endswith(ext) for ext in supported_video_extensions()[:-1] ) ] if not files: notification(get_packed_release_message("AllDebrid")) raise info = {"files": files} return info def create_download_for_link(self, link: str): response = self.client.create_download_link(link) if not response: raise ProviderException("Failed to unlock the download link.") url = response.get("data", {}).get("link") if not url: raise ProviderException("File not cached!") return url def get_info(self) -> None: try: response = self.client.get_user_info() user = response["user"] except Exception as exc: raise ProviderException("Failed to retrieve All-Debrid user info.") from exc status = "Premium" if user["isPremium"] else "Not Active" expires = datetime.fromtimestamp(user["premiumUntil"]) days_remaining = (expires - datetime.today()).days body = [ f"[B]Account:[/B] {user['email']}", f"[B]Username:[/B] {user['username']}", f"[B]Status:[/B] {status}", f"[B]Expires:[/B] {expires}", f"[B]Days Remaining:[/B] {days_remaining}", f"[B]Fidelity Points:[/B] {user['fidelityPoints']}", ] dialog_text(translation(90658), "\n".join(body))