proposal-system/lambdas/pdf-generate/app.py

"""Proposal System - PDF Generate Lambda.

Generates professional branded proposal PDFs using reportlab Platypus.
Triggered via SQS when an admin requests PDF generation.
"""

import json
import logging
import os
import time
from datetime import datetime
from io import BytesIO

import boto3
import httpx
from reportlab.lib import colors
from reportlab.lib.enums import TA_CENTER, TA_RIGHT
from reportlab.lib.pagesizes import letter
from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet
from reportlab.lib.units import inch
from reportlab.platypus import (
    Paragraph,
    SimpleDocTemplate,
    Spacer,
    Table,
    TableStyle,
)

logger = logging.getLogger(__name__)
logger.setLevel(os.environ.get("LOG_LEVEL", "INFO"))

GENERATED_BUCKET = os.environ.get("GENERATED_BUCKET", "")
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")
secrets_client = boto3.client("secretsmanager")

_cached_api_key: str | None = None

COMPANY_NAME = "Sea Haven Industries"
COMPANY_ADDRESS = "Sea Haven Industries LLC"
COMPANY_PHONE = ""
COMPANY_EMAIL = "info@seahavenind.com"

TERMS_AND_CONDITIONS = """
1. This proposal is valid for 30 days from the date of issue.
2. Payment terms: Net 30 days from invoice date.
3. Any changes to the scope of work may result in additional charges.
4. Work will be scheduled upon acceptance of this proposal.
5. All materials and workmanship are guaranteed for one (1) year from completion.
6. Client is responsible for providing access to the work area.
7. This proposal does not include permits unless specifically noted in the line items.
""".strip()


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):
    for record in event.get("Records", []):
        body = json.loads(record["body"])
        payload = body.get("payload", body)
        proposal_id = payload["proposalId"]
        generate_pdf(proposal_id)
    return {"statusCode": 200}


def generate_pdf(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)

    pdf_bytes = build_pdf(proposal, line_items)

    proposal_number = proposal["proposalNumber"]
    revision = proposal.get("currentRevision", 1)
    s3_key = f"{proposal_number}/rev-{revision}.pdf"

    upload_pdf(s3_key, pdf_bytes)

    register_pdf(proposal_id, s3_key)

    logger.info("Generated PDF: %s (%d bytes)", s3_key, len(pdf_bytes))


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 build_pdf(proposal: dict, line_items: list[dict]) -> bytes:
    buffer = BytesIO()

    doc = SimpleDocTemplate(
        buffer,
        pagesize=letter,
        leftMargin=0.75 * inch,
        rightMargin=0.75 * inch,
        topMargin=0.75 * inch,
        bottomMargin=0.75 * inch,
    )

    styles = _get_styles()
    elements = []

    # Header
    elements.extend(_build_header(proposal, styles))
    elements.append(Spacer(1, 0.3 * inch))

    # Proposal metadata
    elements.extend(_build_metadata(proposal, styles))
    elements.append(Spacer(1, 0.3 * inch))

    # Scope of work
    elements.extend(_build_scope(proposal, styles))
    elements.append(Spacer(1, 0.3 * inch))

    # Line items table
    elements.extend(_build_line_items_table(line_items, styles))
    elements.append(Spacer(1, 0.4 * inch))

    # Terms and conditions
    elements.extend(_build_terms(styles))

    doc.build(elements, onFirstPage=_page_footer, onLaterPages=_page_footer)

    return buffer.getvalue()


def _get_styles():
    styles = getSampleStyleSheet()

    styles.add(
        ParagraphStyle(
            "CompanyName",
            parent=styles["Heading1"],
            fontSize=18,
            leading=22,
            textColor=colors.HexColor("#1a237e"),
            spaceAfter=2,
        )
    )

    styles.add(
        ParagraphStyle(
            "CompanyInfo",
            parent=styles["Normal"],
            fontSize=9,
            leading=12,
            textColor=colors.HexColor("#555555"),
        )
    )

    styles.add(
        ParagraphStyle(
            "ProposalTitle",
            parent=styles["Heading2"],
            fontSize=14,
            leading=18,
            textColor=colors.HexColor("#1a237e"),
            spaceBefore=6,
            spaceAfter=12,
        )
    )

    styles.add(
        ParagraphStyle(
            "SectionHeader",
            parent=styles["Heading3"],
            fontSize=11,
            leading=14,
            textColor=colors.HexColor("#1a237e"),
            spaceBefore=8,
            spaceAfter=6,
            borderWidth=0,
        )
    )

    styles.add(
        ParagraphStyle(
            "MetaLabel",
            parent=styles["Normal"],
            fontSize=9,
            leading=12,
            textColor=colors.HexColor("#666666"),
        )
    )

    styles.add(
        ParagraphStyle(
            "MetaValue",
            parent=styles["Normal"],
            fontSize=10,
            leading=13,
            fontName="Helvetica-Bold",
        )
    )

    styles.add(
        ParagraphStyle(
            "ScopeText",
            parent=styles["Normal"],
            fontSize=10,
            leading=14,
            spaceBefore=4,
        )
    )

    styles.add(
        ParagraphStyle(
            "TermsText",
            parent=styles["Normal"],
            fontSize=8,
            leading=11,
            textColor=colors.HexColor("#555555"),
        )
    )

    styles.add(
        ParagraphStyle(
            "TotalLabel",
            parent=styles["Normal"],
            fontSize=11,
            leading=14,
            fontName="Helvetica-Bold",
            alignment=TA_RIGHT,
        )
    )

    styles.add(
        ParagraphStyle(
            "FooterText",
            parent=styles["Normal"],
            fontSize=8,
            leading=10,
            textColor=colors.HexColor("#888888"),
            alignment=TA_CENTER,
        )
    )

    return styles


