"""
Web UI Lambda.
Serves a simple HTML dashboard for viewing purchase orders.
Accessed via Lambda Function URL.
"""
import json
import os
from decimal import Decimal
import boto3
dynamodb = boto3.resource("dynamodb")
PO_TABLE = os.environ.get("PO_TABLE", "purchase-orders")
def get_purchase_orders(limit=500):
table = dynamodb.Table(PO_TABLE)
items = []
response = table.scan()
items.extend(response.get("Items", []))
while "LastEvaluatedKey" in response:
response = table.scan(ExclusiveStartKey=response["LastEvaluatedKey"])
items.extend(response.get("Items", []))
items.sort(key=lambda x: x.get("processed_at", ""), reverse=True)
return items[:limit]
def render_badge(value, color_map):
if not value:
value = "unknown"
color = color_map.get(value.lower(), "#9ca3af")
label = value.replace("_", " ").title()
return f'{label}'
STATUS_COLORS = {
"issued": "#3b82f6",
"pending buyer action": "#f59e0b",
"open": "#3b82f6",
"closed": "#10b981",
"cancelled": "#ef4444",
"soft closed": "#6b7280",
}
EMAIL_TYPE_COLORS = {
"new_po": "#3b82f6",
"revision": "#f59e0b",
"cancellation": "#ef4444",
}
def fmt_currency(val):
if val is None:
return ""
if isinstance(val, Decimal):
val = float(val)
if isinstance(val, (int, float)):
return f"${val:,.2f}"
return str(val)
def render_po_detail(po):
po_number = po.get("po_number", "")
fields = [
("PO Number", po_number),
("Status", render_badge(po.get("po_status", ""), STATUS_COLORS)),
("Email Type", render_badge(po.get("email_type", ""), EMAIL_TYPE_COLORS)),
("Total Amount", fmt_currency(po.get("total_amount"))),
("Currency", po.get("currency")),
("Supplier", (po.get("supplier") or {}).get("name")),
("Site Code", po.get("site_code")),
("State", po.get("state")),
("Trade", po.get("trade")),
("Fiscal Year", po.get("fiscal_year")),
("Coupa Category", po.get("coupa_category")),
("Submitted By", po.get("submitted_by")),
("On Behalf Of", po.get("on_behalf_of")),
("Order Date", po.get("order_date")),
("Revision Date", po.get("revision_date")),
("Payment Terms", po.get("payment_terms")),
("Requisition #", po.get("requisition_number")),
("Department", po.get("department")),
("Data Source", po.get("data_source")),
("Processed At", po.get("processed_at")),
]
ship_to = po.get("ship_to") or {}
if any(ship_to.values()):
ship_parts = []
if ship_to.get("name"):
ship_parts.append(ship_to["name"])
if ship_to.get("address"):
ship_parts.append(ship_to["address"])
if ship_to.get("location_code"):
ship_parts.append(f"Location: {ship_to['location_code']}")
if ship_to.get("attn"):
ship_parts.append(f"Attn: {ship_to['attn']}")
fields.append(("Ship To", "
".join(ship_parts)))
view_url = po.get("view_order_url")
if view_url:
fields.append(("Coupa Link", f'View in Coupa'))
details_html = ""
for label, value in fields:
if value:
details_html += f"""
"""
# Line items
line_items = po.get("line_items") or []
items_html = ""
if line_items:
rows = ""
for item in line_items:
qty = item.get('quantity', '') or ''
unit = item.get('unit', '') or ''
price = item.get('price', '') or ''
rows += f"""
| {item.get('description', '')} |
{qty} |
{unit} |
{price} |
{fmt_currency(item.get('amount'))} |
{item.get('need_by', '') or ''} |
"""
items_html = f"""
Line Items ({len(line_items)})
| Description |
Qty |
Unit |
Price |
Amount |
Need By |
{rows}
"""
return f"""
PO {po_number} - Sea Haven
Purchase Order {po_number}
{details_html}
{items_html}
"""
def render_po_list(purchase_orders):
rows = ""
for po in purchase_orders:
po_number = po.get("po_number", "")
supplier = (po.get("supplier") or {}).get("name", "")
site_code = po.get("site_code", "")
trade = po.get("trade", "")
status = po.get("po_status", "")
total = fmt_currency(po.get("total_amount"))
processed = (po.get("processed_at") or "")[:16]
rows += f"""
| {po_number} |
{supplier} |
{site_code} |
{trade} |
{render_badge(status, STATUS_COLORS)} |
{total} |
{processed} |
"""
return f"""
Purchase Orders - Sea Haven
Purchase Orders
{len(purchase_orders)} most recent
| PO # |
Supplier |
Site |
Trade |
Status |
Amount |
Processed |
{rows if rows else '| No purchase orders yet. |
'}
"""
def handler(event, context):
path = event.get("rawPath", "/")
qs = event.get("queryStringParameters") or {}
if path == "/po" and "id" in qs:
po_number = qs["id"]
table = dynamodb.Table(PO_TABLE)
result = table.get_item(Key={"po_number": po_number})
po = result.get("Item")
if not po:
return {
"statusCode": 404,
"headers": {"Content-Type": "text/html"},
"body": "Purchase order not found
",
}
html = render_po_detail(po)
else:
purchase_orders = get_purchase_orders()
html = render_po_list(purchase_orders)
return {
"statusCode": 200,
"headers": {"Content-Type": "text/html"},
"body": html,
}