"""插件注册中心。 架构说明： - 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') 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 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_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)) @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)) @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) 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) 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._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 文件并清空清单。 必须在任何 importlib.import_module 之前调用，此时 .pyd 未被加载， Windows 文件锁不会触发 WinError 5。但自重启场景下，原进程可能尚未 完全释放文件句柄，因此删除失败时短暂重试。 """ if not UNINSTALL_PENDING_FILE.exists(): return lines = UNINSTALL_PENDING_FILE.read_text(encoding='utf-8').splitlines() pending = [ln.strip() for ln in lines if ln.strip()] failed: list = [] for tool_name in pending: target = LOCAL_PLUGINS_PATH / f"{tool_name}.pyd" if not target.exists(): continue if _remove_with_retry(target): logger.info(f"已删除插件文件: {target}") else: failed.append(tool_name) logger.error(f"删除插件文件失败（重试后仍被占用）: {target}") 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() 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)