"""Detector: unworked leads — leads 48+ hours old with no associated deal."""
from __future__ import annotations
from datetime import timedelta
from typing import List
from .base import BaseDetector, LeakCandidate
from ._common import CLOSED_STAGES, connector_available, days_since, utcnow
LEAD_LIFECYCLE_STAGES = ("lead", "subscriber", "contact")
EXTERNAL_LEAD_SERVICES = ("leadperfection", "five9")
class LeadsUnworkedDetector(BaseDetector):
id = "leads_unworked"
name = "Leads: Unworked Leads"
connector = None # hubspot and/or external lead sources
enabled_by_default = True
description = (
"Flags leads created 48+ hours ago that still have no associated "
"open deal or follow-up."
)
default_params = {"min_age_hours": 48}
def check(self, company_id: str) -> List[LeakCandidate]:
candidates: List[LeakCandidate] = []
candidates.extend(self._check_crm_contacts(company_id))
candidates.extend(self._check_external_leads(company_id))
return candidates
# -- CRM contacts (HubSpot) ------------------------------------------------
def _check_crm_contacts(self, company_id: str) -> List[LeakCandidate]:
if not connector_available(company_id, "hubspot"):
return []
from app.models import CrmContact, CrmDeal
cutoff = utcnow() - timedelta(hours=48)
open_deal_exists = (
CrmDeal.query.filter(
CrmDeal.company_id == company_id,
CrmDeal.contact_id == CrmContact.id,
CrmDeal.stage.notin_(list(CLOSED_STAGES)),
).exists()
)
contacts = (
CrmContact.query.filter(
CrmContact.company_id == company_id,
CrmContact.lifecycle_stage.in_(list(LEAD_LIFECYCLE_STAGES)),
CrmContact.created_at < cutoff,
~open_deal_exists,
).all()
)
out: List[LeakCandidate] = []
for contact in contacts:
days_old = days_since(contact.created_at)
name = contact.full_name or contact.email or contact.external_id
out.append(
LeakCandidate(
detector_id=self.id,
source="Leads: Unworked Lead",
description=(
f"Lead '{name}' created {days_old} days ago with no "
f"associated deal or follow-up"
),
estimated_loss=None,
severity="medium",
metadata_json={
"dedupe_key": f"{self.id}:crm:{contact.id}",
"contact_id": contact.id,
"source_service": "hubspot",
"days_unchanged": days_old,
},
source_connector="hubspot",
rule_params={"min_age_hours": 48},
)
)
return out
# -- External lead sources (LeadPerfection, Five9) --------------------------
def _check_external_leads(self, company_id: str) -> List[LeakCandidate]:
from app.models import ExternalSyncRecord
cutoff = utcnow() - timedelta(hours=48)
records = (
ExternalSyncRecord.query.filter(
ExternalSyncRecord.company_id == company_id,
ExternalSyncRecord.source_service.in_(list(EXTERNAL_LEAD_SERVICES)),
ExternalSyncRecord.entity_type.in_(["lead", "contact"]),
ExternalSyncRecord.created_at < cutoff,
).all()
)
if not records:
return []
# Only report from services whose connector is actually connected.
connected = {
svc for svc in EXTERNAL_LEAD_SERVICES
if connector_available(company_id, svc)
}
out: List[LeakCandidate] = []
for rec in records:
if rec.source_service not in connected:
continue
if self._has_matching_deal(company_id, rec):
continue
days_old = days_since(rec.created_at)
name = rec.name or rec.external_id
out.append(
LeakCandidate(
detector_id=self.id,
source="Leads: Unworked Lead",
description=(
f"Lead '{name}' created {days_old} days ago with no "
f"associated deal or follow-up"
),
estimated_loss=None,
severity="medium",
metadata_json={
"dedupe_key": f"{self.id}:ext:{rec.id}",
"contact_id": rec.id,
"source_service": rec.source_service,
"days_unchanged": days_old,
},
source_connector=rec.source_service,
rule_params={"min_age_hours": 48},
)
)
return out
@staticmethod
def _has_matching_deal(company_id: str, record) -> bool:
"""Best-effort match of an external lead to an open CRM deal.
External records aren't FK-linked to crm_deals; match via a linked
crm contact id in properties_json, falling back to name matching.
"""
from app.models import CrmDeal
props = record.properties_json or {}
contact_id = props.get("crm_contact_id") or props.get("contact_id")
if contact_id:
match = CrmDeal.query.filter(
CrmDeal.company_id == company_id,
CrmDeal.contact_id == contact_id,
CrmDeal.stage.notin_(list(CLOSED_STAGES)),
).first()
if match is not None:
return True
if record.name:
match = CrmDeal.query.filter(
CrmDeal.company_id == company_id,
CrmDeal.name == record.name,
CrmDeal.stage.notin_(list(CLOSED_STAGES)),
).first()
if match is not None:
return True
return False