4サイト分のClaude費用が一つの請求に混ざる — ワークスペース単位で支出を割り当て、絞るパイプラインを見極める

# 組織集計用（配分レポートを回すスクリプトが使う）
export ANTHROPIC_ADMIN_KEY="YOUR_ADMIN_API_KEY"
 
# 各サイトの実行キーとは別物。混ぜない
export ANTHROPIC_WORKSPACE_KEY="YOUR_WORKSPACE_API_KEY"

import os
import httpx
from decimal import Decimal
from datetime import datetime, timedelta, timezone
 
BASE = "https://api.anthropic.com"
HEADERS = {
    "x-api-key": os.environ["ANTHROPIC_ADMIN_KEY"],
    "anthropic-version": "2023-06-01",
}
 
def fetch_cost_report(days: int = 7) -> list[dict]:
    """Cost Report を group_by=workspace_id で days 日分、全ページ取得する。"""
    start = (datetime.now(timezone.utc) - timedelta(days=days)).date().isoformat()
    params = {
        "starting_at": f"{start}T00:00:00Z",
        "group_by[]": "workspace_id",
        "limit": 31,
    }
    buckets: list[dict] = []
    with httpx.Client(timeout=30.0) as client:
        page = None
        while True:
            q = dict(params)
            if page:
                q["page"] = page
            r = client.get(f"{BASE}/v1/organizations/cost_report",
                           headers=HEADERS, params=q)
            r.raise_for_status()
            body = r.json()
            buckets.extend(body.get("data", []))
            if not body.get("has_more"):
                break
            page = body.get("next_page")
    return buckets

from collections import defaultdict
 
WORKSPACE_TO_SITE = {
    "wrkspc_claudelab": "claudelab",
    "wrkspc_gemilab":   "gemilab",
    "wrkspc_antigrav":  "antigravitylab",
    "wrkspc_rorklab":   "rorklab",
}
 
def build_showback(buckets: list[dict]) -> dict[str, dict[str, Decimal]]:
    """{日付: {サイト: 金額}} を返す。未知の workspace は 'unknown' に集約。"""
    table: dict[str, dict[str, Decimal]] = defaultdict(lambda: defaultdict(Decimal))
    for b in buckets:
        day = (b.get("starting_at") or "")[:10]
        for row in b.get("results", []):
            ws = row.get("workspace_id") or "default"
            site = WORKSPACE_TO_SITE.get(ws, "unknown")
            amount = Decimal(str(row.get("amount", "0")))
            table[day][site] += amount
    return table
 
def print_showback(table: dict[str, dict[str, Decimal]]) -> None:
    sites = sorted({s for day in table.values() for s in day})
    header = "date".ljust(12) + "".join(s.rjust(14) for s in sites) + "total".rjust(14)
    print(header)
    grand = defaultdict(Decimal)
    for day in sorted(table):
        row = table[day]
        line = day.ljust(12)
        day_total = Decimal("0")
        for s in sites:
            v = row.get(s, Decimal("0"))
            day_total += v
            grand[s] += v
            line += f"${v:.3f}".rjust(14)
        line += f"${day_total:.3f}".rjust(14)
        print(line)
    total_all = sum(grand.values())
    foot = "TOTAL".ljust(12) + "".join(f"${grand[s]:.2f}".rjust(14) for s in sites)
    foot += f"${total_all:.2f}".rjust(14)
    print("-" * len(header)); print(foot)
    for s in sites:
        share = (grand[s] / total_all * 100) if total_all else Decimal("0")
        print(f"  {s:<16} {share:5.1f}%")

def cost_efficiency(grand: dict[str, Decimal],
                    clicks_7d: dict[str, int]) -> None:
    """サイト別の『クリック100あたりコスト』を出す。低いほど効率が良い。"""
    print(f"{'site':<16}{'cost/7d':>10}{'clicks/7d':>12}{'cost/100clk':>14}")
    for site in sorted(grand):
        cost = grand[site]
        clicks = clicks_7d.get(site, 0)
        per_100 = (cost / clicks * 100) if clicks else Decimal("0")
        print(f"{site:<16}${cost:>8.2f}{clicks:>12}${per_100:>12.3f}")