The AI Data Processing Pipeline: Clean, Transform, Analyze, Report

# 1. Create sandbox
SANDBOX=$(curl -s -X POST "$SANDBOX_URL/api/sandboxes" \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"name":"data-pipeline","instance_type":"ab0t.medium","auto_stop_minutes":30}')
SID=$(echo "$SANDBOX" | jq -r '.sandbox_id')

# 2. Upload the dataset
curl -s -X POST "$SANDBOX_URL/api/sandboxes/$SID/upload" \
  -H "Authorization: Bearer $API_KEY" \
  -F "file=@sales_data_2025.csv"

# 3. Tell the agent what to do
curl -s -X POST "$SANDBOX_URL/api/sandboxes/$SID/execute" \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "command": "cd /workspace && pip install pandas matplotlib seaborn && python -c \"\nimport pandas as pd\nimport matplotlib.pyplot as plt\nimport seaborn as sns\n\ndf = pd.read_csv(\"sales_data_2025.csv\")\n\n# Clean\ndf = df.dropna(subset=[\"amount\", \"date\"])\ndf[\"date\"] = pd.to_datetime(df[\"date\"])\ndf[\"amount\"] = pd.to_numeric(df[\"amount\"], errors=\"coerce\")\n\n# Analyze\nmonthly = df.groupby(df[\"date\"].dt.to_period(\"M\"))[\"amount\"].sum()\ntop_products = df.groupby(\"product\")[\"amount\"].sum().nlargest(10)\n\n# Visualize\nfig, (ax1, ax2) = plt.subplots(1, 2, figsize=(16, 6))\nmonthly.plot(kind=\"bar\", ax=ax1, title=\"Monthly Revenue\")\ntop_products.plot(kind=\"barh\", ax=ax2, title=\"Top 10 Products\")\nplt.tight_layout()\nplt.savefig(\"charts.png\", dpi=150)\n\n# Report\nwith open(\"report.md\", \"w\") as f:\n    f.write(f\"# Sales Report 2025\\n\\n\")\n    f.write(f\"Total revenue: ${df[\"amount\"].sum():,.2f}\\n\\n\")\n    f.write(f\"Transactions: {len(df):,}\\n\\n\")\n    f.write(f\"## Monthly Breakdown\\n\\n\")\n    f.write(monthly.to_markdown())\nprint(\"Done.\")\""
  }'

# 4. Download results
curl -s "$SANDBOX_URL/api/sandboxes/$SID/download" \
  -H "Authorization: Bearer $API_KEY" \
  -o results.tar.gz

> I uploaded sales_data_2025.csv to /workspace. It has columns:
> date, product, region, amount, customer_id, channel.
>
> Analyze this dataset:
> 1. Clean: remove duplicates, handle missing values, fix date formats
> 2. Summary stats by region and by channel
> 3. Find the top 10 customers by total spend
> 4. Monthly revenue trend chart (line) with year-over-year comparison
> 5. Revenue by region (choropleth or bar chart)
> 6. Cohort analysis: when did each customer make their first purchase?
>    What % are still active 6 months later?
> 7. Save everything to /workspace/output/:
>    - report.md (narrative with embedded chart references)
>    - charts/ directory with PNGs
>    - cleaned_data.csv
>    - summary_stats.json

async def extract_dashboard_data(cdp_url, dashboard_url, credentials):
    """Log into a web dashboard and extract the visible data."""
    async with async_playwright() as p:
        browser = await p.chromium.connect_over_cdp(cdp_url)
        page = await browser.contexts[0].new_page()

        # Login
        await page.goto(credentials["login_url"], wait_until="networkidle")
        await page.fill(credentials["username_field"], credentials["username"])
        await page.fill(credentials["password_field"], credentials["password"])
        await page.click(credentials["submit_button"])
        await page.wait_for_load_state("networkidle")

        # Navigate to dashboard
        await page.goto(dashboard_url, wait_until="networkidle")
        await page.wait_for_timeout(3000)

        # Try DOM extraction first
        table_data = await page.evaluate("""() => {
            const tables = document.querySelectorAll('table');
            return Array.from(tables).map(table => {
                const rows = table.querySelectorAll('tr');
                return Array.from(rows).map(row => {
                    const cells = row.querySelectorAll('td, th');
                    return Array.from(cells).map(c => c.innerText.trim());
                });
            });
        }""")

        if table_data and any(len(t) > 1 for t in table_data):
            return {"method": "dom", "tables": table_data}

        # Fallback: vision extraction
        screenshot = await page.screenshot(full_page=True)
        b64 = base64.b64encode(screenshot).decode()

        response = claude.messages.create(
            model="claude-sonnet-4-6-20250514",
            max_tokens=4096,
            messages=[{
                "role": "user",
                "content": [
                    {"type": "image", "source": {
                        "type": "base64", "media_type": "image/png", "data": b64}},
                    {"type": "text", "text": "Extract all data visible in this dashboard screenshot as a JSON array of objects. Include every number, label, and metric you can read."},
                ],
            }],
        )
        return {"method": "vision", "data": json.loads(response.content[0].text)}

