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

"""
系统配置管理器

提供全局配置读取功能，支持缓存和默认值回退
业务代码通过此模块读取数据库中的系统配置
"""
import json
import logging
from typing import Any, Optional, Dict
from datetime import datetime, timedelta
from threading import Lock

from sqlalchemy.orm import Session
from app.database import SessionLocal
from app.models.config import SystemConfig

logger = logging.getLogger(__name__)


class SystemConfigManager:
    """系统配置管理器（单例模式）"""
    
    _instance = None
    _lock = Lock()
    
    # 配置缓存
    _cache: Dict[str, Any] = {}
    _cache_time: Dict[str, datetime] = {}
    _cache_ttl = timedelta(minutes=5)  # 缓存5分钟
    
    # 默认值定义
    DEFAULT_VALUES = {
        # 基础配置
        "system_name": "智创空间",
        "system_logo": "",
        "contact_email": "",
        "icp_number": "",
        # 功能开关
        "enable_registration": True,
        "enable_search": True,
        "enable_verification_reminder": True,
        "enable_local_models": True,
        "enable_openclaw": True,
        "enable_openclaw_client": True,
        "enable_openclaw_web": True,
        "openclaw_client_url": "",
        # 限制配置
        "max_tokens_per_request": 4000,
        "max_images_per_request": 4,
        "max_audio_chars": 5000,
        "max_video_duration": 10,
        # 第三方配置（敏感，默认为空）
        "dashscope_api_key": "",
        "alipay_app_id": "",
        "oss_bucket": "",
        "oss_endpoint": "",
    }
    
    def __new__(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
        return cls._instance
    
    def _get_db(self) -> Session:
        """获取数据库会话"""
        return SessionLocal()
    
    def _parse_value(self, value_str: str, value_type: str) -> Any:
        """解析配置值"""
        try:
            return json.loads(value_str)
        except (json.JSONDecodeError, TypeError):
            return value_str
    
    def get(self, key: str, default: Any = None) -> Any:
        """
        获取配置值

        优先级：Redis缓存 > 本地内存缓存 > 数据库 > 默认值
        Redis 不可用时自动降级到内存缓存。
        """
        # 1. 检查本地内存缓存（最快）
        if key in self._cache:
            cache_time = self._cache_time.get(key)
            if cache_time and datetime.now() - cache_time < self._cache_ttl:
                return self._cache[key]

        # 2. 检查 Redis 缓存
        try:
            from app.core.redis import redis_manager
            r = redis_manager.get_sync_client()
            if r:
                raw = r.get(f"sysconfig:{key}")
                if raw:
                    import json as _json
                    value = _json.loads(raw)
                    # 同步到本地内存缓存
                    self._cache[key] = value
                    self._cache_time[key] = datetime.now()
                    return value
        except Exception:
            pass

        # 3. 从数据库读取
        db = self._get_db()
        try:
            config = db.query(SystemConfig).filter(
                SystemConfig.config_key == key
            ).first()

            if config:
                value = self._parse_value(config.config_value, config.config_type)
                # 写入本地内存缓存
                self._cache[key] = value
                self._cache_time[key] = datetime.now()
                # 写入 Redis 缓存（TTL 5 分钟）
                try:
                    from app.core.redis import redis_manager
                    import json as _json
                    r = redis_manager.get_sync_client()
                    if r:
                        r.setex(f"sysconfig:{key}", 300, _json.dumps(value))
                except Exception:
                    pass
                return value
        except Exception as e:
            logger.warning(f"读取配置 {key} 失败: {e}")
        finally:
            db.close()

        # 4. 返回默认值
        if default is not None:
            return default
        return self.DEFAULT_VALUES.get(key)
    
    def get_string(self, key: str, default: str = "") -> str:
        """获取字符串配置"""
        value = self.get(key, default)
        return str(value) if value is not None else default
    
    def get_int(self, key: str, default: int = 0) -> int:
        """获取整数配置"""
        value = self.get(key, default)
        try:
            return int(value)
        except (ValueError, TypeError):
            return default
    
    def get_float(self, key: str, default: float = 0.0) -> float:
        """获取浮点数配置"""
        value = self.get(key, default)
        try:
            return float(value)
        except (ValueError, TypeError):
            return default
    
    def get_bool(self, key: str, default: bool = False) -> bool:
        """获取布尔配置"""
        value = self.get(key, default)
        if isinstance(value, bool):
            return value
        if isinstance(value, str):
            return value.lower() in ('true', '1', 'yes')
        return bool(value)
    
    def clear_cache(self, key: Optional[str] = None):
        """清除缓存（内存 + Redis）"""
        if key:
            self._cache.pop(key, None)
            self._cache_time.pop(key, None)
            try:
                from app.core.redis import redis_manager
                r = redis_manager.get_sync_client()
                if r:
                    r.delete(f"sysconfig:{key}")
            except Exception:
                pass
        else:
            self._cache.clear()
            self._cache_time.clear()
            try:
                from app.core.redis import redis_manager
                r = redis_manager.get_sync_client()
                if r:
                    # 删除所有 sysconfig: 前缀的 key
                    keys = r.keys("sysconfig:*")
                    if keys:
                        r.delete(*keys)
            except Exception:
                pass
    
    def refresh(self, key: str) -> Any:
        """强制刷新配置"""
        self.clear_cache(key)
        return self.get(key)


# 全局单例实例
config_manager = SystemConfigManager()


# 便捷函数
def get_config(key: str, default: Any = None) -> Any:
    """获取配置值"""
    return config_manager.get(key, default)

def get_config_string(key: str, default: str = "") -> str:
    """获取字符串配置"""
    return config_manager.get_string(key, default)

def get_config_int(key: str, default: int = 0) -> int:
    """获取整数配置"""
    return config_manager.get_int(key, default)

def get_config_float(key: str, default: float = 0.0) -> float:
    """获取浮点数配置"""
    return config_manager.get_float(key, default)

def get_config_bool(key: str, default: bool = False) -> bool:
    """获取布尔配置"""
    return config_manager.get_bool(key, default)

def clear_config_cache(key: Optional[str] = None):
    """清除配置缓存"""
    config_manager.clear_cache(key)