"""Template management — form templates for site creation."""
import json
import logging
logger = logging.getLogger(__name__)
from app.db import get_db
# Import helpers
from app.helpers import (
_check_spam_content,
_current_month,
_decrypt_submission_fields,
_decrypt_user_row,
_encrypt_submission_field,
_get_site_owner_password_hash,
_is_encrypted_value,
parse_user_agent,
resolve_geo,
)
# ─── Column-name allowlist helpers ──────────────────────────────────────
_FORM_TEMPLATES_ALLOWED_COLUMNS = {
"name",
"description",
"fields",
"config",
"thumbnail",
"category",
"active",
"position",
"author",
"price",
"downloads",
"rating",
"reviews",
"tags",
"settings",
"field_config",
"success_message",
"is_featured",
"slug",
"created_by",
"created_at",
"updated_at",
}
def _validate_columns(columns: list, allowed: set) -> list:
"""Validate column names against an allowlist. Returns list of 'col = ?' clauses."""
valid = []
for col in columns:
if col in allowed:
valid.append(f"{col} = ?")
else:
logger.warning(f"Blocked column in UPDATE: {col} (not in allowlist)")
return valid
def list_templates(category=None, featured_only=False, search=None):
"""List form templates with optional filtering.
Args:
category: Filter by category (Business, Events, Hiring, Support, E-commerce, Onboarding)
featured_only: If True, return only featured templates
search: Search in name and description
Returns:
List of template dicts with field_config parsed as JSON.
"""
conn = None
try:
conn = get_db()
query = "SELECT * FROM form_templates WHERE 1=1"
params = []
if featured_only:
query += " AND is_featured = 1"
if category:
query += " AND category = ?"
params.append(category)
if search:
query += " AND (name LIKE ? OR description LIKE ?)"
params.extend([f"%{search}%", f"%{search}%"])
query += " ORDER BY is_featured DESC, name ASC"
rows = conn.execute(query, params).fetchall()
result = []
for row in rows:
d = dict(row)
try:
d["field_config"] = json.loads(d["field_config"]) if d["field_config"] else []
except (json.JSONDecodeError, TypeError):
d["field_config"] = []
d["field_count"] = len(d["field_config"])
result.append(d)
return result
finally:
if conn:
conn.close()
def get_template_by_slug(slug):
"""Get a single template by slug.
Returns:
Template dict with field_config parsed as JSON, or None.
"""
conn = None
try:
conn = get_db()
row = conn.execute("SELECT * FROM form_templates WHERE slug = ?", (slug,)).fetchone()
if not row:
return None
d = dict(row)
try:
d["field_config"] = json.loads(d["field_config"]) if d["field_config"] else []
except (json.JSONDecodeError, TypeError):
d["field_config"] = []
d["field_count"] = len(d["field_config"])
return d
finally:
if conn:
conn.close()
def seed_templates():
"""Seed the form_templates table with launch templates.
Idempotent — skips templates that already exist by slug.
"""
from app.services.template_seeds import LAUNCH_TEMPLATES
conn = None
try:
conn = get_db()
seeded = 0
for tmpl in LAUNCH_TEMPLATES:
existing = conn.execute(
"SELECT id FROM form_templates WHERE slug = ?",
(tmpl["slug"],),
).fetchone()
if existing:
continue
conn.execute(
"""INSERT INTO form_templates
(slug, name, description, category, field_config, success_message, created_by, is_featured)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
(
tmpl["slug"],
tmpl["name"],
tmpl["description"],
tmpl["category"],
json.dumps(tmpl["field_config"]),
tmpl.get("success_message"),
"system",
tmpl.get("is_featured", 0),
),
)
seeded += 1
conn.commit()
if seeded:
print(f"[migration] Seeded {seeded} template(s)")
finally:
if conn:
conn.close()
def create_template(
slug, name, description, category, field_config, success_message=None, is_featured=0, created_by="system"
):
"""Create a new template. Returns template dict or None on duplicate."""
conn = None
try:
conn = get_db()
conn.execute(
"""INSERT INTO form_templates
(slug, name, description, category, field_config, success_message, created_by, is_featured)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
(slug, name, description, category, json.dumps(field_config), success_message, created_by, is_featured),
)
conn.commit()
return get_template_by_slug(slug)
except sqlite3.IntegrityError:
return None
finally:
if conn:
conn.close()
def update_template(slug, **kwargs):
"""Update a template by slug. Returns updated template dict or None if not found."""
conn = None
try:
conn = get_db()
existing = conn.execute("SELECT id, field_config FROM form_templates WHERE slug = ?", (slug,)).fetchone()
if not existing:
return None
updates = []
values = []
column_names = []
for key in ("name", "description", "category", "field_config", "success_message", "is_featured"):
if key in kwargs:
val = kwargs[key]
if key == "field_config" and isinstance(val, (list, dict)):
val = json.dumps(val)
column_names.append(key)
values.append(val)
# Validate column names against allowlist
validated = _validate_columns(column_names, _FORM_TEMPLATES_ALLOWED_COLUMNS)
updates.extend(validated)
if updates:
values.append(slug)
conn.execute(f"UPDATE form_templates SET {', '.join(updates)} WHERE slug = ?", values)
conn.commit()
return get_template_by_slug(slug)
finally:
if conn:
conn.close()
def delete_template(slug):
"""Delete a template by slug. Returns True if deleted."""
conn = None
try:
conn = get_db()
cursor = conn.execute("DELETE FROM form_templates WHERE slug = ?", (slug,))
conn.commit()
return cursor.rowcount > 0
finally:
if conn:
conn.close()