Maas-Model-platform/phone/config_service.py

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