"
)
return "".join(rows)
def dashboard_progress(label, percent, note=""):
p = max(0, min(100, int(percent or 0)))
note_html = f"
{html.escape(str(note))}
" if note else ""
return (
"
"
f"
{html.escape(str(label))}{p}%
"
f"
"
f"{note_html}"
"
"
)
def money(value, decimals=0):
cfg = vertical_config()
symbol = cfg.get("currency_symbol", "")
try:
num = float(value or 0)
except (TypeError, ValueError):
num = 0.0
if decimals <= 0 and abs(num - int(num)) < 0.000001:
return f"{symbol}{int(num):,}"
return f"{symbol}{num:,.{decimals}f}"
def ascii_text(value):
raw = "" if value is None else str(value)
return unicodedata.normalize("NFKD", raw).encode("ascii", "ignore").decode("ascii")
def pdf_escape(value):
return ascii_text(value).replace("\\", "\\\\").replace("(", "\\(").replace(")", "\\)")
def invoice_logo_url(setting_row):
return (setting_row["logo_url"] if setting_row and setting_row["logo_url"] else "").strip()
def get_partner_invoice_settings(conn, partner_id):
row = conn.execute(
"""SELECT pis.*, u.name user_name, u.email user_email, u.phone user_phone, p.company_name partner_company, p.city partner_city
FROM users u
JOIN partners p ON p.user_id=u.id
LEFT JOIN partner_invoice_settings pis ON pis.partner_id=u.id
WHERE u.id=?""",
(partner_id,),
).fetchone()
if not row:
return {
"partner_id": partner_id,
"company_name": "",
"contact_email": "",
"contact_phone": "",
"address": "",
"vat_number": "",
"logo_url": "",
"payment_terms": "Payment due on receipt",
"bank_details": "",
"footer_note": "",
"updated_at": now_iso(),
}
return {
"partner_id": partner_id,
"company_name": (row["company_name"] or row["partner_company"] or row["user_name"] or "").strip(),
"contact_email": (row["contact_email"] or row["user_email"] or "").strip(),
"contact_phone": (row["contact_phone"] or row["user_phone"] or "").strip(),
"address": (row["address"] or row["partner_city"] or "").strip(),
"vat_number": (row["vat_number"] or "").strip(),
"logo_url": invoice_logo_url(row),
"payment_terms": (row["payment_terms"] or "Payment due on receipt").strip(),
"bank_details": (row["bank_details"] or "").strip(),
"footer_note": (row["footer_note"] or "").strip(),
"updated_at": row["updated_at"] if row["updated_at"] else now_iso(),
}
def upsert_partner_invoice_settings(conn, partner_id, data):
conn.execute(
"""INSERT INTO partner_invoice_settings
(partner_id, company_name, contact_email, contact_phone, address, vat_number, logo_url, payment_terms, bank_details, footer_note, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(partner_id) DO UPDATE SET
company_name=excluded.company_name,
contact_email=excluded.contact_email,
contact_phone=excluded.contact_phone,
address=excluded.address,
vat_number=excluded.vat_number,
logo_url=excluded.logo_url,
payment_terms=excluded.payment_terms,
bank_details=excluded.bank_details,
footer_note=excluded.footer_note,
updated_at=excluded.updated_at""",
(
partner_id,
first(data, "invoice_company_name"),
first(data, "invoice_contact_email"),
first(data, "invoice_contact_phone"),
first(data, "invoice_address"),
first(data, "invoice_vat_number"),
first(data, "invoice_logo_url"),
first(data, "invoice_payment_terms", "Payment due on receipt"),
first(data, "invoice_bank_details"),
first(data, "invoice_footer_note"),
now_iso(),
),
)
def calculate_invoice_item_total(quantity, unit_price):
return round(max(0.0, quantity) * max(0.0, unit_price), 2)
def recalculate_invoice_totals(conn, invoice_id):
items = conn.execute(
"SELECT quantity, unit_price, line_total FROM invoice_items WHERE invoice_id=? ORDER BY sort_order ASC, id ASC",
(invoice_id,),
).fetchall()
subtotal = round(
sum(
float(r["line_total"] or calculate_invoice_item_total(float(r["quantity"] or 0), float(r["unit_price"] or 0)))
for r in items
),
2,
)
row = conn.execute(
"SELECT COALESCE(tax_amount,0) tax_amount, COALESCE(amount_paid,0) amount_paid FROM invoices WHERE id=?",
(invoice_id,),
).fetchone()
tax_amount = round(float(row["tax_amount"] or 0), 2) if row else 0.0
amount_paid = round(float(row["amount_paid"] or 0), 2) if row else 0.0
total_amount = round(subtotal + tax_amount, 2)
status = "fully_paid" if total_amount > 0 and amount_paid >= total_amount else ("partial_paid" if amount_paid > 0 else "unpaid")
conn.execute(
"UPDATE invoices SET subtotal=?, total_amount=?, status=?, amount=? WHERE id=?",
(subtotal, total_amount, status, int(round(total_amount)), invoice_id),
)
return subtotal, tax_amount, total_amount, amount_paid, status
def invoice_payment_summary(invoice_row):
total_amount = round(float(invoice_row["total_amount"] or 0), 2)
amount_paid = round(float(invoice_row["amount_paid"] or 0), 2)
outstanding = max(0.0, round(total_amount - amount_paid, 2))
return {
"total": total_amount,
"paid": amount_paid,
"outstanding": outstanding,
"status": normalize_invoice_status(invoice_row["status"]),
}
def build_simple_pdf(title, lines):
stream_lines = ["BT", "/F2 14 Tf", f"48 800 Td ({pdf_escape(title)}) Tj", "0 -20 Td", "/F1 11 Tf"]
for entry in lines:
if isinstance(entry, tuple):
text, font_name = entry
stream_lines.append(f"/{font_name} 11 Tf")
stream_lines.append(f"({pdf_escape(text)}) Tj")
stream_lines.append("0 -16 Td")
stream_lines.append("/F1 11 Tf")
continue
stream_lines.append(f"({pdf_escape(entry)}) Tj")
stream_lines.append("0 -16 Td")
stream_lines.append("ET")
stream = "\n".join(stream_lines).encode("latin-1", "ignore")
objects = []
objects.append(b"<< /Type /Catalog /Pages 2 0 R >>")
objects.append(b"<< /Type /Pages /Kids [3 0 R] /Count 1 >>")
objects.append(b"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 595 842] /Resources << /Font << /F1 4 0 R /F2 5 0 R >> >> /Contents 6 0 R >>")
objects.append(b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>")
objects.append(b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica-Bold >>")
objects.append(b"<< /Length " + str(len(stream)).encode("ascii") + b" >>\nstream\n" + stream + b"\nendstream")
header = b"%PDF-1.4\n%\xe2\xe3\xcf\xd3\n"
chunks = [header]
offsets = [0]
current = len(header)
for index, obj in enumerate(objects, start=1):
offsets.append(current)
block = f"{index} 0 obj\n".encode("ascii") + obj + b"\nendobj\n"
chunks.append(block)
current += len(block)
xref_offset = current
xref_lines = [f"0 {len(objects)+1}\n", "0000000000 65535 f \n"]
for off in offsets[1:]:
xref_lines.append(f"{off:010d} 00000 n \n")
trailer = (
"xref\n"
+ "".join(xref_lines)
+ f"trailer\n<< /Size {len(objects)+1} /Root 1 0 R >>\nstartxref\n{xref_offset}\n%%EOF"
).encode("ascii")
chunks.append(trailer)
return b"".join(chunks)
APP_TITLE = vertical_config()["app_title"]
def local_now():
if ZoneInfo:
try:
return datetime.now(ZoneInfo(APP_TIMEZONE))
except Exception:
pass
return datetime.utcnow() + timedelta(minutes=APP_TZ_OFFSET_MINUTES)
def password_hash(password, salt=None):
if salt is None:
salt = secrets.token_hex(16)
pwd = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt.encode("utf-8"), 150000).hex()
return f"{salt}${pwd}"
def password_verify(password, stored):
if "$" not in stored:
return False
salt, expected = stored.split("$", 1)
check = password_hash(password, salt).split("$", 1)[1]
return hmac.compare_digest(expected, check)
def generate_totp_secret():
return base64.b32encode(secrets.token_bytes(20)).decode("ascii").rstrip("=")
def _normalize_base32(secret):
s = (secret or "").strip().replace(" ", "").upper()
pad = (8 - (len(s) % 8)) % 8
return s + ("=" * pad)
def totp_code(secret, for_time=None, step=30, digits=6):
if for_time is None:
for_time = int(datetime.utcnow().timestamp())
key = base64.b32decode(_normalize_base32(secret), casefold=True)
counter = int(for_time // step)
msg = struct.pack(">Q", counter)
digest = hmac.new(key, msg, hashlib.sha1).digest()
offset = digest[-1] & 0x0F
dbc = struct.unpack(">I", digest[offset : offset + 4])[0] & 0x7FFFFFFF
mod = 10 ** digits
return str(dbc % mod).zfill(digits)
def verify_totp(secret, code, window=1):
raw = (code or "").strip()
norm = []
for ch in raw:
if ch.isdigit():
try:
norm.append(str(unicodedata.digit(ch)))
except Exception:
norm.append(ch)
code = "".join(norm)
if not re.fullmatch(r"\d{6}", code):
return False
now = int(datetime.utcnow().timestamp())
# Wider tolerance for shared-hosting clock drift (up to +/- 3 minutes).
drift_window = max(window, 6)
try:
for w in range(-drift_window, drift_window + 1):
if totp_code(secret, now + (w * 30)) == code:
return True
except Exception:
return False
return False
def db_connect():
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys = ON")
return conn
def backup_db():
if not os.path.exists(DB_PATH):
return None
if os.path.getsize(DB_PATH) == 0:
return None
os.makedirs(BACKUP_DIR, exist_ok=True)
ts = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
base = os.path.basename(DB_PATH)
target = os.path.join(BACKUP_DIR, f"{base}.{ts}.bak")
shutil.copy2(DB_PATH, target)
backups = sorted([x for x in os.listdir(BACKUP_DIR) if x.startswith(base + ".") and x.endswith(".bak")])
if len(backups) > BACKUP_KEEP:
for old in backups[: len(backups) - BACKUP_KEEP]:
try:
os.remove(os.path.join(BACKUP_DIR, old))
except OSError:
pass
return target
def table_has_column(conn, table, column):
cols = conn.execute(f"PRAGMA table_info({table})").fetchall()
return any(c["name"] == column for c in cols)
def table_exists(conn, table):
row = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name=?",
(table,),
).fetchone()
return bool(row)
def parse_date_any(s):
raw = (s or "").strip()
if not raw:
return None
raw = raw.replace("T", " ").split(" ")[0]
for fmt in ("%Y-%m-%d", "%d/%m/%Y", "%m/%d/%Y"):
try:
return datetime.strptime(raw, fmt).date()
except Exception:
continue
return None
def normalize_date_iso(s, fallback=None):
d = parse_date_any(s)
if d:
return d.isoformat()
if fallback:
return fallback
return datetime.utcnow().date().isoformat()
def expire_subscriptions(conn):
today = datetime.utcnow().date()
rows = conn.execute("SELECT id, end_date, status FROM subscriptions").fetchall()
for r in rows:
status = (r["status"] or "").lower()
if status not in {"active", "expired"}:
continue
end_d = parse_date_any(r["end_date"])
if end_d and end_d < today and status != "expired":
conn.execute("UPDATE subscriptions SET status='expired' WHERE id=?", (r["id"],))
def active_subscription(conn, partner_id):
expire_subscriptions(conn)
today = datetime.utcnow().date()
rows = conn.execute(
"""SELECT s.*, p.name AS package_name, p.price_monthly, p.price_annual
FROM subscriptions s
JOIN packages p ON p.id=s.package_id
WHERE s.partner_id=?
AND LOWER(TRIM(COALESCE(s.approval_status,'approved')))='approved'
AND LOWER(TRIM(COALESCE(s.status,'active'))) NOT IN ('cancelled','paused')
ORDER BY s.id DESC""",
(partner_id,),
).fetchall()
row = None
for cand in rows:
start_d = parse_date_any(cand["start_date"])
end_d = parse_date_any(cand["end_date"])
if not start_d or not end_d:
continue
if start_d <= today <= end_d:
row = cand
break
if row:
if (row["status"] or "").lower() == "expired":
conn.execute("UPDATE subscriptions SET status='active' WHERE id=?", (row["id"],))
conn.commit()
row = conn.execute(
"""SELECT s.*, p.name AS package_name, p.price_monthly, p.price_annual
FROM subscriptions s JOIN packages p ON p.id=s.package_id WHERE s.id=?""",
(row["id"],),
).fetchone()
billing_cycle = (row["billing_cycle"] if "billing_cycle" in row.keys() else "yearly") or "yearly"
if billing_cycle == "yearly":
current_ym = local_now().strftime("%Y-%m")
reset_ym = (row["credits_reset_month"] if "credits_reset_month" in row.keys() else "") or (row["start_date"][:7] if row["start_date"] else current_ym)
if reset_ym != current_ym:
conn.execute(
"UPDATE subscriptions SET credits_used=0, credits_remaining=credits_monthly, credits_reset_month=? WHERE id=?",
(current_ym, row["id"]),
)
conn.commit()
row = conn.execute(
"""SELECT s.*, p.name AS package_name, p.price_monthly, p.price_annual
FROM subscriptions s JOIN packages p ON p.id=s.package_id WHERE s.id=?""",
(row["id"],),
).fetchone()
return row
def next_subscription_start(conn, partner_id):
today = datetime.utcnow().date()
rows = conn.execute(
"""SELECT start_date, end_date, status, approval_status
FROM subscriptions
WHERE partner_id=?
ORDER BY id DESC""",
(partner_id,),
).fetchall()
for r in rows:
if ((r["approval_status"] or "approved").strip().lower() != "approved"):
continue
if ((r["status"] or "active").strip().lower() in {"cancelled", "paused"}):
continue
start_d = parse_date_any(r["start_date"])
end_d = parse_date_any(r["end_date"])
if not start_d or not end_d:
continue
if start_d > today:
return start_d
return None
def active_media_subscription(conn, partner_id):
today = datetime.utcnow().date().isoformat()
return conn.execute(
"""SELECT * FROM media_subscriptions
WHERE partner_id=? AND status='active' AND start_date <= ? AND end_date >= ?
ORDER BY id DESC LIMIT 1""",
(partner_id, today, today),
).fetchone()
def add_notification(conn, partner_id, title, message):
conn.execute(
"INSERT INTO notifications (partner_id, title, message, is_read, created_at) VALUES (?, ?, ?, 0, ?)",
(partner_id, title, message, now_iso()),
)
def media_cost_from_impressions(impressions, rate_per_1000):
try:
imp = max(0, int(impressions))
rate = max(0.0, float(rate_per_1000))
except (TypeError, ValueError):
return 0.0
return round((imp / 1000.0) * rate, 2)
def uploads_root():
root = os.path.join(os.path.dirname(__file__), "uploads")
os.makedirs(root, exist_ok=True)
return root
def private_uploads_root():
env_root = os.environ.get("VANLOCALUK_PRIVATE_UPLOADS", "").strip()
if env_root:
root = env_root
else:
root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "vanlocaluk_private_uploads"))
os.makedirs(root, exist_ok=True)
return root
def ensure_upload_access_blocks():
base = uploads_root()
for sub in ("subscription_proofs", "media_proofs"):
folder = os.path.join(base, sub)
os.makedirs(folder, exist_ok=True)
ht = os.path.join(folder, ".htaccess")
try:
if not os.path.exists(ht):
with open(ht, "w", encoding="utf-8") as f:
f.write("Require all denied\n")
except Exception:
pass
def save_uploaded_file(data, field_name, subdir):
b64_list = data.get(field_name, [])
if not b64_list:
return ""
b64 = (b64_list[0] or "").strip()
if not b64:
return ""
try:
blob = base64.b64decode(b64, validate=True)
except Exception:
return ""
if not blob:
return ""
filename = first(data, f"{field_name}__filename", "")
ext = os.path.splitext(filename)[1].lower()
if ext not in {".png", ".jpg", ".jpeg", ".webp", ".gif", ".pdf"}:
ext = ".bin"
is_private = subdir in {"subscription_proofs", "media_proofs"}
folder = os.path.join(private_uploads_root() if is_private else uploads_root(), subdir)
os.makedirs(folder, exist_ok=True)
unique = f"{subdir}_{datetime.utcnow().strftime('%Y%m%d%H%M%S')}_{secrets.token_hex(6)}{ext}"
abs_path = os.path.join(folder, unique)
with open(abs_path, "wb") as f:
f.write(blob)
return f"/protected-uploads/{subdir}/{unique}" if is_private else f"/uploads/{subdir}/{unique}"
def migrate_proof_files_to_private(conn):
old_root = uploads_root()
new_root = private_uploads_root()
if table_exists(conn, "subscriptions"):
rows = conn.execute("SELECT id, payment_proof_url FROM subscriptions WHERE COALESCE(payment_proof_url,'')<>''").fetchall()
for r in rows:
url = (r["payment_proof_url"] or "").strip()
if not url.startswith("/uploads/subscription_proofs/"):
continue
rel = url[len("/uploads/") :]
src = os.path.join(old_root, rel)
dst = os.path.join(new_root, rel)
os.makedirs(os.path.dirname(dst), exist_ok=True)
if os.path.exists(src):
try:
shutil.move(src, dst)
except Exception:
pass
conn.execute("UPDATE subscriptions SET payment_proof_url=? WHERE id=?", ("/protected-uploads/" + rel, r["id"]))
if table_exists(conn, "media_subscriptions"):
rows = conn.execute("SELECT id, payment_proof_path FROM media_subscriptions WHERE COALESCE(payment_proof_path,'')<>''").fetchall()
for r in rows:
url = (r["payment_proof_path"] or "").strip()
if not url.startswith("/uploads/media_proofs/"):
continue
rel = url[len("/uploads/") :]
src = os.path.join(old_root, rel)
dst = os.path.join(new_root, rel)
os.makedirs(os.path.dirname(dst), exist_ok=True)
if os.path.exists(src):
try:
shutil.move(src, dst)
except Exception:
pass
conn.execute("UPDATE media_subscriptions SET payment_proof_path=? WHERE id=?", ("/protected-uploads/" + rel, r["id"]))
def add_days_iso(date_iso, days):
try:
base = datetime.strptime((date_iso or "").strip(), "%Y-%m-%d").date()
return (base + timedelta(days=days)).isoformat()
except Exception:
return (datetime.utcnow().date() + timedelta(days=days)).isoformat()
def init_db():
if AUTO_BACKUP_ENABLED and os.path.exists(DB_PATH):
backup_db()
ensure_upload_access_blocks()
conn = db_connect()
with open(SCHEMA_PATH, "r", encoding="utf-8") as f:
conn.executescript(f.read())
if not table_has_column(conn, "subscriptions", "finalized_amount"):
conn.execute("ALTER TABLE subscriptions ADD COLUMN finalized_amount INTEGER NOT NULL DEFAULT 0")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "created_by_admin_id"):
conn.execute("ALTER TABLE partners ADD COLUMN created_by_admin_id INTEGER")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "vertical_type"):
conn.execute("ALTER TABLE partners ADD COLUMN vertical_type TEXT NOT NULL DEFAULT 'property'")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "service_categories_json"):
conn.execute("ALTER TABLE partners ADD COLUMN service_categories_json TEXT NOT NULL DEFAULT '[]'")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "coverage_postcodes_json"):
conn.execute("ALTER TABLE partners ADD COLUMN coverage_postcodes_json TEXT NOT NULL DEFAULT '[]'")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "supports_domestic"):
conn.execute("ALTER TABLE partners ADD COLUMN supports_domestic INTEGER NOT NULL DEFAULT 1")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "supports_commercial"):
conn.execute("ALTER TABLE partners ADD COLUMN supports_commercial INTEGER NOT NULL DEFAULT 0")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "same_day_available"):
conn.execute("ALTER TABLE partners ADD COLUMN same_day_available INTEGER NOT NULL DEFAULT 0")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "weekend_available"):
conn.execute("ALTER TABLE partners ADD COLUMN weekend_available INTEGER NOT NULL DEFAULT 0")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "preferred_lead_types_json"):
conn.execute("ALTER TABLE partners ADD COLUMN preferred_lead_types_json TEXT NOT NULL DEFAULT '[]'")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "vehicle_types"):
conn.execute("ALTER TABLE partners ADD COLUMN vehicle_types TEXT")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "crew_size"):
conn.execute("ALTER TABLE partners ADD COLUMN crew_size TEXT")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "lead_budget_min"):
conn.execute("ALTER TABLE partners ADD COLUMN lead_budget_min INTEGER NOT NULL DEFAULT 0")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "lead_budget_max"):
conn.execute("ALTER TABLE partners ADD COLUMN lead_budget_max INTEGER NOT NULL DEFAULT 0")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "insurance_details"):
conn.execute("ALTER TABLE partners ADD COLUMN insurance_details TEXT")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "license_details"):
conn.execute("ALTER TABLE partners ADD COLUMN license_details TEXT")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "vertical_type"):
conn.execute("ALTER TABLE partners ADD COLUMN vertical_type TEXT NOT NULL DEFAULT 'property'")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "service_categories_json"):
conn.execute("ALTER TABLE partners ADD COLUMN service_categories_json TEXT NOT NULL DEFAULT '[]'")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "coverage_postcodes_json"):
conn.execute("ALTER TABLE partners ADD COLUMN coverage_postcodes_json TEXT NOT NULL DEFAULT '[]'")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "supports_domestic"):
conn.execute("ALTER TABLE partners ADD COLUMN supports_domestic INTEGER NOT NULL DEFAULT 1")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "supports_commercial"):
conn.execute("ALTER TABLE partners ADD COLUMN supports_commercial INTEGER NOT NULL DEFAULT 0")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "same_day_available"):
conn.execute("ALTER TABLE partners ADD COLUMN same_day_available INTEGER NOT NULL DEFAULT 0")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "weekend_available"):
conn.execute("ALTER TABLE partners ADD COLUMN weekend_available INTEGER NOT NULL DEFAULT 0")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "preferred_lead_types_json"):
conn.execute("ALTER TABLE partners ADD COLUMN preferred_lead_types_json TEXT NOT NULL DEFAULT '[]'")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "vehicle_types"):
conn.execute("ALTER TABLE partners ADD COLUMN vehicle_types TEXT")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "crew_size"):
conn.execute("ALTER TABLE partners ADD COLUMN crew_size TEXT")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "lead_budget_min"):
conn.execute("ALTER TABLE partners ADD COLUMN lead_budget_min INTEGER NOT NULL DEFAULT 0")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "lead_budget_max"):
conn.execute("ALTER TABLE partners ADD COLUMN lead_budget_max INTEGER NOT NULL DEFAULT 0")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "insurance_details"):
conn.execute("ALTER TABLE partners ADD COLUMN insurance_details TEXT")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "license_details"):
conn.execute("ALTER TABLE partners ADD COLUMN license_details TEXT")
if table_exists(conn, "subscriptions") and not table_has_column(conn, "subscriptions", "assigned_by_admin_id"):
conn.execute("ALTER TABLE subscriptions ADD COLUMN assigned_by_admin_id INTEGER")
if table_exists(conn, "subscriptions") and not table_has_column(conn, "subscriptions", "approval_status"):
conn.execute("ALTER TABLE subscriptions ADD COLUMN approval_status TEXT NOT NULL DEFAULT 'approved'")
if table_exists(conn, "subscriptions") and not table_has_column(conn, "subscriptions", "payment_proof_url"):
conn.execute("ALTER TABLE subscriptions ADD COLUMN payment_proof_url TEXT")
if table_exists(conn, "subscriptions") and not table_has_column(conn, "subscriptions", "approved_by_admin_id"):
conn.execute("ALTER TABLE subscriptions ADD COLUMN approved_by_admin_id INTEGER")
if table_exists(conn, "subscriptions") and not table_has_column(conn, "subscriptions", "approved_at"):
conn.execute("ALTER TABLE subscriptions ADD COLUMN approved_at TEXT")
if table_exists(conn, "subscriptions") and not table_has_column(conn, "subscriptions", "billing_cycle"):
conn.execute("ALTER TABLE subscriptions ADD COLUMN billing_cycle TEXT NOT NULL DEFAULT 'yearly'")
if table_exists(conn, "subscriptions") and not table_has_column(conn, "subscriptions", "credits_reset_month"):
conn.execute("ALTER TABLE subscriptions ADD COLUMN credits_reset_month TEXT")
if table_exists(conn, "media_subscriptions") and not table_has_column(conn, "media_subscriptions", "assigned_by_admin_id"):
conn.execute("ALTER TABLE media_subscriptions ADD COLUMN assigned_by_admin_id INTEGER")
if table_exists(conn, "media_subscriptions") and not table_has_column(conn, "media_subscriptions", "approval_status"):
conn.execute("ALTER TABLE media_subscriptions ADD COLUMN approval_status TEXT NOT NULL DEFAULT 'approved'")
if table_exists(conn, "media_subscriptions") and not table_has_column(conn, "media_subscriptions", "payment_proof_path"):
conn.execute("ALTER TABLE media_subscriptions ADD COLUMN payment_proof_path TEXT")
if table_exists(conn, "media_subscriptions") and not table_has_column(conn, "media_subscriptions", "approved_by_admin_id"):
conn.execute("ALTER TABLE media_subscriptions ADD COLUMN approved_by_admin_id INTEGER")
if table_exists(conn, "media_subscriptions") and not table_has_column(conn, "media_subscriptions", "approved_at"):
conn.execute("ALTER TABLE media_subscriptions ADD COLUMN approved_at TEXT")
if not table_has_column(conn, "leads", "lead_type"):
conn.execute("ALTER TABLE leads ADD COLUMN lead_type TEXT NOT NULL DEFAULT 'platform'")
if not table_has_column(conn, "leads", "target_partner_id"):
conn.execute("ALTER TABLE leads ADD COLUMN target_partner_id INTEGER")
if not table_has_column(conn, "leads", "lead_bucket"):
conn.execute("ALTER TABLE leads ADD COLUMN lead_bucket TEXT NOT NULL DEFAULT 'system_pool'")
if not table_has_column(conn, "leads", "vertical_type"):
conn.execute("ALTER TABLE leads ADD COLUMN vertical_type TEXT NOT NULL DEFAULT 'property'")
if not table_has_column(conn, "leads", "lead_score"):
conn.execute("ALTER TABLE leads ADD COLUMN lead_score INTEGER NOT NULL DEFAULT 0")
if not table_has_column(conn, "leads", "service_category"):
conn.execute("ALTER TABLE leads ADD COLUMN service_category TEXT")
if table_exists(conn, "moving_enquiries") and not table_has_column(conn, "moving_enquiries", "job_type"):
conn.execute("ALTER TABLE moving_enquiries ADD COLUMN job_type TEXT")
if table_exists(conn, "moving_enquiries") and not table_has_column(conn, "moving_enquiries", "vehicle_size"):
conn.execute("ALTER TABLE moving_enquiries ADD COLUMN vehicle_size TEXT")
if table_exists(conn, "moving_enquiries") and not table_has_column(conn, "moving_enquiries", "loading_help_needed"):
conn.execute("ALTER TABLE moving_enquiries ADD COLUMN loading_help_needed INTEGER NOT NULL DEFAULT 0")
if table_exists(conn, "moving_enquiries") and not table_has_column(conn, "moving_enquiries", "parking_notes"):
conn.execute("ALTER TABLE moving_enquiries ADD COLUMN parking_notes TEXT")
if table_exists(conn, "moving_enquiries") and not table_has_column(conn, "moving_enquiries", "permit_required"):
conn.execute("ALTER TABLE moving_enquiries ADD COLUMN permit_required INTEGER NOT NULL DEFAULT 0")
if table_exists(conn, "moving_enquiries") and not table_has_column(conn, "moving_enquiries", "waste_type"):
conn.execute("ALTER TABLE moving_enquiries ADD COLUMN waste_type TEXT")
if table_exists(conn, "moving_enquiries") and not table_has_column(conn, "moving_enquiries", "access_notes"):
conn.execute("ALTER TABLE moving_enquiries ADD COLUMN access_notes TEXT")
if table_exists(conn, "move_requirements") and not table_has_column(conn, "move_requirements", "job_type"):
conn.execute("ALTER TABLE move_requirements ADD COLUMN job_type TEXT")
if table_exists(conn, "move_requirements") and not table_has_column(conn, "move_requirements", "vehicle_size"):
conn.execute("ALTER TABLE move_requirements ADD COLUMN vehicle_size TEXT")
if table_exists(conn, "move_requirements") and not table_has_column(conn, "move_requirements", "loading_help_needed"):
conn.execute("ALTER TABLE move_requirements ADD COLUMN loading_help_needed INTEGER NOT NULL DEFAULT 0")
if table_exists(conn, "move_requirements") and not table_has_column(conn, "move_requirements", "parking_notes"):
conn.execute("ALTER TABLE move_requirements ADD COLUMN parking_notes TEXT")
if table_exists(conn, "move_requirements") and not table_has_column(conn, "move_requirements", "permit_required"):
conn.execute("ALTER TABLE move_requirements ADD COLUMN permit_required INTEGER NOT NULL DEFAULT 0")
if table_exists(conn, "move_requirements") and not table_has_column(conn, "move_requirements", "waste_type"):
conn.execute("ALTER TABLE move_requirements ADD COLUMN waste_type TEXT")
if table_exists(conn, "move_requirements") and not table_has_column(conn, "move_requirements", "access_notes"):
conn.execute("ALTER TABLE move_requirements ADD COLUMN access_notes TEXT")
conn.execute(
"""CREATE TABLE IF NOT EXISTS notifications (
id INTEGER PRIMARY KEY AUTOINCREMENT,
partner_id INTEGER NOT NULL,
title TEXT NOT NULL,
message TEXT NOT NULL,
is_read INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL,
FOREIGN KEY(partner_id) REFERENCES users(id)
)"""
)
conn.execute(
"""CREATE TABLE IF NOT EXISTS admin_profiles (
user_id INTEGER PRIMARY KEY,
admin_role TEXT NOT NULL DEFAULT 'super_admin',
created_at TEXT NOT NULL,
FOREIGN KEY(user_id) REFERENCES users(id)
)"""
)
if not table_has_column(conn, "admin_profiles", "manager_user_id"):
conn.execute("ALTER TABLE admin_profiles ADD COLUMN manager_user_id INTEGER")
if not table_has_column(conn, "admin_profiles", "twofa_secret"):
conn.execute("ALTER TABLE admin_profiles ADD COLUMN twofa_secret TEXT")
if not table_has_column(conn, "admin_profiles", "twofa_enabled"):
conn.execute("ALTER TABLE admin_profiles ADD COLUMN twofa_enabled INTEGER NOT NULL DEFAULT 0")
if not table_has_column(conn, "admin_profiles", "twofa_grace_used"):
conn.execute("ALTER TABLE admin_profiles ADD COLUMN twofa_grace_used INTEGER NOT NULL DEFAULT 0")
# Safety net: ensure feature tables exist even if deployed schema.sql is older.
ensure_feature_tables(conn)
cfg = vertical_config()
default_landing = {
"hero_title": cfg["hero_title"],
"hero_subtitle": cfg["hero_subtitle"],
"hero_video_url": "",
"section_how_title": cfg["section_how_title"],
"section_how_text": cfg["section_how_text"],
"buyer_form_title": cfg["buyer_form_title"],
"logo_url": "",
"ad_top_html": "",
"ad_mid_html": "",
"ad_sidebar_html": "",
"buyer_ad_html": "",
}
for k, v in default_landing.items():
conn.execute(
"INSERT OR IGNORE INTO landing_settings (key, value, updated_at) VALUES (?, ?, ?)",
(k, v, now_iso()),
)
expire_subscriptions(conn)
conn.execute("UPDATE subscriptions SET approval_status='approved' WHERE approval_status IS NULL OR approval_status=''")
conn.execute("UPDATE subscriptions SET billing_cycle='yearly' WHERE billing_cycle IS NULL OR billing_cycle=''")
conn.execute("UPDATE subscriptions SET credits_reset_month=substr(start_date,1,7) WHERE credits_reset_month IS NULL OR credits_reset_month=''")
conn.execute("UPDATE media_subscriptions SET approval_status='approved' WHERE approval_status IS NULL OR approval_status=''")
migrate_proof_files_to_private(conn)
conn.commit()
existing = conn.execute("SELECT COUNT(*) AS c FROM users").fetchone()["c"]
if existing == 0:
admin_hash = password_hash("admin123")
partner_hash = password_hash("partner123")
cfg = vertical_config()
conn.execute(
"INSERT INTO users (name, email, phone, password_hash, role, status, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
("Admin User", "admin@vanlocaluk.test", "+447000000001", admin_hash, "admin", "active", now_iso()),
)
admin_user = conn.execute("SELECT id FROM users WHERE email = ?", ("admin@vanlocaluk.test",)).fetchone()["id"]
conn.execute(
"INSERT INTO admin_profiles (user_id, admin_role, created_at) VALUES (?, 'super_admin', ?)",
(admin_user, now_iso()),
)
ensure_admin_permissions_rows(conn, admin_user, full_access=True)
conn.execute(
"INSERT INTO users (name, email, phone, password_hash, role, status, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
(f"Demo {cfg['operator_singular']}", "partner@vanlocaluk.test", "+447000000002", partner_hash, "partner", "active", now_iso()),
)
partner_user = conn.execute("SELECT id FROM users WHERE email = ?", ("partner@vanlocaluk.test",)).fetchone()["id"]
conn.execute(
"""INSERT INTO partners
(user_id, partner_type, company_name, city, areas_json, address, verified_status, created_by_admin_id, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(partner_user, "provider", "VanLocalUK Demo Movers", "London", json.dumps(["SW1", "E14", "N1"]), "1 Demo Road", "verified", admin_user, now_iso()),
)
conn.execute(
"INSERT INTO packages (name, monthly_credits, price_monthly, price_annual, cities_allowed, features_json, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
(f"Growth {cfg['operator_singular']}", 30, 299, 2999, 1, json.dumps({"csv_export": True, "invalid_replacement": True}), now_iso()),
)
package_id = conn.execute("SELECT id FROM packages LIMIT 1").fetchone()["id"]
start = datetime.utcnow().date().isoformat()
end = (datetime.utcnow().date() + timedelta(days=30)).isoformat()
conn.execute(
"""INSERT INTO subscriptions
(partner_id, package_id, start_date, end_date, status, credits_monthly, credits_used, credits_remaining, finalized_amount, assigned_by_admin_id, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(partner_user, package_id, start, end, "active", 30, 0, 30, 29999, admin_user, now_iso()),
)
conn.execute(
"""INSERT INTO leads
(name, phone, email, city, area, budget_min, budget_max, property_type, purpose, timeframe, source, verified_by_admin, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
"James Carter",
"+447700900200",
"james@example.co.uk",
"SW1",
"BS1",
500,
1000,
"Flat",
"buy",
"2026-03-20",
"Website Form",
1,
now_iso(),
),
)
conn.commit()
else:
admins = conn.execute("SELECT id FROM users WHERE role='admin'").fetchall()
super_admin = conn.execute(
"""SELECT ap.user_id
FROM admin_profiles ap JOIN users u ON u.id=ap.user_id
WHERE u.role='admin' AND ap.admin_role='super_admin'
ORDER BY ap.user_id LIMIT 1"""
).fetchone()
default_owner = super_admin["user_id"] if super_admin else None
for a in admins:
exists_ap = conn.execute("SELECT user_id FROM admin_profiles WHERE user_id=?", (a["id"],)).fetchone()
if not exists_ap:
conn.execute(
"INSERT INTO admin_profiles (user_id, admin_role, created_at) VALUES (?, 'staff', ?)",
(a["id"], now_iso()),
)
role_row = conn.execute("SELECT admin_role FROM admin_profiles WHERE user_id=?", (a["id"],)).fetchone()
ensure_admin_permissions_rows(conn, a["id"], full_access=(role_row and role_row["admin_role"] == "super_admin"))
if default_owner:
conn.execute(
"UPDATE partners SET created_by_admin_id=? WHERE created_by_admin_id IS NULL",
(default_owner,),
)
conn.execute(
"""UPDATE subscriptions
SET assigned_by_admin_id=?
WHERE assigned_by_admin_id IS NULL""",
(default_owner,),
)
conn.execute(
"""UPDATE media_subscriptions
SET assigned_by_admin_id=?
WHERE assigned_by_admin_id IS NULL""",
(default_owner,),
)
conn.commit()
conn.close()
def ensure_initialized():
global _APP_INITIALIZED
if not _APP_INITIALIZED:
init_db()
_APP_INITIALIZED = True
def parse_post_data(handler):
length = int(handler.headers.get("Content-Length", "0"))
content_type = handler.headers.get("Content-Type", "")
if "multipart/form-data" in content_type:
raw = handler.rfile.read(length)
boundary_match = re.search(r"boundary=([^;]+)", content_type)
if not boundary_match:
return {}
boundary = boundary_match.group(1).strip().strip('"').encode("utf-8")
delimiter = b"--" + boundary
end_delimiter = delimiter + b"--"
data = {}
parts = raw.split(delimiter)
for part in parts:
if not part or part == b"--" or part.startswith(end_delimiter):
continue
part = part.lstrip(b"\r\n")
if b"\r\n\r\n" not in part:
continue
header_blob, body = part.split(b"\r\n\r\n", 1)
body = body.rstrip(b"\r\n")
headers_text = header_blob.decode("utf-8", errors="ignore")
dispo_match = re.search(r'Content-Disposition:\s*form-data;\s*name="([^"]+)"(?:;\s*filename="([^"]*)")?', headers_text, re.IGNORECASE)
if not dispo_match:
continue
field_name = dispo_match.group(1)
file_name = dispo_match.group(2)
if file_name is not None and file_name != "":
b64_value = base64.b64encode(body).decode("ascii")
data.setdefault(field_name, []).append(b64_value)
data.setdefault(f"{field_name}__filename", []).append(file_name)
else:
field_value = body.decode("utf-8", errors="ignore")
data.setdefault(field_name, []).append(field_value)
return data
raw = handler.rfile.read(length).decode("utf-8")
return urllib.parse.parse_qs(raw, keep_blank_values=True)
def first(data, key, default=""):
value = data.get(key, default)
if isinstance(value, (list, tuple)):
value = value[0] if value else default
if value is None:
value = default
return str(value).strip()
def to_float(value, default=0.0):
try:
return float(value)
except (TypeError, ValueError):
return default
def to_int(value, default=0):
try:
if value is None:
return default
if isinstance(value, str):
cleaned = value.strip().replace(",", "")
if cleaned == "":
return default
if re.fullmatch(r"-?\d+", cleaned):
return int(cleaned)
if re.fullmatch(r"-?\d+\.\d+", cleaned):
return int(float(cleaned))
return default
return int(value)
except (TypeError, ValueError):
return default
def partner_id_from_email(conn, email):
if not email:
return None
row = conn.execute(
"SELECT id FROM users WHERE role='partner' AND lower(email)=lower(?)",
(email.strip(),),
).fetchone()
return row["id"] if row else None
def safe_text(value):
return html.escape("" if value is None else str(value))
def normalize_phone(value):
if not value:
return ""
return re.sub(r"[^0-9]", "", str(value))
def normalize_purpose(value):
v = (value or "").strip().lower()
if v in {"buy", "buyer", "investment", "invest", "investor", "purchase"}:
return "buy"
if v in {"rent", "rental", "tenant"}:
return "rent"
return "buy"
def parse_csv_list(value):
return [x.strip() for x in (value or "").split(",") if x.strip()]
def unique_list(values):
seen = set()
out = []
for item in values:
if item not in seen:
seen.add(item)
out.append(item)
return out
def normalize_yes_no(value):
return 1 if (value or "").strip().lower() in {"1", "true", "yes", "on"} else 0
def normalize_postcode_area(value):
raw = re.sub(r"\s+", " ", (value or "").upper()).strip()
if not raw:
return ""
return raw.split(" ", 1)[0]
def normalize_service_category(value):
raw = (value or "").strip().lower().replace("-", "_").replace("&", "and")
raw = re.sub(r"[^a-z0-9]+", "_", raw).strip("_")
if raw in SERVICE_CATEGORIES:
return raw
alias_map = {
"home_removal": "home_removals",
"office_removal": "office_removals",
"man_with_a_van": "man_with_van",
"man_van": "man_with_van",
"packing": "packing_services",
"waste": "waste_removal",
}
return alias_map.get(raw, raw if raw in SERVICE_CATEGORIES else "home_removals")
def format_service_category(value):
mapping = {
"home_removals": "Home removals",
"office_removals": "Office removals",
"man_with_van": "Man with van",
"waste_removal": "Waste removal",
"storage": "Storage",
"packing_services": "Packing services",
}
return mapping.get((value or "").strip(), (value or "").replace("_", " ").title())
def humanize_token(value):
return (value or "").replace("_", " ").replace("-", " ").title() or "-"
def budget_range_to_values(raw_value):
raw = (raw_value or "").strip()
presets = {
"under_250": (0, 250),
"250_500": (250, 500),
"500_1000": (500, 1000),
"1000_2000": (1000, 2000),
"2000_plus": (2000, 0),
}
if raw in presets:
return presets[raw]
nums = [to_int(x, 0) for x in re.findall(r"\d+", raw)]
if len(nums) >= 2:
return nums[0], nums[1]
if len(nums) == 1:
return nums[0], 0
return 0, 0
def send_simple_email(to_email, subject, body_text):
host = os.environ.get("VANLOCALUK_SMTP_HOST", "localhost")
port = to_int(os.environ.get("VANLOCALUK_SMTP_PORT", "25"), 25)
username = os.environ.get("VANLOCALUK_SMTP_USER", "")
password = os.environ.get("VANLOCALUK_SMTP_PASS", "")
from_email = os.environ.get("VANLOCALUK_SMTP_FROM", "no-reply@vanlocaluk.co.uk")
use_tls = os.environ.get("VANLOCALUK_SMTP_TLS", "0") == "1"
msg = EmailMessage()
msg["Subject"] = subject
msg["From"] = from_email
msg["To"] = to_email
msg.set_content(body_text)
try:
with smtplib.SMTP(host, port, timeout=10) as smtp:
if use_tls:
smtp.starttls()
if username:
smtp.login(username, password)
smtp.send_message(msg)
return True
except Exception:
return False
def landing_setting(conn, key, default_value):
row = conn.execute("SELECT value FROM landing_settings WHERE key=?", (key,)).fetchone()
return row["value"] if row else default_value
def ensure_vertical_tables(conn):
conn.executescript(
"""
CREATE TABLE IF NOT EXISTS moving_enquiries (
id INTEGER PRIMARY KEY AUTOINCREMENT,
enquiry_id INTEGER,
lead_id INTEGER,
from_postcode TEXT NOT NULL,
to_postcode TEXT NOT NULL,
move_date TEXT,
service_type TEXT NOT NULL,
property_type TEXT,
bedrooms TEXT,
move_size TEXT,
floor_number TEXT,
lift_access INTEGER NOT NULL DEFAULT 0,
packing_needed INTEGER NOT NULL DEFAULT 0,
dismantling_needed INTEGER NOT NULL DEFAULT 0,
storage_needed INTEGER NOT NULL DEFAULT 0,
job_type TEXT,
vehicle_size TEXT,
loading_help_needed INTEGER NOT NULL DEFAULT 0,
parking_notes TEXT,
permit_required INTEGER NOT NULL DEFAULT 0,
waste_type TEXT,
access_notes TEXT,
special_items TEXT,
budget_range TEXT,
contact_name TEXT NOT NULL,
contact_phone TEXT NOT NULL,
contact_email TEXT,
created_at TEXT NOT NULL,
FOREIGN KEY(enquiry_id) REFERENCES enquiries(id),
FOREIGN KEY(lead_id) REFERENCES leads(id)
);
CREATE TABLE IF NOT EXISTS move_requirements (
id INTEGER PRIMARY KEY AUTOINCREMENT,
lead_id INTEGER NOT NULL UNIQUE,
from_postcode TEXT NOT NULL,
to_postcode TEXT NOT NULL,
move_date TEXT,
service_type TEXT NOT NULL,
property_type TEXT,
bedrooms TEXT,
move_size TEXT,
floor_number TEXT,
lift_access INTEGER NOT NULL DEFAULT 0,
packing_needed INTEGER NOT NULL DEFAULT 0,
dismantling_needed INTEGER NOT NULL DEFAULT 0,
storage_needed INTEGER NOT NULL DEFAULT 0,
job_type TEXT,
vehicle_size TEXT,
loading_help_needed INTEGER NOT NULL DEFAULT 0,
parking_notes TEXT,
permit_required INTEGER NOT NULL DEFAULT 0,
waste_type TEXT,
access_notes TEXT,
special_items TEXT,
budget_min INTEGER NOT NULL DEFAULT 0,
budget_max INTEGER NOT NULL DEFAULT 0,
is_commercial INTEGER NOT NULL DEFAULT 0,
same_day_requested INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL,
FOREIGN KEY(lead_id) REFERENCES leads(id)
);
CREATE TABLE IF NOT EXISTS partner_service_categories (
id INTEGER PRIMARY KEY AUTOINCREMENT,
partner_id INTEGER NOT NULL,
service_category TEXT NOT NULL,
created_at TEXT NOT NULL,
FOREIGN KEY(partner_id) REFERENCES users(id),
UNIQUE(partner_id, service_category)
);
CREATE TABLE IF NOT EXISTS moving_lead_matches (
id INTEGER PRIMARY KEY AUTOINCREMENT,
lead_id INTEGER NOT NULL,
partner_id INTEGER NOT NULL,
match_score INTEGER NOT NULL DEFAULT 0,
reason_summary TEXT,
status TEXT NOT NULL DEFAULT 'suggested',
created_at TEXT NOT NULL,
FOREIGN KEY(lead_id) REFERENCES leads(id),
FOREIGN KEY(partner_id) REFERENCES users(id)
);
CREATE TABLE IF NOT EXISTS service_areas (
id INTEGER PRIMARY KEY AUTOINCREMENT,
partner_id INTEGER NOT NULL,
postcode_prefix TEXT NOT NULL,
area_label TEXT,
created_at TEXT NOT NULL,
FOREIGN KEY(partner_id) REFERENCES users(id),
UNIQUE(partner_id, postcode_prefix)
);
"""
)
def upsert_partner_vertical_profile(conn, partner_id, data):
categories = unique_list([normalize_service_category(x) for x in parse_csv_list(first(data, "service_categories"))])
coverage_postcodes = unique_list([normalize_postcode_area(x) for x in parse_csv_list(first(data, "coverage_postcodes")) if normalize_postcode_area(x)])
preferred_types = unique_list(parse_csv_list(first(data, "preferred_lead_types")))
lead_budget_min = to_int(first(data, "lead_budget_min", "0"), 0)
lead_budget_max = to_int(first(data, "lead_budget_max", "0"), 0)
conn.execute(
"""UPDATE partners
SET vertical_type=?, service_categories_json=?, coverage_postcodes_json=?, supports_domestic=?, supports_commercial=?,
same_day_available=?, weekend_available=?, preferred_lead_types_json=?, vehicle_types=?, crew_size=?,
lead_budget_min=?, lead_budget_max=?, insurance_details=?, license_details=?
WHERE user_id=?""",
(
VERTICAL_TYPE,
json.dumps(categories),
json.dumps(coverage_postcodes),
normalize_yes_no(first(data, "supports_domestic", "1")),
normalize_yes_no(first(data, "supports_commercial")),
normalize_yes_no(first(data, "same_day_available")),
normalize_yes_no(first(data, "weekend_available")),
json.dumps(preferred_types),
first(data, "vehicle_types"),
first(data, "crew_size"),
lead_budget_min,
lead_budget_max,
first(data, "insurance_details"),
first(data, "license_details"),
partner_id,
),
)
conn.execute("DELETE FROM partner_service_categories WHERE partner_id=?", (partner_id,))
for category in categories:
conn.execute(
"INSERT OR IGNORE INTO partner_service_categories (partner_id, service_category, created_at) VALUES (?, ?, ?)",
(partner_id, category, now_iso()),
)
conn.execute("DELETE FROM service_areas WHERE partner_id=?", (partner_id,))
for postcode in coverage_postcodes:
conn.execute(
"INSERT OR IGNORE INTO service_areas (partner_id, postcode_prefix, area_label, created_at) VALUES (?, ?, ?, ?)",
(partner_id, postcode, postcode, now_iso()),
)
def save_move_requirements(conn, lead_id, data, enquiry_id=None):
service_type = normalize_service_category(first(data, "service_type", "home_removals"))
budget_min, budget_max = budget_range_to_values(first(data, "budget_range"))
move_date = normalize_date_iso(first(data, "move_date"), "")
from_postcode = normalize_postcode_area(first(data, "from_postcode"))
to_postcode = normalize_postcode_area(first(data, "to_postcode"))
property_type = first(data, "property_type") or first(data, "move_property_type")
bedrooms = first(data, "bedrooms")
move_size = first(data, "move_size") or bedrooms
floor_number = first(data, "floor_number")
lift_access = normalize_yes_no(first(data, "lift_access"))
packing_needed = normalize_yes_no(first(data, "packing_needed"))
dismantling_needed = normalize_yes_no(first(data, "dismantling_needed"))
storage_needed = normalize_yes_no(first(data, "storage_needed"))
job_type = first(data, "job_type")
vehicle_size = first(data, "vehicle_size")
loading_help_needed = normalize_yes_no(first(data, "loading_help_needed"))
parking_notes = first(data, "parking_notes")
permit_required = normalize_yes_no(first(data, "permit_required"))
waste_type = first(data, "waste_type")
access_notes = first(data, "access_notes")
special_items = first(data, "special_items")
is_commercial = 1 if service_type == "office_removals" or (property_type or "").strip().lower() == "office" else 0
same_day_requested = 1 if move_date == datetime.utcnow().date().isoformat() else 0
conn.execute(
"""INSERT INTO move_requirements
(lead_id, from_postcode, to_postcode, move_date, service_type, property_type, bedrooms, move_size, floor_number,
lift_access, packing_needed, dismantling_needed, storage_needed, job_type, vehicle_size, loading_help_needed,
parking_notes, permit_required, waste_type, access_notes, special_items, budget_min, budget_max,
is_commercial, same_day_requested, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(lead_id) DO UPDATE SET
from_postcode=excluded.from_postcode,
to_postcode=excluded.to_postcode,
move_date=excluded.move_date,
service_type=excluded.service_type,
property_type=excluded.property_type,
bedrooms=excluded.bedrooms,
move_size=excluded.move_size,
floor_number=excluded.floor_number,
lift_access=excluded.lift_access,
packing_needed=excluded.packing_needed,
dismantling_needed=excluded.dismantling_needed,
storage_needed=excluded.storage_needed,
job_type=excluded.job_type,
vehicle_size=excluded.vehicle_size,
loading_help_needed=excluded.loading_help_needed,
parking_notes=excluded.parking_notes,
permit_required=excluded.permit_required,
waste_type=excluded.waste_type,
access_notes=excluded.access_notes,
special_items=excluded.special_items,
budget_min=excluded.budget_min,
budget_max=excluded.budget_max,
is_commercial=excluded.is_commercial,
same_day_requested=excluded.same_day_requested""",
(
lead_id,
from_postcode,
to_postcode,
move_date,
service_type,
property_type,
bedrooms,
move_size,
floor_number,
lift_access,
packing_needed,
dismantling_needed,
storage_needed,
job_type,
vehicle_size,
loading_help_needed,
parking_notes,
permit_required,
waste_type,
access_notes,
special_items,
budget_min,
budget_max,
is_commercial,
same_day_requested,
now_iso(),
),
)
if enquiry_id:
conn.execute(
"""INSERT INTO moving_enquiries
(enquiry_id, lead_id, from_postcode, to_postcode, move_date, service_type, property_type, bedrooms, move_size, floor_number,
lift_access, packing_needed, dismantling_needed, storage_needed, job_type, vehicle_size, loading_help_needed,
parking_notes, permit_required, waste_type, access_notes, special_items, budget_range, contact_name, contact_phone, contact_email, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
enquiry_id,
lead_id,
from_postcode,
to_postcode,
move_date,
service_type,
property_type,
bedrooms,
move_size,
floor_number,
lift_access,
packing_needed,
dismantling_needed,
storage_needed,
job_type,
vehicle_size,
loading_help_needed,
parking_notes,
permit_required,
waste_type,
access_notes,
special_items,
first(data, "budget_range"),
first(data, "full_name") or first(data, "name"),
first(data, "phone"),
first(data, "email"),
now_iso(),
),
)
lead_score = 25
if from_postcode and to_postcode:
lead_score += 25
if move_date:
lead_score += 20
if special_items:
lead_score += 10
if budget_min or budget_max:
lead_score += 10
if packing_needed or dismantling_needed or storage_needed:
lead_score += 10
if loading_help_needed or permit_required or parking_notes or access_notes:
lead_score += 5
conn.execute(
"UPDATE leads SET vertical_type=?, lead_score=?, service_category=?, city=?, area=?, budget_min=?, budget_max=?, property_type=?, timeframe=? WHERE id=?",
(
VERTICAL_TYPE,
lead_score,
service_type,
from_postcode or first(data, "city") or "UK",
to_postcode or first(data, "area") or "UK",
budget_min,
budget_max,
property_type or format_service_category(service_type),
move_date or first(data, "timeframe") or "Flexible",
lead_id,
),
)
def moving_requirement_summary(conn, lead_id):
row = conn.execute("SELECT * FROM move_requirements WHERE lead_id=?", (lead_id,)).fetchone()
if not row:
return None
return {
"from_postcode": row["from_postcode"],
"to_postcode": row["to_postcode"],
"move_date": row["move_date"],
"service_type": format_service_category(row["service_type"]),
"property_type": row["property_type"] or "-",
"move_size": row["move_size"] or row["bedrooms"] or "-",
"job_type": humanize_token(row["job_type"]),
"vehicle_size": row["vehicle_size"] or "-",
"floor_number": row["floor_number"] or "-",
"lift_access": "Yes" if row["lift_access"] else "No",
"packing_needed": "Yes" if row["packing_needed"] else "No",
"dismantling_needed": "Yes" if row["dismantling_needed"] else "No",
"storage_needed": "Yes" if row["storage_needed"] else "No",
"loading_help_needed": "Yes" if row["loading_help_needed"] else "No",
"permit_required": "Yes" if row["permit_required"] else "No",
"parking_notes": row["parking_notes"] or "-",
"waste_type": row["waste_type"] or "-",
"access_notes": row["access_notes"] or "-",
"special_items": row["special_items"] or "-",
}
def moving_match_candidates(conn, lead_id, required_credits=1):
req = conn.execute("SELECT * FROM move_requirements WHERE lead_id=?", (lead_id,)).fetchone()
if not req:
return []
today = datetime.utcnow().date().isoformat()
from_area = normalize_postcode_area(req["from_postcode"])
category = req["service_type"]
rows = conn.execute(
"""SELECT u.id AS partner_id, p.city, p.same_day_available, p.weekend_available, p.supports_domestic, p.supports_commercial,
p.lead_budget_min, p.lead_budget_max, COALESCE(p.preferred_lead_types_json,'[]') preferred_lead_types_json,
COALESCE(p.vehicle_types,'') vehicle_types, COALESCE(p.license_details,'') license_details,
EXISTS(SELECT 1 FROM partner_service_categories psc WHERE psc.partner_id=u.id AND psc.service_category=?) AS category_match,
EXISTS(SELECT 1 FROM service_areas sa WHERE sa.partner_id=u.id AND sa.postcode_prefix=?) AS postcode_match,
COALESCE((SELECT MAX(s.credits_remaining)
FROM subscriptions s
WHERE s.partner_id=u.id AND s.status='active' AND s.start_date<=? AND s.end_date>=?), 0) AS available_credits
FROM users u
JOIN partners p ON p.user_id=u.id
WHERE u.role='partner' AND u.status='active' AND COALESCE(p.vertical_type,?)=?""",
(category, from_area, today, today, VERTICAL_TYPE, VERTICAL_TYPE),
).fetchall()
candidates = []
for row in rows:
if (row["available_credits"] or 0) < required_credits:
continue
preferred_types = [str(x).strip().lower() for x in parse_csv_list(row["preferred_lead_types_json"])]
if row["preferred_lead_types_json"].strip().startswith("["):
try:
preferred_types = [str(x).strip().lower() for x in json.loads(row["preferred_lead_types_json"] or "[]")]
except Exception:
preferred_types = [str(x).strip().lower() for x in parse_csv_list(row["preferred_lead_types_json"])]
vehicle_types = (row["vehicle_types"] or "").strip().lower()
requested_vehicle = (req["vehicle_size"] or "").strip().lower()
requested_job = (req["job_type"] or "").strip().lower()
score = 0
reasons = []
if row["category_match"]:
score += 45
reasons.append("service category")
if row["postcode_match"]:
score += 30
reasons.append("postcode coverage")
if req["is_commercial"] and row["supports_commercial"]:
score += 10
reasons.append("commercial support")
if not req["is_commercial"] and row["supports_domestic"]:
score += 10
reasons.append("domestic support")
if req["same_day_requested"] and row["same_day_available"]:
score += 10
reasons.append("same-day availability")
if requested_job == "weekend_move" and row["weekend_available"]:
score += 8
reasons.append("weekend availability")
if requested_job and requested_job in preferred_types:
score += 6
reasons.append("preferred lead type")
if requested_vehicle and requested_vehicle in vehicle_types:
score += 8
reasons.append("vehicle fit")
if req["service_type"] == "waste_removal" and row["license_details"]:
score += 8
reasons.append("waste licence")
if row["lead_budget_max"] <= 0 or req["budget_min"] <= row["lead_budget_max"]:
score += 5
reasons.append("budget fit")
if score <= 0:
continue
candidates.append((row["partner_id"], score, ", ".join(reasons)))
candidates.sort(key=lambda item: (-item[1], item[0]))
return candidates
def ensure_feature_tables(conn):
conn.executescript(
"""
CREATE TABLE IF NOT EXISTS password_reset_requests (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
email TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending' CHECK(status IN ('pending','completed','rejected')),
admin_id INTEGER,
admin_notes TEXT,
created_at TEXT NOT NULL,
resolved_at TEXT,
FOREIGN KEY(user_id) REFERENCES users(id),
FOREIGN KEY(admin_id) REFERENCES users(id)
);
CREATE TABLE IF NOT EXISTS morning_meetings (
id INTEGER PRIMARY KEY AUTOINCREMENT,
partner_id INTEGER NOT NULL,
meeting_date TEXT NOT NULL,
agenda TEXT NOT NULL,
tasks TEXT,
status TEXT NOT NULL DEFAULT 'planned',
created_at TEXT NOT NULL,
FOREIGN KEY(partner_id) REFERENCES users(id)
);
CREATE TABLE IF NOT EXISTS admin_permissions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
module TEXT NOT NULL,
can_view INTEGER NOT NULL DEFAULT 1,
can_add INTEGER NOT NULL DEFAULT 0,
can_edit INTEGER NOT NULL DEFAULT 0,
can_delete INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL,
UNIQUE(user_id, module),
FOREIGN KEY(user_id) REFERENCES users(id)
);
CREATE TABLE IF NOT EXISTS landing_settings (
id INTEGER PRIMARY KEY AUTOINCREMENT,
key TEXT NOT NULL UNIQUE,
value TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS partner_badges (
id INTEGER PRIMARY KEY AUTOINCREMENT,
partner_name TEXT NOT NULL,
badge_label TEXT NOT NULL,
city TEXT NOT NULL,
image_url TEXT,
sort_order INTEGER NOT NULL DEFAULT 0,
status TEXT NOT NULL DEFAULT 'active',
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS enquiries (
id INTEGER PRIMARY KEY AUTOINCREMENT,
enquiry_type TEXT NOT NULL CHECK(enquiry_type IN ('partner_application','buyer_enquiry')),
full_name TEXT NOT NULL,
phone TEXT NOT NULL,
email TEXT,
city TEXT,
area TEXT,
property_type TEXT,
budget_min INTEGER,
budget_max INTEGER,
message TEXT,
status TEXT NOT NULL DEFAULT 'new' CHECK(status IN ('new','contacted','qualified','closed','rejected')),
admin_notes TEXT,
email_sent INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS admin_login_challenges (
token TEXT PRIMARY KEY,
user_id INTEGER NOT NULL,
expires_at TEXT NOT NULL,
created_at TEXT NOT NULL,
FOREIGN KEY(user_id) REFERENCES users(id)
);
CREATE TABLE IF NOT EXISTS admin_todos (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
task_text TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
created_at TEXT NOT NULL,
completed_at TEXT,
FOREIGN KEY(user_id) REFERENCES users(id)
);
CREATE TABLE IF NOT EXISTS partner_invoice_settings (
partner_id INTEGER PRIMARY KEY,
company_name TEXT,
contact_email TEXT,
contact_phone TEXT,
address TEXT,
vat_number TEXT,
logo_url TEXT,
payment_terms TEXT,
bank_details TEXT,
footer_note TEXT,
updated_at TEXT NOT NULL,
FOREIGN KEY(partner_id) REFERENCES users(id)
);
CREATE TABLE IF NOT EXISTS invoice_items (
id INTEGER PRIMARY KEY AUTOINCREMENT,
invoice_id INTEGER NOT NULL,
service_name TEXT NOT NULL,
quantity REAL NOT NULL DEFAULT 1,
unit_price REAL NOT NULL DEFAULT 0,
line_total REAL NOT NULL DEFAULT 0,
notes TEXT,
sort_order INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL,
FOREIGN KEY(invoice_id) REFERENCES invoices(id)
);
"""
)
ensure_vertical_tables(conn)
def ensure_runtime_migrations(conn):
# Non-destructive safety migrations for live environments with older DB schema.
ensure_upload_access_blocks()
ensure_feature_tables(conn)
if table_exists(conn, "subscriptions") and not table_has_column(conn, "subscriptions", "finalized_amount"):
conn.execute("ALTER TABLE subscriptions ADD COLUMN finalized_amount INTEGER NOT NULL DEFAULT 0")
if table_exists(conn, "partners") and not table_has_column(conn, "partners", "created_by_admin_id"):
conn.execute("ALTER TABLE partners ADD COLUMN created_by_admin_id INTEGER")
if table_exists(conn, "subscriptions") and not table_has_column(conn, "subscriptions", "assigned_by_admin_id"):
conn.execute("ALTER TABLE subscriptions ADD COLUMN assigned_by_admin_id INTEGER")
if table_exists(conn, "subscriptions") and not table_has_column(conn, "subscriptions", "approval_status"):
conn.execute("ALTER TABLE subscriptions ADD COLUMN approval_status TEXT NOT NULL DEFAULT 'approved'")
if table_exists(conn, "subscriptions") and not table_has_column(conn, "subscriptions", "payment_proof_url"):
conn.execute("ALTER TABLE subscriptions ADD COLUMN payment_proof_url TEXT")
if table_exists(conn, "subscriptions") and not table_has_column(conn, "subscriptions", "approved_by_admin_id"):
conn.execute("ALTER TABLE subscriptions ADD COLUMN approved_by_admin_id INTEGER")
if table_exists(conn, "subscriptions") and not table_has_column(conn, "subscriptions", "approved_at"):
conn.execute("ALTER TABLE subscriptions ADD COLUMN approved_at TEXT")
if table_exists(conn, "subscriptions") and not table_has_column(conn, "subscriptions", "billing_cycle"):
conn.execute("ALTER TABLE subscriptions ADD COLUMN billing_cycle TEXT NOT NULL DEFAULT 'yearly'")
if table_exists(conn, "subscriptions") and not table_has_column(conn, "subscriptions", "credits_reset_month"):
conn.execute("ALTER TABLE subscriptions ADD COLUMN credits_reset_month TEXT")
if table_exists(conn, "media_subscriptions") and not table_has_column(conn, "media_subscriptions", "assigned_by_admin_id"):
conn.execute("ALTER TABLE media_subscriptions ADD COLUMN assigned_by_admin_id INTEGER")
if table_exists(conn, "media_subscriptions") and not table_has_column(conn, "media_subscriptions", "approval_status"):
conn.execute("ALTER TABLE media_subscriptions ADD COLUMN approval_status TEXT NOT NULL DEFAULT 'approved'")
if table_exists(conn, "media_subscriptions") and not table_has_column(conn, "media_subscriptions", "payment_proof_path"):
conn.execute("ALTER TABLE media_subscriptions ADD COLUMN payment_proof_path TEXT")
if table_exists(conn, "media_subscriptions") and not table_has_column(conn, "media_subscriptions", "approved_by_admin_id"):
conn.execute("ALTER TABLE media_subscriptions ADD COLUMN approved_by_admin_id INTEGER")
if table_exists(conn, "media_subscriptions") and not table_has_column(conn, "media_subscriptions", "approved_at"):
conn.execute("ALTER TABLE media_subscriptions ADD COLUMN approved_at TEXT")
if table_exists(conn, "leads") and not table_has_column(conn, "leads", "lead_type"):
conn.execute("ALTER TABLE leads ADD COLUMN lead_type TEXT NOT NULL DEFAULT 'platform'")
if table_exists(conn, "leads") and not table_has_column(conn, "leads", "target_partner_id"):
conn.execute("ALTER TABLE leads ADD COLUMN target_partner_id INTEGER")
if table_exists(conn, "leads") and not table_has_column(conn, "leads", "lead_bucket"):
conn.execute("ALTER TABLE leads ADD COLUMN lead_bucket TEXT NOT NULL DEFAULT 'system_pool'")
if table_exists(conn, "leads") and not table_has_column(conn, "leads", "vertical_type"):
conn.execute("ALTER TABLE leads ADD COLUMN vertical_type TEXT NOT NULL DEFAULT 'property'")
if table_exists(conn, "leads") and not table_has_column(conn, "leads", "lead_score"):
conn.execute("ALTER TABLE leads ADD COLUMN lead_score INTEGER NOT NULL DEFAULT 0")
if table_exists(conn, "leads") and not table_has_column(conn, "leads", "service_category"):
conn.execute("ALTER TABLE leads ADD COLUMN service_category TEXT")
if table_exists(conn, "invoices") and not table_has_column(conn, "invoices", "invoice_number"):
conn.execute("ALTER TABLE invoices ADD COLUMN invoice_number TEXT")
if table_exists(conn, "invoices") and not table_has_column(conn, "invoices", "client_id"):
conn.execute("ALTER TABLE invoices ADD COLUMN client_id INTEGER")
if table_exists(conn, "invoices") and not table_has_column(conn, "invoices", "lead_assignment_id"):
conn.execute("ALTER TABLE invoices ADD COLUMN lead_assignment_id INTEGER")
if table_exists(conn, "invoices") and not table_has_column(conn, "invoices", "customer_name"):
conn.execute("ALTER TABLE invoices ADD COLUMN customer_name TEXT")
if table_exists(conn, "invoices") and not table_has_column(conn, "invoices", "customer_phone"):
conn.execute("ALTER TABLE invoices ADD COLUMN customer_phone TEXT")
if table_exists(conn, "invoices") and not table_has_column(conn, "invoices", "customer_email"):
conn.execute("ALTER TABLE invoices ADD COLUMN customer_email TEXT")
if table_exists(conn, "invoices") and not table_has_column(conn, "invoices", "customer_address"):
conn.execute("ALTER TABLE invoices ADD COLUMN customer_address TEXT")
if table_exists(conn, "invoices") and not table_has_column(conn, "invoices", "issue_date"):
conn.execute("ALTER TABLE invoices ADD COLUMN issue_date TEXT")
if table_exists(conn, "invoices") and not table_has_column(conn, "invoices", "due_date"):
conn.execute("ALTER TABLE invoices ADD COLUMN due_date TEXT")
if table_exists(conn, "invoices") and not table_has_column(conn, "invoices", "subtotal"):
conn.execute("ALTER TABLE invoices ADD COLUMN subtotal REAL NOT NULL DEFAULT 0")
if table_exists(conn, "invoices") and not table_has_column(conn, "invoices", "tax_amount"):
conn.execute("ALTER TABLE invoices ADD COLUMN tax_amount REAL NOT NULL DEFAULT 0")
if table_exists(conn, "invoices") and not table_has_column(conn, "invoices", "total_amount"):
conn.execute("ALTER TABLE invoices ADD COLUMN total_amount REAL NOT NULL DEFAULT 0")
if table_exists(conn, "invoices") and not table_has_column(conn, "invoices", "amount_paid"):
conn.execute("ALTER TABLE invoices ADD COLUMN amount_paid REAL NOT NULL DEFAULT 0")
if table_exists(conn, "invoices") and not table_has_column(conn, "invoices", "notes"):
conn.execute("ALTER TABLE invoices ADD COLUMN notes TEXT")
if table_exists(conn, "invoices") and not table_has_column(conn, "invoices", "share_token"):
conn.execute("ALTER TABLE invoices ADD COLUMN share_token TEXT")
if table_exists(conn, "admin_profiles") and not table_has_column(conn, "admin_profiles", "manager_user_id"):
conn.execute("ALTER TABLE admin_profiles ADD COLUMN manager_user_id INTEGER")
if table_exists(conn, "admin_profiles") and not table_has_column(conn, "admin_profiles", "twofa_secret"):
conn.execute("ALTER TABLE admin_profiles ADD COLUMN twofa_secret TEXT")
if table_exists(conn, "admin_profiles") and not table_has_column(conn, "admin_profiles", "twofa_enabled"):
conn.execute("ALTER TABLE admin_profiles ADD COLUMN twofa_enabled INTEGER NOT NULL DEFAULT 0")
if table_exists(conn, "admin_profiles") and not table_has_column(conn, "admin_profiles", "twofa_grace_used"):
conn.execute("ALTER TABLE admin_profiles ADD COLUMN twofa_grace_used INTEGER NOT NULL DEFAULT 0")
if table_exists(conn, "subscriptions"):
conn.execute("UPDATE subscriptions SET billing_cycle='yearly' WHERE billing_cycle IS NULL OR billing_cycle=''")
conn.execute("UPDATE subscriptions SET credits_reset_month=substr(start_date,1,7) WHERE credits_reset_month IS NULL OR credits_reset_month=''")
conn.execute("UPDATE subscriptions SET approval_status='approved' WHERE approval_status IS NULL OR approval_status=''")
today = datetime.utcnow().date().isoformat()
conn.execute(
"""UPDATE subscriptions
SET status='active'
WHERE COALESCE(approval_status,'approved')='approved'
AND start_date<=? AND end_date>=?
AND COALESCE(status,'active')='expired'""",
(today, today),
)
if table_exists(conn, "media_subscriptions"):
conn.execute("UPDATE media_subscriptions SET approval_status='approved' WHERE approval_status IS NULL OR approval_status=''")
migrate_proof_files_to_private(conn)
def ensure_admin_permissions_rows(conn, user_id, full_access=False):
for module in ADMIN_MODULES:
exists = conn.execute(
"SELECT id FROM admin_permissions WHERE user_id=? AND module=?",
(user_id, module),
).fetchone()
if not exists:
conn.execute(
"""INSERT INTO admin_permissions
(user_id, module, can_view, can_add, can_edit, can_delete, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(
user_id,
module,
1,
1 if full_access else 0,
1 if full_access else 0,
1 if full_access else 0,
now_iso(),
),
)
def page(title, body, user=None):
nav = ""
cfg = vertical_config()
if user:
if user["role"] == "admin":
conn = db_connect()
role_row = conn.execute("SELECT admin_role FROM admin_profiles WHERE user_id=?", (user["id"],)).fetchone()
role_name = role_row["admin_role"] if role_row else "staff"
allowed = set()
if role_name == "super_admin":
allowed = set(ADMIN_MODULES)
else:
rows = conn.execute(
"SELECT module FROM admin_permissions WHERE user_id=? AND can_view=1",
(user["id"],),
).fetchall()
allowed = set([r["module"] for r in rows])
conn.close()
links = [
("dashboard", "/admin/dashboard", "Dashboard"),
("users", "/admin/users", "Users"),
("partners", "/admin/partners", cfg["operator_plural"]),
("packages", "/admin/packages", "Packages"),
("subscriptions", "/admin/subscriptions", "Subscriptions"),
("media", "/admin/media", "Media Ads"),
("leads", "/admin/leads", "Leads"),
("invalid_reports", "/admin/invalid-reports", "Invalid Reports"),
("tickets", "/admin/tickets", "Tickets"),
("enquiries", "/admin/enquiries", "Enquiries"),
("landing_cms", "/admin/landing-cms", "Landing CMS"),
("profile", "/admin/profile", "Profile"),
]
nav = "".join([f'{html.escape(t)}' for m, u, t in links if m in allowed])
else:
if is_removals_vertical():
nav = (
'Dashboard'
'Shift Briefing'
'Operations Diary'
'Enquiries'
'Fresh Leads'
'Customers'
'Won Jobs'
'Invoices'
'Media Performance'
'Billing & Credits'
'Support'
'Profile'
)
else:
nav = (
'Dashboard'
'Morning Meeting'
'Diary'
'Data Bank'
'Leads'
'Fresh Leads'
'Clients'
'Deal'
'Invoices'
'Media Performance'
'Credits & Package'
'Support'
'Profile'
)
nav += 'Logout'
menu_button = '' if user else ""
side_html = f'' if user else ""
overlay_html = '' if user else ""
app_title = cfg["app_title"]
top_subtitle = ""
if user:
top_subtitle = "Admin Portal" if user["role"] == "admin" else cfg["operator_portal_label"]
top_brand = brand_logo_html(cfg["brand_name"], top_subtitle, light=True, compact=True)
menu_script = """
""" if user else ""
return f"""
{html.escape(title)} | {html.escape(app_title)}
{menu_button}{top_brand}
{overlay_html}
{side_html}
{body}
{menu_script}
"""
def svg_bar_chart(title, labels, values, color=None):
color = color or theme_primary()
width = 560
height = 220
max_v = max(values) if values and max(values) > 0 else 1
step = max(1, int(width / (len(values) + 1))) if values else 60
grad_id = f"barGrad{secrets.token_hex(4)}"
bars = []
texts = []
grid = []
for tick in range(5):
y = 170 - int((130 / 4) * tick)
grid.append(f"")
for i, v in enumerate(values):
x = 45 + (i * step)
bar_h = int((v / max_v) * 130)
y = 170 - bar_h
bars.append(f"")
texts.append(f"{html.escape(str(labels[i]))[:10]}")
texts.append(f"{int(v)}")
return f"""
{html.escape(title)}
"""
def svg_line_chart(title, labels, values_a, values_b=None, color_a=None, color_b=None):
color_a = color_a or theme_primary()
color_b = color_b or theme_secondary()
width = 620
height = 240
pad = 36
n = max(len(values_a), len(values_b) if values_b else 0, 2)
max_v = max(values_a + (values_b or [0])) if (values_a or values_b) else 1
max_v = max(max_v, 1)
fill_id = f"lineFill{secrets.token_hex(4)}"
def points(vals):
pts = []
for i, v in enumerate(vals):
x = pad + (i * ((width - pad * 2) / (n - 1)))
y = (height - pad) - ((v / max_v) * (height - pad * 2))
pts.append((x, y))
return pts
pa = points(values_a or [0, 0])
pb = points(values_b or []) if values_b else []
poly_a = " ".join([f"{x:.1f},{y:.1f}" for x, y in pa])
poly_b = " ".join([f"{x:.1f},{y:.1f}" for x, y in pb])
area_a = " ".join([f"{x:.1f},{y:.1f}" for x, y in pa] + [f"{pa[-1][0]:.1f},{height-pad}", f"{pa[0][0]:.1f},{height-pad}"])
x_labels = []
for i, lbl in enumerate(labels[:n]):
x = pad + (i * ((width - pad * 2) / (n - 1)))
x_labels.append(f"{html.escape(str(lbl))}")
grid = []
for tick in range(5):
y = pad + int(((height - pad * 2) / 4) * tick)
grid.append(f"")
dots_a = "".join([f"" for x, y in pa])
dots_b = "".join([f"" for x, y in pb])
return f"""
{html.escape(title)}
"""
def svg_donut_chart(title, percent, color=None):
color = color or theme_primary()
p = max(0, min(100, int(percent)))
r = 70
c = 2 * 3.14159 * r
dash = c * (p / 100)
rest = c - dash
return f"""
{html.escape(title)}
"""
class AppHandler(BaseHTTPRequestHandler):
script_name = ""
def log_message(self, fmt, *args):
return
def _prefix_url(self, url):
if not url or not url.startswith("/"):
return url
prefix = (getattr(self, "script_name", "") or "").rstrip("/")
if not prefix:
return url
if url.startswith(prefix + "/") or url == prefix:
return url
return f"{prefix}{url}"
def absolute_url(self, path):
prefixed = self._prefix_url(path)
scheme = self.headers.get("X-Forwarded-Proto", "https").split(",")[0].strip() or "https"
host = (self.headers.get("X-Forwarded-Host") or self.headers.get("Host") or "").split(",")[0].strip()
if not host:
return prefixed
return f"{scheme}://{host}{prefixed}"
def send_html(self, body, status=200):
prefix = (getattr(self, "script_name", "") or "").rstrip("/")
if prefix:
body = body.replace("href='/", f"href='{prefix}/")
body = body.replace('href="/', f'href="{prefix}/')
body = body.replace("action='/", f"action='{prefix}/")
body = body.replace('action="/', f'action="{prefix}/')
payload = body.encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)
def send_bytes(self, payload, content_type="application/octet-stream", status=200, extra_headers=None):
self.send_response(status)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(payload)))
if extra_headers:
for k, v in extra_headers.items():
self.send_header(k, v)
self.end_headers()
self.wfile.write(payload)
def redirect(self, location):
location = self._prefix_url(location)
self.send_response(302)
self.send_header("Location", location)
self.end_headers()
def current_user(self):
raw = self.headers.get("Cookie", "")
c = cookies.SimpleCookie()
c.load(raw)
sid = c.get("session_id")
if not sid:
return None
conn = db_connect()
try:
ensure_runtime_migrations(conn)
conn.commit()
except Exception:
pass
row = conn.execute(
"""SELECT u.* FROM sessions s
JOIN users u ON u.id=s.user_id
WHERE s.session_id=? AND s.expires_at > ? AND u.status='active'""",
(sid.value, now_iso()),
).fetchone()
conn.close()
return row
def require_auth(self, role=None, allow_expired=False):
user = self.current_user()
if not user:
self.redirect(ADMIN_LOGIN_PATH if role == "admin" else "/login")
return None
if role and user["role"] != role:
self.send_html(page("Unauthorized", "
Unauthorized
", user), 403)
return None
if user["role"] == "admin":
conn = db_connect()
self.admin_2fa_profile(conn, user["id"])
self.admin_role_of(conn, user["id"])
conn.close()
# 2FA enforcement happens only during login challenge flow.
if user["role"] == "partner" and not allow_expired:
conn = db_connect()
sub = active_subscription(conn, user["id"])
if sub:
self.auto_assign_partner_leads(conn, user["id"], sub)
conn.commit()
conn.close()
if not sub:
body = """
Package Expired
Your package has expired. Renew your package to access the portal again.
"""
self.send_html(page("Renew Package", body, user), 403)
return None
return user
def create_session(self, user_id):
sid = secrets.token_urlsafe(32)
exp = (datetime.utcnow() + timedelta(hours=SESSION_HOURS)).replace(microsecond=0).isoformat()
conn = db_connect()
try:
conn.execute("INSERT INTO sessions (session_id, user_id, expires_at, created_at) VALUES (?, ?, ?, ?)", (sid, user_id, exp, now_iso()))
conn.commit()
finally:
conn.close()
return sid
def clear_session(self):
raw = self.headers.get("Cookie", "")
c = cookies.SimpleCookie()
c.load(raw)
sid = c.get("session_id")
if sid:
conn = db_connect()
conn.execute("DELETE FROM sessions WHERE session_id=?", (sid.value,))
conn.commit()
conn.close()
def cookie_value(self, name):
raw = self.headers.get("Cookie", "")
c = cookies.SimpleCookie()
c.load(raw)
v = c.get(name)
return v.value if v else ""
def create_admin_login_challenge(self, user_id):
token = secrets.token_urlsafe(32)
expires = (datetime.utcnow() + timedelta(minutes=5)).replace(microsecond=0).isoformat()
conn = db_connect()
conn.execute("DELETE FROM admin_login_challenges WHERE user_id=?", (user_id,))
conn.execute(
"INSERT INTO admin_login_challenges (token, user_id, expires_at, created_at) VALUES (?, ?, ?, ?)",
(token, user_id, expires, now_iso()),
)
conn.commit()
conn.close()
return token
def consume_admin_login_challenge(self, token):
if not token:
return None
conn = db_connect()
row = conn.execute(
"SELECT * FROM admin_login_challenges WHERE token=? AND expires_at>?",
(token, now_iso()),
).fetchone()
if row:
conn.execute("DELETE FROM admin_login_challenges WHERE token=?", (token,))
conn.commit()
conn.close()
return row
def admin_2fa_profile(self, conn, user_id):
row = conn.execute(
"SELECT twofa_secret, COALESCE(twofa_enabled,0) twofa_enabled, COALESCE(twofa_grace_used,0) twofa_grace_used FROM admin_profiles WHERE user_id=?",
(user_id,),
).fetchone()
if not row:
conn.execute(
"INSERT INTO admin_profiles (user_id, admin_role, manager_user_id, twofa_secret, twofa_enabled, twofa_grace_used, created_at) VALUES (?, 'staff', NULL, '', 0, 0, ?)",
(user_id, now_iso()),
)
conn.commit()
row = conn.execute(
"SELECT twofa_secret, COALESCE(twofa_enabled,0) twofa_enabled, COALESCE(twofa_grace_used,0) twofa_grace_used FROM admin_profiles WHERE user_id=?",
(user_id,),
).fetchone()
return row
def admin_role_of(self, conn, user_id):
row = conn.execute("SELECT admin_role FROM admin_profiles WHERE user_id=?", (user_id,)).fetchone()
return row["admin_role"] if row else "staff"
def admin_has_perm(self, conn, user, module, action="view"):
role = self.admin_role_of(conn, user["id"])
if role == "super_admin":
return True
row = conn.execute(
"SELECT can_view, can_add, can_edit, can_delete FROM admin_permissions WHERE user_id=? AND module=?",
(user["id"], module),
).fetchone()
if not row:
ensure_admin_permissions_rows(conn, user["id"], full_access=False)
conn.commit()
row = conn.execute(
"SELECT can_view, can_add, can_edit, can_delete FROM admin_permissions WHERE user_id=? AND module=?",
(user["id"], module),
).fetchone()
key_map = {"view": "can_view", "add": "can_add", "edit": "can_edit", "delete": "can_delete"}
k = key_map.get(action, "can_view")
return bool(row[k]) if row else False
def require_admin_perm(self, user, module, action="view"):
conn = db_connect()
allowed = self.admin_has_perm(conn, user, module, action)
conn.close()
if not allowed:
self.send_html(page("Forbidden", "
Permission denied.
", user), 403)
return False
return True
def admin_ui_perms(self, conn, user, module):
return {
"view": self.admin_has_perm(conn, user, module, "view"),
"add": self.admin_has_perm(conn, user, module, "add"),
"edit": self.admin_has_perm(conn, user, module, "edit"),
"delete": self.admin_has_perm(conn, user, module, "delete"),
}
def admin_scope_admin_ids(self, conn, user):
role = self.admin_role_of(conn, user["id"])
if role == "super_admin":
return [r["id"] for r in conn.execute("SELECT id FROM users WHERE role='admin'").fetchall()]
if role == "manager":
rows = conn.execute(
"""SELECT u.id
FROM users u
LEFT JOIN admin_profiles ap ON ap.user_id=u.id
WHERE u.role='admin' AND (u.id=? OR ap.manager_user_id=?)""",
(user["id"], user["id"]),
).fetchall()
return [r["id"] for r in rows] or [user["id"]]
return [user["id"]]
def admin_partner_scope_condition(self, conn, user, partner_alias="p"):
role = self.admin_role_of(conn, user["id"])
if role == "super_admin":
return "", []
scoped_admin_ids = self.admin_scope_admin_ids(conn, user)
placeholders = ",".join(["?"] * len(scoped_admin_ids))
return f" AND {partner_alias}.created_by_admin_id IN ({placeholders})", scoped_admin_ids
def partner_in_admin_scope(self, conn, user, partner_id):
role = self.admin_role_of(conn, user["id"])
if role == "super_admin":
return True
scoped_admin_ids = self.admin_scope_admin_ids(conn, user)
placeholders = ",".join(["?"] * len(scoped_admin_ids))
row = conn.execute(
f"SELECT user_id FROM partners WHERE user_id=? AND created_by_admin_id IN ({placeholders})",
[partner_id] + scoped_admin_ids,
).fetchone()
return bool(row)
def do_GET(self):
user = self.current_user()
raw_path = urllib.parse.urlparse(self.path).path
path = raw_path.rstrip("/") or "/"
if path.startswith("/uploads/"):
rel = path[len("/uploads/") :]
if ".." in rel:
return self.send_html("Forbidden", 403)
# Keep payment proofs private: only logged-in admin users can view.
if rel.startswith("subscription_proofs/") or rel.startswith("media_proofs/"):
if not user:
return self.redirect(ADMIN_LOGIN_PATH)
if user["role"] != "admin":
return self.send_html("Forbidden", 403)
abs_path = os.path.join(uploads_root(), rel)
if not os.path.exists(abs_path) or not os.path.isfile(abs_path):
return self.send_html("Not found", 404)
with open(abs_path, "rb") as f:
payload = f.read()
ctype = mimetypes.guess_type(abs_path)[0] or "application/octet-stream"
return self.send_bytes(payload, content_type=ctype, status=200)
if path.startswith("/protected-uploads/"):
rel = path[len("/protected-uploads/") :]
if ".." in rel:
return self.send_html("Forbidden", 403)
if not user:
return self.redirect(ADMIN_LOGIN_PATH)
if user["role"] != "admin":
return self.send_html("Forbidden", 403)
abs_path = os.path.join(private_uploads_root(), rel)
if not os.path.exists(abs_path) or not os.path.isfile(abs_path):
return self.send_html("Not found", 404)
with open(abs_path, "rb") as f:
payload = f.read()
ctype = mimetypes.guess_type(abs_path)[0] or "application/octet-stream"
return self.send_bytes(payload, content_type=ctype, status=200)
if path == "/":
if not user:
return self.public_landing_page()
return self.redirect("/admin/dashboard" if user["role"] == "admin" else "/partner/dashboard")
if path in {"/buyers", "/moving"}:
return self.public_buyer_page()
if path.startswith("/invoice/"):
m = re.match(r"^/invoice/([A-Za-z0-9_-]+)(/pdf)?$", path)
if m:
token = m.group(1)
if m.group(2):
return self.public_invoice_pdf(token)
return self.public_invoice_view(token)
if path == "/login":
return self.get_login(is_admin=False)
if path == ADMIN_LOGIN_PATH:
return self.get_login(is_admin=True)
if path == "/forgot-password":
return self.get_forgot_password()
if path == "/admin/2fa/setup":
return self.get_admin_2fa_setup()
if path == "/admin/2fa/verify":
return self.get_admin_2fa_verify()
if path == "/logout":
self.clear_session()
self.send_response(302)
self.send_header("Location", self._prefix_url("/login"))
self.send_header("Set-Cookie", "session_id=; HttpOnly; Path=/; Max-Age=0")
self.send_header("Set-Cookie", "admin_2fa_token=; HttpOnly; Path=/; Max-Age=0")
self.send_header("Set-Cookie", "admin_2fa_grace_session=; HttpOnly; Path=/; Max-Age=0")
self.end_headers()
return
if path == "/partner/dashboard":
user = self.require_auth("partner")
if user:
return self.partner_dashboard(user)
return
if path == "/partner/morning-meeting":
user = self.require_auth("partner")
if user:
return self.partner_morning_meeting(user)
return
if path == "/partner/diary":
user = self.require_auth("partner")
if user:
return self.partner_diary(user)
return
if path == "/partner/data-bank":
user = self.require_auth("partner")
if user:
return self.partner_data_bank(user)
return
if path == "/partner/leads":
user = self.require_auth("partner")
if user:
return self.partner_leads(user)
return
if path == "/partner/media-leads":
user = self.require_auth("partner")
if user:
return self.partner_media_leads(user)
return
if path == "/partner/clients":
user = self.require_auth("partner")
if user:
return self.partner_clients(user)
return
if path == "/partner/deals":
user = self.require_auth("partner")
if user:
return self.partner_deals(user)
return
if path == "/partner/invoices":
user = self.require_auth("partner")
if user:
return self.partner_invoices(user)
return
if path.startswith("/partner/leads/"):
user = self.require_auth("partner")
if user:
m = re.match(r"^/partner/leads/(\d+)$", path)
if m:
return self.partner_lead_detail(user, int(m.group(1)))
return
if path.startswith("/partner/invoices/"):
user = self.require_auth("partner")
if user:
m = re.match(r"^/partner/invoices/(\d+)(/pdf)?$", path)
if m:
invoice_id = int(m.group(1))
if m.group(2):
return self.partner_invoice_pdf(user, invoice_id)
return self.partner_invoice_detail(user, invoice_id)
return
if path == "/partner/credits":
user = self.require_auth("partner", allow_expired=True)
if user:
return self.partner_credits(user)
return
if path == "/partner/media":
user = self.require_auth("partner")
if user:
return self.partner_media(user)
return
if path == "/partner/support":
user = self.require_auth("partner")
if user:
return self.partner_support(user)
return
if path == "/partner/profile":
user = self.require_auth("partner", allow_expired=True)
if user:
return self.partner_profile(user)
return
if path == "/admin/dashboard":
user = self.require_auth("admin")
if user:
return self.admin_dashboard(user)
return
if path == "/admin/users":
user = self.require_auth("admin")
if user:
return self.admin_users(user)
return
if path == "/admin/partners":
user = self.require_auth("admin")
if user:
return self.admin_partners(user)
return
if path == "/admin/packages":
user = self.require_auth("admin")
if user:
return self.admin_packages(user)
return
if path == "/admin/subscriptions":
user = self.require_auth("admin")
if user:
return self.admin_subscriptions(user)
return
if path == "/admin/media":
user = self.require_auth("admin")
if user:
return self.admin_media(user)
return
if path == "/admin/leads":
user = self.require_auth("admin")
if user:
return self.admin_leads_dashboard(user)
return
if path == "/admin/leads/unused":
user = self.require_auth("admin")
if user:
return self.admin_leads_unused(user)
return
if path == "/admin/leads/create":
user = self.require_auth("admin")
if user:
return self.admin_leads_create(user)
return
if path == "/admin/leads/import-page":
user = self.require_auth("admin")
if user:
return self.admin_leads_import_page(user)
return
if path == "/admin/leads/list":
user = self.require_auth("admin")
if user:
return self.admin_leads_list(user)
return
if path == "/admin/leads/sample.csv":
user = self.require_auth("admin")
if user:
sample = (
"name,phone,email,from_postcode,to_postcode,move_date,service_type,job_type,property_type,move_size,vehicle_size,floor_number,lift_access,packing_needed,dismantling_needed,storage_needed,loading_help_needed,permit_required,parking_notes,waste_type,access_notes,special_items,budget_range,source,lead_type,lead_bucket,partner_email\n"
"Aisha Khan,+447700900123,aisha@example.co.uk,SW1A,BS1,2026-03-20,home_removals,local_move,Flat,2-bed flat,luton van,3,1,1,0,0,1,0,Limited loading bay,,Tight stairs,Grand piano,500_1000,Google Ads,platform,system_pool,\n"
).encode("utf-8") if is_removals_vertical() else (
"name,phone,email,city,area,budget_min,budget_max,property_type,purpose,timeframe,source,lead_type,lead_bucket,partner_email\n"
"Ali Khan,+923001234567,ali@example.com,Karachi,DHA,15000000,25000000,Plot,buy,30 days,Campaign A,platform,system_pool,\n"
).encode("utf-8")
return self.send_bytes(
sample,
content_type="text/csv; charset=utf-8",
extra_headers={"Content-Disposition": "attachment; filename=lead_import_sample.csv"},
)
return
if path == "/admin/leads/sample-specific.csv":
user = self.require_auth("admin")
if user:
sample = (
"name,phone,email,from_postcode,to_postcode,move_date,service_type,job_type,property_type,move_size,vehicle_size,floor_number,lift_access,packing_needed,dismantling_needed,storage_needed,loading_help_needed,permit_required,parking_notes,waste_type,access_notes,special_items,budget_range,source,lead_type,lead_bucket,partner_email\n"
"Tom Client,+447700900124,tom@example.co.uk,E14,M1,2026-03-18,man_with_van,same_day,Flat,Studio,small van,2,0,0,0,0,1,1,Permit needed outside block,,No lift,Boxes and TV,250_500,Facebook Ads,partner_ad,media_ads,partner@vanlocaluk.test\n"
).encode("utf-8") if is_removals_vertical() else (
"name,phone,email,city,area,budget_min,budget_max,property_type,purpose,timeframe,source,lead_type,lead_bucket,partner_email\n"
"Sara Client,+447700900126,sara@example.co.uk,Manchester,Salford,500,900,Flat,buy,15 days,Facebook Ads,partner_ad,media_ads,partner@vanlocaluk.test\n"
).encode("utf-8")
return self.send_bytes(
sample,
content_type="text/csv; charset=utf-8",
extra_headers={"Content-Disposition": "attachment; filename=lead_import_specific_partner_sample.csv"},
)
return
if path == "/admin/media/sample.csv":
user = self.require_auth("admin")
if user:
sample = (
"ad_date,platform,impressions,clicks,views,leads,partner_email\n"
"2026-02-25,Facebook,1000,24,300,3,partner@vanlocaluk.test\n"
"2026-02-25,Google,800,16,220,2,partner@vanlocaluk.test\n"
).encode("utf-8")
return self.send_bytes(
sample,
content_type="text/csv; charset=utf-8",
extra_headers={"Content-Disposition": "attachment; filename=media_performance_sample.csv"},
)
return
if path == "/admin/invalid-reports":
user = self.require_auth("admin")
if user:
return self.admin_invalid_reports(user)
return
if path == "/admin/tickets":
user = self.require_auth("admin")
if user:
return self.admin_tickets(user)
return
if path == "/admin/profile":
user = self.require_auth("admin")
if user:
return self.admin_profile(user)
return
if path == "/admin/enquiries":
user = self.require_auth("admin")
if user:
return self.admin_enquiries(user)
return
if path == "/admin/landing-cms":
user = self.require_auth("admin")
if user:
return self.admin_landing_cms(user)
return
if user:
return self.redirect("/")
return self.redirect("/login")
def do_POST(self):
raw_path = urllib.parse.urlparse(self.path).path
path = raw_path.rstrip("/") or "/"
if path == "/apply-partner":
return self.post_public_partner_application()
if path in {"/submit-buyer-enquiry", "/submit-moving-enquiry"}:
return self.post_public_buyer_enquiry()
if path == "/login":
return self.post_login(is_admin=False)
if path == ADMIN_LOGIN_PATH:
return self.post_login(is_admin=True)
if path == "/forgot-password":
return self.post_forgot_password()
if path == "/admin/2fa/setup":
return self.post_admin_2fa_setup()
if path == "/admin/2fa/verify":
return self.post_admin_2fa_verify()
if path == "/partner/leads/status":
user = self.require_auth("partner")
if user:
return self.post_partner_status(user)
return
if path == "/partner/morning-meeting":
user = self.require_auth("partner")
if user:
return self.post_partner_morning_meeting(user)
return
if path == "/partner/morning-meeting/delete":
user = self.require_auth("partner")
if user:
return self.post_partner_morning_meeting_delete(user)
return
if path == "/partner/diary":
user = self.require_auth("partner")
if user:
return self.post_partner_diary(user)
return
if path == "/partner/diary/delete":
user = self.require_auth("partner")
if user:
return self.post_partner_diary_delete(user)
return
if path == "/partner/data-bank":
user = self.require_auth("partner")
if user:
return self.post_partner_data_bank(user)
return
if path == "/partner/data-bank/delete":
user = self.require_auth("partner")
if user:
return self.post_partner_data_bank_delete(user)
return
if path == "/partner/leads/report-invalid":
user = self.require_auth("partner")
if user:
return self.post_partner_report_invalid(user)
return
if path == "/partner/clients":
user = self.require_auth("partner")
if user:
return self.post_partner_clients(user)
return
if path == "/partner/clients/delete":
user = self.require_auth("partner")
if user:
return self.post_partner_clients_delete(user)
return
if path == "/partner/deals":
user = self.require_auth("partner")
if user:
return self.post_partner_deals(user)
return
if path == "/partner/deals/delete":
user = self.require_auth("partner")
if user:
return self.post_partner_deals_delete(user)
return
if path == "/partner/support":
user = self.require_auth("partner")
if user:
return self.post_partner_support(user)
return
if path == "/partner/profile":
user = self.require_auth("partner", allow_expired=True)
if user:
return self.post_partner_profile(user)
return
if path == "/partner/notifications/ack":
user = self.require_auth("partner")
if user:
return self.post_partner_notification_ack(user)
return
if path == "/partner/media":
user = self.require_auth("partner")
if user:
return self.post_partner_media(user)
return
if path == "/partner/invoices/save":
user = self.require_auth("partner")
if user:
return self.post_partner_invoice_save(user)
return
if path == "/partner/invoices/delete":
user = self.require_auth("partner")
if user:
return self.post_partner_invoice_delete(user)
return
if path == "/admin/partners":
user = self.require_auth("admin")
if user:
return self.post_admin_partners(user)
return
if path == "/admin/users":
user = self.require_auth("admin")
if user:
return self.post_admin_users(user)
return
if path == "/admin/users/update":
user = self.require_auth("admin")
if user:
return self.post_admin_user_update(user)
return
if path == "/admin/users/delete":
user = self.require_auth("admin")
if user:
return self.post_admin_user_delete(user)
return
if path == "/admin/users/permissions":
user = self.require_auth("admin")
if user:
return self.post_admin_user_permissions(user)
return
if path == "/admin/users/2fa/reset":
user = self.require_auth("admin")
if user:
return self.post_admin_user_2fa_reset(user)
return
if path == "/admin/password-requests/resolve":
user = self.require_auth("admin")
if user:
return self.post_admin_password_request_resolve(user)
return
if path == "/admin/partners/update":
user = self.require_auth("admin")
if user:
return self.post_admin_partner_update(user)
return
if path == "/admin/partners/delete":
user = self.require_auth("admin")
if user:
return self.post_admin_partner_delete(user)
return
if path == "/admin/partners/auto-assign":
user = self.require_auth("admin")
if user:
return self.post_admin_partner_auto_assign(user)
return
if path == "/admin/packages":
user = self.require_auth("admin")
if user:
return self.post_admin_packages(user)
return
if path == "/admin/packages/update":
user = self.require_auth("admin")
if user:
return self.post_admin_package_update(user)
return
if path == "/admin/packages/delete":
user = self.require_auth("admin")
if user:
return self.post_admin_package_delete(user)
return
if path == "/admin/subscriptions":
user = self.require_auth("admin")
if user:
return self.post_admin_subscriptions(user)
return
if path == "/admin/media":
user = self.require_auth("admin")
if user:
return self.post_admin_media(user)
return
if path == "/admin/todos/add":
user = self.require_auth("admin")
if user:
return self.post_admin_todo_add(user)
return
if path == "/admin/todos/complete":
user = self.require_auth("admin")
if user:
return self.post_admin_todo_complete(user)
return
if path == "/admin/leads":
user = self.require_auth("admin")
if user:
return self.post_admin_leads(user)
return
if path == "/admin/leads/import":
user = self.require_auth("admin")
if user:
return self.post_admin_leads_import(user)
return
if path == "/admin/leads/assign":
user = self.require_auth("admin")
if user:
return self.post_admin_assign(user)
return
if path == "/admin/leads/assign-bulk":
user = self.require_auth("admin")
if user:
return self.post_admin_assign_bulk(user)
return
if path == "/admin/invalid-reports/review":
user = self.require_auth("admin")
if user:
return self.post_admin_invalid_review(user)
return
if path == "/admin/tickets/reply":
user = self.require_auth("admin")
if user:
return self.post_admin_ticket_reply(user)
return
if path == "/admin/profile":
user = self.require_auth("admin")
if user:
return self.post_admin_profile(user)
return
if path == "/admin/enquiries/update":
user = self.require_auth("admin")
if user:
return self.post_admin_enquiry_update(user)
return
if path == "/admin/landing-cms/update":
user = self.require_auth("admin")
if user:
return self.post_admin_landing_cms_update(user)
return
if path == "/admin/landing-cms/badges":
user = self.require_auth("admin")
if user:
return self.post_admin_landing_badges(user)
return
user = self.current_user()
if user:
return self.redirect("/")
return self.redirect("/login")
def get_login(self, error="", is_admin=False):
cfg = vertical_config()
brand_markup = brand_logo_html(
cfg["brand_name"],
"Verified moving enquiries for UK operators." if is_removals_vertical() else "Verified lead platform for property operators.",
)
portal_label = cfg["operator_portal_label"] if not is_removals_vertical() else f"Moving & Waste {cfg['operator_portal_label']}"
html_page = f"""
{'Admin Login' if is_admin else 'Login'} | {html.escape(cfg['app_title'])}
{brand_markup}
Login
Sign in to your account
{f"
{html.escape(error)}
" if error else ""}
{'Admin Staff Portal' if is_admin else html.escape(portal_label)}
"""
self.send_html(html_page)
def post_forgot_password(self):
data = parse_post_data(self)
email = first(data, "email").lower()
if not email:
return self.get_forgot_password(error="Please enter an email.")
conn = db_connect()
user = conn.execute(
"SELECT id, email FROM users WHERE role='partner' AND lower(email)=lower(?)",
(email,),
).fetchone()
if user:
pending = conn.execute(
"SELECT id FROM password_reset_requests WHERE user_id=? AND status='pending' ORDER BY id DESC LIMIT 1",
(user["id"],),
).fetchone()
if not pending:
conn.execute(
"""INSERT INTO password_reset_requests
(user_id, email, status, created_at)
VALUES (?, ?, 'pending', ?)""",
(user["id"], user["email"], now_iso()),
)
admins = conn.execute("SELECT id FROM users WHERE role='admin' AND status='active'").fetchall()
for a in admins:
add_notification(
conn,
a["id"],
"Password Reset Request",
f"{vertical_config()['operator_singular']} {user['email']} requested password reset.",
)
conn.commit()
conn.close()
return self.get_forgot_password(
message="If your email exists in our system, your request has been sent to admin."
)
def get_admin_2fa_setup(self, error=""):
token = first(urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query), "token")
if not token:
token = self.cookie_value("admin_2fa_token")
conn = db_connect()
ch = None
if token:
ch = conn.execute(
"SELECT * FROM admin_login_challenges WHERE token=? AND expires_at>?",
(token, now_iso()),
).fetchone()
user = None
if ch:
user = conn.execute("SELECT * FROM users WHERE id=? AND role='admin'", (ch["user_id"],)).fetchone()
if not user:
current = self.current_user()
if current and current["role"] == "admin":
user = conn.execute("SELECT * FROM users WHERE id=? AND role='admin'", (current["id"],)).fetchone()
token = ""
if not user:
conn.close()
return self.redirect(ADMIN_LOGIN_PATH)
profile = self.admin_2fa_profile(conn, user["id"])
if int(profile["twofa_enabled"] or 0) == 1:
conn.close()
return self.redirect("/admin/dashboard")
secret = (profile["twofa_secret"] or "").strip()
if not secret:
secret = generate_totp_secret()
conn.execute("UPDATE admin_profiles SET twofa_secret=? WHERE user_id=?", (secret, user["id"]))
conn.commit()
conn.close()
issuer = vertical_config()["brand_name"]
account = user["email"]
otpauth = f"otpauth://totp/{urllib.parse.quote(issuer)}:{urllib.parse.quote(account)}?secret={secret}&issuer={urllib.parse.quote(issuer)}&algorithm=SHA1&digits=6&period=30"
qr_url = "https://api.qrserver.com/v1/create-qr-code/?size=220x220&data=" + urllib.parse.quote(otpauth, safe="")
err_html = f"
" for h, t in cfg["buyer_points"]])
logo_html = brand_logo_html(
cfg["brand_name"],
"Moving, removals and waste removal leads" if is_removals_vertical() else "",
compact=True,
)
form_html = """
"""
self.send_html(html_page)
def post_public_partner_application(self):
data = parse_post_data(self)
conn = db_connect()
area_value = first(data, "areas")
property_value = first(data, "partner_type")
message_value = first(data, "message")
if is_removals_vertical():
area_value = first(data, "coverage_postcodes")
property_value = first(data, "service_categories")
message_lines = [message_value] if message_value else []
message_lines.extend(
[
f"Company: {first(data, 'company_name')}",
f"Domestic: {'Yes' if normalize_yes_no(first(data, 'supports_domestic', '1')) else 'No'}",
f"Commercial: {'Yes' if normalize_yes_no(first(data, 'supports_commercial')) else 'No'}",
f"Same-day: {'Yes' if normalize_yes_no(first(data, 'same_day_available')) else 'No'}",
f"Weekend: {'Yes' if normalize_yes_no(first(data, 'weekend_available')) else 'No'}",
f"Preferred lead types: {first(data, 'preferred_lead_types') or '-'}",
f"Vehicle types: {first(data, 'vehicle_types') or '-'}",
f"Crew size: {first(data, 'crew_size') or '-'}",
f"Budget range: {first(data, 'lead_budget_min') or '0'} to {first(data, 'lead_budget_max') or '0'} {vertical_config()['currency_code']}",
f"Insurance: {first(data, 'insurance_details') or '-'}",
f"Licence: {first(data, 'license_details') or '-'}",
]
)
message_value = "\n".join([line for line in message_lines if line])
conn.execute(
"""INSERT INTO enquiries
(enquiry_type, full_name, phone, email, city, area, property_type, budget_min, budget_max, message, status, admin_notes, email_sent, created_at)
VALUES ('partner_application', ?, ?, ?, ?, ?, ?, 0, 0, ?, 'new', '', 0, ?)""",
(first(data, "full_name"), first(data, "phone"), first(data, "email"), first(data, "city"), area_value, property_value, message_value, now_iso()),
)
sent = send_simple_email(
ENQUIRY_EMAIL_TO,
f"New Verified {vertical_config()['operator_singular']} Application",
(
f"Name: {first(data,'full_name')}\nPhone: {first(data,'phone')}\nCity: {first(data,'city')}\n"
f"Categories/Type: {property_value}\nCoverage: {area_value}\nMessage: {message_value}"
),
)
if sent:
conn.execute("UPDATE enquiries SET email_sent=1 WHERE id=(SELECT MAX(id) FROM enquiries)")
conn.commit()
conn.close()
self.redirect("/?msg=partner_application_submitted")
def post_public_buyer_enquiry(self):
data = parse_post_data(self)
conn = db_connect()
if is_removals_vertical():
from_postcode = normalize_postcode_area(first(data, "from_postcode"))
to_postcode = normalize_postcode_area(first(data, "to_postcode"))
service_type = normalize_service_category(first(data, "service_type", "home_removals"))
budget_min, budget_max = budget_range_to_values(first(data, "budget_range"))
cur = conn.execute(
"""INSERT INTO enquiries
(enquiry_type, full_name, phone, email, city, area, property_type, budget_min, budget_max, message, status, admin_notes, email_sent, created_at)
VALUES ('buyer_enquiry', ?, ?, ?, ?, ?, ?, ?, ?, ?, 'new', '', 0, ?)""",
(
first(data, "full_name"),
first(data, "phone"),
first(data, "email"),
from_postcode or "UK",
to_postcode or "UK",
service_type,
budget_min,
budget_max,
first(data, "message"),
now_iso(),
),
)
enquiry_id = cur.lastrowid
cur = conn.execute(
"""INSERT INTO leads
(name, phone, email, city, area, budget_min, budget_max, property_type, purpose, timeframe, source, lead_type, lead_bucket, target_partner_id, verified_by_admin, created_at, vertical_type, lead_score, service_category)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'platform', 'system_pool', NULL, 1, ?, ?, 0, ?)""",
(
first(data, "full_name"),
first(data, "phone"),
first(data, "email"),
from_postcode or "UK",
to_postcode or "UK",
budget_min,
budget_max,
first(data, "property_type", "House"),
"buy",
first(data, "move_date", "Flexible"),
"Landing Moving Enquiry",
now_iso(),
VERTICAL_TYPE,
service_type,
),
)
lead_id = cur.lastrowid
save_move_requirements(conn, lead_id, data, enquiry_id=enquiry_id)
sent = send_simple_email(
ENQUIRY_EMAIL_TO,
"New Moving Enquiry",
(
f"Name: {first(data,'full_name')}\nPhone: {first(data,'phone')}\nFrom: {from_postcode}\nTo: {to_postcode}\n"
f"Move date: {first(data,'move_date')}\nService: {format_service_category(service_type)}\n"
f"Budget: {first(data,'budget_range')}\nSpecial items: {first(data,'special_items')}\nMessage: {first(data,'message')}"
),
)
if sent:
conn.execute("UPDATE enquiries SET email_sent=1 WHERE id=?", (enquiry_id,))
conn.commit()
conn.close()
return self.redirect("/moving?msg=buyer_enquiry_submitted")
city = first(data, "city") or "Islamabad"
area = first(data, "area") or "Unknown"
purpose = normalize_purpose(first(data, "purpose", "buy"))
conn.execute(
"""INSERT INTO enquiries
(enquiry_type, full_name, phone, email, city, area, property_type, budget_min, budget_max, message, status, admin_notes, email_sent, created_at)
VALUES ('buyer_enquiry', ?, ?, ?, ?, ?, ?, ?, ?, ?, 'new', '', 0, ?)""",
(first(data, "full_name"), first(data, "phone"), first(data, "email"), city, area, first(data, "property_type", "Plot"), to_int(first(data, "budget_min", "0"), 0), to_int(first(data, "budget_max", "0"), 0), first(data, "message"), now_iso()),
)
conn.execute(
"""INSERT INTO leads
(name, phone, email, city, area, budget_min, budget_max, property_type, purpose, timeframe, source, lead_type, lead_bucket, target_partner_id, verified_by_admin, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'platform', 'system_pool', NULL, 1, ?)""",
(first(data, "full_name"), first(data, "phone"), first(data, "email"), city, area, to_int(first(data, "budget_min", "0"), 0), to_int(first(data, "budget_max", "0"), 0), first(data, "property_type", "Plot"), purpose, first(data, "timeframe", "30 days"), "Landing Buyer Enquiry", now_iso()),
)
sent = send_simple_email(ENQUIRY_EMAIL_TO, "New Buyer Enquiry", f"Name: {first(data,'full_name')}\nPhone: {first(data,'phone')}\nCity: {city}\nArea: {area}\nType: {first(data,'property_type','Plot')}\nBudget: {first(data,'budget_min')} - {first(data,'budget_max')}\nMessage: {first(data,'message')}")
if sent:
conn.execute("UPDATE enquiries SET email_sent=1 WHERE id=(SELECT MAX(id) FROM enquiries)")
conn.commit()
conn.close()
self.redirect("/buyers?msg=buyer_enquiry_submitted")
def post_login(self, is_admin=False):
data = parse_post_data(self)
email = first(data, "email").lower()
password = first(data, "password")
conn = db_connect()
try:
ensure_runtime_migrations(conn)
conn.commit()
except Exception:
pass
expire_subscriptions(conn)
user = conn.execute("SELECT * FROM users WHERE email=? AND status='active'", (email,)).fetchone()
if not user or not password_verify(password, user["password_hash"]):
conn.close()
return self.get_login("Invalid credentials.", is_admin=is_admin)
if is_admin and user["role"] != "admin":
conn.close()
return self.get_login("Use partner login at /login.", is_admin=True)
if not is_admin and user["role"] == "admin":
conn.close()
return self.get_login(f"Admin users must login at {ADMIN_LOGIN_PATH}.", is_admin=False)
if is_admin and user["role"] == "admin":
profile = self.admin_2fa_profile(conn, user["id"])
role_name = self.admin_role_of(conn, user["id"])
secret = (profile["twofa_secret"] or "").strip()
enabled = int(profile["twofa_enabled"] or 0) == 1
if role_name == "super_admin":
conn.commit()
conn.close()
sid = self.create_session(user["id"])
self.send_response(302)
self.send_header("Location", self._prefix_url("/admin/dashboard"))
self.send_header("Set-Cookie", f"session_id={sid}; HttpOnly; Path=/; Max-Age={SESSION_HOURS*3600}")
self.send_header("Set-Cookie", "admin_2fa_token=; HttpOnly; Path=/; Max-Age=0")
self.send_header("Set-Cookie", "admin_2fa_grace_session=; HttpOnly; Path=/; Max-Age=0")
self.end_headers()
return
if enabled and secret:
conn.commit()
conn.close()
token = self.create_admin_login_challenge(user["id"])
self.send_response(302)
self.send_header("Location", self._prefix_url("/admin/2fa/verify"))
self.send_header("Set-Cookie", f"admin_2fa_token={token}; HttpOnly; Path=/; Max-Age=300")
self.end_headers()
return
conn.commit()
conn.close()
sid = self.create_session(user["id"])
self.send_response(302)
self.send_header("Location", self._prefix_url("/admin/dashboard"))
self.send_header("Set-Cookie", f"session_id={sid}; HttpOnly; Path=/; Max-Age={SESSION_HOURS*3600}")
self.send_header("Set-Cookie", "admin_2fa_token=; HttpOnly; Path=/; Max-Age=0")
self.send_header("Set-Cookie", "admin_2fa_grace_session=; HttpOnly; Path=/; Max-Age=0")
self.end_headers()
return
if user["role"] == "partner" and not active_subscription(conn, user["id"]):
upcoming = next_subscription_start(conn, user["id"])
conn.commit()
conn.close()
if upcoming:
return self.get_login(
f"Your package is scheduled to start on {upcoming.isoformat()}.",
is_admin=False,
)
return self.get_login("Package expired. Renew your package to access the portal.", is_admin=False)
conn.commit()
conn.close()
try:
sid = self.create_session(user["id"])
except sqlite3.Error:
return self.get_login(
"Login failed: database is not writable. Check cPanel file permissions for vanlocaluk.db and app folder.",
is_admin=is_admin,
)
self.send_response(302)
self.send_header("Location", self._prefix_url("/admin/dashboard" if user["role"] == "admin" else "/partner/dashboard"))
self.send_header("Set-Cookie", f"session_id={sid}; HttpOnly; Path=/; Max-Age={SESSION_HOURS*3600}")
self.end_headers()
def partner_dashboard(self, user):
conn = db_connect()
sub = active_subscription(conn, user["id"])
coverage = conn.execute(
"""SELECT city, areas_json, company_name,
COALESCE(service_categories_json,'[]') service_categories_json,
COALESCE(coverage_postcodes_json,'[]') coverage_postcodes_json,
COALESCE(same_day_available,0) same_day_available,
COALESCE(weekend_available,0) weekend_available,
COALESCE(vehicle_types,'') vehicle_types,
COALESCE(crew_size,'') crew_size
FROM partners
WHERE user_id=?""",
(user["id"],),
).fetchone()
stats = conn.execute(
"""SELECT
SUM(CASE WHEN viewed_at IS NOT NULL THEN 1 ELSE 0 END) AS viewed,
SUM(CASE WHEN contacted_at IS NOT NULL THEN 1 ELSE 0 END) AS contacted,
SUM(CASE WHEN status='converted' THEN 1 ELSE 0 END) AS converted,
COUNT(*) AS total
FROM lead_assignments WHERE partner_id=?""",
(user["id"],),
).fetchone()
notifications = conn.execute(
"SELECT * FROM notifications WHERE partner_id=? AND is_read=0 ORDER BY id DESC LIMIT 12",
(user["id"],),
).fetchall()
status_rows = conn.execute(
"SELECT status, COUNT(*) c FROM lead_assignments WHERE partner_id=? GROUP BY status ORDER BY c DESC",
(user["id"],),
).fetchall()
media_sub = conn.execute(
"""SELECT * FROM media_subscriptions
WHERE partner_id=? AND status='active'
ORDER BY id DESC LIMIT 1""",
(user["id"],),
).fetchone()
media_perf_agg = None
media_daily = []
if media_sub:
media_perf_agg = conn.execute(
"""SELECT
SUM(impressions) impressions,
SUM(clicks) clicks,
SUM(views) views,
SUM(leads) leads
FROM media_performance_daily
WHERE partner_id=? AND media_subscription_id=?""",
(user["id"], media_sub["id"]),
).fetchone()
media_daily = conn.execute(
"""SELECT ad_date,
SUM(impressions) impressions,
SUM(clicks) clicks,
SUM(views) views,
SUM(leads) leads
FROM media_performance_daily
WHERE partner_id=? AND media_subscription_id=?
GROUP BY ad_date
ORDER BY ad_date DESC LIMIT 7""",
(user["id"], media_sub["id"]),
).fetchall()
window_start = (datetime.utcnow().date() - timedelta(days=6)).isoformat()
daily_assigned_rows = conn.execute(
"""SELECT substr(assigned_at,1,10) d, COUNT(*) c
FROM lead_assignments
WHERE partner_id=? AND substr(assigned_at,1,10) >= ?
GROUP BY substr(assigned_at,1,10)
ORDER BY d ASC""",
(user["id"], window_start),
).fetchall()
daily_viewed_rows = conn.execute(
"""SELECT substr(viewed_at,1,10) d, COUNT(*) c
FROM lead_assignments
WHERE partner_id=? AND viewed_at IS NOT NULL AND substr(viewed_at,1,10) >= ?
GROUP BY substr(viewed_at,1,10)
ORDER BY d ASC""",
(user["id"], window_start),
).fetchall()
daily_contacted_rows = conn.execute(
"""SELECT substr(contacted_at,1,10) d, COUNT(*) c
FROM lead_assignments
WHERE partner_id=? AND contacted_at IS NOT NULL AND substr(contacted_at,1,10) >= ?
GROUP BY substr(contacted_at,1,10)
ORDER BY d ASC""",
(user["id"], window_start),
).fetchall()
daily_converted_rows = conn.execute(
"""SELECT substr(lsh.created_at,1,10) d, COUNT(*) c
FROM lead_status_history lsh
JOIN lead_assignments la ON la.id=lsh.assignment_id
WHERE la.partner_id=? AND lsh.status='converted' AND substr(lsh.created_at,1,10) >= ?
GROUP BY substr(lsh.created_at,1,10)
ORDER BY d ASC""",
(user["id"], window_start),
).fetchall()
system_leads_stats = conn.execute(
"""SELECT COUNT(*) total,
SUM(CASE WHEN la.viewed_at IS NOT NULL THEN 1 ELSE 0 END) viewed,
SUM(CASE WHEN la.contacted_at IS NOT NULL THEN 1 ELSE 0 END) contacted,
SUM(CASE WHEN la.status='converted' THEN 1 ELSE 0 END) converted
FROM lead_assignments la
JOIN leads l ON l.id=la.lead_id
WHERE la.partner_id=? AND l.lead_bucket='system_pool'""",
(user["id"],),
).fetchone()
fresh_leads_stats = conn.execute(
"""SELECT COUNT(*) total,
SUM(CASE WHEN la.viewed_at IS NOT NULL THEN 1 ELSE 0 END) viewed,
SUM(CASE WHEN la.contacted_at IS NOT NULL THEN 1 ELSE 0 END) contacted,
SUM(CASE WHEN la.status='converted' THEN 1 ELSE 0 END) converted
FROM lead_assignments la
JOIN leads l ON l.id=la.lead_id
WHERE la.partner_id=? AND l.lead_bucket='media_ads'""",
(user["id"],),
).fetchone()
service_mix_rows = conn.execute(
"""SELECT COALESCE(l.service_category,'general') service_type, COUNT(*) c
FROM lead_assignments la
JOIN leads l ON l.id=la.lead_id
WHERE la.partner_id=?
GROUP BY COALESCE(l.service_category,'general')
ORDER BY c DESC
LIMIT 6""",
(user["id"],),
).fetchall()
origin_rows = conn.execute(
"""SELECT COALESCE(mr.from_postcode, l.city, 'Unknown') area_label, COUNT(*) c
FROM lead_assignments la
JOIN leads l ON l.id=la.lead_id
LEFT JOIN move_requirements mr ON mr.lead_id=l.id
WHERE la.partner_id=?
GROUP BY COALESCE(mr.from_postcode, l.city, 'Unknown')
ORDER BY c DESC
LIMIT 6""",
(user["id"],),
).fetchall()
invoice_stats = conn.execute(
"""SELECT
COUNT(*) total,
SUM(CASE WHEN status='unpaid' THEN 1 ELSE 0 END) unpaid,
SUM(CASE WHEN status='partial_paid' THEN 1 ELSE 0 END) partial_paid,
SUM(CASE WHEN status='fully_paid' THEN 1 ELSE 0 END) fully_paid,
SUM(COALESCE(total_amount,0) - COALESCE(amount_paid,0)) outstanding
FROM invoices
WHERE partner_id=?""",
(user["id"],),
).fetchone()
notif_rows = "".join(
[f"""
{html.escape(n['title'])}
{html.escape(n['message'])}
{html.escape(n['created_at'])}
"""
for n in notifications]
)
date_labels = [(datetime.utcnow().date() - timedelta(days=offset)).isoformat() for offset in range(6, -1, -1)]
short_labels = [d[5:] for d in date_labels]
assigned_map = {r["d"]: r["c"] for r in daily_assigned_rows}
viewed_map = {r["d"]: r["c"] for r in daily_viewed_rows}
contacted_map = {r["d"]: r["c"] for r in daily_contacted_rows}
converted_map = {r["d"]: r["c"] for r in daily_converted_rows}
assigned_series = [assigned_map.get(d, 0) for d in date_labels]
viewed_series = [viewed_map.get(d, 0) for d in date_labels]
contacted_series = [contacted_map.get(d, 0) for d in date_labels]
converted_series = [converted_map.get(d, 0) for d in date_labels]
viewed_rate = int(((stats["viewed"] or 0) * 100) / max(1, stats["total"] or 0))
contacted_rate = int(((stats["contacted"] or 0) * 100) / max(1, stats["total"] or 0))
conversion_rate = int(((stats["converted"] or 0) * 100) / max(1, stats["total"] or 0))
stage_chart = svg_bar_chart(
"Pipeline Volume",
["Assigned", "Viewed", "Contacted", "Converted"],
[stats["total"] or 0, stats["viewed"] or 0, stats["contacted"] or 0, stats["converted"] or 0],
theme_primary(),
)
source_split_chart = svg_bar_chart(
"System vs Fresh Leads",
[lead_bucket_short_label("system_pool"), lead_bucket_short_label("media_ads")],
[system_leads_stats["total"] or 0, fresh_leads_stats["total"] or 0],
theme_secondary(),
)
engagement_chart = svg_line_chart(
"Assigned vs Viewed (7 days)",
short_labels or ["-"],
assigned_series or [0],
viewed_series or [0],
theme_primary(),
theme_secondary(),
)
conversion_chart = svg_line_chart(
"Contacted vs Converted (7 days)",
short_labels or ["-"],
contacted_series or [0],
converted_series or [0],
"#374151",
theme_secondary(),
)
service_mix_chart = svg_bar_chart(
"Service Category Mix",
[format_service_category(r["service_type"]) for r in service_mix_rows] or ["No jobs"],
[r["c"] for r in service_mix_rows] or [0],
theme_primary(),
)
origin_chart = svg_bar_chart(
"Pickup Postcode Demand",
[r["area_label"] for r in origin_rows] or ["No data"],
[r["c"] for r in origin_rows] or [0],
theme_secondary(),
)
media_rows_rev = list(reversed(media_daily))
media_chart = svg_line_chart(
"Media Impressions vs Clicks (7 days)",
[r["ad_date"][-5:] for r in media_rows_rev] or ["-"],
[r["impressions"] for r in media_rows_rev] or [0],
[r["clicks"] for r in media_rows_rev] or [0],
theme_primary(),
theme_secondary(),
)
conn.close()
cfg = vertical_config()
hour = local_now().hour
greet = "Good Morning" if hour < 12 else ("Good Afternoon" if hour < 18 else "Good Evening")
agent_name = user["name"] or cfg["operator_singular"]
company_name = (coverage["company_name"] if coverage and coverage["company_name"] else "").strip()
greet_text = f"{greet}, {agent_name}" + (f" - {company_name}" if company_name else "")
service_categories = []
coverage_postcodes = []
assigned_areas = []
try:
service_categories = json.loads(coverage["service_categories_json"] or "[]") if coverage else []
except Exception:
service_categories = []
try:
coverage_postcodes = json.loads(coverage["coverage_postcodes_json"] or "[]") if coverage else []
except Exception:
coverage_postcodes = []
try:
assigned_areas = json.loads(coverage["areas_json"] or "[]") if coverage else []
except Exception:
assigned_areas = []
top_stat_cards = "".join(
[
dashboard_stat_card(
f"{cfg['lead_plural']} Remaining",
sub["credits_remaining"] if sub else 0,
note="Live credits left in the active package",
badge=html.escape(sub["package_name"]) if sub else "No package",
icon="CR",
tone="primary",
),
dashboard_stat_card(
"System Leads",
system_leads_stats["total"] or 0,
note=f"{system_leads_stats['viewed'] or 0} viewed / {system_leads_stats['converted'] or 0} won",
badge="1 credit on first open",
icon="SY",
tone="dark",
),
dashboard_stat_card(
"Fresh Leads",
fresh_leads_stats["total"] or 0,
note=("No fresh leads available right now" if not (fresh_leads_stats["total"] or 0) else f"{fresh_leads_stats['viewed'] or 0} viewed / {fresh_leads_stats['converted'] or 0} won"),
badge="3 credits on first open",
icon="FR",
tone="secondary",
),
dashboard_stat_card(
"Package Renewal",
html.escape(sub["end_date"]) if sub else "-",
note=f"Plan value {money(sub['finalized_amount']) if sub else money(0)}",
badge=money(sub["finalized_amount"]) if sub else money(0),
icon="EX",
tone="soft",
),
]
)
activity_stat_cards = "".join(
[
dashboard_stat_card(
"Viewed Enquiries",
stats["viewed"] or 0,
note=f"{viewed_rate}% of assigned leads opened",
badge=f"{fresh_leads_stats['viewed'] or 0} fresh views",
icon="VW",
tone="primary",
),
dashboard_stat_card(
"Contacted Enquiries",
stats["contacted"] or 0,
note=f"{contacted_rate}% of assigned enquiries reached",
badge=f"{fresh_leads_stats['contacted'] or 0} fresh contacted",
icon="CT",
tone="secondary",
),
dashboard_stat_card(
"Converted Jobs",
stats["converted"] or 0,
note=f"{conversion_rate}% close rate from assigned enquiries",
badge=f"{system_leads_stats['converted'] or 0} system / {fresh_leads_stats['converted'] or 0} fresh",
icon="CV",
tone="dark",
),
dashboard_stat_card(
"Outstanding Invoices",
invoice_stats["unpaid"] or 0,
note=f"{invoice_stats['partial_paid'] or 0} part-paid / {invoice_stats['fully_paid'] or 0} fully paid",
badge=money(invoice_stats["outstanding"] or 0, 2),
icon="IV",
tone="soft",
),
]
)
profile_items = [
("Primary hub", coverage["city"] if coverage else "-"),
("Coverage postcodes", ", ".join(coverage_postcodes) if coverage_postcodes else (", ".join(assigned_areas) if assigned_areas else "-")),
("Service categories", ", ".join([format_service_category(x) for x in service_categories]) if service_categories else "-"),
("Vehicle types", coverage["vehicle_types"] if coverage and coverage["vehicle_types"] else "-"),
("Crew size", coverage["crew_size"] if coverage and coverage["crew_size"] else "-"),
("Active package", sub["package_name"] if sub else "-"),
]
pipeline_progress = "".join(
[
dashboard_progress("Viewed enquiries", viewed_rate, f"{stats['viewed'] or 0} of {stats['total'] or 0} assigned"),
dashboard_progress("Contacted enquiries", contacted_rate, f"{stats['contacted'] or 0} of {stats['total'] or 0} assigned"),
dashboard_progress("Converted jobs", conversion_rate, f"{stats['converted'] or 0} jobs won"),
dashboard_progress(
"System leads converted",
int(((system_leads_stats["converted"] or 0) * 100) / max(1, system_leads_stats["total"] or 0)),
f"{system_leads_stats['converted'] or 0} of {system_leads_stats['total'] or 0}",
),
dashboard_progress(
"Fresh leads converted",
int(((fresh_leads_stats["converted"] or 0) * 100) / max(1, fresh_leads_stats["total"] or 0)),
f"{fresh_leads_stats['converted'] or 0} of {fresh_leads_stats['total'] or 0}",
),
]
)
media_info = dashboard_info_rows(
[
("Impressions", (media_perf_agg["impressions"] or 0) if media_perf_agg else 0),
("Clicks", (media_perf_agg["clicks"] or 0) if media_perf_agg else 0),
("Views", (media_perf_agg["views"] or 0) if media_perf_agg else 0),
("Leads", (media_perf_agg["leads"] or 0) if media_perf_agg else 0),
("Rate / 1000", money(media_sub["rate_per_1000"], 2) if media_sub else money(0, 2)),
("Campaign package", media_sub["package_name"] if media_sub else "Not active"),
]
)
body = f"""
{html.escape(greet_text)}
See what is happening with your crews, trucks, enquiries, and invoices in one place so you can spot issues early and stay in control.
System Leads are older demand records and cost 1 credit on first open. Fresh Leads come from current VanLocalUK campaigns and cost 3 credits on first open.
ID
Name
Phone
City/Area
Type
Source
Status
Action
{trs or f'
No {html.escape(cfg["lead_plural"].lower())} yet.
'}
{mobile_cards or f"
No {html.escape(cfg['lead_plural'].lower())} yet.
"}
"""
self.send_html(page(f"{cfg['operator_singular']} {cfg['lead_plural']}", body, user))
def partner_media_leads(self, user):
cfg = vertical_config()
parsed = urllib.parse.urlparse(self.path)
q = urllib.parse.parse_qs(parsed.query)
status_filter = q.get("status", [""])[0]
where = "WHERE la.partner_id=? AND l.lead_bucket='media_ads'"
params = [user["id"]]
if status_filter:
where += " AND la.status=?"
params.append(status_filter)
conn = db_connect()
rows = conn.execute(
f"""SELECT la.id AS assignment_id, la.status, la.assigned_at, la.viewed_at, l.*
FROM lead_assignments la
JOIN leads l ON l.id=la.lead_id
{where}
ORDER BY la.id DESC""",
params,
).fetchall()
conn.close()
trs = ""
mobile_cards = ""
for r in rows:
location_text = f"{html.escape(r['city'])} / {html.escape(r['area'])}"
type_text = html.escape(r["property_type"])
if is_removals_vertical():
location_text = f"{html.escape(r['city'])} to {html.escape(r['area'])}"
type_text = html.escape(format_service_category(r["service_category"] or r["property_type"]))
phone_html = html.escape(r["phone"]) if r["viewed_at"] else f"Hidden until View {html.escape(cfg['lead_singular'])}"
trs += f"""
Fresh leads come from current VanLocalUK campaigns. Opening a fresh lead deducts 3 credits from your balance.
ID
Name
Phone
City/Area
Type
Status
Action
{trs or '
No fresh leads available right now.
'}
{mobile_cards or "
No fresh leads available right now.
"}
"""
self.send_html(page("Fresh Leads", body, user))
def partner_lead_detail(self, user, assignment_id):
cfg = vertical_config()
conn = db_connect()
row = conn.execute(
"""SELECT la.*, l.*
FROM lead_assignments la
JOIN leads l ON l.id=la.lead_id
WHERE la.id=? AND la.partner_id=?""",
(assignment_id, user["id"]),
).fetchone()
if not row:
conn.close()
return self.send_html(page("Not Found", f"
{html.escape(cfg['lead_singular'])} not found.
", user), 404)
sub = active_subscription(conn, user["id"])
bucket = normalize_lead_bucket(row["lead_bucket"])
view_cost = lead_bucket_view_cost(bucket)
if row["status"] == "assigned" and not row["viewed_at"]:
if not sub or sub["credits_remaining"] < view_cost:
conn.close()
body = f"""
No Credits Remaining
You need {view_cost} credits to open this {html.escape(lead_bucket_label(bucket)).lower()}. Renew or top up before viewing the enquiry.
"""
return self.send_html(page("No Credits", body, user), 403)
conn.execute(
"UPDATE subscriptions SET credits_used = credits_used + ?, credits_remaining = CASE WHEN credits_remaining >= ? THEN credits_remaining - ? ELSE credits_remaining END WHERE id=?",
(view_cost, view_cost, view_cost, sub["id"]),
)
conn.execute(
"UPDATE lead_assignments SET status=CASE WHEN status='assigned' THEN 'viewed' ELSE status END, viewed_at=COALESCE(viewed_at, ?) WHERE id=?",
(now_iso(), assignment_id),
)
history = conn.execute("SELECT * FROM lead_status_history WHERE assignment_id=? ORDER BY id DESC", (assignment_id,)).fetchall()
move_summary = moving_requirement_summary(conn, row["id"]) if is_removals_vertical() else None
conn.commit()
conn.close()
log_html = "".join([f"
{html.escape(h['status'])}
{html.escape(h['notes'] or '')}
{html.escape(h['created_at'])}
" for h in history])
requirement_html = f"
Requirements: {html.escape(row['property_type'])} for {html.escape(row['purpose'])}, timeframe {html.escape(row['timeframe'])}
"
if move_summary:
requirement_html = (
f"
Route: {html.escape(move_summary['from_postcode'])} to {html.escape(move_summary['to_postcode'])}
"
f"
Move Date: {html.escape(move_summary['move_date'] or '-')}
"""
self.send_html(page("Billing & Credits" if is_removals_vertical() else "Credits & Package", body, user))
def partner_media_credit_summary_html(self, partner_id):
conn = db_connect()
sub = conn.execute(
"""SELECT * FROM media_subscriptions
WHERE partner_id=? AND status='active'
ORDER BY id DESC LIMIT 1""",
(partner_id,),
).fetchone()
conn.close()
if not sub:
return "
"
)
def partner_media(self, user):
parsed = urllib.parse.urlparse(self.path)
q = urllib.parse.parse_qs(parsed.query)
selected_platform = (q.get("platform", ["all"])[0] or "all").strip()
compare_with = (q.get("compare", ["clicks"])[0] or "clicks").strip().lower()
if compare_with not in {"clicks", "views"}:
compare_with = "clicks"
conn = db_connect()
sub = conn.execute(
"""SELECT * FROM media_subscriptions
WHERE partner_id=? AND status='active'
ORDER BY id DESC LIMIT 1""",
(user["id"],),
).fetchone()
rows = []
rows_all = []
agg = None
platforms = []
if sub:
platforms = conn.execute(
"""SELECT DISTINCT platform FROM media_performance_daily
WHERE partner_id=? AND media_subscription_id=?
ORDER BY platform""",
(user["id"], sub["id"]),
).fetchall()
rows_all = conn.execute(
"""SELECT * FROM media_performance_daily
WHERE partner_id=? AND media_subscription_id=?
ORDER BY ad_date DESC, id DESC LIMIT 120""",
(user["id"], sub["id"]),
).fetchall()
if selected_platform == "all":
rows = rows_all[:60]
agg = conn.execute(
"""SELECT
SUM(impressions) impressions,
SUM(clicks) clicks,
SUM(views) views,
SUM(leads) leads,
SUM(charged_amount) spent
FROM media_performance_daily
WHERE partner_id=? AND media_subscription_id=?""",
(user["id"], sub["id"]),
).fetchone()
else:
rows = conn.execute(
"""SELECT * FROM media_performance_daily
WHERE partner_id=? AND media_subscription_id=? AND platform=?
ORDER BY ad_date DESC, id DESC LIMIT 60""",
(user["id"], sub["id"], selected_platform),
).fetchall()
agg = conn.execute(
"""SELECT
SUM(impressions) impressions,
SUM(clicks) clicks,
SUM(views) views,
SUM(leads) leads,
SUM(charged_amount) spent
FROM media_performance_daily
WHERE partner_id=? AND media_subscription_id=? AND platform=?""",
(user["id"], sub["id"], selected_platform),
).fetchone()
conn.close()
# Build chart-ready daily totals from media rows.
daily = {}
platform_totals = {}
for r in rows:
d = r["ad_date"]
if d not in daily:
daily[d] = {"impressions": 0, "clicks": 0, "views": 0, "charged": 0.0}
daily[d]["impressions"] += r["impressions"] or 0
daily[d]["clicks"] += r["clicks"] or 0
daily[d]["views"] += r["views"] or 0
daily[d]["charged"] += r["charged_amount"] or 0.0
p = r["platform"] or "Unknown"
platform_totals[p] = platform_totals.get(p, 0) + (r["impressions"] or 0)
spend_by_platform = {}
for r in rows_all:
p = r["platform"] or "Unknown"
spend_by_platform[p] = spend_by_platform.get(p, 0.0) + (r["charged_amount"] or 0.0)
daily_keys = sorted(daily.keys())[-7:]
impressions_series = [daily[k]["impressions"] for k in daily_keys] if daily_keys else [0]
compare_series = [daily[k][compare_with] for k in daily_keys] if daily_keys else [0]
spend_series = [round(daily[k]["charged"], 2) for k in daily_keys] if daily_keys else [0]
date_labels = [k[-5:] for k in daily_keys] if daily_keys else ["-"]
impressions_chart = svg_line_chart(
f"Impressions vs {compare_with.title()} (Last 7 Days)",
date_labels,
impressions_series,
compare_series,
theme_primary(),
theme_secondary(),
)
spend_chart = svg_bar_chart(
"Daily Ad Spend (Last 7 Days)",
date_labels,
spend_series,
theme_secondary(),
)
platform_labels = list(platform_totals.keys())[:6] if platform_totals else ["-"]
platform_values = [platform_totals[k] for k in platform_labels] if platform_totals else [0]
platform_chart = svg_bar_chart(
"Impressions by Platform",
platform_labels,
platform_values,
theme_primary(),
)
spend_platform_labels = list(spend_by_platform.keys())[:6] if spend_by_platform else ["-"]
spend_platform_values = [round(spend_by_platform[k], 2) for k in spend_platform_labels] if spend_by_platform else [0]
spend_platform_chart = svg_bar_chart(
"Ad Spend by Platform",
spend_platform_labels,
spend_platform_values,
theme_secondary(),
)
utilization = 0
if sub and sub["budget_total"] > 0:
utilization = int((sub["budget_used"] * 100) / sub["budget_total"])
budget_donut = svg_donut_chart("Budget Utilization", utilization, theme_primary())
platform_options = "" + "".join(
[f"" for p in platforms]
)
rows_html = "".join(
[f"
{html.escape(r['ad_date'])}
{html.escape(r['platform'])}
{r['impressions']}
{r['clicks']}
{r['views']}
{r['leads']}
{round(r['charged_amount'],2)}
" for r in rows]
)
body = f"""
Media Performance
Simple pricing: for every 1,000 impressions we deduct your fixed rate (default {money(50)}). Clicks, views, and leads are shown for reporting only and are not charged separately.
Package: {html.escape(sub['package_name']) if sub else 'Not Active'}
Budget Total: {money(sub['budget_total'],2) if sub else money(0,2)} | Used: {money(sub['budget_used'],2) if sub else money(0,2)} | Remaining: {money(sub['budget_remaining'],2) if sub else money(0,2)}
Rate per 1000 Impressions: {round(sub['rate_per_1000'],2) if sub else 0}
Totals: Impressions {(agg['impressions'] or 0) if agg else 0} | Clicks {(agg['clicks'] or 0) if agg else 0} | Views {(agg['views'] or 0) if agg else 0} | Leads {(agg['leads'] or 0) if agg else 0}
Filter Analytics
Showing data for: {html.escape(selected_platform if selected_platform != 'all' else 'All Platforms')}
{impressions_chart}
{spend_chart}
{platform_chart}
{budget_donut}
{spend_platform_chart}
Daily Performance Entries
Date
Platform
Impressions
Clicks
Views
Leads
Charged
{rows_html or '
No media data yet.
'}
"""
self.send_html(page("Media Performance", body, user))
def post_partner_media(self, user):
self.redirect("/partner/media")
def post_partner_notification_ack(self, user):
data = parse_post_data(self)
notification_id = to_int(first(data, "notification_id", "0") or "0", 0)
conn = db_connect()
conn.execute(
"UPDATE notifications SET is_read=1 WHERE id=? AND partner_id=?",
(notification_id, user["id"]),
)
conn.commit()
conn.close()
self.redirect("/partner/dashboard")
def partner_support(self, user):
conn = db_connect()
tickets = conn.execute("SELECT * FROM tickets WHERE partner_id=? ORDER BY id DESC", (user["id"],)).fetchall()
conn.close()
ticket_html = "".join([f"
Update the company details shown on quotes, invoices, share links, and PDF downloads.
"""
self.send_html(page("Profile", body, user))
def post_partner_profile(self, user):
data = parse_post_data(self)
action = first(data, "action")
conn = db_connect()
if action == "profile":
conn.execute(
"UPDATE users SET name=?, email=?, phone=? WHERE id=?",
(first(data, "name"), first(data, "email").lower(), first(data, "phone"), user["id"]),
)
elif action == "invoice_branding":
upsert_partner_invoice_settings(conn, user["id"], data)
elif action == "password":
current = first(data, "current_password")
new_pw = first(data, "new_password")
db_user = conn.execute("SELECT password_hash FROM users WHERE id=?", (user["id"],)).fetchone()
if db_user and password_verify(current, db_user["password_hash"]) and len(new_pw) >= 6:
conn.execute("UPDATE users SET password_hash=? WHERE id=?", (password_hash(new_pw), user["id"]))
conn.commit()
conn.close()
self.redirect("/partner/profile")
def admin_dashboard(self, user):
if not self.require_admin_perm(user, "dashboard", "view"):
return
query = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
twofa_notice = first(query, "twofa_notice")
twofa_enabled_msg = first(query, "twofa_enabled")
conn = db_connect()
expire_subscriptions(conn)
today = datetime.utcnow().date().isoformat()
p_cond, p_params = self.admin_partner_scope_condition(conn, user, "p")
role_name = self.admin_role_of(conn, user["id"])
scoped_admin_ids = self.admin_scope_admin_ids(conn, user)
sid_placeholders = ",".join(["?"] * len(scoped_admin_ids))
stats = {
"partners": conn.execute(f"SELECT COUNT(*) c FROM partners p WHERE 1=1 {p_cond}", p_params).fetchone()["c"],
"active_subs": conn.execute(
f"""SELECT COUNT(*) c FROM subscriptions s
JOIN partners p ON p.user_id=s.partner_id
WHERE s.status='active' AND s.end_date>=? {p_cond}""",
[today] + p_params,
).fetchone()["c"],
"expired_subs": conn.execute(
f"""SELECT COUNT(*) c FROM subscriptions s
JOIN partners p ON p.user_id=s.partner_id
WHERE (s.status='expired' OR s.end_date=? THEN 1 ELSE 0 END), 0)=0
ORDER BY COALESCE(MAX(s.end_date), '0000-00-00') DESC, u.name
LIMIT 25""",
p_params + [today],
).fetchall()
sales_monthly_system = conn.execute(
f"""SELECT substr(created_at,1,7) ym,
COUNT(*) deals,
COALESCE(SUM(finalized_amount),0) amount,
COALESCE(SUM(finalized_amount*0.05),0) commission_total,
COALESCE(SUM(CASE WHEN date(created_at) <= date('now','-3 months') THEN finalized_amount*0.05 ELSE 0 END),0) commission_releasable
FROM subscriptions
WHERE assigned_by_admin_id IN ({sid_placeholders})
GROUP BY substr(created_at,1,7)
ORDER BY ym DESC LIMIT 6""",
scoped_admin_ids,
).fetchall()
sales_monthly_media = conn.execute(
f"""SELECT substr(created_at,1,7) ym,
COUNT(*) deals,
COALESCE(SUM(budget_total),0) amount,
COALESCE(SUM(budget_total*0.05),0) commission_total,
COALESCE(SUM(CASE WHEN date(created_at) <= date('now','-3 months') THEN budget_total*0.05 ELSE 0 END),0) commission_releasable
FROM media_subscriptions
WHERE assigned_by_admin_id IN ({sid_placeholders})
GROUP BY substr(created_at,1,7)
ORDER BY ym DESC LIMIT 6""",
scoped_admin_ids,
).fetchall()
team_rows = []
if role_name in {"manager", "super_admin"}:
team_rows = conn.execute(
f"""SELECT u.id, u.name, u.email,
COALESCE(SUM(s.finalized_amount),0) system_sales,
COALESCE(SUM(s.finalized_amount*0.05),0) system_commission,
COALESCE(SUM(CASE WHEN date(s.created_at)<=date('now','-3 months') THEN s.finalized_amount*0.05 ELSE 0 END),0) system_release,
COALESCE((
SELECT SUM(ms.budget_total) FROM media_subscriptions ms WHERE ms.assigned_by_admin_id=u.id
),0) media_sales,
COALESCE((
SELECT SUM(ms.budget_total*0.05) FROM media_subscriptions ms WHERE ms.assigned_by_admin_id=u.id
),0) media_commission,
COALESCE((
SELECT SUM(CASE WHEN date(ms.created_at)<=date('now','-3 months') THEN ms.budget_total*0.05 ELSE 0 END)
FROM media_subscriptions ms WHERE ms.assigned_by_admin_id=u.id
),0) media_release
FROM users u
LEFT JOIN subscriptions s ON s.assigned_by_admin_id=u.id
WHERE u.role='admin' AND u.id IN ({sid_placeholders})
GROUP BY u.id, u.name, u.email
ORDER BY system_sales DESC, u.id DESC""",
scoped_admin_ids,
).fetchall()
my_monthly_counts = conn.execute(
f"""SELECT substr(created_at,1,7) ym, COUNT(*) c
FROM subscriptions
WHERE assigned_by_admin_id IN ({sid_placeholders})
GROUP BY substr(created_at,1,7)
ORDER BY ym DESC LIMIT 12""",
scoped_admin_ids,
).fetchall()
my_todos = conn.execute(
"SELECT id, task_text, created_at FROM admin_todos WHERE user_id=? AND status='pending' ORDER BY id DESC LIMIT 20",
(user["id"],),
).fetchall()
manager_override = {"system_override": 0, "media_override": 0}
if role_name == "manager":
manager_override = conn.execute(
"""SELECT
COALESCE((SELECT SUM(s.finalized_amount*0.02)
FROM subscriptions s
JOIN admin_profiles ap ON ap.user_id=s.assigned_by_admin_id
WHERE ap.manager_user_id=?),0) system_override,
COALESCE((SELECT SUM(ms.budget_total*0.02)
FROM media_subscriptions ms
JOIN admin_profiles ap2 ON ap2.user_id=ms.assigned_by_admin_id
WHERE ap2.manager_user_id=?),0) media_override""",
(user["id"], user["id"]),
).fetchone()
conn.close()
cfg = vertical_config()
city_chart = svg_bar_chart(
"Demand by Market",
[r["city"] for r in city_rows] or ["none"],
[r["c"] for r in city_rows] or [0],
theme_primary(),
)
delivery_rows_rev = list(reversed(delivery_rows))
delivery_chart = svg_bar_chart(
"Daily Lead Delivery (7 days)",
[r["d"][-5:] for r in delivery_rows_rev] or ["-"],
[r["c"] for r in delivery_rows_rev] or [0],
theme_secondary(),
)
approval_ratio = int(((invalid_ratio["approved"] or 0) * 100) / max(1, (invalid_ratio["approved"] or 0) + (invalid_ratio["rejected"] or 0)))
trend_line = svg_line_chart(
"Lead Delivery Trend",
[r["d"][-5:] for r in delivery_rows_rev] or ["-"],
[r["c"] for r in delivery_rows_rev] or [0],
[max(0, int(r["c"] * 0.7)) for r in delivery_rows_rev] or [0],
theme_primary(),
theme_secondary(),
)
donut = svg_donut_chart("Invalid Approval Ratio", approval_ratio, theme_primary())
media_daily_rev = list(reversed(media_daily))
media_spend_chart = svg_line_chart(
"Media Impressions vs Charged (7 days)",
[r["ad_date"][-5:] for r in media_daily_rev] or ["-"],
[r["impressions"] for r in media_daily_rev] or [0],
[r["charged"] for r in media_daily_rev] or [0],
theme_primary(),
theme_secondary(),
)
category_chart = svg_bar_chart(
"Service Category Demand",
[format_service_category(r["service_type"]) for r in category_rows] or ["No enquiries"],
[r["c"] for r in category_rows] or [0],
theme_primary(),
) if is_removals_vertical() else ""
postcode_chart = svg_bar_chart(
"Origin Postcode Heatmap",
[r["postcode"] for r in postcode_rows] or ["No data"],
[r["c"] for r in postcode_rows] or [0],
theme_secondary(),
) if is_removals_vertical() else ""
sales_rows = ""
monthly_map = {}
for r in sales_monthly_system:
monthly_map[r["ym"]] = {
"system_deals": r["deals"] or 0,
"system_amount": round(r["amount"] or 0, 2),
"system_commission": round(r["commission_total"] or 0, 2),
"system_release": round(r["commission_releasable"] or 0, 2),
"media_deals": 0,
"media_amount": 0,
"media_commission": 0,
"media_release": 0,
}
for r in sales_monthly_media:
row = monthly_map.get(r["ym"], {
"system_deals": 0, "system_amount": 0, "system_commission": 0, "system_release": 0,
"media_deals": 0, "media_amount": 0, "media_commission": 0, "media_release": 0,
})
row["media_deals"] = r["deals"] or 0
row["media_amount"] = round(r["amount"] or 0, 2)
row["media_commission"] = round(r["commission_total"] or 0, 2)
row["media_release"] = round(r["commission_releasable"] or 0, 2)
monthly_map[r["ym"]] = row
for ym in sorted(monthly_map.keys(), reverse=True):
x = monthly_map[ym]
sales_rows += (
f"
{round((r['system_commission'] or 0) + (r['media_commission'] or 0),2)}
{round((r['system_release'] or 0) + (r['media_release'] or 0),2)}
"
for r in team_rows
]
)
+ "
"
)
monthly_count_map = {r["ym"]: (r["c"] or 0) for r in my_monthly_counts}
today_local = local_now().date()
this_ym = today_local.strftime("%Y-%m")
prev_ym = (today_local.replace(day=1) - timedelta(days=1)).strftime("%Y-%m")
this_month_sales = monthly_count_map.get(this_ym, 0)
prev_month_sales = monthly_count_map.get(prev_ym, 0)
target = 3
target_left = max(0, target - this_month_sales)
payout_hold = this_month_sales == 0 and prev_month_sales == 0
bonus_eligible = True
for i in range(12):
ref = (today_local.replace(day=1) - timedelta(days=30 * i)).strftime("%Y-%m")
if monthly_count_map.get(ref, 0) < 3:
bonus_eligible = False
break
sales_stat_cards = "".join(
[
dashboard_stat_card("This Month Sales", this_month_sales, note="Completed subscription sales this month", badge=f"Target {target}", icon="MS", tone="primary"),
dashboard_stat_card("Sales Needed", target_left, note="Deals still needed to hit monthly target", badge="Urgent focus" if target_left else "Target hit", icon="TG", tone="dark"),
dashboard_stat_card("Payout Status", "Hold" if payout_hold else "Eligible", note="Commission pays after the 3-month hold period", badge="2nd zero month" if payout_hold else "Clear", icon="PY", tone="secondary"),
dashboard_stat_card("Year Bonus Track", "On Track" if bonus_eligible else "Not Yet", note="Requires 3 or more sales every month", badge="12-month run", icon="YR", tone="soft"),
]
)
if role_name == "manager":
sales_stat_cards += dashboard_stat_card(
"Manager Override",
money((manager_override["system_override"] or 0) + (manager_override["media_override"] or 0), 2),
note="2% override across assigned staff sales",
badge="Manager bonus",
icon="MO",
tone="primary",
)
sales_progress_stack = "".join(
[
dashboard_progress("Monthly target completion", int((this_month_sales * 100) / max(1, target)), f"{this_month_sales} of {target} sales"),
dashboard_progress("Annual consistency", 100 if bonus_eligible else int((max(0, 12 - target_left) / 12) * 100), "Measures consistency against the bonus threshold"),
]
)
inactive_rows_html = "".join(
[
f"
{html.escape(r['name'])}
{html.escape(r['company_name'] or '-')}
{html.escape(r['email'] or '-')}
{html.escape(r['phone'] or '-')}
{html.escape(r['last_end_date'] or 'No subscription')}
"
for r in inactive_partners
]
)
quality_info = dashboard_info_rows(
[
("Invalid approved", invalid_ratio["approved"] or 0),
("Invalid rejected", invalid_ratio["rejected"] or 0),
("Pending invalid", stats["pending_invalid"]),
("Open tickets", stats["open_tickets"]),
("Approval ratio", f"{approval_ratio}%"),
]
)
media_info = dashboard_info_rows(
[
("Media impressions", media_totals["impressions"] or 0),
("Media clicks", media_totals["clicks"] or 0),
("Media views", media_totals["views"] or 0),
("Media leads", media_totals["leads"] or 0),
("Charged", money(media_totals["charged"] or 0, 2)),
]
)
if role_name != "super_admin":
hour = local_now().hour
greet = "Good Morning" if hour < 12 else ("Good Afternoon" if hour < 18 else "Good Evening")
staff_name = user["name"] or "Team Member"
quotes = [
"Work hard in silence, let results make the noise.",
"Consistency today builds your biggest wins tomorrow.",
"Every follow-up you make moves you closer to your target.",
"Discipline beats motivation when pressure gets real.",
"Small daily actions create massive yearly results.",
"Stay focused, stay positive, and close the next deal.",
]
sid = self.cookie_value("session_id")
idx = (sum(ord(c) for c in (sid or now_iso())) % len(quotes))
quote = quotes[idx]
setup_notice_html = ""
if twofa_notice == "1":
setup_notice_html = (
f"
"
"
Security Notice
"
"
First login is allowed. From next login, Google Authenticator code will be required. Please set it up now.
", user), 403)
my_role = self.admin_role_of(conn, user["id"])
rows = conn.execute(
"""SELECT u.id, u.name, u.email, u.phone, u.status, ap.admin_role,
ap.manager_user_id, mu.name AS manager_name
FROM users u
LEFT JOIN admin_profiles ap ON ap.user_id=u.id
LEFT JOIN users mu ON mu.id=ap.manager_user_id
WHERE u.role='admin'
ORDER BY u.id DESC"""
).fetchall()
managers = conn.execute(
"""SELECT u.id, u.name
FROM users u JOIN admin_profiles ap ON ap.user_id=u.id
WHERE u.role='admin' AND ap.admin_role IN ('manager','super_admin')
ORDER BY u.name"""
).fetchall()
reset_rows = conn.execute(
"""SELECT prr.*, u.name AS partner_name, u.email AS partner_email
FROM password_reset_requests prr
JOIN users u ON u.id=prr.user_id
ORDER BY prr.id DESC LIMIT 100"""
).fetchall()
perm_rows = conn.execute(
"SELECT user_id, module, can_view, can_add, can_edit, can_delete FROM admin_permissions ORDER BY user_id, module"
).fetchall()
except sqlite3.OperationalError:
ensure_feature_tables(conn)
my_role = self.admin_role_of(conn, user["id"])
rows = conn.execute(
"""SELECT u.id, u.name, u.email, u.phone, u.status, ap.admin_role,
ap.manager_user_id, mu.name AS manager_name
FROM users u
LEFT JOIN admin_profiles ap ON ap.user_id=u.id
LEFT JOIN users mu ON mu.id=ap.manager_user_id
WHERE u.role='admin'
ORDER BY u.id DESC"""
).fetchall()
managers = conn.execute(
"""SELECT u.id, u.name
FROM users u JOIN admin_profiles ap ON ap.user_id=u.id
WHERE u.role='admin' AND ap.admin_role IN ('manager','super_admin')
ORDER BY u.name"""
).fetchall()
reset_rows = conn.execute(
"""SELECT prr.*, u.name AS partner_name, u.email AS partner_email
FROM password_reset_requests prr
JOIN users u ON u.id=prr.user_id
ORDER BY prr.id DESC LIMIT 100"""
).fetchall()
perm_rows = conn.execute(
"SELECT user_id, module, can_view, can_add, can_edit, can_delete FROM admin_permissions ORDER BY user_id, module"
).fetchall()
conn.commit()
finally:
conn.close()
manager_opts_base = "" + "".join(
[f"" for m in managers]
)
twofa_state = {}
conn2 = db_connect()
for r2 in conn2.execute(
"SELECT user_id, COALESCE(twofa_enabled,0) twofa_enabled FROM admin_profiles"
).fetchall():
twofa_state[r2["user_id"]] = int(r2["twofa_enabled"] or 0)
conn2.close()
trs = "".join(
[f"""
{r['id']}
{html.escape(r['name'])}
{html.escape(r['email'])}
{html.escape(r['admin_role'] or 'staff')}
{html.escape(r['manager_name'] or '-')}
{html.escape(r['status'])}
{"" if my_role == "super_admin" and (r['admin_role'] or 'staff') != 'super_admin' else ""}
"""
for r in rows]
)
reset_trs = ""
for r in reset_rows:
action_html = ""
if r["status"] == "pending":
action_html = (
f""
)
reset_trs += (
f"
{r['id']}
{safe_text(r['partner_name'])}
"
f"
{safe_text(r['partner_email'])}
{safe_text(r['status'])}
"
f"
{safe_text(r['created_at'])}
{action_html}
"
)
perm_map = {}
for pr in perm_rows:
perm_map[(pr["user_id"], pr["module"])] = pr
permissions_html = ""
for urow in rows:
if urow["id"] == user["id"] and my_role != "super_admin":
continue
rows_html = ""
for module in ADMIN_MODULES:
p = perm_map.get((urow["id"], module), {"can_view": 1, "can_add": 0, "can_edit": 0, "can_delete": 0})
rows_html += (
f"
", user), 403)
my_role = self.admin_role_of(conn, user["id"])
if my_role != "super_admin" and user["id"] != target_user_id:
conn.close()
return self.redirect("/admin/users")
conn.execute(
"UPDATE users SET name=?, email=?, phone=?, status=? WHERE id=? AND role='admin'",
(first(data, "name"), first(data, "email").lower(), first(data, "phone"), first(data, "status", "active"), target_user_id),
)
if my_role == "super_admin":
exists_ap = conn.execute("SELECT user_id FROM admin_profiles WHERE user_id=?", (target_user_id,)).fetchone()
manager_user_id = to_int(first(data, "manager_user_id", "0"), 0)
if manager_user_id <= 0:
manager_user_id = None
if exists_ap:
conn.execute(
"UPDATE admin_profiles SET admin_role=?, manager_user_id=? WHERE user_id=?",
(first(data, "admin_role", "staff"), manager_user_id, target_user_id),
)
else:
conn.execute(
"INSERT INTO admin_profiles (user_id, admin_role, manager_user_id, created_at) VALUES (?, ?, ?, ?)",
(target_user_id, first(data, "admin_role", "staff"), manager_user_id, now_iso()),
)
new_pw = first(data, "new_password")
if new_pw:
conn.execute("UPDATE users SET password_hash=? WHERE id=? AND role='admin'", (password_hash(new_pw), target_user_id))
conn.commit()
conn.close()
self.redirect("/admin/users")
def post_admin_user_delete(self, user):
data = parse_post_data(self)
target_user_id = int(first(data, "user_id", "0") or "0")
conn = db_connect()
if not self.admin_has_perm(conn, user, "users", "delete"):
conn.close()
return self.send_html(page("Forbidden", "
Permission denied.
", user), 403)
if target_user_id == user["id"]:
conn.close()
return self.redirect("/admin/users")
try:
conn.execute("DELETE FROM admin_permissions WHERE user_id=?", (target_user_id,))
conn.execute("DELETE FROM admin_profiles WHERE user_id=?", (target_user_id,))
conn.execute("DELETE FROM users WHERE id=? AND role='admin'", (target_user_id,))
conn.commit()
except sqlite3.IntegrityError:
conn.execute("UPDATE users SET status='suspended' WHERE id=? AND role='admin'", (target_user_id,))
conn.commit()
conn.close()
self.redirect("/admin/users")
def post_admin_user_permissions(self, user):
data = parse_post_data(self)
target_user_id = int(first(data, "user_id", "0") or "0")
conn = db_connect()
if self.admin_role_of(conn, user["id"]) != "super_admin":
conn.close()
return self.send_html(page("Forbidden", "
Permission denied.
", user), 403)
ensure_admin_permissions_rows(conn, target_user_id, full_access=False)
for module in ADMIN_MODULES:
can_view = 1 if first(data, f"perm__{module}__view") else 0
can_add = 1 if first(data, f"perm__{module}__add") else 0
can_edit = 1 if first(data, f"perm__{module}__edit") else 0
can_delete = 1 if first(data, f"perm__{module}__delete") else 0
conn.execute(
"""UPDATE admin_permissions
SET can_view=?, can_add=?, can_edit=?, can_delete=?
WHERE user_id=? AND module=?""",
(can_view, can_add, can_edit, can_delete, target_user_id, module),
)
conn.commit()
conn.close()
self.redirect("/admin/users")
def post_admin_password_request_resolve(self, user):
data = parse_post_data(self)
request_id = to_int(first(data, "request_id", "0") or "0", 0)
decision = first(data, "decision", "completed")
notes = first(data, "admin_notes")
new_password = first(data, "new_password")
if decision not in {"completed", "rejected"}:
return self.redirect("/admin/users")
conn = db_connect()
req = conn.execute(
"SELECT * FROM password_reset_requests WHERE id=?",
(request_id,),
).fetchone()
if not req or req["status"] != "pending":
conn.close()
return self.redirect("/admin/users")
if decision == "completed":
if len(new_password) < 6:
conn.close()
return self.redirect("/admin/users")
conn.execute(
"UPDATE users SET password_hash=? WHERE id=? AND role='partner'",
(password_hash(new_password), req["user_id"]),
)
conn.execute(
"""UPDATE password_reset_requests
SET status=?, admin_id=?, admin_notes=?, resolved_at=?
WHERE id=?""",
(decision, user["id"], notes, now_iso(), request_id),
)
add_notification(
conn,
req["user_id"],
"Password Request Updated",
"Your password reset request was completed." if decision == "completed" else "Your password reset request was rejected.",
)
conn.commit()
conn.close()
self.redirect("/admin/users")
def post_admin_user_2fa_reset(self, user):
data = parse_post_data(self)
target_user_id = to_int(first(data, "user_id", "0"), 0)
conn = db_connect()
if self.admin_role_of(conn, user["id"]) != "super_admin":
conn.close()
return self.send_html(page("Forbidden", "
Permission denied.
", user), 403)
target_role = self.admin_role_of(conn, target_user_id)
if target_role == "super_admin":
conn.close()
return self.redirect("/admin/users")
conn.execute(
"UPDATE admin_profiles SET twofa_secret='', twofa_enabled=0, twofa_grace_used=0 WHERE user_id=?",
(target_user_id,),
)
conn.execute("DELETE FROM admin_login_challenges WHERE user_id=?", (target_user_id,))
conn.commit()
conn.close()
self.redirect("/admin/users")
def admin_profile(self, user):
if not self.require_admin_perm(user, "profile", "view"):
return
conn = db_connect()
role_name = self.admin_role_of(conn, user["id"])
conn.close()
body = f"""
Admin Profile
Role: {html.escape(role_name)}
Change Password
"""
self.send_html(page("Admin Profile", body, user))
def post_admin_profile(self, user):
if not self.require_admin_perm(user, "profile", "edit"):
return
data = parse_post_data(self)
action = first(data, "action")
conn = db_connect()
if action == "profile":
conn.execute(
"UPDATE users SET name=?, email=?, phone=? WHERE id=? AND role='admin'",
(first(data, "name"), first(data, "email").lower(), first(data, "phone"), user["id"]),
)
elif action == "password":
current = first(data, "current_password")
new_pw = first(data, "new_password")
row = conn.execute("SELECT password_hash FROM users WHERE id=? AND role='admin'", (user["id"],)).fetchone()
if row and password_verify(current, row["password_hash"]) and len(new_pw) >= 6:
conn.execute("UPDATE users SET password_hash=? WHERE id=?", (password_hash(new_pw), user["id"]))
conn.commit()
conn.close()
self.redirect("/admin/profile")
def admin_enquiries(self, user):
if not self.require_admin_perm(user, "enquiries", "view"):
return
cfg = vertical_config()
parsed = urllib.parse.urlparse(self.path)
q = urllib.parse.parse_qs(parsed.query)
status_filter = (q.get("status", [""])[0] or "").strip().lower()
type_filter = (q.get("type", [""])[0] or "").strip().lower()
where = "WHERE 1=1"
params = []
if status_filter:
where += " AND status=?"
params.append(status_filter)
if type_filter in {"partner_application", "buyer_enquiry"}:
where += " AND enquiry_type=?"
params.append(type_filter)
conn = db_connect()
rows = conn.execute(
f"""SELECT * FROM enquiries
{where}
ORDER BY id DESC LIMIT 500""",
params,
).fetchall()
conn.close()
trs = "".join(
[f"
{r['id']}
{html.escape(r['enquiry_type'])}
{html.escape(r['full_name'])}
{html.escape(r['phone'])}
{html.escape(r['city'] or '-')}
{html.escape(r['property_type'] or '-')}
{r['budget_min'] or 0}-{r['budget_max'] or 0}
{html.escape(r['status'])}
{'Yes' if r['email_sent'] else 'No'}
" for r in rows]
)
body = f"""
Enquiries
ID
Type
Name
Phone
City
Property
Budget
Status
Email Sent
Action
{trs or '
No enquiries.
'}
"""
self.send_html(page("Enquiries", body, user))
def post_admin_enquiry_update(self, user):
if not self.require_admin_perm(user, "enquiries", "edit"):
return
data = parse_post_data(self)
row_id = to_int(first(data, "id", "0"), 0)
status = first(data, "status", "new")
if status not in {"new", "contacted", "qualified", "closed", "rejected"}:
status = "new"
conn = db_connect()
conn.execute(
"UPDATE enquiries SET status=?, admin_notes=? WHERE id=?",
(status, first(data, "admin_notes"), row_id),
)
conn.commit()
conn.close()
self.redirect("/admin/enquiries")
def admin_landing_cms(self, user):
if not self.require_admin_perm(user, "landing_cms", "view"):
return
cfg = vertical_config()
conn = db_connect()
settings = conn.execute("SELECT * FROM landing_settings ORDER BY key").fetchall()
badges = conn.execute("SELECT * FROM partner_badges ORDER BY sort_order ASC, id DESC").fetchall()
conn.close()
s = {r["key"]: r["value"] for r in settings}
badge_rows = "".join(
[f"
"""
self.send_html(page(cfg["operator_plural"], body, user))
def post_admin_partners(self, user):
data = parse_post_data(self)
conn = db_connect()
if not self.admin_has_perm(conn, user, "partners", "add"):
conn.close()
return self.send_html(page("Forbidden", "
Permission denied.
", user), 403)
name = first(data, "name")
email = first(data, "email").lower()
phone = first(data, "phone")
pw = first(data, "password")
partner_type = "provider" if is_removals_vertical() else first(data, "partner_type", "agent")
company = first(data, "company_name")
city = first(data, "city")
areas = [x.strip() for x in first(data, "areas").split(",") if x.strip()]
try:
conn.execute(
"INSERT INTO users (name, email, phone, password_hash, role, status, created_at) VALUES (?, ?, ?, ?, 'partner', 'active', ?)",
(name, email, phone, password_hash(pw), now_iso()),
)
uid = conn.execute("SELECT id FROM users WHERE email=?", (email,)).fetchone()["id"]
conn.execute(
"""INSERT INTO partners
(user_id, partner_type, company_name, city, areas_json, address, verified_status, created_by_admin_id, created_at)
VALUES (?, ?, ?, ?, ?, '', 'verified', ?, ?)""",
(uid, partner_type, company, city, json.dumps(areas), user["id"], now_iso()),
)
if is_removals_vertical():
upsert_partner_vertical_profile(conn, uid, data)
conn.commit()
finally:
conn.close()
self.redirect("/admin/partners")
def post_admin_partner_update(self, user):
data = parse_post_data(self)
partner_id = int(first(data, "partner_id", "0") or "0")
conn = db_connect()
if not self.admin_has_perm(conn, user, "partners", "edit"):
conn.close()
return self.send_html(page("Forbidden", "
Permission denied.
", user), 403)
if not self.partner_in_admin_scope(conn, user, partner_id):
conn.close()
return self.redirect("/admin/partners")
conn.execute(
"UPDATE users SET name=?, email=?, phone=?, status=? WHERE id=? AND role='partner'",
(first(data, "name"), first(data, "email").lower(), first(data, "phone"), first(data, "status", "active"), partner_id),
)
conn.execute(
"UPDATE partners SET company_name=?, city=? WHERE user_id=?",
(first(data, "company_name"), first(data, "city"), partner_id),
)
if is_removals_vertical():
upsert_partner_vertical_profile(conn, partner_id, data)
new_pw = first(data, "new_password")
if new_pw:
conn.execute("UPDATE users SET password_hash=? WHERE id=? AND role='partner'", (password_hash(new_pw), partner_id))
conn.commit()
conn.close()
self.redirect("/admin/partners")
def post_admin_partner_delete(self, user):
data = parse_post_data(self)
partner_id = int(first(data, "partner_id", "0") or "0")
conn = db_connect()
if not self.admin_has_perm(conn, user, "partners", "delete"):
conn.close()
return self.send_html(page("Forbidden", "
Permission denied.
", user), 403)
if not self.partner_in_admin_scope(conn, user, partner_id):
conn.close()
return self.redirect("/admin/partners")
try:
conn.execute("DELETE FROM partners WHERE user_id=?", (partner_id,))
conn.execute("DELETE FROM users WHERE id=? AND role='partner'", (partner_id,))
conn.commit()
except sqlite3.IntegrityError:
conn.execute("UPDATE users SET status='suspended' WHERE id=? AND role='partner'", (partner_id,))
conn.commit()
conn.close()
self.redirect("/admin/partners")
def post_admin_partner_auto_assign(self, user):
data = parse_post_data(self)
partner_id = int(first(data, "partner_id", "0") or "0")
conn = db_connect()
if not self.admin_has_perm(conn, user, "partners", "edit"):
conn.close()
return self.send_html(page("Forbidden", "
Permission denied.
", user), 403)
if not self.partner_in_admin_scope(conn, user, partner_id):
conn.close()
return self.redirect("/admin/partners")
sub = active_subscription(conn, partner_id)
if sub:
assigned = self.auto_assign_partner_leads(conn, partner_id, sub)
add_notification(
conn,
partner_id,
"Auto Assignment Completed",
f"{assigned} leads were auto-assigned from stored lead pool.",
)
conn.commit()
conn.close()
self.redirect("/admin/partners")
def admin_packages(self, user):
if not self.require_admin_perm(user, "packages", "view"):
return
conn = db_connect()
perms = self.admin_ui_perms(conn, user, "packages")
rows = conn.execute("SELECT * FROM packages ORDER BY id DESC").fetchall()
package_options = conn.execute("SELECT id, name FROM packages ORDER BY name").fetchall()
conn.close()
option_html = "".join([f"" for p in package_options])
trs = "".join(
[f"""
{r['id']}
{html.escape(r['name'])}
{r['monthly_credits']}
{r['price_monthly']}
{r['price_annual']}
{r['cities_allowed']}
{f"" if perms['edit'] else '-'}
{f"" if perms['delete'] else '-'}
"""
for r in rows]
)
body = f"""
{'''
Create Package
''' if perms['add'] else ''}
Packages
ID
Name
Credits
Monthly
Annual
Cities
Edit
Delete
{trs}
"""
self.send_html(page("Packages", body, user))
def post_admin_packages(self, user):
if not self.require_admin_perm(user, "packages", "add"):
return
data = parse_post_data(self)
conn = db_connect()
conn.execute(
"""INSERT INTO packages
(name, monthly_credits, price_monthly, price_annual, cities_allowed, features_json, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(
first(data, "name"),
int(first(data, "monthly_credits", "0") or "0"),
int(first(data, "price_monthly", "0") or "0"),
int(first(data, "price_annual", "0") or "0"),
int(first(data, "cities_allowed", "1") or "1"),
json.dumps({"invalid_replacement": True}),
now_iso(),
),
)
conn.commit()
conn.close()
self.redirect("/admin/packages")
def post_admin_package_update(self, user):
if not self.require_admin_perm(user, "packages", "edit"):
return
data = parse_post_data(self)
package_id = int(first(data, "package_id", "0") or "0")
conn = db_connect()
conn.execute(
"""UPDATE packages
SET name=?, monthly_credits=?, price_monthly=?, price_annual=?, cities_allowed=?
WHERE id=?""",
(
first(data, "name"),
int(first(data, "monthly_credits", "0") or "0"),
int(first(data, "price_monthly", "0") or "0"),
int(first(data, "price_annual", "0") or "0"),
int(first(data, "cities_allowed", "1") or "1"),
package_id,
),
)
conn.commit()
conn.close()
self.redirect("/admin/packages")
def post_admin_package_delete(self, user):
if not self.require_admin_perm(user, "packages", "delete"):
return
data = parse_post_data(self)
package_id = int(first(data, "package_id", "0") or "0")
replacement = first(data, "replacement_package_id")
replacement_id = int(replacement) if replacement else None
conn = db_connect()
in_use = conn.execute("SELECT COUNT(*) c FROM subscriptions WHERE package_id=?", (package_id,)).fetchone()["c"]
if in_use > 0 and not replacement_id:
conn.close()
return self.redirect("/admin/packages")
if replacement_id and replacement_id != package_id:
replacement_pkg = conn.execute("SELECT monthly_credits FROM packages WHERE id=?", (replacement_id,)).fetchone()
if replacement_pkg:
conn.execute(
"""UPDATE subscriptions
SET package_id=?,
credits_monthly=?,
credits_remaining=CASE WHEN ? - credits_used > 0 THEN ? - credits_used ELSE 0 END
WHERE package_id=?""",
(replacement_id, replacement_pkg["monthly_credits"], replacement_pkg["monthly_credits"], replacement_pkg["monthly_credits"], package_id),
)
remaining_refs = conn.execute("SELECT COUNT(*) c FROM subscriptions WHERE package_id=?", (package_id,)).fetchone()["c"]
if remaining_refs == 0:
conn.execute("DELETE FROM packages WHERE id=?", (package_id,))
conn.commit()
conn.close()
self.redirect("/admin/packages")
def admin_media(self, user):
if not self.require_admin_perm(user, "media", "view"):
return
cfg = vertical_config()
conn = db_connect()
my_role = self.admin_role_of(conn, user["id"])
perms = self.admin_ui_perms(conn, user, "media")
can_media_log = (my_role == "super_admin") or perms["edit"]
p_cond, p_params = self.admin_partner_scope_condition(conn, user, "p")
expire_subscriptions(conn)
today = datetime.utcnow().date().isoformat()
conn.execute("UPDATE media_subscriptions SET status='expired' WHERE status='active' AND end_date < ?", (today,))
partners = conn.execute(
"SELECT u.id, u.name, p.company_name FROM users u JOIN partners p ON p.user_id=u.id WHERE 1=1 "
+ p_cond
+ " ORDER BY u.name",
p_params,
).fetchall()
subs = conn.execute(
"""SELECT ms.*, u.name
FROM media_subscriptions ms
JOIN users u ON u.id=ms.partner_id
JOIN partners p ON p.user_id=ms.partner_id
WHERE 1=1 """
+ p_cond
+ " ORDER BY ms.id DESC",
p_params,
).fetchall()
stats = conn.execute(
"""SELECT mpd.id, mpd.ad_date, mpd.platform, mpd.impressions, mpd.clicks, mpd.views, mpd.leads, mpd.charged_amount, u.name
FROM media_performance_daily mpd
JOIN users u ON u.id=mpd.partner_id
JOIN partners p ON p.user_id=mpd.partner_id
WHERE 1=1 """
+ p_cond
+ " ORDER BY mpd.id DESC LIMIT 100",
p_params,
).fetchall()
conn.commit()
conn.close()
partner_opts = "".join([f"" for p in partners])
subs_rows = ""
for s in subs:
approve_actions = ""
if my_role == "super_admin" and (s["approval_status"] if "approval_status" in s.keys() else "approved") == "pending":
approve_actions = (
f""
f""
)
proof_link = s["payment_proof_path"] if "payment_proof_path" in s.keys() else ""
proof_cell = f"View" if proof_link else "No"
subs_rows += (
f"
{s['id']}
{html.escape(s['name'])}
{html.escape(s['package_name'])}
{html.escape(s['status'])}
"
f"
{round(s['budget_total'],2)}
{round(s['budget_used'],2)}
{round(s['budget_remaining'],2)}
{round(s['rate_per_1000'],2)}
"
f"
{html.escape(s['start_date'])} to {html.escape(s['end_date'])}
"
f"
{html.escape((s['approval_status'] if 'approval_status' in s.keys() else 'approved'))}
"
f"
{proof_cell}
"
f"
{approve_actions}
"
)
stats_rows = "".join([f"
{html.escape(r['ad_date'])}
{html.escape(r['name'])}
{html.escape(r['platform'])}
{r['impressions']}
{r['clicks']}
{r['views']}
{r['leads']}
{round(r['charged_amount'],2)}
" for r in stats])
super_admin_budget_card = ""
super_admin_rate_card = ""
if my_role == "super_admin":
super_admin_budget_card = f"""
Correct Media Budget (Super Admin)
Use only if staff entered the wrong values. This directly sets budget total/used/remaining.
"""
super_admin_rate_card = f"""
Update Rate Per 1000 Impressions
This changes only the rate. Budget total/used/remaining are not modified.
"""
manage_forms = ""
if perms["add"]:
manage_forms = f"""
Assign Media Buying Package
Budget is charged by impressions only. Cost formula: `(impressions/1000) * rate_per_1000`
Add Media Budget Credit
{super_admin_rate_card}
{super_admin_budget_card}
"""
if can_media_log:
manage_forms += f"""
''' if can_media_log else ''}
"""
self.send_html(page("Media Ads", body, user))
def post_admin_media(self, user):
if not self.require_admin_perm(user, "media", "add"):
return
data = parse_post_data(self)
action = first(data, "action")
partner_id = int(first(data, "partner_id", "0") or "0")
conn = db_connect()
my_role = self.admin_role_of(conn, user["id"])
can_media_log = (my_role == "super_admin") or self.admin_has_perm(conn, user, "media", "edit")
if partner_id and not self.partner_in_admin_scope(conn, user, partner_id):
conn.close()
return self.redirect("/admin/media")
if action == "assign":
start_date = first(data, "start_date")
end_date = first(data, "end_date")
budget_total = to_float(first(data, "budget_total", "0") or "0", 0.0)
rate = to_float(first(data, "rate_per_1000", "50") or "50", 50.0)
proof_path = save_uploaded_file(data, "payment_proof_file", "media_proofs")
conn.execute("UPDATE media_subscriptions SET status='expired' WHERE partner_id=? AND status='active'", (partner_id,))
approved = my_role == "super_admin"
if not approved and not proof_path:
conn.close()
return self.redirect("/admin/media")
conn.execute(
"""INSERT INTO media_subscriptions
(partner_id, package_name, start_date, end_date, status, budget_total, budget_used, budget_remaining, rate_per_1000, assigned_by_admin_id, approval_status, payment_proof_path, approved_by_admin_id, approved_at, created_at)
VALUES (?, ?, ?, ?, ?, ?, 0, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
partner_id,
first(data, "package_name"),
start_date,
end_date,
"active" if approved else "paused",
budget_total,
budget_total,
rate,
user["id"],
"approved" if approved else "pending",
proof_path,
user["id"] if approved else None,
now_iso() if approved else None,
now_iso(),
),
)
add_notification(conn, partner_id, "Media Package Submitted" if not approved else "Media Package Assigned", f"A new media package has been {'submitted for approval' if not approved else 'assigned'} with budget {budget_total}.")
elif action in {"approve_subscription", "reject_subscription"}:
if my_role != "super_admin":
conn.close()
return self.redirect("/admin/media")
sub_id = to_int(first(data, "subscription_id", "0"), 0)
sub_row = conn.execute("SELECT * FROM media_subscriptions WHERE id=?", (sub_id,)).fetchone()
if not sub_row:
conn.close()
return self.redirect("/admin/media")
if action == "approve_subscription":
conn.execute(
"UPDATE media_subscriptions SET approval_status='approved', status='active', approved_by_admin_id=?, approved_at=? WHERE id=?",
(user["id"], now_iso(), sub_id),
)
add_notification(conn, sub_row["partner_id"], "Media Package Approved", "Your media package has been approved and activated.")
else:
conn.execute(
"UPDATE media_subscriptions SET approval_status='rejected', status='paused' WHERE id=?",
(sub_id,),
)
add_notification(conn, sub_row["partner_id"], "Media Package Rejected", "Your media package was rejected. Contact support.")
elif action == "adjust_budget":
amount = to_float(first(data, "amount", "0") or "0", 0.0)
sub = conn.execute("SELECT * FROM media_subscriptions WHERE partner_id=? AND status='active' ORDER BY id DESC LIMIT 1", (partner_id,)).fetchone()
if sub and amount > 0:
conn.execute(
"UPDATE media_subscriptions SET budget_total=budget_total+?, budget_remaining=budget_remaining+? WHERE id=?",
(amount, amount, sub["id"]),
)
add_notification(conn, partner_id, "Media Budget Added", f"Additional media budget credited: {amount}.")
elif action == "update_rate":
if self.admin_role_of(conn, user["id"]) != "super_admin":
conn.close()
return self.redirect("/admin/media")
new_rate = to_float(first(data, "rate_per_1000", "50") or "50", 50.0)
sub = conn.execute("SELECT * FROM media_subscriptions WHERE partner_id=? AND status='active' ORDER BY id DESC LIMIT 1", (partner_id,)).fetchone()
if sub:
conn.execute(
"UPDATE media_subscriptions SET rate_per_1000=? WHERE id=?",
(new_rate, sub["id"]),
)
add_notification(
conn,
partner_id,
"Media Rate Updated",
f"Rate updated to {new_rate} per 1000 impressions. Budget values unchanged.",
)
elif action == "edit_budget":
if self.admin_role_of(conn, user["id"]) == "super_admin":
sub = conn.execute("SELECT * FROM media_subscriptions WHERE partner_id=? AND status='active' ORDER BY id DESC LIMIT 1", (partner_id,)).fetchone()
if sub:
budget_total = to_float(first(data, "budget_total", "0") or "0", 0.0)
budget_used = to_float(first(data, "budget_used", "0") or "0", 0.0)
budget_remaining = to_float(first(data, "budget_remaining", "0") or "0", 0.0)
conn.execute(
"UPDATE media_subscriptions SET budget_total=?, budget_used=?, budget_remaining=? WHERE id=?",
(budget_total, budget_used, budget_remaining, sub["id"]),
)
add_notification(
conn,
partner_id,
"Media Budget Corrected",
"Your media budget values were corrected by super admin.",
)
elif action == "log_performance":
if not can_media_log:
conn.close()
return self.redirect("/admin/media")
sub = conn.execute("SELECT * FROM media_subscriptions WHERE partner_id=? AND status='active' ORDER BY id DESC LIMIT 1", (partner_id,)).fetchone()
if sub:
impressions = to_int(first(data, "impressions", "0") or "0", 0)
clicks = to_int(first(data, "clicks", "0") or "0", 0)
views = to_int(first(data, "views", "0") or "0", 0)
leads = to_int(first(data, "leads", "0") or "0", 0)
charged = media_cost_from_impressions(impressions, sub["rate_per_1000"])
if charged <= sub["budget_remaining"]:
conn.execute(
"""INSERT INTO media_performance_daily
(partner_id, media_subscription_id, ad_date, platform, impressions, clicks, views, leads, charged_amount, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(partner_id, sub["id"], first(data, "ad_date"), first(data, "platform"), impressions, clicks, views, leads, charged, now_iso()),
)
conn.execute(
"UPDATE media_subscriptions SET budget_used=budget_used+?, budget_remaining=budget_remaining-? WHERE id=?",
(charged, charged, sub["id"]),
)
add_notification(
conn,
partner_id,
"Media Performance Logged",
f"Ad performance recorded. Charged {charged} from media credit based on impressions.",
)
else:
add_notification(
conn,
partner_id,
"Media Spend Skipped",
f"Insufficient media budget for impressions {impressions}.",
)
elif action == "import_performance":
if not can_media_log:
conn.close()
return self.redirect("/admin/media")
csv_data = first(data, "csv_data")
if not csv_data:
csv_data = first(data, "csv_file")
import_mode = first(data, "import_mode", "use_csv_partner")
if csv_data:
reader = csv.DictReader(io.StringIO(csv_data))
imported = 0
skipped = 0
for row in reader:
row_partner_id = None
if import_mode == "single_partner":
row_partner_id = partner_id
else:
row_partner_id = partner_id_from_email(conn, (row.get("partner_email") or "").strip())
if not row_partner_id:
skipped += 1
continue
if not self.partner_in_admin_scope(conn, user, row_partner_id):
skipped += 1
continue
sub = conn.execute(
"SELECT * FROM media_subscriptions WHERE partner_id=? AND status='active' ORDER BY id DESC LIMIT 1",
(row_partner_id,),
).fetchone()
if not sub:
skipped += 1
continue
impressions = to_int((row.get("impressions") or "0").strip() or "0", 0)
clicks = to_int((row.get("clicks") or "0").strip() or "0", 0)
views = to_int((row.get("views") or "0").strip() or "0", 0)
leads = to_int((row.get("leads") or "0").strip() or "0", 0)
charged = media_cost_from_impressions(impressions, sub["rate_per_1000"])
if charged > sub["budget_remaining"]:
skipped += 1
add_notification(
conn,
row_partner_id,
"Media Row Skipped",
f"CSV row skipped due to low budget for impressions {impressions}.",
)
continue
conn.execute(
"""INSERT INTO media_performance_daily
(partner_id, media_subscription_id, ad_date, platform, impressions, clicks, views, leads, charged_amount, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
row_partner_id,
sub["id"],
(row.get("ad_date") or datetime.utcnow().date().isoformat()).strip(),
(row.get("platform") or "Facebook").strip(),
impressions,
clicks,
views,
leads,
charged,
now_iso(),
),
)
conn.execute(
"UPDATE media_subscriptions SET budget_used=budget_used+?, budget_remaining=budget_remaining-? WHERE id=?",
(charged, charged, sub["id"]),
)
imported += 1
add_notification(
conn,
user["id"],
"Media Import Completed",
f"Imported rows: {imported}. Skipped rows: {skipped}.",
)
conn.commit()
conn.close()
self.redirect("/admin/media")
def admin_subscriptions(self, user):
if not self.require_admin_perm(user, "subscriptions", "view"):
return
cfg = vertical_config()
parsed = urllib.parse.urlparse(self.path)
q = urllib.parse.parse_qs(parsed.query)
filter_mode = first(q, "filter", "")
conn = db_connect()
try:
ensure_runtime_migrations(conn)
conn.commit()
except Exception:
pass
my_role = self.admin_role_of(conn, user["id"])
perms = self.admin_ui_perms(conn, user, "subscriptions")
expire_subscriptions(conn)
today = datetime.utcnow().date().isoformat()
p_cond, p_params = self.admin_partner_scope_condition(conn, user, "p")
partners = conn.execute(
"SELECT u.id, u.name, p.company_name FROM users u JOIN partners p ON p.user_id=u.id WHERE 1=1 "
+ p_cond
+ " ORDER BY u.name",
p_params,
).fetchall()
packages = conn.execute("SELECT id, name, monthly_credits FROM packages ORDER BY name").fetchall()
pkg_options = "".join([f"" for x in packages])
if filter_mode == "inactive":
sub_rows = conn.execute(
"""SELECT s.id, s.partner_id, s.package_id, u.name, u.email, p.company_name, p2.name AS package_name, s.start_date, s.end_date, s.status,
s.credits_monthly, s.credits_used, s.credits_remaining, s.finalized_amount,
COALESCE(s.approval_status,'approved') approval_status, COALESCE(s.payment_proof_url,'') payment_proof_url,
COALESCE(s.billing_cycle,'yearly') billing_cycle
FROM subscriptions s
JOIN users u ON u.id=s.partner_id
JOIN partners p ON p.user_id=u.id
JOIN packages p2 ON p2.id=s.package_id
WHERE (s.status='expired' OR s.end_date=? THEN 1 ELSE 0 END) active_count,
SUM(CASE WHEN status='expired' OR end_date=? THEN 1 ELSE 0 END), 0)=0
ORDER BY COALESCE(MAX(s.end_date), '0000-00-00') DESC, u.name""",
p_params + [today],
).fetchall()
conn.close()
inactive_rows_html = "".join(
[
f"
{html.escape(r['name'])}
{html.escape(r['company_name'] or '-')}
{html.escape(r['email'] or '-')}
{html.escape(r['phone'] or '-')}
{html.escape(r['last_end_date'] or 'No subscription')}
"
for r in latest_inactive_partners
]
)
trs = ""
for r in sub_rows:
row_pkg_opts = "".join([f"" for x in packages])
action_html = ""
if perms["edit"]:
action_html += f"""
"""
if perms["delete"]:
action_html += f"""
"""
approve_html = ""
if perms["edit"] and (r["approval_status"] == "pending"):
approve_html = f"""
"""
proof_cell = f"View" if r["payment_proof_url"] else "No"
trs += (
f"
{r['id']}
{html.escape(r['name'])} {html.escape(r['company_name'] or '-')}
"""
self.send_html(page(f"{cfg['lead_plural']} Dashboard", body, user))
def admin_leads_unused(self, user):
if not self.require_admin_perm(user, "leads", "view"):
return
cfg = vertical_config()
parsed = urllib.parse.urlparse(self.path)
q = urllib.parse.parse_qs(parsed.query)
source_filter = (q.get("source_filter", ["all"])[0] or "all").strip().lower()
if source_filter not in {"all", "system_pool", "media_ads"}:
source_filter = "all"
page_no = to_int((q.get("page", ["1"])[0] or "1"), 1)
if page_no < 1:
page_no = 1
per_page = 30
offset = (page_no - 1) * per_page
conn = db_connect()
perms = self.admin_ui_perms(conn, user, "leads")
p_cond, p_params = self.admin_partner_scope_condition(conn, user, "p")
partners = conn.execute(
"""SELECT u.id, u.name, u.email
FROM users u JOIN partners p ON p.user_id=u.id
WHERE u.role='partner' AND u.status='active' """
+ p_cond
+ " ORDER BY u.name",
p_params,
).fetchall()
unused_total = conn.execute(
"""SELECT COUNT(*) c
FROM leads l
LEFT JOIN lead_assignments la ON la.lead_id=l.id
WHERE la.id IS NULL
AND (?='all' OR l.lead_bucket=?)""",
(source_filter, source_filter),
).fetchone()["c"]
unused_rows = conn.execute(
"""SELECT l.*
FROM leads l
LEFT JOIN lead_assignments la ON la.lead_id=l.id
WHERE la.id IS NULL
AND (?='all' OR l.lead_bucket=?)
ORDER BY l.id DESC
LIMIT ? OFFSET ?""",
(source_filter, source_filter, per_page, offset),
).fetchall()
conn.close()
partner_opts = "".join([f"" for p in partners])
total_pages = max(1, (unused_total + per_page - 1) // per_page)
prev_link = f"/admin/leads/unused?source_filter={source_filter}&page={page_no-1}"
next_link = f"/admin/leads/unused?source_filter={source_filter}&page={page_no+1}"
unused_trs_parts = []
for r in unused_rows:
action_cell = "-"
if perms["edit"]:
action_cell = (
""
)
unused_trs_parts.append(
f"
Total Unused {html.escape(cfg['lead_plural'])}: {unused_total}
{f"" if perms['edit'] else '')}
Page {page_no} / {total_pages}
{'Previous' if page_no > 1 else ''}
{'Next' if page_no < total_pages else ''}
"""
self.send_html(page(f"Unused {cfg['lead_plural']} Pool", body, user))
def admin_leads_create(self, user):
if not self.require_admin_perm(user, "leads", "view"):
return
cfg = vertical_config()
conn = db_connect()
perms = self.admin_ui_perms(conn, user, "leads")
if not perms["add"]:
conn.close()
return self.send_html(page("Forbidden", "
Permission denied.
", user), 403)
p_cond, p_params = self.admin_partner_scope_condition(conn, user, "p")
partners = conn.execute(
"""SELECT u.id, u.name, u.email
FROM users u JOIN partners p ON p.user_id=u.id
WHERE u.role='partner' AND u.status='active' """
+ p_cond
+ " ORDER BY u.name",
p_params,
).fetchall()
conn.close()
partner_opts = "".join([f"" for p in partners])
lead_form = f"""
"""
if is_removals_vertical():
lead_form = f"""
"""
body = f"""
Create {'Enquiry' if is_removals_vertical() else 'Lead'}
"""
self.send_html(page(f"Create {cfg['lead_singular']}", body, user))
def admin_leads_import_page(self, user):
if not self.require_admin_perm(user, "leads", "view"):
return
cfg = vertical_config()
conn = db_connect()
perms = self.admin_ui_perms(conn, user, "leads")
if not perms["add"]:
conn.close()
return self.send_html(page("Forbidden", "
Permission denied.
", user), 403)
p_cond, p_params = self.admin_partner_scope_condition(conn, user, "p")
partners = conn.execute(
"""SELECT u.id, u.name, u.email
FROM users u JOIN partners p ON p.user_id=u.id
WHERE u.role='partner' AND u.status='active' """
+ p_cond
+ " ORDER BY u.name",
p_params,
).fetchall()
conn.close()
partner_opts = "".join([f"" for p in partners])
header_text = "name,phone,email,city,area,budget_min,budget_max,property_type,purpose,timeframe,source,lead_type,lead_bucket,partner_email"
if is_removals_vertical():
header_text = "name,phone,email,from_postcode,to_postcode,move_date,service_type,job_type,property_type,move_size,vehicle_size,floor_number,lift_access,packing_needed,dismantling_needed,storage_needed,loading_help_needed,permit_required,parking_notes,waste_type,access_notes,special_items,budget_range,source,lead_type,lead_bucket,partner_email"
body = f"""
"""
self.send_html(page(f"Bulk Import {cfg['lead_plural']}", body, user))
def admin_leads_list(self, user):
if not self.require_admin_perm(user, "leads", "view"):
return
cfg = vertical_config()
parsed = urllib.parse.urlparse(self.path)
q = urllib.parse.parse_qs(parsed.query)
assignment_filter = (q.get("assignment_filter", ["all"])[0] or "all").strip().lower()
if assignment_filter not in {"all", "assigned", "unassigned"}:
assignment_filter = "all"
source_filter = (q.get("source_filter", ["all"])[0] or "all").strip().lower()
if source_filter not in {"all", "system_pool", "media_ads"}:
source_filter = "all"
conn = db_connect()
perms = self.admin_ui_perms(conn, user, "leads")
p_cond, p_params = self.admin_partner_scope_condition(conn, user, "p")
partners = conn.execute(
"""SELECT u.id, u.name, u.email
FROM users u JOIN partners p ON p.user_id=u.id
WHERE u.role='partner' AND u.status='active' """
+ p_cond
+ " ORDER BY u.name",
p_params,
).fetchall()
where_assigned = ""
if assignment_filter == "assigned":
where_assigned = "HAVING COUNT(la.id) > 0"
elif assignment_filter == "unassigned":
where_assigned = "HAVING COUNT(la.id) = 0"
role_name = self.admin_role_of(conn, user["id"])
scope_where = ""
scope_params = []
if role_name != "super_admin":
scoped_admin_ids = self.admin_scope_admin_ids(conn, user)
ph = ",".join(["?"] * len(scoped_admin_ids))
scope_where = f" AND (COUNT(la.id)=0 OR SUM(CASE WHEN p.created_by_admin_id IN ({ph}) THEN 1 ELSE 0 END) > 0)"
scope_params = scoped_admin_ids
rows = conn.execute(
f"""SELECT l.*, u.name AS target_partner_name, COUNT(la.id) AS assigned_count
FROM leads l
LEFT JOIN users u ON u.id=l.target_partner_id
LEFT JOIN lead_assignments la ON la.lead_id=l.id
LEFT JOIN partners p ON p.user_id=la.partner_id
{"WHERE l.lead_bucket='system_pool'" if source_filter=='system_pool' else ("WHERE l.lead_bucket='media_ads'" if source_filter=='media_ads' else "")}
GROUP BY l.id
{(where_assigned + scope_where) if where_assigned else ("HAVING 1=1" + scope_where if scope_where else "")}
ORDER BY l.id DESC"""
,
scope_params,
).fetchall()
conn.close()
partner_opts = "".join([f"" for p in partners])
trs_parts = []
for r in rows:
location_text = f"{html.escape(r['city'])}/{html.escape(r['area'])}"
type_text = html.escape(r["property_type"])
if is_removals_vertical():
location_text = f"{html.escape(r['city'])} to {html.escape(r['area'])}"
type_text = html.escape(format_service_category(r["service_category"] or r["property_type"]))
action_cell = "-"
if perms["edit"]:
action_cell = (
""
)
trs_parts.append(
f"
"""
self.send_html(page(f"{cfg['lead_plural']} List", body, user))
def post_admin_leads(self, user):
if not self.require_admin_perm(user, "leads", "add"):
return
data = parse_post_data(self)
lead_type = first(data, "lead_type", "platform")
if lead_type not in {"platform", "partner_ad"}:
lead_type = "platform"
target_partner_id = first(data, "target_partner_id")
target_partner_id = to_int(target_partner_id, 0) if target_partner_id else None
lead_bucket = normalize_lead_bucket(first(data, "lead_bucket", "system_pool"))
if lead_type == "platform":
target_partner_id = None
lead_bucket = "system_pool"
conn = db_connect()
if target_partner_id and not self.partner_in_admin_scope(conn, user, target_partner_id):
target_partner_id = None
lead_type = "platform"
lead_bucket = "system_pool"
if is_removals_vertical():
budget_min, budget_max = budget_range_to_values(first(data, "budget_range"))
from_postcode = normalize_postcode_area(first(data, "from_postcode"))
to_postcode = normalize_postcode_area(first(data, "to_postcode"))
service_type = normalize_service_category(first(data, "service_type", "home_removals"))
cur = conn.execute(
"""INSERT INTO leads
(name, phone, email, city, area, budget_min, budget_max, property_type, purpose, timeframe, source, lead_type, lead_bucket, target_partner_id, verified_by_admin, created_at, vertical_type, service_category)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?, ?)""",
(
first(data, "name"),
first(data, "phone"),
first(data, "email"),
from_postcode,
to_postcode,
budget_min,
budget_max,
first(data, "property_type", "House"),
"buy",
first(data, "move_date", "Flexible"),
first(data, "source", "Manual"),
lead_type,
lead_bucket,
target_partner_id,
now_iso(),
VERTICAL_TYPE,
service_type,
),
)
lead_id = cur.lastrowid
save_move_requirements(conn, lead_id, data)
if lead_type == "partner_ad" and target_partner_id:
self.assign_specific_partner(conn, lead_id, target_partner_id, consume_credit=False)
conn.commit()
conn.close()
return self.redirect("/admin/leads")
cur = conn.execute(
"""INSERT INTO leads
(name, phone, email, city, area, budget_min, budget_max, property_type, purpose, timeframe, source, lead_type, lead_bucket, target_partner_id, verified_by_admin, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?)""",
(
first(data, "name"),
first(data, "phone"),
first(data, "email"),
first(data, "city"),
first(data, "area"),
to_int(first(data, "budget_min", "0") or "0", 0),
to_int(first(data, "budget_max", "0") or "0", 0),
first(data, "property_type", "Plot"),
first(data, "purpose", "buy"),
first(data, "timeframe", "30 days"),
first(data, "source", "Manual"),
lead_type,
lead_bucket,
target_partner_id,
now_iso(),
),
)
lead_id = cur.lastrowid
if lead_type == "partner_ad" and target_partner_id:
self.assign_specific_partner(conn, lead_id, target_partner_id, consume_credit=False)
conn.commit()
conn.close()
self.redirect("/admin/leads")
def post_admin_leads_import(self, user):
if not self.require_admin_perm(user, "leads", "add"):
return
data = parse_post_data(self)
csv_data = first(data, "csv_data")
if not csv_data:
csv_data = first(data, "csv_file")
if not csv_data:
return self.redirect("/admin/leads?imported=0&skipped=0&msg=no_csv_content_received")
import_mode = first(data, "import_mode", "store_pool")
specific_partner_id = first(data, "specific_partner_id")
specific_partner_id = to_int(specific_partner_id, 0) if specific_partner_id else None
conn = db_connect()
if specific_partner_id and not self.partner_in_admin_scope(conn, user, specific_partner_id):
specific_partner_id = None
existing_rows = conn.execute("SELECT phone FROM leads").fetchall()
existing_keys = set()
for r in existing_rows:
key = normalize_phone(r["phone"])
if key:
existing_keys.add(key)
seen_keys = set()
imported = 0
skipped = 0
skipped_missing = 0
skipped_duplicate = 0
skipped_error = 0
reader = csv.DictReader(io.StringIO(csv_data))
for row in reader:
row_norm = {}
for k, v in row.items():
if k is None:
continue
nk = str(k).replace("\ufeff", "").strip().lower()
row_norm[nk] = (v or "").strip()
if not row_norm.get("name") or not row_norm.get("phone"):
skipped += 1
skipped_missing += 1
continue
phone_norm = normalize_phone(row_norm.get("phone", ""))
key = phone_norm
if not key:
skipped += 1
skipped_missing += 1
continue
if key in existing_keys or key in seen_keys:
skipped += 1
skipped_duplicate += 1
continue
seen_keys.add(key)
lead_type = (row_norm.get("lead_type") or "platform").strip().lower()
lead_bucket = normalize_lead_bucket(row_norm.get("lead_bucket") or "system_pool")
target_partner_id = partner_id_from_email(conn, row_norm.get("partner_email", ""))
if target_partner_id and not self.partner_in_admin_scope(conn, user, target_partner_id):
target_partner_id = None
if import_mode == "store_pool":
lead_type = "platform"
lead_bucket = "system_pool"
target_partner_id = None
if import_mode in {"specific_partner", "specific_partner_media"} and specific_partner_id:
lead_type = "partner_ad"
lead_bucket = "media_ads" if import_mode == "specific_partner_media" else "system_pool"
target_partner_id = specific_partner_id
if lead_type not in {"platform", "partner_ad"}:
lead_type = "platform"
if lead_type == "platform":
lead_bucket = "system_pool"
target_partner_id = None
if lead_bucket not in {"system_pool", "media_ads"}:
lead_bucket = "media_ads" if lead_type == "partner_ad" else "system_pool"
try:
if is_removals_vertical():
budget_min, budget_max = budget_range_to_values(row_norm.get("budget_range"))
cur = conn.execute(
"""INSERT INTO leads
(name, phone, email, city, area, budget_min, budget_max, property_type, purpose, timeframe, source, lead_type, lead_bucket, target_partner_id, verified_by_admin, created_at, vertical_type, service_category)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?, ?)""",
(
row_norm.get("name", ""),
row_norm.get("phone", ""),
row_norm.get("email", ""),
normalize_postcode_area(row_norm.get("from_postcode")) or "UK",
normalize_postcode_area(row_norm.get("to_postcode")) or "UK",
budget_min,
budget_max,
(row_norm.get("property_type") or "House"),
"buy",
(row_norm.get("move_date") or "Flexible"),
(row_norm.get("source") or "Bulk Import"),
lead_type,
lead_bucket,
target_partner_id,
now_iso(),
VERTICAL_TYPE,
normalize_service_category(row_norm.get("service_type")),
),
)
else:
cur = conn.execute(
"""INSERT INTO leads
(name, phone, email, city, area, budget_min, budget_max, property_type, purpose, timeframe, source, lead_type, lead_bucket, target_partner_id, verified_by_admin, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?)""",
(
row_norm.get("name", ""),
row_norm.get("phone", ""),
row_norm.get("email", ""),
(row_norm.get("city") or "Unknown"),
(row_norm.get("area") or "Unknown"),
to_int(row_norm.get("budget_min", "0"), 0),
to_int(row_norm.get("budget_max", "0"), 0),
(row_norm.get("property_type") or "Plot"),
normalize_purpose(row_norm.get("purpose")),
(row_norm.get("timeframe") or "30 days"),
(row_norm.get("source") or "Bulk Import"),
lead_type,
lead_bucket,
target_partner_id,
now_iso(),
),
)
lead_id = cur.lastrowid
if is_removals_vertical():
save_move_requirements(conn, lead_id, row_norm)
if lead_type == "partner_ad" and target_partner_id:
self.assign_specific_partner(conn, lead_id, target_partner_id, consume_credit=False)
imported += 1
except sqlite3.Error:
skipped += 1
skipped_error += 1
continue
add_notification(
conn,
user["id"],
"Lead Import Completed",
f"Imported: {imported}. Skipped: {skipped} (duplicate: {skipped_duplicate}, missing: {skipped_missing}, error: {skipped_error}).",
)
conn.commit()
conn.close()
msg = urllib.parse.quote_plus(
f"duplicate:{skipped_duplicate}, missing:{skipped_missing}, error:{skipped_error}"
)
self.redirect(f"/admin/leads?imported={imported}&skipped={skipped}&msg={msg}")
def auto_assign_partner_leads(self, conn, partner_id, sub=None):
if not sub:
sub = active_subscription(conn, partner_id)
if not sub:
return 0
count_row = conn.execute(
"""SELECT COUNT(*) c FROM lead_assignments
WHERE partner_id=? AND assigned_at >= ? AND assigned_at <= ?""",
(partner_id, sub["start_date"], sub["end_date"]),
).fetchone()
assigned_count = count_row["c"] if count_row else 0
needed = max(0, (sub["credits_monthly"] or 0) - assigned_count)
if needed <= 0:
return 0
if is_removals_vertical():
partner_row = conn.execute(
"SELECT COALESCE(service_categories_json,'[]') service_categories_json, COALESCE(coverage_postcodes_json,'[]') coverage_postcodes_json FROM partners WHERE user_id=?",
(partner_id,),
).fetchone()
categories = json.loads(partner_row["service_categories_json"] or "[]") if partner_row else []
postcodes = json.loads(partner_row["coverage_postcodes_json"] or "[]") if partner_row else []
if not categories and not postcodes:
return 0
rows = conn.execute(
"""SELECT l.id
FROM leads l
LEFT JOIN lead_assignments la ON la.lead_id=l.id AND la.partner_id=?
WHERE l.vertical_type=? AND l.lead_type='platform' AND l.lead_bucket='system_pool' AND la.id IS NULL
ORDER BY COALESCE(l.lead_score,0) DESC, l.id ASC LIMIT 100""",
(partner_id, VERTICAL_TYPE),
).fetchall()
assigned = 0
for r in rows:
candidates = moving_match_candidates(conn, r["id"])
if any(pid == partner_id for pid, _, _ in candidates):
conn.execute(
"""INSERT INTO lead_assignments
(lead_id, partner_id, assigned_at, status, viewed_at, contacted_at)
VALUES (?, ?, ?, 'assigned', NULL, NULL)""",
(r["id"], partner_id, now_iso()),
)
assigned += 1
if assigned >= needed:
break
return assigned
city_row = conn.execute("SELECT city FROM partners WHERE user_id=?", (partner_id,)).fetchone()
city = city_row["city"] if city_row else None
if city:
lead_rows = conn.execute(
"""SELECT l.id FROM leads l
LEFT JOIN lead_assignments la ON la.lead_id=l.id AND la.partner_id=?
WHERE l.lead_type='platform' AND l.lead_bucket='system_pool' AND la.id IS NULL AND l.city=?
ORDER BY l.id ASC LIMIT ?""",
(partner_id, city, needed),
).fetchall()
else:
lead_rows = conn.execute(
"""SELECT l.id FROM leads l
LEFT JOIN lead_assignments la ON la.lead_id=l.id AND la.partner_id=?
WHERE l.lead_type='platform' AND l.lead_bucket='system_pool' AND la.id IS NULL
ORDER BY l.id ASC LIMIT ?""",
(partner_id, needed),
).fetchall()
assigned = 0
for r in lead_rows:
conn.execute(
"""INSERT INTO lead_assignments
(lead_id, partner_id, assigned_at, status, viewed_at, contacted_at)
VALUES (?, ?, ?, 'assigned', NULL, NULL)""",
(r["id"], partner_id, now_iso()),
)
assigned += 1
return assigned
def assign_lead(self, conn, lead_id, max_partners=3):
expire_subscriptions(conn)
today = datetime.utcnow().date().isoformat()
lead = conn.execute("SELECT * FROM leads WHERE id=?", (lead_id,)).fetchone()
if not lead:
return 0
required_credits = lead_bucket_view_cost(lead["lead_bucket"])
if lead["lead_type"] == "partner_ad" and lead["target_partner_id"]:
return self.assign_specific_partner(conn, lead_id, lead["target_partner_id"], consume_credit=False)
if is_removals_vertical() and (lead["vertical_type"] or "") == VERTICAL_TYPE:
candidates = moving_match_candidates(conn, lead_id, required_credits=required_credits)
assigned = 0
for partner_id, score, reason in candidates[:max_partners]:
exists = conn.execute("SELECT id FROM lead_assignments WHERE lead_id=? AND partner_id=?", (lead_id, partner_id)).fetchone()
if exists:
continue
conn.execute(
"""INSERT INTO lead_assignments
(lead_id, partner_id, assigned_at, status, viewed_at, contacted_at)
VALUES (?, ?, ?, 'assigned', NULL, NULL)""",
(lead_id, partner_id, now_iso()),
)
conn.execute(
"""INSERT INTO moving_lead_matches
(lead_id, partner_id, match_score, reason_summary, status, created_at)
VALUES (?, ?, ?, ?, 'assigned', ?)""",
(lead_id, partner_id, score, reason, now_iso()),
)
assigned += 1
return assigned
already = conn.execute("SELECT COUNT(*) c FROM lead_assignments WHERE lead_id=?", (lead_id,)).fetchone()["c"]
slots = max(0, max_partners - already)
if slots == 0:
return 0
candidates = conn.execute(
"""SELECT s.partner_id, s.id AS subscription_id, s.credits_remaining, p.city
FROM subscriptions s
JOIN partners p ON p.user_id=s.partner_id
WHERE s.status='active' AND s.credits_remaining >= ? AND p.city = ? AND s.start_date <= ? AND s.end_date >= ?
ORDER BY s.credits_remaining DESC, s.id ASC""",
(required_credits, lead["city"], today, today),
).fetchall()
assigned = 0
for c in candidates:
if assigned >= slots:
break
exists = conn.execute("SELECT id FROM lead_assignments WHERE lead_id=? AND partner_id=?", (lead_id, c["partner_id"])).fetchone()
if exists:
continue
conn.execute(
"""INSERT INTO lead_assignments
(lead_id, partner_id, assigned_at, status, viewed_at, contacted_at)
VALUES (?, ?, ?, 'assigned', NULL, NULL)""",
(lead_id, c["partner_id"], now_iso()),
)
assigned += 1
return assigned
def assign_specific_partner(self, conn, lead_id, partner_id, consume_credit=False):
lead = conn.execute("SELECT * FROM leads WHERE id=?", (lead_id,)).fetchone()
if not lead:
return 0
required_credits = lead_bucket_view_cost(lead["lead_bucket"])
sub = active_subscription(conn, partner_id)
if not sub:
return 0
if sub["credits_remaining"] < required_credits:
return 0
exists = conn.execute("SELECT id FROM lead_assignments WHERE lead_id=? AND partner_id=?", (lead_id, partner_id)).fetchone()
if exists:
return 0
conn.execute(
"""INSERT INTO lead_assignments
(lead_id, partner_id, assigned_at, status, viewed_at, contacted_at)
VALUES (?, ?, ?, 'assigned', NULL, NULL)""",
(lead_id, partner_id, now_iso()),
)
return 1
def post_admin_assign(self, user):
if not self.require_admin_perm(user, "leads", "edit"):
return
data = parse_post_data(self)
lead_id = int(first(data, "lead_id", "0") or "0")
partner_id = to_int(first(data, "partner_id", "0"), 0)
conn = db_connect()
if partner_id > 0 and not self.partner_in_admin_scope(conn, user, partner_id):
conn.close()
return self.redirect("/admin/leads")
if partner_id > 0:
self.assign_specific_partner(conn, lead_id, partner_id, consume_credit=False)
else:
self.assign_lead(conn, lead_id, 3)
conn.commit()
conn.close()
self.redirect("/admin/leads")
def post_admin_assign_bulk(self, user):
if not self.require_admin_perm(user, "leads", "edit"):
return
data = parse_post_data(self)
partner_id = to_int(first(data, "partner_id", "0"), 0)
raw_ids = data.get("lead_ids", [])
lead_ids = []
for v in raw_ids:
lead_id = to_int(v, 0)
if lead_id > 0:
lead_ids.append(lead_id)
if not lead_ids:
return self.redirect("/admin/leads?msg=no_leads_selected")
conn = db_connect()
if partner_id > 0 and not self.partner_in_admin_scope(conn, user, partner_id):
conn.close()
return self.redirect("/admin/leads")
processed = 0
assigned = 0
for lead_id in lead_ids:
processed += 1
if partner_id > 0:
assigned += self.assign_specific_partner(conn, lead_id, partner_id, consume_credit=False)
else:
assigned += self.assign_lead(conn, lead_id, 3)
conn.commit()
conn.close()
self.redirect(f"/admin/leads?msg=bulk_auto_assign_done_processed:{processed}_assigned:{assigned}")
def admin_invalid_reports(self, user):
if not self.require_admin_perm(user, "invalid_reports", "view"):
return
cfg = vertical_config()
conn = db_connect()
p_cond, p_params = self.admin_partner_scope_condition(conn, user, "p")
rows = conn.execute(
"""SELECT ir.*, u.name AS partner_name, l.city, l.area, l.name AS lead_name
FROM invalid_reports ir
JOIN users u ON u.id=ir.partner_id
JOIN partners p ON p.user_id=ir.partner_id
JOIN lead_assignments la ON la.id=ir.assignment_id
JOIN leads l ON l.id=la.lead_id
WHERE 1=1 """
+ p_cond
+ " ORDER BY ir.id DESC",
p_params,
).fetchall()
conn.close()
trs = ""
for r in rows:
actions = ""
if r["status"] == "pending":
actions = f"""
"""
trs += f"
{r['id']}
{html.escape(r['partner_name'])}
{html.escape(r['lead_name'])}
{html.escape(r['city'])}/{html.escape(r['area'])}
{html.escape(r['reason'])}
{html.escape(r['status'])}
{actions}
"
body = f"""
Invalid Reports
ID
{html.escape(cfg['operator_singular'])}
{html.escape(cfg['lead_singular'])}
Location
Reason
Status
Action
{trs or '
No reports.
'}
"""
self.send_html(page("Invalid Reports", body, user))
def post_admin_invalid_review(self, user):
if not self.require_admin_perm(user, "invalid_reports", "edit"):
return
data = parse_post_data(self)
report_id = int(first(data, "report_id", "0") or "0")
decision = first(data, "decision")
notes = first(data, "admin_notes")
if decision not in {"approved", "rejected"}:
return self.redirect("/admin/invalid-reports")
conn = db_connect()
report = conn.execute("SELECT * FROM invalid_reports WHERE id=?", (report_id,)).fetchone()
if not report or report["status"] != "pending":
conn.close()
return self.redirect("/admin/invalid-reports")
if not self.partner_in_admin_scope(conn, user, report["partner_id"]):
conn.close()
return self.redirect("/admin/invalid-reports")
conn.execute("UPDATE invalid_reports SET status=?, admin_notes=? WHERE id=?", (decision, notes, report_id))
assignment = conn.execute(
"""SELECT la.*, l.city, l.area, COALESCE(l.lead_bucket,'system_pool') lead_bucket FROM lead_assignments la
JOIN leads l ON l.id=la.lead_id
WHERE la.id=?""",
(report["assignment_id"],),
).fetchone()
if decision == "approved":
conn.execute("UPDATE lead_assignments SET status='invalid_approved' WHERE id=?", (report["assignment_id"],))
conn.execute(
"INSERT INTO lead_status_history (assignment_id, status, notes, created_at) VALUES (?, 'invalid_approved', ?, ?)",
(report["assignment_id"], notes or "Invalid approved by admin", now_iso()),
)
if report["reason"] in ALLOWED_INVALID_REASONS:
refund_amount = lead_bucket_view_cost(assignment["lead_bucket"] if assignment else "system_pool")
sub = conn.execute("SELECT id FROM subscriptions WHERE partner_id=? AND status='active' ORDER BY id DESC LIMIT 1", (report["partner_id"],)).fetchone()
if sub:
conn.execute(
"UPDATE subscriptions SET credits_remaining = credits_remaining + ?, credits_used = CASE WHEN credits_used >= ? THEN credits_used - ? ELSE 0 END WHERE id=?",
(refund_amount, refund_amount, refund_amount, sub["id"]),
)
conn.execute(
"INSERT INTO credit_adjustments (partner_id, amount, reason, created_at) VALUES (?, ?, ?, ?)",
(
report["partner_id"],
refund_amount,
f"{lead_bucket_label(assignment['lead_bucket'] if assignment else 'system_pool')} invalid lead approved credit",
now_iso(),
),
)
if assignment:
replacement = conn.execute(
"""SELECT l.id FROM leads l
LEFT JOIN lead_assignments la ON la.lead_id=l.id AND la.partner_id=?
WHERE l.city=? AND l.area=? AND la.id IS NULL
AND COALESCE(l.lead_bucket,'system_pool')=?
ORDER BY l.id ASC LIMIT 1""",
(report["partner_id"], assignment["city"], assignment["area"], normalize_lead_bucket(assignment["lead_bucket"])),
).fetchone()
if replacement:
self.assign_specific_partner(conn, replacement["id"], report["partner_id"], consume_credit=False)
add_notification(
conn,
report["partner_id"],
"Invalid Lead Approved",
f"Report #{report_id} approved. Credit reversed and replacement lead assigned.",
)
else:
add_notification(
conn,
report["partner_id"],
"Invalid Lead Approved",
f"Report #{report_id} approved. Credit reversed. Replacement lead unavailable right now.",
)
else:
add_notification(
conn,
report["partner_id"],
"Invalid Lead Approved",
f"Report #{report_id} approved, but reason is not eligible for replacement credit.",
)
else:
if assignment:
conn.execute("UPDATE lead_assignments SET status='assigned' WHERE id=?", (assignment["id"],))
conn.execute(
"INSERT INTO lead_status_history (assignment_id, status, notes, created_at) VALUES (?, 'invalid_rejected', ?, ?)",
(report["assignment_id"], notes or "Invalid report rejected by admin", now_iso()),
)
add_notification(
conn,
report["partner_id"],
"Invalid Lead Rejected",
f"Report #{report_id} was rejected. Reason: {notes or 'No additional notes'}",
)
conn.commit()
conn.close()
self.redirect("/admin/invalid-reports")
def admin_tickets(self, user):
if not self.require_admin_perm(user, "tickets", "view"):
return
cfg = vertical_config()
conn = db_connect()
perms = self.admin_ui_perms(conn, user, "tickets")
p_cond, p_params = self.admin_partner_scope_condition(conn, user, "p")
rows = conn.execute(
"""SELECT t.*, u.name AS partner_name
FROM tickets t
JOIN users u ON u.id=t.partner_id
JOIN partners p ON p.user_id=t.partner_id
WHERE 1=1 """
+ p_cond
+ " ORDER BY t.id DESC",
p_params,
).fetchall()
msgs = conn.execute(
"""SELECT tm.*, t.partner_id FROM ticket_messages tm
JOIN tickets t ON t.id=tm.ticket_id
JOIN partners p ON p.user_id=t.partner_id
WHERE 1=1 """
+ p_cond
+ " ORDER BY tm.id DESC LIMIT 30",
p_params,
).fetchall()
conn.close()
trs_parts = []
for r in rows:
reply_cell = "-"
if perms["edit"]:
reply_cell = (
""
)
trs_parts.append(
f"
{r['id']}
{html.escape(r['partner_name'])}
"
f"
{html.escape(r['subject'])}
{html.escape(r['priority'])}
"
f"
{html.escape(r['status'])}
{reply_cell}
"
)
trs = "".join(trs_parts)
mtrs = "".join([f"
{m['ticket_id']}
{html.escape(m['sender_role'])}
{html.escape(m['message'])}
{html.escape(m['created_at'])}
" for m in msgs])
body = f"""
Tickets
ID
{html.escape(cfg['operator_singular'])}
Subject
Priority
Status
Reply
{trs}
Recent Messages
Ticket
Sender
Message
Date
{mtrs or '
No messages.
'}
"""
self.send_html(page("Support Tickets", body, user))
def post_admin_ticket_reply(self, user):
if not self.require_admin_perm(user, "tickets", "edit"):
return
data = parse_post_data(self)
ticket_id = int(first(data, "ticket_id", "0") or "0")
message = first(data, "message")
if not message:
return self.redirect("/admin/tickets")
conn = db_connect()
ticket = conn.execute("SELECT partner_id FROM tickets WHERE id=?", (ticket_id,)).fetchone()
if not ticket or not self.partner_in_admin_scope(conn, user, ticket["partner_id"]):
conn.close()
return self.redirect("/admin/tickets")
conn.execute(
"INSERT INTO ticket_messages (ticket_id, sender_role, message, attachments, created_at) VALUES (?, 'admin', ?, '', ?)",
(ticket_id, message, now_iso()),
)
conn.execute("UPDATE tickets SET status='in_progress' WHERE id=?", (ticket_id,))
conn.commit()
conn.close()
self.redirect("/admin/tickets")
def post_admin_todo_add(self, user):
task_text = first(parse_post_data(self), "task_text")
if not task_text:
return self.redirect("/admin/dashboard")
conn = db_connect()
conn.execute(
"INSERT INTO admin_todos (user_id, task_text, status, created_at, completed_at) VALUES (?, ?, 'pending', ?, NULL)",
(user["id"], task_text, now_iso()),
)
conn.commit()
conn.close()
self.redirect("/admin/dashboard")
def post_admin_todo_complete(self, user):
todo_id = to_int(first(parse_post_data(self), "todo_id", "0"), 0)
conn = db_connect()
conn.execute(
"UPDATE admin_todos SET status='completed', completed_at=? WHERE id=? AND user_id=?",
(now_iso(), todo_id, user["id"]),
)
conn.commit()
conn.close()
self.redirect("/admin/dashboard")
class _WSGIAdapter(AppHandler):
"""Adapter to run the same handler logic under WSGI (cPanel Passenger)."""
def __init__(self, environ):
self.environ = environ
self.script_name = (environ.get("SCRIPT_NAME", "") or "").rstrip("/")
self.command = environ.get("REQUEST_METHOD", "GET").upper()
path = environ.get("PATH_INFO", "/") or "/"
query = environ.get("QUERY_STRING", "") or ""
self.path = f"{path}?{query}" if query else path
self.headers = {}
for key, value in environ.items():
if key.startswith("HTTP_"):
hname = key[5:].replace("_", "-").title()
self.headers[hname] = value
if environ.get("CONTENT_TYPE"):
self.headers["Content-Type"] = environ.get("CONTENT_TYPE")
if environ.get("CONTENT_LENGTH"):
self.headers["Content-Length"] = environ.get("CONTENT_LENGTH")
body = b""
if self.command in {"POST", "PUT", "PATCH"}:
clen = int(environ.get("CONTENT_LENGTH") or "0")
if clen > 0:
body = environ["wsgi.input"].read(clen)
self.rfile = io.BytesIO(body)
self.wfile = io.BytesIO()
self._status_code = 200
self._response_headers = []
def send_response(self, code, message=None):
self._status_code = int(code)
def send_header(self, keyword, value):
self._response_headers.append((keyword, value))
def end_headers(self):
return
def wsgi_application(environ, start_response):
ensure_initialized()
handler = _WSGIAdapter(environ)
if handler.command == "GET":
handler.do_GET()
elif handler.command == "POST":
handler.do_POST()
else:
handler.send_html("Method Not Allowed", 405)
body = handler.wfile.getvalue()
headers = list(handler._response_headers)
if not any(k.lower() == "content-length" for k, _ in headers):
headers.append(("Content-Length", str(len(body))))
status_line = f"{handler._status_code} {HTTPStatus(handler._status_code).phrase if handler._status_code in HTTPStatus._value2member_map_ else 'OK'}"
start_response(status_line, headers)
return [body]
def run():
ensure_initialized()
host = os.environ.get("VANLOCALUK_HOST", "127.0.0.1")
port = int(os.environ.get("VANLOCALUK_PORT", "8000"))
server = HTTPServer((host, port), AppHandler)
print(f"Running on http://{host}:{port}")
server.serve_forever()
if __name__ == "__main__":
if len(sys.argv) > 1 and sys.argv[1] == "backup":
b = backup_db()
print(f"Backup created: {b}" if b else "No DB found to backup.")
sys.exit(0)
if len(sys.argv) > 1 and sys.argv[1] == "migrate":
ensure_initialized()
print("Database migration/init completed safely.")
sys.exit(0)
try:
run()
except KeyboardInterrupt:
print("\nStopped.")
sys.exit(0)