diff --git a/lambdas/wo/email_processor/handler.py b/lambdas/wo/email_processor/handler.py index 2abdc8f..3b7efb8 100644 --- a/lambdas/wo/email_processor/handler.py +++ b/lambdas/wo/email_processor/handler.py @@ -7,25 +7,34 @@ then writes the result to DynamoDB. """ import email +import hashlib import json import logging import os import re -from datetime import datetime +from datetime import datetime, timezone from email import policy -import anthropic import boto3 +from template_parser import try_deterministic_parse + logger = logging.getLogger() logger.setLevel(logging.INFO) s3 = boto3.client("s3") dynamodb = boto3.resource("dynamodb") +bedrock = boto3.client("bedrock-runtime") WORK_ORDERS_TABLE = os.environ.get("WORK_ORDERS_TABLE", "WorkOrders") COMMENTS_TABLE = os.environ.get("COMMENTS_TABLE", "WorkOrderComments") -ANTHROPIC_API_KEY_SECRET_ARN = os.environ.get("ANTHROPIC_API_KEY_SECRET_ARN") +BEDROCK_MODEL_ID = os.environ.get( + "BEDROCK_MODEL_ID", "us.anthropic.claude-haiku-4-5-20251001-v1:0" +) + +# CloudWatch EMF namespace + metric for parse-outcome observability. +METRIC_NAMESPACE = "Seahaven/WorkorderIngest" +METRIC_NAME = "ParseOutcome" EXTRACTION_PROMPT = """\ You are an email parser for a facilities maintenance work order system. @@ -65,17 +74,6 @@ Rules: """ -def get_anthropic_client() -> anthropic.Anthropic: - """Create Anthropic client, fetching API key from Secrets Manager if configured.""" - if ANTHROPIC_API_KEY_SECRET_ARN: - secrets = boto3.client("secretsmanager") - secret = secrets.get_secret_value(SecretId=ANTHROPIC_API_KEY_SECRET_ARN) - api_key = secret["SecretString"] - return anthropic.Anthropic(api_key=api_key) - # Fall back to ANTHROPIC_API_KEY env var (for local testing) - return anthropic.Anthropic() - - def parse_raw_email(raw_bytes: bytes) -> dict: """Parse a raw email into subject, sender, body text.""" msg = email.message_from_bytes(raw_bytes, policy=policy.default) @@ -108,10 +106,8 @@ def parse_raw_email(raw_bytes: bytes) -> dict: } -def extract_with_claude(email_data: dict) -> dict: - """Send parsed email to Claude for structured extraction.""" - client = get_anthropic_client() - +def extract_with_bedrock(email_data: dict) -> dict: + """Send parsed email to Claude on Bedrock for structured extraction.""" email_text = ( f"Subject: {email_data['subject']}\n" f"From: {email_data['sender']}\n" @@ -122,18 +118,22 @@ def extract_with_claude(email_data: dict) -> dict: f"{email_data['body']}" ) - response = client.messages.create( - model="claude-haiku-4-5-20251001", - max_tokens=1024, - messages=[ + resp = bedrock.invoke_model( + modelId=BEDROCK_MODEL_ID, + body=json.dumps( { - "role": "user", - "content": f"{EXTRACTION_PROMPT}\n\nEMAIL:\n{email_text}", + "anthropic_version": "bedrock-2023-05-31", + "max_tokens": 1024, + "messages": [ + { + "role": "user", + "content": f"{EXTRACTION_PROMPT}\n\nEMAIL:\n{email_text}", + } + ], } - ], + ), ) - - response_text = response.content[0].text + response_text = json.loads(resp["body"].read())["content"][0]["text"] # Extract JSON from response (handle markdown code blocks) json_match = re.search(r"```(?:json)?\s*(.*?)```", response_text, re.DOTALL) @@ -143,11 +143,37 @@ def extract_with_claude(email_data: dict) -> dict: return json.loads(response_text.strip()) +def emit_parse_metric(method, template_id, reason_code, work_order_id): + """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 + work_order_id ride along as Logs-Insights-queryable properties.""" + emf = { + "_aws": { + "CloudWatchMetrics": [ + { + "Namespace": METRIC_NAMESPACE, + "Dimensions": [["ParseMethod", "TemplateId"]], + "Metrics": [{"Name": METRIC_NAME, "Unit": "Count"}], + } + ], + }, + "ParseMethod": method, + "TemplateId": template_id or "unknown", + "ReasonCode": reason_code or "ok", + "work_order_id": work_order_id or "", + METRIC_NAME: 1, + } + print(json.dumps(emf)) + + def save_work_order(parsed: dict, s3_key: str): """Create or update a work order in DynamoDB.""" table = dynamodb.Table(WORK_ORDERS_TABLE) work_order_id = parsed["work_order_id"] - now = datetime.utcnow().isoformat() + now = datetime.now(timezone.utc).isoformat() # Build update expression dynamically from non-null fields field_map = { @@ -210,14 +236,27 @@ def save_work_order(parsed: dict, s3_key: str): logger.info(f"Saved work order {work_order_id}") -def save_event(parsed: dict, s3_key: str): - """Save an event to the events table. Every email creates an event entry.""" +def save_event(parsed: dict, s3_key: str, object_key: str): + """Save an event to the events table. Every email creates an event entry. + + Issue #23: the range key (attr name stays ``comment_id``) must be unique per + source email AND identical across Lambda async retries of the same S3 object. + The uniqueness suffix is a deterministic hash of the S3 object key alone (not + the s3:// URI, so it is stable across a bucket rename), and wall-clock now() + is kept OUT of the key. When comment_time is absent we use the literal + 'nocomment' segment rather than now() -- otherwise each retry would produce a + different key and duplicate the row. Two distinct emails on the same WO map + to distinct object keys -> distinct rows.""" table = dynamodb.Table(COMMENTS_TABLE) work_order_id = parsed["work_order_id"] email_type = parsed.get("email_type", "unknown") - event_time = parsed.get("comment_time") or datetime.utcnow().isoformat() - event_id = f"{work_order_id}#{event_time}" + key_suffix = hashlib.sha256(object_key.encode("utf-8")).hexdigest()[:12] + comment_time = parsed.get("comment_time") + time_part = comment_time if comment_time else "nocomment" + event_id = f"{work_order_id}#{time_part}#{key_suffix}" + # created_at is display-only; may fall back to now() without affecting the key. + display_time = comment_time or datetime.now(timezone.utc).isoformat() item = { "work_order_id": work_order_id, @@ -225,9 +264,9 @@ def save_event(parsed: dict, s3_key: str): "record_type": email_type, "commenter": parsed.get("commenter") or "", "text": parsed.get("comment_text") or "", - "created_at": event_time, + "created_at": display_time, "source_email_s3_key": s3_key, - "ingested_at": datetime.utcnow().isoformat(), + "ingested_at": datetime.now(timezone.utc).isoformat(), } table.put_item(Item=item) @@ -250,12 +289,19 @@ def handler(event, context): email_data = parse_raw_email(raw_email) logger.info(f"Subject: {email_data['subject']}") - # Extract structured data with Claude - parsed = extract_with_claude(email_data) + # Deterministic template parse first; fall back to the AI extractor only + # on a miss or an invalid (fail-closed) result. + parsed, method, template_id, reason = try_deterministic_parse(email_data) + if parsed is None: + parsed = extract_with_bedrock(email_data) + method = "ai_fallback" logger.info( - f"Parsed: type={parsed.get('email_type')}, wo={parsed.get('work_order_id')}" + f"Parsed ({method}/{template_id}/{reason}): " + f"type={parsed.get('email_type')}, wo={parsed.get('work_order_id')}" ) + emit_parse_metric(method, template_id, reason, parsed.get("work_order_id")) + if not parsed.get("work_order_id"): logger.warning(f"No work order ID found in email, skipping: {key}") continue @@ -265,7 +311,8 @@ def handler(event, context): # Always upsert the work order with any new info save_work_order(parsed, s3_key) - # Save every email as an event for history tracking - save_event(parsed, s3_key) + # Save every email as an event for history tracking. The raw object key + # (not the s3:// URI) drives the retry-idempotent comment_id suffix. + save_event(parsed, s3_key, key) return {"statusCode": 200, "body": "OK"} diff --git a/lambdas/wo/email_processor/requirements.txt b/lambdas/wo/email_processor/requirements.txt index 28fbe99..dd0c8a3 100644 --- a/lambdas/wo/email_processor/requirements.txt +++ b/lambdas/wo/email_processor/requirements.txt @@ -1,2 +1 @@ -anthropic>=0.116.0 boto3>=1.43.47