Track OpenAI API costs: from the Costs API to your desk

"""Read reported organization costs for a complete UTC date range.

Usage: python openai_costs.py 2026-10-01 2026-10-14
Provide OPENAI_ADMIN_KEY through your process environment. The end is exclusive.
Documentation-based example; validated offline, not with a live account.
"""

import argparse
import json
import os
import re
import sys
import time
from datetime import date, datetime, timezone
from decimal import Decimal
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import HTTPRedirectHandler, Request, build_opener

ENDPOINT = "https://api.openai.com/v1/organization/costs"
DAY = 86400


class CostsError(Exception):
    pass


class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def utc_boundary(value):
    if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", value):
        raise CostsError("Use YYYY-MM-DD dates.")
    try:
        day = date.fromisoformat(value)
        return int(datetime(day.year, day.month, day.day, tzinfo=timezone.utc).timestamp())
    except ValueError:
        raise CostsError("Invalid calendar date.") from None


def api_reader(key):
    opener = build_opener(NoRedirect())

    def read_page(params, timeout):
        request = Request(ENDPOINT + "?" + urlencode(params), headers={
            "Authorization": "Bearer " + key, "Accept": "application/json"})
        try:
            with opener.open(request, timeout=timeout) as response:
                raw = response.read(2 * 1024 * 1024 + 1)
            if len(raw) > 2 * 1024 * 1024:
                raise CostsError("Response too large; no total returned.")
            return json.loads(raw, parse_float=Decimal)
        except HTTPError as error:
            raise CostsError(f"OpenAI returned HTTP {error.code}; no total returned.") from None
        except (URLError, TimeoutError, OSError):
            raise CostsError("Request failed or timed out; no total returned.") from None
        except (ValueError, UnicodeError):
            raise CostsError("Invalid JSON response; no total returned.") from None

    return read_page


def collect_costs(start_date, end_date, read_page):
    start, end = utc_boundary(start_date), utc_boundary(end_date)
    if not 0 < end - start <= 366 * DAY:
        raise CostsError("Choose an increasing date range of at most 366 days.")
    params = {"start_time": start, "end_time": end, "bucket_width": "1d", "limit": 31}
    totals, seen_days, seen_cursors = {}, set(), set()
    empty_days = 0
    deadline = time.monotonic() + 120
    for _ in range(100):
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            raise CostsError("Collection time limit reached; no total returned.")
        page = read_page(dict(params), min(15, remaining))
        if (not isinstance(page, dict) or page.get("object") != "page"
                or "error" in page or not isinstance(page.get("data"), list)
                or type(page.get("has_more")) is not bool or "next_page" not in page):
            raise CostsError("Incomplete page response; no total returned.")
        for bucket in page["data"]:
            if not isinstance(bucket, dict) or bucket.get("object") != "bucket":
                raise CostsError("Invalid cost bucket; no total returned.")
            a, b, rows = bucket.get("start_time"), bucket.get("end_time"), bucket.get("results")
            if (type(a) is not int or type(b) is not int or a % DAY or b != a + DAY
                    or a < start or b > end or a in seen_days or not isinstance(rows, list)):
                raise CostsError("Missing, repeated or unexpected day; no total returned.")
            seen_days.add(a)
            empty_days += not rows
            for row in rows:
                if not isinstance(row, dict) or row.get("object") != "organization.costs.result":
                    raise CostsError("Unexpected result type; no total returned.")
                amount = row.get("amount")
                if not isinstance(amount, dict):
                    raise CostsError("Missing amount; no total returned.")
                currency, value = amount.get("currency"), amount.get("value")
                if (not isinstance(currency, str) or not re.fullmatch(r"[a-z]{3}", currency)
                        or isinstance(value, bool) or not isinstance(value, (int, Decimal))):
                    raise CostsError("Missing or invalid currency/value; no total returned.")
                value = Decimal(value)
                if not value.is_finite():
                    raise CostsError("Non-finite amount; no total returned.")
                totals[currency] = totals.get(currency, Decimal(0)) + value
        cursor = page["next_page"]
        if not page["has_more"]:
            if cursor is not None:
                raise CostsError("Conflicting pagination state; no total returned.")
            break
        if not isinstance(cursor, str) or not cursor or cursor in seen_cursors:
            raise CostsError("Missing or repeated page cursor; no total returned.")
        seen_cursors.add(cursor)
        params["page"] = cursor
    else:
        raise CostsError("Page limit reached; no total returned.")
    if seen_days != set(range(start, end, DAY)) or not totals:
        raise CostsError("Incomplete period or no reported cost values; total unavailable.")
    return {
        "period_start_inclusive_utc": start_date + "T00:00:00Z",
        "period_end_exclusive_utc": end_date + "T00:00:00Z",
        "reported_totals": {currency: format(value, "f") for currency, value in sorted(totals.items())},
        "days_without_cost_rows": empty_days,
        "collected_at_utc": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
    }


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("start_date")
    parser.add_argument("end_date", help="Exclusive end date, UTC")
    args = parser.parse_args()
    key = os.environ.get("OPENAI_ADMIN_KEY", "").strip()
    if not key:
        print("OPENAI_ADMIN_KEY is missing; no request made.", file=sys.stderr)
        return 1
    try:
        report = collect_costs(args.start_date, args.end_date, api_reader(key))
    except CostsError as error:
        print(str(error), file=sys.stderr)
        return 1
    print(json.dumps(report, indent=2))
    return 0


if __name__ == "__main__":
    sys.exit(main())