OpenAI / Gemini / 自社 LLM：複数プロバイダの usage を統合する

# llm_usage 統合テーブルcolumns:  - timestamp        # UTC  - provider         # "anthropic" | "openai" | "gemini" | "azure_oai" | "bedrock" | "self_hosted"  - model            # "claude-opus-4-7" | "gpt-5" | "gemini-2-flash" 等  - api_key_id       # 個人/用途識別子（プロバイダ非依存）  - person           # マッピング後の個人 ID  - team             # チーム  - input_tokens  - output_tokens  - cache_read_tokens   # 0 if not supported  - cache_creation_tokens  - cost_usd         # プロバイダの単価で計算  - request_count  - latency_p50_ms   # 任意  - latency_p99_ms   # 任意

import os, requests
OPENAI_KEY = os.environ["OPENAI_API_KEY"]
def fetch_openai_usage(date_str: str) -> dict:    """日次の usage（admin key required）"""    r = requests.get(        "https://api.openai.com/v1/organization/usage/completions",        headers={"Authorization": f"Bearer {OPENAI_KEY}"},        params={"start_time": int(date_str_to_epoch(date_str)),                "bucket_width": "1d", "group_by": ["api_key_id", "model"]},    )    return r.json()

from dataclasses import dataclassfrom datetime import datetime
@dataclassclass UsageRow:    timestamp: datetime    provider: str    model: str    api_key_id: str    input_tokens: int    output_tokens: int    cache_read_tokens: int = 0    cache_creation_tokens: int = 0    request_count: int = 0
def normalize_anthropic(raw: dict) -> list[UsageRow]:    """Anthropic レスポンス → UsageRow"""    rows = []    for bucket in raw["data"]:        ts = datetime.fromisoformat(bucket["starting_at"].replace("Z", "+00:00"))        for r in bucket["results"]:            rows.append(UsageRow(                timestamp=ts, provider="anthropic", model=r["model"],                api_key_id=r["api_key_id"],                input_tokens=r["uncached_input_tokens"],                output_tokens=r["output_tokens"],                cache_read_tokens=r.get("cache_read_input_tokens", 0),                cache_creation_tokens=r.get("cache_creation_input_tokens", 0),            ))    return rows
def normalize_openai(raw: dict) -> list[UsageRow]:    # ... OpenAI 用の adapter    pass
# 全プロバイダから取り込みall_rows = []all_rows.extend(normalize_anthropic(get_anthropic_usage(...)))all_rows.extend(normalize_openai(fetch_openai_usage(...)))# ... 他
# BigQuery / Snowflake へ insertload_to_dwh(all_rows)