"""Action Pipeline — post-submission action execution engine.
Executes configured actions (webhook, email, log, redirect, document)
after a form submission. Actions run in execution_order and are
tracked per-submission for auditing.
"""
import json
import logging
import threading
from datetime import UTC, datetime, timezone
logger = logging.getLogger("agentforms.actions")
logger.setLevel(logging.INFO)
if not logger.handlers:
_h = logging.StreamHandler(__import__("sys").stderr)
_h.setFormatter(logging.Formatter("[actions] %(message)s"))
logger.addHandler(_h)
def _build_submission_data(site, submission, field_config=None):
"""Build normalized submission data dict for action handlers."""
fields = {}
if submission.get("customer_name"):
fields["name"] = submission["customer_name"]
if submission.get("customer_phone"):
fields["phone"] = submission["customer_phone"]
if submission.get("customer_email"):
fields["email"] = submission["customer_email"]
if submission.get("customer_equipment"):
fields["equipment"] = submission["customer_equipment"]
if submission.get("customer_message"):
fields["message"] = submission["customer_message"]
# Merge dynamic data
if submission.get("data"):
try:
dynamic = json.loads(submission["data"])
fields.update(dynamic)
except (json.JSONDecodeError, TypeError):
pass
return {
"site_id": site["id"],
"site_name": site.get("name", ""),
"submission_id": submission["id"],
"submitted_at": submission.get("submitted_at", datetime.now(UTC).isoformat()),
"fields": fields,
"field_config": field_config,
}
def _handle_webhook(action, submission_data):
"""Execute a webhook action — fires a POST to the configured URL."""
config = action.get("config", {})
url = config.get("url")
if not url:
logger.warning(f"Webhook action {action['id']} missing URL, skipping")
return {"success": False, "error": "missing_url"}
headers = config.get("headers", {"Content-Type": "application/json"})
timeout = config.get("timeout", 10)
import requests
try:
resp = requests.post(url, json=submission_data, headers=headers, timeout=timeout)
success = 200 <= resp.status_code < 300
logger.info(f"Webhook action {action['id']}: {url} → {resp.status_code}")
return {
"success": success,
"status_code": resp.status_code,
"response": resp.text[:500],
}
except requests.RequestException as e:
logger.error(f"Webhook action {action['id']} failed: {e}")
return {"success": False, "error": str(e)}
def _handle_email(action, submission_data):
"""Execute an email action — sends notification to configured recipients."""
config = action.get("config", {})
recipients = config.get("recipients", [])
subject = config.get("subject", f"New submission from {submission_data['site_name']}")
body_template = config.get("body", None)
if not recipients:
logger.warning(f"Email action {action['id']} has no recipients, skipping")
return {"success": False, "error": "no_recipients"}
# Build body from template or default
if body_template:
body = body_template
for key, value in submission_data["fields"].items():
body = body.replace(f"{{{key}}}", str(value))
else:
field_lines = []
for key, value in submission_data["fields"].items():
field_lines.append(f" {key}: {value}")
body = f"New submission from {submission_data['site_name']}\n\n" + "\n".join(field_lines)
# Send via SMTP — same mechanism as send_notification in api.py
import os
import smtplib
import ssl
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
smtp_host = os.environ.get("SMTP_HOST")
smtp_port = int(os.environ.get("SMTP_PORT", "587"))
smtp_user = os.environ.get("SMTP_USER", "")
smtp_pass = os.environ.get("SMTP_PASS", "")
smtp_from = config.get("from_email") or config.get("from_name", submission_data["site_name"])
if not smtp_host:
logger.warning(f"Email action {action['id']}: SMTP_HOST not configured, skipping")
return {"success": False, "error": "smtp_not_configured"}
try:
msg = MIMEMultipart()
msg["From"] = smtp_from
msg["To"] = ", ".join(recipients)
msg["Subject"] = subject
msg.attach(MIMEText(body, "plain"))
context = ssl.create_default_context()
with smtplib.SMTP(smtp_host, smtp_port, timeout=15) as server:
server.ehlo()
if smtp_user and smtp_pass:
server.starttls(context=context)
server.ehlo()
server.login(smtp_user, smtp_pass)
server.send_message(msg)
logger.info(f"Email action {action['id']}: sent to {len(recipients)} recipients")
return {"success": True, "recipients": len(recipients)}
except Exception as e:
logger.error(f"Email action {action['id']} failed: {e}")
return {"success": False, "error": str(e)}
def _handle_log(action, submission_data):
"""Execute a log action — writes structured log entry."""
config = action.get("config", {})
level = config.get("level", "info").upper()
message_template = config.get("message", "New submission: {submission_id}")
# Substitute variables
message = message_template
all_data = {
**submission_data["fields"],
**{
"submission_id": str(submission_data["submission_id"]),
"site_id": str(submission_data["site_id"]),
"site_name": submission_data["site_name"],
},
}
for key, value in all_data.items():
message = message.replace(f"{{{key}}}", str(value))
log_method = getattr(logger, level.lower(), logger.info)
log_method(f"[action:{action['id']}] {message}")
return {"success": True, "message": message}
def _handle_redirect(action, submission_data):
"""Execute a redirect action — returns redirect URL (stored for API response)."""
config = action.get("config", {})
url = config.get("url", "")
if not url:
logger.warning(f"Redirect action {action['id']} missing URL, skipping")
return {"success": False, "error": "missing_url"}
# Substitute variables
for key, value in submission_data["fields"].items():
url = url.replace(f"{{{key}}}", str(value))
url = url.replace("{submission_id}", str(submission_data["submission_id"]))
logger.info(f"Redirect action {action['id']}: URL → {url}")
return {"success": True, "redirect_url": url}
def _handle_document(action, submission_data):
"""Execute a document action — triggers document generation from template."""
config = action.get("config", {})
template_id = config.get("template_id")
if not template_id:
logger.warning(f"Document action {action['id']} missing template_id, skipping")
return {"success": False, "error": "missing_template_id"}
from app.models import get_document_template
from app.services.documents import render_document
try:
template = get_document_template(template_id, site_id=submission_data["site_id"])
if not template:
logger.warning(f"Document action {action['id']}: template {template_id} not found")
return {"success": False, "error": "template_not_found"}
# Build submission dict for render_document
submission_dict = {
"submitted_at": submission_data["submitted_at"],
**submission_data["fields"],
}
result = render_document(
template_id=template_id,
submission_id=submission_data["submission_id"],
template_data=template,
submission_data=submission_dict,
)
logger.info(f"Document action {action['id']}: generated doc for template {template_id}")
return {
"success": True,
"document_id": result.get("document_id", "unknown"),
"size_bytes": result.get("size_bytes", 0),
}
except Exception as e:
logger.error(f"Document action {action['id']} failed: {e}")
return {"success": False, "error": str(e)}
def _handle_chain(action, submission_data):
"""Execute a chain action — runs a multi-step workflow.
Chains are queued via RQ for persistence. Falls back to inline
execution if Redis is unavailable.
"""
import os
config = action.get("config", {})
# Ensure config is parsed as dict (may be stored as JSON string in DB)
if isinstance(config, str):
try:
config = json.loads(config)
except (json.JSONDecodeError, TypeError):
logger.warning(f"Chain action {action['id']}: invalid config JSON")
return {"success": False, "error": "invalid_config"}
steps = config.get("steps", {})
if not steps:
logger.warning(f"Chain action {action['id']} has no steps, skipping")
return {"success": False, "error": "empty_chain"}
# Build site dict for chain engine
site = {
"id": submission_data["site_id"],
"name": submission_data["site_name"],
}
submission = {
"id": submission_data["submission_id"],
"submitted_at": submission_data["submitted_at"],
"customer_name": submission_data["fields"].get("name"),
"customer_phone": submission_data["fields"].get("phone"),
"customer_email": submission_data["fields"].get("email"),
"customer_equipment": submission_data["fields"].get("equipment"),
"customer_message": submission_data["fields"].get("message"),
}
# Reconstruct submission data JSON
dynamic_fields = {
k: v
for k, v in submission_data["fields"].items()
if k not in ("name", "phone", "email", "equipment", "message")
}
if dynamic_fields:
submission["data"] = json.dumps(dynamic_fields)
chain_action = {
"id": action["id"],
"config": config,
}
# Enrich chain config with IDs for wait step continuation
if isinstance(steps, dict):
for step_id, step_cfg in steps.items():
if isinstance(step_cfg, dict) and step_cfg.get("action") == "wait":
step_cfg["config"] = step_cfg.get("config", {})
step_cfg["config"]["_chain_id"] = action["id"]
step_cfg["config"]["_next_step_id"] = step_cfg.get("on_success", "")
# Try to queue via RQ
redis_url = os.environ.get("REDIS_URL")
if redis_url:
try:
import redis as redis_lib
from rq import Queue
r = redis_lib.from_url(redis_url)
r.ping()
q = Queue("chains", connection=r)
q.enqueue(
_execute_chain_rq,
args=(chain_action, site, submission),
job_timeout=3600,
result_ttl=86400,
)
logger.info(f"Chain action {action['id']}: queued via RQ")
return {
"success": True,
"message": "Chain execution queued",
"chain_id": action["id"],
}
except Exception as e:
logger.warning(f"RQ unavailable for chain, running inline: {e}")
# Fallback: inline execution
from app.services.chain_engine import execute_chain
result = execute_chain(chain_action, site, submission, submission_data.get("field_config"))
logger.info(
f"Chain action {action['id']}: {result['steps_executed']} steps executed, {result['steps_failed']} failed"
)
return {
"success": result["success"],
"steps_executed": result["steps_executed"],
"steps_failed": result["steps_failed"],
"error": result.get("error"),
}
def _execute_chain_rq(chain_action, site, submission):
"""RQ job wrapper for chain execution."""
from app.services.chain_engine import execute_chain
return execute_chain(chain_action, site, submission)
# Handler registry
_HANDLERS = {
"webhook": _handle_webhook,
"email": _handle_email,
"log": _handle_log,
"redirect": _handle_redirect,
"document": _handle_document,
"chain": _handle_chain,
}
def execute_actions(site, submission, field_config=None):
"""Execute all enabled actions for a site after submission.
Returns a dict with results from each action and any redirect URL.
Args:
site: Site dict from DB
submission: Submission dict from DB (with id)
field_config: Optional parsed field config
Returns:
dict with keys: results (list of {action_id, type, success, data}),
redirect_url (str or None)
"""
from app.models import get_actions_by_site
actions = get_actions_by_site(site["id"])
if not actions:
return {"results": [], "redirect_url": None}
submission_data = _build_submission_data(site, submission, field_config)
results = []
redirect_url = None
for action in actions:
if not action.get("enabled", True):
continue
action_type = action.get("type", "")
handler = _HANDLERS.get(action_type)
if not handler:
logger.warning(f"Unknown action type '{action_type}' for action {action['id']}")
results.append(
{
"action_id": action["id"],
"type": action_type,
"success": False,
"data": {"error": f"unknown_action_type:{action_type}"},
}
)
continue
try:
result = handler(action, submission_data)
result["action_id"] = action["id"]
result["type"] = action_type
results.append(result)
# Capture redirect URL for API response
if action_type == "redirect" and result.get("success"):
redirect_url = result.get("redirect_url")
except Exception as e:
logger.error(f"Action {action['id']} ({action_type}) threw: {e}")
results.append(
{
"action_id": action["id"],
"type": action_type,
"success": False,
"data": {"error": str(e)},
}
)
return {"results": results, "redirect_url": redirect_url}
def execute_actions_async(site, submission, field_config=None):
"""Fire execute_actions in a background thread (non-blocking).
Used by submit() endpoint — redirects are handled inline,
async actions (webhook, email, document) run in background.
"""
thread = threading.Thread(
target=execute_actions,
args=(site, submission, field_config),
daemon=True,
)
thread.start()