# Procurement Ingest Unified email ingestion pipelines for Amazon procurement data. Two independent pipelines — purchase orders (Coupa) and work orders (APM/Hexagon EAM) — share a single repo and CDK app but deploy as separate CloudFormation stacks. ## Pipelines ### Purchase Orders (`po-ingest` stack) Coupa PO emails are received at `amazon_po@int.seahaven.com`, parsed by Claude Haiku 4.5, and written to the `purchase-orders` DynamoDB table. **Flow:** 1. Coupa sends a PO email (new, revision, or cancellation). 2. SES (`INBOUND_MAIL` rule set) drops the raw MIME into `s3://po-ingest-emails-{AccountId}/inbound/`. 3. S3 `ObjectCreated` triggers the `po-email-processor` Lambda. 4. Claude extracts structured JSON (PO number, status, supplier, site code, trade classification, line items, fiscal year). 5. Conditional write to DynamoDB: - `new_po` — idempotent insert (no-op if PO exists) - `revision` — unconditional overwrite - `cancellation` — marks existing row `Cancelled` 6. DynamoDB Streams feeds downstream consumers: - **LedgerFlow** (`seahaven-slack-bot/po-sync`) — daily KB sync - **Site extractor** (`po-ingest-site-extractor`) — real-time site address extraction into `verified-sites` table **Lambdas** (`lambdas/po/`): | Function | Trigger | Purpose | |---|---|---| | `po-email-processor` | S3 ObjectCreated | Claude extraction + DynamoDB write | | `po-ingest-site-extractor` | DynamoDB Streams | Site code/address extraction -> `verified-sites` | | `po-web-ui` | Function URL | HTML dashboard | **Tables:** - `purchase-orders` (PK: `po_number`, Streams: NEW_IMAGE) — shared with payments-dashboard and seahaven-slack-bot - `verified-sites` (PK: `siteCode`, GSI: `by-state`) — ~1,100 unique Amazon facility sites - `pending-site-review` (PK: `po_number`) — unresolvable POs for manual Payee Central verification ### Work Orders (`WorkorderIngestStack` stack) Amazon APM work order emails (from Hexagon EAM / HxGN SmartCloud) are received at `apm@int.seahaven.com`, parsed by Claude Haiku 4.5, and written to the `WorkOrders` DynamoDB table. **Flow:** 1. Hexagon EAM sends email notifications (new assignments, comments, updates, cancellations) to `amazon@seahavenind.com`. 2. Gmail filter forwards APM emails to `apm@int.seahaven.com` (SES). 3. SES drops the raw MIME into `s3://workorder-ingest-emails-{AccountId}/inbound/`. 4. S3 triggers the `workorder-email-processor` Lambda. 5. Claude extracts structured JSON (work order ID, site code, severity, priority, dates, assigned technician). 6. Work order upserted to `WorkOrders`, event/comment appended to `WorkOrderComments`. **Lambdas** (`lambdas/wo/`): | Function | Trigger | Purpose | |---|---|---| | `workorder-email-processor` | S3 ObjectCreated | Claude extraction + DynamoDB write | | `workorder-web-ui` | Function URL | HTML dashboard | **Tables:** - `WorkOrders` (PK: `work_order_id`, GSIs: `site-code-index`, `status-index`) - `WorkOrderComments` (PK: `work_order_id`, SK: `comment_id`) ## Architecture **IaC:** AWS CDK (Python), two stacks in one app, region `us-east-1`. All Lambdas: Python 3.12, ARM64, 60-day log retention. **Secrets:** - `po-ingest/anthropic-api-key` — Anthropic API key for PO parsing - `workorder-ingest/anthropic-api-key` — Anthropic API key for WO parsing **SES:** Both stacks add rules to the shared `INBOUND_MAIL` receipt rule set on `int.seahaven.com`. ## CI/CD GitHub Actions with reusable workflows from `Sea-Haven-Industries/.github`: - **CI** (PR to `main`): linting + `cdk synth` via `ci-python-sam.yaml@main` - **CD** (push to `main`): `cdk deploy --all` via `cd-cdk.yaml@main` (OIDC auth) Branch protection on `main` — all changes through PR. ## Setup 1. Bootstrap CDK: `cdk bootstrap aws://{AccountId}/us-east-1` 2. Store Anthropic API keys: ```bash aws secretsmanager create-secret --name po-ingest/anthropic-api-key --secret-string "sk-ant-..." aws secretsmanager create-secret --name workorder-ingest/anthropic-api-key --secret-string "sk-ant-..." ``` 3. Deploy both stacks: ```bash cd cdk pip install -r requirements.txt cdk deploy --all ``` 4. CloudFormation outputs include `WebUIUrl` for each stack's dashboard. ## Scripts **Reprocess PO emails** (re-run parser against all emails still in S3): ```bash python scripts/reprocess.py # dry-run python scripts/reprocess.py --execute # invoke po-email-processor for each ``` **Backfill verified sites** (one-time scan of historical POs): ```bash python scripts/backfill_sites.py ``` ## Directory Structure ``` cdk/ app.py # Two stacks: po-ingest + WorkorderIngestStack po_stack.py # Purchase order pipeline resources wo_stack.py # Work order pipeline resources lambdas/ po/ # PO pipeline Lambdas email_processor/ site_extractor/ web_ui/ wo/ # WO pipeline Lambdas email_processor/ web_ui/ shared/ models.py # Work order dataclasses/enums scripts/ reprocess.py backfill_sites.py ```