feat: 增强插件配置管理,添加配置 Schema 和更新接口

pull/1389/head
墨梓柒 2025-11-29 00:01:25 +08:00
parent e06a35fe81
commit 321c784434
No known key found for this signature in database
GPG Key ID: 4A65B9DBA35F7635
3 changed files with 767 additions and 10 deletions

View File

@ -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 from dataclasses import dataclass, field
@dataclass @dataclass
class ConfigField: class ConfigField:
"""配置字段定义""" """
配置字段定义
type: type # 字段类型 用于定义插件配置项的元数据支持类型验证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 # 字段类型: str, int, float, bool, list, dict
default: Any # 默认值 default: Any # 默认值
description: str # 字段描述 description: str # 字段描述(也用作默认标签)
example: Optional[str] = None # 示例值
# === 验证相关 ===
example: Optional[str] = None # 示例值(用于生成配置文件注释)
required: bool = False # 是否必需 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
)

View File

@ -12,7 +12,11 @@ from src.plugin_system.base.component_types import (
PluginInfo, PluginInfo,
PythonDependency, 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 from src.plugin_system.utils.manifest_utils import ManifestValidator
logger = get_logger("plugin_base") logger = get_logger("plugin_base")
@ -60,7 +64,10 @@ class PluginBase(ABC):
def config_schema(self) -> Dict[str, Union[Dict[str, ConfigField], str]]: def config_schema(self) -> Dict[str, Union[Dict[str, ConfigField], str]]:
return {} 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): def __init__(self, plugin_dir: str):
"""初始化插件 """初始化插件
@ -564,6 +571,93 @@ class PluginBase(ABC):
return current 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 @abstractmethod
def register_plugin(self) -> bool: def register_plugin(self) -> bool:
""" """

View File

@ -29,8 +29,10 @@ def parse_version(version_str: str) -> tuple[int, int, int]:
Returns: Returns:
(major, minor, patch) 三元组 (major, minor, patch) 三元组
""" """
# 移除 snapshot 等后缀 # 移除 snapshot、dev、alpha、beta 等后缀(支持 - 和 . 分隔符)
base_version = version_str.split(".snapshot")[0].split(".dev")[0].split(".alpha")[0].split(".beta")[0] 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(".") parts = base_version.split(".")
if len(parts) < 3: if len(parts) < 3:
@ -1153,3 +1155,412 @@ async def get_installed_plugins(authorization: Optional[str] = Header(None)) ->
except Exception as e: except Exception as e:
logger.error(f"获取已安装插件列表失败: {e}", exc_info=True) logger.error(f"获取已安装插件列表失败: {e}", exc_info=True)
raise HTTPException(status_code=500, detail=f"服务器错误: {str(e)}") from e 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