def _build_header(proposal: dict, styles) -> list:
    revision = proposal.get("currentRevision", 1)
    revision_text = f"  |  Rev {revision}" if revision > 1 else ""

    header_data = [
        [
            Paragraph(COMPANY_NAME, styles["CompanyName"]),
            Paragraph(f"PROPOSAL{revision_text}", styles["ProposalTitle"]),
        ],
        [
            Paragraph(f"{COMPANY_EMAIL}", styles["CompanyInfo"]),
            Paragraph(f"#{proposal['proposalNumber']}", styles["MetaValue"]),
        ],
    ]

    header_table = Table(header_data, colWidths=[3.5 * inch, 3.5 * inch])
    header_table.setStyle(
        TableStyle(
            [
                ("VALIGN", (0, 0), (-1, -1), "TOP"),
                ("ALIGN", (1, 0), (1, -1), "RIGHT"),
                ("LINEBELOW", (0, -1), (-1, -1), 1.5, colors.HexColor("#1a237e")),
                ("BOTTOMPADDING", (0, -1), (-1, -1), 8),
            ]
        )
    )

    return [header_table]


def _build_metadata(proposal: dict, styles) -> list:
    submitted_at = proposal.get("submittedAt", "")
    if submitted_at:
        try:
            dt = datetime.fromisoformat(submitted_at.replace("Z", "+00:00"))
            submitted_at = dt.strftime("%B %d, %Y")
        except (ValueError, TypeError):
            pass

    approved_at = proposal.get("approvedAt", "")
    if approved_at:
        try:
            dt = datetime.fromisoformat(approved_at.replace("Z", "+00:00"))
            approved_at = dt.strftime("%B %d, %Y")
        except (ValueError, TypeError):
            pass

    meta_data = [
        [
            Paragraph("Customer", styles["MetaLabel"]),
            Paragraph("Site Address", styles["MetaLabel"]),
        ],
        [
            Paragraph(proposal.get("customerName", ""), styles["MetaValue"]),
            Paragraph(proposal.get("customerAddress", ""), styles["MetaValue"]),
        ],
        [
            Paragraph("Work Order #", styles["MetaLabel"]),
            Paragraph("Date", styles["MetaLabel"]),
        ],
        [
            Paragraph(proposal.get("workOrderNumber", ""), styles["MetaValue"]),
            Paragraph(approved_at or submitted_at, styles["MetaValue"]),
        ],
        [
            Paragraph("Category", styles["MetaLabel"]),
            Paragraph("Priority", styles["MetaLabel"]),
        ],
        [
            Paragraph(proposal.get("serviceCategory", ""), styles["MetaValue"]),
            Paragraph(proposal.get("priority", ""), styles["MetaValue"]),
        ],
    ]

    meta_table = Table(meta_data, colWidths=[3.5 * inch, 3.5 * inch])
    meta_table.setStyle(
        TableStyle(
            [
                ("VALIGN", (0, 0), (-1, -1), "TOP"),
                ("TOPPADDING", (0, 0), (-1, -1), 2),
                ("BOTTOMPADDING", (0, 0), (-1, -1), 2),
            ]
        )
    )

    return [meta_table]


def _build_scope(proposal: dict, styles) -> list:
    scope = proposal.get("refinedScope") or proposal.get("scopeOfWork", "")
    if not scope:
        return []

    return [
        Paragraph("Scope of Work", styles["SectionHeader"]),
        Paragraph(scope, styles["ScopeText"]),
    ]


