feat: 添加配置管理API路由和配置架构

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)}")

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)

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 验证请求"""