#!/usr/bin/env python3 # -*- coding: utf-8 -*- import logging, os, shutil, subprocess, tempfile, zipfile, asyncio, json from pathlib import Path from aiohttp import web from ..utils.auth import panel_auth logger = logging.getLogger(__name__) PLUGINS_DIR = Path("plugins") _IMPORT_FAILS: dict[str, str] = {} # plugin_name → error message def _run_in_subprocess(cmd: list, cwd: str, timeout: int = 120) -> tuple[bool, str]: """在子进程中执行命令，不阻塞主事件循环""" try: r = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True, timeout=timeout) return r.returncode == 0, r.stdout[-500:] + "\n" + r.stderr[-500:] except subprocess.TimeoutExpired: return False, "安装超时 (>120s)" except Exception as e: return False, str(e) def _install_plugin_deps(plugin_dir: str) -> tuple[bool, str]: """安装插件依赖 (requirements.txt)""" req_file = os.path.join(plugin_dir, "requirements.txt") if not os.path.exists(req_file): return True, "" return _run_in_subprocess( ["pip", "install", "-r", "requirements.txt", "--break-system-packages"], plugin_dir, ) async def _async_install_deps(plugin_dir: str) -> tuple[bool, str]: """异步包装器 — 在线程池中运行 pip install""" loop = asyncio.get_event_loop() return await loop.run_in_executor(None, _install_plugin_deps, plugin_dir) def setup_routes(app, prefix=""): app.router.add_get(f"{prefix}/api/plugins", panel_auth(list_plugins)) app.router.add_get(f"{prefix}/api/plugin-pages", panel_auth(list_plugin_web_pages)) app.router.add_get(f"{prefix}/api/plugins/stats", panel_auth(plugin_stats)) app.router.add_post(f"{prefix}/api/plugins/{{name}}/{{action}}", panel_auth(manage_plugin)) app.router.add_post(f"{prefix}/api/plugins/import", panel_auth(import_plugin)) app.router.add_get(f"{prefix}/api/plugins/{{name}}/perms", panel_auth(get_perms)) app.router.add_post(f"{prefix}/api/plugins/{{name}}/perms", panel_auth(set_perms)) # ── 统计 ── async def plugin_stats(req): sm = req.app.get("service_manager") ps = sm.get_service("plugin") if sm else None total = len(ps.plugin_info) if ps else 0 loaded = sum(1 for n, info in ps.plugin_info.items() if n in ps.plugins) if ps else 0 disabled = total - loaded # 合并: 导入失败 + 加载失败 from services.plugin_service import LOAD_FAILURES failed = len(_IMPORT_FAILS) fail_details = dict(_IMPORT_FAILS) for name, reason in LOAD_FAILURES.items(): if name not in fail_details: fail_details[name] = reason failed += 1 if ps: for name, info in ps.plugin_info.items(): if getattr(info, "error_count", 0) > 0 and name not in fail_details: fail_details[name] = f"启动失败 (错误数: {info.error_count})" failed += 1 return web.json_response({ "total": total, "loaded": loaded, "disabled": max(0, disabled), "failed": failed, "fail_details": fail_details, }) # ── 导入 ── async def import_plugin(req): """导入插件: multipart zip 或 JSON {url: git地址}""" sm = req.app.get("service_manager") ps = sm.get_service("plugin") if sm else None if not ps: return web.json_response({"ok": False, "error": "Plugin Service 未就绪"}, status=503) content_type = req.content_type or "" # ── Git clone 方式 ── if "application/json" in content_type: try: data = await req.json() git_url = data.get("url", "").strip() if not git_url: return web.json_response({"ok": False, "error": "Git URL 不能为空"}, status=400) # 提取 repo 名作为插件名 name = git_url.rstrip("/").split("/")[-1] if name.endswith(".git"): name = name[:-4] target = PLUGINS_DIR / name if target.exists(): return web.json_response({"ok": False, "error": f"插件目录已存在: {name}"}, status=409) ok, out = _run_in_subprocess( ["git", "clone", git_url, str(target)], str(PLUGINS_DIR) ) if not ok: _IMPORT_FAILS[name] = f"git clone 失败: {out[-200:]}" return web.json_response({"ok": False, "error": "Git clone 失败", "detail": out[-300:]}, status=500) # 安装依赖 dep_ok, dep_out = await _async_install_deps(str(target)) if not dep_ok: logger.warning(f"依赖安装部分失败 {name}: {dep_out[-200:]}") # 加载插件 loaded = await _safe_load_plugin(ps, name) _IMPORT_FAILS.pop(name, None) return web.json_response({ "ok": True, "name": name, "loaded": loaded, "dep_ok": dep_ok, "detail": dep_out[-200:] if not dep_ok else "", }) except Exception as e: return web.json_response({"ok": False, "error": str(e)}, status=500) # ── ZIP 上传方式 ── try: reader = await req.multipart() part = await reader.next() while part: if part.name == "file" and part.filename: fname = part.filename if not fname.endswith(".zip"): return web.json_response({"ok": False, "error": "仅支持 .zip 文件"}, status=400) # 保存到临时文件 tmp = tempfile.NamedTemporaryFile(suffix=".zip", delete=False) while True: chunk = await part.read_chunk(65536) if not chunk: break tmp.write(chunk) tmp.close() # 提取 → plugins/ plugin_name = fname[:-4] # 去掉 .zip target = PLUGINS_DIR / plugin_name if target.exists(): os.unlink(tmp.name) return web.json_response({"ok": False, "error": f"插件目录已存在: {plugin_name}"}, status=409) try: with zipfile.ZipFile(tmp.name, "r") as zf: # 检查 zip 结构 — 如果顶层只有一个目录，用它 members = zf.namelist() top_dirs = set() for m in members: if "/" in m: top_dirs.add(m.split("/")[0]) if len(top_dirs) == 1 and all(m.startswith(list(top_dirs)[0]) for m in members if m): # 有统一顶层目录 zf.extractall(PLUGINS_DIR) if list(top_dirs)[0] != plugin_name: src = PLUGINS_DIR / list(top_dirs)[0] src.rename(target) else: target.mkdir(parents=True) zf.extractall(target) except zipfile.BadZipFile: os.unlink(tmp.name) return web.json_response({"ok": False, "error": "无效的 ZIP 文件"}, status=400) finally: os.unlink(tmp.name) # 安装依赖 dep_ok, dep_out = await _async_install_deps(str(target)) # 加载插件 loaded = await _safe_load_plugin(ps, plugin_name) if loaded: _IMPORT_FAILS.pop(plugin_name, None) return web.json_response({ "ok": True, "name": plugin_name, "loaded": loaded, "dep_ok": dep_ok, "detail": dep_out[-200:] if not dep_ok else "", }) part = await reader.next() return web.json_response({"ok": False, "error": "未找到上传文件 (字段名: file)"}, status=400) except Exception as e: logger.error(f"导入插件失败: {e}") return web.json_response({"ok": False, "error": str(e)}, status=500) async def _safe_load_plugin(ps, name: str) -> bool: """安全加载插件 — 先隔离模式尝试，成功后再切换到正常模式""" try: # 先尝试隔离加载 (子进程，崩溃不影响框架) from services.process_isolated import IsolatedPlugin plugin_path = PLUGINS_DIR / name / "__init__.py" if not plugin_path.exists(): _IMPORT_FAILS[name] = "缺少 __init__.py" return False iso = IsolatedPlugin(name, str(plugin_path), {"settings": {"isolation": True}}) ps.plugins[name] = iso logger.info(f"🔌 插件已导入(隔离模式): {name}") return True except Exception as e: _IMPORT_FAILS[name] = str(e)[:200] logger.error(f"插件加载失败 {name}: {e}") return False # ── 列表 / 管理 / 权限 ── async def list_plugins(req): sm = req.app.get("service_manager") if not sm: return web.json_response({"error": "Service Manager 未初始化"}, status=503) ps = sm.get_service("plugin") if not ps: return web.json_response({"plugins": []}) data = [] for name, info in ps.plugin_info.items(): data.append({ "name": name, "version": getattr(info, "version", "?"), "running": name in ps.plugins, "enabled": getattr(info, "enabled", True), "error": _IMPORT_FAILS.get(name, ""), }) return web.json_response({"plugins": data}) async def manage_plugin(req): sm = req.app.get("service_manager") if not sm: return web.json_response({"error": "SM Missing"}, status=503) name = req.match_info["name"] action = req.match_info["action"] ps = sm.get_service("plugin") if not ps: return web.json_response({"error": "Plugin Service Missing"}, status=503) try: if action in ("disable", "unload"): await ps.unload_plugin(name) elif action == "enable": await ps.load_plugin(name) elif action == "reload": await ps.unload_plugin(name) await ps.load_plugin(name) return web.json_response({"success": True, "msg": "操作成功"}) except Exception as e: logger.error(f"插件操作失败: {e}") return web.json_response({"success": False, "error": str(e)}) async def list_plugin_web_pages(req): sm = req.app.get("service_manager") if not sm: return web.json_response({"pages": []}) ps = sm.get_service("plugin") if not ps: return web.json_response({"pages": []}) pages = [] for name, plugin in ps.plugins.items(): if hasattr(plugin, "get_web_pages"): for path, info in plugin.get_web_pages().items(): pages.append({ "plugin": name, "path": "/plugin/" + name, "title": info.get("title", name), "icon": info.get("icon", "P"), }) return web.json_response({"pages": pages}) async def get_perms(req): return web.json_response({"plugin": req.match_info["name"], "permissions": ["read", "write"]}) async def set_perms(req): return web.json_response({"success": True})