"""Detector: orphaned CRM deals — open deals with value but no owner."""
from __future__ import annotations
from typing import List
from sqlalchemy import or_
from .base import BaseDetector, LeakCandidate
from ._common import (
CLOSED_STAGES,
connector_available,
deal_weighted_value,
)
class CrmOrphanedDealDetector(BaseDetector):
id = "crm_orphaned_deals"
name = "CRM: Orphaned Deals"
connector = "hubspot"
enabled_by_default = True
description = (
"Flags open CRM deals with a positive value but no assigned owner — "
"nobody is working them."
)
default_params = {"min_amount": 0}
def check(self, company_id: str) -> List[LeakCandidate]:
if not connector_available(company_id, self.connector):
return []
from app.models import CrmDeal
deals = (
CrmDeal.query.filter(
CrmDeal.company_id == company_id,
CrmDeal.stage.notin_(list(CLOSED_STAGES)),
CrmDeal.amount > 0,
or_(
CrmDeal.deal_owner_id.is_(None),
CrmDeal.deal_owner_id == "",
),
).all()
)
candidates: List[LeakCandidate] = []
for deal in deals:
amount = deal.amount or 0.0
candidates.append(
LeakCandidate(
detector_id=self.id,
source="CRM: Orphaned Deal",
description=(
f"Deal '{deal.name}' (${amount:,.2f}) has no assigned owner"
),
estimated_loss=deal_weighted_value(deal.amount, deal.probability),
severity="medium",
metadata_json={
"dedupe_key": f"{self.id}:{deal.id}",
"deal_id": deal.id,
},
source_connector="hubspot",
rule_params={"min_amount": 0},
)
)
return candidates