import abc import hashlib import json from copy import deepcopy from typing import Any, Dict, Tuple import xbmcgui from lib.domain.torrent import TorrentStream from lib.utils.debrid.debrid_utils import ( _is_torrent_ready_in_debrid, get_magnet_from_uri, ) from lib.utils.general.utils import Indexer, IndexerType, truncate_text from lib.utils.kodi.logging import summarize_locator_for_log from lib.utils.kodi.utils import ADDON, kodilog, translation from lib.utils.player.utils import resolve_playback_url ACTION_PREVIOUS_MENU = 10 ACTION_PLAYER_STOP = 13 ACTION_NAV_BACK = 92 class BaseWindow(xbmcgui.WindowXMLDialog): def __init__(self, xml_file, location, item_information=None, previous_window=None): super().__init__(xml_file, location) self.item_information = {} self.previous_window = previous_window self.CACHE_KEY = "" self._last_focused_control = (None, None) self.action_exitkeys_id = { ACTION_PREVIOUS_MENU, ACTION_PLAYER_STOP, ACTION_NAV_BACK, } if item_information is None: return self.add_item_information_to_window(item_information) def onInit(self): pass def get_cached_focus(self): if not self.CACHE_KEY: return None, None cache_prop = self._focus_cache_prop() cached_data = xbmcgui.Window(10000).getProperty(cache_prop) if not cached_data: cached_data = ADDON.getSetting(self.CACHE_KEY) if cached_data: xbmcgui.Window(10000).setProperty(cache_prop, cached_data) ADDON.setSetting(self.CACHE_KEY, "") if cached_data: try: return json.loads(cached_data) except json.JSONDecodeError: return None, None return None, None def set_cached_focus(self, control_id, item_id): if not self.CACHE_KEY: return cache_data = json.dumps((control_id, item_id)) xbmcgui.Window(10000).setProperty(self._focus_cache_prop(), cache_data) def _focus_cache_prop(self): cache_key = str(self.CACHE_KEY or "") cache_hash = hashlib.sha1(cache_key.encode("utf-8")).hexdigest() return f"jacktook.focus.{cache_hash}" def set_default_focus(self, control_list=None, control_id=None, control_list_reset=False): try: # Retrieve cached focus if available control_id, item_id = self.get_cached_focus() if control_id and item_id: control = self.getControl(control_id) if isinstance(control, xbmcgui.ControlList): control.selectItem(int(item_id)) self.setFocus(control) return if control_list and control_list.size() > 0: if control_list_reset: control_list.selectItem(0) self.setFocus(control_list) elif control_id: control = self.getControl(control_id) self.setFocus(control) else: raise ValueError("Neither valid control list nor control ID provided.") except (RuntimeError, ValueError) as e: kodilog(f"Could not set focus: {e}") if control_id: self.setFocusId(control_id) def getControlList(self, control_id): try: control = self.getControl(control_id) except RuntimeError as e: kodilog(f"Control does not exist {control_id}", {e}) raise ValueError(f"Control with Id {control_id} does not exist") from e if not isinstance(control, xbmcgui.ControlList): raise AttributeError(f"Control with Id {control_id} should be of type ControlList") return control def add_item_information_to_window(self, item_information): self.item_information = deepcopy(item_information) for i in self.item_information: value = self.item_information[i] if i == "overview": value = truncate_text(str(value)) try: self.setProperty(f"info.{i}", str(value)) except UnicodeEncodeError: self.setProperty(f"info.{i}", value) def getControlProgress(self, control_id): control = self.getControl(control_id) if not isinstance(control, xbmcgui.ControlProgress): raise AttributeError(f"Control with Id {control_id} should be of type ControlProgress") return control def onClick(self, control_id): self.handle_action(7, control_id) def onAction(self, action): action_id = action.getId() if action_id in self.action_exitkeys_id: if self.previous_window: self.previous_window.setProperty("instant_close", "true") self.previous_window.close() self.close() elif action_id != 7: self.handle_action(action_id, self.getFocusId()) def _ensure_playback_info(self, source: TorrentStream): url, magnet, is_torrent = self._extract_source_details(source) if source.addedToDebrid and source.debridType and source.infoHash: is_ready = _is_torrent_ready_in_debrid(source.debridType, source.infoHash) if not is_ready: dialog = xbmcgui.Dialog() message = f"{translation(90696) % source.debridType}\n\n{translation(90697)}" choice = dialog.yesno( translation(90695), message, yeslabel=translation(90698), nolabel=translation(90699), ) if not choice: return {} source.addedToDebrid = False source.debridType = "" source_data = self.prepare_source_data( source=source, url=url, magnet=magnet, is_torrent=is_torrent, ) return resolve_playback_url(source_data) or {} def _extract_source_details(self, source: TorrentStream) -> Tuple[str, str, bool]: url = source.url or "" guid = source.guid or "" magnet = "" indexer = source.indexer source_type = source.type is_torrent = source.type == IndexerType.TORRENT kodilog( "Source extraction input: indexer={!r}, type={!r}, guid={!r}, url={!r}, infohash={!r}".format( indexer, source_type, summarize_locator_for_log(guid), summarize_locator_for_log(url), str(source.infoHash or "").lower()[:12], ) ) if source_type in (IndexerType.DIRECT, IndexerType.STREMIO_DEBRID): return url, magnet, is_torrent # Prefer magnet in guid if guid.startswith("magnet:?"): magnet = guid # Handle Burst indexer elif indexer == Indexer.BURST or guid.endswith(".torrent"): url = guid elif url.endswith(".torrent"): pass # Try to extract magnet from url if url.startswith("http") and not url.endswith(".torrent"): magnet_candidate, _, torrent_url = get_magnet_from_uri(url) if torrent_url: url = torrent_url elif magnet_candidate: magnet = magnet_candidate # Try to extract magnet from guid if it's a details page elif not magnet and guid.startswith("http"): magnet_candidate, _, torrent_url = get_magnet_from_uri(guid) if magnet_candidate: magnet = magnet_candidate elif torrent_url: url = torrent_url # --- Fallback: magnet from url --- if not magnet and url.startswith("magnet:"): magnet, url = url, "" # --- Fallback: magnet from infoHash --- used_infohash_fallback = False has_http_url = bool(url and url.startswith("http")) if not magnet and source.infoHash and not has_http_url: from lib.utils.general.utils import info_hash_to_magnet magnet = info_hash_to_magnet(source.infoHash) used_infohash_fallback = True kodilog( f"Source extraction output: url={summarize_locator_for_log(url)!r}, magnet={summarize_locator_for_log(magnet)!r}, is_torrent={is_torrent}, used_infohash_fallback={used_infohash_fallback}" ) return url, magnet, is_torrent def prepare_source_data( self, source: TorrentStream, url: str, magnet: str, is_torrent: bool, pack_select: bool = False, ) -> Dict[str, Any]: """Prepare the source data dictionary for resolving playback.""" return { "type": source.type, "debrid_type": source.debridType, "indexer": source.indexer, "url": url, "magnet": magnet, "info_hash": source.infoHash, "title": self.item_information.get("title") or self.item_information.get("query") or source.title, "is_torrent": is_torrent, "is_pack": pack_select, "mode": self.item_information.get("mode"), "ids": self.item_information.get("ids"), "tv_data": self.item_information.get("tv_data"), "poster": self.item_information.get("poster"), } @abc.abstractmethod def handle_action(self, action_id, control_id=None): pass