Uber 的 AI 编码工具月度限额：1500 美元背后的定价逻辑与团队预算实践

# ai_budget_policy.yaml
# 每工具独立额度配置
tools:
  cursor:
    monthly_cap_usd: 1500
    warning_threshold_pct: 80  # 80% 时发警告
    billing_source: cursor_api
  claude_code:
    monthly_cap_usd: 1500
    warning_threshold_pct: 80
    billing_source: anthropic_api
  copilot:
    monthly_cap_usd: 100     # 补全类工具预算较低
    warning_threshold_pct: 90
    billing_source: github_api

team:
  notify_channel: slack      # 警告通知渠道
  notify_webhook: "https://hooks.slack.com/services/YOUR/WEBHOOK/HERE"
  report_day: 25             # 每月25日生成月报

#!/usr/bin/env python3
"""
ai_budget_monitor.py
聚合各 AI 编码工具的月度 API 消耗，按工具独立计算额度，
接近阈值时发送 Slack 警告。

运行前：
  pip install requests pyyaml
  修改 ai_budget_policy.yaml 中的 webhook 地址
  填入各平台 API key（环境变量）

使用：
  python ai_budget_monitor.py --policy ai_budget_policy.yaml --user engineer_id
"""

import argparse
import json
import os
import sys
from datetime import datetime
from pathlib import Path

import requests
import yaml


def load_policy(path: str) -> dict:
    with open(path) as f:
        return yaml.safe_load(f)


def fetch_anthropi_cost(month: str, api_key: str) -> float:
    """从 Anthropic API 获取当月 token 消耗并换算为美元。
    实际部署中应调用 Anthropic 的 admin/billing endpoint。
    这里用模拟数据演示逻辑。"""
    # 真实场景：调用 https://api.anthropic.com/v1/organizations/billing
    # 此处为演示，返回模拟值
    mock_input_tokens = 2_000_000   # 当月输入 token 数
    mock_output_tokens = 500_000    # 当月输出 token 数
    cost = (mock_input_tokens / 1_000_000 * 3.0) + \
           (mock_output_tokens / 1_000_000 * 15.0)
    return cost


def fetch_cursor_cost(month: str, api_key: str) -> float:
    """Cursor 目前无公开 billing API，实际需从内部仪表盘导出。
    此处返回模拟值。"""
    return 450.0  # 模拟：含订阅费 + API 调用超额


def fetch_github_copilot_cost(month: str, api_key: str) -> float:
    """Copilot 企业版按席位计费，无超额概念。"""
    return 19.0  # 模拟月费


FETCHERS = {
    "anthropic_api": fetch_anthropi_cost,
    "cursor_api": fetch_cursor_cost,
    "github_api": fetch_github_copilot_cost,
}


def check_budget(tool_name: str, spent: float, cap: float,
                  threshold_pct: float) -> dict:
    pct = spent / cap * 100
    status = "ok"
    if pct >= 100:
        status = "exceeded"
    elif pct >= threshold_pct:
        status = "warning"
    return {
        "tool": tool_name,
        "spent_usd": round(spent, 2),
        "cap_usd": cap,
        "usage_pct": round(pct, 1),
        "status": status,
    }


def send_slack_alert(webhook: str, results: list, user: str):
    blocks = []
    for r in results:
        if r["status"] in ("warning", "exceeded"):
            emoji = "⚠️" if r["status"] == "warning" else "🚨"
            blocks.append({
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": (
                        f"{emoji} *{r['tool']}* — "
                        f"已消耗 ${r['spent_usd']} / ${r['cap_usd']} "
                        f"({r['usage_pct']}%)"
                    ),
                },
            })
    if not blocks:
        print("所有工具额度正常，无需告警。")
        return
    payload = {
        "text": f"AI 工具预算告警 — {user}",
        "blocks": blocks,
    }
    resp = requests.post(webhook, json=payload, timeout=10)
    if resp.status_code != 200:
        print(f"Slack 通知失败: {resp.status_code} {resp.text}")
    else:
        print("告警已发送至 Slack。")


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--policy", required=True)
    parser.add_argument("--user", default="default_engineer")
    args = parser.parse_args()

    policy = load_policy(args.policy)
    month = datetime.now().strftime("%Y-%m")
    results = []

    for tool_name, cfg in policy["tools"].items():
        source = cfg["billing_source"]
        cap = cfg["monthly_cap_usd"]
        threshold = cfg["warning_threshold_pct"]
        fetcher = FETCHERS.get(source)
        if not fetcher:
            print(f"未知 billing_source: {source}, 跳过 {tool_name}")
            continue
        api_key = os.environ.get(f"{source.upper()}_KEY", "")
        spent = fetcher(month, api_key)
        result = check_budget(tool_name, spent, cap, threshold)
        results.append(result)
        print(f"  {tool_name}: ${result['spent_usd']} / ${cap} "
              f"({result['usage_pct']}%) — {result['status']}")

    # 发送告警
    webhook = policy["team"]["notify_webhook"]
    if webhook and "YOUR" not in webhook:
        send_slack_alert(webhook, results, args.user)

    # 输出 JSON 报告，可接入后续分析
    report = {"month": month, "user": args.user, "tools": results}
    report_path = Path(f"budget_report_{month}_{args.user}.json")
    with open(report_path, "w") as f:
        json.dump(report, f, indent=2)
    print(f"\n月度报告已保存: {report_path}")


if __name__ == "__main__":
    main()

# 1. 安装依赖
pip install requests pyyaml

# 2. 配置自己的 Slack webhook（可选，不配则只输出本地报告）

# 3. 运行监控
python ai_budget_monitor.py \
  --policy ai_budget_policy.yaml \
  --user zhangsan

  claude_code: $6.0 / $1500 (0.4%) — ok
  cursor: $450.0 / $1500 (30.0%) — ok
  copilot: $19.0 / $100 (19.0%) — ok
所有工具额度正常，无需告警。

月度报告已保存: budget_report_2025-06_zhangsan.json