import io
import os
import re
import threading
from flask import (
Blueprint,
jsonify,
make_response,
redirect,
render_template,
request,
send_file,
send_from_directory,
session,
url_for,
)
from app.models import add_waitlist, get_site, update_site_branding, waitlist_count
from app.routes.auth import current_user, login_required
from app.services.logging import log
site_bp = Blueprint("site", __name__)
def valid_email(email):
"""Basic email validation."""
return bool(re.match(r"^[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}$", email))
def add_cors_headers(response):
"""Add CORS headers only for API endpoints. CSP is set at app level."""
if not request.path.startswith("/api/"):
return response
origin = request.headers.get("Origin")
if origin and origin.startswith(("http://", "https://")):
response.headers["Access-Control-Allow-Origin"] = origin
else:
response.headers["Access-Control-Allow-Origin"] = "*"
response.headers["Access-Control-Allow-Methods"] = "POST, OPTIONS"
response.headers["Access-Control-Allow-Headers"] = "Content-Type, Authorization"
response.headers["Access-Control-Max-Age"] = "86400"
return response
site_bp.after_request(add_cors_headers)
@site_bp.route("/")
def index():
"""Landing page — redirect authenticated users to dashboard."""
if "user_id" in session:
return redirect(url_for("auth.dashboard"))
resp = make_response(render_template("index.html"))
resp.headers["Cache-Control"] = "no-store, no-cache, must-revalidate"
return resp
@site_bp.route("/signup")
def signup_page():
"""Redirect to registration page."""
return redirect(url_for("auth.register"))
@site_bp.route("/privacy")
def privacy():
"""Privacy Policy page."""
return render_template("privacy.html")
@site_bp.route("/terms")
def terms():
"""Terms of Service page."""
return render_template("terms.html")
@site_bp.route("/features")
def features():
"""Features page — standalone route for real navigation."""
return render_template("features.html")
@site_bp.route("/pricing")
def pricing():
"""Pricing page — standalone route for real navigation."""
return render_template("pricing.html")
@site_bp.route("/api/waitlist", methods=["POST", "OPTIONS"])
def api_waitlist():
"""Add email to waitlist. Accepts JSON or form data."""
if request.method == "OPTIONS":
return make_response("", 204)
if request.is_json:
data = request.json
else:
data = request.form.to_dict()
email = data.get("email", "").strip()
if not email:
return jsonify({"error": "Email is required"}), 400
if not valid_email(email):
return jsonify({"error": "Invalid email address"}), 400
added = add_waitlist(email, "landing")
if added:
return jsonify({"success": True, "message": "You're on the list!"}), 201
else:
return jsonify({"success": True, "message": "Already on the list!"}), 200
@site_bp.route("/embed.js")
@site_bp.route("/embed.v<version>.js")
def serve_embed_sdk(version=None):
"""Serve the embeddable JavaScript SDK.
Versioned URLs (/embed.v1.js) for cache invalidation.
Unversioned (/embed.js) redirects to latest version.
"""
import os as _os
static_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "static")
# If no version specified, redirect to current version
if version is None:
return redirect(url_for("site.serve_embed_sdk", version="1"))
response = send_from_directory(static_dir, "embed.js")
response.headers["Cache-Control"] = f"public, max-age={365 * 86400}" # 1 year for versioned URLs
return response
@site_bp.route("/wp-plugin")
def serve_wp_plugin():
"""Serve the WordPress plugin ZIP file for download."""
static_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "static")
return send_from_directory(static_dir, "agentforms-wp.zip", as_attachment=True)
@site_bp.route("/api/beacon", methods=["POST", "OPTIONS"])
def api_beacon():
"""Lightweight beacon endpoint for SDK impression tracking.
Accepts JSON with: token, event, timestamp, referrer.
Persists to form_impressions table with geo/device enrichment.
"""
if request.method == "OPTIONS":
return make_response("", 204)
data = request.get_json(silent=True) or {}
token = data.get("token", "")
if not token:
return jsonify({"ok": True}), 200
from app.models import add_form_impression, find_site_by_token
site = find_site_by_token(token)
if not site:
return jsonify({"ok": False, "error": "Invalid token"}), 404
client_ip = request.headers.get("X-Forwarded-For", request.remote_addr)
user_agent = request.headers.get("User-Agent")
try:
add_form_impression(site["id"], client_ip=client_ip, user_agent=user_agent)
except Exception as e:
log.error("beacon", f"impression error: {e}", error=str(e))
return jsonify({"ok": True}), 200
@site_bp.route("/api/sessions", methods=["POST", "OPTIONS"])
def api_sessions():
"""Session tracking endpoint for Phase 10 Advanced Analytics.
Beacon-compatible endpoint — accepts form session events from the embed SDK.
Events: session_start, session_update, session_complete.
Payload:
- token: Site embed token (required)
- session_id: Client-generated UUID (required)
- visitor_id: Stable visitor hash for return visitor tracking
- event: Event type (session_start, session_update, session_complete)
- fields_viewed: Array of field keys the user interacted with
- last_field_reached: The furthest field key reached
- step_reached: Current step number (for multi-step forms)
- referrer: Page referrer URL
"""
if request.method == "OPTIONS":
return make_response("", 204)
data = request.get_json(silent=True) or {}
token = data.get("token", "")
if not token:
return jsonify({"ok": True}), 200
from app.models import create_or_update_session, find_site_by_token
site = find_site_by_token(token)
if not site:
return jsonify({"ok": True}), 200
client_ip = request.headers.get("X-Forwarded-For", request.remote_addr)
user_agent = request.headers.get("User-Agent")
try:
create_or_update_session(
site_id=site["id"],
session_id=data.get("session_id", ""),
visitor_id=data.get("visitor_id", ""),
event=data.get("event", "session_start"),
client_ip=client_ip,
user_agent=user_agent,
referrer=data.get("referrer"),
fields_viewed=data.get("fields_viewed"),
last_field_reached=data.get("last_field_reached"),
step_reached=data.get("step_reached"),
)
except Exception as e:
log.error("sessions", f"error: {e}", error=str(e))
return jsonify({"ok": True}), 200
@site_bp.route("/api/v1/variants", methods=["GET", "POST", "OPTIONS"])
def api_variants():
"""A/B test variant management endpoint.
GET /api/v1/variants?token=TOKEN — Get variants for a site
POST /api/v1/variants — Create or update a variant (requires auth)
"""
if request.method == "OPTIONS":
return jsonify({}), 200
if request.method == "GET":
token = request.args.get("token")
if not token:
return jsonify({"error": "token required"}), 400
from app.models import ab_test_stats, assign_variant, find_site_by_token, get_variants
site = find_site_by_token(token)
if not site:
return jsonify({"error": "site not found"}), 404
variants = get_variants(site["id"])
has_active = len(variants) >= 2
# If active test, assign variant based on visitor_id
visitor_id = request.args.get("visitor_id")
assigned = None
if has_active and visitor_id:
assigned = assign_variant(site["id"], visitor_id)
result = {
"site_id": site["id"],
"variants": variants,
"has_active_test": has_active,
"assigned_variant": assigned,
}
return jsonify(result)
# POST — create variant (requires auth)
@login_required
def create_variant_handler():
data = request.get_json(silent=True)
if not data:
return jsonify({"error": "JSON body required"}), 400
from app.models import create_variant, get_site
site_id = data.get("site_id")
variant_key = data.get("variant_key")
name = data.get("name")
field_config = data.get("field_config")
if not all([site_id, variant_key, field_config]):
return jsonify({"error": "site_id, variant_key, field_config required"}), 400
if variant_key not in ("A", "B"):
return jsonify({"error": "variant_key must be A or B"}), 400
site = get_site(site_id)
if not site:
return jsonify({"error": "site not found"}), 404
result = create_variant(site_id, variant_key, name, field_config, data.get("success_message"))
if not result:
return jsonify({"error": "variant already exists"}), 409
return jsonify({"ok": True, "variant": result}), 201
if request.method == "POST":
return create_variant_handler()
@site_bp.route("/api/v1/variants/<int:variant_id>", methods=["PUT", "DELETE", "OPTIONS"])
@login_required
def api_variant_action(variant_id):
"""Update or delete an A/B test variant.
PUT /api/v1/variants/{id} — Update variant (requires auth)
DELETE /api/v1/variants/{id} — Deactivate variant (requires auth)
"""
if request.method == "OPTIONS":
return jsonify({}), 200
from app.models import deactivate_variant, get_variant, update_variant
if request.method == "PUT":
data = request.get_json(silent=True)
if not data:
return jsonify({"error": "JSON body required"}), 400
existing = get_variant(variant_id)
if not existing:
return jsonify({"error": "variant not found"}), 404
result = update_variant(variant_id, **data)
if not result:
return jsonify({"error": "update failed"}), 400
return jsonify({"ok": True, "variant": result})
# DELETE
success = deactivate_variant(variant_id)
if not success:
return jsonify({"error": "variant not found"}), 404
return jsonify({"ok": True})
@site_bp.route("/api/v1/variants/stats", methods=["GET", "OPTIONS"])
def api_variant_stats():
"""Get A/B test comparison stats for a site.
GET /api/v1/variants/stats?token=TOKEN
"""
if request.method == "OPTIONS":
return jsonify({}), 200
token = request.args.get("token")
days = request.args.get("days", 30, type=int)
if not token:
return jsonify({"error": "token required"}), 400
from app.models import ab_test_stats, find_site_by_token
site = find_site_by_token(token)
if not site:
return jsonify({"error": "site not found"}), 404
stats = ab_test_stats(site["id"], days=days)
return jsonify(stats)
# ─── Phase 11: Multi-Channel Embedding ────────────────────────────────────────
def _sanitize_hex_color(color):
"""Validate hex color. Returns None if invalid."""
if not color:
return None
return color if re.match(r"^#[0-9a-fA-F]{6}$", color.strip()) else None
def _sanitize_url(url, allowed_schemes=("http", "https")):
"""Validate URL scheme. Returns None if not a valid HTTP(S) URL."""
if not url:
return None
url = url.strip()
# Check if URL starts with an allowed scheme
for scheme in allowed_schemes:
if url.lower().startswith(scheme + ":"):
return url
return None
def _get_hosted_form_context(token):
"""Build context dict for hosted form rendering. Returns None if site not found."""
from app.models import find_site_by_token, parse_site_fields
site = find_site_by_token(token)
if not site:
return None
fields = parse_site_fields(site) or []
metadata = {}
if site.get("metadata"):
try:
import json as _json
metadata = _json.loads(site["metadata"])
except (_json.JSONDecodeError, TypeError):
pass
# Branding from dedicated columns, fallback to metadata
og_title = site.get("og_title") or metadata.get("og_title")
og_description = site.get("og_description") or metadata.get("og_description")
og_image = site.get("og_image") or metadata.get("og_image")
branding_color = site.get("branding_color") or metadata.get("branding_color")
logo_url = site.get("logo_url") or metadata.get("logo_url")
favicon_url = site.get("favicon_url") or metadata.get("favicon_url")
# ─── Sanitize user-controlled values ──────────────────────────────────
branding_color = _sanitize_hex_color(branding_color) or "#2563eb"
logo_url = _sanitize_url(logo_url)
favicon_url = _sanitize_url(favicon_url)
og_image = _sanitize_url(og_image)
return {
"token": token,
"site_name": site.get("name", "Form"),
"fields": fields,
"honeypot_enabled": site.get("honeypot_enabled", 1),
"metadata": metadata,
"submit_text": metadata.get("submit_text", "Submit"),
# Branding variables for template
"og_title": og_title,
"og_description": og_description,
"og_image": og_image,
"branding_color": branding_color,
"logo_url": logo_url,
"favicon_url": favicon_url,
}
@site_bp.route("/f/<token>")
def hosted_form(token):
"""Short URL / hosted form page — server-rendered HTML form.
Works without JavaScript. POSTs directly to /api/submit.
Supports branding via site metadata (logo, favicon, OG tags).
"""
ctx = _get_hosted_form_context(token)
if ctx is None:
return render_template("form_404.html"), 404
return render_template("hosted_form.html", **ctx)
@site_bp.route("/api/v2/forms/<token>/qr")
def qr_code(token):
"""Generate QR code pointing to the hosted form URL.
GET /api/v2/forms/<token>/qr?format=svg&size=300
Formats: svg (default), png
Size: 200-1000 (default 300)
"""
import qrcode
from qrcode.image.svg import SvgPathImage
ctx = _get_hosted_form_context(token)
if ctx is None:
return jsonify({"error": "Form not found"}), 404
fmt = request.args.get("format", "svg").lower()
size = min(max(int(request.args.get("size", 300)), 200), 1000)
form_url = request.host_url.rstrip("/") + "/f/" + token
try:
if fmt == "png":
qr = qrcode.make(form_url, box_size=10, border=2)
buf = io.BytesIO()
qr.save(buf, format="PNG")
buf.seek(0)
return send_file(
buf,
mimetype="image/png",
as_attachment=False,
download_name=f"qr-{token}.png",
)
else:
# SVG — use SvgPathImage (pure Python, no PIL dependency for SVG)
qr = qrcode.make(
form_url,
image_factory=SvgPathImage,
box_size=10,
border=2,
)
buf = io.BytesIO()
qr.save(buf)
buf.seek(0)
svg_data = buf.read()
return svg_data, 200, {"Content-Type": "image/svg+xml"}
except Exception as e:
log.error("qr", f"generation error: {e}", error=str(e))
return jsonify({"error": "QR generation failed"}), 500
@site_bp.route("/api/v2/forms/<token>/email-html")
def email_html_generator(token):
"""Generate email-optimized form HTML (table-based layout, inline CSS).
GET /api/v2/forms/<token>/email-html
Returns HTML designed for embedding in email clients (Gmail, Outlook, Apple Mail).
Uses table-based layout with inline styles for maximum compatibility.
"""
ctx = _get_hosted_form_context(token)
if ctx is None:
return jsonify({"error": "Form not found"}), 404
form_url = request.host_url.rstrip("/") + "/f/" + token
# For email, we generate a table-based landing page that links to the hosted form
# Email clients can't render interactive forms reliably, so the best approach
# is a branded call-to-action that links to the hosted form page.
html = generate_email_html(ctx, form_url)
return html, 200, {"Content-Type": "text/html; charset=utf-8"}
@site_bp.route("/api/v2/sites/<int:site_id>/branding", methods=["GET", "PUT"])
@login_required
def site_branding(site_id):
"""GET/PUT site branding settings.
GET /api/v2/sites/<int:site_id>/branding
PUT /api/v2/sites/<int:site_id>/branding
Branding fields:
- og_title: Open Graph title
- og_description: Open Graph description
- og_image: Open Graph image URL
- branding_color: Hex color (#rrggbb)
- logo_url: Custom logo URL
- favicon_url: Custom favicon URL
- custom_domain: Custom domain (Pro tier only)
"""
user = current_user()
if user is None:
return jsonify({"error": "Authentication required"}), 401
site = get_site(site_id)
if not site:
return jsonify({"error": "Site not found"}), 404
if site.get("user_id") != user["id"]:
return jsonify({"error": "Access denied"}), 403
# Check Pro tier for custom_domain
is_pro = user.get("tier") in ("pro", "team")
if request.method == "GET":
return jsonify(
{
"og_title": site.get("og_title"),
"og_description": site.get("og_description"),
"og_image": site.get("og_image"),
"branding_color": site.get("branding_color"),
"logo_url": site.get("logo_url"),
"favicon_url": site.get("favicon_url"),
"custom_domain": site.get("custom_domain") if is_pro else None,
"is_pro": is_pro,
}
)
# PUT - update branding
data = request.get_json()
if not data:
return jsonify({"error": "JSON body required"}), 400
# Validate custom domain - Pro only
if "custom_domain" in data and data["custom_domain"]:
if not is_pro:
return jsonify({"error": "Custom domains require Pro tier"}), 403
# Validate domain format
domain = data["custom_domain"].strip().lower()
if not re.match(r"^[a-z0-9]([a-z0-9-]*[a-z0-9])?(\.[a-z]{2,})+$", domain):
return jsonify({"error": "Invalid domain format. Example: forms.yourdomain.com"}), 400
# Validate branding color format
if "branding_color" in data and data["branding_color"]:
color = data["branding_color"].strip()
if not re.match(r"^#[0-9a-fA-F]{6}$", color):
return jsonify({"error": "Invalid color format. Use #rrggbb"}), 400
# Update branding
update_site_branding(
site_id,
og_title=data.get("og_title"),
og_description=data.get("og_description"),
og_image=data.get("og_image"),
branding_color=data.get("branding_color"),
logo_url=data.get("logo_url"),
favicon_url=data.get("favicon_url"),
custom_domain=data.get("custom_domain") if is_pro else None,
)
return jsonify({"success": True, "message": "Branding updated"})
def generate_email_html(ctx, form_url):
"""Generate email-safe HTML with a call-to-action linking to the hosted form.
Email clients can't reliably render interactive forms inline.
Best practice: branded CTA card → hosted form URL.
"""
import html as _html
name = _html.escape(ctx.get("site_name", "Form"))
submit_text = _html.escape(ctx.get("submit_text", "Submit"))
# Build a compact field summary
field_preview = ""
fields = ctx.get("fields", [])
visible_fields = [f for f in fields if not f.get("condition")]
if not visible_fields:
visible_fields = fields
shown = visible_fields[:5]
if shown:
lines = []
for f in shown:
label = _html.escape(f.get("label", f.get("key", "")))
lines.append(f'<span style="color:#666;">•</span> {label}')
field_preview = '<div style="margin:12px 0;font-size:13px;line-height:1.8;">' + "\n".join(lines) + "</div>"
if len(visible_fields) > 5:
field_preview += (
f'<div style="font-size:12px;color:#999;margin-top:4px;">+{len(visible_fields) - 5} more fields</div>'
)
return f'''<table cellpadding="0" cellspacing="0" border="0" width="100%" style="max-width:480px;margin:0 auto;font-family:Arial,Helvetica,sans-serif;">
<tr>
<td align="center" style="padding:24px;">
<table cellpadding="0" cellspacing="0" border="0" width="100%" style="background:#ffffff;border:1px solid #e5e7eb;border-radius:8px;overflow:hidden;">
<tr>
<td style="padding:28px 32px 20px;text-align:center;">
<h1 style="margin:0;font-size:20px;font-weight:700;color:#1a1a2e;line-height:1.3;">{name}</h1>
{field_preview}
<table cellpadding="0" cellspacing="0" border="0" style="margin-top:20px;">
<tr>
<td align="center" style="background:#2563eb;border-radius:6px;">
<a href="{form_url}" style="display:inline-block;padding:12px 32px;font-size:15px;font-weight:600;color:#ffffff;text-decoration:none;">{submit_text}</a>
</td>
</tr>
</table>
</td>
</tr>
</table>
</td>
</tr>
</table>'''