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("配置值必须是布尔类型")