Maas-Model-platform/phone/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_recharge": True,
        "enable_search": True,
        "enable_local_models": True,
        "enable_openclaw": True,
        "enable_openclaw_client": True,
        "enable_openclaw_web": True,
        "openclaw_client_url": "",
        "new_user_bonus": 10.0,
        # 限制配置
        "max_tokens_per_request": 4000,
        "max_images_per_request": 4,
        "max_audio_chars": 5000,
        "max_video_duration": 10,
        "min_balance_required": 0.01,
        # 第三方配置（敏感，默认为空）
        "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:
        """
        获取配置值
        
        优先级：缓存 > 数据库 > 默认值 > 传入的default
        """
        # 检查缓存
        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]
        
        # 从数据库读取
        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()
                return value
        except Exception as e:
            logger.warning(f"读取配置 {key} 失败: {e}")
        finally:
            db.close()
        
        # 返回默认值
        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):
        """清除缓存"""
        if key:
            self._cache.pop(key, None)
            self._cache_time.pop(key, None)
        else:
            self._cache.clear()
            self._cache_time.clear()
    
    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)


def get_tenant_config_bool(db, tenant_id: Optional[int], key: str, default: bool = False) -> bool:
    """
    读取租户级配置（bool），优先租户自己的配置，没有则 fallback 到全局配置，再没有则用默认值。
    """
    from app.models.config import SystemConfig
    import json as _json

    if tenant_id is not None:
        row = db.query(SystemConfig).filter(
            SystemConfig.tenant_id == tenant_id,
            SystemConfig.config_key == key,
        ).first()
        if row:
            try:
                v = _json.loads(row.config_value)
                if isinstance(v, bool):
                    return v
                return str(v).lower() in ('true', '1', 'yes')
            except Exception:
                pass

    # fallback 到全局配置
    return get_config_bool(key, default)


def get_tenant_config_float(db, tenant_id: Optional[int], key: str, default: float = 0.0) -> float:
    """
    读取租户级配置（float），优先租户自己的配置，没有则 fallback 到全局配置，再没有则用默认值。
    """
    from app.models.config import SystemConfig
    import json as _json

    if tenant_id is not None:
        row = db.query(SystemConfig).filter(
            SystemConfig.tenant_id == tenant_id,
            SystemConfig.config_key == key,
        ).first()
        if row:
            try:
                return float(_json.loads(row.config_value))
            except Exception:
                pass

    return get_config_float(key, default)