from sqlalchemy.orm import Session 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, tenant_id: Optional[int] = None): self.db = db self.tenant_id = tenant_id def _base_query(self): return self.db.query(SystemConfig).filter( SystemConfig.tenant_id == self.tenant_id ) _KEY_CATEGORY = { "system_name": "basic", "system_logo": "basic", "contact_email": "basic", "icp_number": "basic", "enable_registration": "feature", "enable_recharge": "feature", "enable_search": "feature", "enable_local_models": "feature", "new_user_bonus": "feature", "max_tokens_per_request": "limit", "max_images_per_request": "limit", "max_audio_chars": "limit", "max_video_duration": "limit", "min_balance_required": "limit", } # 配置项中文描述 _KEY_DESCRIPTION = { "system_name": "平台名称，显示在登录页和导航栏", "system_logo": "平台 Logo 图片 URL", "contact_email": "联系邮箱", "icp_number": "ICP 备案号", "enable_registration": "是否开放用户注册", "enable_recharge": "是否开放充值功能", "enable_search": "是否开放联网搜索功能", "enable_local_models": "是否允许用户使用本地模型", "new_user_bonus": "新用户注册后自动赠送的余额（元）", "max_tokens_per_request": "单次请求最大 Token 数", "max_images_per_request": "单次请求最多生成图片数", "max_audio_chars": "单次音频请求最大字符数", "max_video_duration": "单次视频请求最大时长（秒）", "min_balance_required": "发起请求所需最低余额（元）", } # 需要在前端展示的默认配置项（即使数据库中没有记录也要显示） _DEFAULT_DISPLAY_CONFIGS = [ {"key": "enable_registration", "type": "boolean", "category": "feature", "default": True}, {"key": "enable_recharge", "type": "boolean", "category": "feature", "default": True}, {"key": "enable_search", "type": "boolean", "category": "feature", "default": True}, {"key": "enable_local_models", "type": "boolean", "category": "feature", "default": True}, {"key": "new_user_bonus", "type": "number", "category": "feature", "default": 10.0}, {"key": "max_tokens_per_request", "type": "number", "category": "limit", "default": 4000}, {"key": "max_images_per_request", "type": "number", "category": "limit", "default": 4}, {"key": "max_audio_chars", "type": "number", "category": "limit", "default": 5000}, {"key": "max_video_duration", "type": "number", "category": "limit", "default": 10}, {"key": "min_balance_required", "type": "number", "category": "limit", "default": 0.01}, ] def _get_category(self, key: str) -> str: return self._KEY_CATEGORY.get(key, "basic") def get_all_configs(self) -> Dict[str, List[ConfigItem]]: configs = self._base_query().all() # 先把数据库中已有的配置按 key 索引 db_map = {c.config_key: c for c in configs} grouped: Dict[str, List[ConfigItem]] = {} # 1. 先填入数据库中已有的记录 for config in configs: category = config.category if category not in grouped: grouped[category] = [] grouped[category].append(ConfigItem( key=config.config_key, value=self._parse_value(config.config_value, config.config_type), type=config.config_type, category=config.category, description=config.description or self._KEY_DESCRIPTION.get(config.config_key, ""), updated_at=config.updated_at, )) # 2. 补充数据库中没有记录的默认配置项，确保前端始终能看到这些配置 for meta in self._DEFAULT_DISPLAY_CONFIGS: key = meta["key"] if key in db_map: continue # 已有记录，跳过 category = meta["category"] if category not in grouped: grouped[category] = [] grouped[category].append(ConfigItem( key=key, value=meta["default"], type=meta["type"], category=category, description=self._KEY_DESCRIPTION.get(key, ""), updated_at=None, )) return grouped def get_config(self, key: str) -> Optional[ConfigItem]: config = self._base_query().filter(SystemConfig.config_key == key).first() if not config: return None return ConfigItem( key=config.config_key, value=self._parse_value(config.config_value, config.config_type), type=config.config_type, category=config.category, description=config.description, updated_at=config.updated_at, ) def update_configs(self, configs: List[ConfigUpdate], admin_id: Optional[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: Optional[int]): config = self._base_query().filter(SystemConfig.config_key == key).first() if not config: from app.services.system_config_manager import SystemConfigManager default_value = SystemConfigManager().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( tenant_id=self.tenant_id, config_key=key, config_value=self._serialize_value(default_value, config_type), config_type=config_type, category=self._get_category(key), 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( tenant_id=self.tenant_id, 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 q = self.db.query(ConfigHistory).filter( ConfigHistory.tenant_id == self.tenant_id, ConfigHistory.config_key == key, ) return { "items": q.order_by(ConfigHistory.updated_at.desc()).offset(offset).limit(size).all(), "total": q.count(), "page": page, "size": size, } def reset_config(self, key: str, admin_id: Optional[int]): default_values = { "system_name": "智创空间", "enable_registration": True, "enable_recharge": True, "enable_search": True, "enable_local_models": True, "new_user_bonus": 10, "max_tokens_per_request": 4000, "max_images_per_request": 4, "max_audio_chars": 5000, "max_video_duration": 10, "min_balance_required": 0.01, } 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: return json.loads(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("配置值必须是布尔类型")