""" 配置加载模块 负责读取和解析 config.toml 配置文件,提供全局配置对象。 """ from pathlib import Path import tomllib from pydantic import ValidationError from .config_models import ConfigModel, NapCatWSModel, BotModel, RedisModel, DockerModel, ImageManagerModel, MySQLModel, ReverseWSModel, ThreadingModel, BilibiliModel, LocalFileServerModel, DiscordModel, CrossPlatformModel, LoggingModel from .utils.logger import ModuleLogger from .utils.exceptions import ConfigError, ConfigNotFoundError, ConfigValidationError class Config: """ 配置加载类,负责读取和解析 config.toml 文件 """ def __init__(self, file_path: str = "config.toml"): """ 初始化配置加载器 :param file_path: 配置文件路径,默认为 "config.toml" """ self.path = Path(file_path) self._model: ConfigModel # 创建模块专用日志记录器 self.logger = ModuleLogger("ConfigLoader") self.load() def load(self): """ 加载并验证配置文件 :raises ConfigNotFoundError: 如果配置文件不存在 :raises ConfigValidationError: 如果配置格式不正确 :raises ConfigError: 如果加载配置时发生其他错误 """ if not self.path.exists(): self.logger.warning(f"配置文件 {self.path} 未找到,正在生成示例配置...") self._generate_example_config() self.logger.success(f"示例配置已生成: {self.path}") self.logger.info("请编辑配置文件后重新启动程序") try: self.logger.info(f"正在从 {self.path} 加载配置...") with open(self.path, "rb") as f: raw_config = tomllib.load(f) self._model = ConfigModel(**raw_config) self.logger.success("配置加载并验证成功!") except ValidationError as e: error_details = [] for error in e.errors(): field = " -> ".join(map(str, error["loc"])) error_msg = f"字段 '{field}': {error['msg']}" error_details.append(error_msg) validation_error = ConfigValidationError( message="配置验证失败" ) validation_error.original_error = e self.logger.error("配置验证失败,请检查 `config.toml` 文件中的以下错误:") for detail in error_details: self.logger.error(f" - {detail}") self.logger.log_custom_exception(validation_error) raise validation_error except tomllib.TOMLDecodeError as e: error = ConfigError( message=f"TOML解析错误: {str(e)}" ) error.original_error = e self.logger.error(f"加载配置文件时发生TOML解析错误: {error.message}") self.logger.log_custom_exception(error) raise error except Exception as e: error = ConfigError( message=f"加载配置文件时发生未知错误: {str(e)}" ) error.original_error = e self.logger.exception(f"加载配置文件时发生未知错误: {error.message}") self.logger.log_custom_exception(error) raise error def _generate_example_config(self): """ 生成示例配置文件 """ example_path = Path("config.example.toml") if not example_path.exists(): self.logger.error(f"示例配置文件 {example_path} 不存在,无法生成配置") raise ConfigNotFoundError(message=f"示例配置文件 {example_path} 不存在") content = example_path.read_text() self.path.write_text(content) # 通过属性访问配置 @property def napcat_ws(self) -> NapCatWSModel: """ 获取 NapCat WebSocket 配置 """ return self._model.napcat_ws @property def bot(self) -> BotModel: """ 获取 Bot 基础配置 """ return self._model.bot @property def redis(self) -> RedisModel: """ 获取 Redis 配置 """ return self._model.redis @property def mysql(self) -> MySQLModel: """ 获取 MySQL 配置 """ return self._model.mysql @property def docker(self) -> DockerModel: """ 获取 Docker 配置 """ return self._model.docker @property def image_manager(self) -> ImageManagerModel: """ 获取图片生成管理器配置 """ return self._model.image_manager @property def reverse_ws(self) -> ReverseWSModel: """ 获取反向 WebSocket 配置 """ return self._model.reverse_ws @property def threading(self) -> ThreadingModel: """ 获取线程管理配置 """ return self._model.threading @property def bilibili(self) -> BilibiliModel: """ 获取 Bilibili 配置 """ return self._model.bilibili @property def local_file_server(self) -> LocalFileServerModel: """ 获取本地文件服务器配置 """ return self._model.local_file_server @property def discord(self) -> DiscordModel: """ 获取 Discord 配置 """ return self._model.discord @property def cross_platform(self) -> CrossPlatformModel: """ 获取跨平台配置 """ return self._model.cross_platform @property def logging(self) -> LoggingModel: """ 获取日志配置 """ return self._model.logging # 实例化全局配置对象 global_config = Config()