从"尽情用"到月限 1500 美元：Uber 的 AI 预算刹车说明了什么

# ai_cost_tracker.py
# 运行前：pip install anthropic openai
# 需要设置环境变量 ANTHROPIC_API_KEY 和 OPENAI_API_KEY

import sqlite3
import json
from datetime import datetime, date
from anthropic import Anthropic
from openai import OpenAI

DB_PATH = "ai_usage.db"

def init_db():
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS daily_usage (
                date TEXT,
                provider TEXT,
                user_email TEXT,
                input_tokens INTEGER,
                output_tokens INTEGER,
                cost_usd REAL
            )
        """)

def fetch_anthropi_usage(user_email: str):
    """拉取 Anthropic 当日用量（示例：从本地日志文件解析）"""
    # Anthropic API 目前不提供组织级用量查询端点
    # 实际部署中应从 Claude Code 的本地日志 ~/.claude/logs/ 解析
    # 这里用模拟数据演示结构
    today = date.today().isoformat()
    # 真实场景：解析日志文件提取 token 数
    # with open(f"~/.claude/logs/{today}.jsonl") as f: ...
    return {
        "date": today,
        "provider": "anthropic",
        "user_email": user_email,
        "input_tokens": 0,   # 从日志填充
        "output_tokens": 0,  # 从日志填充
        "cost_usd": 0.0      # 按 $3/M input, $15/M output (Claude 3.5 Sonnet) 计算
    }

def fetch_openai_usage(user_email: str):
    """拉取 OpenAI 当月用量"""
    client = OpenAI()
    # OpenAI 提供 /v1/organization/costs 端点（需要 Admin API Key）
    # 此处用模拟结构演示
    today = date.today().isoformat()
    return {
        "date": today,
        "provider": "openai",
        "user_email": user_email,
        "input_tokens": 0,
        "output_tokens": 0,
        "cost_usd": 0.0
    }

def record_usage(usage: dict):
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute(
            "INSERT INTO daily_usage VALUES (:date, :provider, :user_email, "
            ":input_tokens, :output_tokens, :cost_usd)",
            usage
        )

def monthly_report(user_email: str, month: str = None):
    """生成月度报表，按 provider 分组"""
    if month is None:
        month = datetime.now().strftime("%Y-%m")
    with sqlite3.connect(DB_PATH) as conn:
        rows = conn.execute("""
            SELECT provider, SUM(input_tokens), SUM(output_tokens), SUM(cost_usd)
            FROM daily_usage
            WHERE user_email = ? AND date LIKE ?
            GROUP BY provider
        """, (user_email, f"{month}%")).fetchall()

    print(f"\n=== {user_email} · {month} 月度 AI 用量 ===")
    for provider, in_tok, out_tok, cost in rows:
        print(f"  {provider}: 入 {in_tok:,} tok / 出 {out_tok:,} tok / ${cost:.2f}")
    total = sum(r[3] for r in rows)
    print(f"  合计: $${total:.2f}")

    # 检查是否接近上限
    LIMIT_PER_TOOL = 1500
    for provider, _, _, cost in rows:
        if cost > LIMIT_PER_TOOL * 0.8:
            print(f"  ⚠️  {provider} 已用 {cost/LIMIT_PER_TOOL*100:.0f}%，接近上限！")

if __name__ == "__main__":
    init_db()
    # 示例：记录一条模拟数据
    sample = {
        "date": date.today().isoformat(),
        "provider": "anthropic",
        "user_email": "dev@example.com",
        "input_tokens": 120000,
        "output_tokens": 35000,
        "cost_usd": 120000 * 3 / 1e6 + 35000 * 15 / 1e6  # Sonnet 定价
    }
    record_usage(sample)
    monthly_report("dev@example.com")

export ANTHROPIC_API_KEY="sk-ant-..."
export OPENAI_API_KEY="sk-..."
python ai_cost_tracker.py

=== dev@example.com · 2025-06 月度 AI 用量 ===
  anthropic: 入 120,000 tok / 出 35,000 tok / $0.83
  合计: $0.83

# 加入 crontab：每天 18:00 执行
0 18 * * * cd /path/to/project && python ai_cost_tracker.py >> /var/log/ai_usage.log 2>&1