""" Filename: terminal.py Description: Terminal window Note: Compliant with pylance strict type checking Author: Tyler de Zeeuw License: GPL-3.0 """ # Built-in imports import sys from pathlib import Path from typing import Any, Callable # External library imports from PySide6.QtWidgets import QWidget, QVBoxLayout, QTextEdit, QLineEdit, QMainWindow from PySide6.QtCore import QProcess, Qt, QThread, Signal from file_ext_registration import register_file_association, is_windows_admin from plugin_manager import PluginManager from src.shared.shareddata import API_URL, API_URL_SECONDARY, APP_NAME, CURRENT_VERSION, PLATFORM_NAME from src.window.about import AboutWindow from src.window.plugins import PluginsWindow from updater import UpdateManager class _AssocWorker(QThread): """ Runs register_file_association() off the UI thread. Only actually needed for the force_admin=True path, since that blocks on WaitForSingleObject while the UAC prompt is up and the elevated child process runs — which would otherwise freeze the terminal window. Used for the non-elevated path too for consistency. """ result_ready = Signal(bool, str) def __init__(self, force_admin: bool, parent: QWidget | None = None) -> None: super().__init__(parent) self._force_admin = force_admin def run(self) -> None: ok, msg = register_file_association(force_admin=self._force_admin) self.result_ready.emit(ok, msg) class TerminalWindow(QWidget): def __init__(self, parent: QWidget | None, plugin_manager: PluginManager) -> None: super().__init__(parent, Qt.WindowType.Window) self.setWindowTitle(f"Terminal - {APP_NAME.upper()}") self.resize(320, 180) self.output_area = QTextEdit() self.output_area.setReadOnly(True) self.input_line = QLineEdit() self.input_line.returnPressed.connect(self.handle_command) layout = QVBoxLayout() layout.addWidget(self.output_area) layout.addWidget(self.input_line) self.setLayout(layout) self._process: QProcess | None = None self.plugin_manager = plugin_manager self.commands: dict[str, Callable[..., Any]] = { "hello": self.cmd_hello, "help": self.cmd_help, "version": self.cmd_version, "about": self.cmd_about, "assoc": self.cmd_assoc, "update": self.cmd_update, "plugins": self.cmd_plugins, "utest": self.cmd_utest, } self._pending_assoc_confirmation: bool = False self._assoc_worker: _AssocWorker | None = None self.output_area.append(f"Welcome to {APP_NAME.upper()}. You are running version {CURRENT_VERSION}.") self.output_area.append("Type 'help' for a list of available commands.\n") self.input_line.setFocus() def handle_command(self): command_text = self.input_line.text() self.input_line.clear() self.output_area.append(f"> {command_text}") if self._pending_assoc_confirmation: self._pending_assoc_confirmation = False self._handle_assoc_confirmation(command_text.strip().lower()) return parts = command_text.strip().split() if not parts: return command_name = parts[0] args = parts[1:] func = self.commands.get(command_name) if func: try: result = func(*args) if result: self.output_area.append(str(result)) except Exception as e: self.output_area.append(f"[Error] {e}") else: self.output_area.append(f"[Unknown command] '{command_name}'") def cmd_hello(self, *args: Any) -> str: return "Hello from the terminal!" def cmd_help(self, *args: Any) -> str: available_cmds = [cmd for cmd in self.commands.keys() if cmd != "utest"] return f"Available commands: {', '.join(available_cmds)}" def cmd_version(self, *args: Any) -> str: return f"{APP_NAME.upper()} is running version {CURRENT_VERSION}." def cmd_about(self, *args: Any) -> None: self.about = AboutWindow(self) self.about.show() def cmd_plugins(self, *args: Any) -> None: self.about = PluginsWindow(self, self.plugin_manager) self.about.show() def cmd_update(self, *args: Any) -> str: main_win = self.parent() if not isinstance(main_win, QMainWindow): return "[Error] Main window context not found." self.updater = UpdateManager( main_window=main_win, api_url=API_URL, api_url_sec=API_URL_SECONDARY, current_version=CURRENT_VERSION, platform_name=PLATFORM_NAME, platform_suffix="-" + PLATFORM_NAME, app_name=APP_NAME ) self.output_area.append("Checking for updates...") self.updater.manual_check_for_updates() return "See status bar for update information." def cmd_assoc(self, *args: Any) -> str | None: # Non-Windows platforms don't have the admin/non-admin split — # just register directly. if PLATFORM_NAME != "windows" or is_windows_admin(): self._run_assoc(force_admin=False) return None self._pending_assoc_confirmation = True return "Not running as admin. Register system-wide via UAC elevation? (y/n)" def _handle_assoc_confirmation(self, answer: str) -> None: if answer in ("y", "yes"): self._run_assoc(force_admin=True) elif answer in ("n", "no"): self._run_assoc(force_admin=False) else: self.output_area.append("Please answer 'y' or 'n'. Run 'assoc' again to retry.") def _run_assoc(self, force_admin: bool) -> None: if force_admin: self.output_area.append("Requesting elevation. Check for a UAC prompt...") self._assoc_worker = _AssocWorker(force_admin=force_admin, parent=self) self._assoc_worker.result_ready.connect(self._on_assoc_result) self._assoc_worker.start() def _on_assoc_result(self, ok: bool, msg: str) -> None: self.output_area.append(msg) self._assoc_worker = None def cmd_utest(self, *args: Any) -> str | None: """Executes a specific pre-defined python script non-blockingly.""" if self._process and self._process.state() != QProcess.ProcessState.NotRunning: return "[Error] A process is already running." target_script = Path("main_unit_tests.py") if not target_script.exists(): return f"[Error] Target script not found at: {target_script}" self._process = QProcess(self) # Stream stdout and stderr live to output_area self._process.readyReadStandardOutput.connect(self._handle_stdout) self._process.readyReadStandardError.connect(self._handle_stderr) self._process.finished.connect(self._handle_process_finished) # Use current Python interpreter executable. Works when packaged? python_executable = sys.executable self.output_area.append(f"Starting {target_script.name}...") self._process.start(python_executable, [str(target_script)]) return None def _handle_stdout(self) -> None: if self._process: raw_bytes = bytes(self._process.readAllStandardOutput().data()) data = raw_bytes.decode("utf-8") if data.strip(): self.output_area.append(data.strip()) def _handle_stderr(self) -> None: if self._process: raw_bytes = bytes(self._process.readAllStandardError().data()) data = raw_bytes.decode("utf-8") if data.strip(): self.output_area.append(f"[Error] {data.strip()}") def _handle_process_finished(self, exit_code: int, exit_status: QProcess.ExitStatus) -> None: self.output_area.append(f"Process finished with code {exit_code}.") self._process = None