# ─── Phase 12: Document Routes ─────────────────────────────────────────────────
#
# Document template management + PDF generation API.
#
import json
import uuid
from flask import Blueprint, jsonify, make_response, redirect, render_template, request, send_file, session
from app.helpers import _get_site_owner_password_hash
from app.models import (
advance_schedule_next_run,
check_document_quota,
count_user_documents,
count_user_documents_by_type,
create_document_template,
# Phase 13.5: invoice schedules
create_invoice_schedule,
delete_document_template,
delete_invoice_schedule,
get_auto_generate_template,
get_document,
get_document_template,
get_due_schedules,
get_invoice_schedule,
get_site,
get_submission,
get_user,
get_user_document_count,
get_user_tier,
increment_document_download,
list_document_templates,
list_documents,
list_documents_by_email,
list_invoice_schedules,
list_user_documents,
store_document,
update_document_due_date,
update_document_status,
update_document_stripe_payment,
update_document_template,
update_invoice_schedule,
)
from app.routes.auth import current_user, login_required
from app.services import documents as doc_service
doc_bp = Blueprint("documents", __name__)
def _get_redis():
"""Get Redis connection for RQ queue operations."""
import os
import redis
redis_url = os.environ.get("REDIS_URL", "redis://localhost:6379/0")
return redis.from_url(redis_url)
# ─── Top-Level Documents Dashboard ───────────────────────────────────────────
@doc_bp.route("/documents/")
@login_required
def documents_dashboard():
"""Top-level documents dashboard — aggregates all docs across user sites."""
user = current_user()
documents = list_user_documents(user["id"], limit=50)
counts = count_user_documents(user["id"])
doc_type_counts = count_user_documents_by_type(user["id"])
templates = list_document_templates(user_id=user["id"])
return render_template(
"documents_dashboard.html",
user=user,
documents=documents,
counts=counts,
doc_type_counts=doc_type_counts,
templates=templates,
)
# ─── Standalone Document Routes (no site_id) ──────────────────────────────────
@doc_bp.route("/documents/templates", methods=["GET"])
@login_required
def list_standalone_templates():
"""List user's document templates (standalone)."""
user = current_user()
templates = list_document_templates(user_id=user["id"])
return jsonify({"templates": templates})
@doc_bp.route("/documents/quota", methods=["GET"])
@login_required
def documents_quota():
"""Get document quota information for the current user."""
user = current_user()
tier_key, tier_config = get_user_tier(user["id"])
max_documents = tier_config.get("max_documents", 0)
current_count = get_user_document_count(user["id"])
counts = count_user_documents(user["id"])
return jsonify(
{
"tier": tier_key,
"tier_name": tier_config.get("name", "Unknown"),
"max_documents": max_documents,
"current_count": current_count,
"remaining": max(0, max_documents - current_count) if max_documents > 0 else -1,
"counts": counts,
}
)
@doc_bp.route("/documents/templates", methods=["POST"])
@login_required
def create_standalone_template():
"""Create a standalone document template (not tied to a site)."""
user = current_user()
# Check document quota
allowed, current_count, max_documents = check_document_quota(user["id"])
if not allowed:
if max_documents == 0:
return jsonify(
{
"error": "Document generation not available on your current plan",
"upgrade_required": True,
"current_tier": user.get("tier", "free"),
}
), 403
return jsonify(
{
"error": f"Document limit reached ({current_count}/{max_documents})",
"upgrade_required": True,
"current_count": current_count,
"max_documents": max_documents,
"current_tier": user.get("tier", "free"),
}
), 403
data = request.get_json(force=True)
template_id = create_document_template(
user_id=user["id"],
site_id=None,
name=data.get("name", "Invoice"),
document_type=data.get("document_type", "invoice"),
layout=data.get("layout", "stripe"),
field_mapping=data.get("field_mapping"),
style_config=data.get("style_config"),
auto_generate=data.get("auto_generate", False),
invoice_number_format=data.get("invoice_number_format", "INV-YYYY-####"),
)
return jsonify({"template_id": template_id}), 201
@doc_bp.route("/documents/generate", methods=["POST"])
@login_required
def generate_standalone_document():
"""Generate a standalone document (not tied to a site).
Query params:
- mode=async: queue PDF generation as background task (returns 202)
"""
user = current_user()
# Check document quota
allowed, current_count, max_documents = check_document_quota(user["id"])
if not allowed:
if max_documents == 0:
return jsonify(
{
"error": "Document generation not available on your current plan",
"upgrade_required": True,
"current_tier": user.get("tier", "free"),
}
), 403
return jsonify(
{
"error": f"Document limit reached ({current_count}/{max_documents})",
"upgrade_required": True,
"current_count": current_count,
"max_documents": max_documents,
"current_tier": user.get("tier", "free"),
}
), 403
data = request.get_json(force=True)
template_id = data.get("template_id")
document_type = data.get("document_type", "invoice")
# Allow standalone invoices without a template
template = None
if template_id:
template = get_document_template(template_id, user_id=user["id"])
if not template:
return jsonify({"error": "Template not found"}), 404
line_items = data.get("line_items")
if not line_items:
line_items = [{"description": "Service", "qty": 1, "unit_price": 100.0}]
customer_name = data.get("customer_name", "Customer")
customer_email = data.get("customer_email", "")
# Build template if none provided (standalone invoice)
if not template:
template = {
"id": None,
"name": "Standalone Invoice",
"layout": data.get("layout", "stripe"),
"style_config": data.get("style", {}),
"field_mapping": {},
"next_invoice_number": None,
"invoice_number_format": "INV-YYYY-####",
}
# Async mode — queue the task and return 202
if request.args.get("mode") == "async":
try:
from rq import Queue
from app.tasks.pdf import generate_standalone_pdf_task
r = _get_redis()
q = Queue("pdf", connection=r)
document_id = str(uuid.uuid4())
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={},
pdf_path=None,
)
from app.model_documents import update_document_generation_status
update_document_generation_status(document_id, "pending")
q.enqueue(
generate_standalone_pdf_task,
user_id=user["id"],
template_id=template_id,
document_type=document_type,
data={},
line_items=line_items,
customer_name=customer_name,
customer_email=customer_email,
customer_address=data.get("customer_address", ""),
business_name=data.get("business_name", ""),
business_email=data.get("business_email", ""),
business_phone=data.get("business_phone", ""),
business_address=data.get("business_address", ""),
notes=data.get("notes", ""),
tax_rate=data.get("tax_rate", 0),
payment_terms=data.get("payment_terms", ""),
payment_link=data.get("payment_link", ""),
payment_link_text=data.get("payment_link_text", "Pay Now"),
layout=template.get("layout", "stripe"),
style=template.get("style_config"),
document_id=document_id,
invoice_number_format=template.get("invoice_number_format", "INV-YYYY-####"),
next_invoice_number=template.get("next_invoice_number"),
job_timeout=120,
)
return jsonify({
"document_id": document_id,
"status": "queued",
"generation_status": "pending",
"message": "PDF generation queued — poll /documents/<id>/status for progress",
}), 202
except Exception:
# Fall through to synchronous generation if RQ is unavailable
pass
# Synchronous path (default)
# Build invoice data
invoice_data = doc_service.build_invoice_data(
template,
{
"customer_name": customer_name,
"customer_email": customer_email,
"customer_address": data.get("customer_address", ""),
"business_name": data.get("business_name", ""),
"business_email": data.get("business_email", ""),
"business_phone": data.get("business_phone", ""),
"business_address": data.get("business_address", ""),
"notes": data.get("notes", ""),
"tax_rate": data.get("tax_rate", 0),
"payment_terms": data.get("payment_terms", ""),
"payment_link": data.get("payment_link", ""),
"payment_link_text": data.get("payment_link_text", "Pay Now"),
},
line_items,
"USD",
document_type,
)
# Generate PDF
pdf_path = doc_service.render_document_pdf(invoice_data, template, document_type)
doc_id = store_document(
user_id=user["id"],
site_id=None,
submission_id=None,
template_id=template_id,
document_type=document_type,
data=invoice_data,
pdf_path=pdf_path,
invoice_number=invoice_data.get("invoice_number"),
customer_email=customer_email,
)
# Increment invoice number (only if template-based)
if template_id and template.get("next_invoice_number"):
parts = template["next_invoice_number"].rsplit("-", 1)
if len(parts) == 2 and parts[1].isdigit():
prefix = parts[0] + "-"
seq = int(parts[1])
update_document_template(
template_id,
user["id"],
next_invoice_number=prefix + str(seq + 1).zfill(len(parts[1])),
)
return jsonify(
{
"document_id": doc_id,
"pdf_path": pdf_path,
"invoice_number": invoice_data.get("invoice_number"),
}
), 201
@doc_bp.route("/documents/templates/<int:template_id>", methods=["PUT"])
@login_required
def update_standalone_template(template_id):
"""Update a standalone document template."""
user = current_user()
data = request.get_json(force=True)
updated = update_document_template(
template_id,
user["id"],
name=data.get("name"),
document_type=data.get("document_type"),
layout=data.get("layout"),
field_mapping=data.get("field_mapping"),
style_config=data.get("style_config"),
auto_generate=data.get("auto_generate"),
invoice_number_format=data.get("invoice_number_format"),
)
if not updated:
return jsonify({"error": "Nothing to update"}), 400
return jsonify({"message": "Template updated"})
@doc_bp.route("/documents/templates/<int:template_id>", methods=["DELETE"])
@login_required
def delete_standalone_template(template_id):
"""Delete a standalone document template."""
user = current_user()
template = get_document_template(template_id, user_id=user["id"])
if not template:
return jsonify({"error": "Template not found"}), 404
delete_document_template(template_id, user["id"])
return jsonify({"message": "Template deleted"})
@doc_bp.route("/documents/history", methods=["GET"])
@login_required
def list_standalone_documents():
"""List user's documents (standalone + site-scoped)."""
user = current_user()
status = request.args.get("status")
limit = int(request.args.get("limit", 50))
documents = list_documents(user_id=user["id"], status=status, limit=limit)
return jsonify({"documents": documents})
# ─── Session-Authenticated Routes (UI) ────────────────────────────────────────
@doc_bp.route("/sites/<int:site_id>/documents/templates", methods=["GET"])
@login_required
def templates_list(site_id):
"""List document templates for a site."""
site = get_site(site_id)
if not site or site["user_id"] != current_user()["id"]:
return jsonify({"error": "Site not found"}), 404
user = current_user()
templates = list_document_templates(user_id=user["id"], site_id=site_id)
return jsonify({"templates": templates})
@doc_bp.route("/sites/<int:site_id>/documents/templates", methods=["POST"])
@login_required
def templates_create(site_id):
"""Create a new document template.
Body:
{
"name": "Standard Invoice",
"document_type": "invoice",
"layout": "classic",
"field_mapping": { ... },
"style_config": { ... }
}
"""
site = get_site(site_id)
if not site or site["user_id"] != current_user()["id"]:
return jsonify({"error": "Site not found"}), 404
# Check document quota
user = current_user()
allowed, current_count, max_documents = check_document_quota(user["id"])
if not allowed:
if max_documents == 0:
return jsonify(
{
"error": "Document generation not available on your current plan",
"upgrade_required": True,
"current_tier": user.get("tier", "free"),
}
), 403
return jsonify(
{
"error": f"Document limit reached ({current_count}/{max_documents})",
"upgrade_required": True,
"current_count": current_count,
"max_documents": max_documents,
"current_tier": user.get("tier", "free"),
}
), 403
data = request.get_json()
if not data or not data.get("name"):
return jsonify({"error": "Name is required"}), 400
template_id = create_document_template(
user_id=current_user()["id"],
site_id=site_id,
name=data["name"],
document_type=data.get("document_type", "invoice"),
layout=data.get("layout", "stripe"),
field_mapping=data.get("field_mapping", {}),
style_config=data.get("style_config", {}),
auto_generate=data.get("auto_generate", False),
)
return jsonify({"template_id": template_id, "message": "Template created"}), 201
@doc_bp.route("/sites/<int:site_id>/documents/templates/<int:template_id>", methods=["GET"])
@login_required
def templates_get(site_id, template_id):
"""Get a single document template."""
site = get_site(site_id)
if not site or site["user_id"] != current_user()["id"]:
return jsonify({"error": "Site not found"}), 404
template = get_document_template(template_id, site_id=site_id, user_id=current_user()["id"])
if not template:
return jsonify({"error": "Template not found"}), 404
return jsonify({"template": template})
@doc_bp.route("/sites/<int:site_id>/documents/templates/<int:template_id>", methods=["PUT"])
@login_required
def templates_update(site_id, template_id):
"""Update a document template.
Body (any subset):
{
"name": "...",
"document_type": "...",
"layout": "...",
"field_mapping": { ... },
"style_config": { ... }
}
"""
site = get_site(site_id)
if not site or site["user_id"] != current_user()["id"]:
return jsonify({"error": "Site not found"}), 404
existing = get_document_template(template_id, site_id=site_id, user_id=current_user()["id"])
if not existing:
return jsonify({"error": "Template not found"}), 404
data = request.get_json()
updated = update_document_template(
template_id,
current_user()["id"],
name=data.get("name"),
document_type=data.get("document_type"),
layout=data.get("layout"),
field_mapping=data.get("field_mapping"),
style_config=data.get("style_config"),
auto_generate=data.get("auto_generate"),
)
if not updated:
return jsonify({"error": "Nothing to update"}), 400
return jsonify({"message": "Template updated"})
@doc_bp.route("/sites/<int:site_id>/documents/templates/<int:template_id>", methods=["DELETE"])
@login_required
def templates_delete(site_id, template_id):
"""Delete a document template."""
site = get_site(site_id)
if not site or site["user_id"] != current_user()["id"]:
return jsonify({"error": "Site not found"}), 404
existing = get_document_template(template_id, site_id=site_id, user_id=current_user()["id"])
if not existing:
return jsonify({"error": "Template not found"}), 404
delete_document_template(template_id, user["id"])
return jsonify({"message": "Template deleted"})
# ─── Document Generation ──────────────────────────────────────────────────────
@doc_bp.route("/sites/<int:site_id>/documents/generate", methods=["POST"])
@login_required
def documents_generate(site_id):
"""Generate a document from a submission using a template.
Query params:
- mode=async: queue PDF generation as background task (returns 202)
Body:
{
"template_id": 1,
"submission_id": 42,
"status": "draft"
}
"""
site = get_site(site_id)
if not site or site["user_id"] != current_user()["id"]:
return jsonify({"error": "Site not found"}), 404
# Check document quota
user = current_user()
allowed, current_count, max_documents = check_document_quota(user["id"])
if not allowed:
if max_documents == 0:
return jsonify(
{
"error": "Document generation not available on your current plan",
"upgrade_required": True,
"current_tier": user.get("tier", "free"),
}
), 403
return jsonify(
{
"error": f"Document limit reached ({current_count}/{max_documents})",
"upgrade_required": True,
"current_count": current_count,
"max_documents": max_documents,
"current_tier": user.get("tier", "free"),
}
), 403
data = request.get_json()
template_id = data.get("template_id")
submission_id = data.get("submission_id")
if not template_id or not submission_id:
return jsonify({"error": "template_id and submission_id required"}), 400
template = get_document_template(template_id, site_id=site_id, user_id=current_user()["id"])
if not template:
return jsonify({"error": "Template not found"}), 404
submission = get_submission(submission_id, password_hash=_get_site_owner_password_hash(site_id))
if not submission or submission.get("site_id") != site_id:
return jsonify({"error": "Submission not found"}), 404
# Async mode — queue the task and return 202
if request.args.get("mode") == "async":
try:
from rq import Queue
from app.tasks.pdf import generate_pdf_task
r = _get_redis()
q = Queue("pdf", connection=r)
# Pre-store document record with pending status
document_id = str(uuid.uuid4())
store_document(
user_id=current_user()["id"],
site_id=site_id,
submission_id=submission_id,
template_id=template_id,
document_type=template.get("document_type", "invoice"),
document_id=document_id,
data={},
pdf_path=None,
)
from app.model_documents import update_document_generation_status
update_document_generation_status(document_id, "pending")
q.enqueue(
generate_pdf_task,
user_id=current_user()["id"],
site_id=site_id,
template_id=template_id,
document_type=template.get("document_type", "invoice"),
data=dict(submission),
layout=template.get("layout", "classic"),
document_id=document_id,
style_override=template.get("style_config"),
job_timeout=120,
)
return jsonify({
"document_id": document_id,
"status": "queued",
"generation_status": "pending",
"message": "PDF generation queued — poll /documents/<id>/status for progress",
}), 202
except Exception:
# Fall through to synchronous generation if RQ is unavailable
pass
# Synchronous path (default)
result = doc_service.render_document(
template_id=template_id,
submission_id=submission_id,
template_data=template,
submission_data=dict(submission),
style=template.get("style_config", {}),
)
# Save PDF to disk
pdf_path = doc_service.save_pdf(result["pdf_bytes"], result["document_id"])
# Store document record
doc_id = store_document(
user_id=current_user()["id"],
site_id=site_id,
submission_id=submission_id,
template_id=template_id,
document_type=template.get("document_type", "invoice"),
document_id=result["document_id"],
data=result["data"],
pdf_path=pdf_path,
expires_at=None,
invoice_number=result.get("invoice_number"),
customer_email=result["data"].get("customer_email"),
)
resp_data = {
"document_id": result["document_id"],
"download_url": f"/api/documents/{result['document_id']}/download",
"size_bytes": result["size_bytes"],
"data": result["data"],
"message": "Document generated successfully",
}
if result.get("invoice_number"):
resp_data["invoice_number"] = result["invoice_number"]
return jsonify(resp_data), 201
@doc_bp.route("/sites/<int:site_id>/documents", methods=["GET"])
@login_required
def documents_list(site_id):
"""List documents for a site.
Query params:
- status: filter by status (draft, sent, viewed, paid)
- limit: max results (default 50)
"""
site = get_site(site_id)
if not site or site["user_id"] != current_user()["id"]:
return jsonify({"error": "Site not found"}), 404
status = request.args.get("status")
limit = int(request.args.get("limit", "50"))
documents = list_documents(user_id=current_user()["id"], site_id=site_id, status=status, limit=limit)
return jsonify({"documents": documents})
# ─── Public Document Access ───────────────────────────────────────────────────
@doc_bp.route("/documents/<document_id>/preview", methods=["GET"])
def document_preview(document_id):
"""Get HTML preview of a document. No auth required — shareable link."""
doc = get_document(document_id)
if not doc:
return jsonify({"error": "Document not found"}), 404
data = doc.get("data", {})
if not data:
return jsonify({"error": "No document data available"}), 404
# Re-render HTML preview
layout = "stripe"
doc_type = doc.get("document_type", "invoice")
# Try to get template for layout/style
template_id = doc.get("template_id")
if template_id:
template = get_document_template(template_id)
if template:
layout = template.get("layout", "classic")
style = template.get("style_config", {})
else:
style = {}
else:
style = {}
# Use style from data if not from template
if not style and "style_config" in data:
style = data["style_config"]
html = doc_service.render_document_html(doc_type, layout, data, style)
response = make_response(html)
response.status_code = 200
response.headers["Content-Type"] = "text/html"
response.headers["X-Frame-Options"] = "SAMEORIGIN"
response.headers.pop("Content-Security-Policy", None)
return response
@doc_bp.route("/documents/<document_id>/download", methods=["GET"])
def document_download(document_id):
"""Download a document as PDF. No auth required — shareable link."""
doc = get_document(document_id)
if not doc:
return jsonify({"error": "Document not found"}), 404
# Increment download count
increment_document_download(document_id)
pdf_path = doc.get("pdf_path")
if not pdf_path:
return jsonify({"error": "PDF not available"}), 404
import os
if not os.path.exists(pdf_path):
return jsonify({"error": "PDF file not found"}), 404
return send_file(
pdf_path,
mimetype="application/pdf",
as_attachment=True,
download_name=f"{doc.get('document_type', 'document')}-{document_id}.pdf",
)
@doc_bp.route("/documents/<document_id>/embed", methods=["GET"])
def document_embed(document_id):
"""Serve PDF inline for embedded preview (iframe)."""
import logging
logging.warning(f"EMBED REQUEST: document_id={document_id}")
doc = get_document(document_id)
logging.warning(f"EMBED RESULT: doc={doc is not None}")
if not doc:
return jsonify({"error": "Document not found"}), 404
pdf_path = doc.get("pdf_path")
if not pdf_path:
return jsonify({"error": "PDF not available"}), 404
import os
if not os.path.exists(pdf_path):
return jsonify({"error": "PDF file not found"}), 404
response = make_response(
send_file(
pdf_path,
mimetype="application/pdf",
)
)
response.headers["Content-Disposition"] = 'inline; filename="{0}"'.format(
f"{doc.get('document_type', 'document')}-{document_id}.pdf"
)
return response
@doc_bp.route("/documents/<document_id>/status", methods=["POST"])
def document_status_update(document_id):
"""Update document status (webhook callback).
Body:
{
"status": "viewed" | "paid"
}
"""
doc = get_document(document_id)
if not doc:
return jsonify({"error": "Document not found"}), 404
data = request.get_json()
new_status = data.get("status")
if new_status not in doc_service.DOCUMENT_STATUSES:
return jsonify({"error": f"Invalid status. Must be one of: {doc_service.DOCUMENT_STATUSES}"}), 400
update_document_status(document_id, new_status)
return jsonify({"message": f"Document status updated to {new_status}"})
@doc_bp.route("/documents/<document_id>/status", methods=["GET"])
def document_generation_status(document_id):
"""Poll generation status for async PDF generation.
Returns:
{
"document_id": str,
"generation_status": str, # pending | processing | completed | failed
"pdf_path": str | null,
"error": str | null
}
"""
doc = get_document(document_id)
if not doc:
return jsonify({"error": "Document not found"}), 404
return jsonify({
"document_id": document_id,
"generation_status": doc.get("generation_status", "completed"),
"pdf_path": doc.get("pdf_path"),
"error": doc.get("generation_error"),
})
# ─── Phase 13.3: Payment Status Cycle ─────────────────────────────────────────
@doc_bp.route("/documents/<document_id>/status/cycle", methods=["POST"])
def document_status_cycle(document_id):
"""Cycle document status forward: draft → sent → viewed → paid.
Also supports setting due_date in the request body.
"""
from datetime import datetime
doc = get_document(document_id)
if not doc:
return jsonify({"error": "Document not found"}), 404
current = doc.get("status", "draft")
status_order = ["draft", "sent", "viewed", "paid"]
if current in status_order and status_order.index(current) < len(status_order) - 1:
new_status = status_order[status_order.index(current) + 1]
else:
new_status = "sent" # wrap around
update_document_status(document_id, new_status)
# Also update due_date if provided
data = request.get_json(silent=True) or {}
if data.get("due_date"):
update_document_due_date(document_id, data["due_date"])
return jsonify({"message": f"Status cycled to {new_status}", "status": new_status})
@doc_bp.route("/documents/<document_id>/due-date", methods=["POST"])
def document_set_due_date(document_id):
"""Set or update the due date for a document."""
doc = get_document(document_id)
if not doc:
return jsonify({"error": "Document not found"}), 404
data = request.get_json()
due_date = data.get("due_date")
if not due_date:
return jsonify({"error": "due_date is required"}), 400
update_document_due_date(document_id, due_date)
return jsonify({"message": f"Due date set to {due_date}"})
# ─── Phase 13.4: Bulk Document Generation ─────────────────────────────────────
@doc_bp.route("/sites/<int:site_id>/documents/bulk-generate", methods=["POST"])
@login_required
def documents_bulk_generate(site_id):
"""Generate documents for multiple submissions at once.
Body:
{
"template_id": 1,
"submission_ids": [42, 43, 44]
}
"""
site = get_site(site_id)
if not site or site["user_id"] != current_user()["id"]:
return jsonify({"error": "Site not found"}), 404
data = request.get_json()
template_id = data.get("template_id")
submission_ids = data.get("submission_ids", [])
if not template_id or not submission_ids:
return jsonify({"error": "template_id and submission_ids required"}), 400
template = get_document_template(template_id, site_id=site_id, user_id=current_user()["id"])
if not template:
return jsonify({"error": "Template not found"}), 404
results = []
errors = []
for sub_id in submission_ids:
submission = get_submission(sub_id, password_hash=_get_site_owner_password_hash(site_id))
if not submission or submission.get("site_id") != site_id:
errors.append({"submission_id": sub_id, "error": "Submission not found"})
continue
try:
result = doc_service.render_document(
template_id=template_id,
submission_id=sub_id,
template_data=template,
submission_data=dict(submission),
style=template.get("style_config", {}),
)
pdf_path = doc_service.save_pdf(result["pdf_bytes"], result["document_id"])
doc_id = store_document(
user_id=current_user()["id"],
site_id=site_id,
submission_id=sub_id,
template_id=template_id,
document_type=template.get("document_type", "invoice"),
document_id=result["document_id"],
data=result["data"],
pdf_path=pdf_path,
expires_at=None,
customer_email=result["data"].get("customer_email"),
)
results.append({"submission_id": sub_id, "document_id": result["document_id"], "status": "ok"})
except Exception as e:
errors.append({"submission_id": sub_id, "error": str(e)})
return jsonify(
{
"generated": len(results),
"errors": len(errors),
"results": results,
"error_details": errors,
}
)
# ─── Phase 13.5: Invoice Schedules (Recurring) ────────────────────────────────
@doc_bp.route("/sites/<int:site_id>/invoice-schedules", methods=["GET"])
@login_required
def invoice_schedules_list(site_id):
"""List all invoice schedules for a site."""
site = get_site(site_id)
if not site or site["user_id"] != current_user()["id"]:
return jsonify({"error": "Site not found"}), 404
schedules = list_invoice_schedules(user_id=current_user()["id"], site_id=site_id)
return jsonify({"schedules": schedules})
@doc_bp.route("/sites/<int:site_id>/invoice-schedules", methods=["POST"])
@login_required
def invoice_schedules_create(site_id):
"""Create a recurring invoice schedule.
Body:
{
"template_id": 1,
"customer_name": "John Doe",
"customer_email": "john@example.com",
"line_items": [{"description": "Service", "quantity": 1, "unit_price": 100}],
"document_number_prefix": "INV",
"interval": "monthly",
"interval_count": 1,
"start_date": "2026-07-01"
}
"""
site = get_site(site_id)
if not site or site["user_id"] != current_user()["id"]:
return jsonify({"error": "Site not found"}), 404
data = request.get_json()
required = ["template_id", "customer_name", "customer_email", "start_date"]
for field in required:
if not data.get(field):
return jsonify({"error": f"{field} is required"}), 400
schedule_id = create_invoice_schedule(
user_id=current_user()["id"],
site_id=site_id,
template_id=data["template_id"],
customer_name=data["customer_name"],
customer_email=data["customer_email"],
line_items=data.get("line_items", []),
document_number_prefix=data.get("document_number_prefix", "INV"),
interval=data.get("interval", "monthly"),
interval_count=data.get("interval_count", 1),
start_date=data["start_date"],
)
return jsonify({"schedule_id": schedule_id, "message": "Schedule created"}), 201
@doc_bp.route("/sites/<int:site_id>/invoice-schedules/<int:schedule_id>", methods=["GET"])
@login_required
def invoice_schedules_get(site_id, schedule_id):
"""Get a single invoice schedule."""
site = get_site(site_id)
if not site or site["user_id"] != current_user()["id"]:
return jsonify({"error": "Site not found"}), 404
schedule = get_invoice_schedule(schedule_id, user_id=current_user()["id"], site_id=site_id)
if not schedule:
return jsonify({"error": "Schedule not found"}), 404
return jsonify({"schedule": schedule})
@doc_bp.route("/sites/<int:site_id>/invoice-schedules/<int:schedule_id>", methods=["PUT"])
@login_required
def invoice_schedules_update(site_id, schedule_id):
"""Update an invoice schedule."""
site = get_site(site_id)
if not site or site["user_id"] != current_user()["id"]:
return jsonify({"error": "Site not found"}), 404
existing = get_invoice_schedule(schedule_id, user_id=current_user()["id"], site_id=site_id)
if not existing:
return jsonify({"error": "Schedule not found"}), 404
data = request.get_json()
update_invoice_schedule(schedule_id, current_user()["id"], site_id, **data)
return jsonify({"message": "Schedule updated"})
@doc_bp.route("/sites/<int:site_id>/invoice-schedules/<int:schedule_id>", methods=["DELETE"])
@login_required
def invoice_schedules_delete(site_id, schedule_id):
"""Delete an invoice schedule."""
site = get_site(site_id)
if not site or site["user_id"] != current_user()["id"]:
return jsonify({"error": "Site not found"}), 404
existing = get_invoice_schedule(schedule_id, user_id=current_user()["id"], site_id=site_id)
if not existing:
return jsonify({"error": "Schedule not found"}), 404
delete_invoice_schedule(schedule_id, current_user()["id"], site_id)
return jsonify({"message": "Schedule deleted"})
@doc_bp.route("/sites/<int:site_id>/invoice-schedules/run", methods=["POST"])
@login_required
def invoice_schedules_run(site_id):
"""Manually trigger generation for all due schedules."""
site = get_site(site_id)
if not site or site["user_id"] != current_user()["id"]:
return jsonify({"error": "Site not found"}), 404
due = get_due_schedules()
site_schedules = [s for s in due if s["site_id"] == site_id]
results = []
for schedule in site_schedules:
try:
template = get_document_template(schedule["template_id"], user_id=schedule["user_id"])
if not template:
results.append({"schedule_id": schedule["id"], "error": "Template not found"})
continue
# Build a synthetic submission
submission_data = {
"id": 0,
"site_id": site_id,
"customer_name": schedule["customer_name"],
"customer_email": schedule["customer_email"],
"submitted_at": doc_service.datetime.now(timezone.utc).strftime("%Y-%m-%d"),
"data": {"_schedule_items": json.dumps(schedule["line_items"])},
}
result = doc_service.render_document(
template_id=schedule["template_id"],
submission_id=0,
template_data=template,
submission_data=submission_data,
style=template.get("style_config", {}),
)
pdf_path = doc_service.save_pdf(result["pdf_bytes"], result["document_id"])
doc_id = store_document(
user_id=schedule["user_id"],
site_id=site_id,
submission_id=0,
template_id=schedule["template_id"],
document_type=template.get("document_type", "invoice"),
document_id=result["document_id"],
data=result["data"],
pdf_path=pdf_path,
expires_at=None,
invoice_number=result.get("invoice_number"),
customer_email=schedule.get("customer_email"),
)
# Advance the schedule
next_run = doc_service.calculate_next_run(
schedule.get("next_run", schedule["start_date"]),
schedule["interval"],
schedule["interval_count"],
)
advance_schedule_next_run(schedule["id"], next_run)
results.append({"schedule_id": schedule["id"], "document_id": result["document_id"], "status": "ok"})
except Exception as e:
results.append({"schedule_id": schedule["id"], "error": str(e)})
return jsonify(
{
"processed": len(results),
"results": results,
}
)
# ─── Phase 13.7: Template Gallery ─────────────────────────────────────────────
@doc_bp.route("/documents/gallery", methods=["GET"])
def template_gallery():
"""Return pre-built template layouts available in the gallery."""
gallery = [
{
"name": "Stripe",
"layout": "stripe",
"description": "Minimalist, whitespace-driven — Stripe's invoice aesthetic",
"preview": "Clean, generous whitespace, subtle typography",
},
{
"name": "HubSpot",
"layout": "hubspot",
"description": "Bold header band with color-blocked sections",
"preview": "Strong brand header, color-coded status",
},
{
"name": "Notion",
"layout": "notion",
"description": "Editorial, sidebar accent, clean readability",
"preview": "Notion-style with sidebar and editorial type",
},
{
"name": "Customer Portal",
"layout": "customer_portal",
"description": "Customer-facing invoice with payment link and status",
"preview": "Client-ready invoice with pay button",
},
{
"name": "Classic",
"layout": "classic",
"description": "Clean, professional layout with traditional formatting",
"preview": "Classic invoice style with clear section breaks",
},
{
"name": "Modern",
"layout": "modern",
"description": "Contemporary design with bold headers and color accents",
"preview": "Modern design with accent colors",
},
{
"name": "Minimal",
"layout": "minimal",
"description": "Ultra-clean, whitespace-focused layout for modern brands",
"preview": "Minimal design with generous spacing",
},
{
"name": "Bold",
"layout": "bold",
"description": "High-contrast, attention-grabbing layout with strong typography",
"preview": "Bold design with heavy visual hierarchy",
},
]
return jsonify({"gallery": gallery})
# ─── Phase 13.6: Customer Invoice Portal ──────────────────────────────────────
@doc_bp.route("/documents/portal", methods=["GET"])
def customer_portal():
"""Customer-facing invoice portal. Unauthenticated — email-only lookup.
Query params:
- email: customer email address (required)
Renders a branded page showing all invoices/documents for that customer
with status, download, and pay links.
"""
email = request.args.get("email", "").strip().lower()
if not email:
return render_template("documents/customer_portal.html", documents=None, email="", error="Email is required")
documents = list_documents_by_email(email)
if not documents:
return render_template(
"documents/customer_portal.html",
documents=None,
email=email,
error="No invoices found for this email address.",
)
return render_template("documents/customer_portal.html", documents=documents, email=email, error=None)
@doc_bp.route("/documents/<document_id>/pay", methods=["GET"])
def document_pay(document_id):
"""Create a Stripe Checkout session for a document invoice (one-time payment).
This route is accessible from the customer portal without authentication.
The customer is redirected to Stripe to complete payment.
On success, the webhook handler updates the document status to 'paid'.
"""
import os
import stripe
doc = get_document(document_id)
if not doc:
return jsonify({"error": "Document not found"}), 404
# Only allow payment on unpaid invoices
if doc.get("status") in ("paid", "void"):
return jsonify({"error": "Document already " + doc.get("status") + " — no payment needed"}), 400
# Extract payment amount from document data
data = doc.get("data") or {}
total = data.get("total")
if not total or total <= 0:
return jsonify({"error": "Invalid invoice amount"}), 400
# Check if Stripe is configured
stripe_key = os.environ.get("STRIPE_SECRET_KEY", "")
if not stripe_key:
return jsonify({"error": "Stripe not configured — payments unavailable"}), 503
stripe.api_key = stripe_key
# Get APP_URL for success/cancel redirects
app_url = os.environ.get("APP_URL", "http://localhost:5060")
# Build line items from invoice data or use a single line item
line_items = []
invoice_items = data.get("line_items", [])
if invoice_items:
# Use the actual line items from the invoice
for item in invoice_items:
desc = item.get("description", "Invoice item")
qty = item.get("quantity", 1)
price = item.get("unit_price", total)
# Convert to cents for Stripe
line_items.append(
{
"price_data": {
"currency": data.get("currency", "usd").lower(),
"product_data": {"name": desc},
"unit_amount": int(float(price) * 100),
},
"quantity": qty,
}
)
else:
# Fallback to single line item with total
line_items.append(
{
"price_data": {
"currency": data.get("currency", "usd").lower(),
"product_data": {"name": f"Invoice {doc.get('invoice_number', document_id[:12])}"},
"unit_amount": int(float(total) * 100),
},
"quantity": 1,
}
)
# Create Stripe Checkout session (payment mode, not subscription)
try:
checkout_session = stripe.checkout.Session.create(
line_items=line_items,
mode="payment",
success_url=f"{app_url}/documents/portal?email={doc.get('customer_email', '')}&paid=true",
cancel_url=f"{app_url}/documents/portal?email={doc.get('customer_email', '')}&cancelled=true",
metadata={
"document_id": document_id,
"site_id": doc.get("site_id", ""),
},
customer_email=doc.get("customer_email"),
)
# Track the payment session on the document
update_document_stripe_payment(
document_id,
checkout_session.id,
"pending",
)
# Redirect to Stripe checkout
return redirect(checkout_session.url, code=303)
except stripe.error.StripeError as e:
return jsonify({"error": f"Payment processing error: {str(e)}"}), 500
# ─── Phase 13.6: Customer Portal — Send Invoice ──────────────────────────────
@doc_bp.route("/documents/<document_id>/send", methods=["POST"])
@login_required
def document_send(document_id):
"""Send invoice email to customer with portal link.
Transitions document status draft → sent.
Returns JSON response with success/error.
"""
from app.routes.email import NotificationService
doc = get_document(document_id)
if not doc:
return jsonify({"error": "Document not found"}), 404
# Auth check: user must own this document
user = current_user()
if doc.get("user_id") != user["id"]:
return jsonify({"error": "Unauthorized"}), 403
customer_email = doc.get("customer_email")
if not customer_email:
return jsonify({"error": "No customer email on this document"}), 400
data = doc.get("data") or {}
line_items = data.get("line_items", [])
inv_num = doc.get("invoice_number", document_id[:12])
site_name = doc.get("site_name", "AgentForms")
customer_name = data.get("customer_name", customer_email)
# Build portal link
base_url = request.host_url.rstrip("/")
portal_url = f"{base_url}/documents/portal?email={customer_email}"
# Build email body
items_lines = ""
total = 0
for item in line_items:
desc = item.get("description", "Item")
qty = item.get("quantity", 1)
unit = item.get("unit_price", 0)
line_total = qty * unit
total += line_total
items_lines += f" {desc} x{qty} @ ${unit:.2f} = ${line_total:.2f}\n"
# Build due date line separately to avoid backslash in f-string
due_line = ""
dd = doc.get("due_date")
if dd:
due_line = "Due Date: " + dd + "\n"
body = f"""Hello {customer_name},
You have an invoice from {site_name}.
Invoice: {inv_num}
Amount Due: ${total:.2f}
{due_line}
Items:
{items_lines}
You can view and pay this invoice at your customer portal:
{portal_url}
Thank you for your business.
-- {site_name} via AgentForms
"""
success = NotificationService.send(
customer_email,
f"Invoice {inv_num} from {site_name}",
body,
)
if success:
# Transition draft → sent
update_document_status(document_id, "sent")
return jsonify(
{
"success": True,
"message": f"Invoice sent to {customer_email}",
"portal_url": portal_url,
}
)
else:
return jsonify({"error": "Failed to send email. Check SMTP configuration."}), 502
# ─── Phase 13.7: Customer Portal — Share Link ────────────────────────────────
@doc_bp.route("/documents/<document_id>/share-link", methods=["GET"])
@login_required
def document_share_link(document_id):
"""Return the customer portal URL for a specific document as JSON."""
doc = get_document(document_id)
if not doc:
return jsonify({"error": "Document not found"}), 404
user = current_user()
if doc.get("user_id") != user["id"]:
return jsonify({"error": "Unauthorized"}), 403
customer_email = doc.get("customer_email")
if not customer_email:
return jsonify({"error": "No customer email on this document"}), 400
base_url = request.host_url.rstrip("/")
portal_url = f"{base_url}/documents/portal?email={customer_email}"
return jsonify(
{
"success": True,
"portal_url": portal_url,
"customer_email": customer_email,
}
)