"""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