tech
f534dc52cb
- 新增 ForceUpdateCommand 类实现强制更新功能 - 添加 force-update CLI 命令支持 - 实现参数验证、配置对比、CDN缓存刷新 - 添加完整的测试用例和使用示例 - 更新文档和进度跟踪 - 支持模拟环境测试 - 添加警告提示和错误处理 主要文件: - src/config_man/core/force_update_command.py (新增) - src/config_man/cli/main.py (更新) - examples/test_force_update.py (新增) - examples/force_update_usage.py (新增) - docs/progress-tracker.md (更新)
240 lines
8.2 KiB
Python
240 lines
8.2 KiB
Python
"""
|
||
配置管理模块
|
||
|
||
支持从配置文件和环境变量读取配置信息。
|
||
"""
|
||
|
||
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,
|
||
"cloudflare": {
|
||
"api_token": "",
|
||
"zone_id": ""
|
||
}
|
||
},
|
||
"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_CDN_CLOUDFLARE_API_TOKEN": ("cdn", "cloudflare", "api_token"),
|
||
"CONFIG_MAN_CDN_CLOUDFLARE_ZONE_ID": ("cdn", "cloudflare", "zone_id"),
|
||
"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', {})
|
||
|
||
def get_cloudflare_config(self) -> Dict[str, Any]:
|
||
"""获取Cloudflare配置"""
|
||
return self.get('cdn.cloudflare', {})
|
||
|
||
def get_cloudflare_api_token(self) -> str:
|
||
"""获取Cloudflare API Token"""
|
||
return self.get('cdn.cloudflare.api_token', '')
|
||
|
||
def get_cloudflare_zone_id(self) -> str:
|
||
"""获取Cloudflare Zone ID"""
|
||
return self.get('cdn.cloudflare.zone_id', '')
|
||
|
||
|
||
# 全局配置实例
|
||
config = Config() |