Monitor usage

curl 'https://api.bigdata.com/v1/subscription/quotas' \
  --header 'X-API-KEY: YOUR_API_KEY'

#!/usr/bin/env python3
"""Print subscription quotas / credit usage from GET /v1/subscription/quotas."""

import json
import os
import sys
import urllib.error
import urllib.request

URL = "https://api.bigdata.com/v1/subscription/quotas"

# Omitted from output (legacy / redundant with other metrics).
SKIP_IDS = frozenset({"text:units", "pages:units"})


def unwrap_subscription_body(parsed):
    """Return (subscription_payload, envelope).

    The live API wraps the subscription in ``results`` alongside ``errors`` and
    ``metadata``. Older or direct responses may expose fields at the top level.
    """
    if isinstance(parsed, dict) and isinstance(parsed.get("results"), dict):
        return parsed["results"], parsed
    return parsed, parsed


def money_cents(value):
    if value is None:
        return "—"
    return f"${value / 100:,.2f}"


def fmt_remaining(value):
    if value is None:
        return "—"
    if isinstance(value, str) and value.lower() == "unlimited":
        return "unlimited"
    if isinstance(value, (int, float)):
        return money_cents(value)
    return str(value)


def fmt_quota_scalar(value):
    """Format a quota usage or limit (int, float, or 'unlimited')."""
    if value == "unlimited":
        return "unlimited"
    if isinstance(value, float) and not value.is_integer():
        return f"{value:g}"
    if isinstance(value, (int, float)):
        n = int(value) if isinstance(value, float) else value
        return f"{n:,}"
    return str(value)


def remaining_cell(used, limit):
    if limit == "unlimited":
        return "∞"
    if used is None or limit is None:
        return "—"
    try:
        rem = float(limit) - float(used)
    except (TypeError, ValueError):
        return "—"
    if abs(rem - round(rem)) < 1e-9:
        return f"{int(round(rem)):,}"
    return f"{rem:g}"


def row_from_unit(unit):
    used = unit.get("units_usage")
    limit = unit.get("units_limit")
    return (
        fmt_quota_scalar(used) if used is not None else "—",
        fmt_quota_scalar(limit),
        remaining_cell(used, limit),
    )


def row_from_credit_unit(unit):
    """Table row for per-metric credit usage (amounts in cents); Used column only."""
    used = unit.get("units_usage")
    return (money_cents(used) if used is not None else "—",)


def print_table(rows, headers=("Metric", "Used", "Limit", "Remaining")):
    """Print a simple ASCII table (metric left-aligned; other columns right-aligned)."""
    if not rows:
        return
    n = len(headers)
    cols = [[headers[i]] + [str(r[i]) for r in rows] for i in range(n)]
    widths = [max(len(s) for s in col) for col in cols]
    sep = "+" + "".join(f"-{w * '-'}-+" for w in widths)

    def fmt_row(cells):
        parts = [f"| {cells[0]:<{widths[0]}}"]
        for i in range(1, n):
            parts.append(f" | {cells[i]:>{widths[i]}}")
        parts.append(" |")
        return "".join(parts)

    print(sep)
    print(fmt_row(headers))
    print(sep)
    for r in rows:
        print(fmt_row(r))
    print(sep)


def label_for(unit_id, unit, fallback):
    if unit is None:
        return fallback
    return unit.get("display_name") or fallback


