# ─── Document Data Builders ────────────────────────────────────────────────────
#
# Field mapping and data construction per document type.
# Each build_*_data() function transforms raw submission data into structured
# document data suitable for rendering.
#
# ─── Imports ───────────────────────────────────────────────────────────────────
import json
import logging
import uuid
from datetime import datetime
logger = logging.getLogger("document")
# ─── Constants ─────────────────────────────────────────────────────────────────
DOCUMENT_TYPES = [
"invoice",
"quote",
"receipt",
"certificate",
"report",
"proposal",
"statement",
"work_order",
"delivery_note",
"contract",
]
TEMPLATE_LAYOUTS = [
# ── Invoice / Quote layouts (premium redesign) ──
"stripe", # minimalist, whitespace-driven, Stripe-inspired
"hubspot", # bold header band, color-blocked sections
"notion", # clean, editorial, sidebar accent
"modern", # contemporary flex layout
"professional", # clean single-accent design
"minimal", # ultra-simple
"bold", # high-contrast
"classic", # traditional layout
"receipt", # compact receipt format
"customer_portal", # customer-facing portal view
# ── Proposal layout ──
"proposal", # modern proposal with cover, scope, timeline, pricing, signatures
# ── Statement layout ──
"statement", # financial statement with transaction history and running balance
# ── Work order layout ──
"work_order", # service dispatch with technician info, tasks, materials, signatures
# ── Delivery note layout ──
"delivery_note", # clean packing slip with shipping label and delivery confirmation
# ── Contract layout ──
"contract", # formal agreement with numbered clauses and dual signatures
]
DOCUMENT_STATUSES = ["draft", "sent", "viewed", "paid", "void"]
# ─── Helpers ───────────────────────────────────────────────────────────────────
def _safe_float(value, default=0.0) -> float:
"""Safely parse a float from form data."""
try:
return float(value) if value is not None else default
except (ValueError, TypeError):
return default
def _build_line_items(data: dict, mapping: list) -> list:
"""Build line items from form data based on mapping config.
Mapping format:
[
{"description": "field_key", "quantity": "field_key", "unit_price": "field_key"},
{"description": "field_key", "quantity": "field_key", "unit_price": "field_key"},
]
Each mapping entry produces one line item. Amount is calculated server-side.
"""
line_items = []
for item_map in mapping:
description = data.get(item_map.get("description", ""), "")
quantity = _safe_float(data.get(item_map.get("quantity", "1"), "1"))
unit_price = _safe_float(data.get(item_map.get("unit_price", "0"), "0"))
amount = round(quantity * unit_price, 2)
if description or amount > 0:
line_items.append(
{
"description": description,
"quantity": quantity,
"unit_price": unit_price,
"amount": amount,
}
)
return line_items
# ─── Data Builders ─────────────────────────────────────────────────────────────
def build_invoice_data(
template: dict, submission: dict, line_items: list, currency: str = "USD", doc_type: str = "Invoice"
) -> dict:
"""Build invoice data structure from template, submission, and line items.
Returns a dict suitable for document rendering (HTML/PDF).
"""
from .document_renderer import DEFAULT_STYLE
template_id = template.get("id")
field_mapping = template.get("field_mapping", {})
style_config = template.get("style_config", DEFAULT_STYLE)
# Generate invoice number from template or default
invoice_number = submission.get("invoice_number")
if not invoice_number and template.get("next_invoice_number"):
invoice_number = template["next_invoice_number"]
if not invoice_number:
invoice_number = f"INV-{datetime.now().strftime('%Y')}-{int(datetime.now().timestamp())}"
# Generate document ID
document_id = str(uuid.uuid4())[:8]
# Build line items with calculations
processed_items = []
for item in line_items:
qty = float(item.get("quantity", item.get("qty", 1)))
price = float(item.get("unit_price", item.get("price", 0)))
total = qty * price
processed_items.append(
{
"description": item.get("description", "Item"),
"quantity": qty,
"unit_price": price,
"amount": total,
"total": total,
}
)
# Calculate totals
subtotal = sum(item["total"] for item in processed_items)
tax_rate = float(submission.get("tax_rate", 0))
tax_amount = subtotal * tax_rate / 100 if tax_rate else 0
total = subtotal + tax_amount
# Business info: check field_mapping first, then submission payload, then defaults
from_info = {
"name": field_mapping.get("business_name", "") or submission.get("business_name", ""),
"email": field_mapping.get("business_email", "") or submission.get("business_email", ""),
"phone": field_mapping.get("business_phone", "") or submission.get("business_phone", ""),
"address": field_mapping.get("business_address", "") or submission.get("business_address", ""),
}
result = {
"document_id": document_id,
"invoice_number": invoice_number,
"document_type": doc_type,
"currency": currency,
"issue_date": datetime.now().strftime("%B %d, %Y"),
"due_date": submission.get("due_date", ""),
"from": from_info,
"to": {
"name": submission.get("customer_name", submission.get("name", "Customer")),
"email": submission.get("customer_email", submission.get("email", "")),
"address": submission.get("customer_address", submission.get("address", "")),
},
"customer_name": submission.get("customer_name", submission.get("name", "Customer")),
"customer_email": submission.get("customer_email", submission.get("email", "")),
"customer_address": submission.get("customer_address", submission.get("address", "")),
"line_items": processed_items,
"subtotal": subtotal,
"tax_rate": tax_rate,
"tax_amount": tax_amount,
"tax": tax_amount,
"total": total,
"notes": submission.get("notes", ""),
"payment_terms": submission.get("payment_terms", ""),
"payment_link": submission.get("payment_link", ""),
"payment_link_text": submission.get("payment_link_text", "Pay Now"),
"style_config": style_config,
}
# Pass through additional fields for diverse document types
for field in [
"executive_summary", "scope", "scope_items", "timeline", "terms", "title",
"transactions", "opening_balance", "total_charges", "total_payments", "closing_balance",
"period_start", "period_end",
"tasks", "time_in", "time_out", "total_hours", "work_order_number", "service_address", "scheduled_date",
"po_number", "order_number", "carrier", "tracking_number", "weight",
"contract_title", "preamble", "terms_sections", "effective_date",
"customer_phone",
]:
if field in submission:
result[field] = submission[field]
# Ensure from/to have phone/address
if "from" in submission and isinstance(submission["from"], dict):
result["from"].update(submission["from"])
if "to" in submission and isinstance(submission["to"], dict):
result["to"].update(submission["to"])
return result
def build_proposal_data(
template: dict, submission: dict, line_items: list, currency: str = "USD"
) -> dict:
"""Build proposal data structure with timeline, pricing tiers, and signature blocks."""
field_mapping = template.get("field_mapping", {})
document_id = str(uuid.uuid4())[:8]
proposal_number = submission.get("proposal_number") or submission.get("document_number", "")
if not proposal_number:
proposal_number = f"PROP-{datetime.now().strftime('%Y%m%d')}-{int(datetime.now().timestamp()) % 10000:04d}"
# Build pricing tiers from line_items if structured, else use as single tier
pricing_tiers = []
if line_items:
if isinstance(line_items[0], dict) and "tier_name" in line_items[0]:
pricing_tiers = line_items
else:
# Standard line items → single "Standard" tier
processed = []
for item in line_items:
qty = float(item.get("quantity", item.get("qty", 1)))
price = float(item.get("unit_price", item.get("price", 0)))
processed.append({
"description": item.get("description", "Item"),
"quantity": qty,
"unit_price": price,
"amount": round(qty * price, 2),
})
subtotal = sum(i["amount"] for i in processed)
tax_rate = float(submission.get("tax_rate", 0))
tax = round(subtotal * tax_rate / 100, 2) if tax_rate else 0
pricing_tiers.append({
"tier_name": "Standard",
"line_items": processed,
"subtotal": subtotal,
"tax_rate": tax_rate,
"tax": tax,
"total": round(subtotal + tax, 2),
})
# Build timeline from submission or defaults
timeline = submission.get("timeline", [])
if not timeline:
timeline = []
return {
"document_id": document_id,
"document_number": proposal_number,
"document_type": "proposal",
"currency": currency,
"issue_date": submission.get("issue_date", datetime.now().strftime("%B %d, %Y")),
"valid_until": submission.get("valid_until", ""),
"from": {
"name": field_mapping.get("business_name", "") or submission.get("business_name", ""),
"email": field_mapping.get("business_email", "") or submission.get("business_email", ""),
"phone": field_mapping.get("business_phone", "") or submission.get("business_phone", ""),
"address": field_mapping.get("business_address", "") or submission.get("business_address", ""),
},
"to": {
"name": submission.get("customer_name", submission.get("name", "Client")),
"email": submission.get("customer_email", submission.get("email", "")),
"address": submission.get("customer_address", submission.get("address", "")),
},
"executive_summary": submission.get("executive_summary", submission.get("summary", "")),
"scope_of_work": submission.get("scope_of_work", []),
"line_items": line_items if pricing_tiers == [] else [],
"pricing_tiers": pricing_tiers,
"timeline": timeline,
"terms_conditions": submission.get("terms_conditions", submission.get("terms", "")),
"notes": submission.get("notes", ""),
"signatures": submission.get("signatures", {
"provider_name": field_mapping.get("business_name", ""),
"client_name": submission.get("customer_name", ""),
}),
"payment_terms": submission.get("payment_terms", ""),
"payment_link": submission.get("payment_link", ""),
"payment_link_text": submission.get("payment_link_text", "Accept Proposal"),
}
def build_statement_data(
template: dict, submission: dict, transactions: list, currency: str = "USD"
) -> dict:
"""Build account statement data with transaction history and running balance."""
field_mapping = template.get("field_mapping", {})
document_id = str(uuid.uuid4())[:8]
statement_number = submission.get("statement_number") or submission.get("document_number", "")
if not statement_number:
statement_number = f"STMT-{datetime.now().strftime('%Y%m')}-{int(datetime.now().timestamp()) % 1000:03d}"
# Calculate running balance from transactions
processed_transactions = []
running_balance = _safe_float(submission.get("opening_balance", 0))
for txn in transactions:
debit = _safe_float(txn.get("debit", 0))
credit = _safe_float(txn.get("credit", 0))
running_balance += credit - debit
processed_transactions.append({
"date": txn.get("date", ""),
"description": txn.get("description", ""),
"reference": txn.get("reference", ""),
"debit": debit,
"credit": credit,
"balance": round(running_balance, 2),
})
closing_balance = running_balance if processed_transactions else submission.get("opening_balance", 0)
return {
"document_id": document_id,
"document_number": statement_number,
"document_type": "statement",
"currency": currency,
"issue_date": submission.get("issue_date", datetime.now().strftime("%B %d, %Y")),
"period_start": submission.get("period_start", ""),
"period_end": submission.get("period_end", ""),
"from": {
"name": field_mapping.get("business_name", "") or submission.get("business_name", ""),
"email": field_mapping.get("business_email", "") or submission.get("business_email", ""),
"phone": field_mapping.get("business_phone", "") or submission.get("business_phone", ""),
"address": field_mapping.get("business_address", "") or submission.get("business_address", ""),
},
"to": {
"name": submission.get("customer_name", submission.get("name", "Customer")),
"email": submission.get("customer_email", submission.get("email", "")),
"address": submission.get("customer_address", submission.get("address", "")),
},
"account_number": submission.get("account_number", ""),
"opening_balance": _safe_float(submission.get("opening_balance", 0)),
"total_debits": sum(t["debit"] for t in processed_transactions),
"total_credits": sum(t["credit"] for t in processed_transactions),
"closing_balance": round(closing_balance, 2),
"transactions": processed_transactions,
"line_items": [], # Keep for compatibility
"subtotal": 0,
"tax_rate": 0,
"tax": 0,
"total": round(closing_balance, 2),
"notes": submission.get("notes", ""),
"payment_terms": submission.get("payment_terms", ""),
"payment_link": submission.get("payment_link", ""),
}
def build_work_order_data(
template: dict, submission: dict, line_items: list, currency: str = "USD"
) -> dict:
"""Build work order data with job details, technician info, and time tracking."""
field_mapping = template.get("field_mapping", {})
document_id = str(uuid.uuid4())[:8]
work_order_number = submission.get("work_order_number") or submission.get("document_number", "")
if not work_order_number:
work_order_number = f"WO-{datetime.now().strftime('%Y%m%d')}-{int(datetime.now().timestamp()) % 10000:04d}"
# Process tasks/checklist
tasks = submission.get("tasks", [])
materials = submission.get("materials", [])
if not materials and line_items:
# Use line items as materials
materials = [{
"description": item.get("description", ""),
"quantity": float(item.get("quantity", item.get("qty", 1))),
"unit_price": float(item.get("unit_price", item.get("price", 0))),
"amount": round(float(item.get("quantity", item.get("qty", 1))) * float(item.get("unit_price", item.get("price", 0))), 2),
} for item in line_items]
# Technician info
technician = submission.get("technician", {
"name": "",
"phone": "",
"license": "",
})
return {
"document_id": document_id,
"document_number": work_order_number,
"document_type": "work_order",
"currency": currency,
"issue_date": submission.get("issue_date", datetime.now().strftime("%B %d, %Y")),
"scheduled_date": submission.get("scheduled_date", ""),
"priority": submission.get("priority", "Normal"),
"job_category": submission.get("job_category", ""),
"from": {
"name": field_mapping.get("business_name", "") or submission.get("business_name", ""),
"email": field_mapping.get("business_email", "") or submission.get("business_email", ""),
"phone": field_mapping.get("business_phone", "") or submission.get("business_phone", ""),
"address": field_mapping.get("business_address", "") or submission.get("business_address", ""),
},
"to": {
"name": submission.get("customer_name", submission.get("name", "Customer")),
"email": submission.get("customer_email", submission.get("email", "")),
"phone": submission.get("customer_phone", ""),
"address": submission.get("customer_address", submission.get("address", "")),
},
"job_description": submission.get("job_description", submission.get("description", "")),
"technician": technician,
"tasks": tasks,
"materials": materials,
"time_in": submission.get("time_in", ""),
"time_out": submission.get("time_out", ""),
"labor_hours": submission.get("labor_hours", ""),
"labor_rate": submission.get("labor_rate", 0),
"subtotal": submission.get("subtotal", 0),
"tax_rate": float(submission.get("tax_rate", 0)),
"tax": submission.get("tax", 0),
"total": submission.get("total", 0),
"line_items": line_items,
"notes": submission.get("notes", ""),
"signature_required": submission.get("signature_required", True),
"customer_signature": submission.get("customer_signature", ""),
}
def build_delivery_note_data(
template: dict, submission: dict, line_items: list, currency: str = "USD"
) -> dict:
"""Build delivery note / packing slip data."""
field_mapping = template.get("field_mapping", {})
document_id = str(uuid.uuid4())[:8]
delivery_number = submission.get("delivery_number") or submission.get("document_number", "")
if not delivery_number:
delivery_number = f"DN-{datetime.now().strftime('%Y%m%d')}-{int(datetime.now().timestamp()) % 10000:04d}"
# Process items
processed_items = []
for item in line_items:
processed_items.append({
"description": item.get("description", "Item"),
"sku": item.get("sku", item.get("item_code", "")),
"quantity": float(item.get("quantity", item.get("qty", 1))),
"unit": item.get("unit", "ea"),
"weight": item.get("weight", ""),
})
return {
"document_id": document_id,
"document_number": delivery_number,
"document_type": "delivery_note",
"currency": currency,
"issue_date": submission.get("issue_date", datetime.now().strftime("%B %d, %Y")),
"delivery_date": submission.get("delivery_date", ""),
"po_number": submission.get("po_number", ""),
"reference": submission.get("reference", ""),
"from": {
"name": field_mapping.get("business_name", "") or submission.get("business_name", ""),
"email": field_mapping.get("business_email", "") or submission.get("business_email", ""),
"phone": field_mapping.get("business_phone", "") or submission.get("business_phone", ""),
"address": field_mapping.get("business_address", "") or submission.get("business_address", ""),
},
"to": {
"name": submission.get("customer_name", submission.get("name", "Customer")),
"email": submission.get("customer_email", submission.get("email", "")),
"phone": submission.get("customer_phone", ""),
"address": submission.get("customer_address", submission.get("address", "")),
},
"delivery_address": submission.get("delivery_address", submission.get("customer_address", "")),
"carrier": submission.get("carrier", ""),
"tracking_number": submission.get("tracking_number", ""),
"items": processed_items,
"total_items": sum(item["quantity"] for item in processed_items),
"total_weight": submission.get("total_weight", ""),
"line_items": line_items,
"notes": submission.get("notes", ""),
"signature_required": True,
"received_by": submission.get("received_by", ""),
"received_date": submission.get("received_date", ""),
"received_condition": submission.get("received_condition", ""),
}
def build_contract_data(
template: dict, submission: dict, line_items: list = None, currency: str = "USD"
) -> dict:
"""Build contract/agreement data with parties, clauses, and signatures."""
field_mapping = template.get("field_mapping", {})
document_id = str(uuid.uuid4())[:8]
contract_number = submission.get("contract_number") or submission.get("document_number", "")
if not contract_number:
contract_number = f"CTR-{datetime.now().strftime('%Y%m%d')}-{int(datetime.now().timestamp()) % 10000:04d}"
# Parse clauses from submission
clauses = submission.get("clauses", [])
if not clauses and submission.get("terms_conditions"):
# Split terms into clauses
raw = submission["terms_conditions"]
clauses = [{"title": f"Clause {i+1}", "content": text.strip()}
for i, text in enumerate(raw.split("\n")) if text.strip()]
# Party info
party_a = {
"name": field_mapping.get("business_name", "") or submission.get("business_name", ""),
"email": field_mapping.get("business_email", "") or submission.get("business_email", ""),
"phone": field_mapping.get("business_phone", "") or submission.get("business_phone", ""),
"address": field_mapping.get("business_address", "") or submission.get("business_address", ""),
"title": submission.get("party_a_title", "Provider"),
}
party_b = {
"name": submission.get("customer_name", submission.get("party_b_name", "")),
"email": submission.get("customer_email", submission.get("party_b_email", "")),
"phone": submission.get("customer_phone", submission.get("party_b_phone", "")),
"address": submission.get("customer_address", submission.get("party_b_address", "")),
"title": submission.get("party_b_title", "Client"),
}
# Financial terms if provided
financial_terms = {
"amount": _safe_float(submission.get("contract_amount", 0)),
"currency": currency,
"payment_schedule": submission.get("payment_schedule", ""),
"late_fee": _safe_float(submission.get("late_fee", 0)),
}
return {
"document_id": document_id,
"document_number": contract_number,
"document_type": "contract",
"currency": currency,
"issue_date": submission.get("issue_date", datetime.now().strftime("%B %d, %Y")),
"effective_date": submission.get("effective_date", ""),
"expiration_date": submission.get("expiration_date", ""),
"contract_title": submission.get("contract_title", "Service Agreement"),
"preamble": submission.get("preamble", ""),
"party_a": party_a,
"party_b": party_b,
"from": party_a,
"to": {
"name": party_b["name"],
"email": party_b["email"],
"phone": party_b["phone"],
"address": party_b["address"],
},
"clauses": clauses,
"financial_terms": financial_terms,
"governing_law": submission.get("governing_law", ""),
"line_items": line_items or [],
"subtotal": 0,
"tax_rate": 0,
"tax": 0,
"total": _safe_float(submission.get("contract_amount", 0)),
"notes": submission.get("notes", ""),
"signatures": {
"party_a_name": party_a["name"],
"party_a_title": party_a.get("title", "Provider"),
"party_b_name": party_b["name"],
"party_b_title": party_b.get("title", "Client"),
},
}