"""Angi (HomeAdvisor) Connector - Webhook-based lead feed
Angi delivers leads via webhook POST to your endpoint.
This connector receives leads, processes them, and stores them as CrmContacts
and AngiLead records for full lifecycle tracking.
"""
from __future__ import annotations
import hashlib
import hmac
import logging
import re
import statistics
import time
from datetime import datetime, timezone, timedelta
from typing import Any, Dict, List, Optional
from . import BaseConnector, register_connector, _REGISTRY
logger = logging.getLogger(__name__)
register_connector(
"angi",
{
"service": "angi",
"name": "Angi (HomeAdvisor)",
"category": "lead_gen",
"description": "Lead feed webhook integration for Angi Pro accounts",
"auth_type": "spid",
"auth_fields": ["spid", "webhook_url"],
"capabilities": ["leads", "webhooks"],
"rate_limit": "Webhook-based; rate depends on lead volume",
"docs_url": "https://business.angi.com/developers/",
},
)
class AngiConnector(BaseConnector):
"""Angi lead webhook connector"""
_SERVICE = "angi"
def connect(self) -> Dict[str, Any]:
"""Validate SPID and webhook URL configuration"""
self._log(event_type="connect_attempt", status="pending")
start = time.monotonic()
try:
spid = self.config.get("spid", "")
webhook_url = self.config.get("webhook_url", "")
if not spid:
raise ValueError("Angi SPID is required")
if not webhook_url:
raise ValueError("Webhook URL is required")
# Basic SPID validation (numeric)
spid_clean = spid.replace("_", "").replace("-", "")
if not spid_clean.isdigit():
raise ValueError(f"Invalid SPID format: {spid}")
self._connected = True
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="connect_success",
status="success",
duration_ms=duration_ms,
details={"spid": spid, "webhook_url": webhook_url},
)
return {
"status": "connected",
"service": "angi",
"spid": spid,
"webhook_url": webhook_url,
"message": "Angi connector configured. Register this webhook URL with Angi.",
"duration_ms": duration_ms,
}
except Exception as e:
self._log(event_type="connect_error", status="error", error_message=str(e))
return {"status": "error", "error": str(e)}
def disconnect(self) -> Dict[str, Any]:
"""Disconnect Angi connector"""
self._connected = False
self.config.pop("spid", None)
self.config.pop("webhook_url", None)
self._log(event_type="disconnect", status="success")
return {"status": "disconnected", "service": "angi"}
def sync(self) -> Dict[str, Any]:
"""
Process any pending leads from the webhook queue and refresh analytics.
Since Angi is webhook-based, 'sync' processes any pending leads
that arrived via webhook but haven't been fully processed yet,
pulls lead status updates, and updates lead analytics.
"""
from app.models import db, AngiLead
self._log(event_type="sync_start", status="pending")
start = time.monotonic()
try:
processed = 0
updated = 0
errors = []
# Process any unprocessed leads (leads that are still 'new' and
# have been sitting for a while without response)
try:
leads = AngiLead.query.filter_by(
company_id=self.company_id,
status='new'
).all()
for lead in leads:
try:
# Check if lead is stale (expired without response)
if lead.received_at:
age_hours = (
datetime.now(timezone.utc) - lead.received_at
).total_seconds() / 3600
if age_hours > 72: # 3 days without response
lead.status = 'expired'
lead.internal_notes += (
f"\n[Auto] Expired after 72h without response"
)
updated += 1
except Exception as e:
errors.append(f"Error processing lead {lead.id}: {str(e)}")
processed = len(leads)
db.session.commit()
except Exception as e:
db.session.rollback()
errors.append(f"Error processing leads: {str(e)}")
# Calculate current analytics
analytics = self.get_lead_analytics(period_days=30)
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="sync_complete",
status="success",
record_count=processed,
duration_ms=duration_ms,
details={
"leads_processed": processed,
"leads_updated": updated,
"analytics": analytics,
"errors": errors,
"total_leads_received": self.config.get("total_leads_received", 0),
"last_lead_received": self.config.get("last_lead_received"),
},
)
return {
"status": "success",
"record_count": processed,
"updated": updated,
"duration_ms": duration_ms,
"details": {
"leads_processed": processed,
"leads_updated": updated,
"analytics": analytics,
"total_leads_received": self.config.get("total_leads_received", 0),
"last_lead_received": self.config.get("last_lead_received"),
"note": "Angi uses webhook delivery. Leads are processed in real-time; sync refreshes analytics and flags stale leads.",
},
"errors": errors,
}
except Exception as e:
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="sync_error",
status="error",
duration_ms=duration_ms,
error_message=str(e),
)
return {"status": "error", "error": str(e)}
def status(self) -> Dict[str, Any]:
"""Check connector status"""
spid = self.config.get("spid")
webhook_url = self.config.get("webhook_url")
if not spid:
return {
"service": "angi",
"connected": False,
"status": "not_configured",
"message": "Angi connector not configured.",
}
return {
"service": "angi",
"connected": self._connected,
"status": self.config.get("status", "active"),
"spid": spid,
"webhook_url": webhook_url,
"total_leads_received": self.config.get("total_leads_received", 0),
"last_lead_received": self.config.get("last_lead_received"),
}
# -- Webhook processing --------------------------------------------------
def process_webhook(self, payload: Dict[str, Any], request_headers: Dict[str, str] = None) -> Dict[str, Any]:
"""
Process an incoming Angi lead webhook.
Expected payload structure:
{
"lead": {
"lead_id": "string",
"first_name": "string",
"last_name": "string",
"email": "string",
"phone": "string",
"project_type": "string",
"budget": "string",
"timeline": "string",
"description": "string",
"address": "string",
"city": "string",
"state": "string",
"zip": "string",
"timestamp": "ISO 8601"
}
}
Returns:
Processing result dict
"""
try:
# Validate webhook signature if shared secret is configured
shared_secret = self.config.get("shared_secret")
if shared_secret and request_headers:
signature = request_headers.get("X-Angi-Signature", "")
if not self._validate_signature(payload, signature, shared_secret):
logger.warning("Angi webhook signature validation failed")
return {"success": False, "error": "Invalid webhook signature"}
# Extract lead data
lead_data = payload.get("lead", payload)
if not lead_data:
return {"success": False, "error": "No lead data in payload"}
# Process the lead
lead_id = lead_data.get("lead_id", "unknown")
processed = self._process_lead(lead_data)
if not processed:
return {"success": False, "error": "Lead processing failed: missing or invalid lead_id"}
# Update config stats
self.config["total_leads_received"] = self.config.get("total_leads_received", 0) + 1
self.config["last_lead_received"] = datetime.now(timezone.utc).isoformat()
logger.info("Angi webhook: Lead %s processed successfully", lead_id)
return {
"success": True,
"lead_id": lead_id,
"message": "Lead processed successfully",
}
except Exception as e:
logger.error("Angi webhook processing error: %s", str(e), exc_info=True)
return {"success": False, "error": str(e)}
def _validate_signature(self, payload: Dict, signature: str, shared_secret: str) -> bool:
"""Validate HMAC signature of the webhook payload"""
import json
payload_str = json.dumps(payload, sort_keys=True)
expected = hmac.new(
shared_secret.encode(),
payload_str.encode(),
hashlib.sha256,
).hexdigest()
return hmac.compare_digest(expected, signature)
def _process_lead(self, lead_data: Dict[str, Any]) -> bool:
"""Store lead as CrmContact AND AngiLead in the database. Returns True if lead was processed."""
from app.models import db, CrmContact, AngiLead, AngiLeadAction
lead_id = lead_data.get("lead_id", "")
if not lead_id:
logger.warning("Angi lead has no lead_id, skipping")
return False
# Parse budget
budget_raw = lead_data.get("budget", "")
budget_value = self._parse_budget(budget_raw)
# -- Create/update CrmContact --
properties = {
"source": "angi",
"project_type": lead_data.get("project_type") or lead_data.get("service_type", ""),
"budget": budget_raw,
"timeline": lead_data.get("timeline", ""),
"description": lead_data.get("description") or lead_data.get("project_description", ""),
"address": lead_data.get("address", ""),
"city": lead_data.get("city", ""),
"state": lead_data.get("state", ""),
"zip": lead_data.get("zip", ""),
"timestamp": lead_data.get("timestamp", ""),
"match_type": lead_data.get("match_type", ""),
"interview_qa": lead_data.get("interview_qa"),
"tcpa_compliant": lead_data.get("tcpa_consent", False),
}
self._merge(
CrmContact,
self.company_id,
f"angi_{lead_id}",
{
"email": lead_data.get("email", ""),
"first_name": lead_data.get("first_name", ""),
"last_name": lead_data.get("last_name", ""),
"phone": lead_data.get("phone", ""),
"company_name": "",
"lifecycle_stage": "lead",
"hubspot_owner_id": "",
"properties_json": properties,
},
)
# -- Create AngiLead record --
existing = AngiLead.query.filter_by(
company_id=self.company_id,
angi_lead_id=lead_id,
).first()
if not existing:
# Determine priority based on lead attributes
priority = 'medium'
if lead_data.get("is_premium"):
priority = 'high'
if budget_value and budget_value > 10000:
priority = 'high'
angi_lead = AngiLead(
company_id=self.company_id,
angi_lead_id=lead_id,
connector_id=self.connector_id or '',
first_name=lead_data.get("first_name", ""),
last_name=lead_data.get("last_name", ""),
email=lead_data.get("email", ""),
phone=lead_data.get("phone", ""),
project_type=lead_data.get("project_type", ""),
description=lead_data.get("description", ""),
budget=budget_raw,
budget_value=budget_value,
timeline=lead_data.get("timeline", ""),
address=lead_data.get("address", ""),
city=lead_data.get("city", ""),
state=lead_data.get("state", ""),
zip_code=lead_data.get("zip", ""),
status='new',
priority=priority,
is_premium=lead_data.get("is_premium", False),
is_duplicate=False,
cost_per_lead=lead_data.get("cost_per_lead"),
interview_qa=lead_data.get("interview_qa"),
tcpa_compliance=lead_data.get("tcpa_consent", False),
match_type=lead_data.get("match_type", ""),
received_at=self._parse_timestamp(lead_data.get("timestamp")),
)
db.session.add(angi_lead)
db.session.flush() # Populate angi_lead.id before creating action
# Log the 'received' action
action = AngiLeadAction(
lead_id=angi_lead.id,
company_id=self.company_id,
action_type='received',
details=f"Lead received from Angi webhook",
)
db.session.add(action)
else:
# Update existing lead if data changed
if not existing.first_name:
existing.first_name = lead_data.get("first_name", existing.first_name)
if not existing.last_name:
existing.last_name = lead_data.get("last_name", existing.last_name)
if not existing.email:
existing.email = lead_data.get("email", existing.email)
if not existing.phone:
existing.phone = lead_data.get("phone", existing.phone)
db.session.commit()
logger.info("Angi lead %s stored (contact + lead record)", lead_id)
return True
@staticmethod
def _parse_timestamp(ts: str | None) -> datetime | None:
"""Parse an ISO 8601 timestamp string into a timezone-aware datetime."""
if not ts:
return None
try:
dt = datetime.fromisoformat(ts.replace('Z', '+00:00'))
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt
except (ValueError, AttributeError):
return None
# -- Write-back methods --------------------------------------------------
def respond_to_lead(self, lead_id: str, message: str, performed_by: str = '') -> Dict[str, Any]:
"""Log response to lead, update status to 'responded', track response time."""
from app.models import db, AngiLead, AngiLeadAction
lead = AngiLead.query.filter_by(
company_id=self.company_id,
id=lead_id,
).first()
if not lead:
return {"success": False, "error": "Lead not found"}
now = datetime.now(timezone.utc)
# Calculate response time if this is the first response
if not lead.responded_at and lead.received_at:
received = lead.received_at
if received.tzinfo is None:
received = received.replace(tzinfo=timezone.utc)
diff = now - received
lead.response_time_minutes = diff.total_seconds() / 60
lead.responded_at = now
lead.last_contacted_at = now
# Update status — don't downgrade from higher statuses
if lead.status == 'new':
lead.status = 'responded'
# Log the action
action = AngiLeadAction(
lead_id=lead.id,
company_id=self.company_id,
action_type='responded',
details=f"Lead responded to",
message_content=message,
performed_by=performed_by,
)
db.session.add(action)
db.session.commit()
return {
"success": True,
"data": {
"lead_id": lead.id,
"status": lead.status,
"response_time_minutes": lead.response_time_minutes,
"responded_at": now.isoformat(),
}
}
def update_lead_status(self, lead_id: str, status: str, notes: str = '', performed_by: str = '') -> Dict[str, Any]:
"""Update lead status in DB."""
from app.models import db, AngiLead, AngiLeadAction
valid_statuses = ('new', 'accepted', 'responded', 'contacted', 'scheduled', 'won', 'lost', 'expired', 'rejected')
if status not in valid_statuses:
return {"success": False, "error": f"Invalid status. Must be one of: {', '.join(valid_statuses)}"}
lead = AngiLead.query.filter_by(
company_id=self.company_id,
id=lead_id,
).first()
if not lead:
return {"success": False, "error": "Lead not found"}
old_status = lead.status
lead.status = status
if notes:
lead.internal_notes += f"\n{notes}"
if lead.status in ('responded', 'contacted', 'scheduled'):
lead.last_contacted_at = datetime.now(timezone.utc)
# Log the action
action = AngiLeadAction(
lead_id=lead.id,
company_id=self.company_id,
action_type='status_change',
details=f"Status changed from '{old_status}' to '{status}'",
performed_by=performed_by,
)
db.session.add(action)
db.session.commit()
return {
"success": True,
"data": {
"lead_id": lead.id,
"old_status": old_status,
"new_status": status,
}
}
def accept_lead(self, lead_id: str, performed_by: str = '') -> Dict[str, Any]:
"""Accept lead, update status to 'accepted'."""
from app.models import db, AngiLead, AngiLeadAction
lead = AngiLead.query.filter_by(
company_id=self.company_id,
id=lead_id,
).first()
if not lead:
return {"success": False, "error": "Lead not found"}
old_status = lead.status
lead.status = 'accepted'
lead.last_contacted_at = datetime.now(timezone.utc)
# Log the action
action = AngiLeadAction(
lead_id=lead.id,
company_id=self.company_id,
action_type='accepted',
details=f"Lead accepted (was: {old_status})",
performed_by=performed_by,
)
db.session.add(action)
db.session.commit()
return {
"success": True,
"data": {
"lead_id": lead.id,
"status": 'accepted',
"old_status": old_status,
}
}
def reject_lead(self, lead_id: str, reason: str, performed_by: str = '') -> Dict[str, Any]:
"""Reject lead with reason."""
from app.models import db, AngiLead, AngiLeadAction
lead = AngiLead.query.filter_by(
company_id=self.company_id,
id=lead_id,
).first()
if not lead:
return {"success": False, "error": "Lead not found"}
old_status = lead.status
lead.status = 'rejected'
if reason:
lead.internal_notes += f"\n[Rejected] {reason}"
# Log the action
action = AngiLeadAction(
lead_id=lead.id,
company_id=self.company_id,
action_type='rejected',
details=f"Lead rejected: {reason}",
performed_by=performed_by,
)
db.session.add(action)
db.session.commit()
return {
"success": True,
"data": {
"lead_id": lead.id,
"status": 'rejected',
"reason": reason,
"old_status": old_status,
}
}
# -- Analytics -----------------------------------------------------------
def get_lead_analytics(self, period_days: int = 30) -> Dict[str, Any]:
"""Return lead volume, conversion rate, avg response time for the period."""
from app.models import AngiLead
now = datetime.now(timezone.utc)
since = now - timedelta(days=period_days)
leads = AngiLead.query.filter(
AngiLead.company_id == self.company_id,
AngiLead.received_at >= since,
).all()
total_leads = len(leads)
if total_leads == 0:
return {
"period_days": period_days,
"total_leads": 0,
"new_leads": 0,
"accepted_leads": 0,
"responded_leads": 0,
"contacted_leads": 0,
"scheduled_leads": 0,
"won_leads": 0,
"lost_leads": 0,
"expired_leads": 0,
"rejected_leads": 0,
"conversion_rate": 0.0,
"response_rate": 0.0,
"avg_response_time_minutes": None,
"median_response_time_minutes": None,
"total_cost": 0.0,
"avg_cost_per_lead": None,
"premium_lead_count": 0,
"premium_lead_rate": 0.0,
}
# Count by status
status_counts = {}
for lead in leads:
s = lead.status
status_counts[s] = status_counts.get(s, 0) + 1
responded = status_counts.get('responded', 0) + status_counts.get('contacted', 0) + status_counts.get('scheduled', 0) + status_counts.get('won', 0)
accepted = status_counts.get('accepted', 0)
won = status_counts.get('won', 0)
# Response time stats
response_times = [
lead.response_time_minutes for lead in leads
if lead.response_time_minutes is not None and lead.response_time_minutes >= 0
]
avg_response = statistics.mean(response_times) if response_times else None
median_response = statistics.median(response_times) if response_times else None
# Cost stats
costs = [lead.cost_per_lead for lead in leads if lead.cost_per_lead is not None]
total_cost = sum(costs)
avg_cost = total_cost / len(costs) if costs else None
# Premium stats
premium_count = sum(1 for lead in leads if lead.is_premium)
premium_rate = (premium_count / total_leads * 100) if total_leads > 0 else 0.0
# Conversion rate (responded / total)
response_rate = (responded / total_leads * 100) if total_leads > 0 else 0.0
conversion_rate = (won / total_leads * 100) if total_leads > 0 else 0.0
return {
"period_days": period_days,
"total_leads": total_leads,
"new_leads": status_counts.get('new', 0),
"leads_new": status_counts.get('new', 0),
"accepted_leads": accepted,
"responded_leads": status_counts.get('responded', 0),
"leads_responded": status_counts.get('responded', 0),
"contacted_leads": status_counts.get('contacted', 0),
"scheduled_leads": status_counts.get('scheduled', 0),
"won_leads": won,
"lost_leads": status_counts.get('lost', 0),
"expired_leads": status_counts.get('expired', 0),
"rejected_leads": status_counts.get('rejected', 0),
"conversion_rate": round(conversion_rate, 2),
"response_rate": round(response_rate, 2),
"avg_response_time_minutes": round(avg_response, 1) if avg_response is not None else None,
"median_response_time_minutes": round(median_response, 1) if median_response is not None else None,
"total_cost": round(total_cost, 2),
"avg_cost_per_lead": round(avg_cost, 2) if avg_cost is not None else None,
"premium_lead_count": premium_count,
"premium_lead_rate": round(premium_rate, 2),
}
@staticmethod
def _parse_budget(budget_raw: str) -> Optional[float]:
"""Parse budget string like '$5,000 - $10,000' to float, returning upper bound for ranges."""
if not budget_raw:
return None
# Remove currency symbols and commas
cleaned = budget_raw.replace("$", "").replace(",", "").replace("–", " ").replace("-", " ").strip()
# Extract all numbers
parts = cleaned.split()
numbers: list[float] = []
for part in parts:
try:
numbers.append(float(part))
except ValueError:
continue
if not numbers:
return None
# Return upper bound (max) when a range is detected, otherwise single value
return max(numbers) if len(numbers) > 1 else numbers[0]
# -- Merge helper --------------------------------------------------------
@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 fields."""
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 lead payload for webhook testing"""
test_lead = {
"lead": {
"lead_id": f"ANGI-TEST-{int(time.time())}",
"first_name": "Test",
"last_name": "Customer",
"email": "test@example.com",
"phone": "(555) 555-5555",
"project_type": "Kitchen Remodel",
"budget": "$5,000 - $10,000",
"timeline": "Within 3 months",
"description": "Test lead from Command Sovereignty",
"address": "123 Test St",
"city": "Test City",
"state": "TX",
"zip": "12345",
"timestamp": datetime.now(timezone.utc).isoformat(),
}
}
return {
"test_payload": test_lead,
"webhook_url": f"/api/connectors/webhooks/angi",
"instructions": "POST this payload to your webhook URL to test processing",
}
_REGISTRY["angi"] = AngiConnector