331 lines
12 KiB
Python
331 lines
12 KiB
Python
"""
|
|
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 |