| 插件名称 | 版本 | 作者 | 状态 | 操作 |
|---|
插件管理页面渲染出错：{{e}}
" def _store_content(self) -> str: """渲染插件商店页面 - 纯 HTML/Python 模板""" try: # 获取可用插件列表 available = self._get_available_plugins() installed = self._get_installed_plugins() plugin_cards = "" for pkg_name, info in available.items(): is_installed = pkg_name in installed action_btn = f'' if not is_installed else '' plugin_cards += f"""{info.get('description', '暂无描述')}
插件商店页面渲染出错：{{e}}
" # ==================== API 处理 ==================== def _handle_list_plugins(self, request): """列出所有已安装插件""" plugins = self._scan_all_plugins() return Response(status=200, headers={"Content-Type": "application/json"}, body=json.dumps(plugins, ensure_ascii=False)) def _handle_get_config(self, request): """获取插件配置 schema + 当前值""" plugin_name = request.path_params.get('name', '') schema = self._load_config_schema(plugin_name) current = self._load_plugin_config(plugin_name) return Response(status=200, headers={"Content-Type": "application/json"}, body=json.dumps({ "schema": schema, "current": current }, ensure_ascii=False)) def _handle_save_config(self, request): """保存插件配置""" import json as json_mod try: body = json_mod.loads(request.body) plugin_name = request.path_params.get('name', '') self._save_plugin_config(plugin_name, body) return Response(status=200, headers={"Content-Type": "application/json"}, body='{"ok":true}') except Exception as e: return Response(status=500, headers={"Content-Type": "application/json"}, body=json.dumps({"error": str(e)})) def _handle_get_plugin_info(self, request): """获取插件详细信息""" plugin_name = request.path_params.get('name', '') info = self._get_plugin_detailed_info(plugin_name) return Response(status=200, headers={"Content-Type": "application/json"}, body=json.dumps(info, ensure_ascii=False)) def _handle_uninstall(self, request): """卸载插件""" import shutil plugin_name = request.path_params.get('name', '') # 查找插件目录 plugin_dir = self._find_plugin_dir(plugin_name) if not plugin_dir: return Response(status=404, body='{"error":"插件未安装"}') try: shutil.rmtree(plugin_dir) return Response(status=200, body='{"ok":true}') except Exception as e: return Response(status=500, body=json.dumps({"error": str(e)})) def _handle_remote_store(self, request): """从 Gitee API 获取远程插件列表""" try: plugins = self._fetch_remote_plugins() return Response(status=200, headers={"Content-Type": "application/json"}, body=json.dumps(plugins, ensure_ascii=False)) except Exception as e: return Response(status=500, body=json.dumps({"error": str(e)})) def _handle_store_install(self, request): """安装插件""" import json as json_mod try: body = json_mod.loads(request.body) name = body.get("name", "") author = body.get("author", "FutureOSS") success = self._install_from_gitee(name, author) return Response(status=200, body=json.dumps({"ok": success})) except Exception as e: return Response(status=500, body=json.dumps({"error": str(e)})) # ==================== Gitee 远程商店 ==================== def _fetch_remote_plugins(self) -> list: """从 Gitee 获取所有可用插件（带缓存+限速+重试）""" import time now = time.time() if self._remote_cache and (now - self._cache_time) < self._cache_ttl: return self._remote_cache plugins = [] try: store_url = f"{GITEE_API_BASE}/store" # 重试 3 次，每次间隔增加 for attempt in range(3): try: with _gitee_request(store_url, timeout=15) as resp: dirs = json.loads(resp.read().decode("utf-8")) break except Exception as e: if attempt < 2: time.sleep(1 + attempt) continue raise time.sleep(0.5) for dir_info in dirs: if dir_info.get("type") != "dir": continue author = dir_info.get("name", "") if not author.startswith("@{"): continue author_url = f"{GITEE_API_BASE}/store/{urllib.parse.quote(author, safe='')}" for attempt in range(3): try: with _gitee_request(author_url, timeout=15) as resp: plugin_dirs = json.loads(resp.read().decode("utf-8")) break except Exception: if attempt < 2: time.sleep(1 + attempt) continue raise time.sleep(0.5) for plugin_dir in plugin_dirs: if plugin_dir.get("type") != "dir": continue plugin_name = plugin_dir.get("name", "") manifest_url = f"{GITEE_API_BASE}/store/{urllib.parse.quote(author, safe='')}/{plugin_name}/manifest.json" manifest = {} for attempt in range(3): try: with _gitee_request(manifest_url, timeout=15) as resp: manifest = json.loads(resp.read().decode("utf-8")) break except Exception: if attempt < 2: time.sleep(1 + attempt) continue plugins.append({ "name": plugin_name, "author": author, "full_name": f"{author}/{plugin_name}", "metadata": manifest.get("metadata", {}), "dependencies": manifest.get("dependencies", []), "has_config": False, "is_installed": self._is_plugin_installed(plugin_name, author) }) time.sleep(0.5) self._remote_cache = plugins self._cache_time = now except Exception as e: Log.error("pkg-manager", f"获取远程插件列表失败: {e}") return plugins def _install_from_gitee(self, plugin_name: str, author: str) -> bool: """从 Gitee 下载并安装插件（使用 raw URL）""" import shutil, time install_dir = self.store_dir / author / plugin_name install_dir.mkdir(parents=True, exist_ok=True) try: # 获取目录结构（需要一次 API 调用） api_url = f"{GITEE_API_BASE}/store/{author}/{plugin_name}" with _gitee_request(api_url, timeout=15) as resp: items = json.loads(resp.read().decode("utf-8")) time.sleep(0.5) for item in items: if item.get("type") == "file": # 使用 raw URL 下载文件（不走 API） filename = item.get("name") raw_url = f"{GITEE_RAW_BASE}/store/{author}/{plugin_name}/{filename}" local_file = install_dir / filename try: with _gitee_request(raw_url, timeout=15) as resp: content = resp.read() with open(local_file, 'wb') as f: f.write(content) except: pass elif item.get("type") == "dir": sub_dir = item.get("name") self._download_dir_raw(author, plugin_name, sub_dir, install_dir / sub_dir) time.sleep(0.3) Log.info("pkg-manager", f"已安装: {author}/{plugin_name}") return True except Exception as e: Log.error("pkg-manager", f"安装失败 {plugin_name}: {e}") if install_dir.exists(): shutil.rmtree(install_dir) return False def _download_dir_raw(self, author: str, plugin: str, sub_dir: str, local_dir: Path): """使用 raw URL 递归下载子目录""" import time try: api_url = f"{GITEE_API_BASE}/store/{author}/{plugin}/{sub_dir}" with _gitee_request(api_url, timeout=15) as resp: items = json.loads(resp.read().decode("utf-8")) local_dir.mkdir(parents=True, exist_ok=True) for item in items: if item.get("type") == "file": filename = item.get("name") raw_url = f"{GITEE_RAW_BASE}/store/{author}/{plugin}/{sub_dir}/{filename}" try: with _gitee_request(raw_url, timeout=15) as resp: content = resp.read() with open(local_dir / filename, 'wb') as f: f.write(content) except: pass elif item.get("type") == "dir": self._download_dir_raw(author, plugin, f"{sub_dir}/{item.get('name')}", local_dir / item.get("name")) except: pass # ==================== 辅助方法 ==================== def _scan_all_plugins(self) -> list: """扫描本地已安装插件""" plugins = [] if not self.store_dir.exists(): return plugins for author_dir in self.store_dir.iterdir(): if author_dir.is_dir() and author_dir.name.startswith("@{"): for plugin_dir in author_dir.iterdir(): if plugin_dir.is_dir() and (plugin_dir / "main.py").exists(): manifest_path = plugin_dir / "manifest.json" if manifest_path.exists(): with open(manifest_path, 'r', encoding='utf-8') as f: manifest = json.load(f) plugins.append({ "name": plugin_dir.name, "full_name": f"{author_dir.name}/{plugin_dir.name}", "author": author_dir.name, "metadata": manifest.get("metadata", {}), "dependencies": manifest.get("dependencies", []), "has_config": (plugin_dir / "config.json").exists(), "is_installed": True }) return plugins def _is_plugin_installed(self, plugin_name: str, author: str) -> bool: """检查插件是否已安装""" plugin_dir = self.store_dir / author / plugin_name return (plugin_dir / "main.py").exists() def _find_plugin_dir(self, plugin_name: str) -> Path | None: """查找插件目录""" if not self.store_dir.exists(): return None for author_dir in self.store_dir.iterdir(): if author_dir.is_dir(): plugin_dir = author_dir / plugin_name if plugin_dir.exists() and (plugin_dir / "main.py").exists(): return plugin_dir return None def _load_config_schema(self, plugin_name: str) -> dict: """加载插件 config.json schema""" plugin_dir = self._find_plugin_dir(plugin_name) if not plugin_dir: return {} schema_path = plugin_dir / "config.json" if not schema_path.exists(): return {} with open(schema_path, 'r', encoding='utf-8') as f: return json.load(f) def _load_plugin_config(self, plugin_name: str) -> dict: """加载插件当前配置""" schema = self._load_config_schema(plugin_name) defaults = {} for key, field_def in schema.items(): defaults[key] = field_def.get("default") if self.storage: storage_instance = self.storage.get_storage("pkg-manager") user_config = storage_instance.get(f"plugin_config.{plugin_name}", {}) defaults.update(user_config) return defaults def _save_plugin_config(self, plugin_name: str, config: dict): """保存插件配置""" if self.storage: storage_instance = self.storage.get_storage("pkg-manager") storage_instance.set(f"plugin_config.{plugin_name}", config) def _get_plugin_detailed_info(self, plugin_name: str) -> dict: """获取插件的依赖、事件、页面信息""" dependencies = [] events = [] # 事件 = 功能描述 plugin_dir = self._find_plugin_dir(plugin_name) if plugin_dir: manifest_path = plugin_dir / "manifest.json" if manifest_path.exists(): with open(manifest_path, 'r', encoding='utf-8') as f: manifest = json.load(f) dependencies = manifest.get("dependencies", []) # 从 manifest 的 metadata.description 或 type 中提取功能 metadata = manifest.get("metadata", {}) plugin_type = metadata.get("type", "") if plugin_type: events.append(f"类型: {plugin_type}") # 从 manifest config 推断功能 config = manifest.get("config", {}) if config.get("enabled"): events.append("已启用") # 只返回该插件自己注册的页面（通过插件名匹配） pages = [] if self.webui and hasattr(self.webui, 'server') and self.webui.server: for path, provider in self.webui.server.pages.items(): # 检查 provider 是否属于该插件 provider_name = getattr(provider, '__self__', None) if provider_name and isinstance(provider_name, PkgManagerPlugin): continue # 跳过自己的页面 # 通过路径前缀判断（dashboard 注册 /dashboard） if path == f'/{plugin_name}' or path.startswith(f'/{plugin_name}/'): pages.append({"path": path}) # 特殊处理：首页 if plugin_name == 'webui' and path == '/': pages.append({"path": path}) return { "name": plugin_name, "dependencies": dependencies, "config_fields": list(self._load_config_schema(plugin_name).keys()), "pages": pages, "events": events } register_plugin_type("PkgManagerPlugin", PkgManagerPlugin) def New(): return PkgManagerPlugin()