修复CDN URL和路径构建逻辑
- 修复CDN URL构建，正确处理路径重复问题
- 修复rclone路径构建，使用正确的格式: {rclone_remote}:{storage_path}/{version_path}/{filename}
- 修复CDN路径构建，使用正确的格式: {cdn_base_url}/{version_path}/{filename}
- 修复CDN内容解密处理，支持加密和解密后的内容
- 添加路径构建测试示例
- 完善错误处理和空内容处理
This commit is contained in:
222
src/config_man/utils/config.py
Normal file
222
src/config_man/utils/config.py
Normal file
@@ -0,0 +1,222 @@
|
||||
"""
|
||||
配置管理模块
|
||||
|
||||
支持从配置文件和环境变量读取配置信息。
|
||||
"""
|
||||
|
||||
import os
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Dict, Any, Optional
|
||||
|
||||
|
||||
class Config:
|
||||
"""配置管理类"""
|
||||
|
||||
def __init__(self, config_file: Optional[str] = None):
|
||||
"""
|
||||
初始化配置管理器
|
||||
|
||||
Args:
|
||||
config_file: 配置文件路径，如果为None则使用默认路径
|
||||
"""
|
||||
self.config_file = config_file or self._get_default_config_path()
|
||||
self._config = self._load_config()
|
||||
|
||||
def _get_default_config_path(self) -> str:
|
||||
"""获取默认配置文件路径"""
|
||||
# 优先使用用户主目录下的配置文件
|
||||
home_config = Path.home() / ".config" / "config-man" / "config-man.json"
|
||||
if home_config.exists():
|
||||
return str(home_config)
|
||||
|
||||
# 其次使用项目根目录下的配置文件
|
||||
project_config = Path(__file__).parent.parent.parent.parent / "config-man.json"
|
||||
if project_config.exists():
|
||||
return str(project_config)
|
||||
|
||||
# 最后使用默认配置文件
|
||||
return str(Path(__file__).parent.parent.parent.parent / "config-man.json")
|
||||
|
||||
def _load_config(self) -> Dict[str, Any]:
|
||||
"""加载配置文件"""
|
||||
default_config = {
|
||||
"storage": {
|
||||
"path": "ab",
|
||||
"rclone_remote": "remote",
|
||||
"timeout": 30
|
||||
},
|
||||
"cdn": {
|
||||
"base_url": "",
|
||||
"timeout": 10,
|
||||
"retry_count": 3
|
||||
},
|
||||
"crypto": {
|
||||
"algorithm": "DES",
|
||||
"key": "tbambooz"
|
||||
},
|
||||
"logging": {
|
||||
"level": "INFO",
|
||||
"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
|
||||
},
|
||||
"display": {
|
||||
"table_format": "grid",
|
||||
"max_width": 80,
|
||||
"truncate_length": 25
|
||||
}
|
||||
}
|
||||
|
||||
# 如果配置文件存在，则加载并合并
|
||||
if os.path.exists(self.config_file):
|
||||
try:
|
||||
with open(self.config_file, 'r', encoding='utf-8') as f:
|
||||
file_config = json.load(f)
|
||||
# 深度合并配置
|
||||
self._merge_config(default_config, file_config)
|
||||
except (json.JSONDecodeError, IOError) as e:
|
||||
print(f"警告: 无法加载配置文件 {self.config_file}: {e}")
|
||||
|
||||
# 从环境变量覆盖配置
|
||||
self._load_from_env(default_config)
|
||||
|
||||
return default_config
|
||||
|
||||
def _merge_config(self, base: Dict[str, Any], override: Dict[str, Any]) -> None:
|
||||
"""深度合并配置"""
|
||||
for key, value in override.items():
|
||||
if key in base and isinstance(base[key], dict) and isinstance(value, dict):
|
||||
self._merge_config(base[key], value)
|
||||
else:
|
||||
base[key] = value
|
||||
|
||||
def _load_from_env(self, config: Dict[str, Any]) -> None:
|
||||
"""从环境变量加载配置"""
|
||||
env_mappings = {
|
||||
"CONFIG_MAN_STORAGE_PATH": ("storage", "path"),
|
||||
"CONFIG_MAN_STORAGE_RCLONE_REMOTE": ("storage", "rclone_remote"),
|
||||
"CONFIG_MAN_STORAGE_TIMEOUT": ("storage", "timeout"),
|
||||
"CONFIG_MAN_CDN_BASE_URL": ("cdn", "base_url"),
|
||||
"CONFIG_MAN_CDN_TIMEOUT": ("cdn", "timeout"),
|
||||
"CONFIG_MAN_CDN_RETRY_COUNT": ("cdn", "retry_count"),
|
||||
"CONFIG_MAN_CRYPTO_KEY": ("crypto", "key"),
|
||||
"CONFIG_MAN_CRYPTO_ALGORITHM": ("crypto", "algorithm"),
|
||||
"CONFIG_MAN_LOG_LEVEL": ("logging", "level"),
|
||||
"CONFIG_MAN_DISPLAY_TABLE_FORMAT": ("display", "table_format"),
|
||||
"CONFIG_MAN_DISPLAY_MAX_WIDTH": ("display", "max_width"),
|
||||
"CONFIG_MAN_DISPLAY_TRUNCATE_LENGTH": ("display", "truncate_length"),
|
||||
}
|
||||
|
||||
for env_var, config_path in env_mappings.items():
|
||||
env_value = os.getenv(env_var)
|
||||
if env_value is not None:
|
||||
# 获取配置路径的父级
|
||||
current = config
|
||||
for key in config_path[:-1]:
|
||||
if key not in current:
|
||||
current[key] = {}
|
||||
current = current[key]
|
||||
|
||||
# 设置值，尝试转换为适当类型
|
||||
key = config_path[-1]
|
||||
try:
|
||||
# 尝试转换为整数
|
||||
if isinstance(current.get(key), int):
|
||||
current[key] = int(env_value)
|
||||
# 尝试转换为布尔值
|
||||
elif isinstance(current.get(key), bool):
|
||||
current[key] = env_value.lower() in ('true', '1', 'yes', 'on')
|
||||
else:
|
||||
current[key] = env_value
|
||||
except ValueError:
|
||||
current[key] = env_value
|
||||
|
||||
def get(self, key: str, default: Any = None) -> Any:
|
||||
"""
|
||||
获取配置值
|
||||
|
||||
Args:
|
||||
key: 配置键，支持点号分隔的嵌套键，如 'storage.path'
|
||||
default: 默认值
|
||||
|
||||
Returns:
|
||||
配置值
|
||||
"""
|
||||
keys = key.split('.')
|
||||
value = self._config
|
||||
|
||||
for k in keys:
|
||||
if isinstance(value, dict) and k in value:
|
||||
value = value[k]
|
||||
else:
|
||||
return default
|
||||
|
||||
return value
|
||||
|
||||
def set(self, key: str, value: Any) -> None:
|
||||
"""
|
||||
设置配置值
|
||||
|
||||
Args:
|
||||
key: 配置键，支持点号分隔的嵌套键
|
||||
value: 配置值
|
||||
"""
|
||||
keys = key.split('.')
|
||||
current = self._config
|
||||
|
||||
# 导航到父级
|
||||
for k in keys[:-1]:
|
||||
if k not in current:
|
||||
current[k] = {}
|
||||
current = current[k]
|
||||
|
||||
# 设置值
|
||||
current[keys[-1]] = value
|
||||
|
||||
def save(self) -> None:
|
||||
"""保存配置到文件"""
|
||||
try:
|
||||
# 确保配置目录存在
|
||||
config_dir = os.path.dirname(self.config_file)
|
||||
if config_dir:
|
||||
os.makedirs(config_dir, exist_ok=True)
|
||||
|
||||
with open(self.config_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(self._config, f, indent=2, ensure_ascii=False)
|
||||
except IOError as e:
|
||||
print(f"警告: 无法保存配置文件 {self.config_file}: {e}")
|
||||
|
||||
def get_storage_path(self) -> str:
|
||||
"""获取存储路径"""
|
||||
return self.get('storage.path', 'ab')
|
||||
|
||||
def get_cdn_base_url(self) -> str:
|
||||
"""获取CDN基础URL"""
|
||||
return self.get('cdn.base_url', '')
|
||||
|
||||
def get_crypto_key(self) -> str:
|
||||
"""获取加密密钥"""
|
||||
return self.get('crypto.key', 'tbambooz')
|
||||
|
||||
def get_display_config(self) -> Dict[str, Any]:
|
||||
"""获取显示配置"""
|
||||
return self.get('display', {})
|
||||
|
||||
def get_storage_config(self) -> Dict[str, Any]:
|
||||
"""获取存储配置"""
|
||||
return self.get('storage', {})
|
||||
|
||||
def get_cdn_config(self) -> Dict[str, Any]:
|
||||
"""获取CDN配置"""
|
||||
return self.get('cdn', {})
|
||||
|
||||
def get_crypto_config(self) -> Dict[str, Any]:
|
||||
"""获取加密配置"""
|
||||
return self.get('crypto', {})
|
||||
|
||||
def get_logging_config(self) -> Dict[str, Any]:
|
||||
"""获取日志配置"""
|
||||
return self.get('logging', {})
|
||||
|
||||
|
||||
# 全局配置实例
|
||||
config = Config()
|
||||
Reference in New Issue
Block a user