mirror of
https://github.com/Sea-Haven-Industries/procurement-ingest.git
synced 2026-10-07 16:08:54 +00:00
Wire template-first parse into PO handler with EMF metric
Run try_deterministic_parse ahead of the Bedrock extractor and fall back only on a miss/invalid (fail-closed) result. The shared enrich_parsed post-stage and the save_cancellation/save_revision/ save_new_po routing are untouched, so both paths write identical DynamoDB shapes and the po-ingest-site-extractor stream contract is preserved. Each record emits one ParseMethod EMF line (Seahaven/PoIngest/ ParseOutcome, dimension sets [ParseMethod] and [ParseMethod,TemplateId], ReasonCode/po_number ride-alongs) mirroring the WO idiom. The metric fires before the Bedrock call so a Bedrock-side error still records the ai_fallback outcome. Refs: #99
This commit is contained in:
parent
33b33fe256
commit
3418c7c819
1 changed files with 62 additions and 3 deletions
|
|
@ -2,7 +2,8 @@
|
|||
PO email processor Lambda.
|
||||
|
||||
Triggered by S3 events when SES delivers a Coupa PO email.
|
||||
Parses the raw email, sends it to Claude for structured extraction,
|
||||
Parses the raw email, tries the deterministic template parser first, falls
|
||||
back to Claude on Bedrock for structured extraction on a miss/invalid result,
|
||||
then writes the result to the purchase-orders DynamoDB table.
|
||||
"""
|
||||
|
||||
|
|
@ -17,6 +18,7 @@ from email import policy
|
|||
|
||||
import boto3
|
||||
from ses_auth import authenticate_inbound_email
|
||||
from template_parser import try_deterministic_parse
|
||||
|
||||
logger = logging.getLogger()
|
||||
logger.setLevel(logging.INFO)
|
||||
|
|
@ -30,6 +32,11 @@ BEDROCK_MODEL_ID = os.environ.get(
|
|||
"BEDROCK_MODEL_ID", "us.anthropic.claude-haiku-4-5-20251001-v1:0"
|
||||
)
|
||||
|
||||
# CloudWatch EMF namespace/metric for the parse-outcome metric (the PO
|
||||
# fallback-rate alarm in cdk/po_stack.py reads the ["ParseMethod"] series).
|
||||
METRIC_NAMESPACE = "Seahaven/PoIngest"
|
||||
METRIC_NAME = "ParseOutcome"
|
||||
|
||||
# "Cancelled" is a sticky, authoritative status: once a PO reaches it, a later
|
||||
# new_po/revision may enrich other fields but must never move it back to a
|
||||
# non-cancelled status.
|
||||
|
|
@ -285,6 +292,42 @@ def extract_with_claude(email_data: dict) -> dict:
|
|||
return json.loads(response_text.strip(), parse_float=Decimal)
|
||||
|
||||
|
||||
def _emit_parse_method_metric(method, template_id, reason_code, po_number):
|
||||
"""Emit one CloudWatch EMF line recording the parse outcome.
|
||||
|
||||
Zero-latency (no PutMetricData API call): the extraction path is async and
|
||||
the role already has logs:PutLogEvents. ParseMethod/TemplateId are the only
|
||||
promoted (dimensioned) fields to keep cardinality low; ReasonCode and
|
||||
po_number ride along as Logs-Insights-queryable properties.
|
||||
|
||||
Two dimension sets are published: ["ParseMethod"] (aggregated across all
|
||||
template ids -- the series the fallback-rate alarm queries) AND
|
||||
["ParseMethod", "TemplateId"] (per-template breakdown for Logs Insights /
|
||||
dashboards). CloudWatch materializes only the exact dimension sets listed
|
||||
here and does NOT auto-aggregate, so the alarm's single-dimension query
|
||||
would receive no data unless ["ParseMethod"] is emitted explicitly."""
|
||||
emf = {
|
||||
"_aws": {
|
||||
# EMF requires Timestamp (epoch ms); without it CloudWatch may not
|
||||
# extract the metric datapoint from the log event.
|
||||
"Timestamp": int(datetime.now(timezone.utc).timestamp() * 1000),
|
||||
"CloudWatchMetrics": [
|
||||
{
|
||||
"Namespace": METRIC_NAMESPACE,
|
||||
"Dimensions": [["ParseMethod"], ["ParseMethod", "TemplateId"]],
|
||||
"Metrics": [{"Name": METRIC_NAME, "Unit": "Count"}],
|
||||
}
|
||||
],
|
||||
},
|
||||
"ParseMethod": method,
|
||||
"TemplateId": template_id or "unknown",
|
||||
"ReasonCode": reason_code or "ok",
|
||||
"po_number": po_number or "",
|
||||
METRIC_NAME: 1,
|
||||
}
|
||||
print(json.dumps(emf))
|
||||
|
||||
|
||||
def pad_zip(zip_code: str | None) -> str | None:
|
||||
if not zip_code:
|
||||
return zip_code
|
||||
|
|
@ -477,9 +520,25 @@ def handler(event, context):
|
|||
email_data = parse_raw_email(raw_email)
|
||||
logger.info(f"Subject: {email_data['subject']}")
|
||||
|
||||
parsed = extract_with_claude(email_data)
|
||||
# Deterministic template parse first; fall back to the Bedrock AI
|
||||
# extractor only on a miss or an invalid (fail-closed) result. The
|
||||
# metric is emitted before the Bedrock call, so a Bedrock-side error
|
||||
# still records the ai_fallback outcome (the invocation then errors
|
||||
# into the errors alarm / DLQ as before).
|
||||
parsed, parse_method, template_id, parse_reason = try_deterministic_parse(
|
||||
email_data
|
||||
)
|
||||
_emit_parse_method_metric(
|
||||
parse_method,
|
||||
template_id,
|
||||
parse_reason,
|
||||
parsed.get("po_number") if parsed else None,
|
||||
)
|
||||
if parsed is None:
|
||||
parsed = extract_with_claude(email_data)
|
||||
logger.info(
|
||||
f"Parsed: type={parsed.get('email_type')}, po={parsed.get('po_number')}"
|
||||
f"Parsed ({parse_method}/{template_id}/{parse_reason}): "
|
||||
f"type={parsed.get('email_type')}, po={parsed.get('po_number')}"
|
||||
)
|
||||
|
||||
if not parsed.get("po_number"):
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue