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,
# getattr: cached results are unpickled, and pickle restores __dict__
# directly without applying dataclass defaults -- so sources pickled
# before this field existed genuinely lack the attribute.
"proxyHeaders": getattr(source, "proxyHeaders", None) or {},
"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"),
}
@abc.abstractmethod
def handle_action(self, action_id, control_id=None):
pass