""" Filename: plugin_manager.py Description: Manager file for anything plugin related Note: Compliant with pylance strict type checking Author: Tyler de Zeeuw License: GPL-3.0 """ # Built-in imports import io import ssl import sys import json import shutil import zipfile import platform import urllib.request import importlib.util from pathlib import Path from typing import Any, cast # External library imports import certifi from PySide6.QtCore import QObject, QUrl, Signal from PySide6.QtGui import QDesktopServices from PySide6.QtWidgets import QMainWindow, QMenu, QMessageBox from src.shared.shareddata import APP_NAME, CURRENT_VERSION, PLATFORM_NAME def get_ssl_context() -> ssl.SSLContext: """Returns a cross-platform SSL context configured with certifi's CA bundle.""" return ssl.create_default_context(cafile=certifi.where()) def get_current_platform_id() -> str: """Returns standardized platform identifier (win_x64, darwin_arm64, etc.).""" sys_name = sys.platform arch = platform.machine().lower() if sys_name == "win32": return "win_x64" if "64" in arch or "amd64" in arch else "win_x86" elif sys_name == "darwin": return "darwin_arm64" if "arm" in arch or "aarch64" in arch else "darwin_x64" elif sys_name.startswith("linux"): return "linux_x64" if "64" in arch else "linux_x86" return sys_name def parse_version(v_str: str) -> tuple[int, ...]: """Parses a version string into a comparable integer tuple.""" try: return tuple(int(x) for x in v_str.strip().lstrip("v").split(".")) except ValueError: return (0, 0, 0) class PluginManager(QObject): """ Handles plugin discovery, dynamic loading, menu building, and plugin lifecycle (install, toggle, uninstall). """ plugins_changed = Signal() def __init__(self, main_window: QMainWindow) -> None: super().__init__() self.main_window: QMainWindow = main_window self.plugins_dir: Path = self._resolve_plugins_dir() self.loaded_plugins: list[Any] = [] self.current_platform: str = get_current_platform_id() self.current_app_version: tuple[int, ...] = parse_version(str(CURRENT_VERSION)) def _resolve_plugins_dir(self) -> Path: """Determines the local plugins directory based on execution context.""" if PLATFORM_NAME == "darwin": base_dir = Path(sys.executable).parent / "../../.." else: base_dir = Path.cwd() plugins_path = (base_dir / "plugins").resolve() plugins_path.mkdir(parents=True, exist_ok=True) return plugins_path def reload_plugins(self) -> list[Any]: """Scans directory, imports enabled plugins, and updates the main application menu.""" self.unload_plugins() plugins_dir_str = str(self.plugins_dir) if plugins_dir_str not in sys.path: sys.path.insert(0, plugins_dir_str) failed_plugins: list[tuple[str, str]] = [] for entry in sorted(self.plugins_dir.iterdir(), key=lambda p: p.name.lower()): if entry.name.startswith((".", "__")) or entry.name.endswith(".disabled"): continue module_name = "" init_file: Path | None = None if entry.is_dir(): init_file = entry / "__init__.py" module_name = entry.name elif entry.is_file() and entry.suffix == ".py": init_file = entry module_name = entry.stem if not init_file or not init_file.exists(): continue try: instance = self._import_and_instantiate(module_name, init_file) if instance is not None: self.loaded_plugins.append(instance) except Exception as e: failed_plugins.append((module_name, str(e))) # Report failures after all plugins have attempted to load if failed_plugins: error_details = "\n".join(f"• {name}: {err}" for name, err in failed_plugins) summary_msg = f"One or more plugins failed to load:\n\n{error_details}" print(f"[PluginManager] {summary_msg}") parent_widget = getattr(self, "main_window", None) QMessageBox.warning( parent_widget, "Plugin Load Failures", summary_msg ) self.build_plugins_menu() self.plugins_changed.emit() return self.loaded_plugins def _import_and_instantiate(self, module_name: str, file_path: Path) -> Any | None: """Dynamically imports a plugin module, attaches its metadata, and returns the Plugin instance.""" spec = importlib.util.spec_from_file_location(module_name, file_path) if spec is None or spec.loader is None: raise ImportError(f"Invalid or missing spec for file: {file_path}") module = importlib.util.module_from_spec(spec) sys.modules[module_name] = module spec.loader.exec_module(module) if hasattr(module, "Plugin") and isinstance(module.Plugin, type): instance = module.Plugin(self.main_window) # Attach manifest metadata directly to the plugin instance if plugin.json exists manifest_path = file_path.parent / "plugin.json" if manifest_path.exists(): try: with open(manifest_path, "r", encoding="utf-8") as f: instance.metadata = json.load(f) # Override class name property if specified in manifest if "name" in instance.metadata: instance.name = instance.metadata["name"] except Exception as e: print(f"[PluginManager] Could not attach metadata to '{module_name}': {e}") return instance return None def build_plugins_menu(self) -> None: """Constructs or refreshes submenus under the top-level 'Plugins' menu item.""" menubar = self.main_window.menuBar() # Dynamically locate the existing 'Plugins' menu on the menubar plugins_menu: QMenu | None = None for action in menubar.actions(): clean_text = action.text().replace("&", "").strip().lower() if clean_text == "plugins": menu = action.menu() if isinstance(menu, QMenu): plugins_menu = menu break # If it doesn't exist yet, create it if plugins_menu is None: plugins_menu = menubar.addMenu("&Plugins") plugins_menu.clear() # 1. Populate loaded plugin submenus if not self.loaded_plugins: disabled_action = plugins_menu.addAction("No active plugins") disabled_action.setEnabled(False) else: for plugin in self.loaded_plugins: plugin_name = getattr(plugin, "name", "Unnamed Plugin") plugin_submenu = plugins_menu.addMenu(plugin_name) if hasattr(plugin, "register_menu") and callable(plugin.register_menu): plugin.register_menu(plugin_submenu) # 2. Append entry to open Plugin Manager GUI plugins_menu.addSeparator() manager_action = plugins_menu.addAction("Manage Plugins...") plugins_gui_func = getattr(self.main_window, "plugins_gui", None) if callable(plugins_gui_func): manager_action.triggered.connect(plugins_gui_func) def unload_plugins(self) -> None: """Clears current active plugin instances.""" self.loaded_plugins.clear() def get_installed_plugins_info(self) -> list[dict[str, Any]]: """Returns metadata for all local plugins in the plugins folder by reading their plugin.json.""" results: list[dict[str, Any]] = [] for entry in sorted(self.plugins_dir.iterdir(), key=lambda p: p.name.lower()): if entry.name.startswith((".", "__")): continue is_disabled = entry.name.endswith(".disabled") clean_name = entry.name[:-9] if is_disabled else entry.name if entry.is_dir() or (entry.is_file() and (entry.suffix == ".py" or entry.name.endswith(".py.disabled"))): if entry.is_file() and entry.stem in ("__init__", "__init__.py"): continue # Load local plugin.json if present manifest: dict[str, Any] = {} manifest_path = (entry / "plugin.json") if entry.is_dir() else None if manifest_path and manifest_path.exists(): try: with open(manifest_path, "r", encoding="utf-8") as f: manifest = json.load(f) except Exception as e: print(f"[PluginManager] Warning: Failed to read manifest for '{entry.name}': {e}") display_name = manifest.get("name", clean_name.removesuffix(".py")) version = manifest.get("version", "1.0.0") description = manifest.get("description", "") author = manifest.get("author", "Unknown") results.append({ "id": manifest.get("id", clean_name), "name": display_name, "version": version, "description": description, "author": author, "path": entry, "is_disabled": is_disabled, "manifest": manifest, }) return results def toggle_plugin_state(self, plugin_path: Path) -> None: """Swaps a plugin between enabled and disabled by renaming with/without '.disabled'.""" if not plugin_path.exists(): return if plugin_path.name.endswith(".disabled"): new_path = plugin_path.with_name(plugin_path.name.removesuffix(".disabled")) else: new_path = plugin_path.with_name(f"{plugin_path.name}.disabled") plugin_path.rename(new_path) self.reload_plugins() def uninstall_plugin(self, plugin_path: Path) -> None: """Deletes a plugin directory or file from disk.""" if not plugin_path.exists(): return if plugin_path.is_dir(): shutil.rmtree(plugin_path) else: plugin_path.unlink() self.reload_plugins() def fetch_remote_repositories(self, repo_urls: list[str]) -> tuple[list[dict[str, Any]], bool]: """Fetches remote plugin metadata from repository URLs.""" aggregated: list[dict[str, Any]] = [] failed_count = 0 for url in repo_urls: try: req = urllib.request.Request( url, headers={"User-Agent": f"{APP_NAME}-PluginManager"}, ) with urllib.request.urlopen(req, timeout=5, context=get_ssl_context()) as response: if response.status == 200: raw_data = response.read().decode("utf-8") data = json.loads(raw_data) if isinstance(data, list): aggregated.extend(cast(list[Any], data)) else: failed_count += 1 except Exception: failed_count += 1 all_failed = failed_count == len(repo_urls) and len(repo_urls) > 0 return aggregated, all_failed def install_plugin_from_url(self, download_url: str, plugin_id: str = "") -> Path: """ Downloads a single-folder ZIP archive and extracts it directly into plugins/. """ req = urllib.request.Request( download_url, headers={"User-Agent": f"{APP_NAME}-PluginManager"}, ) with urllib.request.urlopen(req, timeout=15, context=get_ssl_context()) as response: if response.status != 200: raise RuntimeError(f"Download failed with HTTP status code {response.status}") zip_bytes = response.read() with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zip_ref: zip_ref.extractall(self.plugins_dir) self.reload_plugins() return self.plugins_dir def open_plugins_directory(self) -> bool: """Opens the local plugins folder in the operating system's file manager.""" if self.plugins_dir.exists(): return QDesktopServices.openUrl(QUrl.fromLocalFile(str(self.plugins_dir))) return False