# analyze_costs.py
import anthropic
import pandas as pd
import json

def analyze_cur_with_claude(csv_path: str, question: str) -> str:
    """Analyze a Cost and Usage Report CSV with Claude."""
    client = anthropic.Anthropic()

    # Load and summarize CUR (full CUR files are too large for direct input)
    df = pd.read_csv(csv_path, low_memory=False)

    # Aggregate by service and usage type
    summary = (
        df.groupby(['product/serviceName', 'lineItem/UsageType'])
        ['lineItem/UnblendedCost']
        .agg(['sum', 'count'])
        .reset_index()
        .sort_values('sum', ascending=False)
        .head(50)
    )

    # Also get day-over-day trend
    daily = (
        df.groupby('lineItem/UsageStartDate')
        ['lineItem/UnblendedCost']
        .sum()
        .tail(30)
    )

    context = f"""
Cost and Usage Report Summary (top 50 line items):
{summary.to_string()}

Daily spend trend (last 30 days):
{daily.to_string()}

Total period spend: ${df['lineItem/UnblendedCost'].sum():,.2f}
"""

    message = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[
            {
                "role": "user",
                "content": f"Analyze this AWS Cost and Usage Report:\n\n{context}\n\nQuestion: {question}"
            }
        ]
    )

    return message.content[0].text


# Example usage
report = analyze_cur_with_claude(
    "aws-cur-march-2026.csv",
    "Which services have the highest month-over-month growth, and what are the top 3 savings opportunities?"
)
print(report)

# weekly_digest.py
import boto3
import anthropic
from datetime import date, timedelta

def get_weekly_cost_data() -> dict:
    ce = boto3.client("ce", region_name="us-east-1")
    end = date.today()
    start = end - timedelta(days=14)  # 2 weeks for MoM comparison

    response = ce.get_cost_and_usage(
        TimePeriod={"Start": str(start), "End": str(end)},
        Granularity="DAILY",
        Metrics=["UnblendedCost"],
        GroupBy=[{"Type": "DIMENSION", "Key": "SERVICE"}],
    )
    return response["ResultsByTime"]

def generate_digest(cost_data: dict) -> str:
    client = anthropic.Anthropic()

    message = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1500,
        messages=[{
            "role": "user",
            "content": (
                "Generate a concise weekly cloud cost digest for an engineering team. "
                "Highlight top cost drivers, any anomalies vs the prior week, "
                "and 2-3 concrete savings recommendations.\n\n"
                f"Cost data (last 14 days by service/day):\n{cost_data}"
            ),
        }],
    )
    return message.content[0].text

if __name__ == "__main__":
    data = get_weekly_cost_data()
    digest = generate_digest(data)
    print(digest)
    # In production: post to Slack, send email, etc.

#!/usr/bin/env python3
# cost_analyzer.py — Custom AI-powered cost dashboard

import anthropic
import boto3
import json
from datetime import datetime, timedelta

class CostAnalyzer:
    def __init__(self):
        self.ec2 = boto3.client('ec2')
        self.ce = boto3.client('ce')  # Cost Explorer
        self.client = anthropic.Anthropic()

    def get_cost_anomalies(self, days: int = 7) -> dict:
        """Fetch cost data and anomalies from Cost Explorer."""
        end_date = datetime.now().date()
        start_date = end_date - timedelta(days=days)

        # Get daily costs
        response = self.ce.get_cost_and_usage(
            TimePeriod={
                'Start': start_date.isoformat(),
                'End': end_date.isoformat()
            },
            Granularity='DAILY',
            Metrics=['UnblendedCost'],
            GroupBy=[
                {'Type': 'DIMENSION', 'Key': 'SERVICE'}
            ]
        )

        # Format for Claude
        cost_by_service = {}
        for result in response['ResultsByTime']:
            date = result['TimePeriod']['Start']
            for group in result['Groups']:
                service = group['Keys'][0]
                cost = float(group['Metrics']['UnblendedCost']['Amount'])
                if service not in cost_by_service:
                    cost_by_service[service] = []
                cost_by_service[service].append({'date': date, 'cost': cost})

        return cost_by_service

    def get_unused_resources(self) -> dict:
        """Identify unused EC2 and EBS resources."""
        cloudwatch = boto3.client('cloudwatch')
        ec2 = boto3.client('ec2')

        # Find EC2 instances with zero CPU usage
        instances = ec2.describe_instances()['Reservations']
        idle_instances = []

        for reservation in instances:
            for instance in reservation['Instances']:
                instance_id = instance['InstanceId']
                # Check CloudWatch metrics
                response = cloudwatch.get_metric_statistics(
                    Namespace='AWS/EC2',
                    MetricName='CPUUtilization',
                    Dimensions=[{'Name': 'InstanceId', 'Value': instance_id}],
                    StartTime=datetime.now() - timedelta(days=7),
                    EndTime=datetime.now(),
                    Period=3600,
                    Statistics=['Average']
                )

                if response['Datapoints']:
                    avg_cpu = sum(d['Average'] for d in response['Datapoints']) / len(response['Datapoints'])
                    if avg_cpu < 5:  # Less than 5% average CPU
                        idle_instances.append({
                            'instance_id': instance_id,
                            'avg_cpu': avg_cpu,
                            'instance_type': instance['InstanceType'],
                            'state': instance['State']['Name']
                        })

        return {'idle_instances': idle_instances}

    def analyze_with_claude(self) -> str:
        """Use Claude to analyze costs and generate recommendations."""
        costs = self.get_cost_anomalies()
        unused = self.get_unused_resources()

        # Format comprehensive prompt
        analysis_prompt = f"""Analyze our AWS costs and usage for the past week.

Cost Breakdown by Service:
{json.dumps(costs, indent=2)}

Unused Resources:
{json.dumps(unused, indent=2)}

Provide:
1. Top 3 cost drivers and why they increased
2. Specific optimization recommendations with estimated savings
3. Resource utilization concerns (idle instances, oversized types)
4. Priority-ordered action items (quick wins first)
5. Rough implementation effort for each recommendation
"""

        message = self.client.messages.create(
            model="claude-opus-4-6",
            max_tokens=2048,
            messages=[{"role": "user", "content": analysis_prompt}]
        )

        return message.content[0].text


# Usage
if __name__ == "__main__":
    analyzer = CostAnalyzer()
    report = analyzer.analyze_with_claude()
    print(report)

    # Example output from Claude:
    # Top cost driver: RDS increased $5,200 (42% month-over-month)
    # - New db.r6g.2xlarge instance has 0 connections during business hours
    # - Recommendation: Enable Aurora auto-scaling for variable workloads, save ~$1,800/month
    # - Effort: 2 hours