from datetime import timedelta from typing import Any, Dict, Optional, cast from lib.clients.subtitle.submanager import SubtitleManager from lib.db.cached import cache from lib.domain.torrent import TorrentStream from lib.gui.base_window import BaseWindow from lib.gui.source_pack_select import SourcePackSelect from lib.player import JacktookPLayer from lib.utils.debrid.debrid_utils import get_pack_info from lib.utils.kodi.utils import ( ADDON_PATH, get_setting, kodilog, notification, set_property, ) class ResolverWindow(BaseWindow): def __init__( self, xml_file: str, location: str, source: TorrentStream, item_information: Optional[Dict[str, Any]] = None, previous_window: Optional[BaseWindow] = None, close_callback: Optional[Any] = None, is_subtitle_download: bool = False, local_subtitle_path: Optional[str] = None, ) -> None: super().__init__( xml_file, location, item_information=item_information, previous_window=previous_window, ) self.stream_data: Optional[Any] = None self.progress: int = 1 self.resolver: Optional[Any] = None self.source: TorrentStream = source self.pack_select: bool = False self.is_subtitle_download = is_subtitle_download self.local_subtitle_path = local_subtitle_path self.item_information: Dict = item_information or {} self.close_callback: Optional[Any] = close_callback self.playback_info: Optional[Dict[str, Any]] = None self.pack_info: Optional[Any] = None self.previous_window: BaseWindow = previous_window self.setProperty("enable_busy_spinner", "false") self.resolved = False def doModal( self, pack_select: bool = False, ) -> bool: self.pack_select = pack_select self._update_window_properties(self.source) super().doModal() return self.resolved def onInit(self) -> None: super().onInit() self.resolve_source() def close_windows(self): if self.previous_window: self.previous_window.setProperty("instant_close", "true") self.previous_window.close() self.close() def handle_playback_started(self): self.resolved = True self.close_windows() def handle_playback_error(self): self.resolved = False self.setProperty("resolving", "false") self.close() def resolve_source(self) -> Optional[Dict[str, Any]]: try: is_pack = self.source.isPack or self.pack_select has_debrid = bool(self.source.debridType) if is_pack and has_debrid: self.resolve_pack_source() else: self.resolve_single_source() if self.is_subtitle_download: self._download_subtitle() if self.local_subtitle_path and self.playback_info: existing_subtitles = self.playback_info.get("subtitles_path", []) if self.local_subtitle_path not in existing_subtitles: existing_subtitles.append(self.local_subtitle_path) self.playback_info["subtitles_path"] = existing_subtitles set_property("search_subtitles", "true") if not self.playback_info: raise Exception("Failed to resolve source") player = JacktookPLayer( on_started=self.handle_playback_started, on_error=self.handle_playback_error, ) player.run(data=self.playback_info) except Exception as e: import traceback kodilog(f"Error in resolver window: {e}") kodilog(traceback.format_exc()) notification(str(e), time=5000) if self.previous_window: self.previous_window.setProperty("instant_close", "true") self.previous_window.close() self.close() def resolve_single_source(self) -> None: self.playback_info = self._ensure_playback_info(source=self.source) if self.playback_info: self.playback_info.update(self.item_information) self._cache_playback_info() def resolve_pack_source(self) -> None: self.pack_info = get_pack_info( debrid_type=self.source.debridType, info_hash=self.source.infoHash, ) self.window = SourcePackSelect( "source_pack_select.xml", ADDON_PATH, source=self.source, pack_info=self.pack_info, item_information=self.item_information, ) self.playback_info = self.window.doModal() if self.playback_info: self.playback_info.update(self.item_information) self._cache_playback_info() else: self.close_windows() del self.window def _cache_playback_info(self) -> None: playback_info = cast(Dict[str, Any], self.playback_info or {}) if get_setting("super_quick_play", False): kodilog("Super quick play enabled, caching playback info") key = "" if "ids" in playback_info: ids = cast(Dict[str, Any], playback_info["ids"] or {}) key = str(ids.get("original_id") or ids.get("imdb_id") or ids.get("tmdb_id") or "") if key: tv_data = cast(Dict[str, Any], playback_info.get("tv_data", {}) or {}) if tv_data and "season" in tv_data and "episode" in tv_data: key += f"_{tv_data['season']}_{tv_data['episode']}" if key: cache.set(key, playback_info, timedelta(days=30)) kodilog(f"Playback info cached with key: {key}") def _download_subtitle(self): try: subtitle_manager = SubtitleManager(self.playback_info, notification) subtitles_path = subtitle_manager.fetch_subtitles() if subtitles_path: set_property("search_subtitles", "true") if self.playback_info: self.playback_info.update({"subtitles_path": subtitles_path}) else: if subtitle_manager.last_fetch_status == "not_found": kodilog("No subtitles found, continuing without subtitles") elif subtitle_manager.last_fetch_status == "not_selected": kodilog("No subtitles selected, continuing without subtitles") else: kodilog("Subtitle download skipped, continuing without subtitles") except Exception as e: kodilog(f"Subtitle download failed, continuing without subtitles: {e}") def _update_window_properties(self, source: TorrentStream) -> None: self.setProperty("enable_busy_spinner", "true") class SourceException(Exception): def __init__(self, message: str, status_code: Optional[int] = None, error_content: Any = None): self.message = message self.status_code = status_code self.error_content = error_content details = f"{self.message}" if self.status_code is not None: details += f" (Status code: {self.status_code})" if self.error_content is not None: details += f"\nError content: {self.error_content}" super().__init__(details) notification(details)