"""Detector: stale CRM deals — open deals with no activity for 7+ days."""
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,
deal_weighted_value,
utcnow,
)
class CrmStaleDealDetector(BaseDetector):
id = "crm_stale_deals"
name = "CRM: Stale Deals"
connector = "hubspot"
enabled_by_default = True
description = (
"Flags open CRM deals with no updates in 7+ days. Severity escalates "
"to high after 14 days of inactivity."
)
default_params = {"stale_days": 7, "high_severity_days": 14}
def check(self, company_id: str) -> List[LeakCandidate]:
if not connector_available(company_id, self.connector):
return []
from app.models import CrmDeal
cutoff = utcnow() - timedelta(days=7)
deals = (
CrmDeal.query.filter(
CrmDeal.company_id == company_id,
CrmDeal.stage.notin_(list(CLOSED_STAGES)),
CrmDeal.updated_at < cutoff,
).all()
)
candidates: List[LeakCandidate] = []
for deal in deals:
days_stale = days_since(deal.updated_at)
severity = "high" if days_stale > 14 else "medium"
amount = deal.amount or 0.0
candidates.append(
LeakCandidate(
detector_id=self.id,
source="CRM: Stale Deal",
description=(
f"Deal '{deal.name}' hasn't been updated in {days_stale} "
f"days. Value: ${amount:,.2f}"
),
estimated_loss=deal_weighted_value(deal.amount, deal.probability),
severity=severity,
metadata_json={
"dedupe_key": f"{self.id}:{deal.id}",
"deal_id": deal.id,
"days_stale": days_stale,
"stage": deal.stage,
},
source_connector="hubspot",
rule_params={"stale_days": 7, "high_severity_days": 14},
)
)
return candidates