"""Phase 7: Integration formatters and setup guides.
Each integration has:
- A formatter function: (submission, site, field_config, destination) -> (url, headers, body, method)
- A setup guide: config_fields, steps, requires_url
"""
import json
from datetime import UTC, datetime, timezone
import bleach
# ─── Formatter functions ────────────────────────────────────────────────────────
def format_webhook(submission, site, field_config, destination):
"""Generic webhook — sends all submission data as JSON."""
from app.services.form_logic import evaluate_all_conditions
payload = {
"event": "submission",
"site_id": site["id"],
"site_name": site["name"],
"submission_id": submission["id"],
"submitted_at": submission.get("submitted_at", ""),
"fields": _build_flat_fields(submission, field_config),
}
# Add field metadata with conditional info
if field_config:
fields = _build_flat_fields(submission, field_config)
metadata = evaluate_all_conditions(field_config, fields)
if metadata:
payload["field_metadata"] = {
"visible_fields": metadata.get("visible_fields", {}),
"computed_fields": metadata.get("computed_fields", {}),
"applied_conditions": metadata.get("applied_conditions", []),
}
return {
"url": destination["url"],
"headers": {"Content-Type": "application/json"},
"body": json.dumps(payload),
"method": "POST",
}
def format_google_sheets(submission, site, field_config, destination):
"""Google Sheets via Apps Script Web App.
Expects destination.url = Apps Script deployment URL.
Optionally config.sheet_name for the target sheet.
Sends a JSON array of values (one row) that the Apps Script doPost
can parse and append.
"""
config = destination.get("config") or {}
fields = _build_flat_fields(submission, field_config)
sheet_name = config.get("sheet_name") or "Sheet1"
# Header row on first delivery, then data row
payload = {
"sheet": sheet_name,
"headers": list(fields.keys()),
"values": list(fields.values()),
"submitted_at": submission.get("submitted_at", ""),
}
return {
"url": destination["url"],
"headers": {"Content-Type": "application/json"},
"body": json.dumps(payload),
"method": "POST",
}
def format_slack(submission, site, field_config, destination):
"""Slack Incoming Webhook.
Sends a formatted block message with fields as key-value pairs.
"""
config = destination.get("config") or {}
fields = _build_flat_fields(submission, field_config)
form_label = config.get("form_label") or site["name"]
# Build blocks — first a header, then field sections
fields_block = []
for key, val in fields.items():
if val:
label = key.replace("_", " ").title()
fields_block.append(
{
"type": "section",
"fields": [
{"type": "mrkdwn", "text": f"*{label}*\n"},
{"type": "mrkdwn", "text": f"{_slack_escape(str(val))}\n"},
],
}
)
payload = {
"text": f"New form submission from {form_label}",
"blocks": [
{
"type": "header",
"text": {"type": "plain_text", "text": f"📋 {form_label}"},
},
*fields_block,
{
"type": "context",
"elements": [
{
"type": "mrkdwn",
"text": f"Submitted at {_slack_escape(submission.get('submitted_at', ''))}",
}
],
},
],
}
return {
"url": destination["url"],
"headers": {"Content-Type": "application/json"},
"body": json.dumps(payload),
"method": "POST",
}
def format_discord(submission, site, field_config, destination):
"""Discord Webhook.
Sends an embed with fields as Discord embed fields.
"""
config = destination.get("config") or {}
fields = _build_flat_fields(submission, field_config)
# Discord embed fields have 25-char name limit and 1024-char value limit
embed_fields = []
for key, val in fields.items():
if val:
label = key.replace("_", " ").title()[:25]
value = str(val)[:1024] or "\u200b" # zero-width space for empty
embed_fields.append({"name": label, "value": value, "inline": True})
# Discord embed max 25 fields — truncate if needed
embed_fields = embed_fields[:25]
payload = {
"username": config.get("username") or "AgentForms",
"embeds": [
{
"title": site["name"],
"description": f"New form submission — {submission.get('submitted_at', '')}",
"color": 3447003, # blue
"fields": embed_fields,
"footer": {"text": f"AgentForms • Submission #{submission.get('id', '?')}"},
"timestamp": datetime.now(UTC).isoformat(),
}
],
}
return {
"url": destination["url"],
"headers": {"Content-Type": "application/json"},
"body": json.dumps(payload),
"method": "POST",
}
def format_telegram(submission, site, field_config, destination):
"""Telegram Bot API.
Sends a formatted message to the configured chat.
Expects config.bot_token and config.chat_id.
"""
config = destination.get("config") or {}
bot_token = config.get("bot_token") or ""
chat_id = config.get("chat_id") or ""
fields = _build_flat_fields(submission, field_config)
# Build HTML-formatted message
lines = [f"<b>📋 {site['name']}</b>\n"]
for key, val in fields.items():
if val:
label = key.replace("_", " ").title()
lines.append(f"<b>{_tg_escape(label)}:</b> {_tg_escape(str(val))}")
lines.append(f"\n<i>{_tg_escape(submission.get('submitted_at', ''))}</i>")
text = "\n".join(lines)
# Telegram 4096 char limit
text = text[:4096]
# Build the API URL from config
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
payload = {
"chat_id": chat_id,
"text": text,
"parse_mode": "HTML",
}
return {
"url": url,
"headers": {"Content-Type": "application/json"},
"body": json.dumps(payload),
"method": "POST",
}
def format_airtable(submission, site, field_config, destination):
"""Airtable API — create record.
Expects destination.url = https://api.airtable.com/v0/{BASE_ID}/{TABLE_NAME}
config.api_key for Bearer token.
"""
config = destination.get("config") or {}
api_key = config.get("api_key") or ""
fields = _build_flat_fields(submission, field_config)
payload = {"fields": fields}
return {
"url": destination["url"],
"headers": {
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
},
"body": json.dumps(payload),
"method": "POST",
}
def format_notion(submission, site, field_config, destination):
"""Notion API — append blocks to a page.
Expects destination.url = https://api.notion.com/v1/blocks/{BLOCK_ID}/children
config.api_key for Bearer token.
"""
config = destination.get("config") or {}
api_key = config.get("api_key") or ""
fields = _build_flat_fields(submission, field_config)
# Build paragraph blocks for each field
children = []
# Header
children.append(
{
"object": "block",
"heading_2": {
"rich_text": [
{
"type": "text",
"text": {"content": f"Submission from {site['name']}"},
}
],
},
}
)
# Divider
children.append(
{
"object": "block",
"divider": {},
}
)
# Fields as bullet points
for key, val in fields.items():
if val:
label = key.replace("_", " ").title()
children.append(
{
"object": "block",
"paragraph": {
"rich_text": [
{"type": "text", "text": {"content": f"{label}: "}, "annotations": {"bold": True}},
{"type": "text", "text": {"content": str(val)}},
],
},
}
)
# Timestamp
children.append(
{
"object": "block",
"paragraph": {
"rich_text": [
{
"type": "text",
"text": {"content": f"Submitted at {submission.get('submitted_at', '')}"},
"annotations": {"italic": True},
}
],
},
}
)
payload = {"children": children}
return {
"url": destination["url"],
"headers": {
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
"Notion-Version": "2022-06-28",
},
"body": json.dumps(payload),
"method": "POST",
}
# ─── Formatter registry ───────────────────────────────────────────────────────
FORMATTERS = {
"webhook": format_webhook,
"google_sheets": format_google_sheets,
"slack": format_slack,
"discord": format_discord,
"telegram": format_telegram,
"airtable": format_airtable,
"notion": format_notion,
}
# ─── Setup guides ─────────────────────────────────────────────────────────────
SETUP_GUIDES = {
"webhook": {
"config_fields": [],
"requires_url": True,
"steps": [
"Provide the full URL where you want to receive form submissions.",
"Your endpoint will receive a POST request with JSON payload on every submission.",
"Payload includes: event, site_id, site_name, submission_id, submitted_at, fields.",
],
},
"google_sheets": {
"config_fields": [
{
"key": "sheet_name",
"label": "Sheet Name",
"type": "text",
"required": False,
"placeholder": "Sheet1",
},
],
"requires_url": True,
"steps": [
"Open your Google Sheet → Extensions → Apps Script.",
"Paste this script (or adapt your existing doPost handler):",
(
{
"title": "Apps Script",
"step": 'function doPost(e) { var data = JSON.parse(e.postData.contents); var sheet = SpreadsheetApp.getActiveSpreadsheet().getSheetByName(data.sheet || "Sheet1"); if (sheet.getLastRow() === 0) sheet.appendRow(data.headers); sheet.appendRow(data.values); return ContentService.createTextOutput("ok"); }',
}
),
"Deploy → New deployment → Web app → Execute as: Me → Access: Anyone.",
"Paste the deployment URL as the integration URL above.",
],
},
"slack": {
"config_fields": [
{
"key": "form_label",
"label": "Form Label",
"type": "text",
"required": False,
"placeholder": "",
},
],
"requires_url": True,
"steps": [
"Go to https://api.slack.com/apps → Create New App.",
"Add feature: Incoming Webhooks → toggle ON.",
"Select a default channel for webhook messages.",
"Copy the Webhook URL and paste it above.",
],
},
"discord": {
"config_fields": [
{
"key": "username",
"label": "Webhook Username",
"type": "text",
"required": False,
"placeholder": "AgentForms",
},
],
"requires_url": True,
"steps": [
"Open Discord → right-click a channel → Edit Channel.",
"Go to Integrations → Webhooks → New Webhook.",
"Copy the webhook URL and paste it above.",
"Optionally customize the username and avatar.",
],
},
"telegram": {
"config_fields": [
{
"key": "bot_token",
"label": "Bot Token",
"type": "text",
"required": True,
"placeholder": "123456:ABC-DEF...",
},
{
"key": "chat_id",
"label": "Chat ID",
"type": "text",
"required": True,
"placeholder": "-1001234567890",
},
],
"requires_url": False,
"steps": [
"Message @BotFather on Telegram → /newbot → follow prompts to create a bot.",
"Copy your Bot Token and paste it in the config below.",
'Find your Chat ID: message your bot, then visit https://api.telegram.org/bot{TOKEN}/getUpdates — look for "chat": {"id": ...}.',
"For groups/channels: add your bot as admin first, then get the chat ID from getUpdates.",
"Paste your Chat ID below (groups/channels start with -100...).",
],
},
"airtable": {
"config_fields": [
{
"key": "api_key",
"label": "API Key",
"type": "text",
"required": True,
"placeholder": "pat...",
},
],
"requires_url": True,
"steps": [
"Go to Airtable → create a Personal Access Token (https://airtable.com/create/tokens).",
"Scope: data.records:write for your base.",
"Your URL should be: https://api.airtable.com/v0/{BASE_ID}/{TABLE_NAME}",
"Paste your API Key below. Field names in your form should match Airtable field names.",
],
},
"notion": {
"config_fields": [
{
"key": "api_key",
"label": "API Key (Internal Integration Token)",
"type": "text",
"required": True,
"placeholder": "ntn_...",
},
],
"requires_url": True,
"steps": [
"Go to https://www.notion.so/my-integrations → New Integration.",
"Copy the Internal Integration Token and paste it below.",
"Share your target Notion page/database with the integration.",
"Get the Block ID from your page URL (between / and ?).",
"Your URL: https://api.notion.com/v1/blocks/{BLOCK_ID}/children",
"Each submission will append formatted blocks to your page.",
],
},
}
def get_setup_guide(dest_type):
"""Return setup guide for a destination type."""
guide = SETUP_GUIDES.get(dest_type)
if not guide:
return None
return {
"config_fields": guide.get("config_fields", []),
"requires_url": guide.get("requires_url", True),
"steps": guide.get("steps", []),
}
def format_payload(dest_type, destination, site, submission, field_config=None):
"""Dispatch to the correct formatter for the destination type.
Returns (url, payload, headers) or None if unknown type.
"""
formatter = FORMATTERS.get(dest_type)
if formatter is None:
return None
result = formatter(submission, site, field_config, destination)
return (result["url"], result["body"], result["headers"])
# ─── Helpers ────────────────────────────────────────────────────────────────────
def _sanitize_value(val):
"""Strip HTML tags from form field values before sending to webhooks.
Uses bleach to remove all tags/attributes, leaving plain text.
Prevents XSS when webhook targets render HTML.
"""
if val is None:
return None
if not isinstance(val, str):
return val
return bleach.clean(val, tags=[], strip=True)
def _build_flat_fields(submission, field_config):
"""Build a flat dict of field key → value from submission data."""
fields = {}
# Hardcoded field aliases
aliases = {
"customer_name": "name",
"customer_phone": "phone",
"customer_email": "email",
"customer_equipment": "equipment",
"customer_message": "message",
}
for db_key, alias in aliases.items():
val = submission.get(db_key)
if val:
fields[alias] = _sanitize_value(val)
# Dynamic data from JSON column
raw = submission.get("data")
if raw:
try:
dynamic = json.loads(raw) if isinstance(raw, str) else raw
if isinstance(dynamic, dict):
for k, v in dynamic.items():
fields[k] = _sanitize_value(v)
except (json.JSONDecodeError, TypeError):
pass
# If we have field_config, use field keys (preserve order)
if field_config:
ordered = {}
for f in field_config:
key = f.get("key") or f.get("name", "")
if key and key in fields:
ordered[key] = fields[key]
# Add any fields not in config
for k, v in fields.items():
if k not in ordered:
ordered[k] = v
return ordered
return fields
def _slack_escape(text):
"""Escape Slack mrkdwn special characters."""
return text.replace("&", "&").replace("<", "<").replace(">", ">")
def _tg_escape(text):
"""Escape HTML special characters for Telegram HTML parse mode."""
return text.replace("&", "&").replace("<", "<").replace(">", ">")