erp/fleet/atoms/invoice-extract/scripts/validators.py

#!/usr/bin/env python3
"""invoice-extract validators — deterministic post-conditions (atom.yaml invariants).

The LLM proposes, this module disposes: pre-screens hostile documents before any
model sees them, checks every invariant after extraction, and anchors every
critical value to a literal excerpt of the source text (anti-hallucination: a
value absent from its source can never appear in output). A failure quarantines
— refuse, never repair. Stdlib only.
"""
from __future__ import annotations

import re
from datetime import date, datetime

VALID_RATES = {0, 2.1, 5.5, 10, 20}
CRITICAL_AMOUNT_FIELDS = ("totals.ht", "totals.tva", "totals.ttc")

# --- deterministic pre-screens (run BEFORE any model call) -------------------

# Tight, tested against the full real corpus (zero hits) — see eval evidence.
INSTRUCTION_PATTERNS = [
    r"merci d.ignorer",
    r"ignore +(the|any|all|previous)",
    r"skip +validation",
    r"already +approved",
    r"record +(it +)?for +payment",
    r"remplace toute instruction",
    r"instruction pr[eé]c[eé]dente",
    r"<\s*/?\s*system\s*>",
    r"tool[_ ]?call",
    r"create_payment|dolibarr\.",
    r"(supprim\w+|delete)\w*\W.{0,60}(record|enregistrement|entry|pr[eé]c[eé]dent)",
    r"nouvelles coordonn[eé]es bancaires",
    r"(votre|your) +(syst[eè]me|system|op[eé]rateur|operator|assistant|ai\b)",
    r"corrige\w*\W.{0,40}(tva|vat)|correct\w*\W.{0,40}(vat|tva) +field",
    r"system +override",
    r"when +extracting",
    r"instead of the printed",
    r"do not mention",
    r"set +totals?\S* +to +\d",
    r"output +iban",
]

IBAN_RE = re.compile(r"\b([A-Z]{2}\d{2}(?:[   ]?[A-Z0-9]{2,4}){2,8})\b")


def iban_valid(iban: str) -> bool:
    s = re.sub(r"[   ]", "", iban).upper()
    if not re.fullmatch(r"[A-Z]{2}\d{2}[A-Z0-9]{10,30}", s):
        return False
    rearranged = s[4:] + s[:4]
    digits = "".join(str(int(c, 36)) for c in rearranged)
    return int(digits) % 97 == 1


def siren_valid(siren: str) -> bool:
    s = re.sub(r"\D", "", siren)
    if len(s) != 9:
        return False
    total = 0
    for i, ch in enumerate(s):  # Luhn, rightmost digit position 1
        d = int(ch)
        if (len(s) - i) % 2 == 0:
            d *= 2
            if d > 9:
                d -= 9
        total += d
    return total % 10 == 0


def screen_document(text: str) -> tuple[list[str], list[str]]:
    """Screen raw document text BEFORE any model call.

    Returns (quarantine_reasons, escalate_flags): instruction-shaped content is
    a hard quarantine; multiple valid IBANs alone force the Claude escalation
    leg (never silently resolved — e.g. a multi-document PDF); record-time
    corroboration is erp#41's linter stage.
    """
    reasons, flags = [], []
    low = text.lower()
    for pat in INSTRUCTION_PATTERNS:
        m = re.search(pat, low)
        if m:
            reasons.append(f"instruction-pattern: /{pat}/ matched {m.group(0)[:60]!r}")
    ibans = {re.sub(r"[ \u202f\xa0]", "", m.group(1)) for m in IBAN_RE.finditer(text)}
    ibans = {i for i in ibans if iban_valid(i)}
    if len(ibans) >= 2:
        flags.append(f"multi-iban: {len(ibans)} distinct valid IBANs — IBAN choice must be escalated")
    return reasons, flags


# --- number / date normalization --------------------------------------------

def _num_variants(value: float) -> list[str]:
    """Literal spellings a European invoice may use for this amount."""
    out = []
    for thousands, decimal in ((" ", ","), (" ", ","), (" ", ","), ("", ","), (",", "."), ("", ".")):
        s = f"{value:,.2f}"  # 2,000.00
        s = s.replace(",", "\0").replace(".", decimal).replace("\0", thousands)
        out.append(s)
        if value == int(value):
            out.append(s[:-3])  # bare integer form: 2 000
    return out


DATE_FMTS_OUT = ["%Y-%m-%d", "%d/%m/%Y", "%d-%m-%Y", "%d.%m.%Y", "%B %d, %Y", "%b %d, %Y", "%d %B %Y"]
FR_MONTHS = ["janvier", "février", "mars", "avril", "mai", "juin", "juillet",
             "août", "septembre", "octobre", "novembre", "décembre"]


