"""Auto-extracted from models.py — do not edit manually."""
import json
import secrets
import sqlite3
import uuid
from datetime import UTC, 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 site_session_stats(site_id, days=30):
"""Advanced analytics: session-based metrics for a site.
Returns:
dict with:
- total_sessions: count of session_start events
- completed_sessions: count of sessions with completed_at set
- abandonment_rate: pct of sessions that didn't complete
- avg_completion_time_seconds: avg time from start to complete
- return_visitor_count: sessions from repeat visitor_ids
- return_visitor_rate: pct of sessions from return visitors
"""
conn = None
try:
conn = get_db()
cutoff = f"-{days} days"
# Total sessions (distinct session_ids)
total_row = conn.execute(
"""SELECT COUNT(DISTINCT session_id) as cnt FROM form_sessions
WHERE site_id = ? AND started_at >= datetime('now', ?)""",
(site_id, cutoff),
).fetchone()
total_sessions = total_row["cnt"]
# Completed sessions (have completed_at set)
completed_row = conn.execute(
"""SELECT COUNT(DISTINCT session_id) as cnt FROM form_sessions
WHERE site_id = ? AND started_at >= datetime('now', ?)
AND completed_at IS NOT NULL""",
(site_id, cutoff),
).fetchone()
completed_sessions = completed_row["cnt"]
# Avg completion time (for completed sessions)
avg_time_row = conn.execute(
"""SELECT AVG((julianday(completed_at) - julianday(started_at)) * 86400) as avg_seconds
FROM form_sessions
WHERE site_id = ? AND started_at >= datetime('now', ?)
AND completed_at IS NOT NULL AND started_at IS NOT NULL""",
(site_id, cutoff),
).fetchone()
# Return visitors: visitor_ids with >1 session in period
visitor_counts = conn.execute(
"""SELECT visitor_id, COUNT(DISTINCT session_id) as cnt FROM form_sessions
WHERE site_id = ? AND started_at >= datetime('now', ?)
GROUP BY visitor_id HAVING cnt > 1""",
(site_id, cutoff),
).fetchall()
return_visitor_sessions = sum(r["cnt"] for r in visitor_counts)
return_visitor_count = len(visitor_counts)
return {
"total_sessions": total_sessions,
"completed_sessions": completed_sessions,
"abandonment_rate": round((1 - completed_sessions / total_sessions) * 100, 1) if total_sessions > 0 else 0,
"avg_completion_time_seconds": round(avg_time_row["avg_seconds"], 1)
if avg_time_row["avg_seconds"]
else None,
"return_visitor_count": return_visitor_count,
"return_visitor_sessions": return_visitor_sessions,
"return_visitor_rate": round(return_visitor_sessions / total_sessions * 100, 1)
if total_sessions > 0
else 0,
}
finally:
if conn:
conn.close()
def site_field_dropoff(site_id, days=30):
"""Field-level drop-off analysis.
For each field, calculates how many sessions reached it vs how many completed past it.
Drop-off = sessions that reached this field but never submitted.
Returns:
list of {field_key, reached_count, completed_count, drop_off_rate}
ordered by drop_off_rate descending.
"""
conn = None
try:
conn = get_db()
cutoff = f"-{days} days"
# Get all unique last_field_reached values (where sessions stopped)
dropoff_fields = conn.execute(
"""SELECT last_field_reached, COUNT(*) as cnt
FROM form_sessions
WHERE site_id = ? AND started_at >= datetime('now', ?)
AND last_field_reached IS NOT NULL AND last_field_reached != ''
AND completed_at IS NULL
GROUP BY last_field_reached
ORDER BY cnt DESC""",
(site_id, cutoff),
).fetchall()
# Total sessions in period
total_row = conn.execute(
"""SELECT COUNT(DISTINCT session_id) as cnt FROM form_sessions
WHERE site_id = ? AND started_at >= datetime('now', ?)""",
(site_id, cutoff),
).fetchone()
total = total_row["cnt"]
# Completed sessions
completed_row = conn.execute(
"""SELECT COUNT(DISTINCT session_id) as cnt FROM form_sessions
WHERE site_id = ? AND started_at >= datetime('now', ?)
AND completed_at IS NOT NULL""",
(site_id, cutoff),
).fetchone()
completed = completed_row["cnt"]
result = []
for row in dropoff_fields:
field_key = row["last_field_reached"]
abandoned = row["cnt"]
# Estimate reached = abandoned + completed (rough approximation)
reached = abandoned + completed
result.append(
{
"field_key": field_key,
"abandoned": abandoned,
"reached": reached,
"completed": completed,
"drop_off_rate": round(abandoned / reached * 100, 1) if reached > 0 else 0,
}
)
return result
finally:
if conn:
conn.close()
def site_hourly_trends(site_id, days=30):
"""Hourly/day-of-week submission trends for heatmap.
Returns:
dict with:
- hourly: {hour: count} for 0-23
- dow: {day: count} for 0-6 (Monday-Sunday)
"""
conn = None
try:
conn = get_db()
cutoff = f"-{days} days"
# Hourly breakdown
hourly_rows = conn.execute(
"""SELECT CAST(strftime('%H', submitted_at) AS INTEGER) as hour, COUNT(*) as cnt
FROM submissions
WHERE site_id = ? AND spam_flag = 0
AND submitted_at >= datetime('now', ?)
GROUP BY hour ORDER BY hour""",
(site_id, cutoff),
).fetchall()
# Day of week breakdown
dow_rows = conn.execute(
"""SELECT CAST(strftime('%w', submitted_at) AS INTEGER) as dow, COUNT(*) as cnt
FROM submissions
WHERE site_id = ? AND spam_flag = 0
AND submitted_at >= datetime('now', ?)
GROUP BY dow ORDER BY dow""",
(site_id, cutoff),
).fetchall()
hourly = {r["hour"]: r["cnt"] for r in hourly_rows}
dow = {r["dow"]: r["cnt"] for r in dow_rows}
return {"hourly": hourly, "dow": dow}
finally:
if conn:
conn.close()
# ─── Phase 10.9: A/B Testing Framework ─────────────────────────────────────────
def ab_test_stats(site_id, days=30):
"""Get A/B test comparison stats.
Returns:
dict with:
- has_active_test: bool
- variants: [{key, sessions, completions, abandonment_rate, avg_time}]
"""
conn = None
try:
conn = get_db()
variants = conn.execute(
"SELECT * FROM form_variants WHERE site_id = ? AND active = 1 ORDER BY variant_key", (site_id,)
).fetchall()
if len(variants) < 2:
return {"has_active_test": False, "variants": []}
cutoff = f"-{days} days"
result = []
for variant in variants:
total = conn.execute(
"""SELECT COUNT(DISTINCT session_id) as cnt FROM form_sessions
WHERE site_id = ? AND metadata LIKE ? AND started_at >= datetime('now', ?)""",
(site_id, f'%"variant_key":"{variant["variant_key"]}"%', cutoff),
).fetchone()["cnt"]
completed = conn.execute(
"""SELECT COUNT(DISTINCT session_id) as cnt FROM form_sessions
WHERE site_id = ? AND event = 'session_complete'
AND metadata LIKE ? AND started_at >= datetime('now', ?)""",
(site_id, f'%"variant_key":"{variant["variant_key"]}"%', cutoff),
).fetchone()["cnt"]
avg_time_row = conn.execute(
"""SELECT AVG((julianday(completed_at) - julianday(started_at)) * 86400) as avg_sec
FROM form_sessions
WHERE site_id = ? AND event = 'session_complete'
AND metadata LIKE ? AND started_at >= datetime('now', ?)""",
(site_id, f'%"variant_key":"{variant["variant_key"]}"%', cutoff),
).fetchone()
result.append(
{
"key": variant["variant_key"],
"name": variant.get("name"),
"sessions": total,
"completions": completed,
"abandonment_rate": round((1 - completed / total) * 100, 1) if total > 0 else 0,
"avg_completion_time": round(avg_time_row["avg_sec"], 1) if avg_time_row["avg_sec"] else None,
"weight": variant["weight"],
}
)
return {"has_active_test": True, "variants": result}
finally:
if conn:
conn.close()
def get_site_analytics(site_id, days=30):
"""Get submission analytics for a site over the last N days.
Returns list of daily records ordered by day.
"""
conn = None
try:
conn = get_db()
rows = conn.execute(
"""SELECT day, submission_count, spam_count
FROM site_analytics
WHERE site_id = ? AND day >= date('now', ?)
ORDER BY day""",
(site_id, f"-{days} days"),
).fetchall()
finally:
if conn:
conn.close()
return [dict(r) for r in rows]
def get_site_analytics_summary(site_id):
"""Get summary stats for a site: total submissions, spam, avg/day, etc."""
conn = None
try:
conn = get_db()
# Total submissions (non-spam)
row = conn.execute(
"SELECT COUNT(*) as total, SUM(spam_flag) as spam FROM submissions WHERE site_id = ?",
(site_id,),
).fetchone()
total = row["total"] or 0
spam = row["spam"] or 0
# This month stats
now = datetime.now(UTC)
this_month = now.strftime("%Y-%m")
this_month_row = conn.execute(
"""SELECT COALESCE(SUM(submission_count), 0) as subs,
COALESCE(SUM(spam_count), 0) as spam
FROM site_analytics
WHERE site_id = ? AND day LIKE ?""",
(site_id, f"{this_month}-%"),
).fetchone()
# Monthly trend (last 6 months)
trend = []
for i in range(5, -1, -1):
m = (now.replace(day=1) - __import__("datetime").timedelta(days=30 * i)).strftime("%Y-%m")
t_row = conn.execute(
"""SELECT COALESCE(SUM(submission_count), 0) as subs
FROM site_analytics
WHERE site_id = ? AND day LIKE ?""",
(site_id, f"{m}-%"),
).fetchone()
trend.append({"month": m, "submissions": t_row["subs"]})
finally:
if conn:
conn.close()
return {
"total_submissions": total,
"total_spam": spam,
"this_month_submissions": this_month_row["subs"],
"this_month_spam": this_month_row["spam"],
"monthly_trend": trend,
}
def site_geo_stats(site_id, days=30):
"""Top locations for a site's submissions.
Returns:
dict with 'top_cities' (list of {city, country, count}) and
'top_countries' (list of {country, count}).
"""
conn = None
try:
conn = get_db()
cutoff = f"-{days} days"
# Top cities
cities = conn.execute(
"""SELECT geo_city, geo_country, COUNT(*) as cnt
FROM submissions
WHERE site_id = ? AND spam_flag = 0
AND submitted_at >= datetime('now', ?)
AND geo_city IS NOT NULL AND geo_city != ''
GROUP BY geo_city, geo_country
ORDER BY cnt DESC LIMIT 5""",
(site_id, cutoff),
).fetchall()
# Top countries
countries = conn.execute(
"""SELECT geo_country, COUNT(*) as cnt
FROM submissions
WHERE site_id = ? AND spam_flag = 0
AND submitted_at >= datetime('now', ?)
AND geo_country IS NOT NULL AND geo_country != ''
GROUP BY geo_country
ORDER BY cnt DESC LIMIT 5""",
(site_id, cutoff),
).fetchall()
return {
"top_cities": [{"city": r["geo_city"], "country": r["geo_country"], "count": r["cnt"]} for r in cities],
"top_countries": [{"country": r["geo_country"], "count": r["cnt"]} for r in countries],
}
finally:
if conn:
conn.close()
def site_device_stats(site_id, days=30):
"""Device/browser/OS breakdown for a site's submissions.
Returns:
dict with 'devices', 'browsers', 'os' lists of {label, count, pct}.
"""
conn = None
try:
conn = get_db()
cutoff = f"-{days} days"
# Total non-spam submissions in range
total = conn.execute(
"""SELECT COUNT(*) as cnt FROM submissions
WHERE site_id = ? AND spam_flag = 0
AND submitted_at >= datetime('now', ?)""",
(site_id, cutoff),
).fetchone()["cnt"]
def _breakdown(col):
rows = conn.execute(
f"""SELECT {col} as label, COUNT(*) as cnt
FROM submissions
WHERE site_id = ? AND spam_flag = 0
AND submitted_at >= datetime('now', ?)
AND {col} IS NOT NULL AND {col} != ''
GROUP BY {col}
ORDER BY cnt DESC LIMIT 10""",
(site_id, cutoff),
).fetchall()
return [
{"label": r["label"], "count": r["cnt"], "pct": round(r["cnt"] / total * 100, 1) if total > 0 else 0}
for r in rows
]
return {
"devices": _breakdown("device_type"),
"browsers": _breakdown("browser"),
"os": _breakdown("os"),
}
finally:
if conn:
conn.close()
def site_impression_stats(site_id, days=30):
"""Form impression + conversion stats.
Returns:
dict with:
- impressions: total impressions in range
- submissions: total submissions in range
- conversion_rate: submissions/impressions * 100
- daily: list of {day, impressions, submissions} for charting
"""
conn = None
try:
conn = get_db()
cutoff = f"-{days} days"
# Total impressions
imp_row = conn.execute(
"""SELECT COUNT(*) as cnt FROM form_impressions
WHERE site_id = ? AND viewed_at >= datetime('now', ?)""",
(site_id, cutoff),
).fetchone()
impressions = imp_row["cnt"]
# Total submissions (non-spam)
sub_row = conn.execute(
"""SELECT COUNT(*) as cnt FROM submissions
WHERE site_id = ? AND spam_flag = 0
AND submitted_at >= datetime('now', ?)""",
(site_id, cutoff),
).fetchone()
submissions = sub_row["cnt"]
conversion_rate = round(submissions / impressions * 100, 1) if impressions > 0 else 0
# Daily breakdown — UNION of impressions and submissions by day
daily = conn.execute(
"""SELECT day, COALESCE(SUM(imps), 0) as impressions, COALESCE(SUM(subs), 0) as submissions
FROM (
SELECT date(viewed_at) as day, COUNT(*) as imps, 0 as subs
FROM form_impressions
WHERE site_id = ? AND viewed_at >= datetime('now', ?)
GROUP BY date(viewed_at)
UNION ALL
SELECT date(submitted_at) as day, 0 as imps, COUNT(*) as subs
FROM submissions
WHERE site_id = ? AND spam_flag = 0
AND submitted_at >= datetime('now', ?)
GROUP BY date(submitted_at)
)
GROUP BY day
ORDER BY day""",
(site_id, cutoff, site_id, cutoff),
).fetchall()
return {
"impressions": impressions,
"submissions": submissions,
"conversion_rate": conversion_rate,
"daily": [
{"day": r["day"], "impressions": r["impressions"], "submissions": r["submissions"]} for r in daily
],
}
finally:
if conn:
conn.close()
def get_user_analytics(user_id):
"""Get aggregated analytics for all sites owned by a user."""
conn = None
try:
conn = get_db()
sites = conn.execute("SELECT id FROM sites WHERE user_id = ?", (user_id,)).fetchall()
site_ids = [s["id"] for s in sites]
if not site_ids:
conn.close()
return {"total_submissions": 0, "total_spam": 0, "sites": 0}
placeholders = ",".join("?" * len(site_ids))
row = conn.execute(
f"""SELECT COUNT(*) as total, SUM(spam_flag) as spam
FROM submissions WHERE site_id IN ({placeholders})""",
site_ids,
).fetchone()
finally:
if conn:
conn.close()
return {
"total_submissions": row["total"] or 0,
"total_spam": row["spam"] or 0,
"sites": len(site_ids),
}
def get_team_analytics(team_id):
"""Get aggregated analytics for all sites in a team."""
conn = None
try:
conn = get_db()
# Get all site IDs for this team
sites = conn.execute("SELECT id FROM sites WHERE team_id = ?", (team_id,)).fetchall()
site_ids = [s["id"] for s in sites]
if not site_ids:
conn.close()
return {
"total_submissions": 0,
"total_spam": 0,
"this_month_submissions": 0,
"sites": 0,
"daily_data": [],
"monthly_trend": [],
"top_sites": [],
}
placeholders = ",".join("?" * len(site_ids))
# Total submissions
row = conn.execute(
f"""SELECT COUNT(*) as total, COALESCE(SUM(spam_flag), 0) as spam
FROM submissions WHERE site_id IN ({placeholders})""",
site_ids,
).fetchone()
total = row["total"] or 0
spam = row["spam"] or 0
# This month submissions
now = datetime.now(UTC)
this_month = now.strftime("%Y-%m")
month_row = conn.execute(
f"""SELECT COALESCE(SUM(submission_count), 0) as subs
FROM site_analytics WHERE site_id IN ({placeholders}) AND day LIKE ?""",
site_ids + [f"{this_month}-%"],
).fetchone()
# Daily data (last 30 days)
daily_rows = conn.execute(
f"""SELECT day, SUM(submission_count) as subs, SUM(spam_count) as spam
FROM site_analytics
WHERE site_id IN ({placeholders}) AND day >= date('now', '-30 days')
GROUP BY day ORDER BY day""",
site_ids,
).fetchall()
# Monthly trend (last 6 months)
trend = []
import datetime as dt_mod
for i in range(5, -1, -1):
m = (now.replace(day=1) - dt_mod.timedelta(days=30 * i)).strftime("%Y-%m")
t_row = conn.execute(
f"""SELECT COALESCE(SUM(submission_count), 0) as subs
FROM site_analytics
WHERE site_id IN ({placeholders}) AND day LIKE ?""",
site_ids + [f"{m}-%"],
).fetchone()
trend.append({"month": m, "submissions": t_row["subs"]})
# Top sites by submissions
site_stats = conn.execute(
"""SELECT s.id, s.name, COUNT(sub.id) as subs
FROM sites s
LEFT JOIN submissions sub ON sub.site_id = s.id
WHERE s.team_id = ?
GROUP BY s.id
ORDER BY subs DESC""",
(team_id,),
).fetchall()
finally:
if conn:
conn.close()
return {
"total_submissions": total,
"total_spam": spam,
"this_month_submissions": month_row["subs"],
"sites": len(site_ids),
"daily_data": [dict(r) for r in daily_rows],
"monthly_trend": trend,
"top_sites": [{"id": s["id"], "name": s["name"], "submissions": s["subs"]} for s in site_stats],
}
# ─── Email verification ──────────────────────────────────────────────────────
def get_spam_submissions(limit=50, offset=0):
"""Get flagged spam submissions for admin review."""
conn = None
try:
conn = get_db()
rows = conn.execute(
"""SELECT s.*, si.name as site_name, si.owner_email
FROM submissions s
JOIN sites si ON s.site_id = si.id
WHERE s.spam_flag = 1
ORDER BY s.submitted_at DESC
LIMIT ? OFFSET ?""",
(limit, offset),
).fetchall()
finally:
if conn:
conn.close()
return [dict(r) for r in rows]
def dismiss_spam(submission_id):
"""Mark a spam submission as not spam."""
conn = None
try:
conn = get_db()
conn.execute("UPDATE submissions SET spam_flag = 0 WHERE id = ?", (submission_id,))
conn.commit()
finally:
if conn:
conn.close()
def get_admin_stats():
"""Get platform-wide stats for admin dashboard."""
conn = None
try:
conn = get_db()
total_sites = conn.execute("SELECT COUNT(*) as c FROM sites").fetchone()["c"]
total_users = conn.execute("SELECT COUNT(*) as c FROM users").fetchone()["c"]
total_subs = conn.execute("SELECT COUNT(*) as c FROM submissions").fetchone()["c"]
total_spam = conn.execute("SELECT SUM(spam_flag) as c FROM submissions").fetchone()["c"] or 0
# Tier breakdown
tier_rows = conn.execute("SELECT tier, COUNT(*) as count FROM users GROUP BY tier").fetchall()
tiers = {r["tier"]: r["count"] for r in tier_rows}
# Verified users
verified = conn.execute("SELECT COUNT(*) as c FROM users WHERE email_verified = 1").fetchone()["c"]
finally:
if conn:
conn.close()
return {
"total_sites": total_sites,
"total_users": total_users,
"total_submissions": total_subs,
"total_spam": total_spam,
"tier_breakdown": tiers,
"verified_users": verified,
}
# ─── API Key functions ──────────────────────────────────────────────────────
API_KEY_PREFIX = "afk_live_"