# Create a GPU sandbox for ML inference
curl -s -X POST "$SANDBOX_URL/api/sandboxes" \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "ml-inference",
    "instance_type": "ab0t.gpu",
    "auto_stop_minutes": 30
  }'

# Cost: $0.526/hour — use for 30 min of inference, then stop

> I uploaded customer_reviews.csv (10,000 rows, column: "review_text").
>
> Run sentiment analysis on every review:
> 1. Install transformers and torch
> 2. Load a sentiment model (distilbert-base-uncased-finetuned-sst-2-english)
> 3. Classify each review as positive/negative with a confidence score
> 4. Add sentiment and confidence columns to the CSV
> 5. Generate a sentiment distribution chart
> 6. Find the 20 most negative reviews for manual follow-up
> 7. Save: enriched_reviews.csv, sentiment_chart.png, negative_reviews.json

async def monthly_reconciliation():
    async with httpx.AsyncClient(timeout=300) as client:

        # Step 1: Extract Stripe dashboard data (browser container)
        stripe_data = await extract_from_dashboard(
            client, "https://dashboard.stripe.com/reports/revenue",
            {"login_url": "https://dashboard.stripe.com/login", "...": "..."})

        # Step 2: Download internal sales data (API call in terminal)
        internal_data = await execute_on_sandbox(sandbox_id,
            "curl -s https://api.internal.com/sales?month=2026-03 -H 'Authorization: Bearer $INTERNAL_KEY' > /workspace/internal_sales.json")

        # Step 3: Upload partner revenue spreadsheet
        await upload_file(sandbox_id, "partner_revenue_march.xlsx")

        # Step 4: Reconcile everything (terminal sandbox)
        await execute_on_sandbox(sandbox_id, """
            cd /workspace && pip install pandas openpyxl && python -c "
            import pandas as pd, json

            # Load all three sources
            stripe = pd.DataFrame(json.load(open('stripe_data.json')))
            internal = pd.DataFrame(json.load(open('internal_sales.json')))
            partner = pd.read_excel('partner_revenue_march.xlsx')

            # Reconcile: match by transaction ID, flag discrepancies
            merged = internal.merge(stripe, on='transaction_id', how='outer', suffixes=('_internal', '_stripe'))
            discrepancies = merged[merged['amount_internal'] != merged['amount_stripe']]

            # Generate report
            discrepancies.to_csv('discrepancies.csv', index=False)
            print(f'Total discrepancies: {len(discrepancies)}')
            print(f'Total delta: ${(merged[\"amount_internal\"].sum() - merged[\"amount_stripe\"].sum()):,.2f}')
            "
        """)

        # Step 5: Download results
        await download_workspace(sandbox_id, "reconciliation_march.tar.gz")

#!/bin/bash
set -euo pipefail

# Create sandbox, run pipeline, download, cleanup
SID=$(curl -s -X POST "$SANDBOX_URL/api/sandboxes" \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"name":"daily-data","instance_type":"ab0t.small","auto_stop_minutes":15}' \
  | jq -r '.sandbox_id')

sleep 45  # Wait for boot

curl -s -X POST "$SANDBOX_URL/api/sandboxes/$SID/execute" \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"command":"cd /workspace && git clone https://github.com/your-org/data-pipelines.git . && pip install -r requirements.txt && python daily_pipeline.py"}'

curl -s "$SANDBOX_URL/api/sandboxes/$SID/download" \
  -H "Authorization: Bearer $API_KEY" -o /tmp/daily-data.tar.gz

# Upload to S3 / email / Slack
aws s3 cp /tmp/daily-data.tar.gz s3://data-outputs/$(date +%Y-%m-%d).tar.gz

curl -s -X DELETE "$SANDBOX_URL/api/sandboxes/$SID" \
  -H "Authorization: Bearer $API_KEY"