#!/usr/bin/env python3 """ GCP 账单数据导出工具 (main_export.py) 功能: 1. 查询 GCP 项目的各服务每天账单数据 2. 按日期和项目保存为 CSV 文件，按环境（prod/nonprod）分类 3. 支持查询最近一周的数据 4. 支持查询指定日期区间的数据 5. 自动发现所有活跃的 GCP 项目 6. 根据查询日期自动选择合适的数据表 7. 支持环境过滤，按生产和非生产环境分别导出 用法: python -m gcp.main_export --last-week # 最近一周 python -m gcp.main_export --start-date 2025-04-01 --end-date 2025-09-30 # 指定区间 python -m gcp.main_export --days 30 # 最近30天 python -m gcp.main_export --env prod # 仅导出生产环境 python -m gcp.main_export --env nonprod # 仅导出非生产环境 """ import argparse import csv import json import sys from datetime import datetime, timedelta, date from pathlib import Path from typing import Optional, Union import concurrent.futures from google.cloud import bigquery from gcp.mysql_writer import GCPMySQLWriter GCP_DIR = Path(__file__).resolve().parent def _resolve_under_gcp(path: Union[str, Path]) -> Path: """Ensure paths default to the gcp package directory when relative.""" candidate = Path(path) return candidate if candidate.is_absolute() else (GCP_DIR / candidate) class GCPBillingExporter: """GCP 账单数据导出器""" def __init__( self, dataset_name: str = "gcp-digital-bi.gcp_billing_v2", config_file: Optional[str] = None, output_dir: Optional[Union[str, Path]] = None, ): self.bq_client = bigquery.Client() self.dataset_name = dataset_name resolved_output = _resolve_under_gcp(output_dir or "gcp_billing_exports") self.output_dir = resolved_output self.output_dir.mkdir(parents=True, exist_ok=True) self.cache_file = self.output_dir / "gcp_projects_cache.json" self.cache_expiry_hours = 24 # 缓存24小时 self.exclude_recent_days = 2 # 默认排除最近2天，确保数据准确性 # 配置相关 self.config = {} self.environment_config = {} if config_file: self.config = self._load_environment_config(config_file) else: default_config = _resolve_under_gcp("config.json") if default_config.exists(): self.config = self._load_environment_config(str(default_config)) self.environment_config = self.config.get('environments', self.environment_config) self.mysql_writer = GCPMySQLWriter(self.config) # GCP 账单表配置 - 基于 pch_billing.py 中的配置 self.billing_table_v1 = f"`{self.dataset_name}.gcp_billing_export_v1_01A169_D3D9BF_FB82A2`" self.billing_table_v2 = f"`{self.dataset_name}.gcp_billing_export_v1_01D088_8F9CF2_8AF1C6`" self.full_table = f"`{self.dataset_name}.gcp_billing_export_v1_*`" # 账单表版本切换点，便于后续扩展 self.table_versions = [ { "start": date(1, 1, 1), "end": date(2023, 4, 19), "table": self.full_table, }, { "start": date(2023, 4, 20), "end": date(2025, 11, 19), "table": self.billing_table_v1, }, { # 2025-11-20 为迁移日，旧表与新表各存一部分数据 "start": date(2025, 11, 20), "end": date(2025, 11, 20), "table": self.billing_table_v1, }, { "start": date(2025, 11, 20), "end": date(2025, 11, 20), "table": self.billing_table_v2, }, { "start": date(2025, 11, 21), "end": None, # 未来如果有新表，在这里继续追加 "table": self.billing_table_v2, }, ] def _parse_date(self, value: Union[str, datetime, date]) -> date: """将字符串或 datetime 转换为 date""" if isinstance(value, date): return value if isinstance(value, datetime): return value.date() return datetime.strptime(value, '%Y-%m-%d').date() def _get_table_for_date(self, query_date: date) -> str: """根据日期选择对应的账单表""" # 如果某天落在多个版本（如迁移日），优先返回最新表（后定义的版本） for version in reversed(self.table_versions): start = version["start"] end = version["end"] if start <= query_date and (end is None or query_date <= end): return version["table"] # 如果日期超出已知范围，默认使用最后一个版本 return self.table_versions[-1]["table"] def _get_table_segments(self, start_date_obj: date, end_date_obj: date): """按日期范围拆分需要查询的账单表区间""" segments = [] for version in self.table_versions: seg_start = max(start_date_obj, version["start"]) seg_end = min(end_date_obj, version["end"]) if version["end"] else end_date_obj if seg_start <= seg_end: segments.append({ "table": version["table"], "start": seg_start, "end": seg_end, }) return segments def _load_environment_config(self, config_file): """加载环境配置文件""" try: with open(config_file, 'r', encoding='utf-8') as f: config = json.load(f) self.environment_config = config.get('environments', {}) return config except Exception as e: print(f"加载环境配置文件失败: {e}") self.environment_config = {} return {} def get_project_environment(self, project_id): """确定项目属于哪个环境""" if not self.environment_config: # 没有配置文件时，默认所有项目为非生产环境 return 'nonprod' for env_name, env_config in self.environment_config.items(): if project_id in env_config.get('projects', []): return env_name # 如果没有在生产环境列表中找到，默认为非生产环境 return 'nonprod' def filter_projects_by_environment(self, projects, environment): """按环境过滤项目列表""" if environment == 'all' or not self.environment_config: return projects filtered_projects = [] for project in projects: project_env = self.get_project_environment(project['project_id']) if project_env == environment: filtered_projects.append(project) return filtered_projects def _is_cache_valid(self): """检查缓存是否有效""" if not self.cache_file.exists(): return False try: with open(self.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): """从缓存加载项目信息""" try: with open(self.cache_file, 'r', encoding='utf-8') as f: cache_data = json.load(f) return cache_data.get('projects', []) except (FileNotFoundError, json.JSONDecodeError): return [] def _save_cache(self, projects): """保存项目信息到缓存""" cache_data = { 'timestamp': datetime.now().isoformat(), 'projects': projects } with open(self.cache_file, 'w', encoding='utf-8') as f: json.dump(cache_data, f, indent=2, ensure_ascii=False) def get_gcp_projects(self, start_date: Optional[Union[str, date]] = None, end_date: Optional[Union[str, date]] = None): """获取活跃的 GCP 项目信息 如果提供了 start_date/end_date，则按指定区间覆盖对应的账单表（使用 table_versions） 来发现项目，避免只查看最新表导致老表项目缺失。 """ # 自定义区间时不使用缓存，确保项目覆盖完整 use_cache = start_date is None and end_date is None if use_cache and self._is_cache_valid(): print("从缓存加载 GCP 项目信息...") cached_projects = self._load_cache() if cached_projects: print(f"从缓存加载了 {len(cached_projects)} 个项目") return cached_projects print("从 BigQuery 查询 GCP 项目信息...") today = datetime.now().date() if start_date and end_date: range_start = self._parse_date(start_date) range_end = self._parse_date(end_date) else: range_end = today range_start = today - timedelta(days=90) # 针对目标区间落入的各个账单表构建查询，使用 table_versions 分段 segments = self._get_table_segments(range_start, range_end) segment_queries = [] for segment in segments: seg_start = segment["start"].strftime('%Y-%m-%d') seg_end = segment["end"].strftime('%Y-%m-%d') seg_end_next = (segment["end"] + timedelta(days=1)).strftime('%Y-%m-%d') segment_queries.append(f""" SELECT project.id as project_id, project.name as project_name, COUNT(*) as record_count, SUM((cost + ( IFNULL(( SELECT SUM(amount) FROM UNNEST(credits) ), 0) )) * currency_conversion_rate) as total_cost FROM {segment["table"]} WHERE _PARTITIONTIME >= TIMESTAMP('{seg_start}') AND _PARTITIONTIME < TIMESTAMP('{seg_end_next}') GROUP BY project.id, project.name """) union_query = "\nUNION ALL\n".join(segment_queries) query = f""" SELECT project_id, ANY_VALUE(project_name) as project_name, SUM(record_count) as record_count, SUM(total_cost) as total_cost FROM ( {union_query} ) GROUP BY project_id HAVING total_cost > 0 ORDER BY total_cost DESC """ try: results = self.bq_client.query(query).result() gcp_projects = [] for i, row in enumerate(results): project_id = row[0] project_name = row[1] or f"project_{project_id}" record_count = row[2] total_cost = float(row[3]) print(f"发现项目: {project_id} ({project_name}) - 最近30天费用: ${total_cost:.2f}") gcp_projects.append({ 'id': i + 1, # 自动生成ID 'project_id': project_id, 'project_name': project_name, 'record_count': record_count, 'total_cost': total_cost }) print(f"从 BigQuery 发现 {len(gcp_projects)} 个活跃的 GCP 项目") # 保存到缓存（仅默认模式） if use_cache: self._save_cache(gcp_projects) return gcp_projects except Exception as e: print(f"查询 BigQuery 项目信息时出错: {e}") # 如果查询失败，尝试从缓存加载 if use_cache: cached_projects = self._load_cache() if cached_projects: print(f"查询失败，从缓存加载了 {len(cached_projects)} 个项目") return cached_projects return [] def construct_gcp_query(self, project_id, start_date, end_date, exclude_recent_days=1): """构建 GCP 账单查询 SQL，基于 pch_billing.py 中的逻辑 Args: project_id: GCP项目ID start_date: 开始日期 end_date: 结束日期 exclude_recent_days: 排除最近N天的数据（确保数据准确性） """ start_date_obj = self._parse_date(start_date) end_date_obj = self._parse_date(end_date) # 计算排除延迟后的实际结束日期 actual_end_date = self._get_safe_end_date(end_date_obj, exclude_recent_days) # 如果调整后的结束日期早于开始日期，返回空查询 if actual_end_date < start_date_obj: return (f"SELECT DATE('1900-01-01') as day, 0.0 as cost, 'NoData' as product, " f"'{project_id}' as project_id WHERE FALSE") # 根据日期范围拆分不同账单表，避免假设后续不会新增表 segments = self._get_table_segments(start_date_obj, actual_end_date) if not segments: return (f"SELECT DATE('1900-01-01') as day, 0.0 as cost, 'NoData' as product, " f"'{project_id}' as project_id WHERE FALSE") segment_queries = [] for segment in segments: seg_start = segment["start"].strftime('%Y-%m-%d') seg_end = segment["end"].strftime('%Y-%m-%d') seg_end_next = (segment["end"] + timedelta(days=1)).strftime('%Y-%m-%d') segment_queries.append(f""" SELECT DATE(usage_start_time) as day, ROUND(SUM((cost + ( IFNULL(( SELECT SUM(amount) FROM UNNEST(credits) ), 0) )) * currency_conversion_rate), 4) as cost, service.description as product, project.id as project_id FROM {segment["table"]} WHERE project.id = '{project_id}' AND _PARTITIONTIME >= TIMESTAMP('{seg_start}') AND _PARTITIONTIME < TIMESTAMP('{seg_end_next}') AND DATE(usage_start_time) >= '{seg_start}' AND DATE(usage_start_time) <= '{seg_end}' GROUP BY project.id, product, day HAVING cost > 0 """) union_query = "\nUNION ALL\n".join(segment_queries) query = f""" SELECT day, cost, product, project_id FROM ( {union_query} ) ORDER BY day ASC, cost DESC """ return query def _get_safe_end_date(self, end_date, exclude_recent_days=1): """获取考虑延迟的安全结束日期""" end_date_obj = self._parse_date(end_date) # 计算今天减去延迟天数的日期 today = datetime.now().date() safe_date = today - timedelta(days=exclude_recent_days) # 返回较小的日期（不超过安全日期） return min(end_date_obj, safe_date) def query_gcp_billing_data(self, gcp_project, start_date, end_date): """查询单个 GCP 项目的账单数据""" try: start_date_obj = self._parse_date(start_date) end_date_obj = self._parse_date(end_date) query = self.construct_gcp_query( gcp_project['project_id'], start_date_obj, end_date_obj, self.exclude_recent_days ) # 计算实际查询的结束日期 actual_end_date = self._get_safe_end_date(end_date_obj, self.exclude_recent_days) actual_end_date_str = actual_end_date.strftime('%Y-%m-%d') # 显示实际查询范围 if actual_end_date < end_date_obj: print(f"正在查询项目 {gcp_project['project_name']} ({gcp_project['project_id']}) " f"从 {start_date_obj.strftime('%Y-%m-%d')} 到 {actual_end_date_str} 的数据（跳过最近{self.exclude_recent_days}天）...") else: print(f"正在查询项目 {gcp_project['project_name']} ({gcp_project['project_id']}) " f"从 {start_date_obj.strftime('%Y-%m-%d')} 到 {end_date_obj.strftime('%Y-%m-%d')} 的数据...") results = self.bq_client.query(query).result() billing_data = [] for row in results: billing_data.append({ 'date': row[0].strftime('%Y-%m-%d'), 'cost': float(row[1]), 'product': row[2], 'project_id': row[3], 'project_name': gcp_project['project_name'] }) print(f"项目 {gcp_project['project_name']} 查询完成，共 {len(billing_data)} 条记录") return billing_data except Exception as e: print(f"查询项目 {gcp_project['project_name']} 时出错: {e}") return [] def save_to_csv(self, billing_data, filename, subdir=None): """保存数据到 CSV 文件 Args: billing_data: 账单数据 filename: 文件名 subdir: 子目录名（用于环境分类） """ if not billing_data: print(f"没有数据需要保存到 {filename}") return # 确定保存路径 if subdir: target_dir = self.output_dir / subdir target_dir.mkdir(parents=True, exist_ok=True) filepath = target_dir / filename else: filepath = self.output_dir / filename with open(filepath, 'w', newline='', encoding='utf-8') as csvfile: fieldnames = ['date', 'project_id', 'project_name', 'product', 'cost'] writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() for row in billing_data: writer.writerow(row) print(f"数据已保存到: {filepath}") def save_monthly_csv_files(self, billing_data, merge_existing=True, environment=None): """按月份和项目保存CSV文件 Args: billing_data: 账单数据列表 merge_existing: 是否与现有文件合并（去重） """ if not billing_data: print("没有数据需要保存") return [] # 按项目和月份分组数据 monthly_data = {} for item in billing_data: # 提取年月信息 (YYYY-MM) date_parts = item['date'].split('-') year_month = f"{date_parts[0]}-{date_parts[1]}" project_id = item['project_id'] key = (project_id, year_month) if key not in monthly_data: monthly_data[key] = [] monthly_data[key].append(item) # 保存每个项目每月的数据 saved_files = [] for (project_id, year_month), new_data in monthly_data.items(): # 确定项目环境 project_env = self.get_project_environment(project_id) # 如果指定了环境过滤，只处理该环境的项目 if environment != 'all' and environment != project_env: continue # 创建环境目录 env_dir = self.output_dir / project_env env_dir.mkdir(exist_ok=True) filename = f"gcp_{project_id}_{year_month}.csv" filepath = env_dir / filename relative_path = f"{project_env}/{filename}" final_data = new_data # 如果文件已存在且需要合并 if merge_existing and filepath.exists(): existing_data = self._load_existing_csv_data(filepath) if existing_data: # 合并数据并去重，新数据覆盖旧数据 final_data = self._merge_and_deduplicate_data(existing_data, new_data) overlap_count = len(existing_data) + len(new_data) - len(final_data) print(f"与现有文件 {relative_path} 合并数据，覆盖了 {overlap_count} 条重复记录") self.save_to_csv(final_data, filename, project_env) saved_files.append(relative_path) if getattr(self, "mysql_writer", None) and self.mysql_writer.enabled: try: date_values = [ datetime.strptime(item['date'], '%Y-%m-%d').date() for item in final_data if item.get('date') ] if date_values: self.mysql_writer.sync( project_id=project_id, project_name=final_data[0].get('project_name', project_id), environment=project_env, records=final_data, start_date=min(date_values), end_date=max(date_values), source_file=filepath, ) except Exception as exc: print(f"⚠️ 写入 MySQL 失败 ({project_id}): {exc}") return saved_files def _load_existing_csv_data(self, filepath): """加载现有CSV文件数据""" try: existing_data = [] with open(filepath, 'r', encoding='utf-8') as csvfile: reader = csv.DictReader(csvfile) for row in reader: # 确保cost字段是浮点数 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, new_data): """合并并去重数据，新数据优先（覆盖旧数据）""" # 使用字典来去重和覆盖，基于date, project_id, product的组合 data_dict = {} # 先处理现有数据 for item in existing_data: key = (item['date'], item['project_id'], item['product']) data_dict[key] = item # 再处理新数据，会覆盖重复的key（新数据优先） for item in new_data: key = (item['date'], item['project_id'], item['product']) 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, end_date, max_workers=5, merge_existing=True, environment='all'): """导出指定日期范围的账单数据""" start_date_obj = self._parse_date(start_date) end_date_obj = self._parse_date(end_date) # 计算实际的查询结束日期 actual_end_date = self._get_safe_end_date(end_date_obj, self.exclude_recent_days) actual_end_date_str = actual_end_date.strftime('%Y-%m-%d') env_name = self.environment_config.get(environment, {}).get('name', environment) if environment != 'all' else '全部环境' if actual_end_date < end_date_obj: print(f"开始导出 {env_name} GCP 账单数据 ({start_date_obj.strftime('%Y-%m-%d')} 到 {actual_end_date_str})") print(f"⚠️ 为确保数据准确性，已跳过最近{self.exclude_recent_days}天的数据（GCP账单延迟）") else: print(f"开始导出 {env_name} GCP 账单数据 ({start_date_obj.strftime('%Y-%m-%d')} 到 {end_date_obj.strftime('%Y-%m-%d')})") # 获取所有 GCP 项目，按导出区间覆盖多版本账单表 gcp_projects = self.get_gcp_projects(start_date_obj, actual_end_date) if not gcp_projects: print("未找到任何活跃的 GCP 项目") return # 按环境过滤项目 if environment != 'all': original_count = len(gcp_projects) gcp_projects = self.filter_projects_by_environment(gcp_projects, environment) print(f"从 {original_count} 个项目中过滤出 {len(gcp_projects)} 个{env_name}项目") if not gcp_projects: print(f"未找到{env_name}的 GCP 项目") return else: print(f"找到 {len(gcp_projects)} 个活跃的 GCP 项目") # 并发查询所有项目数据 all_billing_data = [] with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: future_to_project = { executor.submit( self.query_gcp_billing_data, project, start_date, end_date ): project for project in gcp_projects } for future in concurrent.futures.as_completed(future_to_project): project = future_to_project[future] try: billing_data = future.result() all_billing_data.extend(billing_data) except Exception as e: print(f"处理项目 {project['project_name']} 时出错: {e}") if not all_billing_data: print("没有查询到任何账单数据") return # 按日期排序 all_billing_data.sort(key=lambda x: (x['date'], x['project_name'], x['product'])) # 使用新的按月份保存方式 saved_files = self.save_monthly_csv_files(all_billing_data, merge_existing, environment) # 统计信息 unique_projects = set(item['project_id'] for item in all_billing_data) unique_months = set() total_cost = sum(item['cost'] for item in all_billing_data) for item in all_billing_data: date_parts = item['date'].split('-') year_month = f"{date_parts[0]}-{date_parts[1]}" unique_months.add(year_month) print("\n导出完成!") print(f"总记录数: {len(all_billing_data)}") print(f"总费用: ${total_cost:.2f}") print(f"项目数: {len(unique_projects)}") print(f"月份数: {len(unique_months)}") print(f"生成文件数: {len(saved_files)}") print(f"输出目录: {self.output_dir.absolute()}") # 显示生成的文件列表 if saved_files: print("\n生成的文件:") for filename in sorted(saved_files): print(f" - {filename}") def parse_args(): """解析命令行参数""" parser = argparse.ArgumentParser( description='GCP 账户账单数据导出工具 (main_export.py)', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 示例用法: %(prog)s --last-week # 最近一周 %(prog)s --days 30 # 最近30天 %(prog)s --start-date 2024-01-01 --end-date 2024-01-07 # 指定区间 %(prog)s --last-week --env prod # 仅导出生产环境 %(prog)s --days 7 --env nonprod --config-file config.json # 非生产环境，使用配置文件 """ ) # 日期选项组 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( '--dataset-name', default='gcp-digital-bi.gcp_billing_v2', help='BigQuery 数据集名称 (默认: gcp-digital-bi.gcp_billing_v2)' ) parser.add_argument( '--refresh-cache', action='store_true', help='强制刷新项目缓存' ) parser.add_argument( '--cache-hours', type=int, default=72, help='缓存有效期（小时）(默认: 72)' ) parser.add_argument( '--overwrite', action='store_true', help='覆盖现有的月度文件，而不是合并数据' ) parser.add_argument( '--exclude-recent-days', type=int, default=2, help='排除最近N天的数据，因为GCP账单有延迟 (默认: 2天)' ) parser.add_argument( '--env', choices=['all', 'prod', 'nonprod'], default='all', help='选择导出的环境 (all: 全部, prod: 生产, nonprod: 非生产)' ) parser.add_argument( '--config-file', help='环境配置文件路径 (JSON格式)' ) return parser.parse_args() def main(): """主函数""" args = parse_args() # 计算查询日期范围（考虑GCP账单延迟） today = datetime.now().date() # 考虑账单延迟，实际的"最新可用日期" latest_available_date = today - timedelta(days=args.exclude_recent_days) if args.last_week: end_date = today # 今天（实际查询时会自动排除最近的延迟天数） start_date = today - timedelta(days=6) # 一周前 elif args.days: end_date = today # 今天（实际查询时会自动排除最近的延迟天数） start_date = today - 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) try: # 创建导出器，支持自定义参数 exporter = GCPBillingExporter(dataset_name=args.dataset_name, config_file=args.config_file) exporter.cache_expiry_hours = args.cache_hours exporter.exclude_recent_days = args.exclude_recent_days # 如果指定刷新缓存，删除现有缓存文件 if args.refresh_cache and exporter.cache_file.exists(): exporter.cache_file.unlink() print("已删除现有缓存，将重新获取项目信息") exporter.export_billing_data( start_date.strftime('%Y-%m-%d'), end_date.strftime('%Y-%m-%d'), args.max_workers, merge_existing=not args.overwrite, # 如果指定覆盖，则不合并 environment=args.env ) except Exception as e: print(f"导出过程中发生错误: {e}") import traceback traceback.print_exc() sys.exit(1) if __name__ == "__main__": main()