tegwick
ea568558a8
Some checks failed
Work Records / validate (push) Has been cancelled
qonto-assistant (sibling repo) already shipped Phase 1 (QONTO-WP-0002) and Phase 2 MCP surface (QONTO-WP-0003), both finished, 31 tests passing, ahead of this workplan's original sequencing. Reconciled T02-T04 against what actually landed there instead of duplicating it. The one piece qonto-assistant explicitly left for this repo -- the CostRunRate consumer cutover -- is implemented here: scripts/qonto-costrunrate-refresh.py calls GET /v1/snapshot and writes dated, redacted evidence under finance/ for a human to reconcile against CostRunRate.md. Verified end-to-end against a local fixture-backed qonto-assistant instance (no real credentials); a real pull against the live account awaits qonto-assistant getting an actual deployment, called out as a known gap rather than faked. Workplan closed (status: finished); remaining gaps (OIDC federation, harness tool profile registration, Phase 3/4) are carried forward in qonto-assistant's own closure notes, not duplicated here. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
99 lines
3.8 KiB
Python
Executable file
99 lines
3.8 KiB
Python
Executable file
#!/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())
|