Maas-Model-platform/backend/app/services/config_service.py

from sqlalchemy.orm import Session
from sqlalchemy import select, update
from typing import Dict, List, Optional, Any
import json
from app.models.config import SystemConfig, ConfigHistory
from app.schemas.config_schema import ConfigItem, ConfigUpdate
from app.services.system_config_manager import clear_config_cache

class ConfigService:
    def __init__(self, db: Session):
        self.db = db
    
    def get_all_configs(self) -> Dict[str, List[ConfigItem]]:
        """获取所有配置项，按分类分组"""
        configs = self.db.query(SystemConfig).all()
        
        grouped = {}
        for config in configs:
            category = config.category
            if category not in grouped:
                grouped[category] = []
            
            value = self._parse_value(config.config_value, config.config_type)
            grouped[category].append(ConfigItem(
                key=config.config_key,
                value=value,
                type=config.config_type,
                category=config.category,
                description=config.description,
                updated_at=config.updated_at
            ))
        
        return grouped
    
    def get_config(self, key: str) -> Optional[ConfigItem]:
        """获取单个配置项"""
        config = self.db.query(SystemConfig).filter(
            SystemConfig.config_key == key
        ).first()
        
        if not config:
            return None
        
        value = self._parse_value(config.config_value, config.config_type)
        return ConfigItem(
            key=config.config_key,
            value=value,
            type=config.config_type,
            category=config.category,
            description=config.description,
            updated_at=config.updated_at
        )
    
    def update_configs(self, configs: List[ConfigUpdate], admin_id: int):
        """批量更新配置项"""
        for config_update in configs:
            self._update_single_config(
                config_update.key,
                config_update.value,
                admin_id
            )
        
        self.db.commit()
        # 清除所有配置缓存
        clear_config_cache()
    
    def _update_single_config(self, key: str, value: Any, admin_id: int):
        """更新单个配置项"""
        config = self.db.query(SystemConfig).filter(
            SystemConfig.config_key == key
        ).first()
        
        if not config:
            # 自动创建配置项
            from app.services.system_config_manager import SystemConfigManager
            config_manager = SystemConfigManager()
            default_value = config_manager.DEFAULT_VALUES.get(key)
            
            if default_value is None:
                raise ValueError(f"配置项 {key} 不存在且没有默认值")
            
            # 确定配置类型
            if isinstance(default_value, bool):
                config_type = "boolean"
            elif isinstance(default_value, (int, float)):
                config_type = "number"
            elif isinstance(default_value, str):
                config_type = "string"
            else:
                config_type = "json"
            
            # 创建配置项
            config = SystemConfig(
                config_key=key,
                config_value=self._serialize_value(default_value, config_type),
                config_type=config_type,
                category="系统设置",
                description="",
                updated_by=admin_id
            )
            self.db.add(config)
            old_value = ""
        else:
            self._validate_value(value, config.config_type)
            old_value = config.config_value
        
        new_value = self._serialize_value(value, config.config_type)
        
        history = ConfigHistory(
            config_key=key,
            old_value=old_value,
            new_value=new_value,
            updated_by=admin_id
        )
        self.db.add(history)
        
        config.config_value = new_value
        config.updated_by = admin_id
    
    def get_config_history(self, key: str, page: int = 1, size: int = 20):
        """获取配置修改历史"""
        offset = (page - 1) * size
        
        history_list = self.db.query(ConfigHistory).filter(
            ConfigHistory.config_key == key
        ).order_by(ConfigHistory.updated_at.desc()).offset(offset).limit(size).all()
        
        total = self.db.query(ConfigHistory).filter(
            ConfigHistory.config_key == key
        ).count()
        
        return {
            "items": history_list,
            "total": total,
            "page": page,
            "size": size
        }
    
    def reset_config(self, key: str, admin_id: int):
        """重置配置为默认值"""
        default_values = {
            "system_name": "智创空间",
            "enable_registration": True,
            "enable_search": True,
            "enable_local_models": True,
            "max_tokens_per_request": 4000,
            "max_images_per_request": 4,
            "max_audio_chars": 5000,
            "max_video_duration": 10
        }
        
        if key not in default_values:
            raise ValueError(f"配置项 {key} 没有默认值")
        
        self._update_single_config(key, default_values[key], admin_id)
        self.db.commit()
    
    def _parse_value(self, value_str: str, value_type: str) -> Any:
        """解析配置值，对空值容错"""
        if not value_str or value_str.strip() == "":
            if value_type == "string":
                return ""
            elif value_type == "number":
                return 0
            elif value_type == "boolean":
                return False
            return ""
        
        if value_type == "string":
            # 字符串类型：如果是 JSON 格式的字符串（带引号），解析它；否则直接返回原值
            try:
                return json.loads(value_str)
            except (json.JSONDecodeError, ValueError):
                return value_str
        elif value_type == "number":
            try:
                return json.loads(value_str)
            except (json.JSONDecodeError, ValueError):
                return 0
        elif value_type == "boolean":
            if value_str.lower() in ("true", "1", "yes"):
                return True
            elif value_str.lower() in ("false", "0", "no"):
                return False
            try:
                return json.loads(value_str)
            except (json.JSONDecodeError, ValueError):
                return False
        elif value_type == "json":
            try:
                return json.loads(value_str)
            except (json.JSONDecodeError, ValueError):
                return {}
        return value_str
    
    def _serialize_value(self, value: Any, value_type: str) -> str:
        """序列化配置值"""
        if value_type == "boolean" and isinstance(value, str):
            value = value.lower() in ("true", "1")
        elif value_type == "number" and isinstance(value, str):
            value = float(value) if '.' in value else int(value)
        return json.dumps(value, ensure_ascii=False)
    
    def _validate_value(self, value: Any, value_type: str):
        """验证配置值类型"""
        if value_type == "string" and not isinstance(value, str):
            raise ValueError("配置值必须是字符串类型")
        elif value_type == "number":
            if isinstance(value, str):
                try:
                    float(value)
                except ValueError:
                    raise ValueError("配置值必须是数字类型")
            elif not isinstance(value, (int, float)):
                raise ValueError("配置值必须是数字类型")
        elif value_type == "boolean":
            if isinstance(value, str):
                if value.lower() not in ("true", "false", "1", "0"):
                    raise ValueError("配置值必须是布尔类型")
            elif not isinstance(value, bool):
                raise ValueError("配置值必须是布尔类型")