#!/usr/bin/env python3 """ AWS 统一账单数据导出工具 功能: 1. 统一导出 AWS 生产和非生产账户的账单数据 2. 支持同时或单独导出不同环境的数据 3. 按日期和账户保存为 CSV 文件 4. 支持查询最近一周或指定日期区间的数据 用法: python aws_unified_billing_export.py --last-week # 导出所有环境最近一周数据 python aws_unified_billing_export.py --env nonprod --days 30 # 只导出非生产环境最近30天 python aws_unified_billing_export.py --env prod --start-date 2025-04-01 --end-date 2025-08-16 python aws_unified_billing_export.py --start-date 2025-04-01 --end-date 2025-08-20 # 导出所有环境指定区间 """ import argparse import csv import json import sys from datetime import datetime, timedelta from pathlib import Path import concurrent.futures from google.cloud import bigquery from typing import Dict, List, Tuple, Optional import pandas as pd class AWSUnifiedBillingExporter: """AWS 统一账单数据导出器""" def __init__(self): self.bq_client = bigquery.Client() # 错位生产账户配置将从配置文件加载 self.misplaced_prod_accounts = {} # Engineering Group 账户详情 self.account_details = {} # 环境配置 self.environments = { 'nonprod': { 'dataset_name': 'gcp-digital-bi.aws_nonprod_billing_native', 'output_dir': Path('aws_dev_billing_exports'), 'file_prefix': 'aws', 'exclude_recent_days': 2, # 非生产环境排除最近2天 'table_pattern': '{account_id}_aws_billing_data', 'cost_field': 'line_item_unblended_cost', 'product_field': 'line_item_product_code', 'date_field': 'line_item_usage_start_date', 'supports_tags': False # 不支持tag数据 }, 'shared': { # 新增：公共测试账户环境 'dataset_name': 'gcp-digital-bi.aws_nonprod_billing_native', 'output_dir': Path('aws_shared_dev_billing_exports'), 'file_prefix': 'aws_shared', 'exclude_recent_days': 2, # 排除最近2天 'table_pattern': '{account_id}_aws_billing_data', 'cost_field': 'line_item_unblended_cost', 'product_field': 'line_item_product_code', 'date_field': 'line_item_usage_start_date', 'supports_tags': True # 支持tag数据 }, 'prod': { 'dataset_name': 'gcp-digital-bi.aws_prod_billing', 'output_dir': Path('aws_prod_billing_exports'), 'file_prefix': 'aws_prod', 'exclude_recent_days': 2, # 生产环境排除最近1天 'cost_field': 'line_item_unblended_cost', 'product_field': 'product_product_name', 'date_field': 'line_item_usage_start_date', 'supports_tags': True, # 支持tag数据 # 4个独立的生产账户表 'accounts': { '143458967504': '143458967504_aws_billing_data', '632330501419': '632330501419_aws_billing_data', '730335266318': '730335266318_aws_billing_data', '380838443567': 'aws_prod_billing_data' # 特殊表名 } }, 'misplaced_prod': { 'dataset_name': 'gcp-digital-bi.aws_nonprod_billing_native', # 数据来源：非生产数据集 'output_dir': Path('aws_misplaced_prod_billing_exports'), # 专门的输出目录 'file_prefix': 'aws_misplaced_prod', # 专门的文件前缀 'exclude_recent_days': 1, # 按生产环境标准排除最近1天 'cost_field': 'line_item_unblended_cost', 'product_field': 'line_item_product_code', 'date_field': 'line_item_usage_start_date', 'supports_tags': True # 支持tag数据 } } # 创建输出目录 for env_config in self.environments.values(): env_config['output_dir'].mkdir(exist_ok=True) # 缓存配置 self.cache_expiry_hours = 24 # 加载账户详情 self.load_account_details() 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() self.account_details[account_id] = { 'account_name': row['account_name'].strip(), 'owner_email': row['owner_email'].strip(), 'environment': row['environment'].strip(), '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 is_engineering_account(self, account_id: str) -> bool: """检查账户是否属于Engineering Group""" return account_id in self.account_details 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 classify_account(self, account_id: str) -> str: """对账户进行分类，决定使用哪种处理方式""" # 1. 检查是否为错位生产账户 if account_id in self.misplaced_prod_accounts and self.misplaced_prod_accounts[account_id].get('enabled', True): return 'misplaced_prod' # 2. 检查是否为正常生产账户 account_info = self.get_account_info(account_id) if account_info.get('environment') == '生产': return 'prod' # 3. 检查是否为公共测试账户 if account_info.get('is_shared_dev_account', False): return 'shared' # 4. 默认为普通非生产账户 return 'nonprod' def _verify_table_has_tags(self, dataset_name: str, table_name: str) -> bool: """验证表是否包含resource_tags字段""" try: check_query = f""" SELECT column_name FROM `{dataset_name}.INFORMATION_SCHEMA.COLUMNS` WHERE table_name = '{table_name}' AND column_name LIKE 'resource_tags_%' LIMIT 1 """ result = self.bq_client.query(check_query).result() return len(list(result)) > 0 except Exception as e: print(f"检查tag字段时出错: {e}") return False def update_from_config(self, config: Dict): """从配置文件更新设置""" if 'export_settings' in config: export_settings = config['export_settings'] # 更新缓存时间 if 'cache_hours' in export_settings: self.cache_expiry_hours = export_settings['cache_hours'] # 更新数据延迟排除天数 if 'exclude_recent_days' in export_settings: exclude_days = export_settings['exclude_recent_days'] for env in self.environments: if env in exclude_days: self.environments[env]['exclude_recent_days'] = exclude_days[env] # 更新数据目录配置 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]['output_dir'] new_dir = Path(data_dirs[env]) self.environments[env]['output_dir'] = new_dir print(f"更新 {env} 环境数据目录: {old_dir} -> {new_dir}") # 创建新目录 new_dir.mkdir(exist_ok=True) # 加载错位生产账户配置 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 get_nonprod_accounts(self) -> List[Dict]: """获取所有 AWS 非生产账户信息（排除错位的生产账户）""" config = self.environments['nonprod'] cache_file = config['output_dir'] / "aws_accounts_cache.json" # 检查缓存 if self._is_cache_valid(cache_file): print("从缓存加载 AWS 非生产账户信息...") cached_accounts = self._load_cache(cache_file) if cached_accounts: print(f"从缓存加载了 {len(cached_accounts)} 个非生产账户") return cached_accounts print("从 BigQuery 查询 AWS 非生产账户信息...") query = f""" SELECT table_name FROM `{config['dataset_name']}.INFORMATION_SCHEMA.TABLES` WHERE table_name LIKE '%_aws_billing_data' ORDER BY table_name """ try: results = self.bq_client.query(query).result() aws_accounts = [] for row in results: table_name = row[0] print if table_name.endswith('_aws_billing_data'): account_id = table_name[:-len('_aws_billing_data')] # 跳过错位的生产账户 if account_id in self.misplaced_prod_accounts: print(f"跳过错位生产账户: {account_id} ({self.misplaced_prod_accounts[account_id]})") continue # 只处理Engineering Group的账户 if not self.is_engineering_account(account_id): print(f"跳过非Engineering Group账户: {account_id}") continue if self._validate_table_has_data(config, account_id): aws_accounts.append({ 'id': len(aws_accounts) + 1, 'account_id': account_id, 'environment': 'nonprod', 'table_name': table_name }) print(f"发现 {len(aws_accounts)} 个有效的 AWS 非生产账户") self._save_cache(cache_file, aws_accounts) return aws_accounts except Exception as e: print(f"查询 BigQuery 非生产账户信息时出错: {e}") cached_accounts = self._load_cache(cache_file) if cached_accounts: print(f"查询失败，从缓存加载了 {len(cached_accounts)} 个账户") return cached_accounts return [] def get_prod_accounts(self) -> List[Dict]: """获取所有 AWS 生产账户信息（包括错位的生产账户）""" config = self.environments['prod'] cache_file = config['output_dir'] / "aws_prod_accounts_cache.json" # 检查缓存 if self._is_cache_valid(cache_file): print("从缓存加载 AWS 生产账户信息...") cached_accounts = self._load_cache(cache_file) if cached_accounts: print(f"从缓存加载了 {len(cached_accounts)} 个生产账户") return cached_accounts print("加载 AWS 生产账户信息...") aws_accounts = [] # 1. 处理正常生产数据集中的账户 print("验证正常生产数据集中的账户...") for account_id, table_name in config['accounts'].items(): # 只处理Engineering Group的账户 if not self.is_engineering_account(account_id): print(f"跳过非Engineering Group生产账户: {account_id}") continue # 验证表是否存在且有数据 if self._validate_prod_table_exists(config, table_name): aws_accounts.append({ 'id': len(aws_accounts) + 1, 'account_id': account_id, 'table_name': table_name, 'environment': 'prod', 'data_source': 'prod_dataset' }) print(f"验证成功：生产账户 {account_id} -> 表 {table_name}") else: print(f"警告：生产账户 {account_id} 对应的表 {table_name} 不存在或无数据") # 2. 处理错位在非生产数据集中的生产账户 print("验证错位在非生产数据集中的生产账户...") nonprod_config = self.environments['nonprod'] for account_id, description in self.misplaced_prod_accounts.items(): # 只处理Engineering Group的账户 if not self.is_engineering_account(account_id): print(f"跳过非Engineering Group错位生产账户: {account_id}") continue table_name = f"{account_id}_aws_billing_data" # 验证该账户在非生产数据集中是否存在且有数据 if self._validate_misplaced_prod_table_exists(nonprod_config, account_id): aws_accounts.append({ 'id': len(aws_accounts) + 1, 'account_id': account_id, 'table_name': table_name, 'environment': 'prod', 'data_source': 'nonprod_dataset', # 标记数据来源 'description': description }) print(f"验证成功：错位生产账户 {account_id} ({description}) -> 非生产数据集中的表 {table_name}") else: print(f"警告：错位生产账户 {account_id} ({description}) 在非生产数据集中不存在或无数据") print(f"发现 {len(aws_accounts)} 个有效的 AWS 生产账户") self._save_cache(cache_file, aws_accounts) return aws_accounts def _validate_prod_table_exists(self, config: Dict, table_name: str) -> bool: """验证生产表是否存在且有数据""" try: full_table_name = f"{config['dataset_name']}.{table_name}" check_query = f""" SELECT COUNT(*) as record_count FROM `{full_table_name}` WHERE DATE({config['date_field']}) >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY) LIMIT 1 """ result = self.bq_client.query(check_query).result() for row in result: return row[0] > 0 return False except Exception as e: # 记录具体的验证错误，但不中断流程 print(f"验证表 {table_name} 数据失败: {e}") return False def _validate_misplaced_prod_table_exists(self, nonprod_config: Dict, account_id: str) -> bool: """验证错位生产账户在非生产数据集中是否存在且有数据""" try: table_name = f"{nonprod_config['dataset_name']}.{account_id}_aws_billing_data" check_query = f""" SELECT COUNT(*) as record_count FROM `{table_name}` WHERE DATE({nonprod_config['date_field']}) >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY) LIMIT 1 """ result = self.bq_client.query(check_query).result() for row in result: return row[0] > 0 return False except Exception as e: print(f"验证错位生产账户 {account_id} 数据失败: {e}") return False def _is_cache_valid(self, cache_file: Path) -> bool: """检查缓存是否有效""" if not cache_file.exists(): return False try: with open(cache_file, 'r', encoding='utf-8') as f: cache_data = json.load(f) cache_time = datetime.fromisoformat(cache_data.get('timestamp', '')) expiry_time = cache_time + timedelta(hours=self.cache_expiry_hours) return datetime.now() < expiry_time except (json.JSONDecodeError, ValueError, KeyError): return False def _load_cache(self, cache_file: Path) -> List[Dict]: """从缓存加载账户信息""" try: with open(cache_file, 'r', encoding='utf-8') as f: cache_data = json.load(f) return cache_data.get('accounts', []) except (FileNotFoundError, json.JSONDecodeError): return [] def _save_cache(self, cache_file: Path, accounts: List[Dict]): """保存账户信息到缓存""" cache_data = { 'timestamp': datetime.now().isoformat(), 'accounts': accounts } with open(cache_file, 'w', encoding='utf-8') as f: json.dump(cache_data, f, indent=2, ensure_ascii=False) def _validate_table_has_data(self, config: Dict, account_id: str) -> bool: """验证非生产表是否有数据""" try: table_name = f"`{config['dataset_name']}.{account_id}_aws_billing_data`" check_query = f""" SELECT COUNT(*) as record_count FROM {table_name} WHERE DATE({config['date_field']}) >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY) LIMIT 1 """ result = self.bq_client.query(check_query).result() for row in result: return row[0] > 0 return False except Exception as e: # 记录具体的验证错误，但不中断流程 print(f"验证表 {account_id} 数据失败: {e}") return False def construct_nonprod_query(self, account_id: str, start_date: str, end_date: str) -> str: """构建非生产账户查询 SQL""" config = self.environments['nonprod'] table = f"`{config['dataset_name']}.{account_id}_aws_billing_data`" actual_end_date = self._get_safe_end_date(end_date, config['exclude_recent_days']) if actual_end_date < start_date: return (f"SELECT DATE('1900-01-01') as day, 0.0 as cost, 'NoData' as product, " f"'{account_id}' as account_id WHERE FALSE") query = f""" SELECT DATE({config['date_field']}) as day, ROUND(SUM({config['cost_field']}), 4) as cost, {config['product_field']} as product, '{account_id}' as account_id FROM {table} WHERE DATE({config['date_field']}) >= '{start_date}' AND DATE({config['date_field']}) <= '{actual_end_date}' AND line_item_line_item_type NOT IN ( 'Tax', 'Credit', 'Refund', 'Fee', 'RIFee' ) AND {config['cost_field']} != 0 GROUP BY day, product HAVING ABS(cost) > 0.001 ORDER BY day ASC, cost DESC """ return query def construct_prod_query(self, account_id: str, table_name: str, start_date: str, end_date: str, data_source: str = 'prod_dataset') -> str: """构建生产账户查询 SQL""" config = self.environments['prod'] # 根据数据来源确定完整表名和字段结构 if data_source == 'nonprod_dataset': # 错位生产账户，使用非生产数据集的结构 nonprod_config = self.environments['nonprod'] full_table_name = f"{nonprod_config['dataset_name']}.{table_name}" date_field = nonprod_config['date_field'] cost_field = nonprod_config['cost_field'] product_field = nonprod_config['product_field'] # 对于错位生产账户，检查是否有详细标签字段 # 如果有标签字段，使用完整查询；否则使用简化查询 has_tags_query = f""" SELECT column_name FROM `{nonprod_config['dataset_name']}.INFORMATION_SCHEMA.COLUMNS` WHERE table_name = '{account_id}_aws_billing_data' AND column_name LIKE 'resource_tags_%' LIMIT 1 """ try: tag_check_result = self.bq_client.query(has_tags_query).result() has_resource_tags = len(list(tag_check_result)) > 0 except: has_resource_tags = False if has_resource_tags: # 使用生产环境的完整查询（含标签） print(f"错位生产账户 {account_id} 包含资源标签，使用完整查询") return self._construct_full_prod_query(account_id, full_table_name, start_date, end_date, date_field, cost_field, product_field) else: # 使用简化查询（无标签） print(f"错位生产账户 {account_id} 不包含资源标签，使用简化查询") return self._construct_simple_prod_query(account_id, full_table_name, start_date, end_date, date_field, cost_field, product_field) else: # 正常生产账户，使用生产数据集的结构 full_table_name = f"{config['dataset_name']}.{table_name}" date_field = config['date_field'] cost_field = config['cost_field'] product_field = config['product_field'] return self._construct_full_prod_query(account_id, full_table_name, start_date, end_date, date_field, cost_field, product_field) def _construct_full_prod_query(self, account_id: str, full_table_name: str, start_date: str, end_date: str, date_field: str, cost_field: str, product_field: str) -> str: """构建包含完整标签的生产查询""" config = self.environments['prod'] actual_end_date = self._get_safe_end_date(end_date, config['exclude_recent_days']) where_clause = f""" WHERE DATE({date_field}) >= '{start_date}' AND DATE({date_field}) <= '{actual_end_date}' AND {cost_field} IS NOT NULL AND line_item_line_item_type NOT IN ( 'Tax', 'Credit', 'Refund', 'Fee', 'RIFee' ) AND {cost_field} != 0 """ query = f""" SELECT DATE({date_field}) as day, '{account_id}' as account_id, {product_field} as product, resource_tags_user_component as component, resource_tags_user_servicetype as servicetype, product_region as region, resource_tags_user_tenant as tenant, resource_tags_user_usedby as usedby, ROUND(SUM({cost_field}), 4) as cost FROM `{full_table_name}` {where_clause} GROUP BY day, product, component, servicetype, region, tenant, usedby HAVING ABS(cost) > 0.001 ORDER BY day DESC, cost DESC """ return query def _construct_simple_prod_query(self, account_id: str, full_table_name: str, start_date: str, end_date: str, date_field: str, cost_field: str, product_field: str) -> str: """构建简化的生产查询（无标签字段）""" config = self.environments['prod'] actual_end_date = self._get_safe_end_date(end_date, config['exclude_recent_days']) where_clause = f""" WHERE DATE({date_field}) >= '{start_date}' AND DATE({date_field}) <= '{actual_end_date}' AND line_item_line_item_type NOT IN ( 'Tax', 'Credit', 'Refund', 'Fee', 'RIFee' ) AND {cost_field} != 0 """ query = f""" SELECT DATE({date_field}) as day, '{account_id}' as account_id, {product_field} as product, '' as component, '' as servicetype, '' as region, '' as tenant, '' as usedby, ROUND(SUM({cost_field}), 4) as cost FROM `{full_table_name}` {where_clause} GROUP BY day, product HAVING ABS(cost) > 0.001 ORDER BY day DESC, cost DESC """ return query def _get_safe_end_date(self, end_date: str, exclude_recent_days: int = 1) -> str: """获取考虑延迟的安全结束日期""" if isinstance(end_date, str): end_date_obj = datetime.strptime(end_date, '%Y-%m-%d').date() else: end_date_obj = end_date today = datetime.now().date() safe_date = today - timedelta(days=exclude_recent_days) return min(end_date_obj, safe_date).strftime('%Y-%m-%d') def construct_shared_account_query(self, account_id: str, start_date: str, end_date: str) -> str: """为公共测试账户构建包含tag的查询""" config = self.environments['shared'] table = f"`{config['dataset_name']}.{account_id}_aws_billing_data`" actual_end_date = self._get_safe_end_date(end_date, config['exclude_recent_days']) # 首先检查表是否真的有tag字段 if not self._verify_table_has_tags(config['dataset_name'], f"{account_id}_aws_billing_data"): print(f"⚠️ 警告：公共测试账户 {account_id} 的表中未找到tag字段，使用简化查询") return self._construct_shared_simple_query(account_id, table, start_date, actual_end_date, config) print(f"✅ 公共测试账户 {account_id} 支持tag数据，使用完整查询") return self._construct_shared_full_query(account_id, table, start_date, actual_end_date, config) def _construct_shared_full_query(self, account_id: str, table: str, start_date: str, end_date: str, config: Dict) -> str: """构建公共测试账户的完整tag查询""" return f""" SELECT DATE({config['date_field']}) as day, '{account_id}' as account_id, {config['product_field']} as product, COALESCE(resource_tags_user_component, '') as component, COALESCE(resource_tags_user_servicetype, '') as servicetype, COALESCE(product_region, '') as region, COALESCE(resource_tags_user_tenant, '') as tenant, COALESCE(resource_tags_user_usedby, '') as usedby, ROUND(SUM({config['cost_field']}), 4) as cost FROM {table} WHERE DATE({config['date_field']}) >= '{start_date}' AND DATE({config['date_field']}) <= '{end_date}' AND line_item_line_item_type NOT IN ('Tax', 'Credit', 'Refund', 'Fee', 'RIFee') AND {config['cost_field']} != 0 GROUP BY day, product, component, servicetype, region, tenant, usedby HAVING ABS(cost) > 0.001 ORDER BY day DESC, cost DESC """ def _construct_shared_simple_query(self, account_id: str, table: str, start_date: str, end_date: str, config: Dict) -> str: """公共测试账户的降级查询（当没有tag字段时）""" return f""" SELECT DATE({config['date_field']}) as day, '{account_id}' as account_id, {config['product_field']} as product, '' as component, '' as servicetype, '' as region, '' as tenant, '' as usedby, ROUND(SUM({config['cost_field']}), 4) as cost FROM {table} WHERE DATE({config['date_field']}) >= '{start_date}' AND DATE({config['date_field']}) <= '{end_date}' AND line_item_line_item_type NOT IN ('Tax', 'Credit', 'Refund', 'Fee', 'RIFee') AND {config['cost_field']} != 0 GROUP BY day, product HAVING ABS(cost) > 0.001 ORDER BY day DESC, cost DESC """ def query_shared_billing_data(self, account_id: str, start_date: str, end_date: str) -> List[Dict]: """查询公共测试账户的账单数据""" try: query = self.construct_shared_account_query(account_id, start_date, end_date) print(f"正在查询公共测试账户 {account_id} " f"从 {start_date} 到 {end_date} 的数据...") results = self.bq_client.query(query).result() billing_data = [] for row in results: billing_data.append({ 'date': row[0].strftime('%Y-%m-%d'), 'account_id': row[1], 'environment': 'shared', 'product': row[2], 'component': row[3] or '', 'servicetype': row[4] or '', 'region': row[5] or '', 'tenant': row[6] or '', 'usedby': row[7] or '', 'cost': float(row[8]) }) print(f"公共测试账户 {account_id} 查询完成，共 {len(billing_data)} 条记录") return billing_data except Exception as e: print(f"查询公共测试账户 {account_id} 时出错: {e}") return [] def query_nonprod_billing_data(self, account: Dict, start_date: str, end_date: str) -> List[Dict]: """查询单个非生产账户的账单数据""" try: query = self.construct_nonprod_query( account['account_id'], start_date, end_date ) print(f"正在查询非生产账户 {account['account_id']} " f"从 {start_date} 到 {end_date} 的数据...") results = self.bq_client.query(query).result() billing_data = [] for row in results: billing_data.append({ 'date': row[0].strftime('%Y-%m-%d'), 'account_id': row[3], 'environment': 'nonprod', 'product': row[2], 'cost': float(row[1]) }) print(f"非生产账户 {account['account_id']} 查询完成，共 {len(billing_data)} 条记录") return billing_data except Exception as e: print(f"查询非生产账户 {account['account_id']} 时出错: {e}") return [] def query_prod_billing_data(self, account: Dict, start_date: str, end_date: str) -> List[Dict]: """查询单个生产账户的账单数据""" try: account_id = account['account_id'] table_name = account['table_name'] data_source = account.get('data_source', 'prod_dataset') description = account.get('description', '') source_info = f"错位账户({description})" if data_source == 'nonprod_dataset' else "正常生产账户" print(f"正在查询生产账户 {account_id} ({source_info}, 表: {table_name}) " f"从 {start_date} 到 {end_date} 的数据...") query = self.construct_prod_query(account_id, table_name, start_date, end_date, data_source) results = self.bq_client.query(query).result() billing_data = [] for row in results: billing_data.append({ 'date': row[0].strftime('%Y-%m-%d'), 'account_id': row[1], 'environment': 'prod', 'product': row[2], 'component': row[3] or '', 'servicetype': row[4] or '', 'region': row[5] or '', 'tenant': row[6] or '', 'usedby': row[7] or '', 'cost': float(row[8]) }) print(f"生产账户 {account_id} 查询完成，共 {len(billing_data)} 条记录") return billing_data except Exception as e: print(f"查询生产账户 {account['account_id']} 时出错: {e}") return [] def save_to_csv(self, billing_data: List[Dict], filepath: Path, environment: str): """保存数据到 CSV 文件""" if not billing_data: print(f"没有数据需要保存到 {filepath}") return # 根据环境确定字段 if environment == 'nonprod': fieldnames = ['date', 'account_id', 'environment', 'product', 'cost'] elif environment in ['shared', 'prod', 'misplaced_prod']: # 支持tag数据的环境 fieldnames = ['date', 'account_id', 'environment', 'product', 'component', 'servicetype', 'region', 'tenant', 'usedby', 'cost'] else: # 默认使用简化格式 fieldnames = ['date', 'account_id', 'environment', 'product', 'cost'] with open(filepath, 'w', newline='', encoding='utf-8') as csvfile: writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() for row in billing_data: # 只写入存在的字段 filtered_row = {k: v for k, v in row.items() if k in fieldnames} writer.writerow(filtered_row) print(f"数据已保存到: {filepath}") def save_monthly_csv_files(self, billing_data: List[Dict], environment: str, merge_existing: bool = True) -> List[str]: """按月份和账户保存CSV文件""" if not billing_data: return [] config = self.environments[environment] monthly_data = {} # 按账户和月份分组数据 for item in billing_data: date_parts = item['date'].split('-') year_month = f"{date_parts[0]}-{date_parts[1]}" account_id = item['account_id'] key = (account_id, year_month) if key not in monthly_data: monthly_data[key] = [] monthly_data[key].append(item) # 保存每个账户每月的数据 saved_files = [] for (account_id, year_month), data in monthly_data.items(): filename = f"{config['file_prefix']}_{account_id}_{year_month}.csv" filepath = config['output_dir'] / filename final_data = data # 如果文件已存在且需要合并 if merge_existing and filepath.exists(): existing_data = self._load_existing_csv_data(filepath, environment) if existing_data: final_data = self._merge_and_deduplicate_data( existing_data, data, environment ) overlap_count = len(existing_data) + len(data) - len(final_data) if overlap_count > 0: print(f"与现有文件 {filename} 合并数据，覆盖了 {overlap_count} 条重复记录") self.save_to_csv(final_data, filepath, environment) saved_files.append(filename) return saved_files def _load_existing_csv_data(self, filepath: Path, environment: str) -> List[Dict]: """加载现有CSV文件数据""" try: existing_data = [] with open(filepath, 'r', encoding='utf-8') as csvfile: reader = csv.DictReader(csvfile) for row in reader: row['cost'] = float(row['cost']) existing_data.append(row) return existing_data except Exception as e: print(f"读取现有文件 {filepath} 失败: {e}") return [] def _merge_and_deduplicate_data(self, existing_data: List[Dict], new_data: List[Dict], environment: str) -> List[Dict]: """合并并去重数据，新数据优先""" data_dict = {} # 确定去重键 if environment == 'nonprod': get_key = lambda x: (x['date'], x['account_id'], x['product']) else: # prod get_key = lambda x: (x['date'], x['account_id'], x['product'], x.get('component', ''), x.get('servicetype', ''), x.get('region', ''), x.get('tenant', ''), x.get('usedby', '')) # 先处理现有数据 for item in existing_data: key = get_key(item) data_dict[key] = item # 再处理新数据，会覆盖重复的key for item in new_data: key = get_key(item) data_dict[key] = item # 转换回列表并排序 merged_data = list(data_dict.values()) merged_data.sort(key=lambda x: (x['date'], x['product'])) return merged_data def export_billing_data(self, start_date: str, end_date: str, environments: List[str], max_workers: int = 5, merge_existing: bool = True, account_filter: Optional[str] = None) -> Dict: """导出指定环境和日期范围的账单数据""" results = {} for env in environments: print(f"\n=== 开始导出 {env.upper()} 环境数据 ===") if env == 'nonprod': result = self._export_nonprod_data( start_date, end_date, max_workers, merge_existing, account_filter ) elif env == 'shared': result = self._export_shared_data( start_date, end_date, max_workers, merge_existing, account_filter ) elif env == 'prod': result = self._export_prod_data( start_date, end_date, merge_existing, account_filter ) else: result = { 'success': False, 'message': f'不支持的环境类型: {env}', 'total_records': 0, 'total_cost': 0.0, 'saved_files': [], 'account_ids': [] } results[env] = result return results def _export_nonprod_data(self, start_date: str, end_date: str, max_workers: int, merge_existing: bool, account_filter: Optional[str]) -> Dict: """导出非生产环境数据""" # 获取账户列表 all_accounts = self.get_nonprod_accounts() if not all_accounts: return {'success': False, 'message': '未找到任何 AWS 非生产账户'} # 过滤账户 if account_filter: accounts = [acc for acc in all_accounts if acc['account_id'] == account_filter] if not accounts: return {'success': False, 'message': f'未找到指定的非生产账户: {account_filter}'} else: accounts = all_accounts print(f"将导出 {len(accounts)} 个非生产账户的数据") # 并发查询 all_billing_data = [] with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: future_to_account = { executor.submit( self.query_nonprod_billing_data, account, start_date, end_date ): account for account in accounts } for future in concurrent.futures.as_completed(future_to_account): account = future_to_account[future] try: billing_data = future.result() all_billing_data.extend(billing_data) except Exception as e: print(f"处理非生产账户 {account['account_id']} 时出错: {e}") if not all_billing_data: return {'success': False, 'message': '没有查询到任何非生产账单数据'} # 保存数据 all_billing_data.sort(key=lambda x: (x['date'], x['account_id'], x['product'])) saved_files = self.save_monthly_csv_files(all_billing_data, 'nonprod', merge_existing) return { 'success': True, 'environment': 'nonprod', 'total_records': len(all_billing_data), 'account_count': len(set(item['account_id'] for item in all_billing_data)), 'saved_files': saved_files, 'total_cost': sum(item['cost'] for item in all_billing_data) } def _export_shared_data(self, start_date: str, end_date: str, max_workers: int, merge_existing: bool, account_filter: Optional[str]) -> Dict: """导出公共测试账户数据""" # 从账户详情中筛选公共测试账户 shared_accounts = [] for account_id, account_info in self.account_details.items(): if account_info.get('is_shared_dev_account', False): # 检查账户过滤器 if account_filter and account_id != account_filter: continue shared_accounts.append(account_id) if not shared_accounts: print("ℹ️ 未找到公共测试账户或所有账户都被过滤器排除") return { 'success': True, 'environment': 'shared', 'total_records': 0, 'account_count': 0, 'saved_files': [], 'total_cost': 0.0 } print(f"将导出 {len(shared_accounts)} 个公共测试账户的数据: {shared_accounts}") # 并发查询 all_billing_data = [] with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: future_to_account = { executor.submit(self.query_shared_billing_data, account_id, start_date, end_date): account_id for account_id in shared_accounts } for future in concurrent.futures.as_completed(future_to_account): account_id = future_to_account[future] try: account_data = future.result() all_billing_data.extend(account_data) except Exception as e: print(f"查询公共测试账户 {account_id} 时出错: {e}") # 保存数据 all_billing_data.sort(key=lambda x: (x['date'], x['account_id'], x['product'])) saved_files = self.save_monthly_csv_files(all_billing_data, 'shared', merge_existing) return { 'success': True, 'environment': 'shared', 'total_records': len(all_billing_data), 'account_count': len(set(item['account_id'] for item in all_billing_data)), 'saved_files': saved_files, 'total_cost': sum(item['cost'] for item in all_billing_data) } def _export_prod_data(self, start_date: str, end_date: str, merge_existing: bool, account_filter: Optional[str]) -> Dict: """导出生产环境数据""" # 获取账户列表 all_accounts = self.get_prod_accounts() if not all_accounts: return {'success': False, 'message': '未找到任何 AWS 生产账户'} # 分离正常生产账户和错位生产账户 normal_prod_accounts = [] misplaced_prod_accounts = [] for account in all_accounts: if account.get('data_source') == 'nonprod_dataset': misplaced_prod_accounts.append(account) else: normal_prod_accounts.append(account) # 过滤账户 if account_filter: normal_prod_accounts = [acc for acc in normal_prod_accounts if acc['account_id'] == account_filter] misplaced_prod_accounts = [acc for acc in misplaced_prod_accounts if acc['account_id'] == account_filter] if not normal_prod_accounts and not misplaced_prod_accounts: return {'success': False, 'message': f'未找到指定的生产账户: {account_filter}'} # 导出正常生产账户数据 normal_result = self._export_normal_prod_accounts( normal_prod_accounts, start_date, end_date, merge_existing ) # 导出错位生产账户数据 misplaced_result = self._export_misplaced_prod_accounts( misplaced_prod_accounts, start_date, end_date, merge_existing ) # 合并结果 total_records = normal_result.get('total_records', 0) + misplaced_result.get('total_records', 0) total_cost = normal_result.get('total_cost', 0) + misplaced_result.get('total_cost', 0) saved_files = normal_result.get('saved_files', []) + misplaced_result.get('saved_files', []) account_count = len(set( normal_result.get('account_ids', []) + misplaced_result.get('account_ids', []) )) return { 'success': True, 'environment': 'prod', 'total_records': total_records, 'account_count': account_count, 'saved_files': saved_files, 'total_cost': total_cost, 'normal_prod': normal_result, 'misplaced_prod': misplaced_result } def _export_normal_prod_accounts(self, accounts: List[Dict], start_date: str, end_date: str, merge_existing: bool) -> Dict: """导出正常生产账户数据""" if not accounts: return {'total_records': 0, 'total_cost': 0, 'saved_files': [], 'account_ids': []} print(f"导出 {len(accounts)} 个正常生产账户的数据") # 串行查询每个账户 all_billing_data = [] account_ids = [] for account in accounts: try: billing_data = self.query_prod_billing_data(account, start_date, end_date) all_billing_data.extend(billing_data) account_ids.append(account['account_id']) except Exception as e: print(f"处理正常生产账户 {account['account_id']} 时出错: {e}") if not all_billing_data: return {'total_records': 0, 'total_cost': 0, 'saved_files': [], 'account_ids': account_ids} # 保存数据到正常生产目录 all_billing_data.sort(key=lambda x: (x['date'], x['account_id'], x['product'])) saved_files = self.save_monthly_csv_files(all_billing_data, 'prod', merge_existing) return { 'total_records': len(all_billing_data), 'total_cost': sum(item['cost'] for item in all_billing_data), 'saved_files': saved_files, 'account_ids': account_ids } def _export_misplaced_prod_accounts(self, accounts: List[Dict], start_date: str, end_date: str, merge_existing: bool) -> Dict: """导出错位生产账户数据""" if not accounts: return {'total_records': 0, 'total_cost': 0, 'saved_files': [], 'account_ids': []} print(f"导出 {len(accounts)} 个错位生产账户的数据到专门目录") # 串行查询每个账户 all_billing_data = [] account_ids = [] for account in accounts: try: billing_data = self.query_prod_billing_data(account, start_date, end_date) all_billing_data.extend(billing_data) account_ids.append(account['account_id']) except Exception as e: print(f"处理错位生产账户 {account['account_id']} 时出错: {e}") if not all_billing_data: return {'total_records': 0, 'total_cost': 0, 'saved_files': [], 'account_ids': account_ids} # 保存数据到错位生产专门目录 all_billing_data.sort(key=lambda x: (x['date'], x['account_id'], x['product'])) saved_files = self.save_monthly_csv_files(all_billing_data, 'misplaced_prod', merge_existing) return { 'total_records': len(all_billing_data), 'total_cost': sum(item['cost'] for item in all_billing_data), 'saved_files': saved_files, 'account_ids': account_ids } def parse_args(): """解析命令行参数""" parser = argparse.ArgumentParser( description='AWS 统一账单数据导出工具', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 示例用法: %(prog)s --last-week # 导出所有环境最近一周 %(prog)s --env nonprod --days 30 # 只导出非生产环境最近30天 %(prog)s --env prod --start-date 2024-01-01 --end-date 2024-01-07 # 只导出生产环境指定区间 %(prog)s --start-date 2024-01-01 --end-date 2024-01-07 # 导出所有环境指定区间 %(prog)s --env nonprod --account-id 385595570414 --last-week # 导出指定非生产账户最近一周 """ ) # 环境选择 parser.add_argument( '--env', choices=['nonprod', 'shared', 'prod', 'all'], default='all', help='要导出的环境 (nonprod=非生产, shared=公共测试, prod=生产, all=全部，默认: all)' ) parser.add_argument( '--account-id', help='只导出指定账户ID的数据' ) # 日期选项组 date_group = parser.add_mutually_exclusive_group(required=True) date_group.add_argument( '--last-week', action='store_true', help='查询最近一周的数据' ) date_group.add_argument( '--days', type=int, help='查询最近N天的数据' ) date_group.add_argument( '--start-date', help='开始日期 (YYYY-MM-DD)' ) parser.add_argument( '--end-date', help='结束日期 (YYYY-MM-DD)，需要与 --start-date 一起使用' ) parser.add_argument( '--max-workers', type=int, default=5, help='非生产环境并发查询的最大线程数 (默认: 5)' ) parser.add_argument( '--overwrite', action='store_true', help='覆盖现有的月度文件，而不是合并数据' ) parser.add_argument( '--refresh-cache', action='store_true', help='强制刷新账户缓存' ) parser.add_argument( '--config', help='配置文件路径 (JSON格式)' ) return parser.parse_args() def load_config_file(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_file(args.config) print(f"使用指定配置文件: {args.config}") else: # 自动查找配置文件 config_file = find_config_file() if config_file: config = load_config_file(config_file) print(f"自动加载配置文件: {config_file}") else: print("未找到配置文件，使用默认配置") # 计算查询日期范围 today = datetime.now().date() if args.last_week: end_date = today - timedelta(days=1) # 昨天 start_date = end_date - timedelta(days=6) # 一周前 elif args.days: end_date = today - timedelta(days=1) # 昨天 start_date = end_date - timedelta(days=args.days - 1) elif args.start_date: if not args.end_date: print("错误: 使用 --start-date 时必须同时指定 --end-date") sys.exit(1) try: start_date = datetime.strptime(args.start_date, '%Y-%m-%d').date() end_date = datetime.strptime(args.end_date, '%Y-%m-%d').date() except ValueError as e: print(f"错误: 日期格式无效 - {e}") sys.exit(1) if start_date > end_date: print("错误: 开始日期不能晚于结束日期") sys.exit(1) # 确定要导出的环境 if args.env == 'all': environments = ['nonprod', 'shared', 'prod'] else: environments = [args.env] try: # 创建导出器 exporter = AWSUnifiedBillingExporter() # 应用配置文件设置 if config: exporter.update_from_config(config) # 如果指定刷新缓存，删除现有缓存文件 if args.refresh_cache: for env_config in exporter.environments.values(): cache_patterns = ['*accounts_cache.json', '*_accounts_cache.json'] for pattern in cache_patterns: for cache_file in env_config['output_dir'].glob(pattern): cache_file.unlink() print(f"已删除缓存文件: {cache_file}") # 获取max_workers参数，优先使用命令行参数，其次使用配置文件 max_workers = args.max_workers if max_workers == 5 and config.get('export_settings', {}).get('max_workers'): max_workers = config['export_settings']['max_workers'] # 导出数据 results = exporter.export_billing_data( start_date.strftime('%Y-%m-%d'), end_date.strftime('%Y-%m-%d'), environments, max_workers, merge_existing=not args.overwrite, account_filter=args.account_id ) # 显示结果 print("\n=== 导出完成 ===") for env, result in results.items(): if result['success']: print(f"\n{env.upper()} 环境:") print(f" 总记录数: {result['total_records']:,}") print(f" 账户数: {result['account_count']}") print(f" 总费用: ${result['total_cost']:,.2f}") print(f" 生成文件数: {len(result['saved_files'])}") if len(result['saved_files']) <= 10: print(" 生成的文件:") for filename in sorted(result['saved_files']): print(f" - {filename}") else: print(f" 生成的文件: {len(result['saved_files'])} 个文件（过多不显示）") else: print(f"\n{env.upper()} 环境导出失败: {result['message']}") except Exception as e: print(f"导出过程中发生错误: {e}") import traceback traceback.print_exc() sys.exit(1) if __name__ == "__main__": main()