"""RQ background tasks for PDF document generation.
Queued via the 'pdf' RQ queue. Falls back to synchronous execution
when Redis/RQ is unavailable (the calling route detects `job is None`).
"""
import json
import logging
import os
import uuid
from app.db import get_db
from app.models import store_document, update_document_pdf_path, update_document_generation_status
from app.services.documents import (
build_invoice_data,
generate_pdf,
render_document_html,
save_pdf,
)
logger = logging.getLogger("agentforms.tasks.pdf")
def generate_pdf_task(
user_id: int,
site_id: int | None,
template_id: int,
document_type: str,
data: dict,
layout: str,
document_id: str,
style_override: dict | None = None,
) -> dict:
"""Background task that renders HTML, generates PDF, saves to disk.
Returns a dict with document_id, pdf_path, and size_bytes on success.
Raises on any failure — RQ will mark the job as failed.
"""
logger.info("generate_pdf_task started", document_id=document_id)
# Step 1: Mark as processing
try:
update_document_generation_status(document_id, "processing")
except Exception:
# Document record might not exist yet if store_document hasn't run
pass
# Step 2: Render HTML
html_content = render_document_html(document_type, layout, data, style_override)
# Step 3: Generate PDF
pdf_bytes = generate_pdf(html_content, style_override)
# Step 4: Save to disk
pdf_path = save_pdf(pdf_bytes, document_id)
# Step 5: Update DB with generation status + pdf_path
update_document_generation_status(document_id, "completed")
update_document_pdf_path(document_id, pdf_path)
logger.info("generate_pdf_task completed", document_id=document_id, size=len(pdf_bytes))
return {
"document_id": document_id,
"pdf_path": pdf_path,
"size_bytes": len(pdf_bytes),
}
def generate_standalone_pdf_task(
user_id: int,
template_id: int | None,
document_type: str,
data: dict,
line_items: list,
customer_name: str,
customer_email: str,
customer_address: str,
business_name: str,
business_email: str,
business_phone: str,
business_address: str,
notes: str,
tax_rate: float,
payment_terms: str,
payment_link: str,
payment_link_text: str,
layout: str,
style: dict | None,
document_id: str,
invoice_number_format: str,
next_invoice_number: str | None,
) -> dict:
"""Background task for standalone document generation (no site/submission).
Rebuilds invoice data from flat parameters, renders HTML, generates PDF,
saves to disk, and stores the document record.
"""
logger.info("generate_standalone_pdf_task started", document_id=document_id)
try:
update_document_generation_status(document_id, "processing")
except Exception:
pass
# Rebuild invoice data from flat params
invoice_data = build_invoice_data(
{"id": template_id, "layout": layout, "style_config": style or {}, "field_mapping": {},
"next_invoice_number": next_invoice_number, "invoice_number_format": invoice_number_format},
{
"customer_name": customer_name,
"customer_email": customer_email,
"customer_address": customer_address,
"business_name": business_name,
"business_email": business_email,
"business_phone": business_phone,
"business_address": business_address,
"notes": notes,
"tax_rate": tax_rate,
"payment_terms": payment_terms,
"payment_link": payment_link,
"payment_link_text": payment_link_text,
},
line_items,
"USD",
document_type,
)
# Render HTML
html_content = render_document_html(document_type, layout, invoice_data, style)
# Generate PDF
pdf_bytes = generate_pdf(html_content, style)
# Save to disk
pdf_path = save_pdf(pdf_bytes, document_id)
# Store document record
row_id = store_document(
user_id=user_id,
site_id=None,
submission_id=None,
template_id=template_id,
document_type=document_type,
document_id=document_id,
data=invoice_data,
pdf_path=pdf_path,
invoice_number=invoice_data.get("invoice_number"),
customer_email=customer_email,
)
# Update generation status
update_document_generation_status(document_id, "completed")
update_document_pdf_path(document_id, pdf_path)
# Increment invoice number if template-based
if template_id and next_invoice_number:
parts = next_invoice_number.rsplit("-", 1)
if len(parts) == 2 and parts[1].isdigit():
prefix = parts[0] + "-"
seq = int(parts[1])
from app.model_documents import update_document_template
update_document_template(
template_id,
user_id,
next_invoice_number=prefix + str(seq + 1).zfill(len(parts[1])),
)
logger.info("generate_standalone_pdf_task completed", document_id=document_id)
return {
"document_id": document_id,
"pdf_path": pdf_path,
"size_bytes": len(pdf_bytes),
"invoice_number": invoice_data.get("invoice_number"),
}