"""插件注册中心。 架构说明： - PluginWorker 运行在独立 QThread 中，仅做纯 I/O（扫描 .pyd、importlib 读元数据、 文件复制），不创建 QObject、不操作 QWidget。 - PluginRegistry 留在主线程，负责 create_plugin()（返回 QWidget，必须在 GUI 线程 创建）与信号转发。 - 跨线程通信全部走 Qt 信号槽（自动 Queued），不再使用 threading.Thread。 """ import os import sys import time import logging import importlib from pathlib import Path from enum import Enum, auto from typing import Optional from PySide6.QtCore import QObject, Signal, Slot, QThread, QTimer LOCAL_PLUGINS_PATH = Path("./plugins") REMOTE_PLUGINS_PATH = Path('Y:/SE/xufeifei/plugins') logger = logging.getLogger(__name__) class Event(Enum): Update = auto() Install = auto() Uninstall = auto() class PluginWorker(QObject): """插件发现与安装 worker，运行在独立 QThread 中。 仅做纯 I/O，结果以纯数据（list[dict]）形式经信号回传主线程。 """ local_discovered = Signal(list) remote_discovered = Signal(list) install_progress = Signal(int, int) install_finished = Signal(bool, str) @Slot(str) def do_discover_local(self, plugins_dir: str) -> None: plugins_dir = Path(plugins_dir) if not self._ensure_dir(plugins_dir): return self.local_discovered.emit(self._scan_plugins(plugins_dir)) @Slot(str) def do_discover_remote(self, plugins_dir: str) -> None: if not os.path.exists(plugins_dir): logger.error("服务器链接错误！") return self.remote_discovered.emit( self._scan_plugins(Path(plugins_dir), with_remote_meta=True) ) @Slot(str, str) def do_install(self, local_path: str, remote_path: str) -> None: try: total_size = os.path.getsize(remote_path) copied_size = 0 with open(remote_path, 'rb') as fsrc, open(local_path, 'wb') as fdst: while True: buf = fsrc.read(1024 * 1024) if not buf: break fdst.write(buf) copied_size += len(buf) self.install_progress.emit(copied_size, total_size) time.sleep(0.1) self.install_finished.emit(True, f"{local_path}安装成功!!!") except Exception as e: logger.error(f"{remote_path}安装失败{e.args}") self.install_finished.emit(False, str(e)) @staticmethod def _ensure_dir(plugins_dir: Path) -> bool: if not os.path.exists(plugins_dir): os.mkdir(plugins_dir) return False return True @staticmethod def _scan_plugins(plugins_dir: Path, with_remote_meta: bool = False) -> list: results: list = [] sys.path.append(str(plugins_dir)) try: for tool in plugins_dir.glob("*.pyd"): tool_name = tool.stem try: module = importlib.import_module(tool_name) item = { "tool_name": tool_name, "name": module.read_plugin_name(), "version": module.read_plugin_version(), } if with_remote_meta: item["description"] = module.read_plugin_description() item["remote_path"] = str(plugins_dir) results.append(item) except Exception as e: logger.error(f"插件{tool_name}加载失败：{e}") finally: sys.path.remove(str(plugins_dir)) return results class PluginRegistry(QObject): """插件注册中心。 主线程负责创建插件实例（create_plugin 返回 QWidget，必须在 GUI 线程创建）； 发现与安装的 I/O 由 PluginWorker 在独立 QThread 中执行，结果通过信号回传 主线程。所有跨线程通信经 Qt 信号槽（Queued）。 """ plugins_loader_signal = Signal(dict) update_plugins_card_signal = Signal(dict) _discover_local_requested = Signal(str) _discover_remote_requested = Signal(str) _install_requested = Signal(str, str) def __init__(self, parent: Optional[QObject] = None) -> None: super().__init__(parent) self.plugins: dict = {} self._worker = PluginWorker() self._thread = QThread() self._worker.moveToThread(self._thread) self._discover_local_requested.connect(self._worker.do_discover_local) self._discover_remote_requested.connect(self._worker.do_discover_remote) self._install_requested.connect(self._worker.do_install) self._worker.local_discovered.connect(self._on_local_discovered) self._worker.remote_discovered.connect(self._on_remote_discovered) self._worker.install_finished.connect(self._on_install_finished) self._thread.start() # 延迟到事件循环启动后触发，保证所有接收方先 connect 后 emit QTimer.singleShot(0, self._start_discovery) def _start_discovery(self) -> None: self._discover_local_requested.emit(str(LOCAL_PLUGINS_PATH)) self._discover_remote_requested.emit(str(REMOTE_PLUGINS_PATH)) @Slot(list) def _on_local_discovered(self, results: list) -> None: for item in results: name = item["name"] if name not in self.plugins: self.plugins[name] = {} module = importlib.import_module(item["tool_name"]) self.plugins[name]["obj"] = module.create_plugin() self.plugins[name]["local version"] = item["version"] self.plugins_loader_signal.emit(self.plugins) @Slot(list) def _on_remote_discovered(self, results: list) -> None: for item in results: name = item["name"] if name not in self.plugins: self.plugins[name] = {} self.plugins[name]["local version"] = None self.plugins[name]["local description"] = None self.plugins[name]["tool_name"] = item["tool_name"] self.plugins[name]["remote version"] = item["version"] self.plugins[name]["remote description"] = item["description"] self.plugins[name]["remote path"] = item["remote_path"] self.update_plugins_card_signal.emit(self.plugins) @Slot(bool, str) def _on_install_finished(self, success: bool, message: str) -> None: if success: logger.info(message) self._start_discovery() # self._discover_local_requested.emit(str(LOCAL_PLUGINS_PATH)) else: logger.error(message) def start_plugins_event(self, event: Event, name: str) -> None: if event == Event.Install: tool_name = self.plugins[name]['tool_name'] local_path = str(LOCAL_PLUGINS_PATH) + '\\' + f"{tool_name}.pyd" remote_path = str(REMOTE_PLUGINS_PATH) + '\\' + f"{tool_name}.pyd" self._install_requested.emit(local_path, remote_path) elif event == Event.Uninstall: tool_name = self.plugins[name]['tool_name'] local_path = str(LOCAL_PLUGINS_PATH) + '\\' + f"{tool_name}.pyd" # os. pass