# ─── Document Service (Orchestrator) ──────────────────────────────────────────
#
# Facade that re-exports everything from the focused sub-modules and provides
# the full-pipeline orchestration functions.
#
# Sub-modules:
# document_builders — build_invoice_data, build_proposal_data, etc.
# document_renderer — render_document_html, apply_field_mapping
# pdf_generator — generate_pdf, save_pdf, get_pdf_path
#
# Pipeline:
# submission → field_mapping → invoice_data → html_render → pdf_generate
#
# ─── Imports ───────────────────────────────────────────────────────────────────
import json
import logging
import os
import re
import uuid
from datetime import UTC, datetime, timezone
from pathlib import Path
from app.models import generate_invoice_number, update_next_invoice_number
logger = logging.getLogger("document")
# ─── Re-exports (backward-compatible API) ─────────────────────────────────────
# ── From document_builders ──
from .document_builders import ( # noqa: F401
DOCUMENT_STATUSES,
DOCUMENT_TYPES,
TEMPLATE_LAYOUTS,
build_contract_data,
build_delivery_note_data,
build_invoice_data,
build_proposal_data,
build_statement_data,
build_work_order_data,
)
# ── From document_renderer ──
from .document_renderer import ( # noqa: F401
DEFAULT_STYLE,
apply_field_mapping,
render_document_html,
)
# ── From pdf_generator ──
from .pdf_generator import ( # noqa: F401
PDF_STORAGE_DIR,
generate_pdf,
get_pdf_path,
save_pdf,
)
# Expose datetime for callers that do doc_service.datetime.now(...)
# (documents.py was imported as a module and accessed datetime this way)
from datetime import datetime as datetime # noqa: F401
# ─── Full Pipeline ─────────────────────────────────────────────────────────────
def render_document(
template_id: int, submission_id: int, template_data: dict, submission_data: dict, style: dict = None
) -> dict:
"""Full document generation pipeline.
Takes a template and submission, produces a document with PDF.
Args:
template_id: Document template ID
submission_id: Form submission ID
template_data: Template config from DB (layout, field_mapping, etc.)
submission_data: Raw submission from DB
style: Style overrides
Returns:
{
"document_id": str,
"pdf_bytes": bytes,
"html_preview": str,
"data": dict, # structured document data
"size_bytes": int,
}
"""
# Step 1: Transform submission data
field_mapping = template_data.get("field_mapping")
if isinstance(field_mapping, str):
field_mapping = json.loads(field_mapping)
elif field_mapping is None:
field_mapping = {}
document_data = apply_field_mapping(submission_data, field_mapping)
# Step 2: Generate document_id early (needed for payment link)
document_id = str(uuid.uuid4())
# Step 3: Add document metadata
document_data["document_number"] = field_mapping.get("document_number", "")
document_data["issue_date"] = submission_data.get("submitted_at", "")
document_data["due_date"] = field_mapping.get("due_date", "")
# Build payment link — auto-generate Stripe link if Stripe is configured
payment_link = field_mapping.get("payment_link", "")
if not payment_link:
app_url = os.environ.get("APP_URL", "https://agentforms.io")
payment_link = f"{app_url}/documents/{document_id}/pay"
document_data["payment_link"] = payment_link
document_data["payment_link_text"] = field_mapping.get("payment_link_text", "Pay Now")
# Step 2b: Generate invoice number for invoice documents
invoice_number = None
if template_data.get("document_type") == "invoice" and template_id:
try:
invoice_number = generate_invoice_number(template_id)
document_data["document_number"] = invoice_number
# Advance the next_invoice_number for the template
current = invoice_number
match = re.search(r"(\d+)$", current)
if match:
seq = int(match.group(1))
seq_len = len(match.group(1))
next_num = current[: match.start()] + str(seq + 1).zfill(seq_len)
update_next_invoice_number(template_id, next_num)
except Exception as e:
logger.warning("invoice_number_gen", f"Failed to generate invoice number: {e}")
# Step 3: Render HTML
layout = template_data.get("layout", "classic")
doc_type = template_data.get("document_type", "invoice")
html_content = render_document_html(doc_type, layout, document_data, style)
# Step 4: Generate PDF
pdf_bytes = generate_pdf(html_content, style)
return {
"document_id": document_id,
"pdf_bytes": pdf_bytes,
"html_preview": html_content,
"data": document_data,
"size_bytes": len(pdf_bytes),
"invoice_number": invoice_number,
}
def render_document_pdf(invoice_data: dict, template: dict, document_type: str = "invoice") -> str:
"""Render a document to PDF from pre-built invoice data.
Returns the relative PDF storage path.
"""
document_id = invoice_data.get("document_id") or str(uuid.uuid4())[:12]
layout = template.get("layout", "classic")
style = template.get("style_config") or {}
html_content = render_document_html(document_type, layout, invoice_data, style)
pdf_bytes = generate_pdf(html_content, style)
pdf_path = save_pdf(pdf_bytes, document_id)
return pdf_path
# ─── Recurring Schedule Helpers ────────────────────────────────────────────────
def calculate_next_run(current_run: str, interval: str, interval_count: int = 1) -> str:
"""Calculate the next run date for a recurring schedule.
Args:
current_run: ISO date string (YYYY-MM-DD)
interval: 'daily', 'weekly', 'monthly', 'yearly'
interval_count: Number of intervals to advance (default 1)
"""
from dateutil.relativedelta import relativedelta
try:
current = datetime.fromisoformat(current_run).date()
except (ValueError, TypeError):
current = datetime.now(UTC).date()
if interval == "daily":
delta = relativedelta(days=interval_count)
elif interval == "weekly":
delta = relativedelta(weeks=interval_count)
elif interval == "monthly":
delta = relativedelta(months=interval_count)
elif interval == "yearly":
delta = relativedelta(years=interval_count)
else:
delta = relativedelta(months=interval_count) # default to monthly
return (current + delta).isoformat()