| """ |
| Structured call-context engine. |
| |
| Kept fully SEPARATE from the neural model: it consumes only business/context |
| fields (caller ID, transaction, account) and produces a deterministic |
| context_risk (0-100) plus explainable signals. It never touches audio. |
| |
| HEURISTIC ONLY: the risk_category mapping is a demo rule set, not a validated |
| model. Nothing here is claimed to be scientifically validated. |
| """ |
|
|
| import threading |
| import uuid |
| import time |
| from dataclasses import asdict, dataclass, field |
| from typing import Any, Dict, List, Optional |
|
|
| from ..config import settings |
|
|
| TRANSACTION_TYPES = ["general_transfer", "high_value_transfer", "account_admin", "credit_card", "other"] |
| RISK_CATEGORIES = ["low", "medium", "high"] |
|
|
|
|
| @dataclass |
| class CallContext: |
| call_id: str = "" |
| caller_id: str = "" |
| known_contact: Optional[bool] = None |
| transaction_type: str = "general_transfer" |
| transaction_amount: float = 0.0 |
| risk_category: str = "low" |
| account_context: str = "" |
| note: str = "" |
|
|
| def to_dict(self) -> Dict[str, Any]: |
| return asdict(self) |
|
|
|
|
| class ContextEngine: |
| """Per-connection call context store + heuristic context risk.""" |
|
|
| def __init__(self): |
| self._lock = threading.RLock() |
| self._contexts: Dict[str, CallContext] = {} |
|
|
| def create_context(self, connection_id: str) -> CallContext: |
| ctx = CallContext( |
| call_id=str(uuid.uuid4()), |
| caller_id="", |
| known_contact=None, |
| transaction_type="general_transfer", |
| transaction_amount=0.0, |
| risk_category="low", |
| ) |
| with self._lock: |
| self._contexts[connection_id] = ctx |
| return ctx |
|
|
| def get(self, connection_id: str) -> Optional[CallContext]: |
| with self._lock: |
| return self._contexts.get(connection_id) |
|
|
| def get_or_create(self, connection_id: str) -> CallContext: |
| ctx = self.get(connection_id) |
| if ctx is None: |
| ctx = self.create_context(connection_id) |
| return ctx |
|
|
| def update( |
| self, |
| connection_id: str, |
| caller_id: Optional[str] = None, |
| known_contact: Optional[bool] = None, |
| transaction_type: Optional[str] = None, |
| transaction_amount: Optional[float] = None, |
| risk_category: Optional[str] = None, |
| account_context: Optional[str] = None, |
| note: Optional[str] = None, |
| ) -> CallContext: |
| ctx = self.get_or_create(connection_id) |
| if caller_id is not None: |
| ctx.caller_id = str(caller_id) |
| if known_contact is not None: |
| ctx.known_contact = bool(known_contact) |
| if transaction_type is not None: |
| ctx.transaction_type = transaction_type if transaction_type in TRANSACTION_TYPES else "other" |
| if transaction_amount is not None: |
| ctx.transaction_amount = float(transaction_amount) |
| if risk_category is not None: |
| ctx.risk_category = risk_category if risk_category in RISK_CATEGORIES else "low" |
| if account_context is not None: |
| ctx.account_context = str(account_context) |
| if note is not None: |
| ctx.note = str(note) |
| return ctx |
|
|
| def get_or_create_with_callid(self, connection_id: str, call_id: str) -> CallContext: |
| ctx = self.get_or_create(connection_id) |
| if call_id: |
| ctx.call_id = str(call_id) |
| return ctx |
|
|
| def context_risk(self, connection_id: str) -> Dict[str, Any]: |
| """Deterministic, explainable context risk (0-100) + signals.""" |
| ctx = self.get_or_create(connection_id) |
| signals: List[str] = [] |
| base = {"low": 5.0, "medium": 40.0, "high": 80.0}.get(ctx.risk_category, 5.0) |
|
|
| amount_risk = 0.0 |
| if ctx.transaction_amount > 0: |
| amount_risk = min(50.0, 100.0 * ctx.transaction_amount / max(settings.context_high_value_threshold, 1e-6)) |
| if ctx.transaction_amount > settings.context_high_value_threshold: |
| signals.append("HIGH_VALUE_TRANSACTION") |
|
|
| contact_risk = 0.0 |
| if ctx.known_contact is False: |
| contact_risk = float(settings.context_unknown_contact_risk) |
| signals.append("UNKNOWN_CALLER") |
| elif ctx.known_contact is True: |
| signals.append("KNOWN_CALLER") |
|
|
| if ctx.transaction_type in ("account_admin", "credit_card"): |
| base = min(100.0, base + 10.0) |
| signals.append(f"SENSITIVE_TRANSACTION_TYPE:{ctx.transaction_type}") |
|
|
| raw = base + 0.5 * amount_risk + contact_risk |
| risk = min(100.0, raw) |
|
|
| return { |
| "context_risk": round(risk, 2), |
| "signals": signals, |
| "context": ctx.to_dict(), |
| } |
|
|
| def remove(self, connection_id: str) -> None: |
| with self._lock: |
| self._contexts.pop(connection_id, None) |
|
|
| def reset(self, connection_id: str) -> CallContext: |
| with self._lock: |
| return self.create_context(connection_id) |
|
|
| def list_contexts(self) -> Dict[str, CallContext]: |
| with self._lock: |
| return dict(self._contexts) |
