"""插件注册中心。 架构说明： - PluginWorker 运行在独立 QThread 中，仅做纯 I/O（扫描 .pyd、importlib 读元数据、 文件复制），不创建 QObject、不操作 QWidget。 - PluginRegistry 留在主线程，负责 create_plugin()（返回 QWidget，必须在 GUI 线程 创建）与信号转发。 - 跨线程通信全部走 Qt 信号槽（自动 Queued），不再使用 threading.Thread。 """ import os import sys import time import shutil import logging import importlib import importlib.util from pathlib import Path from enum import Enum, auto from typing import Optional from PySide6.QtCore import QObject, Signal, Slot, QThread, QTimer WORKSPACE_PATH = Path("./workspace") LOCAL_PLUGINS_PATH = Path("./plugins") REMOTE_PLUGINS_PATH = Path('Y:/SE/xufeifei/plugins') UNINSTALL_PENDING_FILE = LOCAL_PLUGINS_PATH / ".uninstall_pending" logger = logging.getLogger(__name__) def _remove_with_retry(target: Path, retries: int = 5, interval: float = 0.2) -> bool: """删除文件，失败时短暂重试，应对自重启场景下原进程未完全释放句柄的情况。""" for _ in range(retries): try: target.unlink() return True except PermissionError: time.sleep(interval) except FileNotFoundError: return True return False def _load_module_from_path(module_name: str, file_path: Path): """按文件路径加载模块，绕过 sys.modules 缓存。 用 spec_from_file_location 直接从 .pyd 文件构建模块，避免 importlib.import_module 命中同名旧缓存。注意：.pyd 的入口函数 PyInit_ 与文件名绑定，module_name 必须等于文件 stem， 不能用别名。 """ spec = importlib.util.spec_from_file_location(module_name, file_path) if spec is None or spec.loader is None: raise ImportError(f"无法为 {file_path} 创建模块规格") module = importlib.util.module_from_spec(spec) sys.modules[module_name] = module spec.loader.exec_module(module) return module 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) uninstall_finished = Signal(str, bool, str) @Slot(str) def do_discover_local(self, plugins_dir: str) -> None: print(plugins_dir) 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: print(plugins_dir) if not os.path.exists(plugins_dir): logger.error(f"服务器链接错误！路径不存在: {plugins_dir}") return logger.info(f"开始扫描远程插件: {plugins_dir}") 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)) @Slot(str, str) def do_uninstall(self, name: str, local_path: str) -> None: """运行时无法删除已加载的 .pyd（Windows 文件锁），仅记入待删清单。 真正的文件删除由 PluginRegistry._process_pending_uninstalls 在下次启动、 import 之前完成。运行时清理（字典/UI/widget）由主线程在 uninstall_finished 回调中处理。 """ try: tool_name = Path(local_path).stem UNINSTALL_PENDING_FILE.parent.mkdir(parents=True, exist_ok=True) with open(UNINSTALL_PENDING_FILE, 'a', encoding='utf-8') as f: f.write(f"{tool_name}\n") self.uninstall_finished.emit(name, True, f"{name}已加入卸载清单，重启后生效!!!") except Exception as e: logger.error(f"{local_path}卸载失败{e.args}") self.uninstall_finished.emit(name, False, str(e)) @Slot(str, str, str) def do_update(self, name: str, local_path: str, remote_path: str) -> None: """更新=删旧 .pyd + 从服务器复制新 .pyd。 运行时 .pyd 已加载无法删除，记入待处理清单（含 remote_path）， 重启时由 _process_pending_uninstalls 删除旧文件并复制新文件。 运行时 UI 清理复用 uninstall_finished 信号链路，触发与卸载一致的自重启。 """ try: tool_name = Path(local_path).stem UNINSTALL_PENDING_FILE.parent.mkdir(parents=True, exist_ok=True) with open(UNINSTALL_PENDING_FILE, 'a', encoding='utf-8') as f: f.write(f"{tool_name}|{remote_path}\n") self.uninstall_finished.emit(name, True, f"{name}更新已加入清单，重启后生效!!!") except Exception as e: logger.error(f"{local_path}更新失败{e.args}") self.uninstall_finished.emit(name, 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: """扫描目录下的 .pyd 插件，按文件路径加载读取元数据。 .pyd 是 C 扩展，入口函数 PyInit_<文件名> 与文件名绑定，无法用别名 加载。因此本地与远程都用原名加载，但远程扫描前清掉 sys.modules 缓存（避免读到本地旧版本），读完后再次清缓存（避免污染本地逻辑）。 本地扫描用原名加载并保留缓存，供 _on_local_discovered 二次 import。 """ results: list = [] for tool in plugins_dir.glob("*.pyd"): tool_name = tool.stem try: if with_remote_meta: sys.modules.pop(tool_name, None) module = _load_module_from_path(tool_name, tool) 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) sys.modules.pop(tool_name, None) results.append(item) except Exception as e: logger.error(f"插件{tool_name}加载失败：{e}") 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) uninstall_completed_signal = Signal(str) _discover_local_requested = Signal(str) _discover_remote_requested = Signal(str) _install_requested = Signal(str, str) _uninstall_requested = Signal(str, str) _update_requested = Signal(str, str, str) def __init__(self, parent: Optional[QObject] = None) -> None: super().__init__(parent) self.plugins: dict = {} # 启动时先清理上次遗留的待删清单（此时相关 .pyd 尚未 import， # 且 worker 线程未启动，无并发文件访问） self._process_pending_uninstalls() 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._uninstall_requested.connect(self._worker.do_uninstall) self._update_requested.connect(self._worker.do_update) 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._worker.uninstall_finished.connect(self._on_uninstall_finished) self._thread.finished.connect(self._worker.deleteLater) self._thread.start() # 延迟到事件循环启动后触发，保证所有接收方先 connect 后 emit QTimer.singleShot(0, self._start_discovery) def shutdown(self) -> None: """优雅停止工作线程：请求事件循环退出并等待真正结束。 必须在 QThread 析构前调用，否则触发 "QThread: Destroyed while thread is still running" 警告。 """ self._thread.quit() self._thread.wait(3000) @staticmethod def _process_pending_uninstalls() -> None: """启动时执行待处理清单：删除 .pyd 文件，更新项另从服务器复制新文件。 清单行格式： - ``tool_name`` 纯卸载，仅删除本地 .pyd - ``tool_name|remote_path`` 更新，删除后从 remote_path 复制新文件 必须在任何 importlib.import_module 之前调用，此时 .pyd 未被加载， Windows 文件锁不会触发 WinError 5。但自重启场景下，原进程可能尚未 完全释放文件句柄，因此删除失败时短暂重试。失败项保留原始清单行待下次处理。 """ if not UNINSTALL_PENDING_FILE.exists(): return lines = UNINSTALL_PENDING_FILE.read_text(encoding='utf-8').splitlines() failed: list = [] for line in lines: line = line.strip() if not line: continue parts = line.split('|', 1) tool_name = parts[0] target = LOCAL_PLUGINS_PATH / f"{tool_name}.pyd" # 纯卸载且文件已不存在视为已完成；更新项即使文件不存在也要继续复制 if not target.exists() and len(parts) == 1: continue if not _remove_with_retry(target): failed.append(line) logger.error(f"删除插件文件失败（重试后仍被占用）: {target}") continue logger.info(f"已删除插件文件: {target}") if len(parts) == 2: remote_path = Path(parts[1]) try: shutil.copyfile(remote_path, target) logger.info(f"已更新插件文件: {target}") except Exception as e: failed.append(line) logger.error(f"复制更新文件失败: {e}") if failed: UNINSTALL_PENDING_FILE.write_text("\n".join(failed) + "\n", encoding='utf-8') else: UNINSTALL_PENDING_FILE.unlink() 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"] try: module = importlib.import_module(item["tool_name"]) obj = module.create_plugin(WORKSPACE_PATH) except Exception as e: logger.error(f"本地插件 {name} 实例化失败，跳过: {e}") continue if name not in self.plugins: self.plugins[name] = {} self.plugins[name]["obj"] = obj 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) @Slot(str, bool, str) def _on_uninstall_finished(self, name: str, success: bool, message: str) -> None: if not success: logger.error(message) return logger.info(message) info = self.plugins.get(name) if info is not None: info.pop("obj", None) info["local version"] = None self.uninstall_completed_signal.emit(name) 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" self._uninstall_requested.emit(name, local_path) elif event == Event.Update: 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._update_requested.emit(name, local_path, remote_path)