"""Proposal System - Library Ingest Lambda. Processes approved/sent proposals into the Bedrock Knowledge Base library. Formats proposal data as structured markdown and uploads to the library bucket, then triggers a KB sync. """ import json import logging import os import time from datetime import datetime import boto3 import httpx logger = logging.getLogger(__name__) logger.setLevel(os.environ.get("LOG_LEVEL", "INFO")) LIBRARY_BUCKET = os.environ.get("LIBRARY_BUCKET", "") KNOWLEDGE_BASE_ID = os.environ.get("KNOWLEDGE_BASE_ID", "") DATA_SOURCE_ID = os.environ.get("DATA_SOURCE_ID", "") API_BASE_URL = os.environ.get("API_BASE_URL", "") INTERNAL_API_KEY_SECRET_ARN = os.environ.get("INTERNAL_API_KEY_SECRET_ARN", "") s3 = boto3.client("s3") bedrock_agent = boto3.client("bedrock-agent") secrets_client = boto3.client("secretsmanager") _cached_api_key: str | None = None def _get_api_key() -> str: global _cached_api_key if _cached_api_key is None: if INTERNAL_API_KEY_SECRET_ARN: resp = secrets_client.get_secret_value(SecretId=INTERNAL_API_KEY_SECRET_ARN) _cached_api_key = resp["SecretString"] else: _cached_api_key = "" return _cached_api_key def handler(event, context): batch_item_failures = [] for record in event.get("Records", []): try: body = json.loads(record["body"]) payload = body.get("payload", body) proposal_id = payload["proposalId"] process_ingestion(proposal_id) except Exception as e: logger.error("Failed to process record %s: %s", record.get("messageId"), e) batch_item_failures.append({"itemIdentifier": record["messageId"]}) return {"batchItemFailures": batch_item_failures} def process_ingestion(proposal_id: str): proposal = fetch_proposal(proposal_id) if not proposal: logger.warning("Proposal %s not found", proposal_id) return line_items = fetch_line_items(proposal_id) document = format_proposal_document(proposal, line_items) s3_key = upload_to_library(proposal, document) if s3_key: trigger_kb_sync() def fetch_proposal(proposal_id: str) -> dict | None: try: resp = _retry_request( "GET", f"{API_BASE_URL}/api/proposals/{proposal_id}", headers=_api_headers(), ) if resp.status_code == 200: return resp.json() except Exception as e: logger.error("Error fetching proposal: %s", e) return None def fetch_line_items(proposal_id: str) -> list[dict]: try: resp = _retry_request( "GET", f"{API_BASE_URL}/api/proposals/{proposal_id}/line-items", headers=_api_headers(), ) if resp.status_code == 200: return resp.json() except Exception as e: logger.error("Error fetching line items: %s", e) return [] def format_proposal_document(proposal: dict, line_items: list[dict]) -> str: total = sum(li.get("totalPrice", 0) for li in line_items) submitted_at = proposal.get("submittedAt", "") if submitted_at: try: dt = datetime.fromisoformat(submitted_at.replace("Z", "+00:00")) submitted_at = dt.strftime("%Y-%m-%d") except (ValueError, TypeError): pass lines = [ f"# Proposal: {proposal['proposalNumber']}", "", f"**Customer:** {proposal['customerName']}", f"**Address:** {proposal.get('customerAddress', '')}", f"**Service Category:** {proposal['serviceCategory']}", f"**Priority:** {proposal['priority']}", f"**Date:** {submitted_at}", f"**Total Bid Amount:** ${total:,.2f}", f"**Work Order:** {proposal.get('workOrderNumber', '')}", "", "## Scope of Work", "", proposal.get("refinedScope") or proposal.get("scopeOfWork", ""), "", "## Line Items", "", "| # | Description | Qty | Unit | Unit Price | Total |", "|---|---|---|---|---|---|", ] for i, li in enumerate(line_items, 1): desc = li.get("description", "") qty = li.get("quantity", "") unit = li.get("unit", "") unit_price = li.get("unitPrice") total_price = li.get("totalPrice", 0) up_str = f"${unit_price:,.2f}" if unit_price else "-" tp_str = f"${total_price:,.2f}" lines.append(f"| {i} | {desc} | {qty} | {unit} | {up_str} | {tp_str} |") lines.extend( [ "", f"**Total: ${total:,.2f}**", ] ) return "\n".join(lines) def upload_to_library(proposal: dict, document: str) -> str | None: if not LIBRARY_BUCKET: logger.warning("No library bucket configured") return None proposal_number = proposal["proposalNumber"] category = proposal.get("serviceCategory", "General") s3_key = f"proposals/{category.lower()}/{proposal_number}.md" try: s3.put_object( Bucket=LIBRARY_BUCKET, Key=s3_key, Body=document.encode("utf-8"), ContentType="text/markdown", Metadata={ "service-category": category, "proposal-number": proposal_number, "customer-name": proposal.get("customerName", ""), "total-amount": str(proposal.get("totalBidAmount", 0)), "date-submitted": proposal.get("submittedAt", ""), }, ) logger.info("Uploaded %s to library bucket", s3_key) return s3_key except Exception as e: logger.error("Error uploading to library: %s", e) return None def trigger_kb_sync(): if not KNOWLEDGE_BASE_ID or not DATA_SOURCE_ID: logger.info("KB or data source ID not configured, skipping sync") return try: response = bedrock_agent.start_ingestion_job( knowledgeBaseId=KNOWLEDGE_BASE_ID, dataSourceId=DATA_SOURCE_ID, ) job_id = response.get("ingestionJob", {}).get("ingestionJobId", "") logger.info("Started KB ingestion job: %s", job_id) except Exception as e: logger.error("Error triggering KB sync: %s", e) def _api_headers() -> dict: headers = {"Content-Type": "application/json"} api_key = _get_api_key() if api_key: headers["X-Internal-Api-Key"] = api_key return headers def _retry_request( method: str, url: str, *, max_retries: int = 3, **kwargs ) -> httpx.Response: kwargs.setdefault("timeout", 10) for attempt in range(max_retries): try: resp = httpx.request(method, url, **kwargs) if resp.status_code < 500: return resp except (httpx.ConnectError, httpx.ReadTimeout, httpx.WriteTimeout) as exc: if attempt == max_retries - 1: raise logger.warning( "Retryable error (attempt %d/%d): %s", attempt + 1, max_retries, exc ) time.sleep(min(2**attempt, 4)) return resp # type: ignore[possibly-undefined]