"""Proposal PDF generator using ReportLab. Generates a formal cemetery proposal PDF matching the letterhead layout from the INDL-27 PRD. Uploads to S3 under proposals/{tenant_id}/{quote_number}.pdf. Falls back gracefully when S3 credentials are absent. """ from __future__ import annotations import asyncio import io import logging from decimal import Decimal from typing import TYPE_CHECKING, Optional if TYPE_CHECKING: from sqlalchemy.ext.asyncio import AsyncSession from src.apps.sales.models.proposal import Proposal logger = logging.getLogger(__name__) def _build_proposal_pdf_bytes(proposal: "Proposal", tenant_name: str, sales_owner_name: str) -> bytes: """Build proposal PDF bytes using ReportLab.""" from reportlab.lib import colors from reportlab.lib.pagesizes import A4 from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle from reportlab.lib.units import cm from reportlab.platypus import ( HRFlowable, Paragraph, SimpleDocTemplate, Spacer, Table, TableStyle, ) buf = io.BytesIO() doc = SimpleDocTemplate( buf, pagesize=A4, rightMargin=2 * cm, leftMargin=2 * cm, topMargin=2 * cm, bottomMargin=2 * cm, ) PRIMARY = colors.HexColor("#1a4a6e") styles = getSampleStyleSheet() body = ParagraphStyle("Body", parent=styles["Normal"], fontSize=10, leading=14) small = ParagraphStyle("Small", parent=styles["Normal"], fontSize=8, leading=12, textColor=colors.HexColor("#6B7280")) heading = ParagraphStyle("Heading", parent=styles["Normal"], fontSize=11, fontName="Helvetica-Bold", textColor=PRIMARY) title_style = ParagraphStyle("Title", parent=styles["Normal"], fontSize=18, fontName="Helvetica-Bold", textColor=PRIMARY, spaceAfter=4) from datetime import timezone sent = proposal.sent_at if sent and sent.tzinfo is None: sent = sent.replace(tzinfo=timezone.utc) from datetime import timedelta valid_until = None if sent: valid_until = sent + timedelta(days=proposal.valid_days) issued_str = sent.strftime("%B %d, %Y") valid_str = valid_until.strftime("%B %d, %Y") else: from datetime import datetime now = datetime.now(timezone.utc) issued_str = now.strftime("%B %d, %Y") valid_str = (now + timedelta(days=proposal.valid_days)).strftime("%B %d, %Y") story = [] # Letterhead story.append(Paragraph(tenant_name, title_style)) story.append(Paragraph(f"Quote {proposal.quote_number} · Issued {issued_str} · Valid until {valid_str}", small)) story.append(HRFlowable(width="100%", thickness=2, color=PRIMARY, spaceAfter=12)) story.append(Paragraph(proposal.subject or "Sales Proposal", heading)) story.append(Paragraph(f"Reference {proposal.quote_number} · {sales_owner_name}", small)) story.append(Spacer(1, 12)) # Prepared for story.append(Paragraph(f"Prepared for: {proposal.to_email}", body)) story.append(Spacer(1, 8)) # Cover note if proposal.cover_note: note_lines = proposal.cover_note.replace("&", "&").replace("<", "<").replace(">", ">") for line in note_lines.split("\n"): story.append(Paragraph(line or " ", body)) story.append(Spacer(1, 14)) # Quotation table story.append(HRFlowable(width="100%", thickness=0.5, color=colors.HexColor("#E5E7EB"), spaceAfter=6)) story.append(Paragraph("QUOTATION", heading)) story.append(Spacer(1, 4)) tdata = [["Item", "Qty", "Unit (CAD)", "Amount"]] for li in sorted(proposal.line_items, key=lambda x: x.sort_order): tdata.append([ li.description, str(int(li.quantity)), f"${float(li.unit_price):,.2f}", f"${float(li.line_total):,.2f}", ]) subtotal = float(proposal.subtotal) tax = float(proposal.tax_amount) total = float(proposal.total_amount) tdata.append(["", "", "Subtotal", f"${subtotal:,.2f}"]) tdata.append(["", "", "HST (13%)", f"${tax:,.2f}"]) tdata.append(["", "", "Total CAD", f"${total:,.2f}"]) col_widths = [9 * cm, 2 * cm, 3.5 * cm, 3.5 * cm] table = Table(tdata, colWidths=col_widths) table.setStyle(TableStyle([ ("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"), ("FONTSIZE", (0, 0), (-1, -1), 9), ("TEXTCOLOR", (0, 0), (-1, 0), PRIMARY), ("ALIGN", (1, 0), (-1, -1), "RIGHT"), ("LINEBELOW", (0, 0), (-1, 0), 0.5, colors.HexColor("#E5E7EB")), ("FONTNAME", (-2, -3), (-1, -1), "Helvetica-Bold"), ("FONTNAME", (-2, -1), (-1, -1), "Helvetica-Bold"), ("TEXTCOLOR", (-2, -1), (-1, -1), PRIMARY), ("LINEABOVE", (-2, -3), (-1, -3), 0.5, colors.HexColor("#E5E7EB")), ("LINEABOVE", (-2, -1), (-1, -1), 1.5, PRIMARY), ("ROWBACKGROUNDS", (0, 1), (-1, -4), [colors.white, colors.HexColor("#F9FAFB")]), ("TOPPADDING", (0, 0), (-1, -1), 6), ("BOTTOMPADDING", (0, 0), (-1, -1), 6), ])) story.append(table) story.append(Spacer(1, 14)) # Terms story.append(HRFlowable(width="100%", thickness=0.5, color=colors.HexColor("#E5E7EB"), spaceAfter=6)) story.append(Paragraph("TERMS", heading)) terms = [ "This quotation is non-binding and does not reserve the plot.", f"Valid for {proposal.valid_days} days from the issue date above.", "A 20% deposit secures the plot and converts this proposal into a binding contract.", "All sales subject to applicable cemetery by-laws and the Funeral, Burial and Cremation Services Act.", ] for t in terms: story.append(Paragraph(f"• {t}", body)) story.append(Spacer(1, 20)) # Signature blocks sig_data = [ ["Purchaser acceptance", "For " + tenant_name], ["", ""], ["__________________________", "__________________________"], ["Signature · Date", f"{sales_owner_name} · Date"], ] sig_table = Table(sig_data, colWidths=[9 * cm, 9 * cm]) sig_table.setStyle(TableStyle([ ("FONTSIZE", (0, 0), (-1, -1), 8), ("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"), ("TEXTCOLOR", (0, 0), (-1, 0), colors.HexColor("#6B7280")), ("TEXTCOLOR", (0, 3), (-1, 3), colors.HexColor("#6B7280")), ("TOPPADDING", (0, 0), (-1, -1), 4), ("BOTTOMPADDING", (0, 0), (-1, -1), 4), ])) story.append(sig_table) story.append(Spacer(1, 14)) story.append(HRFlowable(width="100%", thickness=0.5, color=colors.HexColor("#E5E7EB"), spaceAfter=4)) story.append(Paragraph("Generated by INDELIS Cemetery Management Platform · Page 1 of 1", small)) doc.build(story) return buf.getvalue() async def generate_and_store_proposal_pdf( proposal: "Proposal", db: "AsyncSession", tenant_id: str, ) -> Optional[str]: """Generate proposal PDF and upload to S3. Returns S3 key or None.""" from src.core.config import settings # Resolve tenant name tenant_name = "Cemetery" try: from sqlalchemy import select as _select from src.apps.tenants.models.account import Account result = await db.execute(_select(Account).where(Account.id == tenant_id)) account = result.scalar_one_or_none() if account: tenant_name = account.organization_name or tenant_name except Exception: pass # Resolve sales owner name sales_owner_name = "Cemetery Admin" if proposal.sales_owner_id: try: from sqlalchemy import select as _select from src.apps.auth.models.user import User result = await db.execute(_select(User).where(User.id == proposal.sales_owner_id)) user = result.scalar_one_or_none() if user: sales_owner_name = f"{user.first_name} {user.last_name}".strip() or user.email except Exception: pass # Build PDF bytes in a thread (CPU-bound) pdf_bytes = await asyncio.to_thread( _build_proposal_pdf_bytes, proposal, tenant_name, sales_owner_name ) # Upload to S3 if not (settings.AWS_ACCESS_KEY_ID and settings.AWS_SECRET_ACCESS_KEY): logger.info("S3 not configured; proposal PDF not stored (quote_number=%s)", proposal.quote_number) return None s3_key = f"proposals/{tenant_id}/{proposal.quote_number}.pdf" def _upload() -> None: import boto3 s3 = boto3.client( "s3", region_name=getattr(settings, "AWS_REGION", "ca-central-1"), aws_access_key_id=settings.AWS_ACCESS_KEY_ID, aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY, ) s3.put_object( Bucket=settings.S3_BUCKET, Key=s3_key, Body=pdf_bytes, ContentType="application/pdf", ) await asyncio.to_thread(_upload) logger.info("Proposal PDF uploaded: %s", s3_key) return s3_key