"""Detector: CRM deals whose expected close date has slipped past."""
from __future__ import annotations
from typing import List
from .base import BaseDetector, LeakCandidate
from ._common import (
CLOSED_STAGES,
as_aware,
connector_available,
days_since,
deal_weighted_value,
utcnow,
)
class CrmSlippedCloseDetector(BaseDetector):
id = "crm_slipped_close"
name = "CRM: Slipped Close Dates"
connector = "hubspot"
enabled_by_default = True
description = (
"Flags open CRM deals whose expected close date has already passed."
)
default_params = {}
def check(self, company_id: str) -> List[LeakCandidate]:
if not connector_available(company_id, self.connector):
return []
from app.models import CrmDeal
now = utcnow()
deals = (
CrmDeal.query.filter(
CrmDeal.company_id == company_id,
CrmDeal.stage.notin_(list(CLOSED_STAGES)),
CrmDeal.expected_close_date.isnot(None),
CrmDeal.expected_close_date < now,
).all()
)
candidates: List[LeakCandidate] = []
for deal in deals:
days_overdue = days_since(deal.expected_close_date)
expected = as_aware(deal.expected_close_date)
expected_str = expected.date().isoformat() if expected else "unknown"
candidates.append(
LeakCandidate(
detector_id=self.id,
source="CRM: Slipped Close Date",
description=(
f"Deal '{deal.name}' missed close date by {days_overdue} "
f"days. Expected close: {expected_str}"
),
estimated_loss=deal_weighted_value(deal.amount, deal.probability),
severity="high",
metadata_json={
"dedupe_key": f"{self.id}:{deal.id}",
"deal_id": deal.id,
"days_overdue": days_overdue,
"original_date": expected_str,
},
source_connector="hubspot",
rule_params={},
)
)
return candidates