def _build_line_items_table(line_items: list[dict], styles) -> list:
    if not line_items:
        return [Paragraph("No line items", styles["Normal"])]

    elements = [Paragraph("Itemized Pricing", styles["SectionHeader"])]

    header = ["#", "Description", "Qty", "Unit", "Unit Price", "Total"]

    table_data = [header]
    subtotal = 0.0

    for i, li in enumerate(line_items, 1):
        qty = li.get("quantity", "")
        unit = li.get("unit", "")
        unit_price = li.get("unitPrice")
        total_price = li.get("totalPrice", 0)
        pricing_mode = li.get("pricingMode", "TotalPrice")

        subtotal += float(total_price or 0)

        if pricing_mode == "TotalPrice":
            up_str = "-"
        elif unit_price is not None:
            up_str = f"${float(unit_price):,.2f}"
        else:
            up_str = "-"

        tp_str = f"${float(total_price):,.2f}" if total_price else "-"

        row = [
            str(i),
            Paragraph(li.get("description", ""), styles["Normal"]),
            str(qty) if qty else "",
            unit,
            up_str,
            tp_str,
        ]
        table_data.append(row)

    col_widths = [
        0.35 * inch,
        3.15 * inch,
        0.55 * inch,
        0.7 * inch,
        1.0 * inch,
        1.0 * inch,
    ]
    table = Table(table_data, colWidths=col_widths, repeatRows=1)

    table.setStyle(
        TableStyle(
            [
                # Header row
                ("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#1a237e")),
                ("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
                ("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
                ("FONTSIZE", (0, 0), (-1, 0), 9),
                ("BOTTOMPADDING", (0, 0), (-1, 0), 6),
                ("TOPPADDING", (0, 0), (-1, 0), 6),
                # Data rows
                ("FONTSIZE", (0, 1), (-1, -1), 9),
                ("TOPPADDING", (0, 1), (-1, -1), 4),
                ("BOTTOMPADDING", (0, 1), (-1, -1), 4),
                ("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
                # Alignment
                ("ALIGN", (0, 0), (0, -1), "CENTER"),
                ("ALIGN", (2, 0), (2, -1), "CENTER"),
                ("ALIGN", (3, 0), (3, -1), "CENTER"),
                ("ALIGN", (4, 0), (4, -1), "RIGHT"),
                ("ALIGN", (5, 0), (5, -1), "RIGHT"),
                # Grid
                ("LINEBELOW", (0, 0), (-1, 0), 1, colors.HexColor("#1a237e")),
                ("LINEBELOW", (0, 1), (-1, -2), 0.5, colors.HexColor("#e0e0e0")),
                ("LINEBELOW", (0, -1), (-1, -1), 1, colors.HexColor("#1a237e")),
                # Alternating row colors
                *[
                    ("BACKGROUND", (0, i), (-1, i), colors.HexColor("#f5f5f5"))
                    for i in range(2, len(table_data), 2)
                ],
            ]
        )
    )

    elements.append(table)
    elements.append(Spacer(1, 0.15 * inch))

    # Total row
    total_data = [
        ["", "", "", "", "TOTAL:", f"${subtotal:,.2f}"],
    ]
    total_table = Table(total_data, colWidths=col_widths)
    total_table.setStyle(
        TableStyle(
            [
                ("FONTNAME", (0, 0), (-1, -1), "Helvetica-Bold"),
                ("FONTSIZE", (0, 0), (-1, -1), 11),
                ("ALIGN", (4, 0), (4, 0), "RIGHT"),
                ("ALIGN", (5, 0), (5, 0), "RIGHT"),
                ("TOPPADDING", (0, 0), (-1, -1), 4),
                ("LINEABOVE", (4, 0), (5, 0), 1.5, colors.HexColor("#1a237e")),
            ]
        )
    )
    elements.append(total_table)

    return elements


def _build_terms(styles) -> list:
    elements = [
        Spacer(1, 0.2 * inch),
        Paragraph("Terms & Conditions", styles["SectionHeader"]),
    ]

    for line in TERMS_AND_CONDITIONS.split("\n"):
        elements.append(Paragraph(line, styles["TermsText"]))

    return elements


def _page_footer(canvas, doc):
    canvas.saveState()
    page_num = canvas.getPageNumber()
    footer_text = f"Page {page_num}"
    canvas.setFont("Helvetica", 8)
    canvas.setFillColor(colors.HexColor("#888888"))
    canvas.drawCentredString(letter[0] / 2, 0.4 * inch, footer_text)
    canvas.drawString(
        0.75 * inch,
        0.4 * inch,
        f"{COMPANY_NAME} — Confidential",
    )
    canvas.restoreState()


def upload_pdf(s3_key: str, pdf_bytes: bytes):
    try:
        s3.put_object(
            Bucket=GENERATED_BUCKET,
            Key=s3_key,
            Body=pdf_bytes,
            ContentType="application/pdf",
        )
    except Exception as e:
        logger.error("Error uploading PDF: %s", e)
        raise


def register_pdf(proposal_id: str, s3_key: str):
    try:
        resp = _retry_request(
            "POST",
            f"{API_BASE_URL}/api/generated-pdfs",
            json={"proposalId": proposal_id, "s3Key": s3_key},
            headers=_api_headers(),
        )
        if resp.status_code not in (200, 201):
            logger.error("Failed to register PDF: %s %s", resp.status_code, resp.text)
    except Exception as e:
        logger.error("Error registering PDF: %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]