mirror of
https://github.com/Sea-Haven-Industries/procurement-ingest.git
synced 2026-10-07 16:08:54 +00:00
Adam Moussa
fc5b1bc0ba
Mirror WO PR #99's template-first approach for the Coupa PO processor. Two templates identified from a full 3,448-email triage: - coupa_new_po (95.5%): scaffolded; fails closed to the LLM until extract_new_po lands. - coupa_cancellation (2.9%): implemented. Nested contract with recursive validation, Decimal money, and a fail-closed gate. Derived fields (site_code/trade/fiscal_year) are deferred to a shared post-stage. Comments, revisions, multi-line, and non-USD emails fall back to Bedrock. docs/po-template-parser.md records the investigation, decisions, and remaining work. Signed-off-by: Adam Moussa <166072409+amoussa1229@users.noreply.github.com>
394 lines
16 KiB
Python
394 lines
16 KiB
Python
"""Deterministic template parser for Coupa purchase order emails.
|
||
|
||
Pure module: no boto3, no network. Runs ahead of the AI extraction path in the
|
||
purchase-order email processor. Only returns a parsed result when it is proven
|
||
conformant to one of the known Coupa templates; otherwise it fails closed and
|
||
signals the caller to fall back to the Bedrock AI extractor.
|
||
|
||
Mirrors the WO parser idioms (lambdas/wo/email_processor/template_parser.py):
|
||
classify_template -> extract -> validate (FAIL CLOSED) -> try_deterministic_parse
|
||
returning (parsed|None, method, template_id, reason). A failure is NEVER a
|
||
parsed result.
|
||
|
||
Full-bucket triage of all 3,448 inbound emails (2026-07-16) fixed the scope:
|
||
|
||
T1 coupa_new_po -- 3,294 / 3,448 (95.5%). Subject
|
||
"***Copy for Reference*** New Purchase Order <PO#> has been issued".
|
||
multipart/alternative; the text/plain part is a stable label-delimited
|
||
layout (PO ID / Status / Order Date / Revision Date / Payment Term / Req # /
|
||
Submitted By / On Behalf Of / Supplier block / Shipping block w/ Location
|
||
Code + Attn / a Lines section whose per-line metadata is U+2022-delimited:
|
||
Need By / Category / Account / Period [/ optional Part Number]).
|
||
Maps to email_type "new_po". 99.8% single-line-item, 100% USD.
|
||
|
||
T2 coupa_cancellation -- 100 / 3,448 (2.9%). Subject
|
||
"<SiteName> Purchase Order #<PO#> has been cancelled". Minimal body; the
|
||
only field the handler's save_cancellation() needs is po_number.
|
||
Maps to email_type "cancellation".
|
||
|
||
DELIBERATELY OUT OF SCOPE -> always AI fallback (never template-parsed):
|
||
* "New Comment on Purchase Order for Amazon" (19 / 3,448) -- an email_type the
|
||
handler enum does not model; do not fabricate a PO record deterministically.
|
||
* revision -- 0 distinct emails in 3,448; no template to build.
|
||
* any multi-line-item new_po (6 / 3,448) -- Lines-array structure unobserved.
|
||
* any non-USD new_po (0 observed) -- non-USD path entirely unexercised.
|
||
* non-Coupa senders (35 / 3,448) -- already rejected at the ses_auth layer.
|
||
|
||
Derived fields (site_code, trade, fiscal_year) are NOT computed here: the parser
|
||
leaves them None and a shared post-stage (handler enrich_parsed(), the pad_zip
|
||
precedent) fills them identically on both the template and the LLM path, so the
|
||
gate judges extraction fidelity only. coupa_category is a verbatim label capture,
|
||
not a classifier, and IS extracted here.
|
||
|
||
Entry point: try_deterministic_parse(email_data) -> (parsed|None, method,
|
||
template_id, reason_code).
|
||
"""
|
||
|
||
import re
|
||
from decimal import Decimal, InvalidOperation
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# IMPLEMENTATION STATUS (scaffold)
|
||
# [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.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# The contract keys (23), EXACTLY -- mirrors the AI EXTRACTION_PROMPT fields.
|
||
CONTRACT_KEYS = (
|
||
"email_type",
|
||
"po_number",
|
||
"po_status",
|
||
"source_system",
|
||
"submitted_by",
|
||
"on_behalf_of",
|
||
"order_date",
|
||
"revision_date",
|
||
"last_opened",
|
||
"acknowledged_at",
|
||
"payment_terms",
|
||
"requisition_number",
|
||
"department",
|
||
"view_order_url",
|
||
"supplier",
|
||
"site_code",
|
||
"ship_to",
|
||
"total_amount",
|
||
"currency",
|
||
"fiscal_year",
|
||
"trade",
|
||
"coupa_category",
|
||
"line_items",
|
||
)
|
||
|
||
SUPPLIER_KEYS = ("name",)
|
||
|
||
SHIP_TO_KEYS = (
|
||
"name",
|
||
"address",
|
||
"street",
|
||
"city",
|
||
"state",
|
||
"zip",
|
||
"location_code",
|
||
"attn",
|
||
)
|
||
|
||
LINE_ITEM_KEYS = (
|
||
"description",
|
||
"amount",
|
||
"currency",
|
||
"need_by",
|
||
"category",
|
||
"account_code",
|
||
"period",
|
||
"quantity",
|
||
"unit",
|
||
"price",
|
||
)
|
||
|
||
# Derived fields the parser must leave None (filled by the shared post-stage).
|
||
DERIVED_KEYS = ("site_code", "trade", "fiscal_year")
|
||
|
||
SOURCE_SYSTEM = "coupa"
|
||
|
||
# email_type per template.
|
||
_TEMPLATE_EMAIL_TYPE = {
|
||
"coupa_new_po": "new_po",
|
||
"coupa_cancellation": "cancellation",
|
||
}
|
||
VALID_EMAIL_TYPES = {"new_po", "revision", "cancellation"}
|
||
|
||
# Only these Status strings were observed on new_po (2,223 + 1,071 of 3,294).
|
||
# Anything else fails closed to the LLM -- NEVER default-to-new_po.
|
||
NEW_PO_SAFE_STATUSES = {"Issued - Created", "Issued - Scheduled for email"}
|
||
|
||
# Subject classifiers (Python unfolds header continuation lines before we see them).
|
||
_NEW_PO_SUBJECT = re.compile(
|
||
r"^\*\*\*Copy for Reference\*\*\* New Purchase Order\s+(?P<po>\S+)\s+has been issued$"
|
||
)
|
||
_CANCELLATION_SUBJECT = re.compile(
|
||
r"Purchase Order\s+#(?P<po>[A-Z0-9-]+)\s+has been cancelled\s*$"
|
||
)
|
||
|
||
# PO number shape, e.g. 2D-21456967, FK-21920384, B187-17955555.
|
||
_PO_ID_RE = re.compile(r"^[A-Z0-9]{1,6}-\d+$")
|
||
|
||
# U+2022 bullet delimiting per-line metadata in the Lines section.
|
||
_BULLET = "•"
|
||
|
||
# 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__"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Skeleton helpers (recursive -- unlike WO's flat contract)
|
||
# ---------------------------------------------------------------------------
|
||
def _empty_line_item():
|
||
return {k: None for k in LINE_ITEM_KEYS}
|
||
|
||
|
||
def _empty_candidate():
|
||
"""Full nested skeleton: every contract key present, None where absent."""
|
||
cand = {k: None for k in CONTRACT_KEYS}
|
||
cand["supplier"] = {k: None for k in SUPPLIER_KEYS}
|
||
cand["ship_to"] = {k: None for k in SHIP_TO_KEYS}
|
||
cand["line_items"] = [_empty_line_item()]
|
||
cand["source_system"] = SOURCE_SYSTEM
|
||
return cand
|
||
|
||
|
||
def _normalize(candidate):
|
||
"""Guarantee exact nested key presence before returning."""
|
||
out = _empty_candidate()
|
||
for k in CONTRACT_KEYS:
|
||
if k in candidate and k not in ("supplier", "ship_to", "line_items"):
|
||
out[k] = candidate[k]
|
||
supplier = candidate.get("supplier") or {}
|
||
out["supplier"] = {k: supplier.get(k) for k in SUPPLIER_KEYS}
|
||
ship_to = candidate.get("ship_to") or {}
|
||
out["ship_to"] = {k: ship_to.get(k) for k in SHIP_TO_KEYS}
|
||
items = candidate.get("line_items") or [{}]
|
||
out["line_items"] = [{k: (it or {}).get(k) for k in LINE_ITEM_KEYS} for it in items]
|
||
return out
|
||
|
||
|
||
def _clean(value):
|
||
"""Strip trailing CR and U+00A0 nbsp that every captured Coupa value carries;
|
||
collapse nothing else. Returns None for empty/placeholder 'None'."""
|
||
if value is None:
|
||
return None
|
||
v = value.replace("\r", "").replace(" ", " ").strip()
|
||
if v == "" or v == "None":
|
||
return None
|
||
return v
|
||
|
||
|
||
def _plain_lines(body):
|
||
return body.replace("\r\n", "\n").replace("\r", "\n").split("\n")
|
||
|
||
|
||
def _to_decimal(raw):
|
||
"""Parse a Coupa money token ('18,624.05') to Decimal, stripping thousands
|
||
separators. Returns _UNPARSEABLE if it does not parse (gate must reject)."""
|
||
if raw is None:
|
||
return None
|
||
token = raw.replace(",", "").strip()
|
||
try:
|
||
return Decimal(token)
|
||
except (InvalidOperation, ValueError):
|
||
return _UNPARSEABLE
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Subject helpers
|
||
# ---------------------------------------------------------------------------
|
||
def classify_template(email_data):
|
||
"""Return (template_id, reason). template_id in
|
||
{coupa_new_po, coupa_cancellation, unknown}."""
|
||
subject = _clean(email_data.get("subject")) or ""
|
||
if _NEW_PO_SUBJECT.match(subject):
|
||
return "coupa_new_po", "ok"
|
||
if _CANCELLATION_SUBJECT.search(subject):
|
||
return "coupa_cancellation", "ok"
|
||
return "unknown", "subject_no_match"
|
||
|
||
|
||
def _subject_po_id(email_data):
|
||
"""PO number parsed from the subject, or None."""
|
||
subject = _clean(email_data.get("subject")) or ""
|
||
m = _NEW_PO_SUBJECT.match(subject)
|
||
if m:
|
||
return m.group("po")
|
||
m = _CANCELLATION_SUBJECT.search(subject)
|
||
if m:
|
||
return m.group("po")
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# T2: coupa_cancellation -> cancellation (simple + safe: po_number only)
|
||
# ---------------------------------------------------------------------------
|
||
def extract_cancellation(email_data):
|
||
"""Cancellation carries no PO detail we trust beyond the id; the handler's
|
||
save_cancellation() only needs po_number. Everything else stays None."""
|
||
candidate = _empty_candidate()
|
||
candidate["email_type"] = "cancellation"
|
||
candidate["po_number"] = _subject_po_id(email_data)
|
||
return _normalize(candidate)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# T1: coupa_new_po -> new_po
|
||
# ---------------------------------------------------------------------------
|
||
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.
|
||
"""
|
||
candidate = _empty_candidate()
|
||
candidate["email_type"] = "new_po"
|
||
candidate["po_number"] = _subject_po_id(email_data)
|
||
# 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)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Validation gate -- FAIL CLOSED
|
||
# ---------------------------------------------------------------------------
|
||
def _structural_keys_ok(candidate):
|
||
if set(candidate.keys()) != set(CONTRACT_KEYS):
|
||
return False
|
||
if set((candidate.get("supplier") or {}).keys()) != set(SUPPLIER_KEYS):
|
||
return False
|
||
if set((candidate.get("ship_to") or {}).keys()) != set(SHIP_TO_KEYS):
|
||
return False
|
||
items = candidate.get("line_items")
|
||
if not isinstance(items, list) or not items:
|
||
return False
|
||
return all(set((it or {}).keys()) == set(LINE_ITEM_KEYS) for it in items)
|
||
|
||
|
||
def validate(candidate, template_id, email_data):
|
||
"""Return (True, 'ok') only if provably conformant; else (False, reason).
|
||
Every rule must hold. See failClosedGateRules in the investigation report."""
|
||
# (1) known template
|
||
if template_id not in _TEMPLATE_EMAIL_TYPE:
|
||
return False, "subject_no_match"
|
||
|
||
# (2) exact nested key-set
|
||
if not _structural_keys_ok(candidate):
|
||
return False, "key_set_mismatch"
|
||
|
||
# (3) derived fields must NOT be populated by the parser
|
||
for k in DERIVED_KEYS:
|
||
if candidate.get(k) is not None:
|
||
return False, "derived_field_set"
|
||
|
||
# (4) email_type matches the template's expected type
|
||
expected_type = _TEMPLATE_EMAIL_TYPE[template_id]
|
||
et = candidate.get("email_type")
|
||
if et not in VALID_EMAIL_TYPES:
|
||
return False, "missing_required_field"
|
||
if et != expected_type:
|
||
return False, "email_type_mismatch"
|
||
|
||
# (5) po_number: valid shape AND byte-equals the subject id
|
||
subject_po = _subject_po_id(email_data)
|
||
po = candidate.get("po_number")
|
||
if not po or not _PO_ID_RE.match(str(po)):
|
||
return False, "missing_required_field"
|
||
if po != subject_po:
|
||
return False, "po_id_mismatch"
|
||
|
||
if template_id == "coupa_cancellation":
|
||
# Cancellation is complete once po_number is proven; save_cancellation
|
||
# sets status=Cancelled by po_number and needs nothing else.
|
||
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:
|
||
return False, "unrecognized_status"
|
||
|
||
# (7) single-line only -- multi-line Lines structure is unobserved.
|
||
if len(candidate.get("line_items") or []) != 1:
|
||
return False, "multiline_unsupported"
|
||
|
||
# (8) currency must be exactly USD (non-USD path entirely unexercised).
|
||
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"
|
||
|
||
|
||
def try_deterministic_parse(email_data):
|
||
"""Entry point. Returns (parsed|None, parse_method, template_id, reason).
|
||
|
||
On a proven-conformant parse returns (dict, 'template', template_id, 'ok').
|
||
On any miss/invalid/exception returns (None, 'ai_fallback', template_id,
|
||
reason) -- a failure is NEVER a parsed result."""
|
||
template_id = "unknown"
|
||
try:
|
||
template_id, reason = classify_template(email_data)
|
||
if template_id == "unknown":
|
||
return None, "ai_fallback", template_id, reason
|
||
|
||
if template_id == "coupa_new_po":
|
||
candidate = extract_new_po(email_data)
|
||
else:
|
||
candidate = extract_cancellation(email_data)
|
||
|
||
ok, reason = validate(candidate, template_id, email_data)
|
||
if not ok:
|
||
return None, "ai_fallback", template_id, reason
|
||
return candidate, "template", template_id, "ok"
|
||
except Exception: # noqa: BLE001 -- fail closed on ANY extractor error
|
||
return None, "ai_fallback", template_id, "extractor_raised"
|