#!/usr/bin/env python3 """ Cost Report Analyzer - Aggregate workspace-level cost data. CLI script that fetches and aggregates cost report data. """ import argparse import json import sys from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any # Add parent directory to path for imports sys.path.insert(0, str(Path(__file__).parent.parent)) from src.cost_report.cost_report_fetcher import fetch_cost_report_for_date_range from src.cost_report.cost_report_types import CostBucket def load_workspaces_metadata() -> dict[str, dict[str, Any]]: """ Load workspace metadata from output/usage/raw/workspaces.json. Returns: Dict mapping workspace_id to workspace metadata (name, color, created_at) """ workspaces_path = Path('output/usage/raw/workspaces.json') if not workspaces_path.exists(): print(f'⚠️ Warning: {workspaces_path} not found, workspace metadata will not be hydrated', file=sys.stderr) return {} try: with open(workspaces_path, 'r') as f: response = json.load(f) workspaces_metadata = {} for workspace_data in response.get('data', []): workspaces_metadata[workspace_data['id']] = { 'name': workspace_data.get('name', 'Default'), 'display_color': workspace_data.get('display_color', '#9B87F5'), 'created_at': workspace_data.get('created_at'), } return workspaces_metadata except Exception as e: print(f'⚠️ Warning: Failed to load workspaces metadata: {e}', file=sys.stderr) return {} def aggregate_cost_buckets(buckets: list[CostBucket], workspaces_metadata: dict[str, dict[str, Any]] | None = None) -> dict[str, Any]: """ Aggregate cost buckets by workspace with detailed breakdowns. Creates hierarchical structure: - workspace -> time_period -> model -> token_type -> cost - workspace -> time_period -> other_costs (web_search, etc.) Args: buckets: List of CostBucket objects workspaces_metadata: Optional dict mapping workspace_id to workspace metadata Returns: Dict with workspaces, total_cost, date_range, metadata """ if workspaces_metadata is None: workspaces_metadata = {} workspaces: dict[str, Any] = {} for bucket in buckets: workspace_id = bucket.workspace_id or 'default' cost = float(bucket.amount) # Initialize workspace if not exists if workspace_id not in workspaces: # Get workspace metadata if available ws_meta = workspaces_metadata.get(workspace_id, {}) # For 'default' workspace, use defaults if no metadata found if workspace_id == 'default': workspace_name = ws_meta.get('name', 'Default') workspace_color = ws_meta.get('display_color', '#9B87F5') workspace_created_at = ws_meta.get('created_at') else: # For other workspaces, try metadata first, fallback to description workspace_name = ws_meta.get('name', bucket.description) workspace_color = ws_meta.get('display_color', '#9B87F5') workspace_created_at = ws_meta.get('created_at') workspaces[workspace_id] = { 'workspace_id': workspace_id, 'workspace_name': workspace_name, 'workspace_color': workspace_color, 'workspace_created_at': workspace_created_at, 'total_cost': 0.0, 'models': {}, 'time_periods': {}, } # Add to total cost workspaces[workspace_id]['total_cost'] += cost # Add to aggregate model breakdown (for workspace-level totals) model = bucket.model or 'unknown' token_type = bucket.token_type or 'unknown' # Initialize model if not exists in aggregate if model not in workspaces[workspace_id]['models']: workspaces[workspace_id]['models'][model] = {} # Add cost to the appropriate token_type in aggregate if token_type not in workspaces[workspace_id]['models'][model]: workspaces[workspace_id]['models'][model][token_type] = 0.0 workspaces[workspace_id]['models'][model][token_type] += cost # Add to time period with detailed breakdown period_key = f'{bucket.starting_at}_{bucket.ending_at}' if period_key not in workspaces[workspace_id]['time_periods']: workspaces[workspace_id]['time_periods'][period_key] = { 'starting_at': bucket.starting_at, 'ending_at': bucket.ending_at, 'cost': 0.0, 'models': {}, 'other_costs': {}, } tp = workspaces[workspace_id]['time_periods'][period_key] tp['cost'] += cost # Handle token costs vs other costs (web_search, etc.) if bucket.cost_type == 'tokens' and bucket.model: # Token costs: group by model -> token_type if model not in tp['models']: tp['models'][model] = {} if token_type not in tp['models'][model]: tp['models'][model][token_type] = 0.0 tp['models'][model][token_type] += cost else: # Non-token costs (web_search, etc.) cost_type = bucket.cost_type or 'unknown' if cost_type not in tp['other_costs']: tp['other_costs'][cost_type] = 0.0 tp['other_costs'][cost_type] += cost # Convert to final result structure result = {} total_cost = 0.0 for workspace_id, data in workspaces.items(): # Sort time periods by starting_at and convert to list time_periods_list = sorted(data['time_periods'].values(), key=lambda x: x['starting_at']) result[workspace_id] = { 'workspace_id': data['workspace_id'], 'workspace_name': data['workspace_name'], 'workspace_color': data['workspace_color'], 'workspace_created_at': data['workspace_created_at'], 'total_cost': round(data['total_cost'], 2), 'models': data['models'], 'time_periods': time_periods_list, } total_cost += data['total_cost'] return { 'workspaces': result, 'total_cost': round(total_cost, 2), 'date_range': { 'start': buckets[0].starting_at if buckets else '', 'end': buckets[-1].ending_at if buckets else '', }, 'metadata': { 'fetched_at': datetime.now(timezone.utc).isoformat(), 'bucket_count': len(buckets), 'workspace_count': len(result), }, } def main() -> None: """Main entry point.""" parser = argparse.ArgumentParser(description='Fetch and analyze cost report data from Claude Admin API') # Date arguments date_group = parser.add_mutually_exclusive_group(required=True) date_group.add_argument('--days', type=int, help='Fetch last N days') date_group.add_argument('--start', type=str, help='Start date (YYYY-MM-DD)') parser.add_argument('--end', type=str, help='End date (YYYY-MM-DD, defaults to today)') parser.add_argument('--output', type=Path, default=Path('output/cost_report/cost_report.json'), help='Output file path') parser.add_argument('--debug', action='store_true', help='Enable debug output') args = parser.parse_args() # Calculate date range if args.days: end_date = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d') start_date = (datetime.now() - timedelta(days=args.days)).strftime('%Y-%m-%d') else: start_date = args.start end_date = args.end or (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d') if args.debug: print(f'Fetching cost report from {start_date} to {end_date}') # Fetch cost report data (uses fetcher functions) try: buckets = fetch_cost_report_for_date_range( start_date=start_date, end_date=end_date, group_by=['workspace_id', 'description'], ) if args.debug: print(f'Fetched {len(buckets)} cost buckets') # Load workspace metadata for hydration print('Loading workspace metadata from output/usage/raw/workspaces.json...') workspaces_metadata = load_workspaces_metadata() if workspaces_metadata: print(f'Loaded metadata for {len(workspaces_metadata)} workspaces') # Aggregate data with workspace metadata aggregated = aggregate_cost_buckets(buckets, workspaces_metadata) # Save to output file args.output.parent.mkdir(parents=True, exist_ok=True) with open(args.output, 'w') as f: json.dump(aggregated, f, indent=2) print('\n✅ Cost report generated!') print(f' JSON: {args.output}') print(f' Total cost: ${aggregated["total_cost"] / 100:.2f} (from {aggregated["total_cost"]:.2f} cents)') print(f' Workspaces: {aggregated["metadata"]["workspace_count"]}') print(' Raw data: output/cost_report/raw/') except Exception as e: import traceback print(f'❌ Error: {e}', file=sys.stderr) if args.debug: traceback.print_exc() sys.exit(1) if __name__ == '__main__': main()