import contextlib
import threading
from typing import Dict, List, Optional
import xbmc
import xbmcgui
from lib.clients.stremio.constants import STREMIO_ADDON_ALIASES_KEY
from lib.db.cached import cache
from lib.domain.torrent import TorrentStream
from lib.gui.base_window import BaseWindow
from lib.gui.filter_items_window import FilterWindow
from lib.gui.filter_type_window import FilterTypeWindow
from lib.gui.qr_progress_dialog import QRProgressDialog
from lib.gui.resolver_window import ResolverWindow
from lib.gui.resume_window import ResumeDialog
# For web subtitle upload
from lib.services.subtitle_server import SubtitleUploadServer, get_local_ip
from lib.utils.debrid.debrid_utils import (
add_source_to_debrid,
get_source_status,
get_torrent_data_from_uri,
)
from lib.utils.debrid.qrcode_utils import make_qrcode
from lib.utils.general.utils import (
IndexerType,
extract_publish_date,
format_season_episode,
get_colored_languages,
get_info_hash_from_magnet,
get_provider_color,
get_random_color,
normalize_tv_data,
safe_json_loads,
)
from lib.utils.kodi.utils import (
ADDON_PATH,
action_url_run,
bytes_to_human_readable,
get_setting,
kodilog,
notification,
translation,
)
from lib.utils.parsers.title_parser import parse_title_info
from lib.utils.torrent.torrserver_utils import add_source_to_torrserver
THEMES = {
"0": {"card_bg": "FF362e33", "card_focus": "992A3E5C", "card_accent": "FF00559D"},
"1": {"card_bg": "FF000000", "card_focus": "FF1A1A1A", "card_accent": "FF333333"},
"2": {"card_bg": "FF1A0B2E", "card_focus": "994D004D", "card_accent": "FF00F3FF"},
"3": {"card_bg": "FF1B261B", "card_focus": "992D402D", "card_accent": "FF7CFC00"},
"4": {"card_bg": "FF141414", "card_focus": "992B2510", "card_accent": "FFD4AF37"},
}
def _get_source_indexer_label(source: TorrentStream, aliases: Dict) -> str:
current_alias = aliases.get(source.addonKey) if isinstance(aliases, dict) else ""
if current_alias:
if source.addonInstanceLabel and source.addonName:
return source.addonInstanceLabel.replace(source.addonName, current_alias, 1)
return current_alias
return source.addonInstanceLabel or source.indexer
class SourceSelect(BaseWindow):
def __init__(
self,
xml_file: str,
location: str,
item_information: Optional[Dict] = None,
sources: Optional[List[TorrentStream]] = None,
uncached: Optional[List[TorrentStream]] = None,
):
super().__init__(xml_file, location, item_information=item_information)
self.uncached_sources: List[TorrentStream] = uncached or []
self.position: int = -1
self.sources: List[TorrentStream] = sources or []
self.list_sources: List[TorrentStream] = []
self.item_information: Dict = item_information or {}
self.playback_info: Optional[Dict] = None
self.resume: Optional[bool] = None
self.CACHE_KEY: str = str(self.item_information.get("tv_data", "")) or str(
self.item_information.get("ids", "")
)
self.setProperty("instant_close", "false")
self.setProperty("resolving", "false")
self.filtered_sources: Optional[List[TorrentStream]] = None
self.filter_applied: bool = False
self.resolved = False
def onInit(self) -> None:
theme_index = get_setting("source_select_theme", "0")
theme = THEMES.get(str(theme_index), THEMES["0"])
self.setProperty("style.card_bg", theme["card_bg"])
self.setProperty("style.card_focus", theme["card_focus"])
self.setProperty("style.card_accent", theme["card_accent"])
self.display_list: xbmcgui.ControlList = self.getControlList(1000)
self.populate_qualities_header()
self.populate_sources_list()
self.set_default_focus(self.display_list, 1000, control_list_reset=True)
super().onInit()
def populate_qualities_header(self):
from collections import Counter
fixed_qualities = [
("[B][COLOR yellow]4k[/COLOR][/B]", "4k"),
("[B][COLOR blue]1080p[/COLOR][/B]", "1080p"),
("[B][COLOR orange]720p[/COLOR][/B]", "720p"),
("[B][COLOR yellow]N/A[/COLOR][/B]", "N/A"),
]
qualities = [s.quality for s in self.sources if s.quality]
counts = Counter(qualities)
qualities_list = self.getControl(1300)
qualities_list.reset()
# Set fixed width for 4 items
self.setProperty("quality_item_width", str(int(800 / 4)))
for display, key in fixed_qualities:
count = counts.get(display, 0)
label = f"{display} ({count})"
list_item = xbmcgui.ListItem(label)
list_item.setProperty("quality", key)
qualities_list.addItem(list_item)
def doModal(self) -> bool:
super().doModal()
return self.resolved
def handle_action(self, action_id: int, control_id: Optional[int] = None) -> None:
if action_id == 1 and control_id == 1000:
self._handle_filter_action()
elif action_id == 117: # Context menu action
self._handle_context_menu_action()
elif control_id == 1300:
self._handle_quality_select_action()
elif action_id == 7 and control_id == 1000: # Select action
self._handle_select_action(control_id)
def _handle_filter_action(self):
filter_type_popup = FilterTypeWindow("filter_type.xml", ADDON_PATH)
filter_type_popup.doModal()
selected_type = filter_type_popup.selected_type
del filter_type_popup
def get_unique(attr):
return sorted(set(getattr(s, attr) for s in self.sources if getattr(s, attr)))
filter_map = {
"quality": {
"items": lambda: get_unique("quality"),
"filter": lambda val: [s for s in self.sources if s.quality == val],
},
"provider": {
"items": lambda: get_unique("provider"),
"filter": lambda val: [s for s in self.sources if s.provider == val],
},
"type": {
"items": lambda: get_unique("type"),
"filter": lambda val: [s for s in self.sources if s.type == val],
},
"indexer": {
"items": lambda: get_unique("indexer"),
"filter": lambda val: [s for s in self.sources if s.indexer == val],
},
"language": {
"items": self._get_all_languages,
"filter": lambda val: [
s
for s in self.sources
if val in getattr(s, "languages", []) or val in getattr(s, "fullLanguages", [])
],
},
}
one_click_filters = {
"only_torrents": lambda: [s for s in self.sources if s.type == "Torrent"],
"only_debrid": lambda: [s for s in self.sources if s.type == "Debrid"],
}
if selected_type in one_click_filters:
filtered_sources = one_click_filters[selected_type]()
if not filtered_sources:
kodilog("No sources found matching the filter")
return
self.filtered_sources = filtered_sources
self.filter_applied = True
elif selected_type in filter_map:
items = filter_map[selected_type]["items"]()
if selected_type == "language" and not items:
notification(translation(90406))
return
popup = FilterWindow("filter_items.xml", ADDON_PATH, filter=items)
popup.doModal()
selected_filter = popup.selected_filter
del popup
if selected_filter is not None:
filtered_sources = filter_map[selected_type]["filter"](selected_filter)
if not filtered_sources:
kodilog("No sources found matching the filter")
return
self.filtered_sources = filtered_sources
self.filter_applied = True
else:
self.filtered_sources = None
self.filter_applied = False
else:
self.filtered_sources = None
self.filter_applied = False
self.populate_sources_list()
self.set_default_focus(self.display_list, 1000, control_list_reset=True)
def _handle_context_menu_action(self):
self.position = self.display_list.getSelectedPosition()
selected_source = self.list_sources[self.position]
menu_items = []
menu_actions = []
if selected_source.type == IndexerType.TORRENT:
menu_items.extend(
[
translation(90365),
translation(90359),
translation(90082),
translation(90744),
]
)
menu_actions.extend(
[
"download_to_debrid",
"add_to_torrserver",
"subtitle_download",
"upload_subtitle",
]
)
elif selected_source.type in (IndexerType.DIRECT, IndexerType.STREMIO_DEBRID):
menu_items.extend([translation(90083), translation(90082), translation(90744)])
menu_actions.extend(["download_file", "subtitle_download", "upload_subtitle"])
else:
menu_items.extend(
[
translation(90084),
translation(90083),
translation(90082),
translation(90744),
]
)
menu_actions.extend(
[
"pack_select",
"download_file",
"subtitle_download",
"upload_subtitle",
]
)
response = xbmcgui.Dialog().contextmenu(menu_items)
if response < 0 or response >= len(menu_actions):
return
action = menu_actions[response]
kodilog(
f"SourceSelect context action selected: action={action}, mode={self.item_information.get('mode')}, query={self.item_information.get('query')}, year={self.item_information.get('year')}"
)
if action == "download_to_debrid":
try:
self._download_to_debrid(selected_source)
except Exception as e:
kodilog(f"SourceSelect _download_to_debrid raised: {e}")
notification(str(e))
elif action == "add_to_torrserver":
self._add_to_torrserver(selected_source)
elif action == "download_file":
self._download_file(selected_source)
elif action == "subtitle_download":
self._resolve_item(selected_source, is_subtitle_download=True)
elif action == "pack_select":
self._resolve_item(selected_source, pack_select=True)
elif action == "upload_subtitle":
self._upload_subtitle(selected_source)
def _handle_quality_select_action(self):
quality_list = self.getControl(1300)
selected_item = quality_list.getSelectedItem()
if selected_item:
selected_quality = selected_item.getProperty("quality")
filtered_sources = [s for s in self.sources if selected_quality in s.quality]
if not filtered_sources:
kodilog("No sources found matching the filter")
return
self.filtered_sources = filtered_sources
self.filter_applied = True
self.populate_sources_list()
def _handle_select_action(self, control_id):
self.position = self.display_list.getSelectedPosition()
selected_source = self.list_sources[self.position]
control_list = self.getControl(control_id)
self.set_cached_focus(control_id, control_list.getSelectedPosition())
self._resolve_item(selected_source, pack_select=False)
def populate_sources_list(self) -> None:
self.display_list.reset()
self.list_sources = (
self.filtered_sources
if self.filter_applied and self.filtered_sources is not None
else self.sources
)
aliases = {}
with contextlib.suppress(Exception):
aliases = cache.get(STREMIO_ADDON_ALIASES_KEY) or {}
for source in self.list_sources:
info = parse_title_info(source.title)
menu_item = xbmcgui.ListItem(label=source.title)
menu_item.setProperty("title", source.title)
menu_item.setProperty("display_title", info["clean_title"])
menu_item.setProperty("codec", info["codec"])
menu_item.setProperty("audio", info["audio"])
menu_item.setProperty("hdr_info", info["badges"])
menu_item.setProperty("release_group", info["release_group"])
kodilog(f"SourceSelect populate_sources_list: source.type={source.type}")
if source.type in IndexerType.TORRENT:
if source.addedToDebrid and source.debridType:
provider_name = source.debridType
else:
provider_name = source.type or source.subindexer
elif source.type == IndexerType.DEBRID:
provider_name = source.debridType or source.type
elif source.type == IndexerType.STREMIO_DEBRID:
if source.debridType:
provider_name = source.debridType
elif source.isCached and not source.infoHash:
provider_name = source.type
else:
provider_name = source.subindexer or source.type
elif source.type == IndexerType.DIRECT:
provider_name = source.indexer or source.type
else:
provider_name = source.debridType or source.type
indexer_label = _get_source_indexer_label(source, aliases)
menu_item.setProperty("type", get_provider_color(provider_name))
menu_item.setProperty("indexer", get_random_color(indexer_label))
menu_item.setProperty("guid", source.guid)
menu_item.setProperty("infoHash", source.infoHash)
menu_item.setProperty("size", bytes_to_human_readable(int(source.size)))
if source.seeders and not source.isCached:
menu_item.setProperty("seeders", str(source.seeders))
if source.peers and not source.isCached:
menu_item.setProperty("peers", str(source.peers))
menu_item.setProperty("fullLanguages", get_colored_languages(source.fullLanguages))
menu_item.setProperty("provider", get_random_color(source.provider))
menu_item.setProperty("publishDate", extract_publish_date(source.publishDate))
menu_item.setProperty("quality", source.quality)
menu_item.setProperty("status", get_source_status(source))
menu_item.setProperty("isPack", str(source.isPack))
self.display_list.addItem(menu_item)
def _refresh_sources_list(self) -> None:
if not hasattr(self, "display_list"):
return
current_position = max(getattr(self, "position", 0), 0)
self.populate_sources_list()
if self.list_sources:
self.set_default_focus(self.display_list, 1000, control_list_reset=True)
self.display_list.selectItem(min(current_position, len(self.list_sources) - 1))
def _download_to_debrid(self, selected_source) -> None:
kodilog("SourceSelect _download_to_debrid entered")
url, magnet, _ = self._extract_source_details(selected_source)
info_hash = selected_source.infoHash or ""
torrent_data = b""
if not info_hash and magnet:
info_hash = get_info_hash_from_magnet(magnet).lower()
if not info_hash and url.startswith("magnet:"):
info_hash = get_info_hash_from_magnet(url).lower()
if url:
kodilog(f"Fetching torrent URL to extract info_hash: {url}")
torrent_data, magnet_candidate, extracted_hash, torrent_url = get_torrent_data_from_uri(
url
)
if magnet_candidate and not magnet:
magnet = magnet_candidate
if extracted_hash and not info_hash:
info_hash = extracted_hash
if torrent_url:
selected_source.url = torrent_url
if not info_hash and not torrent_data:
notification(translation(90361))
return
debrid_type = add_source_to_debrid(
info_hash,
selected_source.debridType,
torrent_data=torrent_data,
torrent_name=selected_source.title or self.item_information.get("title", ""),
)
if debrid_type:
kodilog(f"SourceSelect download_to_debrid succeeded via {debrid_type}")
if info_hash:
selected_source.infoHash = info_hash
selected_source.debridType = debrid_type
selected_source.addedToDebrid = True
self._refresh_sources_list()
def _add_to_torrserver(self, selected_source) -> None:
url, magnet, _ = self._extract_source_details(selected_source)
title = selected_source.title or self.item_information.get("title", "")
poster = self.item_information.get("poster", "")
import json as _json
metadata = {
"title": title,
"mode": self.item_information.get("mode", ""),
"ids": self.item_information.get("ids", {}),
"tv_data": self.item_information.get("tv_data", {}),
}
data = _json.dumps(metadata)
add_source_to_torrserver(
magnet=magnet,
url=url,
info_hash=selected_source.infoHash or "",
title=title,
poster=poster,
data=data,
)
def _download_file(self, selected_source) -> None:
self.playback_info = self._ensure_playback_info(selected_source)
if self.playback_info:
self.playback_info.update(self.item_information)
else:
notification(translation(90407))
return
try:
title = (
self.playback_info.get("title") or self.item_information.get("title") or ""
).strip()
year = self.item_information.get("year", "")
tv_data = self.playback_info.get("tv_data")
if isinstance(tv_data, str):
tv_data = safe_json_loads(tv_data, {})
tv_data = normalize_tv_data(tv_data)
if tv_data and tv_data.get("season") and tv_data.get("episode"):
season = int(tv_data["season"])
episode = int(tv_data["episode"])
file_name = f"{title} - {format_season_episode(season, episode)}"
elif year:
file_name = f"{title} ({year})"
else:
file_name = title
mode = self.playback_info.get("mode", "movie")
dest_data = {
"title": title,
"mode": "movies" if mode in ("movie", "movies") else mode,
}
if tv_data:
dest_data["tv_data"] = tv_data
from lib.downloader import get_destination_path
destination = get_destination_path(dest_data)
if not destination:
notification(translation(90407))
return
xbmc.executebuiltin(
action_url_run(
"handle_download_file",
file_name=file_name,
url=self.playback_info["url"],
destination=destination,
)
)
except Exception as e:
kodilog(f"Failed to start download: {e!s}")
xbmcgui.Dialog().notification(translation(90653), translation(90654) % str(e))
def _upload_subtitle(self, selected_source) -> None:
"""Show upload subtitle dialog with local/web options."""
kodilog("[UPLOAD_SUBTITLE] Starting subtitle upload process")
dialog = xbmcgui.Dialog()
options = [
translation(90746), # Local File
translation(90747), # Upload from Browser
]
choice = dialog.select(translation(90744), options) # Upload Subtitle
if choice == 0: # Local file
self._upload_subtitle_local(selected_source)
elif choice == 1: # Web upload
self._upload_subtitle_from_web(selected_source)
def _upload_subtitle_local(self, selected_source) -> None:
"""Open file browser to select a local subtitle file and resolve the source with it."""
dialog = xbmcgui.Dialog()
subtitle_file = dialog.browse(
1, translation(90745), "files", ".srt|.ass|.ssa|.sub|.vtt|.txt"
)
if not subtitle_file:
kodilog("[UPLOAD_SUBTITLE] User cancelled subtitle upload (no file selected)")
return
self._resolve_item(selected_source, local_subtitle_path=subtitle_file)
def _upload_subtitle_from_web(self, selected_source) -> None:
"""Start web server and show QR dialog for browser-based subtitle upload."""
kodilog("[UPLOAD_SUBTITLE] Starting web upload server")
local_ip = get_local_ip()
server = SubtitleUploadServer(port=8082)
server.start()
if not server.is_running:
kodilog("[UPLOAD_SUBTITLE] Failed to start upload server")
notification(translation(90744))
return
url = f"http://{local_ip}:{server.port}/"
subtitle_result = [None] # Use list for mutability in nested function
dialog = None
upload_complete = threading.Event()
try:
kodilog(f"[UPLOAD_SUBTITLE] Server started at {url}")
# Generate QR code
qr_path = make_qrcode(url)
# Show QR dialog
dialog = QRProgressDialog("qr_dialog.xml", ADDON_PATH)
dialog.setup(
title=translation(90748), # Upload subtitle from your browser
qr_code=qr_path or "",
url=url,
user_code="",
is_debrid=False,
custom_message=translation(90749) % url,
)
# Start polling thread to detect upload completion
def poll_for_upload():
kodilog("[UPLOAD_SUBTITLE] Polling thread started")
import time
poll_count = 0
while not upload_complete.is_set():
time.sleep(0.5)
poll_count += 1
# Check if file was uploaded
file_path = server.get_uploaded_file()
if file_path:
kodilog(f"[UPLOAD_SUBTITLE] File detected via polling: {file_path}")
subtitle_result[0] = file_path
upload_complete.set()
# Close dialog from thread
try:
dialog.close()
except Exception as e:
kodilog(f"[UPLOAD_SUBTITLE] Error closing dialog from thread: {e}")
break
# Timeout after 10 minutes
if poll_count > 1200:
kodilog("[UPLOAD_SUBTITLE] Polling timeout")
upload_complete.set()
with contextlib.suppress(BaseException):
dialog.close()
break
kodilog("[UPLOAD_SUBTITLE] Polling thread ended")
# Start polling thread
poll_thread = threading.Thread(target=poll_for_upload, daemon=True)
poll_thread.start()
# Show modal dialog (blocks until closed)
kodilog("[UPLOAD_SUBTITLE] Showing modal dialog...")
dialog.doModal()
# Signal polling thread to stop
upload_complete.set()
# Check result
if subtitle_result[0]:
kodilog(f"[UPLOAD_SUBTITLE] Upload successful: {subtitle_result[0]}")
elif dialog.iscanceled:
kodilog("[UPLOAD_SUBTITLE] User cancelled")
else:
kodilog("[UPLOAD_SUBTITLE] Dialog closed without upload")
finally:
upload_complete.set()
server.stop()
kodilog("[UPLOAD_SUBTITLE] Web upload server stopped")
# If we got a file, resolve the source with it
if subtitle_result[0]:
self._resolve_item(selected_source, local_subtitle_path=subtitle_result[0])
def _resolve_item(
self,
selected_source,
pack_select: bool = False,
is_subtitle_download=False,
local_subtitle_path: Optional[str] = None,
) -> None:
self.setProperty("resolving", "true")
resolver_window = ResolverWindow(
"resolver.xml",
ADDON_PATH,
source=selected_source,
previous_window=self,
item_information=self.item_information,
is_subtitle_download=is_subtitle_download,
local_subtitle_path=local_subtitle_path,
)
self.resolved = resolver_window.doModal(pack_select)
del resolver_window
def show_resume_dialog(self, playback_percent: float) -> Optional[bool]:
try:
resume_window = ResumeDialog(
"resume_dialog.xml",
ADDON_PATH,
resume_percent=playback_percent,
)
resume_window.doModal()
return resume_window.resume
finally:
del resume_window
def _get_all_languages(self):
all_languages = set()
for s in self.sources:
all_languages.update(getattr(s, "languages", []))
all_languages.update(getattr(s, "fullLanguages", []))
return sorted(all_languages)