""" 配置管理模块 ----------- 提供统一的配置加载、访问和保存功能。 """ import os import configparser from typing import Dict, List, Optional, Any from dotenv import load_dotenv from ..core.utils.log_utils import get_logger from .defaults import DEFAULT_CONFIG # 加载 .env 文件 load_dotenv() logger = get_logger(__name__) class ConfigManager: """ 配置管理类,负责加载和保存配置 单例模式确保全局只有一个配置实例 """ _instance = None def __new__(cls, config_file=None): """单例模式实现""" if cls._instance is None: cls._instance = super(ConfigManager, cls).__new__(cls) cls._instance._init(config_file) return cls._instance def _init(self, config_file): """初始化配置管理器""" # 计算应用根目录(不依赖 os.getcwd()) import sys if getattr(sys, 'frozen', False): # PyInstaller 打包后,根目录是 exe 所在目录 self.app_root = os.path.dirname(sys.executable) else: # 源码运行,根目录是 app/config/ 的上两级 self.app_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) self.config_file = config_file or os.path.join(self.app_root, 'config.ini') self.config = configparser.ConfigParser() self.load_config() def load_config(self) -> None: """ 加载配置文件,如果不存在则创建默认配置 API 密钥优先从环境变量 (.env) 读取 """ if not os.path.exists(self.config_file): self.create_default_config() else: try: # 先读取现有配置 self.config.read(self.config_file, encoding='utf-8') # 检查是否有缺失的配置项,只添加缺失的项 for section, options in DEFAULT_CONFIG.items(): if not self.config.has_section(section): self.config.add_section(section) for option, value in options.items(): if not self.config.has_option(section, option): self.config.set(section, option, value) # API 密钥优先从环境变量读取 self._override_from_env() # 保存更新后的配置 self.save_config() logger.info(f"已加载并更新配置文件: {self.config_file}") except Exception as e: logger.error(f"加载配置文件时出错: {e}") logger.info("使用默认配置") self.create_default_config(save=False) def _override_from_env(self) -> None: """从环境变量覆盖敏感配置""" env_mapping = { ('API', 'api_key'): 'BAIDU_API_KEY', ('API', 'secret_key'): 'BAIDU_SECRET_KEY', ('Gitea', 'token'): 'GITEA_TOKEN', } for (section, option), env_key in env_mapping.items(): env_val = os.getenv(env_key, '').strip() if env_val: self.config.set(section, option, env_val) def create_default_config(self, save: bool = True) -> None: """创建默认配置""" for section, options in DEFAULT_CONFIG.items(): if not self.config.has_section(section): self.config.add_section(section) for option, value in options.items(): self.config.set(section, option, value) if save: self.save_config() logger.info(f"已创建默认配置文件: {self.config_file}") def save_config(self) -> None: """保存配置到文件(API 密钥不写入文件,Gitea token 需要持久化)""" # 保存前临时清空 API 密钥,避免写入文件(这些从 .env 读取) saved_keys = {} for option in ('api_key', 'secret_key'): try: saved_keys[option] = self.config.get('API', option, fallback='') except Exception: saved_keys[option] = '' self.config.set('API', option, '') try: with open(self.config_file, 'w', encoding='utf-8') as f: self.config.write(f) logger.info(f"配置已保存到: {self.config_file}") finally: # 恢复内存中的值(即使写入失败也恢复) for option, val in saved_keys.items(): if val: self.config.set('API', option, val) def get(self, section: str, option: str, fallback: Any = None) -> Any: """获取配置值""" return self.config.get(section, option, fallback=fallback) def getint(self, section: str, option: str, fallback: int = 0) -> int: """获取整数配置值""" return self.config.getint(section, option, fallback=fallback) def getfloat(self, section: str, option: str, fallback: float = 0.0) -> float: """获取浮点数配置值""" return self.config.getfloat(section, option, fallback=fallback) def getboolean(self, section: str, option: str, fallback: bool = False) -> bool: """获取布尔配置值""" return self.config.getboolean(section, option, fallback=fallback) def get_list(self, section: str, option: str, fallback: str = "", delimiter: str = ",") -> List[str]: """获取列表配置值(逗号分隔的字符串转为列表)""" value = self.get(section, option, fallback) return [item.strip() for item in value.split(delimiter) if item.strip()] def update(self, section: str, option: str, value: Any) -> None: """更新配置选项""" if not self.config.has_section(section): self.config.add_section(section) self.config.set(section, option, str(value)) logger.debug(f"更新配置: [{section}] {option} = {value}") def get_path(self, section: str, option: str, fallback: str = "", create: bool = False) -> str: """ 获取路径配置并确保它是一个有效的绝对路径 如果create为True,则自动创建该目录 """ from pathlib import Path path_str = self.get(section, option, fallback) path = Path(path_str) if not path.is_absolute(): # 相对路径,转为绝对路径(相对于应用根目录) path = Path(self.app_root) / path if create: try: # 智能判断是文件还是目录 # 如果有后缀名则认为是文件,创建其父目录 if path.suffix: directory = path.parent if not directory.exists(): directory.mkdir(parents=True, exist_ok=True) logger.info(f"已创建父目录: {directory}") else: # 否则认为是目录路径 if not path.exists(): path.mkdir(parents=True, exist_ok=True) logger.info(f"已创建目录: {path}") except Exception as e: logger.error(f"创建目录失败: {path}, 错误: {e}") return str(path.absolute())