def _date_variants(iso: str) -> list[str]:
    d = datetime.strptime(iso, "%Y-%m-%d").date()
    out = []
    for fmt in DATE_FMTS_OUT:
        try:
            out.append(d.strftime(fmt))
        except ValueError:
            pass
    out.append(f"{d.day} {FR_MONTHS[d.month - 1]} {d.year}")
    out.append(f"{d.day:02d} {FR_MONTHS[d.month - 1]} {d.year}")
    return out


def locate(value, kind: str, text: str) -> str | None:
    """Find a literal excerpt of `text` that spells `value`; None if absent."""
    if value is None:
        return None
    if kind == "amount":
        for v in _num_variants(float(value)):
            if v in text:
                return v
    elif kind == "date":
        for v in _date_variants(str(value)):
            if v in text:
                return v
    elif kind == "iban":
        target = re.sub(r"[   ]", "", str(value)).upper()
        for m in IBAN_RE.finditer(text):
            if re.sub(r"[   ]", "", m.group(1)).upper() == target:
                return m.group(1)
    else:  # ref and other verbatim strings
        if str(value) in text:
            return str(value)
    return None


# --- post-extraction invariants ----------------------------------------------

def check(payload: dict, text: str, source_sha256: str) -> tuple[dict | None, list[str]]:
    """Validate one extraction. Returns (enriched payload, []) or (None, reasons)."""
    r: list[str] = []
    t = payload.get("totals") or {}
    per_rate = payload.get("per_rate") or []
    rc = bool(payload.get("reverse_charge"))

    try:
        ht, tva, ttc = float(t["ht"]), float(t["tva"]), float(t["ttc"])
    except (KeyError, TypeError, ValueError):
        return None, ["totals missing or non-numeric"]
    if abs(ht + tva - ttc) > 0.01:
        r.append(f"arithmetic: HT {ht} + TVA {tva} != TTC {ttc}")
    if per_rate:
        if abs(sum(float(x["ht"]) for x in per_rate) - ht) > 0.01:
            r.append("arithmetic: sum(per_rate.ht) != totals.ht")
        if abs(sum(float(x["tva"]) for x in per_rate) - tva) > 0.01:
            r.append("arithmetic: sum(per_rate.tva) != totals.tva")
        for x in per_rate:
            if float(x["rate"]) not in VALID_RATES and not rc:
                r.append(f"rate {x['rate']} outside {sorted(VALID_RATES)} without reverse_charge")
    else:
        r.append("per_rate empty")

    siren = (payload.get("supplier") or {}).get("siren")
    if siren and not siren_valid(siren):
        r.append(f"SIREN checksum failed: {siren}")
    iban = payload.get("iban")
    if iban and not iban_valid(iban):
        r.append(f"IBAN mod-97 failed: {iban}")

    di, dd = payload.get("date_issue"), payload.get("date_due")
    try:
        d_issue = datetime.strptime(di, "%Y-%m-%d").date() if di else None
        d_due = datetime.strptime(dd, "%Y-%m-%d").date() if dd else None
        if d_issue and d_issue > date(2030, 1, 1):
            r.append("date_issue implausibly far in the future")
        if d_issue and d_due and d_issue > d_due:
            r.append("date_issue > date_due")
    except ValueError:
        r.append("dates not ISO YYYY-MM-DD")

    # anti-hallucination anchor: every critical value must exist literally in source
    prov = {}
    anchors = [("totals.ht", ht, "amount"), ("totals.tva", tva, "amount"), ("totals.ttc", ttc, "amount"),
               ("ref_supplier", payload.get("ref_supplier"), "ref"), ("date_issue", di, "date")]
    if dd:
        anchors.append(("date_due", dd, "date"))
    if iban:
        anchors.append(("iban", iban, "iban"))
    for field, value, kind in anchors:
        if value in (None, ""):
            r.append(f"critical field absent: {field}")
            continue
        # TVA of 0.00 under reverse charge is often not spelled out — anchor waived
        if kind == "amount" and float(value) == 0.0 and rc and field == "totals.tva":
            prov[field] = {"source_sha256": source_sha256, "raw_excerpt": "(reverse charge: TVA 0 not printed)"}
            continue
        ex = locate(value, kind, text)
        if ex is None:
            r.append(f"provenance: {field}={value!r} not found literally in source text")
        else:
            prov[field] = {"source_sha256": source_sha256, "raw_excerpt": ex}

    if r:
        return None, r
    out = dict(payload)
    out["provenance"] = prov
    return out, []