diff --git a/README.md b/README.md index 44b0bce..e11c80d 100644 --- a/README.md +++ b/README.md @@ -10,15 +10,35 @@ Coupa purchase-order email ingestion pipeline. SES receives Amazon PO emails, Cl 4. The Lambda parses the email, sends it to Claude Haiku 4.5 for structured JSON extraction, and writes to DynamoDB. - `email_type: new_po` — conditional `PutItem` on `purchase-orders` (idempotent on `po_number`). - `email_type: cancellation` — `UpdateItem` marking the existing row `Cancelled`. -5. LedgerFlow consumes the table via DynamoDB Streams → `po-sync`. This repo only writes. +5. DynamoDB Streams (NEW_AND_OLD_IMAGES) on `purchase-orders` feeds two downstream consumers: + - **LedgerFlow** (`seahaven-slack-bot/po-sync`) — daily KB sync. + - **Verified-sites pipeline** (`po-ingest-site-extractor`) — real-time site address extraction (see below). A separate `po-web-ui` Lambda (Function URL, unauthenticated) renders a simple HTML dashboard scanning the table. +### Verified-sites pipeline + +The `po-ingest-site-extractor` Lambda is triggered by the DynamoDB Stream on every PO INSERT/MODIFY. It: + +1. Extracts an Amazon facility site code from `ship_to.name` using a regex cascade (parentheses, `LLC - CODE`, `Station CODE`, `DS - CODE`) with a fallback to the first `line_items` description. +2. Parses `ship_to.address` into structured fields (street, city, state, zip). +3. Upserts to the `verified-sites` DynamoDB table — atomically increments `poCount` and appends the PO number to `sourcePOs`. + +POs with no extractable site code are logged and skipped. + +Backfill stats (initial run): 14,825 POs scanned → 9,900 with extractable site codes → 1,100 unique sites. + ## Architecture - **IaC:** AWS CDK (Python), stack name `PoIngestStack`, region `us-east-1`. -- **Lambdas:** `po-email-processor` (S3-triggered) and `po-web-ui` (Function URL). Python 3.12, 256 MB, 60s timeout. -- **Storage:** S3 `po-ingest-emails-{AccountId}` with 90-day lifecycle expiry; DynamoDB `purchase-orders` (shared, not owned by this stack). +- **Lambdas** (all Python 3.12, arm64, 60-day log retention): + - `po-email-processor` — S3-triggered, parses PO emails via Claude Haiku. + - `po-web-ui` — Function URL, HTML dashboard. + - `po-ingest-site-extractor` — DynamoDB Streams-triggered, extracts site addresses. +- **Storage:** + - S3 `po-ingest-emails-{AccountId}` — 90-day lifecycle expiry. + - DynamoDB `purchase-orders` — owned by this stack, Streams enabled (NEW_AND_OLD_IMAGES). + - DynamoDB `verified-sites` — PK `siteCode`, GSI `by-state` on `state`. - **Secrets:** Anthropic API key in Secrets Manager at `po-ingest/anthropic-api-key`. - **SES:** adds the `PoEmailRule` to the existing `INBOUND_MAIL` receipt rule set (shared with `workorder-ingest`). @@ -53,3 +73,13 @@ python scripts/reprocess.py --execute # invokes po-email-processor for each ``` Inserts are conditional on `po_number`, so re-processing existing POs is a no-op. + +## Backfilling verified sites + +The stream Lambda handles all future POs automatically. To backfill from historical PO data (one-time): + +```bash +python scripts/backfill_sites.py +``` + +Uses the same extraction logic as the Lambda. Idempotent — safe to re-run.