#!/usr/bin/env python3 """Alibaba Cloud billing export CLI.""" from __future__ import annotations import argparse import sys from pathlib import Path from typing import List, Optional from alibaba_cloud.src.export.exporter import AlibabaCloudBillingExporter def parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace: parser = argparse.ArgumentParser( description="Export Alibaba Cloud billing data from BigQuery into CSV snapshots.", formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) date_group = parser.add_mutually_exclusive_group() date_group.add_argument("--start-date", type=str, help="Start date (YYYY-MM-DD).") date_group.add_argument("--days", type=int, help="Export the most recent N days.") date_group.add_argument("--last-week", action="store_true", help="Shortcut for --days 7.") parser.add_argument("--end-date", type=str, help="End date (YYYY-MM-DD).") parser.add_argument( "--env", action="append", dest="environments", help="Environment filter; repeatable. Use 'all' for every environment.", ) parser.add_argument( "--account-id", action="append", dest="account_ids", help="Filter to one or more Alibaba Cloud account IDs.", ) parser.add_argument( "--config", type=str, help="Path to config.json. Defaults to alibaba_cloud/config.json.", ) parser.add_argument( "--dry-run", action="store_true", help="Run the query without writing CSV files.", ) parser.add_argument( "--no-merge", action="store_true", help="Overwrite monthly CSVs instead of merging with existing data.", ) parser.add_argument( "--output-dir", type=Path, help="Custom export directory (overrides config).", ) return parser.parse_args(argv) def main(argv: Optional[List[str]] = None) -> int: args = parse_args(argv) try: exporter = AlibabaCloudBillingExporter( config_path=args.config, output_dir=args.output_dir, ) start_date, end_date = exporter.resolve_date_range( start_date=args.start_date, end_date=args.end_date, days=args.days, last_week=args.last_week, ) stats = exporter.export( start_date=start_date, end_date=end_date, environments=args.environments, account_ids=args.account_ids, merge_existing=not args.no_merge, dry_run=args.dry_run, ) print( f"Export finished: {stats.rows_returned} rows across {stats.queried_days} days. " f"Wrote {stats.files_written} files covering {stats.start_date} → {stats.end_date}." ) return 0 except Exception as exc: # pylint: disable=broad-except print(f"Alibaba Cloud export failed: {exc}") return 1 if __name__ == "__main__": sys.exit(main())