import os import yaml from typing import Dict, Any, Optional from pathlib import Path class Config: """统一的配置管理类""" def __init__(self): self._config = self._load_config() self._init_config_attributes() def _load_config(self) -> Dict[str, Any]: """加载配置,优先从环境变量读取,否则从配置文件读取""" # 优先从环境变量读取 if "aws_key_id" in os.environ: return os.environ # 从配置文件读取 config_path = self._get_config_path() if config_path.exists(): try: with open(config_path, encoding="utf-8", mode="r") as f: return yaml.load(f, Loader=yaml.FullLoader) except Exception as e: print(f"读取配置文件失败: {e}") return {} else: print(f"配置文件不存在: {config_path}") return {} def _get_config_path(self) -> Path: """获取配置文件路径""" # 获取当前文件所在目录的上级目录 current_dir = Path(__file__).parent.parent return current_dir / "config.yaml" def _init_config_attributes(self): """初始化配置属性""" # 腾讯云COS配置 self.cos_secret_id = self._config.get("cos_secret_id") self.cos_secret_key = self._config.get("cos_secret_key") self.cos_region = self._config.get("cos_region") self.cos_sucai_bucket_name = self._config.get("cos_sucai_bucket_name") # AWS S3配置 self.aws_key_id = self._config.get("aws_key_id") self.aws_access_key = self._config.get("aws_access_key") # JM API配置 self.jm_api_key = self._config.get("jm_api_key") # VOD配置(使用COS相同的密钥) self.vod_secret_id = self._config.get("vod_secret_id") or self.cos_secret_id self.vod_secret_key = self._config.get("vod_secret_key") or self.cos_secret_key self.vod_region = self._config.get("vod_region") or self.cos_region # 其他配置项 self._other_configs = { k: v for k, v in self._config.items() if k not in [ "cos_secret_id", "cos_secret_key", "cos_region", "cos_sucai_bucket_name", "aws_key_id", "aws_access_key", "jm_api_key", "vod_secret_id", "vod_secret_key", "vod_region", ] } def get(self, key: str, default: Any = None) -> Any: """获取配置值""" return self._config.get(key, default) def get_config(self, key: str, default: Any = None) -> Any: """获取配置值(别名方法)""" return self.get(key, default) def get_cos_config(self) -> Dict[str, str]: """获取腾讯云COS配置""" return { "secret_id": self.cos_secret_id, "secret_key": self.cos_secret_key, "region": self.cos_region, "bucket_name": self.cos_sucai_bucket_name, } def get_aws_config(self) -> Dict[str, str]: """获取AWS S3配置""" return { "access_key_id": self.aws_key_id, "secret_access_key": self.aws_access_key, } def get_jm_config(self) -> Dict[str, str]: """获取JM API配置""" return {"api_key": self.jm_api_key} def get_vod_config(self) -> Dict[str, str]: """获取腾讯云VOD配置""" return { "secret_id": self.vod_secret_id, "secret_key": self.vod_secret_key, "region": self.vod_region, } def has_cos_config(self) -> bool: """检查是否有完整的腾讯云COS配置""" return all([self.cos_secret_id, self.cos_secret_key, self.cos_region]) def has_aws_config(self) -> bool: """检查是否有完整的AWS S3配置""" return all([self.aws_key_id, self.aws_access_key]) def has_jm_config(self) -> bool: """检查是否有JM API配置""" return bool(self.jm_api_key) def has_vod_config(self) -> bool: """检查是否有完整的腾讯云VOD配置""" return all([self.vod_secret_id, self.vod_secret_key, self.vod_region]) def reload(self): """重新加载配置""" self._config = self._load_config() self._init_config_attributes() def to_dict(self) -> Dict[str, Any]: """将配置转换为字典""" return self._config.copy() def __str__(self) -> str: """字符串表示""" config_info = { "cos_config": self.get_cos_config() if self.has_cos_config() else "不完整", "aws_config": self.get_aws_config() if self.has_aws_config() else "不完整", "jm_config": self.get_jm_config() if self.has_jm_config() else "不完整", } return f"Config({config_info})" # 创建全局配置实例 config = Config()