"""Configuration management module""" import json from pathlib import Path from typing import Dict, Optional, Any from .constants import DEFAULT_SETTINGS, DEFAULT_THRESHOLDS, ENVIRONMENTS class ConfigManager: """管理配置文件和设置""" def __init__(self, config_path: Optional[str] = None): """初始化配置管理器 Args: config_path: 配置文件路径,如果为None则尝试自动查找 """ self.config = {} self.config_path = config_path or self._find_config_file() if self.config_path and Path(self.config_path).exists(): self.load_config() else: self._use_defaults() def _find_config_file(self) -> Optional[str]: """查找配置文件""" search_paths = [ 'config.json', '../config.json', 'aws/config.json', Path.home() / '.aws_billing' / 'config.json' ] for path in search_paths: if Path(path).exists(): return str(path) return None def load_config(self): """加载配置文件""" try: with open(self.config_path, 'r', encoding='utf-8') as f: self.config = json.load(f) print(f"✓ 配置文件加载成功: {self.config_path}") except Exception as e: print(f"⚠️ 配置文件加载失败: {e}") self._use_defaults() def _use_defaults(self): """使用默认配置""" self.config = { 'settings': DEFAULT_SETTINGS.copy(), 'thresholds': DEFAULT_THRESHOLDS.copy(), 'environments': ENVIRONMENTS.copy(), 'webhook_urls': {} } print("ℹ️ 使用默认配置") def get(self, key: str, default: Any = None) -> Any: """获取配置项 Args: key: 配置键,支持点号分隔的多级键 (如 'thresholds.account_warning') default: 默认值 Returns: 配置值 """ keys = key.split('.') value = self.config for k in keys: if isinstance(value, dict): value = value.get(k) if value is None: return default else: return default return value def set(self, key: str, value: Any): """设置配置项 Args: key: 配置键,支持点号分隔的多级键 value: 配置值 """ keys = key.split('.') config = self.config for k in keys[:-1]: if k not in config: config[k] = {} config = config[k] config[keys[-1]] = value def get_environment_config(self, env: str) -> Dict: """获取环境配置 Args: env: 环境名称 Returns: 环境配置字典 """ env_config = self.config.get('environments', {}).get(env, ENVIRONMENTS.get(env, {})) return env_config def get_webhook_url(self, webhook_type: str) -> Optional[str]: """获取webhook URL Args: webhook_type: webhook类型 Returns: webhook URL """ return self.config.get('webhook_urls', {}).get(webhook_type) def get_thresholds(self) -> Dict: """获取所有阈值配置""" return self.config.get('thresholds', DEFAULT_THRESHOLDS.copy()) def save_config(self, path: Optional[str] = None): """保存配置到文件 Args: path: 保存路径,如果为None则使用当前配置路径 """ save_path = path or self.config_path if not save_path: save_path = 'config.json' try: with open(save_path, 'w', encoding='utf-8') as f: json.dump(self.config, f, indent=2, ensure_ascii=False) print(f"✓ 配置已保存到: {save_path}") except Exception as e: print(f"✗ 配置保存失败: {e}") def validate(self) -> bool: """验证配置完整性 Returns: 配置是否有效 """ required_keys = ['environments', 'thresholds'] for key in required_keys: if key not in self.config: print(f"✗ 缺少必需配置: {key}") return False # 检查至少有一个环境配置 if not self.config.get('environments'): print("✗ 至少需要配置一个环境") return False return True