diff --git a/src/webui/config_routes.py b/src/webui/config_routes.py new file mode 100644 index 00000000..41784d4e --- /dev/null +++ b/src/webui/config_routes.py @@ -0,0 +1,312 @@ +""" +配置管理API路由 +""" + +import os +import tomlkit +from fastapi import APIRouter, HTTPException, Body +from typing import Any + +from src.common.logger import get_logger +from src.config.config import Config, APIAdapterConfig, CONFIG_DIR +from src.config.official_configs import ( + BotConfig, + PersonalityConfig, + RelationshipConfig, + ChatConfig, + MessageReceiveConfig, + EmojiConfig, + ExpressionConfig, + KeywordReactionConfig, + ChineseTypoConfig, + ResponsePostProcessConfig, + ResponseSplitterConfig, + TelemetryConfig, + ExperimentalConfig, + MaimMessageConfig, + LPMMKnowledgeConfig, + ToolConfig, + MemoryConfig, + DebugConfig, + MoodConfig, + VoiceConfig, + JargonConfig, +) +from src.config.api_ada_configs import ( + ModelTaskConfig, + ModelInfo, + APIProvider, +) +from src.webui.config_schema import ConfigSchemaGenerator + +logger = get_logger("webui.config_routes") + +router = APIRouter(prefix="/config", tags=["config"]) + + +# ===== 架构获取接口 ===== + + +@router.get("/schema/bot") +async def get_bot_config_schema(): + """获取麦麦主程序配置架构""" + try: + # Config 类包含所有子配置 + schema = ConfigSchemaGenerator.generate_config_schema(Config) + return {"success": True, "schema": schema} + except Exception as e: + logger.error(f"获取配置架构失败: {e}") + raise HTTPException(status_code=500, detail=f"获取配置架构失败: {str(e)}") + + +@router.get("/schema/model") +async def get_model_config_schema(): + """获取模型配置架构(包含提供商和模型任务配置)""" + try: + schema = ConfigSchemaGenerator.generate_config_schema(APIAdapterConfig) + return {"success": True, "schema": schema} + except Exception as e: + logger.error(f"获取模型配置架构失败: {e}") + raise HTTPException(status_code=500, detail=f"获取模型配置架构失败: {str(e)}") + + +# ===== 子配置架构获取接口 ===== + + +@router.get("/schema/section/{section_name}") +async def get_config_section_schema(section_name: str): + """ + 获取指定配置节的架构 + + 支持的section_name: + - bot: BotConfig + - personality: PersonalityConfig + - relationship: RelationshipConfig + - chat: ChatConfig + - message_receive: MessageReceiveConfig + - emoji: EmojiConfig + - expression: ExpressionConfig + - keyword_reaction: KeywordReactionConfig + - chinese_typo: ChineseTypoConfig + - response_post_process: ResponsePostProcessConfig + - response_splitter: ResponseSplitterConfig + - telemetry: TelemetryConfig + - experimental: ExperimentalConfig + - maim_message: MaimMessageConfig + - lpmm_knowledge: LPMMKnowledgeConfig + - tool: ToolConfig + - memory: MemoryConfig + - debug: DebugConfig + - mood: MoodConfig + - voice: VoiceConfig + - jargon: JargonConfig + - model_task_config: ModelTaskConfig + - api_provider: APIProvider + - model_info: ModelInfo + """ + section_map = { + "bot": BotConfig, + "personality": PersonalityConfig, + "relationship": RelationshipConfig, + "chat": ChatConfig, + "message_receive": MessageReceiveConfig, + "emoji": EmojiConfig, + "expression": ExpressionConfig, + "keyword_reaction": KeywordReactionConfig, + "chinese_typo": ChineseTypoConfig, + "response_post_process": ResponsePostProcessConfig, + "response_splitter": ResponseSplitterConfig, + "telemetry": TelemetryConfig, + "experimental": ExperimentalConfig, + "maim_message": MaimMessageConfig, + "lpmm_knowledge": LPMMKnowledgeConfig, + "tool": ToolConfig, + "memory": MemoryConfig, + "debug": DebugConfig, + "mood": MoodConfig, + "voice": VoiceConfig, + "jargon": JargonConfig, + "model_task_config": ModelTaskConfig, + "api_provider": APIProvider, + "model_info": ModelInfo, + } + + if section_name not in section_map: + raise HTTPException(status_code=404, detail=f"配置节 '{section_name}' 不存在") + + try: + config_class = section_map[section_name] + schema = ConfigSchemaGenerator.generate_schema(config_class, include_nested=False) + return {"success": True, "schema": schema} + except Exception as e: + logger.error(f"获取配置节架构失败: {e}") + raise HTTPException(status_code=500, detail=f"获取配置节架构失败: {str(e)}") + + +# ===== 配置读取接口 ===== + + +@router.get("/bot") +async def get_bot_config(): + """获取麦麦主程序配置""" + try: + config_path = os.path.join(CONFIG_DIR, "bot_config.toml") + if not os.path.exists(config_path): + raise HTTPException(status_code=404, detail="配置文件不存在") + + with open(config_path, "r", encoding="utf-8") as f: + config_data = tomlkit.load(f) + + return {"success": True, "config": config_data} + except HTTPException: + raise + except Exception as e: + logger.error(f"读取配置文件失败: {e}") + raise HTTPException(status_code=500, detail=f"读取配置文件失败: {str(e)}") + + +@router.get("/model") +async def get_model_config(): + """获取模型配置(包含提供商和模型任务配置)""" + try: + config_path = os.path.join(CONFIG_DIR, "model_config.toml") + if not os.path.exists(config_path): + raise HTTPException(status_code=404, detail="配置文件不存在") + + with open(config_path, "r", encoding="utf-8") as f: + config_data = tomlkit.load(f) + + return {"success": True, "config": config_data} + except HTTPException: + raise + except Exception as e: + logger.error(f"读取配置文件失败: {e}") + raise HTTPException(status_code=500, detail=f"读取配置文件失败: {str(e)}") + + +# ===== 配置更新接口 ===== + + +@router.post("/bot") +async def update_bot_config(config_data: dict[str, Any] = Body(...)): + """更新麦麦主程序配置""" + try: + # 验证配置数据 + try: + Config.from_dict(config_data) + except Exception as e: + raise HTTPException(status_code=400, detail=f"配置数据验证失败: {str(e)}") + + # 保存配置文件 + config_path = os.path.join(CONFIG_DIR, "bot_config.toml") + with open(config_path, "w", encoding="utf-8") as f: + tomlkit.dump(config_data, f) + + logger.info("麦麦主程序配置已更新") + return {"success": True, "message": "配置已保存"} + except HTTPException: + raise + except Exception as e: + logger.error(f"保存配置文件失败: {e}") + raise HTTPException(status_code=500, detail=f"保存配置文件失败: {str(e)}") + + +@router.post("/model") +async def update_model_config(config_data: dict[str, Any] = Body(...)): + """更新模型配置""" + try: + # 验证配置数据 + try: + APIAdapterConfig.from_dict(config_data) + except Exception as e: + raise HTTPException(status_code=400, detail=f"配置数据验证失败: {str(e)}") + + # 保存配置文件 + config_path = os.path.join(CONFIG_DIR, "model_config.toml") + with open(config_path, "w", encoding="utf-8") as f: + tomlkit.dump(config_data, f) + + logger.info("模型配置已更新") + return {"success": True, "message": "配置已保存"} + except HTTPException: + raise + except Exception as e: + logger.error(f"保存配置文件失败: {e}") + raise HTTPException(status_code=500, detail=f"保存配置文件失败: {str(e)}") + + +# ===== 配置节更新接口 ===== + + +@router.post("/bot/section/{section_name}") +async def update_bot_config_section(section_name: str, section_data: dict[str, Any] = Body(...)): + """更新麦麦主程序配置的指定节""" + try: + # 读取现有配置 + config_path = os.path.join(CONFIG_DIR, "bot_config.toml") + if not os.path.exists(config_path): + raise HTTPException(status_code=404, detail="配置文件不存在") + + with open(config_path, "r", encoding="utf-8") as f: + config_data = tomlkit.load(f) + + # 更新指定节 + if section_name not in config_data: + raise HTTPException(status_code=404, detail=f"配置节 '{section_name}' 不存在") + + config_data[section_name] = section_data + + # 验证完整配置 + try: + Config.from_dict(config_data) + except Exception as e: + raise HTTPException(status_code=400, detail=f"配置数据验证失败: {str(e)}") + + # 保存配置 + with open(config_path, "w", encoding="utf-8") as f: + tomlkit.dump(config_data, f) + + logger.info(f"配置节 '{section_name}' 已更新") + return {"success": True, "message": f"配置节 '{section_name}' 已保存"} + except HTTPException: + raise + except Exception as e: + logger.error(f"更新配置节失败: {e}") + raise HTTPException(status_code=500, detail=f"更新配置节失败: {str(e)}") + + +@router.post("/model/section/{section_name}") +async def update_model_config_section(section_name: str, section_data: dict[str, Any] = Body(...)): + """更新模型配置的指定节""" + try: + # 读取现有配置 + config_path = os.path.join(CONFIG_DIR, "model_config.toml") + if not os.path.exists(config_path): + raise HTTPException(status_code=404, detail="配置文件不存在") + + with open(config_path, "r", encoding="utf-8") as f: + config_data = tomlkit.load(f) + + # 更新指定节 + if section_name not in config_data: + raise HTTPException(status_code=404, detail=f"配置节 '{section_name}' 不存在") + + config_data[section_name] = section_data + + # 验证完整配置 + try: + APIAdapterConfig.from_dict(config_data) + except Exception as e: + raise HTTPException(status_code=400, detail=f"配置数据验证失败: {str(e)}") + + # 保存配置 + with open(config_path, "w", encoding="utf-8") as f: + tomlkit.dump(config_data, f) + + logger.info(f"配置节 '{section_name}' 已更新") + return {"success": True, "message": f"配置节 '{section_name}' 已保存"} + except HTTPException: + raise + except Exception as e: + logger.error(f"更新配置节失败: {e}") + raise HTTPException(status_code=500, detail=f"更新配置节失败: {str(e)}") diff --git a/src/webui/config_schema.py b/src/webui/config_schema.py new file mode 100644 index 00000000..c1608bc4 --- /dev/null +++ b/src/webui/config_schema.py @@ -0,0 +1,336 @@ +""" +配置架构生成器 - 自动从配置类生成前端表单架构 +""" + +import inspect +from dataclasses import fields, MISSING +from typing import Any, get_origin, get_args, Literal, Optional +from enum import Enum + +from src.config.config_base import ConfigBase + + +class FieldType(str, Enum): + """字段类型枚举""" + + STRING = "string" + NUMBER = "number" + INTEGER = "integer" + BOOLEAN = "boolean" + SELECT = "select" + ARRAY = "array" + OBJECT = "object" + TEXTAREA = "textarea" + + +class FieldSchema: + """字段架构""" + + def __init__( + self, + name: str, + type: FieldType, + label: str, + description: str = "", + default: Any = None, + required: bool = True, + options: Optional[list[str]] = None, + min_value: Optional[float] = None, + max_value: Optional[float] = None, + items: Optional[dict] = None, + properties: Optional[dict] = None, + ): + self.name = name + self.type = type + self.label = label + self.description = description + self.default = default + self.required = required + self.options = options + self.min_value = min_value + self.max_value = max_value + self.items = items + self.properties = properties + + def to_dict(self) -> dict: + """转换为字典""" + result = { + "name": self.name, + "type": self.type.value, + "label": self.label, + "description": self.description, + "required": self.required, + } + + if self.default is not None: + result["default"] = self.default + + if self.options is not None: + result["options"] = self.options + + if self.min_value is not None: + result["minValue"] = self.min_value + + if self.max_value is not None: + result["maxValue"] = self.max_value + + if self.items is not None: + result["items"] = self.items + + if self.properties is not None: + result["properties"] = self.properties + + return result + + +class ConfigSchemaGenerator: + """配置架构生成器""" + + @staticmethod + def _extract_field_description(config_class: type, field_name: str) -> str: + """ + 从类定义中提取字段的文档字符串描述 + + Args: + config_class: 配置类 + field_name: 字段名 + + Returns: + str: 字段描述 + """ + try: + # 获取源代码 + source = inspect.getsource(config_class) + lines = source.split("\n") + + # 查找字段定义 + field_found = False + description_lines = [] + + for i, line in enumerate(lines): + # 匹配字段定义行,例如: platform: str + if f"{field_name}:" in line and "=" in line: + field_found = True + # 查找下一行的文档字符串 + if i + 1 < len(lines): + next_line = lines[i + 1].strip() + if next_line.startswith('"""') or next_line.startswith("'''"): + # 单行文档字符串 + if next_line.count('"""') == 2 or next_line.count("'''") == 2: + description_lines.append(next_line.strip('"""').strip("'''").strip()) + else: + # 多行文档字符串 + quote = '"""' if next_line.startswith('"""') else "'''" + description_lines.append(next_line.strip(quote).strip()) + for j in range(i + 2, len(lines)): + if quote in lines[j]: + description_lines.append(lines[j].split(quote)[0].strip()) + break + description_lines.append(lines[j].strip()) + break + elif f"{field_name}:" in line and "=" not in line: + # 没有默认值的字段 + field_found = True + if i + 1 < len(lines): + next_line = lines[i + 1].strip() + if next_line.startswith('"""') or next_line.startswith("'''"): + if next_line.count('"""') == 2 or next_line.count("'''") == 2: + description_lines.append(next_line.strip('"""').strip("'''").strip()) + else: + quote = '"""' if next_line.startswith('"""') else "'''" + description_lines.append(next_line.strip(quote).strip()) + for j in range(i + 2, len(lines)): + if quote in lines[j]: + description_lines.append(lines[j].split(quote)[0].strip()) + break + description_lines.append(lines[j].strip()) + break + + if field_found and description_lines: + return " ".join(description_lines) + + except Exception: + pass + + return "" + + @staticmethod + def _get_field_type_and_options(field_type: type) -> tuple[FieldType, Optional[list[str]], Optional[dict]]: + """ + 获取字段类型和选项 + + Args: + field_type: 字段类型 + + Returns: + tuple: (FieldType, options, items) + """ + origin = get_origin(field_type) + args = get_args(field_type) + + # 处理 Literal 类型(枚举选项) + if origin is Literal: + return FieldType.SELECT, [str(arg) for arg in args], None + + # 处理 list 类型 + if origin is list: + item_type = args[0] if args else str + if item_type is str: + items = {"type": "string"} + elif item_type is int: + items = {"type": "integer"} + elif item_type is float: + items = {"type": "number"} + elif item_type is bool: + items = {"type": "boolean"} + elif item_type is dict: + items = {"type": "object"} + else: + items = {"type": "string"} + return FieldType.ARRAY, None, items + + # 处理 set 类型(与 list 类似) + if origin is set: + item_type = args[0] if args else str + if item_type is str: + items = {"type": "string"} + else: + items = {"type": "string"} + return FieldType.ARRAY, None, items + + # 处理基本类型 + if field_type is bool or field_type == bool: + return FieldType.BOOLEAN, None, None + elif field_type is int or field_type == int: + return FieldType.INTEGER, None, None + elif field_type is float or field_type == float: + return FieldType.NUMBER, None, None + elif field_type is str or field_type == str: + return FieldType.STRING, None, None + elif field_type is dict or origin is dict: + return FieldType.OBJECT, None, None + + # 默认为字符串 + return FieldType.STRING, None, None + + @staticmethod + def _format_field_name(name: str) -> str: + """ + 格式化字段名为可读的标签 + + Args: + name: 原始字段名 + + Returns: + str: 格式化后的标签 + """ + # 将下划线替换为空格,并首字母大写 + return " ".join(word.capitalize() for word in name.split("_")) + + @staticmethod + def generate_schema(config_class: type[ConfigBase], include_nested: bool = True) -> dict: + """ + 从配置类生成前端表单架构 + + Args: + config_class: 配置类(必须继承自 ConfigBase) + include_nested: 是否包含嵌套的配置对象 + + Returns: + dict: 前端表单架构 + """ + if not issubclass(config_class, ConfigBase): + raise ValueError(f"{config_class.__name__} 必须继承自 ConfigBase") + + schema_fields = [] + nested_schemas = {} + + for field in fields(config_class): + # 跳过私有字段和内部字段 + if field.name.startswith("_") or field.name in ["MMC_VERSION"]: + continue + + # 提取字段描述 + description = ConfigSchemaGenerator._extract_field_description(config_class, field.name) + + # 判断是否必填 + required = field.default is MISSING and field.default_factory is MISSING + + # 获取默认值 + default_value = None + if field.default is not MISSING: + default_value = field.default + elif field.default_factory is not MISSING: + try: + default_value = field.default_factory() + except Exception: + default_value = None + + # 检查是否为嵌套的 ConfigBase + if isinstance(field.type, type) and issubclass(field.type, ConfigBase): + if include_nested: + # 递归生成嵌套配置的架构 + nested_schema = ConfigSchemaGenerator.generate_schema(field.type, include_nested=True) + nested_schemas[field.name] = nested_schema + + field_schema = FieldSchema( + name=field.name, + type=FieldType.OBJECT, + label=ConfigSchemaGenerator._format_field_name(field.name), + description=description or field.type.__doc__ or "", + default=default_value, + required=required, + properties=nested_schema, + ) + else: + continue + else: + # 获取字段类型和选项 + field_type, options, items = ConfigSchemaGenerator._get_field_type_and_options(field.type) + + # 特殊处理:长文本使用 textarea + if field_type == FieldType.STRING and field.name in [ + "personality", + "reply_style", + "interest", + "plan_style", + "visual_style", + "private_plan_style", + "emotion_style", + "reaction", + "filtration_prompt", + ]: + field_type = FieldType.TEXTAREA + + field_schema = FieldSchema( + name=field.name, + type=field_type, + label=ConfigSchemaGenerator._format_field_name(field.name), + description=description, + default=default_value, + required=required, + options=options, + items=items, + ) + + schema_fields.append(field_schema.to_dict()) + + return { + "className": config_class.__name__, + "classDoc": config_class.__doc__ or "", + "fields": schema_fields, + "nested": nested_schemas if nested_schemas else None, + } + + @staticmethod + def generate_config_schema(config_class: type[ConfigBase]) -> dict: + """ + 生成完整的配置架构(包含所有嵌套的子配置) + + Args: + config_class: 配置类 + + Returns: + dict: 完整的配置架构 + """ + return ConfigSchemaGenerator.generate_schema(config_class, include_nested=True) diff --git a/src/webui/routes.py b/src/webui/routes.py index 37f82e5d..64d033e8 100644 --- a/src/webui/routes.py +++ b/src/webui/routes.py @@ -4,12 +4,16 @@ from pydantic import BaseModel, Field from typing import Optional from src.common.logger import get_logger from .token_manager import get_token_manager +from .config_routes import router as config_router logger = get_logger("webui.api") # 创建路由器 router = APIRouter(prefix="/api/webui", tags=["WebUI"]) +# 注册配置管理路由 +router.include_router(config_router) + class TokenVerifyRequest(BaseModel): """Token 验证请求"""