"""Zapier Connector — webhook-based automation bridge.
Zapier delivers events via webhook POST to your endpoint.
This connector receives Zap triggers, maps them to Command Center actions,
and stores results as CrmContacts / ActivityLogs.
Triggers Command Center can expose to Zaps:
- new_lead
- project_updated
- goal_completed
- kpi_threshold
Actions Command Center can receive from Zaps:
- create_lead
- update_contact
- log_activity
"""
from __future__ import annotations
import logging
import time
import uuid
from datetime import datetime, timezone
from typing import Any, Dict, Optional
import requests
from . import BaseConnector, _REGISTRY, register_connector
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Registration
# ---------------------------------------------------------------------------
register_connector(
"zapier",
{
"service": "zapier",
"name": "Zapier",
"category": "automation",
"description": (
"Connect Zapier to automate workflows between Command Center "
"and 5000+ apps. Trigger actions from Command Center data or "
"push Zap data into Command Center."
),
"auth_type": "api_key",
"auth_fields": ["api_key"],
"capabilities": ["webhooks", "triggers", "actions"],
"rate_limit": "Webhook-based; rate depends on Zap volume",
"docs_url": "https://platform.zapier.com/",
},
)
class ZapierConnector(BaseConnector):
"""Zapier webhook connector.
Users configure their Zaps to POST to:
{base_url}/api/connectors/webhooks/zapier
The connector validates the API key from Zapier Platform, processes
the incoming payload, and maps it to internal models.
"""
_SERVICE = "zapier"
_API_BASE = "https://platform.zapier.com/api/v1"
# ------------------------------------------------------------------
# Public API (implements BaseConnector interface)
# ------------------------------------------------------------------
def connect(self) -> Dict[str, Any]:
"""Validate the Zapier Platform API key.
Sends a GET to the Zapier Platform account endpoint to confirm
the API key is valid.
Returns ``{'status': 'connected', 'account': ...}`` on success.
"""
self._log(event_type="connect_attempt", status="pending")
start = time.monotonic()
api_key = self.config.get("api_key", "").strip()
if not api_key:
self._log(event_type="connect_error", status="error",
error_message="API key is required")
return {"status": "error", "error": "API key is required"}
try:
resp = self._retry(
requests.get,
f"{self._API_BASE}/account/",
headers=self._auth_headers(),
timeout=15,
)
resp.raise_for_status()
except requests.HTTPError as exc:
status_code = getattr(exc.response, "status_code", 0)
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="connect_error",
status="error",
duration_ms=duration_ms,
error_message=f"Zapier API returned {status_code}: {exc}",
)
return {
"status": "error",
"error": f"Zapier API returned {status_code} — check your API key",
}
except Exception as exc:
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="connect_error",
status="error",
duration_ms=duration_ms,
error_message=str(exc),
)
return {"status": "error", "error": str(exc)}
self._connected = True
duration_ms = int((time.monotonic() - start) * 1000)
# Extract account info (best-effort)
account_info = {}
try:
account_info = resp.json()
except Exception:
pass
self._log(
event_type="connect_success",
status="success",
duration_ms=duration_ms,
details={"account": account_info},
)
return {
"status": "connected",
"account": account_info,
"duration_ms": duration_ms,
"message": (
"Zapier connected. Configure your Zaps to POST to "
f"{self._webhook_url()}."
),
}
def disconnect(self) -> Dict[str, Any]:
"""Disconnect Zapier connector — clear in-memory state."""
self._connected = False
# Wipe sensitive config
self.config.pop("api_key", None)
self._log(event_type="disconnect", status="success")
return {"status": "disconnected", "service": "zapier"}
def sync(self) -> Dict[str, Any]:
"""No traditional sync — Zapier is event-driven via webhooks.
Returns a status summary of webhook activity.
"""
start = time.monotonic()
total_received = self.config.get("total_webhooks_received", 0)
last_received = self.config.get("last_webhook_received")
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="sync_complete",
status="success",
record_count=0,
duration_ms=duration_ms,
details={
"total_webhooks_received": total_received,
"last_webhook_received": last_received,
"note": "Zapier uses webhook delivery. Events are processed in real-time.",
},
)
return {
"status": "success",
"record_count": 0,
"duration_ms": duration_ms,
"details": {
"total_webhooks_received": total_received,
"last_webhook_received": last_received,
"note": (
"Zapier uses webhook delivery. "
"Events are processed in real-time as they arrive."
),
},
}
def status(self) -> Dict[str, Any]:
"""Check current connection health."""
api_key = self.config.get("api_key")
if not api_key:
return {
"service": "zapier",
"connected": False,
"status": "not_configured",
"message": "Zapier connector not configured — no API key set.",
}
# Light check against Zapier API
try:
resp = requests.get(
f"{self._API_BASE}/account/",
headers=self._auth_headers(),
timeout=10,
)
healthy = resp.status_code < 400
except Exception as exc:
return {
"service": "zapier",
"connected": self._connected,
"status": "error",
"message": f"API check failed: {exc}",
"total_webhooks_received": self.config.get("total_webhooks_received", 0),
"last_webhook_received": self.config.get("last_webhook_received"),
}
return {
"service": "zapier",
"connected": self._connected and healthy,
"status": "active" if healthy else "api_error",
"total_webhooks_received": self.config.get("total_webhooks_received", 0),
"last_webhook_received": self.config.get("last_webhook_received"),
"webhook_url": self._webhook_url(),
}
# ------------------------------------------------------------------
# Webhook processing
# ------------------------------------------------------------------
def process_webhook(
self,
payload: Dict[str, Any],
request_headers: Optional[Dict[str, str]] = None,
) -> Dict[str, Any]:
"""Process an incoming Zapier webhook.
Expected payload structure:
{
"action": "create_lead" | "update_contact" | "log_activity",
"data": { ... fields for the specific action ... }
}
Zapier also sends an X-Zapier-Api-Key header when using API key auth.
"""
try:
# Optional: validate Zapier API key from header
zapier_api_key = (
request_headers.get("X-Zapier-Api-Key", "")
if request_headers
else ""
)
if zapier_api_key and self.config.get("api_key"):
if zapier_api_key != self.config["api_key"]:
logger.warning("Zapier webhook: API key mismatch")
return {
"success": False,
"error": "Invalid API key",
}
# Extract action and data
action = payload.get("action", "")
data = payload.get("data", payload)
if not action:
return {
"success": False,
"error": "No action specified in payload",
}
# Dispatch to handler
handler = getattr(self, f"_handle_{action}", None)
if handler is None:
logger.warning("Zapier webhook: Unknown action '%s'", action)
return {
"success": False,
"error": f"Unknown action: {action}",
}
result = handler(data)
# Update webhook stats
self.config["total_webhooks_received"] = (
self.config.get("total_webhooks_received", 0) + 1
)
self.config["last_webhook_received"] = (
datetime.now(timezone.utc).isoformat()
)
logger.info(
"Zapier webhook: action='%s' processed", action
)
return {
"success": True,
"action": action,
"result": result,
}
except Exception as exc:
logger.error(
"Zapier webhook processing error: %s", str(exc), exc_info=True
)
return {"success": False, "error": str(exc)}
# -- Action handlers -----------------------------------------------------
def _handle_create_lead(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""Store incoming data as a CrmContact lead."""
from app.models import db, CrmContact
external_id = data.get("id", data.get("contact_id", f"zap_{uuid.uuid4().hex[:12]}"))
# Build properties dict from any extra fields
known_fields = {
"id", "contact_id", "email", "first_name", "last_name",
"phone", "company_name", "subject", "message",
}
properties = {
k: v for k, v in data.items() if k not in known_fields
}
properties["source"] = "zapier"
self._merge(
CrmContact,
self.company_id,
f"zapier_{external_id}",
{
"email": data.get("email", ""),
"first_name": data.get("first_name", ""),
"last_name": data.get("last_name", ""),
"phone": data.get("phone", ""),
"company_name": data.get("company_name", ""),
"lifecycle_stage": "lead",
"hubspot_owner_id": "",
"properties_json": properties,
},
)
db.session.commit()
logger.info("Zapier create_lead: stored contact %s", external_id)
return {"contact_id": f"zapier_{external_id}"}
def _handle_update_contact(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""Update an existing CrmContact by external_id."""
from app.models import db, CrmContact
external_id = data.get("id", data.get("contact_id"))
if not external_id:
return {"error": "contact_id or id is required for update"}
record = CrmContact.query.filter_by(
company_id=self.company_id,
external_id=f"zapier_{external_id}",
).first()
if not record:
logger.warning("Zapier update_contact: not found %s", external_id)
return {"error": f"Contact not found: {external_id}"}
# Update mutable fields
for field in ("email", "first_name", "last_name", "phone",
"company_name", "lifecycle_stage"):
if field in data:
setattr(record, field, data[field])
# Merge extra properties
if "properties" in data:
record.properties_json = record.properties_json or {}
record.properties_json.update(data["properties"])
db.session.commit()
logger.info("Zapier update_contact: updated %s", external_id)
return {"contact_id": f"zapier_{external_id}"}
def _handle_log_activity(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""Log an activity note from Zapier."""
from app.models import db, ActivityLog
activity = ActivityLog(
company_id=self.company_id,
user_id="",
activity_type="zapier_webhook",
description=data.get("description", data.get("message", "")),
details_json={
"source": "zapier",
"action": data.get("action", "unknown"),
"data": data,
},
)
db.session.add(activity)
db.session.commit()
logger.info("Zapier log_activity: recorded activity %s", activity.id)
return {"activity_id": activity.id}
# -- Helpers -------------------------------------------------------------
def _auth_headers(self) -> Dict[str, str]:
"""Build auth headers for Zapier Platform API."""
api_key = self.config.get("api_key", "")
return {"Authorization": f"Bearer {api_key}"}
def _webhook_url(self) -> str:
"""Return the webhook URL for Zaps to POST to."""
return "/api/connectors/webhooks/zapier"
@staticmethod
def _merge(
model,
company_id: str,
external_id: str,
defaults: Dict[str, Any],
) -> int:
"""Find or create a record by (company_id, external_id) and update."""
from app.models import db
record = model.query.filter_by(
company_id=company_id,
external_id=external_id,
).first()
if record:
for key, value in defaults.items():
setattr(record, key, value)
else:
record = model(
company_id=company_id,
external_id=external_id,
**defaults,
)
db.session.add(record)
return 1
# -- Test helpers --------------------------------------------------------
def test_webhook(self) -> Dict[str, Any]:
"""Generate a test payload for webhook testing."""
test_payload = {
"action": "create_lead",
"data": {
"id": f"ZAP-TEST-{int(time.time())}",
"first_name": "Test",
"last_name": "Contact",
"email": "test@example.com",
"phone": "(555) 555-5555",
"company_name": "Test Company",
"source": "zapier_test",
},
}
return {
"test_payload": test_payload,
"webhook_url": self._webhook_url(),
"instructions": (
"POST this payload to your webhook URL to test processing"
),
}
# Auto-register on import
_REGISTRY["zapier"] = ZapierConnector