"""Detector: speed-to-lead gap — new leads with no call/SMS within 5 minutes. Cross-connector: joins lead sources (HubSpot / LeadPerfection / Angi) with communication activity (Five9 call logs, Marlimar SMS, Slack alerts). """ from __future__ import annotations from datetime import datetime, timedelta from typing import Dict, List, Optional, Tuple from .base import BaseDetector, LeakCandidate from ._common import ( any_connector_available, as_aware, normalize_phone, parse_dt, utcnow, ) COMM_SERVICES = ("five9", "marlimar") LEAD_SERVICES = ("hubspot", "leadperfection", "angi_leads") SLACK_GRACE_MINUTES = 15 def _minutes_severity(minutes: float) -> str: if minutes > 120: return "critical" if minutes > 30: return "high" return "medium" class SpeedToLeadDetector(BaseDetector): id = "speed_to_lead" name = "Speed-to-Lead: Slow Response" connector = ["five9", "marlimar"] # at least one comm channel + a lead source enabled_by_default = True description = ( "Cross-references new leads (HubSpot, LeadPerfection, Angi) with call " "and SMS activity (Five9, Marlimar, Slack) and flags leads not " "contacted within 5 minutes. Industry benchmark: 5 minutes." ) default_params = {"speed_to_lead_minutes": 5, "benchmark_minutes": 5, "window_days": 7, "slack_grace_minutes": SLACK_GRACE_MINUTES} def check(self, company_id: str) -> List[LeakCandidate]: # Need at least one communication channel and at least one lead source. if not any_connector_available(company_id, COMM_SERVICES): return [] if not any_connector_available(company_id, LEAD_SERVICES): return [] from app.models import AngiLead, CrmContact, ExternalSyncRecord cutoff = utcnow() - timedelta(days=7) # ------------------------------------------------------------------ # 1. Collect candidate leads from all available sources. # Each entry: (lead_id, source, name, phone, email, created, cost) # ------------------------------------------------------------------ leads: List[Tuple[str, str, str, str, str, datetime, Optional[float]]] = [] for contact in CrmContact.query.filter( CrmContact.company_id == company_id, CrmContact.lifecycle_stage == "lead", CrmContact.created_at > cutoff, ).all(): leads.append(( contact.id, "hubspot", contact.full_name or contact.email or "Unknown", contact.phone or "", contact.email or "", as_aware(contact.created_at), None, )) for rec in ExternalSyncRecord.query.filter( ExternalSyncRecord.company_id == company_id, ExternalSyncRecord.source_service == "leadperfection", ExternalSyncRecord.entity_type == "lead", ExternalSyncRecord.created_at > cutoff, ).all(): props = rec.properties_json or {} leads.append(( rec.id, "leadperfection", rec.name or "Unknown", props.get("phone") or "", props.get("email") or "", as_aware(rec.created_at), None, )) for angi in AngiLead.query.filter( AngiLead.company_id == company_id, AngiLead.created_at > cutoff, ).all(): leads.append(( angi.id, angi.lead_source or "angi", angi.full_name or "Unknown", angi.phone or "", angi.email or "", as_aware(angi.received_at or angi.created_at), angi.cost_per_lead, )) if not leads: return [] # ------------------------------------------------------------------ # 2. Build contact-event indexes: phone/email -> sorted timestamps. # Five9 calls + Marlimar messages count as first contact. # Slack notifications count with a 15-minute grace window. # ------------------------------------------------------------------ contact_events: Dict[str, List[datetime]] = {} slack_events: Dict[str, List[datetime]] = {} def _index(store: Dict[str, List[datetime]], key: str, ts: Optional[datetime]): if key and ts is not None: store.setdefault(key, []).append(ts) comm_records = ExternalSyncRecord.query.filter( ExternalSyncRecord.company_id == company_id, ExternalSyncRecord.source_service.in_(["five9", "marlimar", "slack"]), ExternalSyncRecord.created_at > cutoff - timedelta(days=1), ).all() for rec in comm_records: props = rec.properties_json or {} ts = ( parse_dt(props.get("timestamp")) or parse_dt(props.get("call_time")) or parse_dt(props.get("sent_at")) or as_aware(rec.created_at) ) phone = normalize_phone( props.get("phone") or props.get("contact_phone") or props.get("to") or props.get("dnis") or props.get("ani") ) email = (props.get("email") or props.get("contact_email") or "").strip().lower() if rec.source_service == "slack": _index(slack_events, phone, ts) _index(slack_events, email, ts) # Slack messages often only carry a name/text mention _index(slack_events, (rec.name or "").strip().lower(), ts) elif rec.source_service == "five9" and rec.entity_type == "call_log": _index(contact_events, phone, ts) _index(contact_events, email, ts) elif rec.source_service == "marlimar" and rec.entity_type == "message": direction = str(props.get("direction", "")).lower() if direction != "inbound": # outbound SMS = contact attempt _index(contact_events, phone, ts) _index(contact_events, email, ts) for store in (contact_events, slack_events): for key in store: store[key].sort() def _first_after(store: Dict[str, List[datetime]], keys, after: datetime): best = None for key in keys: for ts in store.get(key, ()): if ts >= after: if best is None or ts < best: best = ts break return best # ------------------------------------------------------------------ # 3. Evaluate each lead's time-to-first-contact. # ------------------------------------------------------------------ now = utcnow() candidates: List[LeakCandidate] = [] for lead_id, lead_source, name, phone, email, created, cost in leads: if created is None: continue minutes_to_contact: Optional[float] = None # Angi leads carry their own response tracking — trust it first. if lead_source in ("angi", "homeadvisor"): angi = AngiLead.query.get(lead_id) if angi is not None: if angi.response_time_minutes is not None: minutes_to_contact = float(angi.response_time_minutes) else: first = as_aware(angi.first_contact_at or angi.responded_at) if first is not None: minutes_to_contact = (first - created).total_seconds() / 60.0 if minutes_to_contact is None: keys = [k for k in (normalize_phone(phone), (email or "").strip().lower()) if k] first_contact = _first_after(contact_events, keys, created) if keys else None first_slack = _first_after( slack_events, keys + [(name or "").strip().lower()], created ) if first_contact is not None: minutes_to_contact = (first_contact - created).total_seconds() / 60.0 elif first_slack is not None: slack_minutes = (first_slack - created).total_seconds() / 60.0 if slack_minutes <= SLACK_GRACE_MINUTES: continue # Slack alert acted on fast enough minutes_to_contact = slack_minutes else: # Never contacted — clock is still running. minutes_to_contact = (now - created).total_seconds() / 60.0 minutes_to_contact = max(round(minutes_to_contact, 1), 0.0) if minutes_to_contact <= 5: continue candidates.append( LeakCandidate( detector_id=self.id, source="Speed-to-Lead: Slow Response", description=( f"Lead '{name}' waited {int(minutes_to_contact)} minutes " f"for first contact. Industry benchmark: 5 min." ), estimated_loss=cost, severity=_minutes_severity(minutes_to_contact), metadata_json={ "dedupe_key": f"{self.id}:{lead_source}:{lead_id}", "lead_id": lead_id, "lead_source": lead_source, "minutes_to_contact": minutes_to_contact, "lead_phone": phone, }, source_connector="five9" if any_connector_available( company_id, ["five9"] ) else "marlimar", rule_params={"speed_to_lead_minutes": 5, "benchmark_minutes": 5}, ) ) return candidates