191 lines
4.9 KiB
Python
191 lines
4.9 KiB
Python
"""
|
|
统一配置管理器
|
|
|
|
单例模式，提供:
|
|
- 配置缓存，避免重复读取文件
|
|
- 配置热更新检测
|
|
- 类型安全的配置访问
|
|
"""
|
|
|
|
import tomllib
|
|
from pathlib import Path
|
|
from threading import Lock
|
|
from typing import Any, Dict, Optional
|
|
|
|
from loguru import logger
|
|
|
|
|
|
class ConfigManager:
|
|
"""
|
|
配置管理器 (线程安全单例)
|
|
|
|
使用示例:
|
|
from utils.config_manager import get_config
|
|
|
|
# 获取单个配置项
|
|
admins = get_config().get("Bot", "admins", [])
|
|
|
|
# 获取整个配置节
|
|
bot_config = get_config().get_section("Bot")
|
|
|
|
# 检查并重新加载
|
|
if get_config().reload_if_changed():
|
|
logger.info("配置已更新")
|
|
"""
|
|
|
|
_instance: Optional["ConfigManager"] = None
|
|
_lock = Lock()
|
|
|
|
def __new__(cls):
|
|
if cls._instance is None:
|
|
with cls._lock:
|
|
if cls._instance is None:
|
|
instance = super().__new__(cls)
|
|
instance._initialized = False
|
|
cls._instance = instance
|
|
return cls._instance
|
|
|
|
def __init__(self):
|
|
if self._initialized:
|
|
return
|
|
|
|
self._config: Dict[str, Any] = {}
|
|
self._config_path = Path("main_config.toml")
|
|
self._file_mtime: float = 0
|
|
self._config_lock = Lock()
|
|
self._reload()
|
|
self._initialized = True
|
|
logger.debug("ConfigManager 初始化完成")
|
|
|
|
def _reload(self) -> bool:
|
|
"""重新加载配置文件"""
|
|
try:
|
|
if not self._config_path.exists():
|
|
logger.warning(f"配置文件不存在: {self._config_path}")
|
|
return False
|
|
|
|
current_mtime = self._config_path.stat().st_mtime
|
|
if current_mtime == self._file_mtime and self._config:
|
|
return False # 文件未变化
|
|
|
|
with self._config_lock:
|
|
with open(self._config_path, "rb") as f:
|
|
self._config = tomllib.load(f)
|
|
self._file_mtime = current_mtime
|
|
|
|
logger.debug("配置文件已重新加载")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"加载配置文件失败: {e}")
|
|
return False
|
|
|
|
def get(self, section: str, key: str, default: Any = None) -> Any:
|
|
"""
|
|
获取配置项
|
|
|
|
Args:
|
|
section: 配置节名称，如 "Bot"
|
|
key: 配置项名称，如 "admins"
|
|
default: 默认值
|
|
|
|
Returns:
|
|
配置值或默认值
|
|
"""
|
|
return self._config.get(section, {}).get(key, default)
|
|
|
|
def get_section(self, section: str) -> Dict[str, Any]:
|
|
"""
|
|
获取整个配置节
|
|
|
|
Args:
|
|
section: 配置节名称
|
|
|
|
Returns:
|
|
配置节字典的副本
|
|
"""
|
|
return self._config.get(section, {}).copy()
|
|
|
|
def get_all(self) -> Dict[str, Any]:
|
|
"""获取完整配置（只读副本）"""
|
|
return self._config.copy()
|
|
|
|
def reload_if_changed(self) -> bool:
|
|
"""
|
|
如果文件有变化则重新加载
|
|
|
|
Returns:
|
|
是否重新加载了配置
|
|
"""
|
|
try:
|
|
if not self._config_path.exists():
|
|
return False
|
|
current_mtime = self._config_path.stat().st_mtime
|
|
if current_mtime != self._file_mtime:
|
|
return self._reload()
|
|
except Exception:
|
|
pass
|
|
return False
|
|
|
|
def force_reload(self) -> bool:
|
|
"""强制重新加载配置"""
|
|
self._file_mtime = 0
|
|
return self._reload()
|
|
|
|
|
|
# ==================== 便捷函数 ====================
|
|
|
|
def get_config() -> ConfigManager:
|
|
"""获取配置管理器实例"""
|
|
return ConfigManager()
|
|
|
|
|
|
def get_bot_config() -> Dict[str, Any]:
|
|
"""快捷获取 [Bot] 配置节"""
|
|
return get_config().get_section("Bot")
|
|
|
|
|
|
def get_performance_config() -> Dict[str, Any]:
|
|
"""快捷获取 [Performance] 配置节"""
|
|
return get_config().get_section("Performance")
|
|
|
|
|
|
def get_database_config() -> Dict[str, Any]:
|
|
"""快捷获取 [Database] 配置节"""
|
|
return get_config().get_section("Database")
|
|
|
|
|
|
def get_scheduler_config() -> Dict[str, Any]:
|
|
"""快捷获取 [Scheduler] 配置节"""
|
|
return get_config().get_section("Scheduler")
|
|
|
|
|
|
def get_queue_config() -> Dict[str, Any]:
|
|
"""快捷获取 [Queue] 配置节"""
|
|
return get_config().get_section("Queue")
|
|
|
|
|
|
def get_concurrency_config() -> Dict[str, Any]:
|
|
"""快捷获取 [Concurrency] 配置节"""
|
|
return get_config().get_section("Concurrency")
|
|
|
|
|
|
def get_webui_config() -> Dict[str, Any]:
|
|
"""快捷获取 [WebUI] 配置节"""
|
|
return get_config().get_section("WebUI")
|
|
|
|
|
|
# ==================== 导出列表 ====================
|
|
|
|
__all__ = [
|
|
'ConfigManager',
|
|
'get_config',
|
|
'get_bot_config',
|
|
'get_performance_config',
|
|
'get_database_config',
|
|
'get_scheduler_config',
|
|
'get_queue_config',
|
|
'get_concurrency_config',
|
|
'get_webui_config',
|
|
]
|