import threading
from concurrent.futures import Future, ThreadPoolExecutor
from dataclasses import dataclass
from typing import Callable, List, Optional
import xbmc
from lib.domain.torrent import TorrentStream
from lib.gui.base_window import BaseWindow
from lib.utils.kodi.utils import kodilog, translation
STATUS_MAP = {
"Pending": 90244,
"In Progress": 90245,
"Completed": 90246,
"Failed": 90247,
"Cancelled": 90248,
}
@dataclass
class SearchTask:
name: str
indexer_key: str
status: str = "Pending" # "Pending", "In Progress", "Completed", "Failed", "Cancelled"
result_count: int = 0
error: str = ""
future: Optional[Future] = None
class SearchTaskManager:
def __init__(self, executor: ThreadPoolExecutor):
self.executor = executor
self.tasks: List[SearchTask] = []
self._cancel_event = threading.Event()
self._lock = threading.RLock()
def submit_task(self, name: str, indexer_key: str, fn: Callable, *args, **kwargs) -> SearchTask:
task = SearchTask(name=name, indexer_key=indexer_key)
with self._lock:
self.tasks.append(task)
def wrapped_fn(*a, **kw):
if self._cancel_event.is_set():
self._update_task(task, status="Cancelled")
return []
self._update_task(task, status="In Progress")
try:
# ARTIFICIAL DELAY FOR TESTING GUI UPDATES
# xbmc.sleep(8000)
results = fn(*a, **kw)
if self._cancel_event.is_set():
self._update_task(task, status="Cancelled")
return results if results else []
self._update_task(
task,
status="Completed",
result_count=len(results) if results else 0,
)
return results
except Exception as e:
self._update_task(task, status="Failed", error=str(e))
kodilog(f"Task {name} failed: {e}")
return []
future = self.executor.submit(wrapped_fn, *args, **kwargs)
task.future = future
return task
def cancel_pending(self):
self._cancel_event.set()
with self._lock:
for task in self.tasks:
if task.status in ["Pending", "In Progress"]:
if not task.future or getattr(task.future, "cancel", lambda: False)():
task.status = "Cancelled"
@property
def is_cancelled(self) -> bool:
return self._cancel_event.is_set()
@property
def is_complete(self) -> bool:
with self._lock:
return all(t.status in ["Completed", "Failed", "Cancelled"] for t in self.tasks)
def _update_task(
self,
task: SearchTask,
status: Optional[str] = None,
result_count: Optional[int] = None,
error: Optional[str] = None,
):
with self._lock:
if status is not None:
task.status = status
if result_count is not None:
task.result_count = result_count
if error is not None:
task.error = error
def snapshot_tasks(self) -> List[SearchTask]:
with self._lock:
return [
SearchTask(
name=task.name,
indexer_key=task.indexer_key,
status=task.status,
result_count=task.result_count,
error=task.error,
)
for task in self.tasks
]
def collect_results(self) -> List[TorrentStream]:
total_results = []
with self._lock:
tasks = list(self.tasks)
for task in tasks:
if task.future and task.future.done():
try:
res = task.future.result()
if res:
total_results.extend(res)
except Exception as e:
kodilog(f"Error collecting result for {task.name}: {e}")
return total_results
class SearchStatusWindow(BaseWindow):
MAX_TASK_SLOTS = 10
def __init__(
self,
xml_file: str,
location: str,
task_manager: SearchTaskManager,
item_information=None,
):
super().__init__(xml_file, location, item_information=item_information)
self.task_manager = task_manager
self._monitor = xbmc.Monitor()
def onInit(self):
self.getControl(13001).setLabel(translation(90241))
self.setProperty("instant_close", "false")
# Set initial task names so they appear immediately
self._update_ui_state()
def _get_status_color(self, status: str) -> str:
if status == "Pending":
return "FF888888" # Grey
elif status == "In Progress":
return "FFFFD700" # Yellow
elif status == "Completed":
return "FF00FF00" # Green
elif status == "Failed":
return "FFFF0000" # Red
elif status == "Cancelled":
return "FFFFA500" # Orange
return "FFFFFFFF" # White
def run_until_complete(self, poll_interval_ms=250, final_delay_ms=500):
while not self._monitor.abortRequested() and not self.task_manager.is_complete:
xbmc.sleep(poll_interval_ms)
if self.task_manager.is_cancelled:
break
self._update_ui_state()
# Final update
self._update_ui_state()
# Give user a brief moment to see final state if not cancelled
if not self.task_manager.is_cancelled:
xbmc.sleep(final_delay_ms)
self.close()
def _update_ui_state(self):
tasks = self.task_manager.snapshot_tasks()
total_tasks = len(tasks)
completed_tasks = 0
total_results = 0
for i, task in enumerate(tasks):
if i >= self.MAX_TASK_SLOTS:
break
status_text = translation(STATUS_MAP.get(task.status, 90244))
self.setProperty(f"task_{i}_name", task.name)
self.setProperty(f"task_{i}_status", status_text)
self.setProperty(f"task_{i}_color", self._get_status_color(task.status))
total_results += task.result_count
if task.status == "Completed":
completed_tasks += 1
self.setProperty(f"task_{i}_results", str(task.result_count))
self.setProperty(f"task_{i}_show_spinner", "")
elif task.status in ["Failed", "Cancelled"]:
completed_tasks += 1
self.setProperty(f"task_{i}_results", "")
self.setProperty(f"task_{i}_show_spinner", "")
elif task.status == "In Progress":
self.setProperty(
f"task_{i}_results",
str(task.result_count) if task.result_count > 0 else "",
)
show_spinner = "true" if task.result_count == 0 else ""
self.setProperty(f"task_{i}_show_spinner", show_spinner)
else:
self.setProperty(f"task_{i}_results", "")
self.setProperty(f"task_{i}_show_spinner", "")
# Update progress
if total_tasks > 0:
percent = int((completed_tasks / total_tasks) * 100)
self.getControl(13002).setPercent(percent)
self.setProperty("search_results_count", f"{translation(90243)}{total_results}")
def onAction(self, action):
action_id = action.getId()
if action_id in self.action_exitkeys_id:
self.task_manager.cancel_pending()
super().onAction(action)
def handle_action(self, action_id, control_id=None):
if action_id == 7 and control_id == 13004: # Cancel button clicked
self.task_manager.cancel_pending()
self.close()