def print_units_grouped(quota_units, row_fn, *, credit_tables=False):
    """Print usage tables grouped by product area.

    ``row_fn(unit)`` returns a tuple: (used, limit, remaining) for quota, or (used) only
    for credit tables.
    """
    by_id = {u["id"]: u for u in quota_units if u.get("id")}
    table_headers = (
        ("Metric", "Used") if credit_tables else ("Metric", "Used", "Limit", "Remaining")
    )

    def append_row(metric, unit):
        vals = row_fn(unit)
        return (metric,) + tuple(vals)

    # --- Search Service ---
    search_spec = [
        ("search:smart_unit_read", "Smart Search"),
        ("search:unit_read", "Fast Search"),
        ("search:batch_unit_read", "Batch Search"),
    ]
    search_rows = []
    for uid, short in search_spec:
        u = by_id.get(uid)
        if u is None:
            continue
        search_rows.append(append_row(short, u))

    if search_rows:
        print("Search Service (API Query Units):")
        print_table(search_rows, headers=table_headers)
        print()

    # --- Research Agent ---
    research_sections = [
        (
            "Base model",
            [
                "chat:input_tokens",
                "chat:output_tokens",
                "chat:cached_input_tokens",
            ],
        ),
        (
            "Pro model",
            [
                "chat:input_tokens_pro_model",
                "chat:output_tokens_pro_model",
                "chat:cached_input_tokens_pro_model",
            ],
        ),
        (
            "Web Searches",
            ["chat:external_search"],
        ),
    ]

    any_research = False
    for subsection, ids in research_sections:
        rows = []
        for uid in ids:
            u = by_id.get(uid)
            if u is None:
                continue
            name = label_for(uid, u, uid)
            rows.append(append_row(name, u))
        if rows:
            if not any_research:
                print("Research Agent:")
                any_research = True
            print(f"  {subsection}")
            print_table(rows, headers=table_headers)
            print()

    # --- Content Enrichment & Indexing ---
    content_spec = [
        ("private_content:emails", "Emails uploaded"),
    ]
    content_rows = []
    for uid, short in content_spec:
        u = by_id.get(uid)
        if u is None:
            continue
        content_rows.append(append_row(short, u))

    if content_rows:
        print("Content Enrichment & Indexing:")
        print_table(content_rows, headers=table_headers)
        print()

    # --- Anything else (except skipped) ---
    known = {uid for uid, _ in search_spec}
    known.update(uid for _, ids in research_sections for uid in ids)
    known.update(uid for uid, _ in content_spec)
    known.update(SKIP_IDS)

    other_rows = []
    for u in quota_units:
        uid = u.get("id")
        if not uid or uid in known or uid in SKIP_IDS:
            continue
        name = u.get("display_name") or uid
        other_rows.append(append_row(name, u))

    if other_rows:
        print("Other:")
        print_table(other_rows, headers=table_headers)
        print()


def print_quota_grouped(quota_units):
    print_units_grouped(quota_units, row_from_unit)


def main():
    api_key = os.environ.get("BIGDATA_API_KEY")
    if not api_key:
        print("Set BIGDATA_API_KEY to your API key.", file=sys.stderr)
        sys.exit(1)

    req = urllib.request.Request(
        URL,
        headers={"X-API-KEY": api_key, "Accept": "application/json"},
    )
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            raw = resp.read().decode()
    except urllib.error.HTTPError as e:
        print(f"HTTP {e.code}: {e.read().decode()}", file=sys.stderr)
        sys.exit(1)

    parsed = json.loads(raw)
    data, envelope = unwrap_subscription_body(parsed)

    errors = envelope.get("errors") if isinstance(envelope, dict) else None
    if errors:
        print("API errors:", file=sys.stderr)
        print(json.dumps(errors, indent=2), file=sys.stderr)

    sub_type = data.get("subscription_type", "unknown")

    print("=" * 72)
    print(" Bigdata — subscription usage")
    print("=" * 72)
    print(f"Organization:      {data.get('org_id', '—')}")
    print(f"Subscription type: {sub_type}")
    print()

    if sub_type == "credits":
        c = data.get("credits") or {}
        print("Credits (organization total)")
        print(f"  Limit:       {money_cents(c.get('limit'))}")
        print(f"  Used:        {money_cents(c.get('usage'))}")
        print(f"  Remaining:   {fmt_remaining(c.get('remaining'))}")
        print()
        for i, bill in enumerate(data.get("billing") or [], 1):
            period = bill.get("period") or {}
            start, end = period.get("start"), period.get("end")
            print(f"Billing period {i}: {start} → {end}")
            print(f"  Credit usage (this period): {money_cents(bill.get('credit_usage'))}")
            print()
            bill_units = [
                u
                for u in (bill.get("units") or [])
                if u.get("id") not in SKIP_IDS
            ]
            print_units_grouped(bill_units, row_from_credit_unit, credit_tables=True)

    elif sub_type == "quota":
        units = [
            u
            for u in (data.get("quota_units") or [])
            if u.get("id") not in SKIP_IDS
        ]
        print_quota_grouped(units)

    else:
        print("Raw response:")
        print(json.dumps(parsed, indent=2))

    print("=" * 72)


if __name__ == "__main__":
    main()

"usage": {
  "api_query_units": 0.1
}