buff2.0/backend-admin/app/core/config_cache.py

"""
配置缓存服务 - 管理端专用
用于缓存 system_config 表的配置信息，支持定时刷新
"""
from typing import Dict, Any, Optional
from app.core.logger import logger
from app.models.system_config import SystemConfig

class ConfigCache:
    """配置缓存单例"""

    def __init__(self):
        self._configs: Dict[str, Any] = {}
        self._last_refresh: Optional[str] = None
        self._refresh_interval_seconds: int = 60  # 默认60秒

    def load_configs(self) -> Dict[str, Any]:
        """从数据库加载所有配置"""
        try:
            self._configs = SystemConfig.get_all_configs()
            from datetime import datetime
            self._last_refresh = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            logger.info(f"配置加载成功，共 {len(self._configs)} 项配置")
            logger.debug(f"配置内容: {self._configs}")
            return self._configs
        except Exception as e:
            logger.exception("配置加载失败")
            return self._configs

    def get(self, key: str, default: Any = None) -> Any:
        """获取配置值（从缓存）"""
        return self._configs.get(key, default)

    def get_all(self) -> Dict[str, Any]:
        """获取所有配置"""
        return self._configs

    def set_refresh_interval(self, seconds: int):
        """设置刷新间隔（秒）"""
        self._refresh_interval_seconds = max(1, seconds)
        logger.info(f"配置刷新间隔设置为 {self._refresh_interval_seconds} 秒")

    def get_refresh_interval(self) -> int:
        """获取刷新间隔（秒）"""
        return self._refresh_interval_seconds

    def get_last_refresh_time(self) -> Optional[str]:
        """获取上次刷新时间"""
        return self._last_refresh

    def refresh(self) -> bool:
        """刷新配置缓存"""
        try:
            old_configs = self._configs.copy()
            self.load_configs()

            # 检查配置变化
            new_keys = set(self._configs.keys()) - set(old_configs.keys())
            removed_keys = set(old_configs.keys()) - set(self._configs.keys())
            changed_keys = []

            for key in set(self._configs.keys()) & set(old_configs.keys()):
                if self._configs[key] != old_configs[key]:
                    changed_keys.append(key)

            if new_keys or removed_keys or changed_keys:
                logger.info(f"配置已更新:")
                if new_keys:
                    logger.info(f"  新增配置: {new_keys}")
                if removed_keys:
                    logger.info(f"  移除配置: {removed_keys}")
                if changed_keys:
                    logger.info(f"  变更配置: {changed_keys}")

            return True
        except Exception as e:
            logger.exception("配置刷新失败")
            return False

# 全局配置缓存实例
config_cache = ConfigCache()


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


def get_all_configs() -> Dict[str, Any]:
    """便捷函数：获取所有配置"""
    return config_cache.get_all()


def load_all_configs() -> Dict[str, Any]:
    """便捷函数：加载所有配置"""
    return config_cache.load_configs()