151 lines
5.0 KiB
Python
151 lines
5.0 KiB
Python
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()
|