From 321c784434c700158d79c4dd96e118e5f959a64c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A2=A8=E6=A2=93=E6=9F=92?= <1787882683@qq.com> Date: Sat, 29 Nov 2025 00:01:25 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=A2=9E=E5=BC=BA=E6=8F=92=E4=BB=B6?= =?UTF-8?q?=E9=85=8D=E7=BD=AE=E7=AE=A1=E7=90=86=EF=BC=8C=E6=B7=BB=E5=8A=A0?= =?UTF-8?q?=E9=85=8D=E7=BD=AE=20Schema=20=E5=92=8C=E6=9B=B4=E6=96=B0?= =?UTF-8?q?=E6=8E=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/plugin_system/base/config_types.py | 264 +++++++++++++++- src/plugin_system/base/plugin_base.py | 98 +++++- src/webui/plugin_routes.py | 415 ++++++++++++++++++++++++- 3 files changed, 767 insertions(+), 10 deletions(-) diff --git a/src/plugin_system/base/config_types.py b/src/plugin_system/base/config_types.py index 752b3345..ef0f656c 100644 --- a/src/plugin_system/base/config_types.py +++ b/src/plugin_system/base/config_types.py @@ -1,18 +1,270 @@ """ 插件系统配置类型定义 + +提供插件配置的类型定义，支持 WebUI 可视化配置编辑。 """ -from typing import Any, Optional, List +from typing import Any, Optional, List, Dict, Union, Callable from dataclasses import dataclass, field @dataclass class ConfigField: - """配置字段定义""" + """ + 配置字段定义 + + 用于定义插件配置项的元数据，支持类型验证、UI 渲染等功能。 + + 基础示例: + ConfigField(type=str, default="", description="API密钥") + + 完整示例: + ConfigField( + type=str, + default="", + description="API密钥", + input_type="password", + placeholder="请输入API密钥", + required=True, + hint="从服务商控制台获取", + order=1 + ) + """ - type: type # 字段类型 + # === 基础字段（必需） === + type: type # 字段类型: str, int, float, bool, list, dict default: Any # 默认值 - description: str # 字段描述 - example: Optional[str] = None # 示例值 + description: str # 字段描述（也用作默认标签） + + # === 验证相关 === + example: Optional[str] = None # 示例值（用于生成配置文件注释） required: bool = False # 是否必需 - choices: Optional[List[Any]] = field(default_factory=list) # 可选值列表 + choices: Optional[List[Any]] = field(default_factory=list) # 可选值列表（用于下拉选择） + min: Optional[float] = None # 最小值（数字类型） + max: Optional[float] = None # 最大值（数字类型） + step: Optional[float] = None # 步进值（数字类型） + pattern: Optional[str] = None # 正则验证（字符串类型） + max_length: Optional[int] = None # 最大长度（字符串类型） + + # === UI 显示控制 === + label: Optional[str] = None # 显示标签（默认使用 description） + placeholder: Optional[str] = None # 输入框占位符 + hint: Optional[str] = None # 字段下方的提示文字 + icon: Optional[str] = None # 字段图标名称 + hidden: bool = False # 是否在 UI 中隐藏 + disabled: bool = False # 是否禁用编辑 + order: int = 0 # 排序权重（数字越小越靠前） + + # === 输入控件类型 === + # 可选值: text, password, textarea, number, color, code, file, json + # 不指定时根据 type 和 choices 自动推断 + input_type: Optional[str] = None + + # === textarea 专用 === + rows: int = 3 # 文本域行数 + + # === 分组与布局 === + group: Optional[str] = None # 字段分组（在 section 内再细分） + + # === 条件显示 === + depends_on: Optional[str] = None # 依赖的字段路径，如 "section.field" + depends_value: Any = None # 依赖字段需要的值（当依赖字段等于此值时显示） + + def get_ui_type(self) -> str: + """ + 获取 UI 控件类型 + + 如果指定了 input_type 则直接返回，否则根据 type 和 choices 自动推断。 + + Returns: + 控件类型字符串 + """ + if self.input_type: + return self.input_type + + # 根据 type 和 choices 自动推断 + if self.type == bool: + return "switch" + elif self.type in (int, float): + if self.min is not None and self.max is not None: + return "slider" + return "number" + elif self.type == str: + if self.choices: + return "select" + return "text" + elif self.type == list: + return "list" + elif self.type == dict: + return "json" + else: + return "text" + + def to_dict(self) -> Dict[str, Any]: + """ + 转换为可序列化的字典（用于 API 传输） + + Returns: + 包含所有配置信息的字典 + """ + return { + "type": self.type.__name__ if isinstance(self.type, type) else str(self.type), + "default": self.default, + "description": self.description, + "example": self.example, + "required": self.required, + "choices": self.choices if self.choices else None, + "min": self.min, + "max": self.max, + "step": self.step, + "pattern": self.pattern, + "max_length": self.max_length, + "label": self.label or self.description, + "placeholder": self.placeholder, + "hint": self.hint, + "icon": self.icon, + "hidden": self.hidden, + "disabled": self.disabled, + "order": self.order, + "input_type": self.input_type, + "ui_type": self.get_ui_type(), + "rows": self.rows, + "group": self.group, + "depends_on": self.depends_on, + "depends_value": self.depends_value, + } + + +@dataclass +class ConfigSection: + """ + 配置节定义 + + 用于描述配置文件中一个 section 的元数据。 + + 示例: + ConfigSection( + title="API配置", + description="外部API连接参数", + icon="cloud", + order=1 + ) + """ + title: str # 显示标题 + description: Optional[str] = None # 详细描述 + icon: Optional[str] = None # 图标名称 + collapsed: bool = False # 默认是否折叠 + order: int = 0 # 排序权重 + + def to_dict(self) -> Dict[str, Any]: + """转换为可序列化的字典""" + return { + "title": self.title, + "description": self.description, + "icon": self.icon, + "collapsed": self.collapsed, + "order": self.order, + } + + +@dataclass +class ConfigTab: + """ + 配置标签页定义 + + 用于将多个 section 组织到一个标签页中。 + + 示例: + ConfigTab( + id="general", + title="通用设置", + icon="settings", + sections=["plugin", "api"] + ) + """ + id: str # 标签页 ID + title: str # 显示标题 + sections: List[str] = field(default_factory=list) # 包含的 section 名称列表 + icon: Optional[str] = None # 图标名称 + order: int = 0 # 排序权重 + badge: Optional[str] = None # 角标文字（如 "Beta", "New"） + + def to_dict(self) -> Dict[str, Any]: + """转换为可序列化的字典""" + return { + "id": self.id, + "title": self.title, + "sections": self.sections, + "icon": self.icon, + "order": self.order, + "badge": self.badge, + } + + +@dataclass +class ConfigLayout: + """ + 配置页面布局定义 + + 用于定义插件配置页面的整体布局结构。 + + 布局类型: + - "auto": 自动布局，sections 作为折叠面板显示 + - "tabs": 标签页布局 + - "pages": 分页布局（左侧导航 + 右侧内容） + + 简单示例（标签页布局）: + ConfigLayout( + type="tabs", + tabs=[ + ConfigTab(id="basic", title="基础", sections=["plugin", "api"]), + ConfigTab(id="advanced", title="高级", sections=["debug"]), + ] + ) + """ + type: str = "auto" # 布局类型: auto, tabs, pages + tabs: List[ConfigTab] = field(default_factory=list) # 标签页列表 + + def to_dict(self) -> Dict[str, Any]: + """转换为可序列化的字典""" + return { + "type": self.type, + "tabs": [tab.to_dict() for tab in self.tabs], + } + + +def section_meta( + title: str, + description: Optional[str] = None, + icon: Optional[str] = None, + collapsed: bool = False, + order: int = 0 +) -> Union[str, ConfigSection]: + """ + 便捷函数：创建 section 元数据 + + 可以在 config_section_descriptions 中使用，提供比纯字符串更丰富的信息。 + + Args: + title: 显示标题 + description: 详细描述 + icon: 图标名称 + collapsed: 默认是否折叠 + order: 排序权重 + + Returns: + ConfigSection 实例 + + 示例: + config_section_descriptions = { + "api": section_meta("API配置", icon="cloud", order=1), + "debug": section_meta("调试设置", collapsed=True, order=99), + } + """ + return ConfigSection( + title=title, + description=description, + icon=icon, + collapsed=collapsed, + order=order + ) diff --git a/src/plugin_system/base/plugin_base.py b/src/plugin_system/base/plugin_base.py index 0b7f15d1..1fe99b8a 100644 --- a/src/plugin_system/base/plugin_base.py +++ b/src/plugin_system/base/plugin_base.py @@ -12,7 +12,11 @@ from src.plugin_system.base.component_types import ( PluginInfo, PythonDependency, ) -from src.plugin_system.base.config_types import ConfigField +from src.plugin_system.base.config_types import ( + ConfigField, + ConfigSection, + ConfigLayout, +) from src.plugin_system.utils.manifest_utils import ManifestValidator logger = get_logger("plugin_base") @@ -60,7 +64,10 @@ class PluginBase(ABC): def config_schema(self) -> Dict[str, Union[Dict[str, ConfigField], str]]: return {} - config_section_descriptions: Dict[str, str] = {} + config_section_descriptions: Dict[str, Union[str, ConfigSection]] = {} + + # 布局配置（可选，不定义则使用自动布局） + config_layout: ConfigLayout = None def __init__(self, plugin_dir: str): """初始化插件 @@ -564,6 +571,93 @@ class PluginBase(ABC): return current + def get_webui_config_schema(self) -> Dict[str, Any]: + """ + 获取 WebUI 配置 Schema + + 返回完整的配置 schema，包含： + - 插件基本信息 + - 所有 section 及其字段定义 + - 布局配置 + + 用于 WebUI 动态生成配置表单。 + + Returns: + Dict: 完整的配置 schema + """ + schema = { + "plugin_id": self.plugin_name, + "plugin_info": { + "name": self.display_name, + "version": self.plugin_version, + "description": self.plugin_description, + "author": self.plugin_author, + }, + "sections": {}, + "layout": None, + } + + # 处理 sections + for section_name, fields in self.config_schema.items(): + if not isinstance(fields, dict): + continue + + section_data = { + "name": section_name, + "title": section_name, + "description": None, + "icon": None, + "collapsed": False, + "order": 0, + "fields": {}, + } + + # 获取 section 元数据 + section_meta = self.config_section_descriptions.get(section_name) + if section_meta: + if isinstance(section_meta, str): + section_data["title"] = section_meta + elif isinstance(section_meta, ConfigSection): + section_data["title"] = section_meta.title + section_data["description"] = section_meta.description + section_data["icon"] = section_meta.icon + section_data["collapsed"] = section_meta.collapsed + section_data["order"] = section_meta.order + elif isinstance(section_meta, dict): + section_data.update(section_meta) + + # 处理字段 + for field_name, field_def in fields.items(): + if isinstance(field_def, ConfigField): + field_data = field_def.to_dict() + field_data["name"] = field_name + section_data["fields"][field_name] = field_data + + schema["sections"][section_name] = section_data + + # 处理布局 + if self.config_layout: + schema["layout"] = self.config_layout.to_dict() + else: + # 自动布局：按 section order 排序 + schema["layout"] = { + "type": "auto", + "tabs": [], + } + + return schema + + def get_current_config_values(self) -> Dict[str, Any]: + """ + 获取当前配置值 + + 返回插件当前的配置值（已从配置文件加载）。 + + Returns: + Dict: 当前配置值 + """ + return self.config.copy() + @abstractmethod def register_plugin(self) -> bool: """ diff --git a/src/webui/plugin_routes.py b/src/webui/plugin_routes.py index 0d8ab19b..5c49483e 100644 --- a/src/webui/plugin_routes.py +++ b/src/webui/plugin_routes.py @@ -29,8 +29,10 @@ def parse_version(version_str: str) -> tuple[int, int, int]: Returns: (major, minor, patch) 三元组 """ - # 移除 snapshot 等后缀 - base_version = version_str.split(".snapshot")[0].split(".dev")[0].split(".alpha")[0].split(".beta")[0] + # 移除 snapshot、dev、alpha、beta 等后缀（支持 - 和 . 分隔符） + import re + # 匹配 -snapshot.X, .snapshot, -dev, .dev, -alpha, .alpha, -beta, .beta 等后缀 + base_version = re.split(r'[-.](?:snapshot|dev|alpha|beta|rc)', version_str, flags=re.IGNORECASE)[0] parts = base_version.split(".") if len(parts) < 3: @@ -1153,3 +1155,412 @@ async def get_installed_plugins(authorization: Optional[str] = Header(None)) -> except Exception as e: logger.error(f"获取已安装插件列表失败: {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"服务器错误: {str(e)}") from e + + +# ============ 插件配置管理 API ============ + + +class UpdatePluginConfigRequest(BaseModel): + """更新插件配置请求""" + + config: Dict[str, Any] = Field(..., description="配置数据") + + +@router.get("/config/{plugin_id}/schema") +async def get_plugin_config_schema( + plugin_id: str, authorization: Optional[str] = Header(None) +) -> Dict[str, Any]: + """ + 获取插件配置 Schema + + 返回插件的完整配置 schema，包含所有 section、字段定义和布局信息。 + 用于前端动态生成配置表单。 + """ + # Token 验证 + token = authorization.replace("Bearer ", "") if authorization else None + token_manager = get_token_manager() + if not token or not token_manager.verify_token(token): + raise HTTPException(status_code=401, detail="未授权：无效的访问令牌") + + logger.info(f"获取插件配置 Schema: {plugin_id}") + + try: + # 尝试从已加载的插件中获取 + from src.plugin_system.core.plugin_manager import plugin_manager + + # 查找插件实例 + plugin_instance = None + + # 遍历所有已加载的插件 + for loaded_plugin_name in plugin_manager.list_loaded_plugins(): + instance = plugin_manager.get_plugin_instance(loaded_plugin_name) + if instance: + # 匹配 plugin_name 或 manifest 中的 id + if instance.plugin_name == plugin_id: + plugin_instance = instance + break + # 也尝试匹配 manifest 中的 id + manifest_id = instance.get_manifest_info("id", "") + if manifest_id == plugin_id: + plugin_instance = instance + break + + if plugin_instance and hasattr(plugin_instance, 'get_webui_config_schema'): + # 从插件实例获取 schema + schema = plugin_instance.get_webui_config_schema() + return {"success": True, "schema": schema} + + # 如果插件未加载，尝试从文件系统读取 + # 查找插件目录 + plugins_dir = Path("plugins") + plugin_path = None + + for p in plugins_dir.iterdir(): + if p.is_dir(): + manifest_path = p / "_manifest.json" + if manifest_path.exists(): + try: + with open(manifest_path, "r", encoding="utf-8") as f: + manifest = json.load(f) + if manifest.get("id") == plugin_id or p.name == plugin_id: + plugin_path = p + break + except Exception: + continue + + if not plugin_path: + raise HTTPException(status_code=404, detail=f"未找到插件: {plugin_id}") + + # 读取配置文件获取当前配置 + config_path = plugin_path / "config.toml" + current_config = {} + if config_path.exists(): + import toml + current_config = toml.load(config_path) + + # 构建基础 schema（无法获取完整的 ConfigField 信息） + schema = { + "plugin_id": plugin_id, + "plugin_info": { + "name": plugin_id, + "version": "", + "description": "", + "author": "", + }, + "sections": {}, + "layout": {"type": "auto", "tabs": []}, + "_note": "插件未加载，仅返回当前配置结构", + } + + # 从当前配置推断 schema + for section_name, section_data in current_config.items(): + if isinstance(section_data, dict): + schema["sections"][section_name] = { + "name": section_name, + "title": section_name, + "description": None, + "icon": None, + "collapsed": False, + "order": 0, + "fields": {}, + } + for field_name, field_value in section_data.items(): + # 推断字段类型 + field_type = type(field_value).__name__ + ui_type = "text" + if isinstance(field_value, bool): + ui_type = "switch" + elif isinstance(field_value, (int, float)): + ui_type = "number" + elif isinstance(field_value, list): + ui_type = "list" + elif isinstance(field_value, dict): + ui_type = "json" + + schema["sections"][section_name]["fields"][field_name] = { + "name": field_name, + "type": field_type, + "default": field_value, + "description": field_name, + "label": field_name, + "ui_type": ui_type, + "required": False, + "hidden": False, + "disabled": False, + "order": 0, + } + + return {"success": True, "schema": schema} + + except HTTPException: + raise + except Exception as e: + logger.error(f"获取插件配置 Schema 失败: {e}", exc_info=True) + raise HTTPException(status_code=500, detail=f"服务器错误: {str(e)}") from e + + +@router.get("/config/{plugin_id}") +async def get_plugin_config( + plugin_id: str, authorization: Optional[str] = Header(None) +) -> Dict[str, Any]: + """ + 获取插件当前配置值 + + 返回插件的当前配置值。 + """ + # Token 验证 + token = authorization.replace("Bearer ", "") if authorization else None + token_manager = get_token_manager() + if not token or not token_manager.verify_token(token): + raise HTTPException(status_code=401, detail="未授权：无效的访问令牌") + + logger.info(f"获取插件配置: {plugin_id}") + + try: + # 查找插件目录 + plugins_dir = Path("plugins") + plugin_path = None + + for p in plugins_dir.iterdir(): + if p.is_dir(): + manifest_path = p / "_manifest.json" + if manifest_path.exists(): + try: + with open(manifest_path, "r", encoding="utf-8") as f: + manifest = json.load(f) + if manifest.get("id") == plugin_id or p.name == plugin_id: + plugin_path = p + break + except Exception: + continue + + if not plugin_path: + raise HTTPException(status_code=404, detail=f"未找到插件: {plugin_id}") + + # 读取配置文件 + config_path = plugin_path / "config.toml" + if not config_path.exists(): + return {"success": True, "config": {}, "message": "配置文件不存在"} + + import toml + config = toml.load(config_path) + + return {"success": True, "config": config} + + except HTTPException: + raise + except Exception as e: + logger.error(f"获取插件配置失败: {e}", exc_info=True) + raise HTTPException(status_code=500, detail=f"服务器错误: {str(e)}") from e + + +@router.put("/config/{plugin_id}") +async def update_plugin_config( + plugin_id: str, + request: UpdatePluginConfigRequest, + authorization: Optional[str] = Header(None) +) -> Dict[str, Any]: + """ + 更新插件配置 + + 保存新的配置值到插件的配置文件。 + """ + # Token 验证 + token = authorization.replace("Bearer ", "") if authorization else None + token_manager = get_token_manager() + if not token or not token_manager.verify_token(token): + raise HTTPException(status_code=401, detail="未授权：无效的访问令牌") + + logger.info(f"更新插件配置: {plugin_id}") + + try: + # 查找插件目录 + plugins_dir = Path("plugins") + plugin_path = None + + for p in plugins_dir.iterdir(): + if p.is_dir(): + manifest_path = p / "_manifest.json" + if manifest_path.exists(): + try: + with open(manifest_path, "r", encoding="utf-8") as f: + manifest = json.load(f) + if manifest.get("id") == plugin_id or p.name == plugin_id: + plugin_path = p + break + except Exception: + continue + + if not plugin_path: + raise HTTPException(status_code=404, detail=f"未找到插件: {plugin_id}") + + config_path = plugin_path / "config.toml" + + # 备份旧配置 + import shutil + import datetime + if config_path.exists(): + backup_name = f"config.toml.backup.{datetime.datetime.now().strftime('%Y%m%d%H%M%S')}" + backup_path = plugin_path / backup_name + shutil.copy(config_path, backup_path) + logger.info(f"已备份配置文件: {backup_path}") + + # 写入新配置 + import toml + with open(config_path, "w", encoding="utf-8") as f: + toml.dump(request.config, f) + + logger.info(f"已更新插件配置: {plugin_id}") + + return { + "success": True, + "message": "配置已保存", + "note": "配置更改将在插件重新加载后生效" + } + + except HTTPException: + raise + except Exception as e: + logger.error(f"更新插件配置失败: {e}", exc_info=True) + raise HTTPException(status_code=500, detail=f"服务器错误: {str(e)}") from e + + +@router.post("/config/{plugin_id}/reset") +async def reset_plugin_config( + plugin_id: str, authorization: Optional[str] = Header(None) +) -> Dict[str, Any]: + """ + 重置插件配置为默认值 + + 删除当前配置文件，下次加载插件时将使用默认配置。 + """ + # Token 验证 + token = authorization.replace("Bearer ", "") if authorization else None + token_manager = get_token_manager() + if not token or not token_manager.verify_token(token): + raise HTTPException(status_code=401, detail="未授权：无效的访问令牌") + + logger.info(f"重置插件配置: {plugin_id}") + + try: + # 查找插件目录 + plugins_dir = Path("plugins") + plugin_path = None + + for p in plugins_dir.iterdir(): + if p.is_dir(): + manifest_path = p / "_manifest.json" + if manifest_path.exists(): + try: + with open(manifest_path, "r", encoding="utf-8") as f: + manifest = json.load(f) + if manifest.get("id") == plugin_id or p.name == plugin_id: + plugin_path = p + break + except Exception: + continue + + if not plugin_path: + raise HTTPException(status_code=404, detail=f"未找到插件: {plugin_id}") + + config_path = plugin_path / "config.toml" + + if not config_path.exists(): + return {"success": True, "message": "配置文件不存在，无需重置"} + + # 备份并删除 + import shutil + import datetime + backup_name = f"config.toml.reset.{datetime.datetime.now().strftime('%Y%m%d%H%M%S')}" + backup_path = plugin_path / backup_name + shutil.move(config_path, backup_path) + + logger.info(f"已重置插件配置: {plugin_id}，备份: {backup_path}") + + return { + "success": True, + "message": "配置已重置，下次加载插件时将使用默认配置", + "backup": str(backup_path) + } + + except HTTPException: + raise + except Exception as e: + logger.error(f"重置插件配置失败: {e}", exc_info=True) + raise HTTPException(status_code=500, detail=f"服务器错误: {str(e)}") from e + + +@router.post("/config/{plugin_id}/toggle") +async def toggle_plugin( + plugin_id: str, authorization: Optional[str] = Header(None) +) -> Dict[str, Any]: + """ + 切换插件启用状态 + + 切换插件配置中的 enabled 字段。 + """ + # Token 验证 + token = authorization.replace("Bearer ", "") if authorization else None + token_manager = get_token_manager() + if not token or not token_manager.verify_token(token): + raise HTTPException(status_code=401, detail="未授权：无效的访问令牌") + + logger.info(f"切换插件状态: {plugin_id}") + + try: + # 查找插件目录 + plugins_dir = Path("plugins") + plugin_path = None + + for p in plugins_dir.iterdir(): + if p.is_dir(): + manifest_path = p / "_manifest.json" + if manifest_path.exists(): + try: + with open(manifest_path, "r", encoding="utf-8") as f: + manifest = json.load(f) + if manifest.get("id") == plugin_id or p.name == plugin_id: + plugin_path = p + break + except Exception: + continue + + if not plugin_path: + raise HTTPException(status_code=404, detail=f"未找到插件: {plugin_id}") + + config_path = plugin_path / "config.toml" + + import toml + + # 读取当前配置 + config = {} + if config_path.exists(): + config = toml.load(config_path) + + # 切换 enabled 状态 + if "plugin" not in config: + config["plugin"] = {} + + current_enabled = config["plugin"].get("enabled", True) + new_enabled = not current_enabled + config["plugin"]["enabled"] = new_enabled + + # 写入配置 + with open(config_path, "w", encoding="utf-8") as f: + toml.dump(config, f) + + status = "启用" if new_enabled else "禁用" + logger.info(f"已{status}插件: {plugin_id}") + + return { + "success": True, + "enabled": new_enabled, + "message": f"插件已{status}", + "note": "状态更改将在下次加载插件时生效" + } + + except HTTPException: + raise + except Exception as e: + logger.error(f"切换插件状态失败: {e}", exc_info=True) + raise HTTPException(status_code=500, detail=f"服务器错误: {str(e)}") from e