binky-control/scripts/qonto-costrunrate-refresh.py

#!/usr/bin/env python3
"""Pull a governed Qonto snapshot for the CostRunRate.md refresh cycle.

Calls GET /v1/snapshot on a running qonto-assistant instance (BINKY-WP-0007 /
QONTO-WP-0002) and writes the redacted response as dated evidence under
finance/. Does not edit CostRunRate.md itself — a human (or a follow-on
task) reviews the evidence and updates the TBC rows/prose, same as the
2026-07-21 first pull.
"""
from __future__ import annotations

import argparse
import json
import sys
import urllib.error
import urllib.request
from datetime import UTC, datetime
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parents[1]


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--base-url",
        default="http://127.0.0.1:8080",
        help="qonto-assistant base URL (default: %(default)s)",
    )
    parser.add_argument("--window-days", type=int, default=90)
    parser.add_argument("--page-size", type=int, default=50)
    parser.add_argument("--actor-id", default="finance-steward")
    parser.add_argument("--tenant-id", default="binky")
    parser.add_argument(
        "--token",
        default=None,
        help="Bearer token if the assistant enforces QONTO_ASSISTANT_MCP_TOKEN-style auth on REST too "
        "(REST is unauthenticated by default in the current build; pass only if that changes).",
    )
    parser.add_argument(
        "--out-dir",
        default=str(REPO_ROOT / "finance"),
        help="Directory to write the dated evidence file into (default: %(default)s)",
    )
    return parser.parse_args()


def fetch_snapshot(args: argparse.Namespace) -> dict:
    url = f"{args.base_url}/v1/snapshot?window_days={args.window_days}&page_size={args.page_size}"
    headers = {"X-Actor-ID": args.actor_id, "X-Tenant-ID": args.tenant_id}
    if args.token:
        headers["Authorization"] = f"Bearer {args.token}"
    request = urllib.request.Request(url, headers=headers)
    try:
        with urllib.request.urlopen(request, timeout=15) as response:  # noqa: S310
            return json.load(response)
    except urllib.error.HTTPError as exc:
        body = exc.read().decode("utf-8", errors="replace")
        raise SystemExit(f"qonto-assistant returned {exc.code}: {body}") from exc
    except urllib.error.URLError as exc:
        raise SystemExit(
            f"could not reach qonto-assistant at {args.base_url} ({exc.reason}). "
            "Is the service running? See docs/operator-runbook.md in qonto-assistant."
        ) from exc


def main() -> int:
    args = parse_args()
    snapshot = fetch_snapshot(args)

    pulled_at = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
    evidence = {
        "pulled_at": pulled_at,
        "source": "qonto-assistant GET /v1/snapshot (governed, read-only, BINKY-WP-0007/QONTO-WP-0002)",
        "base_url": args.base_url,
        "window_days": args.window_days,
        **snapshot,
    }

    out_dir = Path(args.out_dir)
    out_dir.mkdir(parents=True, exist_ok=True)
    date_stamp = datetime.now(UTC).strftime("%Y-%m-%d")
    out_path = out_dir / f"qonto-snapshot-{date_stamp}.json"
    out_path.write_text(json.dumps(evidence, indent=2, sort_keys=False) + "\n", encoding="utf-8")

    summary = snapshot.get("summary", {})
    print(f"Wrote {out_path.relative_to(REPO_ROOT)}")
    print(
        f"Total balance: {summary.get('total_balance')} EUR "
        f"(authorized: {summary.get('authorized_balance')} EUR, window: {summary.get('window_days')}d)"
    )
    for hint in snapshot.get("cost_run_rate_hints", {}).get("recurring_debits", []) or []:
        print(f"  recurring debit: {hint}")
    print("Review the evidence file and update finance/CostRunRate.md by hand if figures moved.")
    return 0


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