""" data_source.py ────────────── Pluggable storage layer for the Fraud Report module. The abstract `FraudRepository` defines the contract that the API expects. `JsonFraudRepository` implements it on top of the legacy `data/fraud_log.json` file. To migrate to a database later, write a new subclass that fulfils the same contract (e.g. `PostgresFraudRepository`, `MongoFraudRepository`) and swap the `get_repository()` factory in `fraudreport.py`. The route layer, response models, and the frontend do NOT change. """ from __future__ import annotations import json import logging import os from abc import ABC, abstractmethod from pathlib import Path from typing import Iterable, Optional from .models import FraudEvent log = logging.getLogger("fraud_report.data_source") # ── Domain exceptions ─────────────────────────────────────────────── # # Routes translate these to HTTP statuses (404 / 422 / 500). Keeping # them inside this module means every repository (JSON, SQL, Mongo…) # raises the same exception types — so error handling never has to # branch on the backing store. class FraudDataError(Exception): """Base class for fraud-data problems.""" class FraudDataNotFoundError(FraudDataError): """Underlying store is missing (e.g. JSON file not found).""" class FraudDataCorruptError(FraudDataError): """Underlying store can't be parsed (e.g. invalid JSON).""" class FraudDataEmptyError(FraudDataError): """Store exists and is parseable but contains zero rows.""" class AccessKeyNotFoundError(FraudDataError): """The requested access key has no events.""" # ── Abstract repository ───────────────────────────────────────────── class FraudRepository(ABC): """Contract every fraud-data backend must satisfy. The methods are deliberately narrow — just what the API needs. Don't add helpers here unless every backend can implement them cheaply, otherwise it leaks JSON-specific assumptions into the abstract layer. """ @abstractmethod def list_access_keys(self) -> list[str]: """All distinct access keys, sorted, most-recent-event first.""" @abstractmethod def fetch_events(self, access_key: Optional[str] = None) -> list[FraudEvent]: """All events, optionally filtered by access key. Returns an empty list if the key has no events — does NOT raise. Callers (the service layer) decide whether empty → 404. """ # ── JSON-file backend ─────────────────────────────────────────────── class JsonFraudRepository(FraudRepository): """Reads fraud events from a JSON array on disk. Designed to handle the messy reality of the existing `data/fraud_log.json`: • Some rows are missing `access_key` entirely (we resolve them via `metadata.access_key` or `candidate_id`). • Timestamps appear in two formats (ISO with Z, ISO with +00:00). • Rows can be reordered freely — sorting happens at the service layer, not here. """ def __init__(self, path: str | os.PathLike): self.path = Path(path) self._cache: list[FraudEvent] | None = None self._cache_mtime: float | None = None # Public ---------------------------------------------------------------- def list_access_keys(self) -> list[str]: events = self._load() # Use the resolved access_key (handles legacy rows where the # top-level key was missing but metadata had it). seen: dict[str, str] = {} # access_key → last_event_at for ev in events: ak = self._resolve_access_key(ev) if not ak: continue ts = ev.created_at or ev.timestamp or "" if ak not in seen or ts > seen[ak]: seen[ak] = ts return sorted(seen.keys(), key=lambda k: seen[k], reverse=True) def fetch_events(self, access_key: Optional[str] = None) -> list[FraudEvent]: events = self._load() if access_key is None: return events ak = access_key.strip() if not ak: return [] return [e for e in events if self._resolve_access_key(e) == ak] # Internal -------------------------------------------------------------- def _load(self) -> list[FraudEvent]: """Read & cache the JSON file. Hot-reloads when the mtime changes — useful in development. In production with a real DB the caching/invalidation strategy lives in that repository.""" if not self.path.exists(): raise FraudDataNotFoundError( f"fraud_log file not found at {self.path}" ) try: mtime = self.path.stat().st_mtime except OSError as e: raise FraudDataNotFoundError(f"cannot stat {self.path}: {e}") from e if self._cache is not None and self._cache_mtime == mtime: return self._cache try: raw = self.path.read_text(encoding="utf-8") except OSError as e: raise FraudDataNotFoundError(f"cannot read {self.path}: {e}") from e if not raw.strip(): raise FraudDataEmptyError(f"{self.path} is empty") try: parsed = json.loads(raw) except json.JSONDecodeError as e: raise FraudDataCorruptError( f"invalid JSON in {self.path}: {e.msg} at line {e.lineno}" ) from e if not isinstance(parsed, list): raise FraudDataCorruptError( f"{self.path} root must be a JSON array, got {type(parsed).__name__}" ) if not parsed: raise FraudDataEmptyError(f"{self.path} contains zero records") events: list[FraudEvent] = [] for idx, row in enumerate(parsed): if not isinstance(row, dict): log.warning("skipping non-object row #%d in fraud_log.json", idx) continue try: events.append(FraudEvent(**row)) except Exception as e: # noqa: BLE001 — be tolerant of one bad row log.warning("skipping malformed fraud event #%d: %s", idx, e) self._cache = events self._cache_mtime = mtime return events @staticmethod def _resolve_access_key(ev: FraudEvent) -> Optional[str]: """Real-world fix-up: the top-level `access_key` is sometimes absent (notably on DISQUALIFIED rows) — pull it from metadata or fall back to candidate_id, since this codebase treats those as identical in practice.""" if ev.access_key: return ev.access_key meta = ev.metadata or {} if isinstance(meta, dict) and meta.get("access_key"): return str(meta["access_key"]) if ev.candidate_id: return ev.candidate_id return None # ── Factory ──────────────────────────────────────────────────────── # # Centralised so swapping to a DB backend is a one-line change. # Production deployments can override the path via env var: # FRAUD_LOG_PATH=/var/lib/peoplehub/fraud_log.json def get_repository() -> FraudRepository: """Return the configured FraudRepository implementation. Today: JSON file. Tomorrow: dispatch on env (`FRAUD_BACKEND=mongo`) and return the matching repository instance. """ default_path = ( Path(__file__).resolve().parent.parent / "data" / "fraud_log.json" ) path = os.getenv("FRAUD_LOG_PATH", str(default_path)) return JsonFraudRepository(path=path)