Get a weekly OpenAI cost breakdown

from collections import defaultdict
from datetime import datetime, timedelta, timezone
from clickruntime import on_trigger, credentials, http, notify, state

API = "https://api.openai.com/v1/organization"

@on_trigger
def run(event):
    headers = {"Authorization": f"Bearer {credentials.secret('OPENAI_ADMIN_KEY')}"}
    end = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0)
    params = {"start_time": int((end - timedelta(days=7)).timestamp()), "end_time": int(end.timestamp()), "limit": 7}
    tokens, cost = defaultdict(int), defaultdict(float)
    usage = http.get(f"{API}/usage/completions", headers=headers, params={**params, "group_by": ["model"]}).json()
    for b in usage["data"]:
        for r in b["results"]:
            tokens[r["model"]] += r["input_tokens"] + r["output_tokens"]
    for b in http.get(f"{API}/costs", headers=headers, params={**params, "group_by": ["project_id"]}).json()["data"]:
        for r in b["results"]:
            cost[r.get("project_id") or "no project"] += r["amount"]["value"]
    last = state.get("last_week") or {"tokens": {}, "cost": {}}
    lines = ["Tokens by model:"] + [f"  {m}: {n:,} ({n - last['tokens'].get(m, 0):+,})" for m, n in tokens.items()]
    lines += ["Cost by project:"] + [f"  {p}: ${v:,.2f} ({v - last['cost'].get(p, 0):+,.2f})" for p, v in cost.items()]
    notify.email_me("OpenAI usage last week", "\n".join(lines))
    state.set("last_week", {"tokens": tokens, "cost": cost})
    return {"models": len(tokens), "projects": len(cost)}