"""Auto-extracted from models.py — do not edit manually."""
import json
import secrets
import sqlite3
import uuid
from datetime import datetime, timezone
import bcrypt
from app.crypto import (
decrypt_entity,
decrypt_user_value,
encrypt_entity,
encrypt_user_submission,
encrypt_user_value,
encrypt_value,
hash_value,
is_encrypted,
key_is_configured,
try_decrypt,
try_decrypt_entity,
try_decrypt_user_submission,
try_decrypt_user_value,
)
from app.db import DB_PATH, get_db
def create_document_template(
user_id,
site_id=None,
name="invoice",
document_type="invoice",
layout="classic",
field_mapping=None,
style_config=None,
auto_generate=False,
invoice_number_format="INV-YYYY-####",
) -> int:
"""Create a new document template.
Returns template id.
"""
conn = None
try:
conn = get_db()
cursor = conn.execute(
"""
INSERT INTO document_templates
(user_id, site_id, name, document_type, layout, field_mapping, style_config,
auto_generate, invoice_number_format)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
user_id,
site_id,
name,
document_type,
layout,
json.dumps(field_mapping or {}),
json.dumps(style_config or {}),
1 if auto_generate else 0,
invoice_number_format or "INV-YYYY-####",
),
)
conn.commit()
return cursor.lastrowid
finally:
if conn:
conn.close()
def get_document_template(template_id, site_id=None, user_id=None):
"""Get a document template by id. Optionally validates site/user ownership."""
conn = None
try:
conn = get_db()
if site_id and user_id:
row = conn.execute(
"SELECT * FROM document_templates WHERE id = ? AND site_id = ? AND user_id = ?",
(template_id, site_id, user_id),
).fetchone()
elif site_id:
row = conn.execute(
"SELECT * FROM document_templates WHERE id = ? AND site_id = ?", (template_id, site_id)
).fetchone()
elif user_id:
row = conn.execute(
"SELECT * FROM document_templates WHERE id = ? AND user_id = ?", (template_id, user_id)
).fetchone()
else:
row = conn.execute("SELECT * FROM document_templates WHERE id = ?", (template_id,)).fetchone()
finally:
if conn:
conn.close()
if row:
row = dict(row)
row["field_mapping"] = json.loads(row.get("field_mapping", "{}"))
row["style_config"] = json.loads(row.get("style_config", "{}"))
return row
return None
def list_document_templates(user_id, site_id=None) -> list:
"""List document templates filtered by user_id, optionally scoped to site_id."""
conn = None
try:
conn = get_db()
if site_id is not None:
rows = conn.execute(
"SELECT * FROM document_templates WHERE user_id = ? AND site_id = ? ORDER BY created_at DESC",
(user_id, site_id),
).fetchall()
else:
rows = conn.execute(
"SELECT * FROM document_templates WHERE user_id = ? ORDER BY created_at DESC", (user_id,)
).fetchall()
finally:
if conn:
conn.close()
result = []
for row in rows:
row = dict(row)
row["field_mapping"] = json.loads(row.get("field_mapping", "{}"))
row["style_config"] = json.loads(row.get("style_config", "{}"))
result.append(row)
return result
def get_auto_generate_template(site_id):
"""Get the first active document template with auto_generate=1 for a site.
Returns the template dict or None.
"""
conn = None
try:
conn = get_db()
row = conn.execute(
"SELECT * FROM document_templates WHERE site_id = ? AND auto_generate = 1 AND is_active = 1 ORDER BY created_at DESC LIMIT 1",
(site_id,),
).fetchone()
finally:
if conn:
conn.close()
if row:
row = dict(row)
row["field_mapping"] = json.loads(row.get("field_mapping", "{}"))
row["style_config"] = json.loads(row.get("style_config", "{}"))
return row
return None
def update_document_template(
template_id,
user_id,
name=None,
document_type=None,
layout=None,
field_mapping=None,
style_config=None,
auto_generate=None,
invoice_number_format=None,
next_invoice_number=None,
) -> bool:
"""Update a document template. Returns True if updated."""
conn = None
try:
conn = get_db()
updates = []
params = []
if name is not None:
updates.append("name = ?")
params.append(name)
if document_type is not None:
updates.append("document_type = ?")
params.append(document_type)
if layout is not None:
updates.append("layout = ?")
params.append(layout)
if field_mapping is not None:
updates.append("field_mapping = ?")
params.append(json.dumps(field_mapping))
if style_config is not None:
updates.append("style_config = ?")
params.append(json.dumps(style_config))
if auto_generate is not None:
updates.append("auto_generate = ?")
params.append(1 if auto_generate else 0)
if invoice_number_format is not None:
updates.append("invoice_number_format = ?")
params.append(invoice_number_format)
if next_invoice_number is not None:
updates.append("next_invoice_number = ?")
params.append(next_invoice_number)
if not updates:
conn.close()
return False
updates.append("updated_at = CURRENT_TIMESTAMP")
params.append(template_id)
params.append(user_id)
conn.execute(f"UPDATE document_templates SET {', '.join(updates)} WHERE id = ? AND user_id = ?", params)
conn.commit()
finally:
if conn:
conn.close()
return True
def delete_document_template(template_id, user_id) -> bool:
"""Delete a document template. Returns True if deleted."""
conn = None
try:
conn = get_db()
conn.execute("DELETE FROM document_templates WHERE id = ? AND user_id = ?", (template_id, user_id))
deleted = conn.total_changes > 0
conn.commit()
finally:
if conn:
conn.close()
return deleted
def increment_document_download(document_id) -> bool:
"""Increment download count for a document."""
conn = None
try:
conn = get_db()
conn.execute(
"UPDATE documents SET download_count = COALESCE(download_count, 0) + 1 WHERE id = ?", (document_id,)
)
conn.commit()
return conn.total_changes > 0
finally:
if conn:
conn.close()
def store_document(
user_id,
site_id=None,
submission_id=None,
template_id=None,
document_type="invoice",
document_id=None,
data=None,
pdf_path=None,
expires_at=None,
invoice_number=None,
customer_email=None,
) -> int:
"""Store a generated document record.
Returns document row id.
"""
conn = None
try:
conn = get_db()
if data is None:
data = {}
cursor = conn.execute(
"""
INSERT INTO documents
(document_id, site_id, submission_id, template_id, document_type, data, pdf_path, expires_at,
user_id, invoice_number, customer_email)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
document_id or f"doc-{uuid.uuid4().hex[:12]}",
site_id,
submission_id,
template_id,
document_type,
json.dumps(data),
pdf_path,
expires_at,
user_id,
invoice_number,
customer_email,
),
)
conn.commit()
return cursor.lastrowid
finally:
if conn:
conn.close()
def get_document(document_id, site_id=None, user_id=None):
"""Get a document by its UUID. Optionally validates site/user ownership."""
conn = None
try:
conn = get_db()
if site_id and user_id:
row = conn.execute(
"SELECT * FROM documents WHERE document_id = ? AND site_id = ? AND user_id = ?",
(document_id, site_id, user_id),
).fetchone()
elif site_id:
row = conn.execute(
"SELECT * FROM documents WHERE document_id = ? AND site_id = ?", (document_id, site_id)
).fetchone()
elif user_id:
row = conn.execute(
"SELECT * FROM documents WHERE document_id = ? AND user_id = ?", (document_id, user_id)
).fetchone()
else:
row = conn.execute("SELECT * FROM documents WHERE document_id = ?", (document_id,)).fetchone()
finally:
if conn:
conn.close()
if row:
row = dict(row)
row["data"] = json.loads(row.get("data", "{}")) if row.get("data") else {}
return row
return None
def list_documents(user_id, site_id=None, status=None, limit=50) -> list:
"""List documents for a user, optionally filtered by site and status."""
conn = None
try:
conn = get_db()
query = "SELECT * FROM documents WHERE user_id = ?"
params = [user_id]
if site_id:
query += " AND site_id = ?"
params.append(site_id)
if status:
query += " AND status = ?"
params.append(status)
query += " ORDER BY created_at DESC LIMIT ?"
params.append(limit)
rows = conn.execute(query, params).fetchall()
docs = [dict(r) for r in rows]
for doc in docs:
if isinstance(doc.get("data"), str):
doc["data"] = json.loads(doc["data"])
return docs
finally:
if conn:
conn.close()
"""Increment download count for a document."""
conn = None
try:
conn = get_db()
conn.execute("UPDATE documents SET download_count = download_count + 1 WHERE document_id = ?", (document_id,))
conn.commit()
finally:
if conn:
conn.close()
def update_document_status(document_id, status):
"""Update document status (draft → sent → viewed → paid)."""
conn = None
try:
conn = get_db()
conn.execute("UPDATE documents SET status = ? WHERE document_id = ?", (status, document_id))
conn.commit()
finally:
if conn:
conn.close()
def update_document_due_date(document_id, due_date):
"""Update document due_date for overdue tracking."""
conn = None
try:
conn = get_db()
conn.execute("UPDATE documents SET due_date = ? WHERE document_id = ?", (due_date, document_id))
conn.commit()
finally:
if conn:
conn.close()
def update_document_generation_status(document_id, status):
"""Update document generation_status (pending, processing, completed, failed).
Status values:
- pending: document queued for PDF generation
- processing: PDF generation in progress
- completed: PDF generated and saved to disk
- failed: generation failed (see generation_error for details)
"""
conn = None
try:
conn = get_db()
conn.execute(
"UPDATE documents SET generation_status = ? WHERE document_id = ?", (status, document_id)
)
conn.commit()
finally:
if conn:
conn.close()
def update_document_generation_error(document_id, error_message):
"""Record the error message for a failed document generation."""
conn = None
try:
conn = get_db()
conn.execute(
"UPDATE documents SET generation_error = ? WHERE document_id = ?", (error_message, document_id)
)
conn.commit()
finally:
if conn:
conn.close()
def update_document_pdf_path(document_id, pdf_path):
"""Update the pdf_path for an existing document record."""
conn = None
try:
conn = get_db()
conn.execute(
"UPDATE documents SET pdf_path = ? WHERE document_id = ?", (pdf_path, document_id)
)
conn.commit()
finally:
if conn:
conn.close()
# ─── Phase 13.5: Invoice Schedule models ──────────────────────────────────────
def create_invoice_schedule(
user_id,
site_id,
template_id,
customer_name,
customer_email,
line_items,
document_number_prefix,
interval,
interval_count,
start_date,
):
"""Create a recurring invoice schedule. Returns schedule id."""
from datetime import timedelta
conn = None
try:
conn = get_db()
# Calculate next_run based on start_date and interval
start_dt = datetime.strptime(start_date, "%Y-%m-%d")
if interval == "weekly":
next_run = start_dt + timedelta(weeks=interval_count)
elif interval == "monthly":
# Simple month addition
month = start_dt.month - 1 + interval_count
year = start_dt.year + month // 12
month = month % 12 + 1
import calendar
day = min(start_dt.day, calendar.monthrange(year, month)[1])
next_run = start_dt.replace(year=year, month=month, day=day)
elif interval == "quarterly":
month = start_dt.month - 1 + (interval_count * 3)
year = start_dt.year + month // 12
month = month % 12 + 1
import calendar
day = min(start_dt.day, calendar.monthrange(year, month)[1])
next_run = start_dt.replace(year=year, month=month, day=day)
elif interval == "yearly":
next_run = start_dt.replace(year=start_dt.year + interval_count)
else:
next_run = start_dt + timedelta(days=interval_count * 30)
cursor = conn.execute(
"""
INSERT INTO invoice_schedules
(user_id, site_id, template_id, customer_name, customer_name_encrypted, customer_email, customer_email_encrypted, line_items,
document_number_prefix, interval, interval_count, start_date, next_run)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
user_id,
site_id,
template_id,
customer_name,
None,
customer_email,
None,
json.dumps(line_items),
document_number_prefix,
interval,
interval_count,
start_date,
next_run.strftime("%Y-%m-%d"),
),
)
conn.commit()
schedule_id = cursor.lastrowid
# Encrypt PII after insert
if customer_name:
enc = encrypt_entity("invoice_schedule", schedule_id, customer_name)
if enc:
conn.execute(
"UPDATE invoice_schedules SET customer_name_encrypted = ? WHERE id = ?", (enc, schedule_id)
)
if customer_email:
enc = encrypt_entity("invoice_schedule", schedule_id, customer_email)
if enc:
conn.execute(
"UPDATE invoice_schedules SET customer_email_encrypted = ? WHERE id = ?", (enc, schedule_id)
)
conn.commit()
return schedule_id
finally:
if conn:
conn.close()
def get_invoice_schedule(schedule_id, user_id=None, site_id=None):
"""Get a single invoice schedule."""
conn = None
try:
conn = get_db()
if site_id:
row = conn.execute(
"SELECT * FROM invoice_schedules WHERE id = ? AND site_id = ? AND user_id = ?",
(schedule_id, site_id, user_id),
).fetchone()
elif user_id:
row = conn.execute(
"SELECT * FROM invoice_schedules WHERE id = ? AND user_id = ?", (schedule_id, user_id)
).fetchone()
else:
row = conn.execute("SELECT * FROM invoice_schedules WHERE id = ?", (schedule_id,)).fetchone()
finally:
if conn:
conn.close()
if row:
row = dict(row)
row["line_items"] = json.loads(row.get("line_items", "[]"))
return _decrypt_invoice_schedule(row, schedule_id)
return None
def list_invoice_schedules(user_id, site_id=None):
"""List all invoice schedules for a user, optionally filtered by site."""
conn = None
try:
conn = get_db()
if site_id:
rows = conn.execute(
"SELECT * FROM invoice_schedules WHERE user_id = ? AND site_id = ? ORDER BY next_run ASC",
(user_id, site_id),
).fetchall()
else:
rows = conn.execute(
"SELECT * FROM invoice_schedules WHERE user_id = ? ORDER BY next_run ASC", (user_id,)
).fetchall()
finally:
if conn:
conn.close()
result = []
for row in rows:
row = dict(row)
row["line_items"] = json.loads(row.get("line_items", "[]"))
result.append(_decrypt_invoice_schedule(row, row["id"]))
return result
def update_invoice_schedule(schedule_id, user_id, site_id, **kwargs):
"""Update an invoice schedule. Supported fields: customer_name, customer_email,
line_items, document_number_prefix, interval, interval_count, start_date, status, template_id."""
allowed = {
"customer_name",
"customer_email",
"line_items",
"document_number_prefix",
"interval",
"interval_count",
"start_date",
"status",
"template_id",
}
updates = {k: v for k, v in kwargs.items() if k in allowed and v is not None}
if not updates:
return False
conn = None
try:
conn = get_db()
set_clauses = []
params = []
for key, value in updates.items():
if key == "line_items":
set_clauses.append(f"{key} = ?")
params.append(json.dumps(value))
elif key == "customer_name" and value:
set_clauses.append(f"{key} = ?")
params.append(value)
enc = encrypt_entity("invoice_schedule", schedule_id, value)
if enc:
set_clauses.append("customer_name_encrypted = ?")
params.append(enc)
elif key == "customer_email" and value:
set_clauses.append(f"{key} = ?")
params.append(value)
enc = encrypt_entity("invoice_schedule", schedule_id, value)
if enc:
set_clauses.append("customer_email_encrypted = ?")
params.append(enc)
else:
set_clauses.append(f"{key} = ?")
params.append(value)
set_clauses.append("updated_at = CURRENT_TIMESTAMP")
params.extend([schedule_id, site_id, user_id])
conn.execute(
f"UPDATE invoice_schedules SET {', '.join(set_clauses)} WHERE id = ? AND site_id = ? AND user_id = ?",
params,
)
conn.commit()
finally:
if conn:
conn.close()
return True
def delete_invoice_schedule(schedule_id, user_id, site_id):
"""Delete an invoice schedule."""
conn = None
try:
conn = get_db()
conn.execute(
"DELETE FROM invoice_schedules WHERE id = ? AND site_id = ? AND user_id = ?",
(schedule_id, site_id, user_id),
)
deleted = conn.total_changes > 0
conn.commit()
finally:
if conn:
conn.close()
return deleted
def get_due_schedules():
"""Get all schedules due for today (next_run <= today and status = active)."""
today = datetime.now().strftime("%Y-%m-%d")
conn = None
try:
conn = get_db()
rows = conn.execute(
"SELECT * FROM invoice_schedules WHERE next_run <= ? AND status = 'active'", (today,)
).fetchall()
finally:
if conn:
conn.close()
result = []
for row in rows:
row = dict(row)
row["line_items"] = json.loads(row.get("line_items", "[]"))
result.append(_decrypt_invoice_schedule(row, row["id"]))
return result
def advance_schedule_next_run(schedule_id, new_next_run):
"""Update next_run and last_run for a schedule after successful generation."""
conn = None
try:
conn = get_db()
conn.execute(
"UPDATE invoice_schedules SET next_run = ?, last_run = ? WHERE id = ?",
(new_next_run, datetime.now().strftime("%Y-%m-%d"), schedule_id),
)
conn.commit()
finally:
if conn:
conn.close()
# ─── Phase 11: Multi-Channel helpers ──────────────────────────────────────────
import io as _io
def generate_qr_code(url, size=300, border=2):
"""Generate a QR code as PNG bytes for the given URL."""
import qrcode
qr = qrcode.QRCode(
version=1,
error_correction=qrcode.constants.ERROR_CORRECT_H,
box_size=10,
border=border,
)
qr.add_data(url)
qr.make(fit=True)
img = qr.make_image(fill_color="black", back_color="white")
buf = _io.BytesIO()
img.save(buf, format="PNG")
return buf.getvalue()
def generate_email_html(site_name, fields, form_url, branding_color=None):
"""Generate table-based HTML for email embedding.
Uses inline CSS and table layout for Gmail/Outlook compatibility.
"""
color = branding_color or "#2563eb"
field_rows = ""
for field in fields:
ftype = field.get("type", "text")
label = field.get("label", field.get("name", "Field"))
req = field.get("required", False)
req_mark = '<span style="color:#e53e3e;">*</span>' if req else ""
if ftype == "textarea":
field_html = '<td colspan="2"><div contenteditable="false" style="border:1px solid #d1d5db; border-radius:4px; padding:8px; min-height:60px; background:#f9fafb; color:#9ca3af; font-size:14px;">Type your answer here...</div></td>'
elif ftype in ("select", "radio"):
options = field.get("options", [])
opts = "<br>".join([f"◊ {opt}" for opt in options])
field_html = f'<td colspan="2"><div style="padding:8px; color:#6b7280; font-size:14px; line-height:1.8;">{opts}</div></td>'
elif ftype == "checkbox":
options = field.get("options", [])
opts = "<br>".join([f"☐ {opt}" for opt in options])
field_html = f'<td colspan="2"><div style="padding:8px; color:#6b7280; font-size:14px; line-height:1.8;">{opts}</div></td>'
else:
field_html = f'<td colspan="2"><div contenteditable="false" style="border:1px solid #d1d5db; border-radius:4px; padding:8px; background:#f9fafb; color:#9ca3af; font-size:14px;">Enter {label.lower()}...</div></td>'
field_rows += f""" <tr>
<td style="padding:10px 12px 10px 0; font-weight:600; color:#374151; font-size:14px; vertical-align:top;">{label} {req_mark}</td>
{field_html}
</tr>
"""
return f'''<table cellpadding="0" cellspacing="0" border="0" width="100%" style="max-width:600px; margin:0 auto; font-family:Arial,Helvetica,sans-serif; background:#ffffff; border-radius:8px; overflow:hidden; border:1px solid #e5e7eb;">
<tr>
<td style="background:{color}; padding:24px; text-align:center;">
<h1 style="margin:0; color:#ffffff; font-size:22px;">{site_name}</h1>
</td>
</tr>
<tr>
<td style="padding:24px;">
{field_rows} </td>
</tr>
<tr>
<td style="padding:0 24px 24px; text-align:center;">
<a href="{form_url}" target="_blank" style="display:inline-block; padding:12px 32px; background:{color}; color:#ffffff; text-decoration:none; border-radius:6px; font-size:16px; font-weight:600;">Open Full Form →</a>
</td>
</tr>
</table>'''
def update_site_branding(
site_id,
og_title=None,
og_description=None,
og_image=None,
branding_color=None,
logo_url=None,
favicon_url=None,
custom_domain=None,
):
"""Update branding columns for a site."""
conn = None
try:
conn = get_db()
updates = []
params = []
for key, val in [
("og_title", og_title),
("og_description", og_description),
("og_image", og_image),
("branding_color", branding_color),
("logo_url", logo_url),
("favicon_url", favicon_url),
("custom_domain", custom_domain),
]:
if val is not None:
updates.append(f"{key} = ?")
params.append(val)
if not updates:
conn.close()
return
params.append(site_id)
conn.execute(f"UPDATE sites SET {', '.join(updates)} WHERE id = ?", params)
conn.commit()
finally:
if conn:
conn.close()
def list_user_documents(user_id, limit=50):
"""List documents for a user. Returns list of dicts."""
conn = None
try:
conn = get_db()
docs = [
dict(r)
for r in conn.execute(
"SELECT * FROM documents WHERE user_id = ? ORDER BY created_at DESC LIMIT ?",
(user_id, limit),
).fetchall()
]
for doc in docs:
if isinstance(doc.get("data"), str):
doc["data"] = json.loads(doc["data"])
return docs
finally:
if conn:
conn.close()
def count_user_documents(user_id):
"""Count documents by status. Returns dict with counts."""
conn = None
try:
conn = get_db()
rows = conn.execute(
"SELECT status, COUNT(*) as cnt FROM documents WHERE user_id = ? GROUP BY status",
(user_id,),
).fetchall()
counts = {r["status"]: r["cnt"] for r in rows}
counts["total"] = sum(counts.values())
return counts
finally:
if conn:
conn.close()
def count_user_documents_by_type(user_id):
"""Count documents by document_type. Returns list of (type, count) tuples."""
conn = None
try:
conn = get_db()
rows = conn.execute(
"SELECT document_type, COUNT(*) as cnt FROM documents WHERE user_id = ? GROUP BY document_type ORDER BY cnt DESC",
(user_id,),
).fetchall()
return [(r["document_type"], r["cnt"]) for r in rows]
finally:
if conn:
conn.close()
def list_documents_by_email(email):
"""List documents by customer email. Returns list of dicts."""
conn = None
try:
conn = get_db()
return [
dict(r)
for r in conn.execute(
"SELECT * FROM documents WHERE customer_email = ? ORDER BY created_at DESC",
(email,),
).fetchall()
]
finally:
if conn:
conn.close()
def update_document_stripe_payment(document_id, session_id, status):
"""Update document with Stripe payment info. Returns True on success."""
conn = None
try:
conn = get_db()
conn.execute(
"UPDATE documents SET stripe_session_id = ?, payment_status = ? WHERE id = ?",
(session_id, status, document_id),
)
conn.commit()
return True
finally:
if conn:
conn.close()
def generate_invoice_number(template_id):
"""Generate next invoice number for a template. Returns string."""
conn = None
try:
conn = get_db()
row = conn.execute(
"SELECT next_invoice_num FROM invoice_counters WHERE template_id = ?",
(template_id,),
).fetchone()
if not row:
conn.execute(
"INSERT INTO invoice_counters (template_id, next_invoice_num) VALUES (?, 1)",
(template_id,),
)
num = 1
else:
num = row["next_invoice_num"]
invoice_num = f"INV-{template_id}-{num:04d}"
return invoice_num
finally:
if conn:
conn.close()
def update_next_invoice_number(template_id, next_num):
"""Update the next invoice number counter."""
conn = None
try:
conn = get_db()
conn.execute(
"UPDATE invoice_counters SET next_invoice_num = ? WHERE template_id = ?",
(next_num, template_id),
)
conn.commit()
finally:
if conn:
conn.close()