""" Web UI Lambda. Serves a simple HTML dashboard for viewing work orders and comments. Accessed via Lambda Function URL. """ import json import logging import os from html import escape as esc import boto3 from web_ui_auth import is_authenticated logger = logging.getLogger() logger.setLevel(logging.INFO) dynamodb = boto3.resource("dynamodb") WORK_ORDERS_TABLE = os.environ.get("WORK_ORDERS_TABLE", "WorkOrders") COMMENTS_TABLE = os.environ.get("COMMENTS_TABLE", "WorkOrderComments") # Defense-in-depth auth gate. The public Function URL was removed (INFRA-74), but # the handler must still refuse unauthenticated requests so any future invocation # path does not re-expose the whole WO DB. Callers must present the shared secret # in the X-Auth-Token header (or Authorization: Bearer ). The secret is # fetched at runtime from Secrets Manager to keep it out of CloudFormation # templates and Lambda env vars. If the ARN is unset or the secret is missing, the # handler fails closed and denies every request. def get_work_orders(limit=500): table = dynamodb.Table(WORK_ORDERS_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("updated_at", ""), reverse=True) return items[:limit] def get_comments(work_order_id): table = dynamodb.Table(COMMENTS_TABLE) items = [] kwargs = { "KeyConditionExpression": "work_order_id = :woid", "ExpressionAttributeValues": {":woid": work_order_id}, } response = table.query(**kwargs) items.extend(response.get("Items", [])) while "LastEvaluatedKey" in response: response = table.query(**kwargs, ExclusiveStartKey=response["LastEvaluatedKey"]) items.extend(response.get("Items", [])) items.sort(key=lambda x: x.get("created_at", ""), reverse=True) return items def render_badge(value, color_map): if not value: value = "unknown" color = color_map.get(value.lower(), "#9ca3af") label = esc(value.replace("_", " ").title()) return f'{label}' STATUS_COLORS = { "new": "#3b82f6", "assigned": "#8b5cf6", "in_progress": "#f59e0b", "on_hold": "#6b7280", "completed": "#10b981", "cancelled": "#ef4444", "unknown": "#9ca3af", } RECORD_TYPE_COLORS = { "new_work_order": "#3b82f6", "comment": "#8b5cf6", "update": "#f59e0b", "cancellation": "#ef4444", "unknown": "#9ca3af", } def render_work_order_detail(wo, events): events_html = "" if events: for e in events: record_type = e.get("record_type", "unknown") commenter = esc(e.get("commenter", "")) created = esc(e.get("created_at", "")) text = esc(e.get("text", "")) badge = render_badge(record_type, RECORD_TYPE_COLORS) commenter_str = ( f"{commenter} — " if commenter else "" ) border_colors = { "new_work_order": "#3b82f6", "comment": "#8b5cf6", "update": "#f59e0b", "cancellation": "#ef4444", } border = border_colors.get(record_type, "#94a3b8") events_html += f"""
{badge} {commenter_str}{created}
{"
" + text + "
" if text else ""}
""" else: events_html = '

No events yet.

' wo_id = esc(wo.get("work_order_id", "")) fields = [ ("Description", esc(wo.get("description", "")) or None), ("Status", render_badge(wo.get("wo_status", "unknown"), STATUS_COLORS)), ( "Record Type", render_badge(wo.get("record_type", "unknown"), RECORD_TYPE_COLORS), ), ("Site Code", esc(wo.get("site_code", "")) or None), ("Building", esc(wo.get("building", "")) or None), ("Address", esc(wo.get("address", "")) or None), ("Severity", esc(wo.get("severity", "")) or None), ("Priority", esc(wo.get("priority", "")) or None), ("Due Date", esc(wo.get("due_date", "")) or None), ("Date Reported", esc(wo.get("date_reported", "")) or None), ("Scheduled Start", esc(wo.get("scheduled_start", "")) or None), ("Assigned To", esc(wo.get("assigned_to", "")) or None), ("Created", esc(wo.get("created_at", "")) or None), ("Last Updated", esc(wo.get("updated_at", "")) or None), ] details_html = "" for label, value in fields: if value: details_html += f"""
{label}
{value}
""" return f""" WO {wo_id} - Sea Haven
← All Work Orders

Work Order {wo_id}

{details_html}

Events ({len(events)})

{events_html}
""" def render_work_orders_list(work_orders): rows = "" for wo in work_orders: wo_id = esc(wo.get("work_order_id", "")) desc = esc(wo.get("description", "")) site = esc(wo.get("site_code", "")) status = wo.get("wo_status", "unknown") record_type = wo.get("record_type", "unknown") due = esc(wo.get("due_date", "")) updated = esc((wo.get("updated_at") or "")[:16]) rows += f""" {wo_id} {desc} {site} {render_badge(record_type, RECORD_TYPE_COLORS)} {render_badge(status, STATUS_COLORS)} {due} {updated} """ return f""" Work Orders - Sea Haven

Work Orders

{len(work_orders)} total
{rows if rows else ''}
WO # Description Site Last Action Status Due Date Updated
No work orders yet.
""" def handler(event, context): if not is_authenticated(event): return { "statusCode": 401, "headers": {"Content-Type": "text/html"}, "body": "

401 Unauthorized

", } path = event.get("rawPath", "/") qs = event.get("queryStringParameters") or {} if path == "/wo" and "id" in qs: wo_id = qs["id"] # Get work order table = dynamodb.Table(WORK_ORDERS_TABLE) result = table.get_item(Key={"work_order_id": wo_id}) wo = result.get("Item") if not wo: return { "statusCode": 404, "headers": {"Content-Type": "text/html"}, "body": "

Work order not found

", } events = get_comments(wo_id) html = render_work_order_detail(wo, events) else: work_orders = get_work_orders() html = render_work_orders_list(work_orders) return { "statusCode": 200, "headers": {"Content-Type": "text/html"}, "body": html, }