"""插件注册中心。 架构说明： - PluginWorker 运行在独立 QThread 中，仅做纯 I/O（扫描 zip/.pyd、下载解压、 读 manifest 元数据），不创建 QObject、不操作 QWidget。 - PluginRegistry 留在主线程，负责 create_plugin()（返回 QWidget，必须在 GUI 线程 创建）与信号转发。 - 跨线程通信全部走 Qt 信号槽（自动 Queued），不再使用 threading.Thread。 分发模型： - 远程：每插件一个 --.zip（export_runtime + pack_zip 产物） - 本地：解压到 plugins// 子目录，卸载删整个子目录 """ import os import sys import time import json import shutil import zipfile 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 _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 def _pyd_module_name(file_path: Path) -> str: """从 .pyd 文件名提取 Python 模块名（PyInit_ 后缀）。 Nuitka 产物形如 mil.cp311-win_amd64.pyd，C 扩展入口函数为 PyInit_mil，因此模块名必须取第一个点之前的部分，不能用完整 stem。 """ return file_path.name.split('.', 1)[0] 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: """扫描 plugins/<插件名>/ 子目录下的主插件 .pyd。""" plugins_dir = Path(plugins_dir) if not self._ensure_dir(plugins_dir): return self.local_discovered.emit(self._scan_local_plugins(plugins_dir)) @Slot(str) def do_discover_remote(self, plugins_dir: str) -> None: """扫描远程 *.zip，读 zip 内 manifest.json 取插件元数据。""" if not os.path.exists(plugins_dir): logger.error(f"服务器链接错误！路径不存在: {plugins_dir}") return logger.info(f"开始扫描远程插件 zip: {plugins_dir}") self.remote_discovered.emit( self._scan_remote_zips(Path(plugins_dir)) ) @Slot(str, str) def do_install(self, local_dir: str, remote_zip: str) -> None: """下载 zip 流式写本地临时文件，再解压到 plugins/<插件名>/。 流式下载保留 install_progress 信号，便于 UI 显示进度条。 """ try: total_size = os.path.getsize(remote_zip) target = Path(local_dir) target.mkdir(parents=True, exist_ok=True) tmp_zip = target / ".download.tmp" copied = 0 with open(remote_zip, 'rb') as fsrc, open(tmp_zip, 'wb') as fdst: while True: buf = fsrc.read(1024 * 1024) if not buf: break fdst.write(buf) copied += len(buf) self.install_progress.emit(copied, total_size) with zipfile.ZipFile(tmp_zip, 'r') as zf: zf.extractall(target) tmp_zip.unlink() self.install_finished.emit(True, f"安装成功: {local_dir}") except Exception as e: logger.error(f"{remote_zip}安装失败{e.args}") self.install_finished.emit(False, str(e)) @Slot(str, str) def do_uninstall(self, name: str, local_dir: str) -> None: """运行时无法删除已加载 .pyd，记入待删清单（目录路径），重启时删整个目录。 真正的目录删除由 PluginRegistry._process_pending_uninstalls 在下次启动、 import 之前完成。运行时清理（字典/UI/widget）由主线程在 uninstall_finished 回调中处理。 """ try: UNINSTALL_PENDING_FILE.parent.mkdir(parents=True, exist_ok=True) with open(UNINSTALL_PENDING_FILE, 'a', encoding='utf-8') as f: f.write(f"{local_dir}\n") self.uninstall_finished.emit(name, True, f"{name}已加入卸载清单，重启后生效!!!") except Exception as e: logger.error(f"{local_dir}卸载失败{e.args}") self.uninstall_finished.emit(name, False, str(e)) @Slot(str, str, str) def do_update(self, name: str, local_dir: str, remote_zip: str) -> None: """更新=删旧目录+从远程 zip 解压新版本。运行时记清单，重启时执行。 运行时 .pyd 已加载无法删除，记入待处理清单（含 remote_zip）， 重启时由 _process_pending_uninstalls 删除旧目录并解压新 zip。 运行时 UI 清理复用 uninstall_finished 信号链路，触发与卸载一致的自重启。 """ try: UNINSTALL_PENDING_FILE.parent.mkdir(parents=True, exist_ok=True) with open(UNINSTALL_PENDING_FILE, 'a', encoding='utf-8') as f: f.write(f"{local_dir}|{remote_zip}\n") self.uninstall_finished.emit(name, True, f"{name}更新已加入清单，重启后生效!!!") except Exception as e: logger.error(f"{local_dir}更新失败{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_local_plugins(plugins_dir: Path) -> list: """扫描 plugins/<插件名>/ 子目录，读主插件 .pyd 元数据。 跳过伴生包（openpyxl/et_xmlfile）与 C 扩展（下划线开头）。 每个插件子目录下只应有一个主插件 .pyd。 """ results: list = [] for sub in plugins_dir.iterdir(): if not sub.is_dir(): continue for tool in sub.glob("*.pyd"): # .pyd 文件名形如 mil.cp311-win_amd64.pyd， # 模块名取第一个点之前的部分（PyInit_ 后缀） tool_name = _pyd_module_name(tool) # 伴生包与 C 扩展不是主插件，跳过 if tool_name in ("openpyxl", "et_xmlfile") or tool_name.startswith("_"): continue try: module = _load_module_from_path(tool_name, tool) results.append({ "tool_name": tool_name, "name": module.read_plugin_name(), "version": module.read_plugin_version(), "local_dir": str(sub), }) except Exception as e: logger.error(f"插件{tool_name}加载失败：{e}") return results @staticmethod def _scan_remote_zips(plugins_dir: Path) -> list: """扫描远程 *.zip，读 zip 内 manifest.json 取插件元数据。 zip 文件名约定：--.zip manifest.json 由 export_runtime.py 生成，含 plugin_name/version/description。 """ results: list = [] for zip_file in plugins_dir.glob("*.zip"): try: with zipfile.ZipFile(zip_file, "r") as zf: manifest = json.loads(zf.read("manifest.json")) results.append({ "tool_name": manifest["tool_name"], "name": manifest["plugin_name"], "version": manifest["plugin_version"], "description": manifest["plugin_description"], "remote_path": str(zip_file), }) except Exception as e: logger.error(f"远程 zip {zip_file.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) install_progress = Signal(str, int, int) # name, copied_bytes, total_bytes install_finished = Signal(bool, 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) _PENDING_CLEANUP_DELAY = 1500 def __init__(self, parent: Optional[QObject] = None) -> None: super().__init__(parent) self.plugins: dict = {} self._installing_name: str = "" self._last_logged_pct: int = 0 # 节流：记录上次上报的百分比 # Worker 在线程启动前创建（信号连接必须先于线程启动）， # 但实际启动延迟至清理完成后，避免新进程在 shutil.rmtree 之前 # 就把插件 .pyd 加载进内存导致自锁。 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._worker.install_progress.connect(self._on_install_progress) self._thread.finished.connect(self._worker.deleteLater) # 延迟启动：先等旧进程彻底退出并释放 .pyd 句柄， # 再执行待删清单清理，最后才扫描并加载插件。 # 顺序必须是 清理 → 启动线程 → 发现插件，否则新进程自锁。 QTimer.singleShot(self._PENDING_CLEANUP_DELAY, self._post_init) def _post_init(self) -> None: """延迟初始化：清理 → 启动线程 → 发现插件。 必须在旧进程退出后执行，且清理必须在任何插件 import 之前完成， 否则新进程自己持有 .pyd 文件锁导致 shutil.rmtree 失败。 """ self._process_pending_uninstalls() self._thread.start() 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: """启动时执行待处理清单：删除插件目录，更新项另从远程 zip 解压。 清单行格式： - ```` 纯卸载，仅删除本地目录 - ``|`` 更新，删除后从 remote_zip 解压新版本 必须在任何 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) target = Path(parts[0]) # 纯卸载且目录已不存在视为已完成；更新项即使目录不存在也要继续解压 if not target.exists() and len(parts) == 1: continue # 删除整个插件目录（带重试：Windows 文件锁可能在短时间后释放） deleted = False last_error = None for attempt in range(3): try: shutil.rmtree(target) logger.info(f"已删除插件目录: {target}") deleted = True break except Exception as e: last_error = e if attempt < 2: time.sleep(0.5) if not deleted: failed.append(line) logger.error(f"删除插件目录失败（可能仍被占用）: {target} {last_error}") continue # 更新项：从远程 zip 解压新版本 if len(parts) == 2: remote_zip = Path(parts[1]) try: target.mkdir(parents=True, exist_ok=True) with zipfile.ZipFile(remote_zip, 'r') as zf: zf.extractall(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[name]["local_dir"] = item["local_dir"] self.plugins_loader_signal.emit(self.plugins) # 本地扫描完成后也刷新卡片（安装后 local version 更新，卡片需重绘状态） self.update_plugins_card_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) for sub in LOCAL_PLUGINS_PATH.iterdir(): if sub.is_dir() and str(sub) not in sys.path: sys.path.insert(0, str(sub)) self._discover_local_requested.emit(str(LOCAL_PLUGINS_PATH)) else: logger.error(message) self._installing_name = "" self.install_finished.emit(success, message) @Slot(int, int) def _on_install_progress(self, copied: int, total: int) -> None: """将安装进度写入日志（显示在 FrameLog），按百分比节流避免刷屏。""" if not self._installing_name or total == 0: return pct = int(copied / total * 100) if pct - self._last_logged_pct >= 10 or pct == 100: mb_copied = copied / 1024 / 1024 mb_total = total / 1024 / 1024 logger.info(f"[{self._installing_name}] 安装进度: {pct}% ({mb_copied:.1f}/{mb_total:.1f} MB)") self._last_logged_pct = pct @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: """本地子目录路径作为 install/uninstall/update 的目标。 本地布局：plugins//（zip 解压产物） 远程来源：plugins[name]['remote path'] 指向 .zip 文件 """ tool_name = self.plugins[name]['tool_name'] local_dir = str(LOCAL_PLUGINS_PATH / tool_name) if event == Event.Install: self._installing_name = name self._last_logged_pct = 0 logger.info(f"[{name}] 开始安装...") remote_path = self.plugins[name]['remote path'] self._install_requested.emit(local_dir, remote_path) elif event == Event.Uninstall: self._uninstall_requested.emit(name, local_dir) elif event == Event.Update: self._installing_name = name remote_path = self.plugins[name]['remote path'] self._update_requested.emit(name, local_dir, remote_path)