"""LeadPerfection connector.
LeadPerfection is a lead management / lead distribution platform for home
improvement contractors. It ingests leads from marketplaces (Angi,
HomeAdvisor, etc.) and routes them to contractors.
Auth: token endpoint on ``https://api.leadperfection.com/token`` using a
password grant with LeadPerfection-specific ``appkey`` and ``clientid``
parameters. The user supplies:
* ``username`` — LeadPerfection API user
* ``password`` — LeadPerfection API password
* ``app_key`` — application key issued by LeadPerfection
* ``client_id`` — LeadPerfection client identifier
* ``base_url`` — optional override (default ``https://api.leadperfection.com``)
Key endpoints:
* ``POST /api/Leads/GetLeadsForwardLook`` — pull upcoming/appointment leads
* ``POST /api/Leads/LeadAdd`` — push a new lead in
Synced leads are upserted into ``ExternalSyncRecord`` keyed by
``(company_id, external_id)`` so repeated syncs are idempotent.
"""
from __future__ import annotations
import logging
import time
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
import requests
from . import BaseConnector, register_connector, _REGISTRY
logger = logging.getLogger(__name__)
# -- Registration metadata ----------------------------------------------------
register_connector(
"leadperfection",
{
"service": "leadperfection",
"display_name": "LeadPerfection",
"name": "LeadPerfection",
"category": "crm",
"description": "Lead management for home improvement contractors — pull marketplace leads, push new leads, and track lead status.",
"auth_type": "api_key",
"auth_fields": ["username", "password", "app_key", "client_id"],
"config_fields": [
{"name": "username", "label": "API Username", "type": "text", "required": True},
{"name": "password", "label": "API Password", "type": "password", "required": True},
{"name": "app_key", "label": "App Key (issued by LeadPerfection)", "type": "password", "required": True},
{"name": "client_id", "label": "Client ID", "type": "text", "required": True},
{"name": "base_url", "label": "API Base URL", "type": "text", "required": False},
],
"capabilities": ["leads", "lead_push", "lead_status"],
"data_types": ["leads"],
"rate_limit": "Not documented; conservative back-off applied.",
"docs_url": "https://api.leadperfection.com/Help",
},
)
_DEFAULT_BASE = "https://api.leadperfection.com"
class LeadPerfectionConnector(BaseConnector):
"""LeadPerfection integration — pull marketplace leads, push new leads."""
_SERVICE = "leadperfection"
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
self._token: str = ""
self._token_obtained_at: float = 0.0
self._token_ttl: float = 0.0
# -- Config helpers ---------------------------------------------------------
@property
def _base_url(self) -> str:
return (self.config.get("base_url") or _DEFAULT_BASE).rstrip("/")
# -- Auth --------------------------------------------------------------------
def _get_token(self, force: bool = False) -> str:
"""Fetch (and cache) a bearer token from the LeadPerfection token endpoint.
LeadPerfection uses a password grant with extra ``appkey`` and
``clientid`` form fields.
"""
now = time.time()
if (
not force
and self._token
and (self._token_ttl <= 0 or now - self._token_obtained_at < self._token_ttl - 60)
):
return self._token
username = self.config.get("username", "")
password = self.config.get("password", "")
app_key = self.config.get("app_key", "")
client_id = self.config.get("client_id", "")
if not all([username, password, app_key, client_id]):
raise ValueError(
"LeadPerfection credentials incomplete — username, password, "
"app_key, and client_id are all required."
)
resp = requests.post(
f"{self._base_url}/token",
data={
"grant_type": "password",
"username": username,
"password": password,
"appkey": app_key,
"clientid": client_id,
},
headers={"Accept": "application/json"},
timeout=30,
)
resp.raise_for_status()
data = resp.json()
self._token = data.get("access_token", "")
if not self._token:
raise ValueError("LeadPerfection token endpoint returned no access_token")
self._token_obtained_at = now
try:
self._token_ttl = float(data.get("expires_in") or 0)
except (TypeError, ValueError):
self._token_ttl = 0.0
return self._token
def _api_post(self, path: str, payload: Optional[Dict[str, Any]] = None) -> Any:
"""POST to a LeadPerfection API endpoint with bearer auth."""
token = self._get_token()
url = f"{self._base_url}{path}"
resp = requests.post(
url,
json=payload or {},
headers={
"Authorization": f"Bearer {token}",
"Accept": "application/json",
},
timeout=30,
)
if resp.status_code == 401:
# Token expired mid-session — refresh once and retry.
token = self._get_token(force=True)
resp = requests.post(
url,
json=payload or {},
headers={
"Authorization": f"Bearer {token}",
"Accept": "application/json",
},
timeout=30,
)
if resp.status_code == 429:
retry_after = resp.headers.get("Retry-After")
logger.warning("LeadPerfection rate limited: 429, Retry-After=%s", retry_after)
resp.raise_for_status()
if not resp.content:
return {}
return resp.json()
# -- connect / disconnect ------------------------------------------------------
def connect(self) -> Dict[str, Any]:
"""Validate credentials by acquiring a token."""
self._log(event_type="connect_attempt", status="pending")
start = time.monotonic()
try:
self._retry(self._get_token, True)
self._connected = True
duration_ms = int((time.monotonic() - start) * 1000)
self._log(event_type="connect_success", status="success", duration_ms=duration_ms)
return {
"status": "connected",
"service": "leadperfection",
"duration_ms": duration_ms,
}
except Exception as exc:
self._log(event_type="connect_error", status="error", error_message=str(exc))
return {"status": "error", "error": str(exc)}
def test_connection(self) -> Dict[str, Any]:
"""Validate credentials without side effects — used by the test route."""
return self.connect()
def disconnect(self) -> Dict[str, Any]:
"""Drop the cached token and clear stored credentials."""
self._connected = False
self._token = ""
for key in ("password", "app_key"):
self.config.pop(key, None)
self._log(event_type="disconnect", status="success")
return {"status": "disconnected", "service": "leadperfection"}
# -- sync --------------------------------------------------------------------------
def sync(self) -> Dict[str, Any]:
"""Pull leads via GetLeadsForwardLook and upsert into ExternalSyncRecord.
Idempotent — leads are upserted by (company_id, external_id).
"""
self._log(event_type="sync_start", status="pending")
start = time.monotonic()
total = 0
try:
leads = self._sync_leads()
total += leads
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="sync_complete",
status="success",
record_count=total,
duration_ms=duration_ms,
details={"leads": leads},
)
return {
"status": "success",
"record_count": total,
"duration_ms": duration_ms,
"details": {"leads": leads},
"last_sync_at": datetime.now(timezone.utc).isoformat(),
}
except Exception as exc:
self._log(event_type="sync_error", status="error", error_message=str(exc))
return {"status": "error", "error": str(exc), "record_count": total}
def _sync_leads(self) -> int:
"""Pull forward-look leads and upsert them locally."""
from app.models import db, ExternalSyncRecord
self._check_rate_limit()
data = self._retry(self._api_post, "/api/Leads/GetLeadsForwardLook", {})
# Response may be a bare list or wrapped in a key.
if isinstance(data, dict):
items = data.get("leads") or data.get("Leads") or data.get("items") or []
else:
items = data or []
if not isinstance(items, list):
items = []
count = 0
for item in items:
if not isinstance(item, dict):
continue
raw_id = (
item.get("leadid")
or item.get("LeadId")
or item.get("lead_id")
or item.get("id")
)
if raw_id in (None, ""):
continue
external_id = f"leadperfection:lead:{raw_id}"
record = ExternalSyncRecord.query.filter_by(
company_id=self.company_id, external_id=external_id
).first()
if record is None:
record = ExternalSyncRecord(
company_id=self.company_id,
external_id=external_id,
source_service="leadperfection",
entity_type="lead",
)
db.session.add(record)
first = str(item.get("firstname") or item.get("FirstName") or "")
last = str(item.get("lastname") or item.get("LastName") or "")
record.name = (f"{first} {last}".strip() or str(raw_id))[:255]
record.status = str(
item.get("status") or item.get("Status") or item.get("apptresult") or ""
)[:100]
record.properties_json = {
"phone": item.get("phone") or item.get("Phone", ""),
"email": item.get("email") or item.get("Email", ""),
"product": item.get("productid") or item.get("ProductID", ""),
"source": item.get("srs_id") or item.get("source", ""),
"appt_date": item.get("apptdate") or item.get("ApptDate", ""),
"zip": item.get("zip") or item.get("Zip", ""),
"raw": item,
}
count += 1
db.session.commit()
return count
# -- lead push --------------------------------------------------------------------
def push_lead(self, lead: Dict[str, Any]) -> Dict[str, Any]:
"""Push a new lead into LeadPerfection via the LeadAdd endpoint.
Expected keys (LeadPerfection LeadAdd schema — pass-through for any
additional fields the account supports): firstname, lastname,
address1, city, state, zip, phone, email, productid, proddescr,
srs_id (source), notes.
"""
self._log(event_type="lead_push_start", status="pending")
try:
payload = {
"firstname": lead.get("first_name", lead.get("firstname", "")),
"lastname": lead.get("last_name", lead.get("lastname", "")),
"address1": lead.get("address", lead.get("address1", "")),
"city": lead.get("city", ""),
"state": lead.get("state", ""),
"zip": lead.get("zip", lead.get("zip_code", "")),
"phone": lead.get("phone", ""),
"email": lead.get("email", ""),
"productid": lead.get("product_id", lead.get("productid", "")),
"proddescr": lead.get("product_description", lead.get("proddescr", "")),
"srs_id": lead.get("source_id", lead.get("srs_id", "")),
"notes": lead.get("notes", ""),
}
# Pass through any extra LeadPerfection-native fields.
for key, value in lead.items():
if key not in payload and value not in (None, ""):
payload[key] = value
result = self._retry(self._api_post, "/api/Leads/LeadAdd", payload)
self._log(
event_type="lead_push_success",
status="success",
record_count=1,
details={"result": result if isinstance(result, dict) else {}},
)
return {"status": "success", "result": result}
except Exception as exc:
self._log(event_type="lead_push_error", status="error", error_message=str(exc))
return {"status": "error", "error": str(exc)}
# -- status -----------------------------------------------------------------------
def status(self) -> Dict[str, Any]:
"""Report connection health."""
return {
"service": "leadperfection",
"connected": self._connected or bool(self._token),
"has_credentials": bool(
self.config.get("username")
and self.config.get("password")
and self.config.get("app_key")
and self.config.get("client_id")
),
"base_url": self._base_url,
}
_REGISTRY["leadperfection"] = LeadPerfectionConnector