erp/.claude/skills/dolibarr-sandbox-write/replay-packs/2026-07-11-books-regularization/verify-provenance.py

#!/usr/bin/env python3
"""Provenance check — every critical value in the write pack must be
mechanically corroborated by (a) the source PDF text and/or (b) a FRESH
bank-feed pull. No value is trusted because an LLM typed it.
Exit 0 = all green; any FAIL = exit 1."""
import json, re, subprocess, sys, os

SCRATCH = os.path.dirname(os.path.abspath(__file__))
PDF = os.path.join(SCRATCH, "pdfs")
ERP = os.path.expanduser("~/Work/Arcodange/erp")
BC = os.path.join(ERP, ".claude/skills/arcodange-bank-reco/scripts/bank-curl.sh")

results = []
def check(name, ok, detail=""):
    results.append((name, ok, detail))

def pdftext(fname):
    p = subprocess.run(["pdftotext", "-layout", os.path.join(PDF, fname), "-"],
                       capture_output=True, text=True)
    return p.stdout

def norm(t):  # 1 234,56 / 1,234.56 / 219.50 -> comparable tokens
    return t.replace(" ", " ").replace(",", ".")

def has_amount(text, amount):
    # match 219.50 or 219,50, tolerant to thousand separators/spacing
    a = f"{amount:.2f}"
    pat = a.replace(".", r"[.,]")
    return re.search(pat, norm(text)) is not None

# ---------- (a) PDF-anchored facts ----------
DOCS = {
  "F1045_ARCODANGE_2026-06-30.pdf": {
      "ref": "F1045", "date_str": "30/06/2026",
      "amounts": [214.70, 42.94, 257.64],
      "line_hint": "Apport d'affaire"},
  "F1046_ARCODANGE_2026-06-29.pdf": {
      "ref": "F1046", "date_str": "29/06/2026",
      "amounts": [219.50, 43.90, 263.40],
      "line_hint": "Apport d'affaire"},
  "Invoice-9BF0758D-695749.pdf": {
      "ref": "9BF0758D-695749", "date_str": "April 12, 2026",
      "amounts": [180.00],
      "line_hint": "Claude Pro"},
  "invoice-MSTRL-API-814045-001.pdf": {
      "ref": "MSTRL-API-814045-001", "date_str": "2 avr. 2026",
      "amounts": [143.90, 28.78, 172.68],
      "line_hint": "Le Chat Pro"},
}
for fname, exp in DOCS.items():
    t = pdftext(fname)
    if not t.strip():
        check(f"{fname}: text layer", False, "empty pdftotext"); continue
    check(f"{fname}: ref {exp['ref']}", exp["ref"] in t)
    check(f"{fname}: date {exp['date_str']}", exp["date_str"] in t)
    for a in exp["amounts"]:
        check(f"{fname}: amount {a:.2f}", has_amount(t, a))
    check(f"{fname}: line '{exp['line_hint']}'", exp["line_hint"] in t)

# ---------- (b) fresh bank-feed corroboration ----------
def bank(pathsvc, path):
    p = subprocess.run([BC, pathsvc, path], capture_output=True, text=True)
    return json.loads(p.stdout) if p.returncode == 0 and p.stdout.strip() else {}

org = bank("qonto", "/v2/organization")
acct = next(a["id"] for a in org["organization"]["bank_accounts"] if a["status"] == "active")
qtx = []
for window in [("2026-04-01", "2026-04-15"), ("2026-06-28", "2026-06-30")]:
    d = bank("qonto", f"/v2/transactions?bank_account_id={acct}&settled_at_from={window[0]}T00:00:00Z&settled_at_to={window[1]}T23:59:59Z&per_page=100&current_page=1")
    qtx += d.get("transactions", [])
qonto_feed = {(t["transaction_id"].split("transaction-")[-1]): (float(t["amount"]), t["side"]) for t in qtx}

envf = os.path.join(ERP, ".claude/skills/dolibarr/.env")
wpid = ""
for line in open(envf):
    if line.startswith("WISE_PROFILE_ID"):
        wpid = line.split("=", 1)[1].strip().strip('"')
wact = bank("wise", f"/v1/profiles/{wpid}/activities?since=2026-05-25T00:00:00.000Z&until=2026-06-30T23:59:59.000Z")
wise_feed = {}
for a in wact.get("activities", []):
    if a.get("type") == "TRANSFER":
        m = re.search(r"([\d,]+(?:\.\d{1,2})?)", a.get("primaryAmount", ""))
        if not m:
            continue
        amt = float(m.group(1).replace(",", ""))  # "2,195.97"->2195.97 ; "2,147"->2147.0 (comma = thousands sep)
        wise_feed[str((a.get("resource") or {}).get("id", ""))] = amt

manA = json.load(open(os.path.join(SCRATCH, "manifest-A-km-payments.json")))
manB = json.load(open(os.path.join(SCRATCH, "manifest-B-suppliers.json")))
EXPECT_A = {"2159468139": 2147.00, "2210434850": 2195.97}  # full wire amounts (invoice pays remaining 2145.92; delta = FX, booked separately)
for op in manA:
    tx = op["input"]["transaction_id"]
    check(f"manifest A: Wise tx {tx} exists in fresh feed", tx in wise_feed,
          f"feed={wise_feed.get(tx)}")
    if tx in wise_feed:
        check(f"manifest A: Wise tx {tx} amount == feed", abs(wise_feed[tx] - EXPECT_A[tx]) < 0.005,
              f"feed {wise_feed[tx]} vs expected wire {EXPECT_A[tx]}")
for op in manB:
    if op["op"] != "payment":
        # invoice op: amounts corroborated by the F1045 PDF above
        li = op["input"]["lines"][0]
        check("manifest B: F1045 line HT in PDF", has_amount(pdftext("F1045_ARCODANGE_2026-06-30.pdf"), float(li["price_ht"])))
        check("manifest B: F1045 ref_supplier matches PDF ref", op["input"]["ref_supplier"] == "F1045")
        continue
    tx = op["input"]["transaction_id"]; amt = float(op["input"]["amount"])
    check(f"manifest B: Qonto tx …{tx[-12:]} exists in fresh feed", tx in qonto_feed, f"known={list(qonto_feed)[:2]}…")
    if tx in qonto_feed:
        famt, side = qonto_feed[tx]
        check(f"manifest B: tx …{tx[-12:]} amount {amt:.2f} == feed", abs(famt - amt) < 0.005 and side == "debit",
              f"feed {famt} {side}")

# ---------- report ----------
w = max(len(n) for n, _, _ in results)
fails = 0
for n, ok, det in results:
    print(f"{'PASS' if ok else 'FAIL':4}  {n:<{w}}  {det if not ok else ''}")
    fails += 0 if ok else 1
print(f"\n{len(results)-fails}/{len(results)} checks green")
sys.exit(1 if fails else 0)