""" 配置缓存服务 - 管理端专用 用于缓存 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()