"""Shared helpers for Phase 1 leak detectors."""
from __future__ import annotations
from datetime import datetime, timezone
from typing import Optional
#: CRM deal stages considered closed (won or lost) — excluded from open-deal scans.
CLOSED_STAGES = {
'Closed Won', 'Closed Lost',
'closedwon', 'closedlost',
'closed_won', 'closed_lost',
'won', 'lost',
}
def utcnow() -> datetime:
return datetime.now(timezone.utc)
def as_aware(dt: Optional[datetime]) -> Optional[datetime]:
"""Coerce a (possibly naive) DB datetime to UTC-aware for arithmetic."""
if dt is None:
return None
if dt.tzinfo is None:
return dt.replace(tzinfo=timezone.utc)
return dt
def days_since(dt: Optional[datetime]) -> int:
"""Whole days elapsed since ``dt`` (0 if dt is None/future)."""
aware = as_aware(dt)
if aware is None:
return 0
delta = utcnow() - aware
return max(delta.days, 0)
def connector_available(company_id: str, service: str) -> bool:
"""True if the company has a connected connector for ``service``."""
from app.models import Connector
return (
Connector.query.filter_by(
company_id=company_id, service=service, status='connected'
).first()
is not None
)
def any_connector_available(company_id: str, services) -> bool:
"""True if any of the listed connector services is connected."""
from app.models import Connector
return (
Connector.query.filter(
Connector.company_id == company_id,
Connector.service.in_(list(services)),
Connector.status == 'connected',
).first()
is not None
)
def all_connectors_available(company_id: str, services) -> bool:
"""True if ALL of the listed connector services are connected."""
return all(connector_available(company_id, s) for s in services)
# Spec-compat alias used by multi-connector (Phase 3) detectors.
_connectors_available = all_connectors_available
def normalize_phone(phone) -> str:
"""Normalize a phone number to its last 10 digits for fuzzy matching."""
if not phone:
return ""
digits = "".join(ch for ch in str(phone) if ch.isdigit())
return digits[-10:] if len(digits) >= 10 else digits
def parse_dt(value):
"""Best-effort parse of an ISO-ish timestamp from properties_json."""
from datetime import datetime
if value is None:
return None
if isinstance(value, datetime):
return as_aware(value)
if isinstance(value, (int, float)):
try:
# Heuristic: epoch millis vs seconds
ts = float(value)
if ts > 1e12:
ts /= 1000.0
return datetime.fromtimestamp(ts, tz=timezone.utc)
except (ValueError, OSError, OverflowError):
return None
if isinstance(value, str):
try:
return as_aware(datetime.fromisoformat(value.replace("Z", "+00:00")))
except ValueError:
return None
return None
def deal_weighted_value(amount: Optional[float], probability: Optional[float]) -> float:
"""Expected value of a deal: amount × probability (defaults: 0 / 0.5)."""
amt = amount or 0.0
prob = probability if probability is not None else 0.5
# Clamp probability to sane range
prob = min(max(prob, 0.0), 1.0)
return round(amt * prob, 2)