"""Environment configuration management""" from pathlib import Path from typing import Dict, Any, Optional, List from ..common.constants import ENVIRONMENTS, AWS_DIR, DEFAULT_EXPORT_DIR from ..common.config import ConfigManager class EnvironmentManager: """环境配置管理器""" def __init__(self, config_manager: Optional[ConfigManager] = None): """初始化环境管理器 Args: config_manager: 配置管理器实例 """ self.config_manager = config_manager or ConfigManager() self.environments = self._load_environments() self._create_output_directories() def _load_environments(self) -> Dict[str, Dict]: """加载环境配置 Returns: 环境配置字典 """ # 从配置文件加载，如果没有则使用默认值 env_config = self.config_manager.get('environments') if not env_config: env_config = ENVIRONMENTS.copy() export_dir_override = self.config_manager.get('export_settings.outputDir') if 'misplaced_prod' in env_config: env_config.pop('misplaced_prod', None) # 确保所有环境配置都有必需的字段 for env_name, env_settings in env_config.items(): if export_dir_override: env_settings = dict(env_settings) env_settings['output_dir'] = export_dir_override env_config[env_name] = self._validate_environment_config(env_name, env_settings) return env_config def _validate_environment_config(self, env_name: str, config: Dict) -> Dict: """验证并补充环境配置 Args: env_name: 环境名称 config: 环境配置 Returns: 验证后的配置 """ # 必需字段及其默认值 required_fields = { 'dataset_name': None, 'output_dir': DEFAULT_EXPORT_DIR, 'file_prefix': f'aws_{env_name}', 'exclude_recent_days': 2, 'cost_field': 'line_item_unblended_cost', 'product_field': 'line_item_product_code', 'date_field': 'line_item_usage_start_date', 'supports_tags': False } # 补充缺失字段 for field, default_value in required_fields.items(): if field not in config: if default_value is None: print(f"⚠️ 环境 {env_name} 缺少必需字段: {field}") else: config[field] = default_value # 确保output_dir是Path对象 output_dir = config.get('output_dir') if isinstance(output_dir, str): output_dir = Path(output_dir) if isinstance(output_dir, Path) and not output_dir.is_absolute(): output_dir = AWS_DIR / output_dir config['output_dir'] = output_dir return config def _create_output_directories(self): """创建所有环境的输出目录""" for env_config in self.environments.values(): output_dir = env_config.get('output_dir') if output_dir and isinstance(output_dir, Path): output_dir.mkdir(exist_ok=True, parents=True) def get_environment(self, env_name: str) -> Optional[Dict]: """获取指定环境的配置 Args: env_name: 环境名称 Returns: 环境配置字典 """ return self.environments.get(env_name) def list_environments(self) -> List[str]: """列出所有环境名称 Returns: 环境名称列表 """ return list(self.environments.keys()) def get_dataset_name(self, env_name: str) -> Optional[str]: """获取环境的数据集名称 Args: env_name: 环境名称 Returns: 数据集名称 """ env = self.get_environment(env_name) return env.get('dataset_name') if env else None def get_output_dir(self, env_name: str) -> Optional[Path]: """获取环境的输出目录 Args: env_name: 环境名称 Returns: 输出目录路径 """ env = self.get_environment(env_name) return env.get('output_dir') if env else None def get_file_prefix(self, env_name: str) -> str: """获取环境的文件前缀 Args: env_name: 环境名称 Returns: 文件前缀 """ env = self.get_environment(env_name) return env.get('file_prefix', 'aws') if env else 'aws' def get_exclude_recent_days(self, env_name: str) -> int: """获取环境的排除天数配置 Args: env_name: 环境名称 Returns: 排除的天数 """ env = self.get_environment(env_name) return env.get('exclude_recent_days', 2) if env else 2 def get_field_mappings(self, env_name: str) -> Dict[str, str]: """获取环境的字段映射 Args: env_name: 环境名称 Returns: 字段映射字典 """ env = self.get_environment(env_name) if not env: return {} return { 'cost_field': env.get('cost_field', 'line_item_unblended_cost'), 'product_field': env.get('product_field', 'line_item_product_code'), 'date_field': env.get('date_field', 'line_item_usage_start_date') } def supports_tags(self, env_name: str) -> bool: """检查环境是否支持标签数据 Args: env_name: 环境名称 Returns: 是否支持标签 """ env = self.get_environment(env_name) return env.get('supports_tags', False) if env else False def get_table_pattern(self, env_name: str) -> Optional[str]: """获取表名模式 Args: env_name: 环境名称 Returns: 表名模式字符串 """ env = self.get_environment(env_name) return env.get('table_pattern') if env else None def build_output_filename(self, env_name: str, account_id: str, year_month: str) -> str: """构建输出文件名 Args: env_name: 环境名称 account_id: 账户ID year_month: 年月字符串 (YYYY-MM) Returns: 文件名 """ prefix = self.get_file_prefix(env_name) return f"{prefix}_{account_id}_{year_month}.csv" def build_output_path(self, env_name: str, account_id: str, year_month: str) -> Path: """构建完整的输出文件路径 Args: env_name: 环境名称 account_id: 账户ID year_month: 年月字符串 Returns: 完整文件路径 """ output_dir = self.get_output_dir(env_name) filename = self.build_output_filename(env_name, account_id, year_month) if output_dir: return output_dir / filename else: return Path(filename) def get_cache_file_path(self, env_name: str) -> Path: """获取环境的缓存文件路径 Args: env_name: 环境名称 Returns: 缓存文件路径 """ output_dir = self.get_output_dir(env_name) if output_dir: return output_dir / f"{env_name}_accounts_cache.json" else: return Path(f"{env_name}_accounts_cache.json") def validate_environment(self, env_name: str) -> bool: """验证环境配置是否完整 Args: env_name: 环境名称 Returns: 配置是否有效 """ env = self.get_environment(env_name) if not env: print(f"✗ 环境 {env_name} 不存在") return False # 检查必需字段 required_fields = ['dataset_name', 'output_dir', 'cost_field', 'product_field', 'date_field'] for field in required_fields: if field not in env or env[field] is None: print(f"✗ 环境 {env_name} 缺少必需字段: {field}") return False return True