"""Base classes for the auto leak-detection framework (Phase 0).
A detector inspects a company's synced connector data and emits
LeakCandidate objects. The framework upserts them into the RevenueLeak
table idempotently — re-running a scan never duplicates open leaks.
Phase 0 ships only the framework; concrete detectors arrive in Phase 1+.
"""
from __future__ import annotations
import json
import logging
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple
logger = logging.getLogger(__name__)
VALID_SEVERITIES = ("low", "medium", "high", "critical")
@dataclass
class LeakCandidate:
"""A potential revenue leak found by a detector.
metadata_json SHOULD include a stable ``dedupe_key`` (e.g. an invoice id
or deal id) so re-scans match the same underlying entity. If omitted,
the canonical JSON of metadata_json is used as the identity.
"""
detector_id: str
source: str
description: str = ""
estimated_loss: Optional[float] = None
severity: str = "medium"
metadata_json: Dict[str, Any] = field(default_factory=dict)
source_connector: Optional[str] = None
rule_params: Dict[str, Any] = field(default_factory=dict)
def dedupe_key(self) -> str:
"""Stable identity for idempotent upserts."""
key = (self.metadata_json or {}).get("dedupe_key")
if key:
return str(key)
return json.dumps(self.metadata_json or {}, sort_keys=True, default=str)
class BaseDetector(ABC):
"""Abstract base class all leak detectors inherit from.
Subclasses must set the class attributes and implement check().
"""
#: Unique detector name, e.g. 'qb_overdue_30d'
id: str = ""
#: Human-readable name
name: str = ""
#: Connector service this detector depends on (e.g. 'quickbooks'), or
#: None for cross-connector / connector-independent detectors.
connector: Optional[str] = None
#: Whether the detector is enabled when a company has no explicit setting
enabled_by_default: bool = True
#: Human-readable description for the admin UI
description: str = ""
#: Default threshold parameters (overridable per company via settings)
default_params: Dict[str, Any] = {}
@abstractmethod
def check(self, company_id: str) -> List[LeakCandidate]:
"""Inspect company data and return leak candidates (may be empty)."""
raise NotImplementedError
# -- Persistence helpers ---------------------------------------------------
def to_revenue_leak(self, candidate: LeakCandidate, company_id: str):
"""Build a new (unsaved) RevenueLeak model instance from a candidate."""
from app.models import RevenueLeak
severity = candidate.severity if candidate.severity in VALID_SEVERITIES else "medium"
metadata = dict(candidate.metadata_json or {})
metadata.setdefault("dedupe_key", candidate.dedupe_key())
return RevenueLeak(
company_id=company_id,
source=candidate.source,
description=candidate.description or "",
estimated_loss=candidate.estimated_loss,
severity=severity,
metadata_json=metadata,
detection_type="auto",
detector_id=candidate.detector_id,
source_connector=candidate.source_connector or self.connector,
rule_params_json=dict(candidate.rule_params or {}),
)
def _upsert(self, company_id: str, candidate: LeakCandidate) -> Tuple[str, Any]:
"""Idempotently upsert a candidate into revenue_leaks.
Matching: same company + detector_id + dedupe key (from metadata_json).
- Existing UNRESOLVED match → update it in place → ('updated', leak)
- Only RESOLVED matches → create a fresh leak → ('created', leak)
- No match → create → ('created', leak)
Does NOT commit — callers batch-commit after a scan.
"""
from app.models import db, RevenueLeak
key = candidate.dedupe_key()
existing = (
RevenueLeak.query.filter_by(
company_id=company_id,
detection_type="auto",
detector_id=candidate.detector_id,
)
.order_by(RevenueLeak.created_at.desc())
.all()
)
def _leak_key(leak) -> str:
meta = leak.metadata_json or {}
k = meta.get("dedupe_key")
if k:
return str(k)
return json.dumps(meta, sort_keys=True, default=str)
open_match = None
resolved_match = None
for leak in existing:
if _leak_key(leak) != key:
continue
if leak.resolved:
resolved_match = resolved_match or leak
else:
open_match = leak
break
if open_match is not None:
# Refresh the open leak with the latest detection data
open_match.source = candidate.source
open_match.description = candidate.description or ""
open_match.estimated_loss = candidate.estimated_loss
if candidate.severity in VALID_SEVERITIES:
open_match.severity = candidate.severity
metadata = dict(candidate.metadata_json or {})
metadata.setdefault("dedupe_key", key)
open_match.metadata_json = metadata
open_match.rule_params_json = dict(candidate.rule_params or {})
open_match.source_connector = candidate.source_connector or self.connector
return "updated", open_match
leak = self.to_revenue_leak(candidate, company_id)
db.session.add(leak)
status = "created_after_resolved" if resolved_match is not None else "created"
return status, leak