""" 配置管理服务 提供统一的系统配置访问接口，支持缓存和热更新 """ import threading import time from typing import Any, Optional, Dict from app.models.system import SystemConfig from app.database.db_manager import db import logging logger = logging.getLogger(__name__) class ConfigService: """ 系统配置管理服务 特性: - 缓存配置数据，减少数据库查询 - 支持热更新，配置变更时自动刷新缓存 - 类型安全的配置值获取 - 线程安全 """ _instance = None _lock = threading.Lock() def __new__(cls): if cls._instance is None: with cls._lock: if cls._instance is None: cls._instance = super().__new__(cls) cls._instance._initialized = False return cls._instance def __init__(self): if not self._initialized: self._config_cache: Dict[str, Any] = {} self._cache_timestamp = 0 self._cache_ttl = 300 # 缓存5分钟 self._cache_lock = threading.RLock() self._initialized = True logger.info("配置服务已初始化") def _is_cache_valid(self) -> bool: """检查缓存是否有效""" return (time.time() - self._cache_timestamp) < self._cache_ttl def _refresh_cache(self) -> None: """刷新配置缓存""" try: # 检查是否有Flask应用上下文 try: from flask import has_app_context if not has_app_context(): logger.warning("没有Flask应用上下文，跳过配置缓存刷新") return except ImportError: logger.warning("Flask未加载，跳过配置缓存刷新") return with self._cache_lock: logger.debug("开始刷新配置缓存...") # 查询所有活跃的配置 configs = SystemConfig.query.filter_by(is_active=True).all() # 清空缓存并重新填充 self._config_cache.clear() for config in configs: self._config_cache[config.parameter_name] = config.get_typed_value() self._cache_timestamp = time.time() logger.info(f"配置缓存已刷新，共加载 {len(self._config_cache)} 个配置参数") except Exception as e: logger.error(f"刷新配置缓存失败: {e}") # 如果刷新失败，保留旧缓存 def get_config(self, parameter_name: str, default_value: Any = None, force_refresh: bool = False) -> Any: """ 获取配置值 Args: parameter_name: 参数名称 default_value: 默认值，当配置不存在时返回 force_refresh: 是否强制刷新缓存 Returns: 配置的实际类型值 """ with self._cache_lock: # 检查是否需要刷新缓存 if force_refresh or not self._is_cache_valid() or not self._config_cache: self._refresh_cache() # 从缓存获取值 if parameter_name in self._config_cache: value = self._config_cache[parameter_name] logger.debug(f"从缓存获取配置 {parameter_name} = {value}") return value else: # 缓存中不存在，尝试查询数据库（可能是新添加的配置） logger.debug(f"缓存中不存在配置 {parameter_name}，尝试查询数据库") # 检查是否有Flask应用上下文 try: from flask import has_app_context if not has_app_context(): logger.warning(f"没有Flask应用上下文，无法查询配置 {parameter_name}，返回默认值 {default_value}") return default_value except ImportError: logger.warning(f"Flask未加载，无法查询配置 {parameter_name}，返回默认值 {default_value}") return default_value try: config = SystemConfig.query.filter_by( parameter_name=parameter_name, is_active=True ).first() if config: value = config.get_typed_value() # 更新缓存 self._config_cache[parameter_name] = value logger.debug(f"从数据库获取配置 {parameter_name} = {value}") return value else: logger.warning(f"配置参数 {parameter_name} 不存在，返回默认值 {default_value}") return default_value except Exception as e: logger.error(f"查询配置参数 {parameter_name} 失败: {e}") return default_value def get_str(self, parameter_name: str, default_value: str = '') -> str: """获取字符串类型配置""" value = self.get_config(parameter_name, default_value) return str(value) if value is not None else default_value def get_int(self, parameter_name: str, default_value: int = 0) -> int: """获取整数类型配置""" value = self.get_config(parameter_name, default_value) try: return int(value) if value is not None else default_value except (ValueError, TypeError): logger.warning(f"配置 {parameter_name} 无法转换为整数，返回默认值 {default_value}") return default_value def get_float(self, parameter_name: str, default_value: float = 0.0) -> float: """获取浮点数类型配置""" value = self.get_config(parameter_name, default_value) try: return float(value) if value is not None else default_value except (ValueError, TypeError): logger.warning(f"配置 {parameter_name} 无法转换为浮点数，返回默认值 {default_value}") return default_value def get_bool(self, parameter_name: str, default_value: bool = False) -> bool: """获取布尔类型配置""" value = self.get_config(parameter_name, default_value) if isinstance(value, bool): return value elif isinstance(value, str): return value.upper() in ['TRUE', 'YES', '1', 'ON'] elif isinstance(value, (int, float)): return bool(value) else: return default_value def get_list(self, parameter_name: str, default_value: list = None) -> list: """获取列表类型配置""" if default_value is None: default_value = [] value = self.get_config(parameter_name, default_value) return value if isinstance(value, list) else default_value def set_config(self, parameter_name: str, value: Any) -> bool: """ 设置配置值 Args: parameter_name: 参数名称 value: 要设置的值 Returns: 是否设置成功 """ try: config = SystemConfig.query.filter_by(parameter_name=parameter_name).first() if config: config.set_typed_value(value) db.session.commit() # 更新缓存 with self._cache_lock: self._config_cache[parameter_name] = config.get_typed_value() logger.info(f"配置参数 {parameter_name} 已更新为: {value}") return True else: logger.error(f"配置参数 {parameter_name} 不存在") return False except Exception as e: logger.error(f"设置配置参数 {parameter_name} 失败: {e}") db.session.rollback() return False def reload_config(self) -> None: """重新加载所有配置（强制刷新缓存）""" logger.info("强制重新加载配置缓存") self._refresh_cache() def get_config_by_category(self, category: str) -> Dict[str, Any]: """ 获取指定分类的所有配置 Args: category: 配置分类 Returns: 该分类下的所有配置字典 """ try: configs = SystemConfig.query.filter_by( category=category, is_active=True ).all() result = {} for config in configs: result[config.parameter_name] = config.get_typed_value() return result except Exception as e: logger.error(f"获取分类 {category} 的配置失败: {e}") return {} def get_cache_info(self) -> Dict[str, Any]: """获取缓存状态信息""" with self._cache_lock: return { 'cache_size': len(self._config_cache), 'cache_timestamp': self._cache_timestamp, 'cache_age_seconds': time.time() - self._cache_timestamp, 'cache_ttl_seconds': self._cache_ttl, 'is_cache_valid': self._is_cache_valid() } # 创建全局配置服务实例 config_service = ConfigService() # 便捷函数，方便在其他模块中使用 def get_config(parameter_name: str, default_value: Any = None) -> Any: """获取配置值的便捷函数""" return config_service.get_config(parameter_name, default_value) def get_str_config(parameter_name: str, default_value: str = '') -> str: """获取字符串配置的便捷函数""" return config_service.get_str(parameter_name, default_value) def get_int_config(parameter_name: str, default_value: int = 0) -> int: """获取整数配置的便捷函数""" return config_service.get_int(parameter_name, default_value) def get_float_config(parameter_name: str, default_value: float = 0.0) -> float: """获取浮点数配置的便捷函数""" return config_service.get_float(parameter_name, default_value) def get_bool_config(parameter_name: str, default_value: bool = False) -> bool: """获取布尔配置的便捷函数""" return config_service.get_bool(parameter_name, default_value) def get_list_config(parameter_name: str, default_value: list = None) -> list: """获取列表配置的便捷函数""" return config_service.get_list(parameter_name, default_value or []) def set_config(parameter_name: str, value: Any) -> bool: """设置配置值的便捷函数""" return config_service.set_config(parameter_name, value) def reload_config() -> None: """重新加载配置的便捷函数""" config_service.reload_config()