Files
treg/scripts/catalog_fx_update.py
Jason ZhouandClaude Opus 5 a6d756759a feat(catalog): curation tooling — ingest, validate, verify, fx
- catalog_ingest.py: bulk-generate the extended tier from a provider's own
  spec. Re-runnable (same input, byte-identical output), core wins on
  collision, and verification stamps survive regeneration
- catalog_validate.py: schema + referential integrity, two rule sets for the
  two tiers, exactly-one-state, and a credential leak scan
- catalog_verify.py / catalog_verify_extended.py: live-test endpoints and
  capture examples. --via-treg routes through the /call proxy so an OAuth
  provider verifies with the token treg already holds; the bulk runner adds
  budget caps, --out/--merge for parallel batches, and balance readback
- catalog_fx_update.py: refresh the FX table so costs display in one currency

Verification lessons are documented, not just implemented: a uniform block of
identical failures is our tooling until proven otherwise; after fixing such a
bug, re-run every batch that ran before the fix; retries are free only under
per-success billing; a spend cap that reads a missing price as zero cannot
fire.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-07-28 22:51:23 +10:00

45 lines
1.4 KiB
Python

#!/usr/bin/env python3
"""Refresh src/treg/catalog/fx.yaml from open.er-api.com (free, keyless).
Run on deploy or cron: `uv run python scripts/catalog_fx_update.py`. Only currencies already
listed in fx.yaml are refreshed — adding a new billing currency = add its key once, rerun.
"""
from __future__ import annotations
import json
import urllib.request
from datetime import date
from pathlib import Path
import yaml
FX = Path(__file__).resolve().parent.parent / "src" / "treg" / "catalog" / "fx.yaml"
def main() -> int:
doc = yaml.safe_load(FX.read_text())
live = json.load(urllib.request.urlopen("https://open.er-api.com/v6/latest/USD", timeout=30))
if live.get("result") != "success":
print(f"rate API answered {live.get('result')!r} — fx.yaml left unchanged")
return 1
usd_to = live["rates"]
for cur in doc["rates_to_usd"]:
if cur == "USD" or cur not in usd_to:
continue
doc["rates_to_usd"][cur] = round(1 / usd_to[cur], 4)
doc["updated"] = date.today().isoformat()
FX.write_text(
FX.read_text().split("rates_to_usd:")[0] # keep the header comment block
+ "rates_to_usd:\n"
+ "".join(f" {c}: {v}\n" for c, v in doc["rates_to_usd"].items())
+ f"updated: {doc['updated']}\n"
+ f"source: {doc['source']}\n"
)
print(f"fx.yaml refreshed: {doc['rates_to_usd']} ({doc['updated']})")
return 0
if __name__ == "__main__":
raise SystemExit(main())