"""Main billing analyzer module""" from collections import defaultdict import re from datetime import datetime, timedelta from pathlib import Path from typing import Any, Dict, List, Optional, Tuple, Set from ..common.config import ConfigManager from ..common.utils import parse_date_range, format_currency, safe_float from ..data.csv_handler import CSVHandler from ..export.environments import EnvironmentManager from ..export.account_manager import AccountManager from .anomaly import AnomalyDetector from .baseline import BaselineCalculator from ..notification.feishu import FeishuNotifier class BillingAnalyzer: """账单分析器""" def __init__(self, config_path: Optional[str] = None): """初始化分析器 Args: config_path: 配置文件路径 """ self.config_manager = ConfigManager(config_path) self.env_manager = EnvironmentManager(self.config_manager) self.account_manager = AccountManager(config_manager=self.config_manager) self.csv_handler = CSVHandler() self.cloud_webhook_url = self.config_manager.get_webhook_url('cloud') self.account_scoped_service_webhook_url = self.config_manager.get_webhook_url( 'accountScopedServiceMonitor' ) self.business_type_summary_webhook_url = self.config_manager.get_webhook_url( 'businessTypeSummary' ) # 获取配置 thresholds = self.config_manager.get_thresholds() # 初始化组件 self.anomaly_detector = AnomalyDetector( warning_threshold=thresholds.get('account_warning', 1.5), critical_threshold=thresholds.get('account_critical', 2.0), min_cost_threshold=thresholds.get('min_cost_threshold', 100.0) ) self.baseline_calculator = BaselineCalculator( baseline_days=30, min_sample_size=7 ) # 获取账户详情用于通知显示 account_details = self.account_manager.get_account_details() if hasattr(self.account_manager, 'get_account_details') else {} self.feishu_notifier = FeishuNotifier(account_details=account_details) # 开发/测试区域监控配置 common_regions_config = self.config_manager.get( 'settings.devTestCommonRegions', ['us-east-1', 'us-west-2', 'global'] ) or [] self.dev_test_common_regions = { str(region).strip().lower() for region in common_regions_config if str(region).strip() } self.dev_test_common_regions_display = [ str(region).strip() for region in common_regions_config if str(region).strip() ] or ['us-east-1', 'us-west-2', 'global'] if self.dev_test_common_regions_display: self.dev_test_common_regions_display = list( dict.fromkeys(self.dev_test_common_regions_display) ) environments_config = self.config_manager.get( 'settings.devTestRegionEnvironments', ['nonprod', 'shared'] ) or [] self.dev_test_region_environments: Set[str] = { str(env).strip().lower() for env in environments_config if str(env).strip() } min_cost_config = self.config_manager.get( 'settings.devTestRegionMinCost', 1.0 ) self.dev_test_region_min_cost = safe_float(min_cost_config, 1.0) self.enable_untagged_service_monitor = bool( self.config_manager.get('settings.enableAccountScopedServiceMonitor', False) ) self.enable_untagged_tag_monitor = bool( self.config_manager.get('settings.enableUntaggedTagMonitor', False) ) self.untagged_service_monitor_environments = [ str(env).strip().lower() for env in self.config_manager.get( 'settings.accountScopedServiceMonitorEnvironments', ['prod'] ) or [] if str(env).strip() ] self.untagged_service_monitor_services = [ str(service).strip() for service in self.config_manager.get( 'settings.accountScopedServiceMonitorServices', [] ) or [] if str(service).strip() ] self.untagged_service_monitor_warning = thresholds.get( 'service_warning', thresholds.get('account_warning', 1.5) ) self.untagged_service_monitor_critical = thresholds.get( 'service_critical', thresholds.get('account_critical', 2.0) ) self.untagged_service_monitor_min_cost = safe_float( thresholds.get('min_cost_threshold', 100.0), 100.0 ) self.untagged_service_monitor_min_total_cost = safe_float( self.config_manager.get( 'settings.accountScopedServiceMonitorMinTotalCost', 5.0 ), 5.0 ) self.untagged_tag_monitor_environments = [ str(env).strip().lower() for env in self.config_manager.get( 'settings.untaggedTagMonitorEnvironments', ['prod'] ) or [] if str(env).strip() ] self.untagged_tag_monitor_accounts = [ str(account).strip() for account in self.config_manager.get( 'settings.untaggedTagMonitorAccounts', [] ) or [] if str(account).strip() ] self.untagged_tag_monitor_account_set = set(self.untagged_tag_monitor_accounts) self.untagged_tag_monitor_min_cost = safe_float( self.config_manager.get( 'settings.untaggedTagMonitorMinCost', 0.0 ), 0.0 ) self.untagged_tag_monitor_ignore_services = [ str(service).strip() for service in self.config_manager.get( 'settings.untaggedTagMonitorIgnoredServices', [] ) or [] if str(service).strip() ] self.untagged_tag_monitor_ignore_map = set( self._normalize_service_name(service) for service in self.untagged_tag_monitor_ignore_services ) self.untagged_service_monitor_map = self._normalize_service_list( self.untagged_service_monitor_services ) # 分析结果 self.analysis_results = {} def analyze(self, environments: List[str] = None, target_date: datetime = None, account_ids: List[str] = None, dry_run: bool = False, send_notifications: bool = True) -> Dict: """执行分析任务 Args: environments: 要分析的环境列表 target_date: 目标分析日期 account_ids: 指定账户ID列表 dry_run: 试运行模式 send_notifications: 是否发送通知 Returns: 分析结果字典 """ # 默认分析前天的数据 if target_date is None: target_date = datetime.now() - timedelta(days=2) # 确定要分析的环境 if not environments: environments = self.env_manager.list_environments() print(f"\n{'='*60}") print(f"AWS 账单异常分析") print(f"{'='*60}") print(f"分析日期: {target_date.strftime('%Y-%m-%d')}") print(f"环境: {', '.join(environments)}") print(f"试运行模式: {'是' if dry_run else '否'}") print(f"发送通知: {'是' if send_notifications and not dry_run else '否'}") # 重置分析结果 self.analysis_results = { 'analysis_date': target_date.strftime('%Y-%m-%d'), 'environments': environments, 'account_analyses': {}, 'environment_summaries': {}, 'overall_summary': {}, 'notifications_sent': [] } # 分析每个环境 for env_name in environments: if not self.env_manager.validate_environment(env_name): continue print(f"\n分析环境: {env_name}") print("-" * 40) env_results = self._analyze_environment( env_name, target_date, account_ids, dry_run, send_notifications ) self.analysis_results['environment_summaries'][env_name] = env_results # 发送环境级通知（支持合并特定环境） if send_notifications and not dry_run: self._dispatch_environment_notifications(target_date) self._send_untagged_service_monitor(target_date) self._send_untagged_tag_monitor(target_date) # 生成整体汇总 self._generate_overall_summary() # 发送汇总通知 if send_notifications and not dry_run: self._send_summary_notifications() # 打印分析结果 self._print_analysis_summary() return self.analysis_results def _analyze_environment(self, env_name: str, target_date: datetime, account_ids: List[str] = None, dry_run: bool = False, send_notifications: bool = True) -> Dict: """分析单个环境 Args: env_name: 环境名称 target_date: 目标日期 account_ids: 指定账户列表 dry_run: 试运行模式 send_notifications: 是否发送通知 Returns: 环境分析结果 """ # 获取账户列表 if account_ids: accounts = [{'account_id': aid, 'environment': env_name} for aid in account_ids] else: validate_tables = True if env_name == 'prod': cutover_date_str = self.config_manager.get( 'settings.awsBillingCutoverDate', '2025-10-01' ) try: cutover_date = datetime.strptime(cutover_date_str, '%Y-%m-%d').date() except ValueError: cutover_date = datetime(2025, 10, 1).date() if target_date.date() >= cutover_date: validate_tables = False accounts = self.account_manager.get_accounts_by_environment( env_name, validate_tables=validate_tables ) if not accounts: print(f"⚠️ 环境 {env_name} 没有找到账户") return {'accounts': [], 'anomalies': [], 'summary': {}} print(f"发现账户: {len(accounts)} 个") env_anomalies = [] account_results = {} # 分析每个账户 for account in accounts: account_id = account['account_id'] try: account_result = self._analyze_account( env_name, account_id, target_date, dry_run, send_notifications ) account_results[account_id] = account_result if account_result['anomalies']: env_anomalies.extend(account_result['anomalies']) print(f" ✓ {account_id}: {len(account_result['anomalies'])} 个异常") else: print(f" ✓ {account_id}: 无异常") except Exception as e: error_msg = f"账户 {account_id} 分析失败: {e}" print(f" ✗ {error_msg}") account_results[account_id] = {'error': error_msg} # 计算当月成本汇总 monthly_costs = self._calculate_monthly_costs(env_name, accounts, target_date) # 生成环境汇总 env_summary = self._generate_environment_summary( env_name, account_results, env_anomalies ) env_summary['monthly_costs'] = monthly_costs region_monitor = None if env_name in self.dev_test_region_environments: region_monitor = self._analyze_dev_test_regions(env_name, account_results, target_date) if region_monitor: env_summary['region_monitor'] = region_monitor return { 'accounts': account_results, 'anomalies': env_anomalies, 'summary': env_summary, 'region_monitor': region_monitor } def _dispatch_environment_notifications(self, target_date: datetime): """根据环境结果发送通知，支持按需合并""" env_summaries = self.analysis_results.get('environment_summaries', {}) if not env_summaries: return ordered_envs = self.analysis_results.get('environments', []) def handle_environment(env_name: str): if env_name not in env_summaries: return env_data = env_summaries[env_name] anomalies = list(env_data.get('anomalies', [])) account_results = dict(env_data.get('accounts', {})) monthly_costs = env_data.get('summary', {}).get('monthly_costs') region_monitor = env_data.get('region_monitor') self._send_environment_notifications( env_name, anomalies, account_results, target_date, monthly_costs, region_monitor=region_monitor ) if env_name == 'prod': self._send_prod_business_type_summary( account_results, target_date ) for env_name in ordered_envs: handle_environment(env_name) for env_name in env_summaries.keys(): if env_name not in ordered_envs: handle_environment(env_name) def _analyze_account(self, env_name: str, account_id: str, target_date: datetime, dry_run: bool = False, send_notifications: bool = True) -> Dict: """分析单个账户 Args: env_name: 环境名称 account_id: 账户ID target_date: 目标日期 dry_run: 试运行模式 send_notifications: 是否发送通知 Returns: 账户分析结果 """ # 加载当前日期数据 current_data = self._load_account_data(env_name, account_id, target_date) if not current_data: return {'current_data': [], 'historical_data': [], 'anomalies': []} # 加载历史数据 historical_data = self._load_historical_data(env_name, account_id, target_date) # 执行基础异常检测 anomalies = self._detect_basic_anomalies( current_data, historical_data, account_id ) # 不在账户级发送通知，而是在环境级汇总发送 return { 'current_data': current_data, 'historical_data': historical_data, 'anomalies': anomalies, 'analysis_date': target_date.strftime('%Y-%m-%d') } def _get_account_month_file(self, env_name: str, account_id: str, target_date: datetime) -> Optional[Path]: """获取账户当月CSV文件路径""" output_dir = self.env_manager.get_output_dir(env_name) if not output_dir: return None env_config = self.env_manager.get_environment(env_name) file_prefix = env_config.get('file_prefix', 'aws') if env_config else 'aws' year_month = target_date.strftime('%Y-%m') filename = f"{file_prefix}_{account_id}_{year_month}.csv" return output_dir / filename def _load_account_month_records(self, env_name: str, account_id: str, target_date: datetime) -> List[Dict]: """加载账户当月的全部记录""" file_path = self._get_account_month_file(env_name, account_id, target_date) if not file_path or not file_path.exists(): return [] return self.csv_handler.read_file(str(file_path)) def _load_account_data(self, env_name: str, account_id: str, target_date: datetime) -> List[Dict]: """加载账户当天数据 Args: env_name: 环境名称 account_id: 账户ID target_date: 目标日期 Returns: 数据记录列表 """ # 读取当月所有记录 all_records = self._load_account_month_records(env_name, account_id, target_date) if not all_records: return [] # 过滤目标日期的数据 target_date_str = target_date.strftime('%Y-%m-%d') target_records = [] for record in all_records: if record.get('date') == target_date_str: target_records.append(record) return target_records def _load_historical_data(self, env_name: str, account_id: str, target_date: datetime) -> List[Dict]: """加载历史数据 Args: env_name: 环境名称 account_id: 账户ID target_date: 目标日期 Returns: 历史数据记录列表 """ output_dir = self.env_manager.get_output_dir(env_name) return self.baseline_calculator.load_historical_data( [output_dir], account_id, target_date, env_name ) def _detect_basic_anomalies(self, current_data: List[Dict], historical_data: List[Dict], account_id: str) -> List[Dict]: """检测基础异常（账户级和服务级） Args: current_data: 当前数据 historical_data: 历史数据 account_id: 账户ID Returns: 异常列表 """ anomalies = [] if not historical_data: return anomalies # 计算当前成本 current_total_cost = sum(safe_float(record.get('cost', 0)) for record in current_data) current_service_costs = {} for record in current_data: service = record.get('product', 'Unknown') cost = safe_float(record.get('cost', 0)) current_service_costs[service] = current_service_costs.get(service, 0) + cost # 计算基线 account_baseline = self.anomaly_detector.calculate_account_baseline(historical_data) service_baselines = self.anomaly_detector.calculate_service_baselines(historical_data) # 检测账户级异常 account_anomaly = self.anomaly_detector.detect_account_anomalies( current_total_cost, account_baseline, account_id ) if account_anomaly: anomalies.append(account_anomaly) # 检测服务级异常 service_anomalies = self.anomaly_detector.detect_service_anomalies( current_service_costs, service_baselines, account_id ) anomalies.extend(service_anomalies) return anomalies def _calculate_monthly_costs(self, env_name: str, accounts: List[Dict], target_date: datetime) -> Dict: """计算当月成本汇总（从月初到目标日期） Args: env_name: 环境名称 accounts: 账户列表 target_date: 目标日期 Returns: 月度成本汇总数据 """ output_dir = self.env_manager.get_output_dir(env_name) if not output_dir: return {} # 获取环境配置以确定文件前缀 env_config = self.env_manager.get_environment(env_name) file_prefix = env_config.get('file_prefix', 'aws') if env_config else 'aws' # 月初日期 month_start = target_date.replace(day=1) year_month = target_date.strftime('%Y-%m') # 初始化月度成本数据 monthly_data = { 'month': year_month, 'start_date': month_start.strftime('%Y-%m-%d'), 'end_date': target_date.strftime('%Y-%m-%d'), 'total_cost': 0, 'account_costs': {}, # {account_id: cost} 'service_costs': {}, # {service: cost} } # 计算每个账户的月度成本 for account in accounts: account_id = account['account_id'] filename = f"{file_prefix}_{account_id}_{year_month}.csv" file_path = output_dir / filename if not file_path.exists(): continue # 读取整月数据 all_records = self.csv_handler.read_file(str(file_path)) # 累计月度成本 account_monthly_cost = 0 for record in all_records: record_date_str = record.get('date') if not record_date_str: continue # 检查日期是否在范围内（月初到目标日期） try: record_date = datetime.strptime(record_date_str, '%Y-%m-%d') if month_start <= record_date <= target_date: cost = safe_float(record.get('cost', 0)) account_monthly_cost += cost # 累计服务成本 service = record.get('product', 'Unknown') monthly_data['service_costs'][service] = monthly_data['service_costs'].get(service, 0) + cost except ValueError: continue # 保存账户成本 if account_monthly_cost > 0: monthly_data['account_costs'][account_id] = account_monthly_cost monthly_data['total_cost'] += account_monthly_cost # 按成本从高到低排序账户 sorted_accounts = sorted(monthly_data['account_costs'].items(), key=lambda x: x[1], reverse=True) monthly_data['sorted_accounts'] = sorted_accounts # 按成本从高到低排序服务 sorted_services = sorted(monthly_data['service_costs'].items(), key=lambda x: x[1], reverse=True) monthly_data['sorted_services'] = sorted_services[:10] # Top 10 services # 打印月度成本摘要 print(f"\n月度成本汇总 ({month_start.strftime('%Y-%m-%d')} 至 {target_date.strftime('%Y-%m-%d')})") print("-" * 40) print(f"总成本: {format_currency(monthly_data['total_cost'])}") print(f"账户数: {len(monthly_data['account_costs'])}") if sorted_accounts: print("\nTop 5 账户成本:") for i, (acc_id, cost) in enumerate(sorted_accounts[:5], 1): acc_display = self._get_simple_account_name(acc_id) print(f" {i}. {acc_display}: {format_currency(cost)}") return monthly_data def _analyze_dev_test_regions(self, env_name: str, account_results: Dict[str, Dict], target_date: datetime) -> Dict[str, Any]: """分析开发/测试账户在非常用区域的费用情况""" total_accounts = len(account_results) region_totals: Dict[str, float] = defaultdict(float) account_region_costs: Dict[str, Dict[str, float]] = {} involved_region_keys: Set[str] = set() region_labels: Dict[str, str] = {} detail_rows: List[Dict[str, Any]] = [] has_region_data = False mtd_region_totals: Dict[Tuple[str, str], float] = defaultdict(float) if total_accounts == 0: return { 'total_accounts': 0, 'abnormal_accounts': 0, 'regions': [], 'details': [], 'top_regions': [], 'top_regions_mtd': [], 'common_regions_display': self.dev_test_common_regions_display, 'threshold': self.dev_test_region_min_cost, 'has_region_data': False, 'region_labels': {} } month_start = target_date.replace(day=1) for account_id, result in account_results.items(): current_data = result.get('current_data') or [] if not current_data: # 即使当日无数据，也尝试累加当月区域费用 month_records = self._load_account_month_records(env_name, account_id, target_date) if self._accumulate_mtd_region_costs( month_records, month_start, target_date, account_id, region_labels, mtd_region_totals ): has_region_data = True continue region_accumulator: Dict[str, float] = defaultdict(float) for record in current_data: region_raw = (record.get('region') or record.get('product_region') or '').strip() if not region_raw: continue cost = safe_float(record.get('cost', 0)) if cost <= 0: continue region_key = region_raw.lower() region_labels.setdefault(region_key, region_raw) region_totals[region_key] += cost has_region_data = True if region_key not in self.dev_test_common_regions: region_accumulator[region_key] += cost filtered_regions = { region: cost for region, cost in region_accumulator.items() if cost >= self.dev_test_region_min_cost } if filtered_regions: account_region_costs[account_id] = filtered_regions involved_region_keys.update(filtered_regions.keys()) month_records = self._load_account_month_records(env_name, account_id, target_date) if self._accumulate_mtd_region_costs( month_records, month_start, target_date, account_id, region_labels, mtd_region_totals ): has_region_data = True for account_id, regions in account_region_costs.items(): sorted_regions = sorted(regions.items(), key=lambda item: item[1], reverse=True) for region_key, cost in sorted_regions: region_label = region_labels.get(region_key, region_key) detail_rows.append({ 'account_id': account_id, 'region': region_label, 'region_key': region_key, 'cost': cost }) detail_rows.sort(key=lambda item: item['cost'], reverse=True) non_common_region_totals = [ (region, cost) for region, cost in region_totals.items() if region not in self.dev_test_common_regions ] sorted_non_common_regions = sorted( non_common_region_totals, key=lambda item: item[1], reverse=True ) top_regions_display = [ (region_labels.get(region, region), cost) for region, cost in sorted_non_common_regions[:5] ] involved_regions = [ region_labels.get(region, region) for region in sorted(involved_region_keys) ] top_mtd_entries = sorted( mtd_region_totals.items(), key=lambda item: item[1], reverse=True )[:5] top_regions_mtd = [ { 'account_id': account_id, 'region': region_labels.get(region_key, region_key), 'region_key': region_key, 'cost': cost } for (account_id, region_key), cost in top_mtd_entries ] if not has_region_data and mtd_region_totals: has_region_data = True return { 'total_accounts': total_accounts, 'abnormal_accounts': len(account_region_costs), 'regions': involved_regions, 'details': detail_rows, 'top_regions': top_regions_display, 'top_regions_mtd': top_regions_mtd, 'common_regions_display': self.dev_test_common_regions_display, 'threshold': self.dev_test_region_min_cost, 'has_region_data': has_region_data, 'region_labels': region_labels } def _accumulate_mtd_region_costs( self, records: List[Dict[str, Any]], month_start: datetime, target_date: datetime, account_id: str, region_labels: Dict[str, str], accumulator: Dict[Tuple[str, str], float] ) -> bool: """累计单账户在非常用区域的当月费用""" if not records: return False updated = False for record in records: date_str = record.get('date') if not date_str: continue try: record_date = datetime.strptime(date_str, '%Y-%m-%d') except ValueError: continue if not (month_start <= record_date <= target_date): continue region_raw = (record.get('region') or record.get('product_region') or '').strip() if not region_raw: continue region_key = region_raw.lower() if region_key in self.dev_test_common_regions: continue cost = safe_float(record.get('cost', 0)) if cost <= 0: continue region_labels.setdefault(region_key, region_raw) accumulator[(account_id, region_key)] += cost updated = True return updated def _get_simple_account_name(self, account_id: str) -> str: """获取账户简单名称 Args: account_id: 账户ID Returns: 账户简单名称 """ if hasattr(self.account_manager, 'get_account_details'): account_details = self.account_manager.get_account_details() if account_id in account_details: account_info = account_details[account_id] if hasattr(account_info, 'account_name'): return f"{account_id} ({account_info.account_name})" elif 'account_name' in account_info: return f"{account_id} ({account_info.get('account_name', 'Unknown')})" return account_id def _send_environment_notifications(self, env_name: str, anomalies: List[Dict], account_results: Dict, target_date: datetime, monthly_costs: Optional[Dict] = None, region_monitor: Optional[Dict[str, Any]] = None): """发送环境级汇总通知 Args: env_name: 环境名称 anomalies: 所有异常列表 account_results: 账户分析结果 target_date: 目标日期 monthly_costs: 月度成本汇总数据 """ # 分离基础异常 basic_anomalies = [a for a in anomalies if a.get('type') != 'tag'] # 发送环境汇总通知 - 无论是否有异常都发送 webhook_url = self._get_webhook_url(env_name) if webhook_url: success = self.feishu_notifier.send_environment_summary( env_name, basic_anomalies, account_results, target_date.strftime('%Y-%m-%d'), webhook_url ) if success: self.analysis_results['notifications_sent'].append({ 'type': 'environment_summary', 'environment': env_name, 'anomaly_count': len(basic_anomalies), 'account_count': len([aid for aid, result in account_results.items() if result.get('anomalies')]) }) # 开发/测试区域监控通知 if env_name in self.dev_test_region_environments and region_monitor is not None: region_webhook = self._get_region_monitor_webhook_url(env_name) if region_webhook: success = self.feishu_notifier.send_dev_test_region_monitor( env_name, region_monitor, target_date.strftime('%Y-%m-%d'), region_webhook ) if success: self.analysis_results['notifications_sent'].append({ 'type': 'region_monitor', 'environment': env_name, 'abnormal_accounts': region_monitor.get('abnormal_accounts', 0), 'regions': region_monitor.get('regions', []) }) # 发送月度成本汇总通知（独立通知）- 仅在nonprod环境启用 if monthly_costs and env_name == 'nonprod': monthly_webhook_url = self._get_monthly_cost_webhook_url(env_name) if monthly_webhook_url: success = self.feishu_notifier.send_monthly_cost_summary( env_name, monthly_costs, target_date.strftime('%Y-%m-%d'), monthly_webhook_url ) if success: self.analysis_results['notifications_sent'].append({ 'type': 'monthly_cost_summary', 'environment': env_name, 'total_cost': monthly_costs['total_cost'], 'account_count': len(monthly_costs.get('account_costs', {})) }) @staticmethod def _normalize_service_name(service_name: str) -> str: """标准化服务名称，便于与配置列表匹配""" if not service_name: return '' return re.sub(r'[^a-z0-9]+', '', service_name.strip().lower()) def _normalize_service_list(self, services: List[str]) -> Dict[str, str]: """构建标准化服务名称映射表""" normalized_map: Dict[str, str] = {} for service in services: normalized = self._normalize_service_name(service) if not normalized or normalized in normalized_map: continue normalized_map[normalized] = service return normalized_map def _aggregate_monitored_service_costs(self, records: List[Dict]) -> Dict[str, float]: """聚合配置服务的当天成本""" if not self.untagged_service_monitor_map: return {} costs = defaultdict(float) for record in records: service_raw = (record.get('product') or '').strip() if not service_raw: continue normalized = self._normalize_service_name(service_raw) display_name = self.untagged_service_monitor_map.get(normalized) if not display_name: continue costs[display_name] += safe_float(record.get('cost', 0)) return dict(costs) def _calculate_monitored_service_baselines(self, records: List[Dict]) -> Dict[str, Dict[str, float]]: """计算配置服务的历史基线均值""" if not self.untagged_service_monitor_map: return {} daily_costs = defaultdict(lambda: defaultdict(float)) for record in records: service_raw = (record.get('product') or '').strip() if not service_raw: continue normalized = self._normalize_service_name(service_raw) display_name = self.untagged_service_monitor_map.get(normalized) if not display_name: continue record_date = record.get('date') if not record_date: continue daily_costs[display_name][record_date] += safe_float(record.get('cost', 0)) baselines: Dict[str, Dict[str, float]] = {} for service, per_day in daily_costs.items(): costs = [cost for cost in per_day.values() if cost > 0] if not costs: continue baselines[service] = { 'avg_cost': sum(costs) / len(costs), 'sample_size': len(costs) } return baselines def _format_untagged_service_cell(self, current_cost: float, baseline: Optional[Dict[str, float]]) -> str: """格式化不可打标签服务监控单元格""" if current_cost <= 0 and not baseline: return "-" if baseline and baseline.get('avg_cost', 0) > 0: avg_cost = baseline['avg_cost'] ratio = current_cost / avg_cost if avg_cost > 0 else 0 if current_cost >= self.untagged_service_monitor_min_cost: if ratio >= self.untagged_service_monitor_critical: status = "🚨" elif ratio >= self.untagged_service_monitor_warning: status = "⚠️" else: status = "✅" else: status = "ℹ️" return f"{status} {format_currency(current_cost)} ({ratio:.1f}x)" if current_cost > 0: return f"ℹ️ {format_currency(current_cost)} (no baseline)" return "-" def _send_untagged_service_monitor(self, target_date: datetime): """发送账户级服务监控通知（不可打标签）""" if not self.enable_untagged_service_monitor: return webhook_url = self.account_scoped_service_webhook_url or self.cloud_webhook_url if not webhook_url: return if not self.untagged_service_monitor_services: return env_summaries = self.analysis_results.get('environment_summaries', {}) monitor_envs = [env.lower() for env in self.untagged_service_monitor_environments] included_envs = [] rows = [] for env_name in monitor_envs: env_data = env_summaries.get(env_name) if not env_data: continue included_envs.append(env_name) account_results = env_data.get('accounts', {}) for account_id, result in account_results.items(): current_data = result.get('current_data', []) or [] historical_data = result.get('historical_data', []) or [] current_service_costs = self._aggregate_monitored_service_costs(current_data) baselines = self._calculate_monitored_service_baselines(historical_data) total_cost = sum(current_service_costs.values()) if total_cost < self.untagged_service_monitor_min_total_cost: continue service_cells = {} for service_name in self.untagged_service_monitor_services: current_cost = current_service_costs.get(service_name, 0) baseline = baselines.get(service_name) service_cells[service_name] = self._format_untagged_service_cell( current_cost, baseline ) rows.append({ 'account_id': account_id, 'total_cost': total_cost, 'service_cells': service_cells }) if not rows: return rows.sort(key=lambda item: item['total_cost'], reverse=True) success = self.feishu_notifier.send_untagged_service_monitor( analysis_date=target_date.strftime('%Y-%m-%d'), environments=included_envs, services=self.untagged_service_monitor_services, rows=rows, webhook_url=webhook_url ) if success: self.analysis_results['notifications_sent'].append({ 'type': 'untagged_service_monitor', 'environments': included_envs, 'account_count': len(rows), 'service_count': len(self.untagged_service_monitor_services) }) def _send_untagged_tag_monitor(self, target_date: datetime): """发送漏打标签资源监控通知（业务类型标签）""" if not self.enable_untagged_tag_monitor: return webhook_url = self.business_type_summary_webhook_url or self.cloud_webhook_url if not webhook_url: return cutover_date_str = self.config_manager.get( 'settings.awsBillingCutoverDate', '2025-10-01' ) try: cutover_date = datetime.strptime(cutover_date_str, '%Y-%m-%d').date() except ValueError: cutover_date = datetime(2025, 10, 1).date() if target_date.date() < cutover_date: return env_summaries = self.analysis_results.get('environment_summaries', {}) monitor_envs = [ env.lower() for env in (self.untagged_tag_monitor_environments or ['prod']) ] included_envs: List[str] = [] rows: List[Dict[str, Any]] = [] total_untagged_cost = 0.0 for env_name in monitor_envs: env_data = env_summaries.get(env_name) if not env_data: continue included_envs.append(env_name) account_results = env_data.get('accounts', {}) or {} for account_id, result in account_results.items(): if self.untagged_tag_monitor_account_set: if account_id not in self.untagged_tag_monitor_account_set: continue current_data = result.get('current_data', []) or [] if not current_data: continue account_total = 0.0 per_service_untagged: Dict[str, float] = defaultdict(float) for record in current_data: cost = safe_float(record.get('cost', 0)) if cost == 0: continue account_total += cost usedby = (record.get('tag_usedby') or '').strip() servicetype = (record.get('tag_servicetype') or '').strip() if not usedby and not servicetype: product = (record.get('product') or 'Unknown').strip() or 'Unknown' if self._normalize_service_name(product) in self.untagged_tag_monitor_ignore_map: continue per_service_untagged[product] += cost if not per_service_untagged: continue if account_total <= 0: continue for product, untagged_cost in per_service_untagged.items(): if untagged_cost <= self.untagged_tag_monitor_min_cost: continue rows.append({ 'environment': env_name, 'account_id': account_id, 'product': product, 'untagged_cost': untagged_cost, 'total_cost': account_total }) total_untagged_cost += untagged_cost if not rows: return rows.sort( key=lambda item: (item['account_id'], -item['untagged_cost']) ) success = self.feishu_notifier.send_untagged_tag_monitor( analysis_date=target_date.strftime('%Y-%m-%d'), environments=included_envs, rows=rows, total_untagged_cost=total_untagged_cost, webhook_url=webhook_url ) if success: self.analysis_results['notifications_sent'].append({ 'type': 'untagged_tag_monitor', 'environments': included_envs, 'account_count': len(rows), 'total_untagged_cost': total_untagged_cost }) def _build_prod_business_type_summary( self, account_results: Dict[str, Dict], target_date: datetime ) -> Dict[str, Any]: """聚合生产账户业务类型标签成本（按日）""" cutover_date_str = self.config_manager.get( 'settings.awsBillingCutoverDate', '2025-10-01' ) try: cutover_date = datetime.strptime(cutover_date_str, '%Y-%m-%d').date() except ValueError: cutover_date = datetime(2025, 10, 1).date() if target_date.date() < cutover_date: return { 'enabled': False, 'cutover_date': cutover_date_str, 'has_tag_data': False, 'types': [] } business_types = [ 'Global Control', 'Regional Control', 'Cloud Platform', 'DataPanel Dedicated', 'DataPanel Essential&Starter', 'DataPanel Premium' ] control_usedby = { 'dbaas-control-plane-base': 'Global Control', 'dbaas-control-plane-seed': 'Regional Control' } control_usedby_set = set(control_usedby.keys()) type_costs = { name: { 'total_cost': 0.0, 'account_costs': defaultdict(float) } for name in business_types } control_servicetype_breakdown = { 'Global Control': defaultdict(float), 'Regional Control': defaultdict(float) } has_tag_data = False for account_id, result in account_results.items(): current_data = result.get('current_data') or [] for record in current_data: usedby = (record.get('tag_usedby') or '').strip() servicetype = (record.get('tag_servicetype') or '').strip() if usedby or servicetype: has_tag_data = True cost = safe_float(record.get('cost', 0)) if cost == 0: continue matched_types = [] usedby_key = usedby.lower() if usedby else '' servicetype_key = servicetype.lower() if servicetype else '' control_type = control_usedby.get(usedby_key) if control_type: matched_types.append(control_type) if servicetype_key: control_servicetype_breakdown[control_type][servicetype_key] += cost if servicetype_key == 'platform': matched_types.append('Cloud Platform') if servicetype_key == 'dedicated' and usedby_key not in control_usedby_set: matched_types.append('DataPanel Dedicated') if servicetype_key in {'serverless', 'devtier'}: matched_types.append('DataPanel Essential&Starter') if servicetype_key in {'next-gen', 'tidb-next-gen'} and usedby_key not in control_usedby_set: matched_types.append('DataPanel Premium') for matched_type in matched_types: type_costs[matched_type]['total_cost'] += cost type_costs[matched_type]['account_costs'][account_id] += cost type_entries = [] for name in business_types: data = type_costs.get(name, {}) account_costs = [ (account_id, cost) for account_id, cost in data.get('account_costs', {}).items() if cost != 0 ] account_costs.sort(key=lambda item: item[1], reverse=True) type_entries.append({ 'name': name, 'total_cost': data.get('total_cost', 0.0), 'accounts': account_costs }) return { 'enabled': True, 'cutover_date': cutover_date_str, 'has_tag_data': has_tag_data, 'types': type_entries, 'control_servicetype_breakdown': { control_type: dict(breakdown) for control_type, breakdown in control_servicetype_breakdown.items() } } def _send_prod_business_type_summary( self, account_results: Dict[str, Dict], target_date: datetime ) -> None: """发送生产账户业务类型标签成本汇总""" business_type_summary = self._build_prod_business_type_summary( account_results, target_date ) if not business_type_summary.get('enabled'): return webhook_url = self._get_webhook_url('prod') if self.business_type_summary_webhook_url: webhook_url = self.business_type_summary_webhook_url if not webhook_url: return success = self.feishu_notifier.send_business_type_summary( analysis_date=target_date.strftime('%Y-%m-%d'), summary=business_type_summary, webhook_url=webhook_url ) if success: self.analysis_results['notifications_sent'].append({ 'type': 'business_type_summary', 'environment': 'prod', 'type_count': len(business_type_summary.get('types', [])) }) def _get_webhook_url(self, env_name: str) -> Optional[str]: """获取环境对应的webhook URL Args: env_name: 环境名称 Returns: webhook URL """ # 云级通道（唯一通道） return self.cloud_webhook_url def _get_monthly_cost_webhook_url(self, env_name: str) -> Optional[str]: """获取月度成本通知的webhook URL Args: env_name: 环境名称 Returns: 月度成本webhook URL """ return self.cloud_webhook_url def _get_region_monitor_webhook_url(self, env_name: str) -> Optional[str]: """获取区域监控通知的webhook URL""" return self.cloud_webhook_url def _generate_environment_summary(self, env_name: str, account_results: Dict, anomalies: List[Dict]) -> Dict: """生成环境汇总 Args: env_name: 环境名称 account_results: 账户分析结果 anomalies: 异常列表 Returns: 环境汇总信息 """ # 统计异常 critical_count = len([a for a in anomalies if a.get('severity') == 'critical']) warning_count = len([a for a in anomalies if a.get('severity') == 'warning']) # 统计账户 total_accounts = len(account_results) affected_accounts = len([acc for acc, result in account_results.items() if result.get('anomalies')]) # 计算总成本 total_current_cost = 0 for result in account_results.values(): if 'current_data' in result: total_current_cost += sum(safe_float(record.get('cost', 0)) for record in result['current_data']) return { 'environment': env_name, 'total_accounts': total_accounts, 'affected_accounts': affected_accounts, 'total_anomalies': len(anomalies), 'critical_anomalies': critical_count, 'warning_anomalies': warning_count, 'total_current_cost': total_current_cost } def _generate_overall_summary(self): """生成整体汇总""" env_summaries = self.analysis_results['environment_summaries'] total_accounts = sum(env['summary'].get('total_accounts', 0) for env in env_summaries.values()) total_affected = sum(env['summary'].get('affected_accounts', 0) for env in env_summaries.values()) total_anomalies = sum(len(env.get('anomalies', [])) for env in env_summaries.values()) total_critical = sum(env['summary'].get('critical_anomalies', 0) for env in env_summaries.values()) total_warning = sum(env['summary'].get('warning_anomalies', 0) for env in env_summaries.values()) total_cost = sum(env['summary'].get('total_current_cost', 0) for env in env_summaries.values()) self.analysis_results['overall_summary'] = { 'total_accounts': total_accounts, 'affected_accounts': total_affected, 'total_anomalies': total_anomalies, 'critical_anomalies': total_critical, 'warning_anomalies': total_warning, 'total_cost': total_cost, 'environments_analyzed': len(env_summaries), 'notifications_sent': len(self.analysis_results['notifications_sent']) } def _send_summary_notifications(self): """发送汇总通知（已禁用，因为环境级通知已包含足够详细信息）""" # 汇总通知已被禁用，因为环境级通知已包含Top5数据和详细分析 # 避免发送重复且信息量不足的通知 pass def _print_analysis_summary(self): """打印分析汇总""" overall = self.analysis_results['overall_summary'] print(f"\n{'='*60}") print("分析汇总") print(f"{'='*60}") print(f"分析日期: {self.analysis_results['analysis_date']}") print(f"分析环境: {', '.join(self.analysis_results['environments'])}") print(f"总账户数: {overall['total_accounts']}") print(f"异常账户: {overall['affected_accounts']}") print(f"总异常数: {overall['total_anomalies']}") print(f" - 严重: {overall['critical_anomalies']}") print(f" - 警告: {overall['warning_anomalies']}") print(f"总成本: {format_currency(overall['total_cost'])}") print(f"通知发送: {overall['notifications_sent']} 个") # 按环境显示详情 for env_name, env_data in self.analysis_results['environment_summaries'].items(): summary = env_data['summary'] print(f"\n{env_name} 环境:") print(f" 账户数: {summary['total_accounts']}") print(f" 异常数: {summary['total_anomalies']}") print(f" 成本: {format_currency(summary['total_current_cost'])}") if overall['total_anomalies'] == 0: print("\n✅ 未发现异常，所有账户成本正常") else: print(f"\n⚠️ 发现 {overall['total_anomalies']} 个异常，建议关注")