From 33b33fe25696504e2679d7bc7a3995775cbd7f05 Mon Sep 17 00:00:00 2001 From: Adam Moussa Date: Thu, 16 Jul 2026 15:14:25 -0400 Subject: [PATCH] Implement PO new_po extraction and value-level gate Replace the extract_new_po scaffold stub with the full section-windowed extractor (duplicate-label anchoring, sentinel ship-to, label-keyed U+2022 bullet split, Decimal money from three anchored contexts only) and add value-level gate rules V1-V13. Both new_po_not_implemented scaffold guards are removed; rules 6-8 (unrecognized_status, multiline_unsupported, non_usd) go live. The gate re-derives every byte proof from the email body so an extractor bug cannot vouch for itself: amount re-serialization with a digit/comma border check (the thousands-separator truncation kill switch), sum(lines)==total against both Total blocks, anchor/supplier identity proofs, USPS address shape on the raw pre-enrichment zip, bullet label discipline, and sentinel/artifact hygiene. Any failure falls closed to the LLM; a validation failure is never a parsed result. Refs: #99 --- lambdas/po/email_processor/template_parser.py | 667 ++++++++++++++++-- 1 file changed, 618 insertions(+), 49 deletions(-) diff --git a/lambdas/po/email_processor/template_parser.py b/lambdas/po/email_processor/template_parser.py index ad4df58..7a9962a 100644 --- a/lambdas/po/email_processor/template_parser.py +++ b/lambdas/po/email_processor/template_parser.py @@ -48,15 +48,14 @@ import re from decimal import Decimal, InvalidOperation # --------------------------------------------------------------------------- -# IMPLEMENTATION STATUS (scaffold) +# IMPLEMENTATION STATUS # [x] contract + recursive skeleton/normalize # [x] classify_template (both templates) -# [x] coupa_cancellation extract + gate (simple, safe) -# [ ] coupa_new_po extract -- labeled fields, duplicate-label anchoring, -# U+2022 line split, Decimal amounts <-- TODO, needs fixtures -# [ ] coupa_new_po value-level gate rules <-- TODO, currently FAILS CLOSED -# Until the new_po TODOs land, a new_po email deterministically returns -# reason 'new_po_not_implemented' and falls back to the LLM -- safe by design. +# [x] coupa_cancellation extract + gate +# [x] coupa_new_po extract -- labeled fields, duplicate-label anchoring, +# U+2022 line split, Decimal amounts (built against the scrubbed real +# fixture corpus in tests/fixtures/) +# [x] coupa_new_po value-level gate rules V1-V13 (see validate()) # --------------------------------------------------------------------------- # The contract keys (23), EXACTLY -- mirrors the AI EXTRACTION_PROMPT fields. @@ -142,6 +141,69 @@ _PO_ID_RE = re.compile(r"^[A-Z0-9]{1,6}-\d+$") # U+2022 bullet delimiting per-line metadata in the Lines section. _BULLET = "•" +# --- new_po layout patterns (built against the scrubbed real fixture corpus) --- +# Money tokens are read ONLY from three anchored contexts: the Lines-section +# ' for ' line, the Total-block standalone amount line, and the +# Items-summary ' x ' line. NEVER free money-shaped scanning: +# the Items summary carries unit-price tokens distinct from line amounts. +_MONEY_RE = re.compile(r"\d{1,3}(?:,\d{3})*\.\d{2}") +_CURRENCY_RE = re.compile(r"[A-Z]{3}") +# Items-summary quantity line, e.g. '1.0 EACH x 55,206.00'. +_SUMMARY_ITEM_RE = re.compile( + r"^(?P\d+(?:\.\d+)?) (?P[A-Z]+) x (?P\d{1,3}(?:,\d{3})*\.\d{2})$" +) +# Lines-block quantity evidence line, e.g. '1.0 EA' (gate rule V13 cross-check). +_LINE_QTY_RE = re.compile(r"^(?P\d+(?:\.\d+)?) (?P[A-Z]+)$") +# Lines-block description/amount line. GREEDY desc: '.+' binds the LAST ' for ', +# so a description containing the word 'for' can never shift the amount. +_LINE_DESC_AMT_RE = re.compile( + r"^(?P.+) for (?P\d{1,3}(?:,\d{3})*\.\d{2}) (?P[A-Z]{3})$" +) +_VIEW_ORDER_URL_RE = re.compile(r"^https://supplier\.coupahost\.com/orders/\S+$") +_ORDER_URL_ID_RE = re.compile(r"^https://supplier\.coupahost\.com/orders/(\d+)\b") +_LOCATION_CODE_RE = re.compile(r"^Location Code: (?P\d+)$") +_ATTN_RE = re.compile(r"^Attn: (?P.+)$") +# Ship-to city line immediately preceding 'United States'. +_CITY_STATE_ZIP_RE = re.compile( + r"^(?P.+), (?P[A-Z]{2}) (?P\d{5}(?:-\d{4})?)$" +) +_US_SENTINEL = "United States" +_SUPPLIER_MARKER = ( + "SEA HAVEN" # case-sensitive; drift becomes fallback, never wrong data +) + +# More Detail block: label lines, each expected exactly once (gate rule V9). +_MORE_DETAIL_FIELDS = { + "Department": "department", + "Status": "po_status", + "Last Opened": "last_opened", + "Order Date": "order_date", + "Acknowledged At": "acknowledged_at", + "Revision Date": "revision_date", + "Payment Term": "payment_terms", + "Req #": "requisition_number", +} +_MORE_DETAIL_LABELS = ("PO ID", *_MORE_DETAIL_FIELDS) + +# Per-line bullet metadata: closed label set, assigned purely by leading label +# (longest label first), NEVER by ordinal position -- real data has an optional +# 'Part Number' segment between 'Category' and 'Account', and every run starts +# with a 'Supplier ' segment. +_BULLET_LABELS = ("Part Number", "Need By", "Category", "Account", "Period", "Supplier") +_BULLET_FIELDS = { + "Need By": "need_by", + "Category": "category", + "Account": "account_code", + "Period": "period", +} + +# Frozen USPS state/territory codes (50 states + DC + territories). +_USPS_STATES = frozenset( + """AL AK AZ AR CA CO CT DE FL GA HI ID IL IN IA KS KY LA ME MD MA MI MN MS + MO MT NE NV NH NJ NM NY NC ND OH OK OR PA RI SC SD TN TX UT VT VA WA WV WI + WY DC PR VI GU AS MP""".split() +) + # Sentinel: a label was present but its value did not parse. Must FAIL the gate # (present-but-unparseable), distinct from an absent value (None). _UNPARSEABLE = "__UNPARSEABLE__" @@ -206,6 +268,74 @@ def _to_decimal(raw): return _UNPARSEABLE +# --------------------------------------------------------------------------- +# Line navigation helpers (shared by the extractor and the gate; both operate +# on email_data["body"] only -- the gate re-derives its own byte evidence and +# never trusts extractor-carried state it can re-derive) +# --------------------------------------------------------------------------- +def _visible(line): + """True when the raw line carries visible content. The literal placeholder + 'None' IS visible (unlike _clean, which maps it to None), so label values + of 'None' are found -- not skipped over into the next label line.""" + return bool(line.replace("\r", "").replace("\xa0", "").strip()) + + +def _indices(lines, label): + """All indices whose _clean-ed content equals the label exactly.""" + return [i for i, ln in enumerate(lines) if _clean(ln) == label] + + +def _find_after(lines, label, start): + """First index >= start whose _clean-ed content equals label, or None.""" + for i in range(start, len(lines)): + if _clean(lines[i]) == label: + return i + return None + + +def _next_visible(lines, idx, end=None): + """Index of the first visible line strictly after idx (before end), or None.""" + stop = len(lines) if end is None else min(end, len(lines)) + for j in range(idx + 1, stop): + if _visible(lines[j]): + return j + return None + + +def _split_bullet_segments(raw_line): + """Split a Lines-section metadata line on the bare U+2022 bullet; strip each + segment of spaces and nbsp; drop empty segments.""" + segments = [] + for seg in raw_line.replace("\r", "").split(_BULLET): + seg = seg.replace("\xa0", " ").strip() + if seg: + segments.append(seg) + return segments + + +def _match_bullet_label(segment): + """(label, value) by leading-label prefix match (longest label first) + against the closed _BULLET_LABELS set, or (None, None) if unrecognized.""" + for label in sorted(_BULLET_LABELS, key=len, reverse=True): + if segment == label: + return label, None + if segment.startswith(label + " "): + return label, segment[len(label) :].strip() + return None, None + + +def _walk_leaves(obj): + """Yield every scalar leaf of a nested dict/list candidate.""" + if isinstance(obj, dict): + for v in obj.values(): + yield from _walk_leaves(v) + elif isinstance(obj, list): + for v in obj: + yield from _walk_leaves(v) + else: + yield obj + + # --------------------------------------------------------------------------- # Subject helpers # --------------------------------------------------------------------------- @@ -247,35 +377,191 @@ def extract_cancellation(email_data): # --------------------------------------------------------------------------- # T1: coupa_new_po -> new_po # --------------------------------------------------------------------------- +def _assign_bullet_metadata(item, raw_line): + """Assign the U+2022 metadata segments to the item BY LEADING LABEL. + + 'Supplier' is recognized but not stored on the item (gate rule V5 proves it + byte-equals supplier.name from the body); 'Part Number' is recognized but + discarded (LINE_ITEM_KEYS has no slot -- inventing one would break the + key_set_mismatch rule and LLM-path shape parity). Unrecognized or duplicate + segments are the gate's job to reject (rule V8).""" + for segment in _split_bullet_segments(raw_line): + label, value = _match_bullet_label(segment) + field = _BULLET_FIELDS.get(label) + if field and item[field] is None: + item[field] = _clean(value) + + def extract_new_po(email_data): """Extract the new_po contract from the text/plain body. - TODO(new_po): this is the core extraction work and needs fixtures before it - can be trusted. Implement, in order, guarding each with a gate rule below: - * labeled single-occurrence fields (PO ID, Status, Order Date, Revision - Date, Payment Term, Req #, Submitted By, On Behalf Of, view_order_url, - Department/Last Opened/Acknowledged At with literal 'None' -> null). - * DUPLICATE labels: 'Supplier', 'Shipping', 'Total' each appear TWICE - (summary placeholder + detail block; first 'Shipping' is literally - 'None'). Anchor supplier.name on the FIRST 'Supplier', the ship_to block - on the SECOND 'Shipping', and pick a deterministic 'Total'. - * ship_to by SENTINEL anchors ('United States' / 'Location Code:' / - 'Attn:'), never fixed line offsets (multi-line street unobserved). - * line_items: split the metadata line on the bare U+2022 (_BULLET) and - assign each segment BY LEADING LABEL word ('Need By '/'Category '/ - 'Account '/'Period '/'Part Number '), never by ordinal position. - * amounts via _to_decimal (thousands-separator safe); currency = the exact - 3-letter token adjacent to the amount. - * coupa_category = verbatim 'Category' bullet value. - Leave DERIVED_KEYS (site_code, trade, fiscal_year) as None -- filled later by - the shared post-stage. + Permissive capture, section-windowed: each anchor line is located by exact + _clean-ed full-line equality, STRICTLY AFTER the previous anchor. Missing + anchors leave fields None -- the gate then fails closed. Representation- + agnostic: matches only on _clean-ed lines, never on '\\r'-suffixed literals + (body line endings are decode-path dependent). + + Duplicate labels: 'Supplier', 'Shipping', 'Total' each appear TWICE (summary + placeholder + detail block; the first 'Shipping' value is literally 'None'). + supplier.name anchors on the FIRST 'Supplier'; ship_to on the SECOND + 'Shipping'; total on the SECOND 'Total'. + + DERIVED_KEYS (site_code, trade, fiscal_year) stay None -- filled later by + the shared post-stage identically on both paths. """ candidate = _empty_candidate() candidate["email_type"] = "new_po" candidate["po_number"] = _subject_po_id(email_data) + + lines = _plain_lines(email_data["body"]) + + # --- Summary section (start .. 'More Detail') --- + more_detail = _find_after(lines, "More Detail", 0) + summary_end = more_detail if more_detail is not None else len(lines) + for label, field in ( + ("Submitted By", "submitted_by"), + ("On Behalf Of", "on_behalf_of"), + ): + idx = _find_after(lines, label, 0) + if idx is not None and idx < summary_end: + j = _next_visible(lines, idx, summary_end) + if j is not None: + candidate[field] = _clean(lines[j]) + sup1 = _find_after(lines, "Supplier", 0) + if sup1 is not None and sup1 < summary_end: + j = _next_visible(lines, sup1, summary_end) + if j is not None: + candidate["supplier"]["name"] = _clean(lines[j]) + # view_order_url: the unique orders link in the summary (0 or >1 -> None). + url_lines = [ + _clean(lines[i]) + for i in range(summary_end) + if _VIEW_ORDER_URL_RE.match(_clean(lines[i]) or "") + ] + if len(url_lines) == 1: + candidate["view_order_url"] = url_lines[0] + # Items-summary quantity lines ('1.0 EACH x 55,206.00'), collected in order. + summary_items = [ + m + for i in range(summary_end) + if (m := _SUMMARY_ITEM_RE.fullmatch(_clean(lines[i]) or "")) + ] + + # --- More Detail block ('More Detail' .. second 'Supplier') --- + sup2 = None + if more_detail is not None: + sup2 = _find_after(lines, "Supplier", more_detail + 1) + md_end = sup2 if sup2 is not None else len(lines) + for label, field in _MORE_DETAIL_FIELDS.items(): + idx = _find_after(lines, label, more_detail + 1) + if idx is not None and idx < md_end: + j = _next_visible(lines, idx, md_end) + if j is not None: + candidate[field] = _clean(lines[j]) + # The FIRST 'Shipping' lives in this block; its value must be the literal + # 'None' placeholder (gate rule V4 tripwire) and is never used for ship_to. + + # --- ship_to (second 'Shipping' .. 'Lines'), sentinel-anchored --- + ship2 = _find_after(lines, "Shipping", sup2 + 1) if sup2 is not None else None + lines_anchor = _find_after(lines, "Lines", ship2 + 1) if ship2 is not None else None + st_end = lines_anchor if lines_anchor is not None else len(lines) + ship_to = candidate["ship_to"] + if ship2 is not None: + name_idx = _next_visible(lines, ship2, st_end) + if name_idx is not None: + ship_to["name"] = _clean(lines[name_idx]) + us_idx = _find_after(lines, _US_SENTINEL, name_idx + 1) + if us_idx is not None and us_idx < st_end: + city_idx = us_idx - 1 + m = None + if city_idx > name_idx: + m = _CITY_STATE_ZIP_RE.fullmatch(_clean(lines[city_idx]) or "") + if m: + ship_to["city"] = m.group("city") + ship_to["state"] = m.group("state") + ship_to["zip"] = m.group("zip") + street = [ + _clean(lines[j]) + for j in range(name_idx + 1, city_idx) + if _visible(lines[j]) + ] + if street: + ship_to["street"] = "\n".join(street) + ship_to["address"] = "\n".join( + _clean(lines[j]) + for j in range(name_idx, us_idx + 1) + if _visible(lines[j]) + ) + for j in range(us_idx + 1, st_end): + cl = _clean(lines[j]) or "" + lc = _LOCATION_CODE_RE.fullmatch(cl) + if lc and ship_to["location_code"] is None: + ship_to["location_code"] = lc.group("lc") + attn = _ATTN_RE.fullmatch(cl) + if attn and ship_to["attn"] is None: + ship_to["attn"] = _clean(attn.group("attn")) + + # --- Lines section ('Lines' .. second 'Total') --- + # Item blocks are delimited by the lone U+00A0 line(s). EVERY block is + # extracted, even when >1, so gate rule 7 fires with honest + # multiline_unsupported data (never silently keep item 0). + total2 = ( + _find_after(lines, "Total", lines_anchor + 1) + if lines_anchor is not None + else None + ) + items = [] + if lines_anchor is not None: + end = total2 if total2 is not None else len(lines) + blocks, block = [], [] + for j in range(lines_anchor + 1, end): + if lines[j].replace("\r", "") == "\xa0": + blocks.append(block) + block = [] + else: + block.append(lines[j]) + blocks.append(block) + for block in blocks: + visible = [ln for ln in block if _visible(ln)] + if not visible: + continue + item = _empty_line_item() + for raw in visible: + dm = _LINE_DESC_AMT_RE.fullmatch(_clean(raw) or "") + if dm and item["description"] is None: + item["description"] = _clean(dm.group("desc")) + item["amount"] = _to_decimal(dm.group("amt")) + item["currency"] = dm.group("cur") + elif _BULLET in raw: + _assign_bullet_metadata(item, raw) + # The optional ' EA' evidence line is not stored: quantity/ + # unit/price come from the Items summary; gate rule V13 + # cross-checks the EA line against it from the body. + items.append(item) + if items: + candidate["line_items"] = items + # coupa_category is a VERBATIM copy of item 0's Category bullet value. + candidate["coupa_category"] = items[0]["category"] + if len(summary_items) == 1 and len(items) == 1: + m = summary_items[0] + items[0]["quantity"] = _to_decimal(m.group("qty")) + items[0]["unit"] = m.group("unit") + items[0]["price"] = _to_decimal(m.group("price")) + + # --- Total block (second 'Total' .. end) --- + if total2 is not None: + amt_idx = _next_visible(lines, total2) + if amt_idx is not None: + token = _clean(lines[amt_idx]) or "" + if _MONEY_RE.fullmatch(token): + candidate["total_amount"] = _to_decimal(token) + cur_idx = _next_visible(lines, amt_idx) + if cur_idx is not None: + cur_token = _clean(lines[cur_idx]) or "" + if _CURRENCY_RE.fullmatch(cur_token): + candidate["currency"] = cur_token + # DERIVED_KEYS intentionally left None (shared post-stage fills them). - # TODO(new_po): populate the remaining fields per the docstring, then remove - # the fail-closed guard in validate(). return _normalize(candidate) @@ -333,11 +619,6 @@ def validate(candidate, template_id, email_data): return True, "ok" # ---- coupa_new_po ---- - # SCAFFOLD GUARD: extract_new_po is not implemented yet, so a new_po parse is - # never provable -> fail closed to the LLM with an honest reason. Remove this - # line once extract_new_po + the value-level rules below are implemented. - return False, "new_po_not_implemented" - # (6) status must be one of the confirmed-safe strings. status = candidate.get("po_status") if status not in NEW_PO_SAFE_STATUSES: @@ -351,22 +632,310 @@ def validate(candidate, template_id, email_data): if candidate.get("currency") != "USD": return False, "non_usd" - # TODO(new_po) value-level gate rules -- implement alongside extract_new_po: - # * amount fidelity: re-serialize each Decimal to the source grouped format - # and require byte-equality with the matched substring; reject if the - # token is bordered by a digit/comma (guards '18,624.05' -> '624.05'). - # Then require sum(line_items[].amount) == total_amount. - # * duplicate-label anchoring: reject Shipping=='None'; require ship_to has - # a 'Location Code:' line; supplier.name contains 'SEA HAVEN'; - # ship_to.name != supplier.name. - # * bullet metadata: every U+2022 segment carries a recognized leading - # label; fail on any unlabeled segment. - # * address shape: ship_to.state null-or-valid-USPS; ship_to.zip null-or - # ^\d{5}(-\d{4})?$ after pad_zip; tokenize city/state/zip by pattern. - # * _UNPARSEABLE discipline: reject any field still == _UNPARSEABLE. - # * whitespace hygiene: reject any value still containing \r or \xa0. - # Until the above land, new_po is not provable -> fail closed to the LLM. - return False, "new_po_not_implemented" + # (V1-V13) value-level rules: every byte proof is RE-DERIVED from + # email_data["body"] -- the gate never trusts extractor-carried state it + # can re-derive, so an extractor bug cannot vouch for itself. + return _validate_new_po_values(candidate, email_data) + + +def _money_border_ok(line_text, serialized): + """True when `serialized` occurs in the source line with a character before + it that is not a digit or comma (the '18,624.05' -> '624.05' kill switch: + a truncated capture re-serializes as '624.05', but every occurrence of that + string in its source line is preceded by a comma or digit).""" + idx = line_text.find(serialized) + while idx != -1: + prev = line_text[idx - 1] if idx > 0 else "" + if prev not in "0123456789,": + return True + idx = line_text.find(serialized, idx + 1) + return False + + +def _validate_new_po_values(candidate, email_data): # noqa: PLR0911, PLR0912, PLR0915 + """Value-level gate rules V1-V13 for coupa_new_po. FAIL CLOSED. + + Anchor integrity (V4) runs first because every later byte proof needs the + anchor frame; the remaining rules run in spec order. Each rule re-derives + its evidence from email_data["body"] -- duplicated regex work, negligible + at ~57 emails/day, in exchange for extractor-bug independence.""" + lines = _plain_lines(email_data["body"]) + items = candidate.get("line_items") or [] + ship_to = candidate.get("ship_to") or {} + supplier_name = (candidate.get("supplier") or {}).get("name") + + # --- V4 anchor integrity -> anchor_violation --- + more_detail_idxs = _indices(lines, "More Detail") + supplier_idxs = _indices(lines, "Supplier") + shipping_idxs = _indices(lines, "Shipping") + total_idxs = _indices(lines, "Total") + if len(more_detail_idxs) != 1: + return False, "anchor_violation" + if len(supplier_idxs) != 2 or len(shipping_idxs) != 2 or len(total_idxs) != 2: + return False, "anchor_violation" + more_detail = more_detail_idxs[0] + # The FIRST Shipping value must be the literal 'None' placeholder -- the + # dup-label-swap tripwire (a real address under the first label means the + # layout drifted; ship_to would have been read from the wrong block). + first_ship_val = "" + if shipping_idxs[0] + 1 < len(lines): + first_ship_val = ( + lines[shipping_idxs[0] + 1].replace("\r", "").replace("\xa0", "").strip() + ) + if first_ship_val != "None": + return False, "anchor_violation" + lines_anchor = _find_after(lines, "Lines", shipping_idxs[1] + 1) + if lines_anchor is None: + return False, "anchor_violation" + # Section ordering: summary Supplier < More Detail < detail Supplier < + # second Shipping < Lines < second Total; summary Total before More Detail. + if not ( + supplier_idxs[0] + < more_detail + < supplier_idxs[1] + < shipping_idxs[1] + < lines_anchor + < total_idxs[1] + ): + return False, "anchor_violation" + if not (total_idxs[0] < more_detail < shipping_idxs[0] < supplier_idxs[1]): + return False, "anchor_violation" + lines_end = total_idxs[1] + + # --- V1 money fidelity -> amount_mismatch --- + # Every captured money value must re-locate its raw source token in the + # body: token fullmatches the grouped money shape, format(value, ',.2f') + # byte-equals it, and the character before it is not a digit/comma. + for_matches = [] + for j in range(lines_anchor + 1, lines_end): + m = _LINE_DESC_AMT_RE.fullmatch(_clean(lines[j]) or "") + if m: + for_matches.append(m) + if len(for_matches) != len(items): + return False, "amount_mismatch" + for item, m in zip(items, for_matches): + amount = item.get("amount") + if not isinstance(amount, Decimal): + return False, "amount_mismatch" + serialized = format(amount, ",.2f") + if serialized != m.group("amt"): + return False, "amount_mismatch" + if not _money_border_ok(m.string, serialized): + return False, "amount_mismatch" + summary_matches = [ + m + for i in range(more_detail) + if (m := _SUMMARY_ITEM_RE.fullmatch(_clean(lines[i]) or "")) + ] + price = items[0].get("price") if items else None + if price is not None: + if not isinstance(price, Decimal) or len(summary_matches) != 1: + return False, "amount_mismatch" + serialized = format(price, ",.2f") + if serialized != summary_matches[0].group("price"): + return False, "amount_mismatch" + if not _money_border_ok(summary_matches[0].string, serialized): + return False, "amount_mismatch" + total = candidate.get("total_amount") + if not isinstance(total, Decimal): + return False, "amount_mismatch" + + # --- V3 dual-Total proof -> amount_mismatch (V2 needs its token, so it + # derives here; sum proof follows immediately) --- + total_tokens = [] + for t_idx in total_idxs: + a_idx = _next_visible(lines, t_idx) + if a_idx is None: + return False, "amount_mismatch" + token = _clean(lines[a_idx]) or "" + if not _MONEY_RE.fullmatch(token): + return False, "amount_mismatch" + c_idx = _next_visible(lines, a_idx) + cur_token = (_clean(lines[c_idx]) or "") if c_idx is not None else "" + if not _CURRENCY_RE.fullmatch(cur_token): + return False, "amount_mismatch" + total_tokens.append((token, cur_token)) + if total_tokens[0] != total_tokens[1]: + return False, "amount_mismatch" + serialized = format(total, ",.2f") + if serialized != total_tokens[1][0]: + return False, "amount_mismatch" + if candidate.get("currency") != total_tokens[1][1]: + return False, "amount_mismatch" + + # --- V2 sum proof -> amount_mismatch (exact Decimal equality; deliberately + # NO qty*price==amount rule -- corpus shows partial quantities) --- + if sum(item["amount"] for item in items) != total: + return False, "amount_mismatch" + + # --- V5 supplier proof -> anchor_violation --- + if not supplier_name or _SUPPLIER_MARKER not in supplier_name: + return False, "anchor_violation" + det_idx = _next_visible(lines, supplier_idxs[1], shipping_idxs[1]) + if det_idx is None or _clean(lines[det_idx]) != supplier_name: + return False, "anchor_violation" + sum_idx = _next_visible(lines, supplier_idxs[0], more_detail) + if sum_idx is None or _clean(lines[sum_idx]) != supplier_name: + return False, "anchor_violation" + if ship_to.get("name") == supplier_name: + return False, "anchor_violation" + + # --- V8 bullet discipline -> bullet_label_unrecognized (V5's per-item + # Supplier-segment proof rides the same walk) --- + bullet_lines = [ + lines[j] for j in range(lines_anchor + 1, lines_end) if _BULLET in lines[j] + ] + if len(bullet_lines) != len(items): + return False, "bullet_label_unrecognized" + for raw in bullet_lines: + seen = {} + for segment in _split_bullet_segments(raw): + label, value = _match_bullet_label(segment) + if label is None or label in seen: + return False, "bullet_label_unrecognized" + seen[label] = value + if set(seen) - { + "Supplier", + "Need By", + "Category", + "Account", + "Period", + "Part Number", + }: + return False, "bullet_label_unrecognized" + if not {"Supplier", "Need By", "Category", "Account", "Period"} <= set(seen): + return False, "bullet_label_unrecognized" + if _clean(seen["Supplier"]) != supplier_name: + return False, "anchor_violation" + + # --- V6 ship_to required -> missing_required_field --- + for field in ("name", "street", "city", "state", "zip", "location_code"): + if not ship_to.get(field): + return False, "missing_required_field" + if not re.fullmatch(r"\d+", ship_to["location_code"]): + return False, "missing_required_field" + lc_values = [ + m.group("lc") + for j in range(shipping_idxs[1] + 1, lines_anchor) + if (m := _LOCATION_CODE_RE.fullmatch(_clean(lines[j]) or "")) + ] + if ship_to["location_code"] not in lc_values: + return False, "missing_required_field" + attn_values = [ + _clean(m.group("attn")) + for j in range(shipping_idxs[1] + 1, lines_anchor) + if (m := _ATTN_RE.fullmatch(_clean(lines[j]) or "")) + ] + if attn_values: + if ship_to.get("attn") not in attn_values: + return False, "missing_required_field" + elif ship_to.get("attn") is not None: + return False, "missing_required_field" + + # --- V7 address shape -> address_shape_invalid (gated on the RAW + # pre-enrichment zip: validate() runs BEFORE enrich_parsed, so a short zip + # like '2149' fails closed to the LLM path where pad_zip repairs it -- + # both paths then get identical pad_zip treatment downstream) --- + us_idx = _find_after(lines, _US_SENTINEL, shipping_idxs[1] + 1) + if us_idx is None or us_idx >= lines_anchor: + return False, "address_shape_invalid" + m = _CITY_STATE_ZIP_RE.fullmatch(_clean(lines[us_idx - 1]) or "") + if not m: + return False, "address_shape_invalid" + if ( + m.group("city") != ship_to["city"] + or m.group("state") != ship_to["state"] + or m.group("zip") != ship_to["zip"] + ): + return False, "address_shape_invalid" + if ship_to["state"] not in _USPS_STATES: + return False, "address_shape_invalid" + if not re.fullmatch(r"\d{5}(-\d{4})?", ship_to["zip"]): + return False, "address_shape_invalid" + + # --- V9 required labeled fields -> missing_required_field --- + for field in ( + "po_status", + "order_date", + "payment_terms", + "requisition_number", + "submitted_by", + "view_order_url", + ): + if candidate.get(field) is None: + return False, "missing_required_field" + for item in items: + for field in ( + "description", + "amount", + "currency", + "need_by", + "category", + "account_code", + "period", + ): + if item.get(field) is None: + return False, "missing_required_field" + for label in _MORE_DETAIL_LABELS: + if len(_indices(lines, label)) != 1: + return False, "missing_required_field" + if candidate.get("coupa_category") != items[0].get("category"): + return False, "missing_required_field" + # Nullable by design: on_behalf_of, department, last_opened, acknowledged_at, + # revision_date, attn, quantity, unit, price. + + # --- V10 PO identity proofs -> po_id_mismatch --- + po = candidate["po_number"] + po_id_idx = _indices(lines, "PO ID")[0] + v_idx = _next_visible(lines, po_id_idx) + if v_idx is None or _clean(lines[v_idx]) != po: + return False, "po_id_mismatch" + if not _indices(lines, f"Amazon Purchase Order #{po}"): + return False, "po_id_mismatch" + um = _ORDER_URL_ID_RE.match(candidate.get("view_order_url") or "") + if not um or um.group(1) != po.split("-", 1)[1]: + return False, "po_id_mismatch" + + # --- V11 sentinel discipline -> unparseable_value --- + for leaf in _walk_leaves(candidate): + if isinstance(leaf, str) and leaf == _UNPARSEABLE: + return False, "unparseable_value" + + # --- V12 hygiene -> residual_artifact --- + for leaf in _walk_leaves(candidate): + if isinstance(leaf, str) and ("\r" in leaf or "\xa0" in leaf): + return False, "residual_artifact" + + # --- V13 quantity/unit/price coherence -> amount_mismatch --- + quantity = items[0].get("quantity") + unit = items[0].get("unit") + if quantity is not None or unit is not None or price is not None: + if not isinstance(quantity, Decimal) or quantity <= 0: + return False, "amount_mismatch" + if not unit or not re.fullmatch(r"[A-Z]+", unit): + return False, "amount_mismatch" + if not isinstance(price, Decimal): + return False, "amount_mismatch" + if len(summary_matches) != 1: + return False, "amount_mismatch" + if _to_decimal(summary_matches[0].group("qty")) != quantity: + return False, "amount_mismatch" + if summary_matches[0].group("unit") != unit: + return False, "amount_mismatch" + # The Lines-block ' EA' evidence line must numeric-equal the + # summary quantity when present. + ea_matches = [ + m + for j in range(lines_anchor + 1, lines_end) + if (m := _LINE_QTY_RE.fullmatch(_clean(lines[j]) or "")) + ] + if ea_matches: + if len(ea_matches) != 1: + return False, "amount_mismatch" + if _to_decimal(ea_matches[0].group("qty")) != quantity: + return False, "amount_mismatch" + + return True, "ok" def try_deterministic_parse(email_data):