| """ |
| Embeddable CTA forms: create in Settings, embed on WordPress, submissions → Leads. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| import re |
| import uuid |
| from datetime import datetime |
| from typing import Any, Dict, List, Optional |
|
|
| from fastapi import APIRouter, Depends, HTTPException, Request |
| from fastapi.responses import JSONResponse, PlainTextResponse, Response |
| from sqlalchemy.orm import Session |
|
|
| from .database import CrmLead, CtaForm, db_commit_with_retry, get_db |
| from .crm_lead_contacts import link_lead_to_contact |
| from .models import CtaFormCreateRequest, CtaFormPatchRequest, CtaFormSubmitRequest |
| from .tenant_deps import TenantContext, get_tenant_context |
|
|
| router = APIRouter(tags=["cta-forms"]) |
|
|
| CTA_LEAD_SOURCE = "cta_form" |
| CTA_LEAD_ID_PREFIX = "cta-" |
|
|
| DEFAULT_FIELDS: List[dict] = [ |
| {"key": "name", "label": "Name", "type": "text", "required": True, "map_to": "name"}, |
| {"key": "email", "label": "Email", "type": "email", "required": True, "map_to": "email"}, |
| { |
| "key": "company_name", |
| "label": "Company Name", |
| "type": "text", |
| "required": False, |
| "map_to": "company_name", |
| }, |
| ] |
|
|
| _CORS_HEADERS = { |
| "Access-Control-Allow-Origin": "*", |
| "Access-Control-Allow-Methods": "GET, POST, OPTIONS", |
| "Access-Control-Allow-Headers": "Content-Type", |
| } |
|
|
|
|
| def _safe_str(val: Any) -> str: |
| if val is None: |
| return "" |
| return str(val).strip() |
|
|
|
|
| def _slug_key(label: str, existing: set[str]) -> str: |
| base = re.sub(r"[^a-z0-9]+", "_", label.lower()).strip("_") or "field" |
| key = base |
| n = 2 |
| while key in existing: |
| key = f"{base}_{n}" |
| n += 1 |
| existing.add(key) |
| return key |
|
|
|
|
| def _normalize_fields(raw: List[Any]) -> List[dict]: |
| out: List[dict] = [] |
| seen: set[str] = set() |
| for item in raw or []: |
| if isinstance(item, dict): |
| label = _safe_str(item.get("label")) or "Field" |
| key = _safe_str(item.get("key")) or _slug_key(label, seen) |
| if key in seen: |
| key = _slug_key(key, seen) |
| else: |
| seen.add(key) |
| ftype = _safe_str(item.get("type")) or "text" |
| if ftype not in ("text", "email", "tel", "textarea"): |
| ftype = "text" |
| map_to = _safe_str(item.get("map_to")) or "custom" |
| if map_to not in ("name", "email", "company_name", "title", "phone", "message", "custom"): |
| map_to = "custom" |
| out.append( |
| { |
| "key": key[:64], |
| "label": label[:120], |
| "type": ftype, |
| "required": bool(item.get("required")), |
| "map_to": map_to, |
| } |
| ) |
| return out or list(DEFAULT_FIELDS) |
|
|
|
|
| def _build_embed_codes( |
| *, |
| origin: str, |
| public_id: str, |
| name: str, |
| ) -> dict: |
| title = (name or "Form").replace('"', """) |
| iframe_style = ( |
| "width:100%;max-width:100%;border:none;border-radius:12px;display:block;min-height:280px;" |
| ) |
| return { |
| "embed_iframe": ( |
| f'<iframe src="{origin}/embed/cta/{public_id}" title="{title}" ' |
| f'style="{iframe_style}" loading="lazy"></iframe>' |
| if origin |
| else "" |
| ), |
| "embed_script": ( |
| f'<div id="ezofis-cta-{public_id}"></div>\n' |
| f'<script src="{origin}/api/embed/cta/{public_id}.js" async></script>' |
| if origin |
| else "" |
| ), |
| } |
|
|
|
|
| def _form_to_dict(row: CtaForm, *, request: Optional[Request] = None) -> dict: |
| origin = "" |
| if request is not None: |
| origin = _public_origin(request) |
| pid = row.public_id |
| codes = _build_embed_codes( |
| origin=origin, |
| public_id=pid, |
| name=row.name or "", |
| ) |
| return { |
| "id": row.id, |
| "public_id": pid, |
| "name": row.name or "", |
| "fields": row.fields if isinstance(row.fields, list) else [], |
| "is_active": bool(row.is_active), |
| "created_at": row.created_at.isoformat() if row.created_at else None, |
| "updated_at": row.updated_at.isoformat() if row.updated_at else None, |
| **codes, |
| } |
|
|
|
|
| def _public_origin(request: Request) -> str: |
| fwd = request.headers.get("x-forwarded-proto") |
| proto = (fwd.split(",")[0].strip() if fwd else request.url.scheme or "https").lower() |
| host = request.headers.get("x-forwarded-host") or request.headers.get("host") or request.url.netloc |
| host = host.split(",")[0].strip() if host else "" |
| if not host: |
| return str(request.base_url).rstrip("/") |
| base = f"{proto}://{host}" |
| if base.startswith("http://") and ".hf.space" in base: |
| base = "https://" + base[7:] |
| return base.rstrip("/") |
|
|
|
|
| def _split_name(full: str) -> tuple[str, str]: |
| parts = _safe_str(full).split(None, 1) |
| if not parts: |
| return "", "" |
| if len(parts) == 1: |
| return parts[0], "" |
| return parts[0], parts[1] |
|
|
|
|
| def _submission_to_lead_fields( |
| form: CtaForm, |
| payload: Dict[str, str], |
| ) -> dict: |
| fields_def = form.fields if isinstance(form.fields, list) else [] |
| mapped: Dict[str, str] = {} |
| custom_lines: List[str] = [] |
|
|
| for fd in fields_def: |
| if not isinstance(fd, dict): |
| continue |
| key = _safe_str(fd.get("key")) |
| label = _safe_str(fd.get("label")) or key |
| val = _safe_str(payload.get(key)) |
| map_to = _safe_str(fd.get("map_to")) or "custom" |
| if map_to == "custom": |
| if val: |
| custom_lines.append(f"{label}: {val}") |
| else: |
| if val: |
| mapped[map_to] = val |
|
|
| first_name, last_name = "", "" |
| if mapped.get("name"): |
| first_name, last_name = _split_name(mapped["name"]) |
|
|
| email = mapped.get("email", "") |
| company = mapped.get("company_name", "") |
| title = mapped.get("title", "") |
| phone = mapped.get("phone", "") |
| message = mapped.get("message", "") |
|
|
| body_lines = [f"Form: {form.name or 'CTA Form'}", "---"] |
| for fd in fields_def: |
| if not isinstance(fd, dict): |
| continue |
| key = _safe_str(fd.get("key")) |
| label = _safe_str(fd.get("label")) or key |
| val = _safe_str(payload.get(key)) |
| if val: |
| body_lines.append(f"{label}: {val}") |
|
|
| body = "\n".join(body_lines).strip() |
| if message and message not in body: |
| body = f"{body}\n\nMessage:\n{message}".strip() |
|
|
| return { |
| "first_name": first_name, |
| "last_name": last_name, |
| "email": email, |
| "company_name": company, |
| "title": title, |
| "phone": phone, |
| "last_reply_subject": f"CTA: {form.name or 'Form submission'}", |
| "last_reply_body": body, |
| "custom_lines": custom_lines, |
| } |
|
|
|
|
| def _create_lead_from_submission( |
| db: Session, |
| form: CtaForm, |
| payload: Dict[str, str], |
| *, |
| page_url: str = "", |
| when: Optional[datetime] = None, |
| ) -> CrmLead: |
| when = when or datetime.utcnow() |
| parsed = _submission_to_lead_fields(form, payload) |
| if not parsed["email"]: |
| raise HTTPException(status_code=400, detail="Email is required") |
|
|
| submission_id = uuid.uuid4().hex[:16] |
| lead_key = f"{CTA_LEAD_ID_PREFIX}{form.public_id}-{submission_id}" |
|
|
| raw: Dict[str, Any] = { |
| "source": CTA_LEAD_SOURCE, |
| "form_id": form.id, |
| "form_public_id": form.public_id, |
| "form_name": form.name, |
| "submission": payload, |
| "page_url": page_url or None, |
| "messages": [ |
| { |
| "direction": "inbound", |
| "channel": "cta_form", |
| "subject": parsed["last_reply_subject"], |
| "body": parsed["last_reply_body"], |
| "at": when.replace(microsecond=0).isoformat() + "Z", |
| } |
| ], |
| } |
|
|
| row = CrmLead( |
| tenant_id=form.tenant_id, |
| smartlead_lead_id=lead_key, |
| campaign_id=form.public_id, |
| campaign_name=form.name or "CTA Form", |
| email=parsed["email"], |
| first_name=parsed["first_name"], |
| last_name=parsed["last_name"], |
| company_name=parsed["company_name"], |
| title=parsed["title"], |
| last_reply_subject=parsed["last_reply_subject"], |
| last_reply_body=parsed["last_reply_body"], |
| last_reply_at=when, |
| crm_status="new_lead", |
| raw_webhook=raw, |
| ) |
| db.add(row) |
| db.flush() |
| link_lead_to_contact(db, row) |
| return row |
|
|
|
|
| def _get_active_form(db: Session, public_id: str) -> CtaForm: |
| row = ( |
| db.query(CtaForm) |
| .filter(CtaForm.public_id == public_id, CtaForm.is_active == 1) |
| .first() |
| ) |
| if not row: |
| raise HTTPException(status_code=404, detail="Form not found") |
| return row |
|
|
|
|
| def _validate_submission(form: CtaForm, body: CtaFormSubmitRequest) -> Dict[str, str]: |
| fields_def = form.fields if isinstance(form.fields, list) else [] |
| out: Dict[str, str] = {} |
| for fd in fields_def: |
| if not isinstance(fd, dict): |
| continue |
| key = _safe_str(fd.get("key")) |
| if not key: |
| continue |
| val = _safe_str(body.fields.get(key)) |
| if fd.get("required") and not val: |
| label = _safe_str(fd.get("label")) or key |
| raise HTTPException(status_code=400, detail=f"{label} is required") |
| if val: |
| out[key] = val |
| return out |
|
|
|
|
| |
|
|
|
|
| @router.get("/api/cta-forms") |
| def list_cta_forms(request: Request, tc: TenantContext = Depends(get_tenant_context)): |
| rows = ( |
| tc.db.query(CtaForm) |
| .filter(CtaForm.tenant_id == tc.tenant_id) |
| .order_by(CtaForm.updated_at.desc(), CtaForm.id.desc()) |
| .all() |
| ) |
| return {"forms": [_form_to_dict(r, request=request) for r in rows]} |
|
|
|
|
| @router.post("/api/cta-forms") |
| def create_cta_form( |
| body: CtaFormCreateRequest, |
| request: Request, |
| tc: TenantContext = Depends(get_tenant_context), |
| ): |
| fields = _normalize_fields([f.model_dump() for f in body.fields] if body.fields else DEFAULT_FIELDS) |
| row = CtaForm( |
| tenant_id=tc.tenant_id, |
| user_id=tc.user_id, |
| public_id=uuid.uuid4().hex, |
| name=body.name.strip(), |
| fields=fields, |
| is_active=1, |
| ) |
| tc.db.add(row) |
| db_commit_with_retry(tc.db) |
| tc.db.refresh(row) |
| return _form_to_dict(row, request=request) |
|
|
|
|
| @router.get("/api/cta-forms/{form_id}") |
| def get_cta_form(form_id: int, request: Request, tc: TenantContext = Depends(get_tenant_context)): |
| row = ( |
| tc.db.query(CtaForm) |
| .filter(CtaForm.tenant_id == tc.tenant_id, CtaForm.id == form_id) |
| .first() |
| ) |
| if not row: |
| raise HTTPException(status_code=404, detail="Form not found") |
| return _form_to_dict(row, request=request) |
|
|
|
|
| @router.patch("/api/cta-forms/{form_id}") |
| def patch_cta_form( |
| form_id: int, |
| body: CtaFormPatchRequest, |
| request: Request, |
| tc: TenantContext = Depends(get_tenant_context), |
| ): |
| row = ( |
| tc.db.query(CtaForm) |
| .filter(CtaForm.tenant_id == tc.tenant_id, CtaForm.id == form_id) |
| .first() |
| ) |
| if not row: |
| raise HTTPException(status_code=404, detail="Form not found") |
| if body.name is not None: |
| row.name = body.name.strip() |
| if body.fields is not None: |
| row.fields = _normalize_fields([f.model_dump() for f in body.fields]) |
| if body.is_active is not None: |
| row.is_active = 1 if body.is_active else 0 |
| row.updated_at = datetime.utcnow() |
| db_commit_with_retry(tc.db) |
| tc.db.refresh(row) |
| return _form_to_dict(row, request=request) |
|
|
|
|
| @router.delete("/api/cta-forms/{form_id}") |
| def delete_cta_form(form_id: int, tc: TenantContext = Depends(get_tenant_context)): |
| row = ( |
| tc.db.query(CtaForm) |
| .filter(CtaForm.tenant_id == tc.tenant_id, CtaForm.id == form_id) |
| .first() |
| ) |
| if not row: |
| raise HTTPException(status_code=404, detail="Form not found") |
| tc.db.delete(row) |
| db_commit_with_retry(tc.db) |
| return {"ok": True} |
|
|
|
|
| |
|
|
|
|
| @router.get("/api/public/cta-forms/{public_id}") |
| def public_cta_form(public_id: str, request: Request, db: Session = Depends(get_db)): |
| row = _get_active_form(db, public_id) |
| return JSONResponse( |
| content={ |
| "public_id": row.public_id, |
| "name": row.name or "", |
| "fields": row.fields if isinstance(row.fields, list) else [], |
| }, |
| headers=_CORS_HEADERS, |
| ) |
|
|
|
|
| @router.options("/api/public/cta-forms/{public_id}/submit") |
| async def public_cta_submit_options(): |
| return Response(headers=_CORS_HEADERS) |
|
|
|
|
| @router.post("/api/public/cta-forms/{public_id}/submit") |
| def public_cta_submit( |
| public_id: str, |
| body: CtaFormSubmitRequest, |
| db: Session = Depends(get_db), |
| ): |
| form = _get_active_form(db, public_id) |
| payload = _validate_submission(form, body) |
| lead = _create_lead_from_submission( |
| db, |
| form, |
| payload, |
| page_url=_safe_str(body.page_url), |
| ) |
| db_commit_with_retry(db) |
| return JSONResponse( |
| content={"ok": True, "lead_id": lead.id, "message": "Thank you — we will be in touch soon."}, |
| headers=_CORS_HEADERS, |
| ) |
|
|
|
|
| @router.get("/api/embed/cta/{public_id}.js") |
| def embed_cta_script(public_id: str, request: Request, db: Session = Depends(get_db)): |
| form = _get_active_form(db, public_id) |
| origin = _public_origin(request) |
| fields_json = json.dumps(form.fields if isinstance(form.fields, list) else []) |
| js = f"""(function() {{ |
| var FORM_ID = {json.dumps(public_id)}; |
| var API = {json.dumps(origin)}; |
| var FIELDS = {fields_json}; |
| var mount = document.getElementById('ezofis-cta-' + FORM_ID); |
| if (!mount) {{ |
| var s = document.currentScript; |
| mount = document.createElement('div'); |
| mount.id = 'ezofis-cta-' + FORM_ID; |
| if (s && s.parentNode) s.parentNode.insertBefore(mount, s); |
| else document.body.appendChild(mount); |
| }} |
| var css = document.createElement('style'); |
| css.textContent = '.ezofis-cta{{font-family:system-ui,-apple-system,sans-serif;max-width:100%;margin:0 auto;padding:16px;border:1px solid #e2e8f0;border-radius:12px;background:#fff;box-sizing:border-box}}.ezofis-cta *{{box-sizing:border-box}}.ezofis-cta label{{display:block;font-size:0.8125rem;font-weight:500;color:#475569;margin-bottom:4px}}.ezofis-cta input,.ezofis-cta textarea{{width:100%;padding:10px 12px;border:1px solid #cbd5e1;border-radius:8px;font-size:0.9375rem;margin-bottom:12px}}.ezofis-cta textarea{{min-height:96px;resize:vertical}}.ezofis-cta button{{width:100%;padding:12px;background:#7c3aed;color:#fff;border:none;border-radius:8px;font-weight:600;font-size:0.9375rem;cursor:pointer}}.ezofis-cta button:disabled{{opacity:0.6;cursor:not-allowed}}.ezofis-cta .ok{{color:#15803d;font-size:0.875rem;text-align:center;padding:12px}}.ezofis-cta .err{{color:#b91c1c;font-size:0.8125rem;margin-bottom:8px}}'; |
| document.head.appendChild(css); |
| var root = document.createElement('div'); |
| root.className = 'ezofis-cta'; |
| mount.appendChild(root); |
| var err = document.createElement('div'); |
| err.className = 'err'; |
| err.style.display = 'none'; |
| root.appendChild(err); |
| var form = document.createElement('form'); |
| root.appendChild(form); |
| FIELDS.forEach(function(f) {{ |
| var wrap = document.createElement('div'); |
| var lab = document.createElement('label'); |
| lab.textContent = f.label + (f.required ? ' *' : ''); |
| wrap.appendChild(lab); |
| var inp; |
| if (f.type === 'textarea') {{ |
| inp = document.createElement('textarea'); |
| }} else {{ |
| inp = document.createElement('input'); |
| inp.type = f.type || 'text'; |
| }} |
| inp.name = f.key; |
| inp.required = !!f.required; |
| wrap.appendChild(inp); |
| form.appendChild(wrap); |
| }}); |
| var btn = document.createElement('button'); |
| btn.type = 'submit'; |
| btn.textContent = 'Submit'; |
| form.appendChild(btn); |
| form.addEventListener('submit', function(e) {{ |
| e.preventDefault(); |
| err.style.display = 'none'; |
| btn.disabled = true; |
| var data = {{ fields: {{}}, page_url: window.location.href }}; |
| FIELDS.forEach(function(f) {{ |
| var el = form.querySelector('[name="' + f.key + '"]'); |
| if (el) data.fields[f.key] = el.value || ''; |
| }}); |
| fetch(API + '/api/public/cta-forms/' + FORM_ID + '/submit', {{ |
| method: 'POST', |
| headers: {{ 'Content-Type': 'application/json' }}, |
| body: JSON.stringify(data) |
| }}).then(function(r) {{ |
| return r.json().then(function(j) {{ return {{ ok: r.ok, j: j }}; }}); |
| }}).then(function(res) {{ |
| if (res.ok) {{ |
| form.innerHTML = '<div class="ok">' + (res.j.message || 'Thank you!') + '</div>'; |
| }} else {{ |
| err.textContent = (res.j && res.j.detail) ? (typeof res.j.detail === 'string' ? res.j.detail : 'Submission failed') : 'Submission failed'; |
| err.style.display = 'block'; |
| btn.disabled = false; |
| }} |
| }}).catch(function() {{ |
| err.textContent = 'Network error. Please try again.'; |
| err.style.display = 'block'; |
| btn.disabled = false; |
| }}); |
| }}); |
| }})(); |
| """ |
| return PlainTextResponse(js, media_type="application/javascript", headers=_CORS_HEADERS) |
|
|
|
|
| def lead_is_cta_form(row: CrmLead) -> bool: |
| raw = row.raw_webhook if isinstance(row.raw_webhook, dict) else {} |
| if raw.get("source") == CTA_LEAD_SOURCE: |
| return True |
| sid = _safe_str(row.smartlead_lead_id) |
| return sid.startswith(CTA_LEAD_ID_PREFIX) |
|
|
