#!/usr/bin/env python3 """ AWS 统一账单数据每日分析和告警系统 功能: 1. 统一分析生产和非生产环境的账单数据，检测异常费用 2. 按账户维度分析每日总费用是否有异常 3. 按服务维度分析各服务费用是否有异常 4. 检测新出现的服务和突然停用的服务 5. 通过飞书webhook分别发送告警通知 6. 支持合并环境报告或分别发送 用法: python aws_unified_daily_analyzer.py --date 2025-08-20 # 分析指定日期的所有环境 python aws_unified_daily_analyzer.py # 分析前天的所有环境数据 python aws_unified_daily_analyzer.py --env nonprod --dry-run # 仅分析非生产环境不发送通知 python aws_unified_daily_analyzer.py --env prod --account-id 730335266318 # 分析指定生产账户 python aws_unified_daily_analyzer.py --separate-reports # 分别发送环境报告 """ import argparse import csv import json import statistics import sys from collections import defaultdict from datetime import datetime, timedelta from pathlib import Path from typing import Dict, List, Optional import requests import pandas as pd import re import urllib.parse class AWSUnifiedBillingAnalyzer: """AWS 统一账单数据分析器""" def __init__(self, account_filter: Optional[str] = None, config: Optional[Dict] = None): # 错位生产账户配置将从配置文件加载 self.misplaced_prod_accounts = {} # Engineering Group 账户详情 self.account_details = {} # 加载账户详情 self.load_account_details() # 环境配置 self.environments = { 'nonprod': { 'data_dir': Path('aws_billing_exports'), 'file_prefix': 'aws', 'display_name': 'AWS Dev/Test Accounts', 'icon': '🟠', 'color_normal': 'green', 'color_warning': 'orange', 'color_critical': 'red', # 非生产环境相对宽松的阈值 'thresholds': { 'account_warning': 1.5, # 150% 'account_critical': 2.0, # 200% 'service_warning': 2.0, # 200% 'service_critical': 3.0, # 300% 'min_cost_threshold': 1.0, 'new_service_threshold': 5.0 } }, 'prod': { 'data_dir': Path('aws_prod_billing_exports'), 'file_prefix': 'aws_prod', 'display_name': 'AWS Production Accounts', 'icon': '🔴', 'color_normal': 'green', 'color_warning': 'orange', 'color_critical': 'red', # 生产环境相对保守的阈值 'thresholds': { 'account_warning': 1.3, # 130% 'account_critical': 1.6, # 160% 'service_warning': 1.5, # 150% 'service_critical': 2.0, # 200% 'min_cost_threshold': 5.0, 'new_service_threshold': 20.0, # no-tenant维度异常阈值 'no_tenant_warning': 1.5, # 150% 'no_tenant_critical': 2.0, # 200% 'min_no_tenant_cost_threshold': 10.0, 'new_no_tenant_threshold': 20.0, 'total_no_tenant_threshold': 100.0 } }, 'misplaced_prod': { 'data_dir': Path('aws_misplaced_prod_billing_exports'), 'file_prefix': 'aws_misplaced_prod', 'display_name': 'AWS Misplaced Production Accounts', 'icon': '🟡', 'color_normal': 'green', 'color_warning': 'orange', 'color_critical': 'red', # 使用生产环境的阈值 'thresholds': { 'account_warning': 1.3, # 130% 'account_critical': 1.6, # 160% 'service_warning': 1.5, # 150% 'service_critical': 2.0, # 200% 'min_cost_threshold': 5.0, 'new_service_threshold': 20.0 } }, 'shared': { 'data_dir': Path('aws_shared_billing_exports'), 'file_prefix': 'aws_shared', 'display_name': 'AWS Shared Test Accounts', 'icon': '🟢', 'color_normal': 'green', 'color_warning': 'orange', 'color_critical': 'red', # 公共测试账户使用中等阈值 'thresholds': { 'account_warning': 1.4, # 140% 'account_critical': 1.8, # 180% 'service_warning': 1.8, # 180% 'service_critical': 2.5, # 250% 'min_cost_threshold': 2.0, 'new_service_threshold': 10.0, # tag维度异常阈值 'tag_warning': 2.0, # 200% 'tag_critical': 3.0, # 300% 'min_tag_cost_threshold': 1.0, 'new_tag_threshold': 5.0 } } } self.account_filter = account_filter self.anomaly_detector = AnomalyDetector() # 初始化默认忽略服务列表（非资源费用） self.ignored_services = { 'ComputeSavingsPlans', 'Savings Plans for Compute', 'EC2-Instance Savings Plans', 'SageMaker Savings Plans', 'AWS Support (Business)', 'AWS Support (Enterprise)', 'Tax' } # 从配置更新设置（包括可能的忽略服务列表更新） if config: self._update_from_config(config) def load_account_details(self): """加载Engineering Group账户详情""" try: account_details_file = Path(__file__).parent / 'aws_account_details.csv' if not account_details_file.exists(): print(f"警告：账户详情文件不存在: {account_details_file}") return df = pd.read_csv(account_details_file) for _, row in df.iterrows(): account_id = str(row['account_id']).strip() # 处理环境字段的中英文映射 env_raw = row['environment'].strip() if env_raw == '生产': environment = 'prod' elif env_raw == '开发测试': environment = 'nonprod' else: environment = env_raw.lower() # 其他情况转为小写 self.account_details[account_id] = { 'account_name': row['account_name'].strip(), 'owner_email': row['owner_email'].strip(), 'environment': environment, 'is_shared_dev_account': str(row['is_shared_dev_account']).strip() == '是' } print(f"已加载 {len(self.account_details)} 个Engineering Group账户详情") except Exception as e: print(f"加载账户详情失败: {e}") self.account_details = {} def get_account_display_name(self, account_id: str) -> str: """获取账户友好显示名称""" # 如果账户ID包含环境前缀，提取纯账户ID pure_account_id = account_id if '_' in account_id and account_id.split('_')[0] in ['nonprod', 'prod']: pure_account_id = account_id.split('_', 1)[1] if pure_account_id in self.account_details: account_info = self.account_details[pure_account_id] name = account_info['account_name'] email = account_info['owner_email'] shared_tag = " [公共测试账户]" if account_info['is_shared_dev_account'] else "" return f"`{pure_account_id}` **{name}**{shared_tag} ({email})" return f"`{pure_account_id}` **Unknown-{pure_account_id}** (unknown@pingcap.com)" def get_account_info(self, account_id: str) -> Dict: """获取账户详细信息""" if account_id in self.account_details: return self.account_details[account_id] return { 'account_name': f'Unknown-{account_id}', 'owner_email': 'unknown@pingcap.com', 'environment': 'unknown', 'is_shared_dev_account': False } def is_engineering_account(self, account_id: str) -> bool: """检查账户是否属于Engineering Group""" return account_id in self.account_details def _update_from_config(self, config: Dict): """从配置文件更新设置""" # 更新阈值配置 if 'thresholds' in config: custom_thresholds = config['thresholds'] for env in self.environments: self.environments[env]['thresholds'].update(custom_thresholds) # 更新数据目录配置 if 'data_directories' in config: data_dirs = config['data_directories'] for env in self.environments: if env in data_dirs: old_dir = self.environments[env]['data_dir'] new_dir = Path(data_dirs[env]) self.environments[env]['data_dir'] = new_dir print(f"更新 {env} 环境数据目录: {old_dir} -> {new_dir}") # 更新生产环境特殊配置 if 'prod_accounts' in config: # 支持配置文件中指定生产账户列表 self.prod_accounts = config['prod_accounts'] else: # 默认生产账户列表（基于文档分析） self.prod_accounts = { '143458967504': '143458967504_aws_billing_data', '632330501419': '632330501419_aws_billing_data', '730335266318': '730335266318_aws_billing_data', '380838443567': 'aws_prod_billing_data' # 特殊表名 } # 确保 prod_accounts 属性总是存在 if not hasattr(self, 'prod_accounts'): self.prod_accounts = { '143458967504': '143458967504_aws_billing_data', '632330501419': '632330501419_aws_billing_data', '730335266318': '730335266318_aws_billing_data', '380838443567': 'aws_prod_billing_data' } print(f"使用默认生产账户配置: {len(self.prod_accounts)} 个账户") else: print(f"使用配置文件中的生产账户配置: {len(self.prod_accounts)} 个账户") # 更新忽略服务列表 if 'ignored_services' in config: if isinstance(config['ignored_services'], list): # 配置文件中的服务列表会添加到默认列表中 self.ignored_services.update(config['ignored_services']) elif isinstance(config['ignored_services'], dict): self.ignored_services.update(config['ignored_services'].keys()) elif isinstance(config['ignored_services'], set): self.ignored_services.update(config['ignored_services']) print(f"应用配置后的忽略服务列表: {len(self.ignored_services)} 个服务") # 加载错位生产账户配置 if 'misplaced_prod_accounts' in config: misplaced_config = config['misplaced_prod_accounts'] for account_id, account_info in misplaced_config.items(): # 只加载启用的账户 if account_info.get('enabled', True): self.misplaced_prod_accounts[account_id] = account_info.get('description', account_id) if self.misplaced_prod_accounts: print(f"从配置文件加载了 {len(self.misplaced_prod_accounts)} 个错位生产账户:") for account_id, description in self.misplaced_prod_accounts.items(): print(f" - {account_id}: {description}") else: print("配置文件中未发现启用的错位生产账户") else: print("配置文件中未找到错位生产账户配置段") def load_billing_data(self, environment: str, target_date: str) -> Dict[str, List[Dict]]: """加载指定环境的账单数据""" env_config = self.environments[environment] target_month = target_date[:7] # YYYY-MM account_data = {} # 根据环境加载不同的数据源 if environment == 'prod': # 生产环境需要从两个位置加载数据：正常生产目录 + 错位生产账户专门目录 account_data.update(self._load_normal_prod_data(target_month)) account_data.update(self._load_misplaced_prod_from_dedicated_dir(target_month)) elif environment == 'shared': # 公共测试账户，从专门目录加载包含tag数据的账单 account_data = self._load_shared_data(target_month) else: # 非生产环境，排除错位的生产账户 account_data = self._load_nonprod_data(target_month) return account_data def _load_shared_data(self, target_month: str) -> Dict[str, List[Dict]]: """加载公共测试账户数据（包含tag信息）""" env_config = self.environments['shared'] account_data = {} # 确定文件模式 if self.account_filter: csv_pattern = f"{env_config['file_prefix']}_{self.account_filter}_{target_month}.csv" csv_files = list(env_config['data_dir'].glob(csv_pattern)) if not csv_files: print(f"未找到公共测试账户 {self.account_filter} 在 {target_month} 的数据文件") return {} else: csv_files = list(env_config['data_dir'].glob(f"{env_config['file_prefix']}_*_{target_month}.csv")) for csv_file in csv_files: account_id = self._extract_account_id_from_shared_file(csv_file) # 只处理Engineering Group的账户 if account_id and not self.is_engineering_account(account_id): print(f"跳过非Engineering Group公共测试账户: {account_id}") continue if account_id and (not self.account_filter or account_id == self.account_filter): data = self._load_csv_file(csv_file, account_id, 'shared') if data: account_data[account_id] = data return account_data def _extract_account_id_from_shared_file(self, csv_file: Path) -> Optional[str]: """从公共测试账户文件名中提取账户ID""" try: # aws_shared_385595570414_2025-08.csv -> 385595570414 parts = csv_file.stem.split('_') if len(parts) >= 4 and parts[0] == 'aws' and parts[1] == 'shared': return parts[2] # account_id return None except Exception as e: print(f"从公共测试文件名 {csv_file.name} 提取账户ID失败: {e}") return None def _load_normal_prod_data(self, target_month: str) -> Dict[str, List[Dict]]: """加载正常生产环境数据""" env_config = self.environments['prod'] account_data = {} # 确定文件模式 if self.account_filter: csv_pattern = f"{env_config['file_prefix']}_{self.account_filter}_{target_month}.csv" csv_files = list(env_config['data_dir'].glob(csv_pattern)) if not csv_files: print(f"未找到生产环境账户 {self.account_filter} 在 {target_month} 的数据文件") return {} else: csv_files = list(env_config['data_dir'].glob(f"{env_config['file_prefix']}_*_{target_month}.csv")) for csv_file in csv_files: account_id = self._extract_account_id_from_file(csv_file, 'prod') # 只处理Engineering Group的账户 if account_id and not self.is_engineering_account(account_id): print(f"跳过非Engineering Group生产账户: {account_id}") continue if account_id and (not self.account_filter or account_id == self.account_filter): data = self._load_csv_file(csv_file, account_id, 'prod') if data: account_data[account_id] = data return account_data def _load_misplaced_prod_from_dedicated_dir(self, target_month: str) -> Dict[str, List[Dict]]: """从专门目录加载错位生产账户数据""" misplaced_config = self.environments['misplaced_prod'] account_data = {} # 确定文件模式 if self.account_filter: csv_pattern = f"{misplaced_config['file_prefix']}_{self.account_filter}_{target_month}.csv" csv_files = list(misplaced_config['data_dir'].glob(csv_pattern)) if not csv_files: print(f"未找到错位生产账户 {self.account_filter} 在 {target_month} 的数据文件") return {} else: csv_files = list(misplaced_config['data_dir'].glob(f"{misplaced_config['file_prefix']}_*_{target_month}.csv")) for csv_file in csv_files: account_id = self._extract_account_id_from_misplaced_file(csv_file) # 只处理Engineering Group的账户 if account_id and not self.is_engineering_account(account_id): print(f"跳过非Engineering Group错位生产账户: {account_id}") continue if account_id: # 验证账户是否在错位生产账户列表中 if account_id in self.misplaced_prod_accounts: if not self.account_filter or account_id == self.account_filter: print(f"加载错位生产账户 {account_id} ({self.misplaced_prod_accounts[account_id]}) 的数据") data = self._load_csv_file(csv_file, account_id, 'prod') # 按生产环境处理 if data: account_data[account_id] = data else: print(f"警告: 在错位生产目录中发现未配置的账户 {account_id}") return account_data def _extract_account_id_from_misplaced_file(self, csv_file: Path) -> Optional[str]: """从错位生产账户文件名中提取账户ID""" try: # aws_misplaced_prod_632330501419_2025-08.csv -> 632330501419 parts = csv_file.stem.split('_') if len(parts) >= 4 and parts[0] == 'aws' and parts[1] == 'misplaced' and parts[2] == 'prod': return parts[3] # account_id return None except Exception as e: print(f"从错位生产文件名 {csv_file.name} 提取账户ID失败: {e}") return None def _load_nonprod_data(self, target_month: str) -> Dict[str, List[Dict]]: """加载非生产环境数据（排除错位的生产账户）""" env_config = self.environments['nonprod'] account_data = {} # 确定文件模式 if self.account_filter: csv_pattern = f"{env_config['file_prefix']}_{self.account_filter}_{target_month}.csv" csv_files = list(env_config['data_dir'].glob(csv_pattern)) if not csv_files: print(f"未找到非生产环境账户 {self.account_filter} 在 {target_month} 的数据文件") return {} else: csv_files = list(env_config['data_dir'].glob(f"{env_config['file_prefix']}_*_{target_month}.csv")) for csv_file in csv_files: account_id = self._extract_account_id_from_file(csv_file, 'nonprod') # 只处理Engineering Group的账户 if account_id and not self.is_engineering_account(account_id): print(f"跳过非Engineering Group非生产账户: {account_id}") continue if account_id: # 跳过错位的生产账户 if account_id in self.misplaced_prod_accounts: print(f"跳过错位生产账户: {account_id} ({self.misplaced_prod_accounts[account_id]})") continue if not self.account_filter or account_id == self.account_filter: data = self._load_csv_file(csv_file, account_id, 'nonprod') if data: account_data[account_id] = data return account_data def _extract_account_id_from_file(self, csv_file: Path, environment: str) -> Optional[str]: """从文件名中提取账户ID""" try: parts = csv_file.stem.split('_') if environment == 'nonprod': # aws_385595570414_2025-07.csv -> 385595570414 account_id = parts[1] if len(parts) >= 3 else None # 验证账户ID是否为数字（非生产环境应该都是纯数字） if account_id and not account_id.isdigit(): print(f"警告: 非生产环境账户ID格式异常: {account_id}") return None return account_id else: # prod or shared # 支持多种生产环境文件命名格式 # aws_prod_385595570414_2025-07.csv -> 385595570414 # 或 aws_prod_380838443567_2025-07.csv -> 380838443567 (特殊表 aws_prod_billing_data) # aws_shared_385595570414_2025-08.csv -> 385595570414 (shared环境) if len(parts) >= 4 and parts[1] == 'prod': return parts[2] # aws_prod_account_date.csv elif len(parts) >= 3: return parts[1] # aws_account_date.csv (fallback) return None except Exception as e: print(f"从文件名 {csv_file.name} 提取账户ID失败: {e}") return None def _load_csv_file(self, csv_file: Path, account_id: str, environment: str) -> List[Dict]: """加载单个CSV文件的数据""" try: with open(csv_file, 'r', encoding='utf-8') as f: reader = csv.DictReader(f) records = [] for row in reader: # 统一字段名称和数据类型 if environment == 'nonprod': # 非生产环境字段: date, account_id, environment, product, cost record = { 'date': row['date'], 'account_id': row['account_id'], 'product': row['product'], 'cost': float(row['cost']), 'environment': 'nonprod' } else: # prod or shared # 生产环境和公共测试环境字段: date, account_id, environment, product, component, servicetype, region, tenant, usedby, cost # 支持多种数据表结构字段名 cost_field = 'cost' if 'cost' not in row: cost_field = 'daily_cost' if 'daily_cost' in row else 'line_item_unblended_cost' product_field = 'product' if 'product' not in row: product_field = 'service_name' if 'service_name' in row else 'product_name' # 验证数据质量 try: cost_value = float(row[cost_field]) if cost_value == 0: continue # 跳过零成本记录 except (ValueError, TypeError): print(f"警告: 账户 {account_id} 数据中成本字段格式错误: {row.get(cost_field)}") continue record = { 'date': row['date'], 'account_id': row['account_id'], 'product': row[product_field], 'cost': cost_value, 'environment': environment, # 使用传入的环境参数 'component': row.get('component', ''), 'servicetype': row.get('servicetype', ''), 'region': row.get('region', ''), 'tenant': row.get('tenant', ''), 'usedby': row.get('usedby', '') } records.append(record) print(f"已加载{environment}环境账户 {account_id} 的数据，共 {len(records)} 条记录") return records except Exception as e: print(f"加载{environment}环境文件 {csv_file} 失败: {e}") return [] def filter_data_by_date(self, account_data: Dict[str, List[Dict]], target_date: str) -> Dict[str, List[Dict]]: """按日期过滤数据""" filtered_data = {} for account_id, records in account_data.items(): filtered_records = [r for r in records if r['date'] == target_date] if filtered_records: filtered_data[account_id] = filtered_records return filtered_data def get_historical_data(self, account_data: Dict[str, List[Dict]], end_date: str, days: int = 37) -> Dict[str, List[Dict]]: """获取历史数据用于建立基线（包含检查停用服务的天数）""" end_date_obj = datetime.strptime(end_date, '%Y-%m-%d').date() start_date_obj = end_date_obj - timedelta(days=days) historical_data = {} for account_id, records in account_data.items(): historical_records = [] for record in records: record_date = datetime.strptime(record['date'], '%Y-%m-%d').date() if start_date_obj <= record_date < end_date_obj: historical_records.append(record) if historical_records: historical_data[account_id] = historical_records return historical_data def analyze_environment(self, environment: str, target_date: str) -> Dict: """分析指定环境指定日期的账单数据""" env_config = self.environments[environment] print(f"开始分析 {env_config['display_name']} {target_date} 的账单数据...") # 加载数据 all_account_data = self.load_billing_data(environment, target_date) if not all_account_data: return { 'success': False, 'environment': environment, 'message': f'未找到 {target_date} 的{environment}环境账单数据' } # 过滤目标日期数据 target_data = self.filter_data_by_date(all_account_data, target_date) if not target_data: return { 'success': False, 'environment': environment, 'message': f'{target_date} 没有{environment}环境账单数据' } # 获取历史数据建立基线 historical_data = self.get_historical_data(all_account_data, target_date) # 分析账户维度异常 account_anomalies = self.analyze_account_anomalies( target_data, historical_data, target_date, env_config['thresholds'] ) # 分析服务维度异常 service_anomalies = self.analyze_service_anomalies( target_data, historical_data, all_account_data, target_date, env_config['thresholds'] ) # 分析tag维度异常（仅对支持tag数据的环境） tag_anomalies = [] if environment in ['prod', 'shared']: tag_anomalies = self.analyze_tag_anomalies( target_data, historical_data, target_date, env_config['thresholds'] ) # 分析没有tenant标签的资源（仅对生产环境） no_tenant_anomalies = [] if environment == 'prod': no_tenant_anomalies = self.analyze_no_tenant_tags( target_data, historical_data, target_date, env_config['thresholds'] ) # 统计信息 total_cost = sum( sum(record['cost'] for record in records) for records in target_data.values() ) account_costs = { account_id: sum(record['cost'] for record in records) for account_id, records in target_data.items() } # 忽略非资源费用的服务 service_costs = defaultdict(float) for records in target_data.values(): for record in records: service = record['product'] if service not in self.ignored_services: service_costs[service] += record['cost'] return { 'success': True, 'environment': environment, 'date': target_date, 'total_cost': total_cost, 'account_count': len(target_data), 'service_count': len(service_costs), 'account_costs': account_costs, 'service_costs': dict(service_costs), 'account_anomalies': account_anomalies, 'service_anomalies': service_anomalies, 'tag_anomalies': tag_anomalies, 'no_tenant_anomalies': no_tenant_anomalies, 'total_anomalies': len(account_anomalies) + len(service_anomalies) + len(tag_anomalies) + len(no_tenant_anomalies) } def analyze_account_anomalies(self, target_data: Dict[str, List[Dict]], historical_data: Dict[str, List[Dict]], target_date: str, thresholds: Dict) -> List[Dict]: """分析账户维度的异常""" anomalies = [] for account_id in target_data: # 计算目标日期的总费用 target_cost = sum(record['cost'] for record in target_data[account_id]) # 计算历史基线 if account_id in historical_data: baseline = self.anomaly_detector.calculate_account_baseline( historical_data[account_id] ) # 检测异常 anomaly = self.anomaly_detector.detect_account_anomaly( target_cost, baseline, thresholds ) if anomaly: anomaly.update({ 'account_id': account_id, 'date': target_date, 'type': 'account_anomaly' }) anomalies.append(anomaly) else: # 新账户 if target_cost > thresholds['new_service_threshold']: anomalies.append({ 'account_id': account_id, 'date': target_date, 'type': 'new_account', 'severity': 'warning', 'current_cost': target_cost, 'message': f'New account incurred cost: ${target_cost:,.2f}' }) return anomalies def analyze_service_anomalies(self, target_data: Dict[str, List[Dict]], historical_data: Dict[str, List[Dict]], all_account_data: Dict[str, List[Dict]], target_date: str, thresholds: Dict) -> List[Dict]: """分析服务维度的异常""" anomalies = [] for account_id in target_data: # 按服务聚合目标日期的费用 target_services = {} for record in target_data[account_id]: service = record['product'] # 忽略非资源费用的服务 if service in self.ignored_services: continue target_services[service] = target_services.get(service, 0) + record['cost'] # 计算历史服务基线 if account_id in historical_data: service_baselines = self.anomaly_detector.calculate_service_baselines( historical_data[account_id], self.ignored_services ) # 检测服务异常 for service, cost in target_services.items(): if service in service_baselines: anomaly = self.anomaly_detector.detect_service_anomaly( cost, service_baselines[service], thresholds ) if anomaly: anomaly.update({ 'account_id': account_id, 'service': service, 'date': target_date, 'type': 'service_anomaly' }) anomalies.append(anomaly) else: # 新服务 if cost > thresholds['new_service_threshold']: anomalies.append({ 'account_id': account_id, 'service': service, 'date': target_date, 'type': 'new_service', 'severity': 'warning', 'current_cost': cost, 'message': f'New service {service} incurred cost: ${cost:.2f}' }) # 检测停用的服务 stopped_services = self._detect_stopped_services( account_id, historical_data[account_id], target_date, service_baselines, thresholds ) anomalies.extend(stopped_services) return anomalies def _detect_stopped_services(self, account_id: str, historical_records: List[Dict], target_date: str, service_baselines: Dict[str, Dict], thresholds: Dict) -> List[Dict]: """检测连续7天没有费用的服务（可能已停用）""" anomalies = [] # 计算目标日期前7天的日期列表 target_date_obj = datetime.strptime(target_date, '%Y-%m-%d').date() check_dates = [] for i in range(7): check_date = target_date_obj - timedelta(days=i) check_dates.append(check_date.strftime('%Y-%m-%d')) # 按服务和日期聚合费用 service_daily_costs = defaultdict(lambda: defaultdict(float)) for record in historical_records: service = record['product'] date = record['date'] if date in check_dates and service not in self.ignored_services: service_daily_costs[service][date] += record['cost'] # 检查每个历史服务是否连续7天没有费用 for service, baseline in service_baselines.items(): if baseline['avg_cost'] <= thresholds['min_cost_threshold']: continue # 跳过历史费用很低的服务 consecutive_zero_days = 0 for check_date in check_dates: daily_cost = service_daily_costs[service].get(check_date, 0.0) if daily_cost <= 0.001: consecutive_zero_days += 1 else: break if consecutive_zero_days >= 7: anomalies.append({ 'account_id': account_id, 'service': service, 'date': target_date, 'type': 'service_stopped', 'severity': 'info', 'baseline_cost': baseline['avg_cost'], 'consecutive_zero_days': consecutive_zero_days, 'message': (f'Service {service} may have been stopped ' f'(no cost for {consecutive_zero_days} consecutive days, ' f'historical average: ${baseline["avg_cost"]:.2f})') }) return anomalies def analyze_tag_anomalies(self, target_data: Dict[str, List[Dict]], historical_data: Dict[str, List[Dict]], target_date: str, thresholds: Dict) -> List[Dict]: """分析tag维度的异常（仅支持有tag数据的环境）""" anomalies = [] for account_id in target_data: # 按usedby标签聚合目标日期的费用 target_tags = {} for record in target_data[account_id]: usedby_tag = record.get('usedby', '').strip() if not usedby_tag: # 跳过空的usedby标签 continue # 忽略非资源费用的服务 service = record['product'] if service in self.ignored_services: continue target_tags[usedby_tag] = target_tags.get(usedby_tag, 0) + record['cost'] # 计算历史tag基线 if account_id in historical_data: tag_baselines = self.anomaly_detector.calculate_tag_baselines( historical_data[account_id], self.ignored_services ) # 检测tag异常 for usedby_tag, cost in target_tags.items(): # 获取tag维度的阈值，如果没有则使用默认值 tag_thresholds = { 'tag_warning': thresholds.get('tag_warning', 2.0), 'tag_critical': thresholds.get('tag_critical', 3.0), 'min_tag_cost_threshold': thresholds.get('min_tag_cost_threshold', 1.0) } if usedby_tag in tag_baselines: anomaly = self.anomaly_detector.detect_tag_anomaly( cost, tag_baselines[usedby_tag], tag_thresholds ) if anomaly: anomaly.update({ 'account_id': account_id, 'usedby_tag': usedby_tag, 'date': target_date, 'type': 'tag_anomaly' }) anomalies.append(anomaly) else: # 新的usedby标签 new_tag_threshold = thresholds.get('new_tag_threshold', 5.0) if cost > new_tag_threshold: anomalies.append({ 'account_id': account_id, 'usedby_tag': usedby_tag, 'date': target_date, 'type': 'new_tag', 'severity': 'warning', 'current_cost': cost, 'message': f'New usedby tag "{usedby_tag}" incurred cost: ${cost:.2f}' }) # 检测停用的usedby标签 stopped_tags = self._detect_stopped_tags( account_id, historical_data[account_id], target_date, tag_baselines, thresholds ) anomalies.extend(stopped_tags) return anomalies def analyze_no_tenant_tags(self, target_data: Dict[str, List[Dict]], historical_data: Dict[str, List[Dict]], target_date: str, thresholds: Dict) -> List[Dict]: """分析没有tenant标签的资源，按服务分组（仅生产环境）""" anomalies = [] for account_id in target_data: # 按服务分组统计没有tenant标签的资源费用 no_tenant_services = {} total_no_tenant_cost = 0 for record in target_data[account_id]: # 检查tenant标签是否为空 tenant_tag = record.get('tenant', '').strip() if tenant_tag: # 跳过有tenant标签的记录 continue # 忽略非资源费用的服务 service = record['product'] if service in self.ignored_services: continue # 累计该服务没有tenant标签的费用 no_tenant_services[service] = no_tenant_services.get(service, 0) + record['cost'] total_no_tenant_cost += record['cost'] # 计算历史基线（没有tenant标签的资源） if account_id in historical_data: historical_no_tenant_services = {} historical_no_tenant_daily = defaultdict(lambda: defaultdict(float)) for record in historical_data[account_id]: tenant_tag = record.get('tenant', '').strip() if tenant_tag: # 跳过有tenant标签的记录 continue service = record['product'] if service in self.ignored_services: continue date = record['date'] historical_no_tenant_daily[service][date] += record['cost'] # 计算每个服务的历史平均值 for service, daily_costs in historical_no_tenant_daily.items(): costs = list(daily_costs.values()) if costs: historical_no_tenant_services[service] = { 'avg_cost': statistics.mean(costs), 'median_cost': statistics.median(costs), 'max_cost': max(costs), 'min_cost': min(costs) } # 检测异常 no_tenant_threshold = thresholds.get('no_tenant_warning', 1.5) no_tenant_critical = thresholds.get('no_tenant_critical', 2.0) min_no_tenant_cost = thresholds.get('min_no_tenant_cost_threshold', 10.0) for service, cost in no_tenant_services.items(): if cost < min_no_tenant_cost: continue if service in historical_no_tenant_services: baseline = historical_no_tenant_services[service] if baseline['avg_cost'] > 0: ratio = cost / baseline['avg_cost'] if ratio >= no_tenant_critical: anomalies.append({ 'account_id': account_id, 'service': service, 'date': target_date, 'type': 'no_tenant_critical', 'severity': 'critical', 'current_cost': cost, 'baseline_cost': baseline['avg_cost'], 'ratio': ratio, 'message': (f'Critical no-tenant resource anomaly for {service}: ' f'${cost:,.2f} (baseline: ${baseline["avg_cost"]:,.2f}, ' f'ratio: {ratio:.1f}x)') }) elif ratio >= no_tenant_threshold: anomalies.append({ 'account_id': account_id, 'service': service, 'date': target_date, 'type': 'no_tenant_warning', 'severity': 'warning', 'current_cost': cost, 'baseline_cost': baseline['avg_cost'], 'ratio': ratio, 'message': (f'No-tenant resource anomaly for {service}: ' f'${cost:,.2f} (baseline: ${baseline["avg_cost"]:,.2f}, ' f'ratio: {ratio:.1f}x)') }) else: # 新出现的没有tenant标签的服务 new_no_tenant_threshold = thresholds.get('new_no_tenant_threshold', 20.0) if cost > new_no_tenant_threshold: anomalies.append({ 'account_id': account_id, 'service': service, 'date': target_date, 'type': 'new_no_tenant_service', 'severity': 'warning', 'current_cost': cost, 'message': f'New service {service} with no tenant tag incurred cost: ${cost:,.2f}' }) # 添加总体统计信息（如果有显著的无tenant标签费用） if total_no_tenant_cost > thresholds.get('total_no_tenant_threshold', 100.0): service_breakdown = ', '.join([ f'{service}: ${cost:,.2f}' for service, cost in sorted(no_tenant_services.items(), key=lambda x: x[1], reverse=True)[:5] ]) anomalies.append({ 'account_id': account_id, 'date': target_date, 'type': 'no_tenant_summary', 'severity': 'info', 'total_cost': total_no_tenant_cost, 'service_count': len(no_tenant_services), 'message': (f'Total no-tenant resources cost: ${total_no_tenant_cost:,.2f} ' f'across {len(no_tenant_services)} services. ' f'Top services: {service_breakdown}') }) return anomalies def _detect_stopped_tags(self, account_id: str, historical_records: List[Dict], target_date: str, tag_baselines: Dict[str, Dict], thresholds: Dict) -> List[Dict]: """检测连续7天没有费用的usedby标签（可能已停用）""" anomalies = [] # 计算目标日期前7天的日期列表 target_date_obj = datetime.strptime(target_date, '%Y-%m-%d').date() check_dates = [] for i in range(7): check_date = target_date_obj - timedelta(days=i) check_dates.append(check_date.strftime('%Y-%m-%d')) # 按usedby标签和日期聚合费用 tag_daily_costs = defaultdict(lambda: defaultdict(float)) for record in historical_records: usedby_tag = record.get('usedby', '').strip() if not usedby_tag: continue date = record['date'] service = record['product'] if date in check_dates and service not in self.ignored_services: tag_daily_costs[usedby_tag][date] += record['cost'] # 检查每个历史usedby标签是否连续7天没有费用 min_cost_threshold = thresholds.get('min_tag_cost_threshold', 1.0) for usedby_tag, baseline in tag_baselines.items(): if baseline['avg_cost'] <= min_cost_threshold: continue # 跳过历史费用很低的标签 consecutive_zero_days = 0 for check_date in check_dates: daily_cost = tag_daily_costs[usedby_tag].get(check_date, 0.0) if daily_cost <= 0.001: consecutive_zero_days += 1 else: break if consecutive_zero_days >= 7: anomalies.append({ 'account_id': account_id, 'usedby_tag': usedby_tag, 'date': target_date, 'type': 'tag_stopped', 'severity': 'info', 'baseline_cost': baseline['avg_cost'], 'consecutive_zero_days': consecutive_zero_days, 'message': (f'UsedBy tag "{usedby_tag}" may have been stopped ' f'(no cost for {consecutive_zero_days} consecutive days, ' f'historical average: ${baseline["avg_cost"]:.2f})') }) return anomalies def analyze_all_environments(self, target_date: str, environments: List[str]) -> Dict[str, Dict]: """分析所有指定环境的数据""" results = {} for env in environments: result = self.analyze_environment(env, target_date) results[env] = result return results def combine_environment_results(self, results: Dict[str, Dict]) -> Dict: """合并多个环境的分析结果为统一报告""" successful_results = {env: result for env, result in results.items() if result['success']} if not successful_results: return { 'success': False, 'message': '所有环境的分析都失败了' } # 合并统计 total_cost = sum(result['total_cost'] for result in successful_results.values()) total_accounts = sum(result['account_count'] for result in successful_results.values()) total_anomalies = sum(result['total_anomalies'] for result in successful_results.values()) # 合并异常 all_account_anomalies = [] all_service_anomalies = [] all_tag_anomalies = [] all_no_tenant_anomalies = [] combined_account_costs = {} combined_service_costs = defaultdict(float) for env, result in successful_results.items(): # 添加环境标识到异常记录 for anomaly in result['account_anomalies']: anomaly['environment'] = env all_account_anomalies.append(anomaly) for anomaly in result['service_anomalies']: anomaly['environment'] = env all_service_anomalies.append(anomaly) # 合并tag异常（如果存在） for anomaly in result.get('tag_anomalies', []): anomaly['environment'] = env all_tag_anomalies.append(anomaly) # 合并no_tenant异常（如果存在） for anomaly in result.get('no_tenant_anomalies', []): anomaly['environment'] = env all_no_tenant_anomalies.append(anomaly) # 合并费用统计（账户ID加环境前缀避免冲突） for account_id, cost in result['account_costs'].items(): combined_key = f"{env}_{account_id}" combined_account_costs[combined_key] = cost for service, cost in result['service_costs'].items(): combined_service_costs[service] += cost return { 'success': True, 'combined': True, 'environments': list(successful_results.keys()), 'date': list(successful_results.values())[0]['date'], 'total_cost': total_cost, 'account_count': total_accounts, 'service_count': len(combined_service_costs), 'account_costs': combined_account_costs, 'service_costs': dict(combined_service_costs), 'account_anomalies': all_account_anomalies, 'service_anomalies': all_service_anomalies, 'tag_anomalies': all_tag_anomalies, 'no_tenant_anomalies': all_no_tenant_anomalies, 'total_anomalies': total_anomalies, 'environment_results': successful_results } class AnomalyDetector: """异常检测器""" def calculate_account_baseline(self, historical_records: List[Dict]) -> Dict: """计算账户费用基线""" # 按日期聚合费用 daily_costs = defaultdict(float) for record in historical_records: daily_costs[record['date']] += record['cost'] costs = list(daily_costs.values()) if not costs: return {'avg_cost': 0, 'median_cost': 0, 'std_cost': 0} return { 'avg_cost': statistics.mean(costs), 'median_cost': statistics.median(costs), 'std_cost': statistics.stdev(costs) if len(costs) > 1 else 0, 'max_cost': max(costs), 'min_cost': min(costs), 'sample_size': len(costs) } def calculate_service_baselines(self, historical_records: List[Dict], ignored_services: set = None) -> Dict[str, Dict]: """计算服务费用基线""" if ignored_services is None: ignored_services = set() service_daily_costs = defaultdict(lambda: defaultdict(float)) # 按服务和日期聚合费用 for record in historical_records: service = record['product'] if service in ignored_services: continue date = record['date'] service_daily_costs[service][date] += record['cost'] baselines = {} for service, daily_costs in service_daily_costs.items(): costs = list(daily_costs.values()) if costs: baselines[service] = { 'avg_cost': statistics.mean(costs), 'median_cost': statistics.median(costs), 'std_cost': statistics.stdev(costs) if len(costs) > 1 else 0, 'max_cost': max(costs), 'min_cost': min(costs), 'sample_size': len(costs) } return baselines def detect_account_anomaly(self, current_cost: float, baseline: Dict, thresholds: Dict) -> Optional[Dict]: """检测账户费用异常""" if baseline['avg_cost'] == 0: return None ratio = current_cost / baseline['avg_cost'] if (ratio >= thresholds['account_critical'] and current_cost > thresholds['min_cost_threshold']): return { 'severity': 'critical', 'current_cost': current_cost, 'baseline_cost': baseline['avg_cost'], 'ratio': ratio, 'message': (f'Critical account cost anomaly: ${current_cost:,.2f} ' f'(baseline: ${baseline["avg_cost"]:,.2f}, ' f'ratio: {ratio:.1f}x)') } elif (ratio >= thresholds['account_warning'] and current_cost > thresholds['min_cost_threshold']): return { 'severity': 'warning', 'current_cost': current_cost, 'baseline_cost': baseline['avg_cost'], 'ratio': ratio, 'message': (f'Account cost anomaly: ${current_cost:,.2f} ' f'(baseline: ${baseline["avg_cost"]:,.2f}, ' f'ratio: {ratio:.1f}x)') } return None def detect_service_anomaly(self, current_cost: float, baseline: Dict, thresholds: Dict) -> Optional[Dict]: """检测服务费用异常""" if baseline['avg_cost'] == 0: return None ratio = current_cost / baseline['avg_cost'] if (ratio >= thresholds['service_critical'] and current_cost > thresholds['min_cost_threshold']): return { 'severity': 'critical', 'current_cost': current_cost, 'baseline_cost': baseline['avg_cost'], 'ratio': ratio, 'message': (f'Critical service cost anomaly: ${current_cost:,.2f} ' f'(baseline: ${baseline["avg_cost"]:,.2f}, ' f'ratio: {ratio:.1f}x)') } elif (ratio >= thresholds['service_warning'] and current_cost > thresholds['min_cost_threshold']): return { 'severity': 'warning', 'current_cost': current_cost, 'baseline_cost': baseline['avg_cost'], 'ratio': ratio, 'message': (f'Service cost anomaly: ${current_cost:,.2f} ' f'(baseline: ${baseline["avg_cost"]:,.2f}, ' f'ratio: {ratio:.1f}x)') } return None def calculate_tag_baselines(self, historical_records: List[Dict], ignored_services: set = None) -> Dict[str, Dict]: """计算usedby标签费用基线""" if ignored_services is None: ignored_services = set() tag_daily_costs = defaultdict(lambda: defaultdict(float)) # 按usedby标签和日期聚合费用 for record in historical_records: usedby_tag = record.get('usedby', '').strip() if not usedby_tag: # 跳过空的usedby标签 continue service = record['product'] if service in ignored_services: continue date = record['date'] tag_daily_costs[usedby_tag][date] += record['cost'] baselines = {} for usedby_tag, daily_costs in tag_daily_costs.items(): costs = list(daily_costs.values()) if costs: baselines[usedby_tag] = { 'avg_cost': statistics.mean(costs), 'median_cost': statistics.median(costs), 'std_cost': statistics.stdev(costs) if len(costs) > 1 else 0, 'max_cost': max(costs), 'min_cost': min(costs), 'sample_size': len(costs) } return baselines def detect_tag_anomaly(self, current_cost: float, baseline: Dict, thresholds: Dict) -> Optional[Dict]: """检测usedby标签费用异常""" if baseline['avg_cost'] == 0: return None ratio = current_cost / baseline['avg_cost'] if (ratio >= thresholds['tag_critical'] and current_cost > thresholds['min_tag_cost_threshold']): return { 'severity': 'critical', 'current_cost': current_cost, 'baseline_cost': baseline['avg_cost'], 'ratio': ratio, 'message': (f'Critical usedby tag cost anomaly: ${current_cost:,.2f} ' f'(baseline: ${baseline["avg_cost"]:,.2f}, ' f'ratio: {ratio:.1f}x)') } elif (ratio >= thresholds['tag_warning'] and current_cost > thresholds['min_tag_cost_threshold']): return { 'severity': 'warning', 'current_cost': current_cost, 'baseline_cost': baseline['avg_cost'], 'ratio': ratio, 'message': (f'UsedBy tag cost anomaly: ${current_cost:,.2f} ' f'(baseline: ${baseline["avg_cost"]:,.2f}, ' f'ratio: {ratio:.1f}x)') } return None class FeishuNotifier: """飞书通知器""" def __init__(self, webhook_urls: Dict[str, str], analyzer: 'AWSUnifiedBillingAnalyzer' = None): """ 初始化通知器 Args: webhook_urls: 环境到webhook URL的映射，格式如： { 'nonprod': 'webhook_url_for_nonprod', 'prod': 'webhook_url_for_prod', 'combined': 'webhook_url_for_combined_report' } """ self.webhook_urls = webhook_urls self.analyzer = analyzer # 验证webhook URLs validation_errors = self._validate_webhook_urls() if validation_errors: print("⚠️ 警告：发现webhook URL配置问题：") for error in validation_errors: print(f" - {error}") print(" 建议检查config.json中的webhook_urls配置") # 初始化环境配置 self.environments_config = { 'nonprod': { 'display_name': 'AWS Dev/Test Accounts', 'icon': '🟠' }, 'prod': { 'display_name': 'AWS Production Accounts', 'icon': '🔴' }, 'shared': { 'display_name': 'AWS Shared Test Accounts', 'icon': '🟢' }, 'misplaced_prod': { 'display_name': 'AWS Misplaced Production Accounts', 'icon': '🟡' } } def get_account_display_name(self, account_id: str) -> str: """获取账户友好显示名称""" if self.analyzer: return self.analyzer.get_account_display_name(account_id) return f"`{account_id}` **Unknown-{account_id}** (unknown@pingcap.com)" def _validate_webhook_urls(self) -> List[str]: """验证webhook URLs的有效性""" errors = [] # 检查是否存在示例/占位符URL placeholder_patterns = [ 'your-nonprod-webhook-url', 'your-prod-webhook-url', 'your-combined-webhook-url', 'example.com', 'localhost' ] for env, url in self.webhook_urls.items(): # 检查是否为空或None if not url or not url.strip(): errors.append(f"{env}: webhook URL为空") continue # 检查是否包含占位符文本 for pattern in placeholder_patterns: if pattern in url.lower(): errors.append(f"{env}: webhook URL似乎是示例地址，包含'{pattern}'") break # 检查URL格式 if not self._is_valid_feishu_webhook_url(url): errors.append(f"{env}: webhook URL格式不正确，应为飞书webhook格式") return errors def _is_valid_feishu_webhook_url(self, url: str) -> bool: """检查是否为有效的飞书webhook URL格式""" try: parsed = urllib.parse.urlparse(url) # 检查协议 if parsed.scheme not in ['http', 'https']: return False # 检查域名（飞书webhook通常使用open.feishu.cn） valid_domains = ['open.feishu.cn', 'open.larksuite.com'] if parsed.netloc in valid_domains: # 检查路径格式 if not parsed.path.startswith('/open-apis/bot/v2/hook/'): return False # 允许测试环境的其他域名 return True except Exception: return False def send_analysis_reports(self, results: Dict[str, Dict], separate_reports: bool = False, dry_run: bool = False) -> Dict[str, bool]: """发送分析报告""" send_results = {} if separate_reports: # 分别发送各环境的报告 for env, result in results.items(): if env in self.webhook_urls: success = self._send_single_environment_report( env, result, dry_run ) send_results[env] = success else: # 发送合并报告 if 'combined' in self.webhook_urls: # 先合并结果 temp_analyzer = AWSUnifiedBillingAnalyzer() combined_result = temp_analyzer.combine_environment_results(results) success = self._send_combined_report(combined_result, dry_run) send_results['combined'] = success else: print("未配置合并报告的webhook URL") send_results['combined'] = False # 发送tag维度的独立通知（仅对支持tag的环境） tag_send_results = self.send_tag_analysis_reports(results, dry_run) send_results.update(tag_send_results) # 发送no-tenant维度的独立通知（仅对生产环境） no_tenant_send_results = self.send_no_tenant_analysis_reports(results, dry_run) send_results.update(no_tenant_send_results) return send_results def send_tag_analysis_reports(self, results: Dict[str, Dict], dry_run: bool = False) -> Dict[str, bool]: """发送tag维度的独立分析报告""" send_results = {} # 检查是否有tag webhook配置 tag_webhook_url = self.webhook_urls.get('tag_notifications') if not tag_webhook_url: print("未配置tag维度通知的webhook URL，跳过tag通知") return {} # 筛选有tag异常的环境 tag_results = {} for env, result in results.items(): if result.get('success', False) and env in ['prod', 'shared']: tag_anomalies = result.get('tag_anomalies', []) if tag_anomalies: # 只发送有tag异常的环境 tag_results[env] = result if not tag_results: print("未发现tag维度异常，跳过tag通知") return {} # 构建并发送tag通知 card = self._build_tag_analysis_card(tag_results) success = self._send_notification(tag_webhook_url, card, dry_run) send_results['tag_notifications'] = success return send_results def send_no_tenant_analysis_reports(self, results: Dict[str, Dict], dry_run: bool = False) -> Dict[str, bool]: """发送no-tenant维度的独立分析报告（仅生产环境）""" send_results = {} # 检查是否有no-tenant webhook配置 no_tenant_webhook_url = self.webhook_urls.get('no_tenant_notifications') if not no_tenant_webhook_url: print("未配置no-tenant维度通知的webhook URL，跳过no-tenant通知") return {} # 筛选有no-tenant异常的环境（仅生产环境） no_tenant_results = {} for env, result in results.items(): if result.get('success', False) and env == 'prod': no_tenant_anomalies = result.get('no_tenant_anomalies', []) if no_tenant_anomalies: # 只发送有no-tenant异常的环境 no_tenant_results[env] = result if not no_tenant_results: print("未发现no-tenant维度异常，跳过no-tenant通知") return {} # 构建并发送no-tenant通知 card = self._build_no_tenant_analysis_card(no_tenant_results) success = self._send_notification(no_tenant_webhook_url, card, dry_run) send_results['no_tenant_notifications'] = success return send_results def _send_single_environment_report(self, environment: str, result: Dict, dry_run: bool) -> bool: """发送单个环境的报告""" if not result['success']: return self._send_error_notification( environment, result['message'], dry_run ) card = self._build_environment_card(environment, result) return self._send_notification(self.webhook_urls[environment], card, dry_run) def _send_combined_report(self, result: Dict, dry_run: bool) -> bool: """发送合并报告""" if not result['success']: return self._send_error_notification('combined', result['message'], dry_run) card = self._build_combined_card(result) webhook_url = self.webhook_urls.get('combined', list(self.webhook_urls.values())[0]) return self._send_notification(webhook_url, card, dry_run) def _send_error_notification(self, environment: str, error_message: str, dry_run: bool) -> bool: """发送错误通知""" if environment == 'combined': title = "🔴 AWS All Environments | 🚨 Billing Analysis Failed" else: env_config = self.environments_config.get(environment, {}) icon = env_config.get('icon', '❓') display_name = env_config.get('display_name', environment.upper()) title = f"{icon} {display_name} | 🚨 Billing Analysis Failed" card = { "msg_type": "interactive", "card": { "config": {"wide_screen_mode": True}, "header": { "template": "red", "title": {"tag": "plain_text", "content": title} }, "elements": [ { "tag": "div", "text": {"tag": "plain_text", "content": error_message} } ] } } webhook_url = self.webhook_urls.get(environment, list(self.webhook_urls.values())[0]) return self._send_notification(webhook_url, card, dry_run) def _build_environment_card(self, environment: str, result: Dict) -> Dict: """构建单环境分析报告卡片""" env_config = self.environments_config[environment] # 只统计账户和服务异常，不包括tag和no_tenant异常（它们有独立的通知） account_anomalies_count = len(result.get('account_anomalies', [])) service_anomalies_count = len(result.get('service_anomalies', [])) display_anomalies = account_anomalies_count + service_anomalies_count # 确定卡片颜色和标题 if display_anomalies == 0: header_color = "green" title = f"{env_config['icon']} {env_config['display_name']} | ✅ Billing Normal" status_text = "No anomalies detected" else: has_critical = any( anomaly.get('severity') == 'critical' for anomaly in result['account_anomalies'] + result['service_anomalies'] ) if has_critical: header_color = "red" title = f"{env_config['icon']} {env_config['display_name']} | 🚨 Critical Billing Anomaly" status_text = f"{display_anomalies} anomalies detected, including critical ones" else: header_color = "orange" title = f"{env_config['icon']} {env_config['display_name']} | ⚠️ Billing Anomaly Warning" status_text = f"{display_anomalies} anomalies detected" # 构建卡片内容 return self._build_card_content(title, header_color, result, status_text, environment) def _build_combined_card(self, result: Dict) -> Dict: """构建合并环境分析报告卡片""" # 只统计账户和服务异常，不包括tag和no_tenant异常（它们有独立的通知） account_anomalies_count = len(result.get('account_anomalies', [])) service_anomalies_count = len(result.get('service_anomalies', [])) display_anomalies = account_anomalies_count + service_anomalies_count environments = result['environments'] # 动态生成标题前缀，避免"All Environments"与单一环境的矛盾 if len(environments) == 1: # 单个环境：显示具体环境名称 env_name = environments[0].upper() env_display_map = { 'NONPROD': 'Nonprod Environment', 'PROD': 'Production Environment', 'SHARED': 'Shared Environment' } env_display = env_display_map.get(env_name, f'{env_name} Environment') title_prefix = f"🔵 AWS {env_display}" else: # 多个环境：显示环境列表 env_list = ', '.join([env.upper() for env in environments]) title_prefix = f"🔵 AWS Multiple Environments ({env_list})" # 确定卡片颜色和标题 if display_anomalies == 0: header_color = "green" title = f"{title_prefix} | ✅ Billing Normal" status_text = f"No anomalies detected across {len(environments)} environment(s)" else: has_critical = any( anomaly.get('severity') == 'critical' for anomaly in result['account_anomalies'] + result['service_anomalies'] ) if has_critical: header_color = "red" title = f"{title_prefix} | 🚨 Critical Billing Anomaly" status_text = f"{display_anomalies} anomalies detected across {len(environments)} environment(s), including critical ones" else: header_color = "orange" title = f"{title_prefix} | ⚠️ Billing Anomaly Warning" status_text = f"{display_anomalies} anomalies detected across {len(environments)} environment(s)" return self._build_card_content(title, header_color, result, status_text, 'combined') def _build_card_content(self, title: str, header_color: str, result: Dict, status_text: str, environment: str) -> Dict: """构建卡片内容""" elements = [ { "tag": "div", "text": { "tag": "lark_md", "content": f"📋 **AWS Daily Cost Report**" } }, { "tag": "div", "fields": [ {"is_short": True, "text": {"tag": "lark_md", "content": f"**Analysis Date**\n{result['date']}"}}, {"is_short": True, "text": {"tag": "lark_md", "content": f"**Status**\n{status_text}"}}, {"is_short": True, "text": {"tag": "lark_md", "content": f"**Total Cost**\n${result['total_cost']:,.0f}"}}, {"is_short": True, "text": {"tag": "lark_md", "content": f"**Accounts**\n{result['account_count']}"}} ] } ] # 添加异常详情（只统计账户和服务异常，不包括tag和no_tenant异常） account_anomalies_count = len(result.get('account_anomalies', [])) service_anomalies_count = len(result.get('service_anomalies', [])) display_anomalies = account_anomalies_count + service_anomalies_count if display_anomalies > 0: elements.append({"tag": "hr"}) # 账户异常表格 if result['account_anomalies']: account_anomaly_rows = [] for anomaly in result['account_anomalies']: icon = "🔴" if anomaly['severity'] == 'critical' else "🟠" env_prefix = f"[{anomaly.get('environment', '').upper()}] " if environment == 'combined' else "" display_name = self.get_account_display_name(anomaly['account_id']) # 解析异常消息中的费用信息，转为整数格式 message = anomaly['message'] # 提取baseline和current cost，转为整数显示 cost_match = re.search(r'\$(\d+\.?\d*)\s*\(baseline:\s*\$(\d+\.?\d*),.*?ratio:\s*([\d\.]+)x\)', message) if cost_match: current_cost = float(cost_match.group(1)) baseline_cost = float(cost_match.group(2)) ratio = cost_match.group(3) cost_display = f"${current_cost:,.0f}" baseline_display = f"${baseline_cost:,.0f}" anomaly_desc = f"Current: {cost_display}, Baseline: {baseline_display}, Ratio: {ratio}x" else: anomaly_desc = message account_anomaly_rows.append({ "severity": icon, "environment": env_prefix.strip("[] ") if env_prefix.strip("[] ") else "NONPROD", "account": display_name, "details": anomaly_desc }) elements.append({ "tag": "div", "text": { "tag": "lark_md", "content": "**🚨 Account Anomalies**" } }) elements.append({ "tag": "table", "page_size": 10, "row_height": "low", "header_style": { "text_align": "left", "text_size": "normal", "background_style": "red", "text_color": "white" }, "columns": [ { "name": "severity", "display_name": "⚠️", "horizontal_align": "center", "data_type": "text" }, { "name": "environment", "display_name": "Env", "horizontal_align": "center", "data_type": "text" }, { "name": "account", "display_name": "Account", "horizontal_align": "left", "data_type": "text" }, { "name": "details", "display_name": "Cost Anomaly Details", "horizontal_align": "left", "data_type": "text" } ], "rows": account_anomaly_rows }) # 服务异常表格 if result['service_anomalies']: service_anomaly_rows = [] for anomaly in result['service_anomalies']: if anomaly['severity'] == 'critical': icon = "🔴" elif anomaly['severity'] == 'warning': icon = "🟠" else: icon = "🔵" env_prefix = f"[{anomaly.get('environment', '').upper()}] " if environment == 'combined' else "" display_name = self.get_account_display_name(anomaly['account_id']) # 解析服务异常消息中的费用信息，转为整数格式 message = anomaly['message'] cost_match = re.search(r'\$(\d+\.?\d*)\s*\(baseline:\s*\$(\d+\.?\d*),.*?ratio:\s*([\d\.]+)x\)', message) if cost_match: current_cost = float(cost_match.group(1)) baseline_cost = float(cost_match.group(2)) ratio = cost_match.group(3) cost_display = f"${current_cost:,.0f}" baseline_display = f"${baseline_cost:,.0f}" cost_info = f"{cost_display} (was {baseline_display}, {ratio}x)" else: cost_info = message.split(": ", 1)[1] if ": " in message else message service_anomaly_rows.append({ "severity": icon, "environment": env_prefix.strip("[] ") if env_prefix.strip("[] ") else "NONPROD", "account": display_name.split(' ')[0] if ' ' in display_name else display_name[:15] + "...", # 简化账户名 "service": anomaly['service'], "cost_info": cost_info }) elements.append({ "tag": "div", "text": { "tag": "lark_md", "content": "**⚙️ Service Anomalies**" } }) elements.append({ "tag": "table", "page_size": 10, "row_height": "low", "header_style": { "text_align": "left", "text_size": "normal", "background_style": "orange", "text_color": "white" }, "columns": [ { "name": "severity", "display_name": "⚠️", "horizontal_align": "center", "data_type": "text" }, { "name": "environment", "display_name": "Env", "horizontal_align": "center", "data_type": "text" }, { "name": "account", "display_name": "Account", "horizontal_align": "left", "data_type": "text" }, { "name": "service", "display_name": "Service", "horizontal_align": "left", "data_type": "text" }, { "name": "cost_info", "display_name": "Cost Details", "horizontal_align": "left", "data_type": "text" } ], "rows": service_anomaly_rows }) # 添加 TOP 费用信息表格 elements.append({"tag": "hr"}) # TOP 5 账户费用表格 top_accounts = sorted( result['account_costs'].items(), key=lambda x: x[1], reverse=True )[:5] if top_accounts: account_rows = [] for account_id, cost in top_accounts: # 提取账户信息 pure_account_id = account_id if '_' in account_id and account_id.split('_')[0] in ['nonprod', 'prod']: pure_account_id = account_id.split('_', 1)[1] if pure_account_id in self.analyzer.account_details: account_info = self.analyzer.account_details[pure_account_id] account_name = account_info['account_name'] owner_email = account_info['owner_email'] shared_tag = " 🔄" if account_info['is_shared_dev_account'] else "" else: account_name = f"Unknown-{pure_account_id}" owner_email = "unknown@pingcap.com" shared_tag = "" account_rows.append({ "account": f"{pure_account_id}", "name": f"{account_name}{shared_tag}", "owner": owner_email, "cost": f"${cost:,.0f}" }) elements.append({ "tag": "div", "text": { "tag": "lark_md", "content": "**💰 TOP 5 Account Costs**" } }) elements.append({ "tag": "table", "page_size": 5, "row_height": "low", "header_style": { "text_align": "left", "text_size": "normal", "background_style": "grey", "text_color": "default" }, "columns": [ { "name": "account", "display_name": "Account ID", "horizontal_align": "left", "data_type": "text" }, { "name": "name", "display_name": "Account Name", "horizontal_align": "left", "data_type": "text" }, { "name": "owner", "display_name": "Owner", "horizontal_align": "left", "data_type": "text" }, { "name": "cost", "display_name": "Cost", "horizontal_align": "right", "data_type": "text" } ], "rows": account_rows }) # TOP 5 服务费用表格 top_services = sorted( result['service_costs'].items(), key=lambda x: x[1], reverse=True )[:5] if top_services: service_rows = [] for service, cost in top_services: # 添加服务图标, 保留，暂时不显示 # service_icon = self._get_service_icon(service) service_rows.append({ "service": f"{service}", "cost": f"${cost:,.0f}", "percentage": f"{(cost / result['total_cost'] * 100):.2f}%" }) elements.append({ "tag": "div", "text": { "tag": "lark_md", "content": "**⚙️ TOP 5 Service Costs**" } }) elements.append({ "tag": "table", "page_size": 5, "row_height": "low", "header_style": { "text_align": "left", "text_size": "normal", "background_style": "grey", "text_color": "default" }, "columns": [ { "name": "service", "display_name": "Service", "horizontal_align": "left", "data_type": "text" }, { "name": "cost", "display_name": "Cost", "horizontal_align": "right", "data_type": "text" }, { "name": "percentage", "display_name": "Percentage of Total Cost", "horizontal_align": "right", "data_type": "text" } ], "rows": service_rows }) return { "msg_type": "interactive", "card": { "config": {"wide_screen_mode": True}, "header": { "template": header_color, "title": {"tag": "plain_text", "content": title} }, "elements": elements } } def _send_notification(self, webhook_url: str, card: Dict, dry_run: bool) -> bool: """发送通知""" if dry_run: print("=== 飞书通知内容 (dry-run) ===") print(json.dumps(card, indent=2, ensure_ascii=False)) return True # 检查webhook URL是否为占位符 placeholder_patterns = [ 'your-nonprod-webhook-url', 'your-prod-webhook-url', 'your-combined-webhook-url' ] if any(placeholder in webhook_url.lower() for placeholder in placeholder_patterns): print("❌ 飞书通知发送失败: webhook URL是示例地址，请在config.json中配置正确的webhook URL") print(f" 当前URL: {webhook_url}") return False try: print(f"📤 正在发送飞书通知到: {self._mask_webhook_url(webhook_url)}") response = requests.post( webhook_url, json=card, headers={'Content-Type': 'application/json'}, timeout=10 ) # 检查HTTP状态码 response.raise_for_status() # 解析飞书API响应 try: result = response.json() except json.JSONDecodeError: print("❌ 飞书通知发送失败: 响应不是有效的JSON格式") print(f" 响应内容: {response.text[:200]}") return False # 检查飞书API响应码 if result.get('code') == 0: print("✅ 飞书通知发送成功") return True else: error_code = result.get('code', 'unknown') error_msg = result.get('msg', 'unknown error') print(f"❌ 飞书通知发送失败: code={error_code}, msg={error_msg}") return False except requests.exceptions.Timeout: print("❌ 飞书通知发送失败: 请求超时（10秒）") return False except requests.exceptions.ConnectionError: print("❌ 飞书通知发送失败: 网络连接错误，请检查网络和webhook URL") return False except requests.exceptions.HTTPError: print(f"❌ 飞书通知发送失败: HTTP错误 {response.status_code}") if response.status_code == 404: print(" 可能原因: webhook URL不存在或已失效") elif response.status_code == 401: print(" 可能原因: webhook URL认证失败") return False except Exception as e: print(f"❌ 飞书通知发送失败: {str(e)}") return False def _mask_webhook_url(self, url: str) -> str: """对webhook URL进行脱敏处理""" try: parsed = urllib.parse.urlparse(url) if parsed.path.startswith('/open-apis/bot/v2/hook/'): # 只显示hook ID的前6位和后4位 hook_id = parsed.path.split('/')[-1] if len(hook_id) > 10: masked_id = hook_id[:6] + "****" + hook_id[-4:] return f"{parsed.scheme}://{parsed.netloc}/open-apis/bot/v2/hook/{masked_id}" return url[:50] + "..." if len(url) > 50 else url except Exception: return url[:50] + "..." if len(url) > 50 else url def _build_tag_analysis_card(self, tag_results: Dict[str, Dict]) -> Dict: """构建tag维度分析报告卡片""" # 统计tag异常总数 total_tag_anomalies = sum( len(result.get('tag_anomalies', [])) for result in tag_results.values() ) # 确定卡片颜色和标题 has_critical = False for result in tag_results.values(): if any(anomaly.get('severity') == 'critical' for anomaly in result.get('tag_anomalies', [])): has_critical = True break if has_critical: header_color = "red" title = "🏷️ AWS UsedBy Tag Monitoring | 🚨 Critical Tag Anomaly" status_text = f"{total_tag_anomalies} tag anomalies detected, including critical ones" else: header_color = "orange" title = "🏷️ AWS UsedBy Tag Monitoring | ⚠️ Tag Anomaly Warning" status_text = f"{total_tag_anomalies} tag anomalies detected" # 构建异常详情 elements = [ { "tag": "div", "text": { "tag": "lark_md", "content": f"📋 **AWS UsedBy Tag Daily Analysis Report**" } }, { "tag": "div", "fields": [ {"is_short": True, "text": {"tag": "lark_md", "content": f"**Analysis Date**\n{list(tag_results.values())[0]['date']}"}}, {"is_short": True, "text": {"tag": "lark_md", "content": f"**Status**\n{status_text}"}}, {"is_short": True, "text": {"tag": "lark_md", "content": f"**Environments**\n{', '.join([env.upper() for env in tag_results.keys()])}"}}, {"is_short": True, "text": {"tag": "lark_md", "content": f"**Total Tag Anomalies**\n{total_tag_anomalies}"}} ] } ] # 添加tag异常详情 elements.append({"tag": "hr"}) for env, result in tag_results.items(): env_config = self.environments_config.get(env, {'display_name': env.upper(), 'icon': '❓'}) env_display = f"{env_config['icon']} {env_config['display_name']}" tag_anomalies_text = f"**{env_display} Tag Anomalies:**\n" for anomaly in result.get('tag_anomalies', []): if anomaly['severity'] == 'critical': icon = "🔴" elif anomaly['severity'] == 'warning': icon = "🟠" else: icon = "🔵" display_name = self.get_account_display_name(anomaly['account_id']) usedby_tag = anomaly.get('usedby_tag', 'Unknown') tag_anomalies_text += f"{icon} {display_name} - `{usedby_tag}`: {anomaly['message']}\n" elements.append({ "tag": "div", "text": {"tag": "lark_md", "content": tag_anomalies_text.strip()} }) return { "msg_type": "interactive", "card": { "config": {"wide_screen_mode": True}, "header": { "template": header_color, "title": {"tag": "plain_text", "content": title} }, "elements": elements } } def _build_no_tenant_analysis_card(self, no_tenant_results: Dict[str, Dict]) -> Dict: """构建no-tenant维度分析报告卡片""" # 统计no-tenant异常总数 total_no_tenant_anomalies = sum( len(result.get('no_tenant_anomalies', [])) for result in no_tenant_results.values() ) # 确定卡片颜色和标题 has_critical = False for result in no_tenant_results.values(): if any(anomaly.get('severity') == 'critical' for anomaly in result.get('no_tenant_anomalies', [])): has_critical = True break if has_critical: header_color = "red" title = "🏷️ AWS No-Tenant Tag Monitoring | 🚨 Critical No-Tenant Anomaly" status_text = f"{total_no_tenant_anomalies} no-tenant anomalies detected, including critical ones" else: header_color = "orange" title = "🏷️ AWS No-Tenant Tag Monitoring | ⚠️ No-Tenant Anomaly Warning" status_text = f"{total_no_tenant_anomalies} no-tenant anomalies detected" # 构建异常详情 elements = [ { "tag": "div", "text": { "tag": "lark_md", "content": f"📋 **AWS No-Tenant Resources Daily Analysis Report**" } }, { "tag": "div", "fields": [ {"is_short": True, "text": {"tag": "lark_md", "content": f"**Analysis Date**\n{list(no_tenant_results.values())[0]['date']}"}}, {"is_short": True, "text": {"tag": "lark_md", "content": f"**Status**\n{status_text}"}}, {"is_short": True, "text": {"tag": "lark_md", "content": f"**Environment**\nProduction"}}, {"is_short": True, "text": {"tag": "lark_md", "content": f"**Total Anomalies**\n{total_no_tenant_anomalies}"}} ] } ] # 添加no-tenant异常详情表格 elements.append({"tag": "hr"}) no_tenant_rows = [] for env, result in no_tenant_results.items(): for anomaly in result.get('no_tenant_anomalies', []): # 处理具体的服务异常，跳过汇总信息 if anomaly['type'] in ['no_tenant_warning', 'no_tenant_critical', 'new_no_tenant_service']: if anomaly['severity'] == 'critical': icon = "🔴" elif anomaly['severity'] == 'warning': icon = "🟠" else: icon = "🔵" display_name = self.get_account_display_name(anomaly['account_id']) service = anomaly.get('service', 'N/A') # 解析费用信息 current_cost = anomaly.get('current_cost', 0) baseline_cost = anomaly.get('baseline_cost', 0) ratio = anomaly.get('ratio', 0) if baseline_cost > 0: cost_info = f"${current_cost:,.0f} (was ${baseline_cost:,.0f}, {ratio:.1f}x)" else: cost_info = f"${current_cost:,.0f} (new)" # 生成说明信息 if anomaly['type'] == 'new_no_tenant_service': description = "New service with no tenant tag" else: description = f"Cost increased {ratio:.1f}x from baseline" no_tenant_rows.append({ "severity": icon, "account": display_name.split(' ')[0] if ' ' in display_name else display_name[:20] + "...", "service": service, "cost_info": cost_info, "description": description }) if no_tenant_rows: elements.append({ "tag": "div", "text": { "tag": "lark_md", "content": "**🚨 No-Tenant Resource Anomalies by Service**" } }) elements.append({ "tag": "table", "page_size": 10, "row_height": "low", "header_style": { "text_align": "left", "text_size": "normal", "background_style": "orange", "text_color": "white" }, "columns": [ { "name": "severity", "display_name": "⚠️", "horizontal_align": "center", "data_type": "text" }, { "name": "account", "display_name": "Account", "horizontal_align": "left", "data_type": "text" }, { "name": "service", "display_name": "Service", "horizontal_align": "left", "data_type": "text" }, { "name": "cost_info", "display_name": "Cost Details", "horizontal_align": "left", "data_type": "text" }, { "name": "description", "display_name": "Anomaly Description", "horizontal_align": "left", "data_type": "text" } ], "rows": no_tenant_rows[:20] # 限制显示前20条 }) # 添加汇总信息 for env, result in no_tenant_results.items(): for anomaly in result.get('no_tenant_anomalies', []): if anomaly['type'] == 'no_tenant_summary': elements.append({"tag": "hr"}) elements.append({ "tag": "div", "text": { "tag": "lark_md", "content": f"**📊 Summary for Account {self.get_account_display_name(anomaly['account_id'])}**\n{anomaly['message']}" } }) return { "msg_type": "interactive", "card": { "config": {"wide_screen_mode": True}, "header": { "template": header_color, "title": {"tag": "plain_text", "content": title} }, "elements": elements } } def _get_service_icon(self, service: str) -> str: """获取AWS服务对应的图标""" service_icons = { "AmazonEC2": "🖥️", "Amazon Elastic Compute Cloud": "🖥️", "AmazonS3": "🗂️", "Amazon Simple Storage Service": "🗂️", "AWSELB": "⚖️", "Elastic Load Balancing": "⚖️", "AmazonVPC": "🌐", "Amazon Virtual Private Cloud": "🌐", "AmazonEKS": "⚙️", "Amazon Elastic Kubernetes Service": "⚙️", "AmazonRDS": "💾", "Amazon Relational Database Service": "💾", "AWSCertificateManager": "🔐", "AWS Certificate Manager": "🔐", "AmazonCloudWatch": "📊", "Amazon CloudWatch": "📊", "AWSCloudTrail": "👁️", "AWS CloudTrail": "👁️", "AmazonRoute53": "🌍", "Amazon Route 53": "🌍", "AWSWAF": "🛡️", "AWS WAF": "🛡️", "AWSLambda": "⚡", "AWS Lambda": "⚡", "AmazonCloudFront": "🚀", "Amazon CloudFront": "🚀", "AmazonEBS": "💿", "Amazon Elastic Block Store": "💿", "AmazonSQS": "📬", "Amazon Simple Queue Service": "📬", "AmazonSNS": "📢", "Amazon Simple Notification Service": "📢", "AWSSecretsManager": "🔑", "AWS Secrets Manager": "🔑", "AWSKMS": "🔒", "AWS Key Management Service": "🔒", "AWSSupport": "🎧", "AWS Support": "🎧", "Savings Plans for AWS Compute usage": "💰", "AWS Savings Plans": "💰" } return service_icons.get(service, "☁️") def validate_config(config: Dict) -> List[str]: """验证配置文件的有效性""" errors = [] # 检查webhook_urls配置 if 'webhook_urls' in config: webhook_urls = config['webhook_urls'] if not isinstance(webhook_urls, dict): errors.append("webhook_urls必须是字典格式") else: # 临时创建通知器来验证webhook URLs try: FeishuNotifier(webhook_urls) # 这里会自动运行webhook URL验证 except Exception as e: errors.append(f"webhook_urls配置错误: {e}") # 检查阈值配置 if 'thresholds' in config: thresholds = config['thresholds'] required_threshold_keys = [ 'account_warning', 'account_critical', 'service_warning', 'service_critical', 'min_cost_threshold' ] for key in required_threshold_keys: if key not in thresholds: errors.append(f"缺少阈值配置: {key}") elif not isinstance(thresholds[key], (int, float)): errors.append(f"阈值配置{key}必须是数字") return errors def parse_args(): """解析命令行参数""" parser = argparse.ArgumentParser( description='AWS 统一账单数据每日分析和告警系统', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 示例用法: %(prog)s # 分析所有环境前天数据 %(prog)s --date 2024-01-15 # 分析所有环境指定日期 %(prog)s --env nonprod --dry-run # 仅分析非生产环境不发送通知 %(prog)s --env prod --account-id 730335266318 # 分析指定生产账户 %(prog)s --separate-reports # 分别发送环境报告 %(prog)s --config config.json # 使用配置文件 """ ) parser.add_argument( '--date', help='分析日期 (YYYY-MM-DD)，默认为前天' ) parser.add_argument( '--env', choices=['nonprod', 'prod', 'shared', 'all'], default='all', help='要分析的环境 (nonprod=非生产, prod=生产, shared=公共测试, all=全部，默认: all)' ) parser.add_argument( '--account-id', help='仅分析指定账户ID (用于测试单个账户)' ) parser.add_argument( '--separate-reports', action='store_true', help='分别发送各环境的报告，而不是合并报告' ) parser.add_argument( '--webhook-urls', help='飞书 webhook URLs (JSON格式字符串)' ) parser.add_argument( '--config', help='配置文件路径 (JSON格式)' ) parser.add_argument( '--dry-run', action='store_true', help='仅分析不发送通知' ) parser.add_argument( '--thresholds', help='异常阈值配置 (JSON字符串)' ) return parser.parse_args() def load_config(config_path: str) -> Dict: """加载配置文件""" try: with open(config_path, 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: print(f"加载配置文件失败: {e}") return {} def find_config_file() -> Optional[str]: """查找配置文件""" possible_paths = [ 'aws/config.json', 'config.json', Path(__file__).parent / 'config.json' ] for path in possible_paths: if Path(path).exists(): return str(path) return None def main(): """主函数""" args = parse_args() # 加载配置 config = {} if args.config: config = load_config(args.config) else: # 自动查找配置文件 config_file = find_config_file() if config_file: config = load_config(config_file) print(f"自动加载配置文件: {config_file}") # 验证配置文件 if config: config_errors = validate_config(config) if config_errors: print("❌ 配置文件验证失败:") for error in config_errors: print(f" - {error}") print("\n请检查并修正config.json中的配置错误后重试") if not args.dry_run: # dry-run模式下允许继续执行 sys.exit(1) # 确定分析日期（默认前天） if args.date: try: target_date = datetime.strptime(args.date, '%Y-%m-%d').date() except ValueError: print("错误: 日期格式无效，请使用 YYYY-MM-DD 格式") sys.exit(1) else: # 默认分析前天的数据 target_date = datetime.now().date() - timedelta(days=2) target_date_str = target_date.strftime('%Y-%m-%d') # 确定要分析的环境 if args.env == 'all': environments = ['nonprod', 'prod', 'shared'] else: environments = [args.env] # 创建分析器 analyzer = AWSUnifiedBillingAnalyzer(args.account_id, config) # 更新异常阈值（命令行参数优先） if args.thresholds: try: custom_thresholds = json.loads(args.thresholds) for env in environments: analyzer.environments[env]['thresholds'].update(custom_thresholds) except json.JSONDecodeError: print("错误: 阈值配置JSON格式无效") sys.exit(1) # 分析数据 results = analyzer.analyze_all_environments(target_date_str, environments) # 打印分析结果 for env, result in results.items(): if result['success']: env_display = analyzer.environments[env]['display_name'] if args.account_id: print(f"\n=== {target_date_str} {env_display} 账户 {args.account_id} 分析结果 ===") else: print(f"\n=== {target_date_str} {env_display} 分析结果 ===") print(f"总费用: ${result['total_cost']:,.2f}") print(f"账户数: {result['account_count']}") print(f"服务数: {result['service_count']}") print(f"异常数: {result['total_anomalies']}") if result['total_anomalies'] > 0: print(f"\n{env}环境异常详情:") all_anomalies = result['account_anomalies'] + result['service_anomalies'] # 添加tag异常（如果存在） if 'tag_anomalies' in result: all_anomalies += result['tag_anomalies'] # 添加no_tenant异常（如果存在） if 'no_tenant_anomalies' in result: all_anomalies += result['no_tenant_anomalies'] for anomaly in all_anomalies: print(f" - {anomaly['message']}") else: print(f"\n{env}环境分析失败: {result['message']}") # 发送飞书通知 webhook_urls = {} if args.webhook_urls: try: webhook_urls = json.loads(args.webhook_urls) except json.JSONDecodeError: print("错误: webhook URLs JSON格式无效") elif config.get('webhook_urls'): webhook_urls = config['webhook_urls'] if webhook_urls and not args.dry_run: notifier = FeishuNotifier(webhook_urls, analyzer) send_results = notifier.send_analysis_reports( results, args.separate_reports, dry_run=False ) for target, success in send_results.items(): if success: print(f"飞书通知发送成功: {target}") else: print(f"飞书通知发送失败: {target}") elif args.dry_run: # dry-run 模式，显示通知内容 dummy_webhooks = {env: "dummy_url" for env in environments} if not args.separate_reports: dummy_webhooks['combined'] = "dummy_url" notifier = FeishuNotifier(dummy_webhooks, analyzer) notifier.send_analysis_reports(results, args.separate_reports, dry_run=True) else: print("未配置飞书 webhook URLs，跳过通知") if __name__ == "__main__": main()