backend/app/observability.py

"""Observability: online metrics/KPIs, an in-process structured log buffer, and
offline metrics publishing.

  • Online tracking   — every processed document updates the DB; KPIs are computed
                        live (OCR completion, straight-through-processing rate, HITL
                        rate, avg confidence, throughput, cost/doc, leading indicators).
  • Online logs       — a ring buffer of structured events surfaced in the Admin tab.
  • Offline publishing— periodic KPI snapshots written to disk (and retained) so the
                        IT team can ship them to a warehouse / BI tool out-of-band.
"""
from __future__ import annotations

import json
import logging
import time
from collections import deque
from pathlib import Path

# --- online structured log ring buffer ---------------------------------------
_LOG = deque(maxlen=500)


def log_event(level: str, message: str, **fields) -> None:
    _LOG.appendleft({"ts": time.time(), "level": level.upper(), "message": message, **fields})


def recent_logs(limit: int = 200, level: str | None = None) -> list[dict]:
    items = list(_LOG)
    if level:
        items = [e for e in items if e["level"] == level.upper()]
    return items[:limit]


class _BufferHandler(logging.Handler):
    def emit(self, record):  # pragma: no cover - thin
        try:
            log_event(record.levelname, record.getMessage(), logger=record.name)
        except Exception:
            pass


def install_log_handler() -> None:
    root = logging.getLogger("aperture")
    if not any(isinstance(h, _BufferHandler) for h in root.handlers):
        root.addHandler(_BufferHandler())
        root.setLevel(logging.INFO)


# --- business KPIs (leading indicators) --------------------------------------
def business_kpis(db, metrics) -> dict:
    docs = db.list_documents(limit=100000)
    n = len(docs)
    if n == 0:
        return {"total_documents": 0, "note": "no documents processed yet"}

    posted = sum(1 for d in docs if d["posted"])
    review = sum(1 for d in docs if d["requires_review"])
    # "completion" = usable text obtained (digital text-layer OR OCR succeeded)
    ocr_ok = sum(1 for d in docs if "no_text_extracted" not in (d.get("flags") or []))
    confs = [d["confidence"] for d in docs if d.get("confidence") is not None]
    low_conf = sum(1 for c in confs if c < 0.85)
    costs = [d.get("total_cost_usd", 0.0) for d in docs]
    times = [d.get("created_at", 0) for d in docs if d.get("created_at")]
    span_min = (max(times) - min(times)) / 60.0 if len(times) > 1 else 0.0

    def by(field):
        out: dict[str, int] = {}
        for d in docs:
            out[d.get(field) or "unknown"] = out.get(d.get(field) or "unknown", 0) + 1
        return out

    return {
        "total_documents": n,
        # leading / completion indicators
        "ocr_completion_rate": round(ocr_ok / n, 3),
        "straight_through_rate": round(posted / n, 3),       # auto-posted, no human
        "hitl_rate": round(review / n, 3),                   # routed to review
        "avg_confidence": round(sum(confs) / len(confs), 3) if confs else None,
        "low_confidence_share": round(low_conf / n, 3),      # leading indicator
        "throughput_docs_per_min": round(n / span_min, 2) if span_min > 0 else None,
        "cost_per_document_usd": round(sum(costs) / n, 6),
        "total_cost_usd": round(sum(costs), 6),
        # distributions
        "by_category": by("category"),
        "by_doc_type": by("doc_type"),
        "by_ocr_backend": by("ocr_backend"),
        # cross-link to inference metrics
        "cache_hit_rate": metrics.summary().get("cache_hit_rate"),
        "inference_savings_pct": metrics.summary().get("savings_pct"),
    }


# --- offline publishing -------------------------------------------------------
def publish_offline_snapshot(db, metrics, out_dir: Path) -> dict:
    out_dir = Path(out_dir)
    out_dir.mkdir(parents=True, exist_ok=True)
    snapshot = {
        "published_at": time.time(),
        "kpis": business_kpis(db, metrics),
        "inference": metrics.summary(),
        "category_counts": db.category_counts(),
    }
    ts = time.strftime("%Y%m%dT%H%M%S")
    path = out_dir / f"kpi_snapshot_{ts}.json"
    path.write_text(json.dumps(snapshot, indent=2))
    # keep a stable 'latest' pointer
    (out_dir / "latest.json").write_text(json.dumps(snapshot, indent=2))
    log_event("info", "offline KPI snapshot published", path=str(path))
    return {"path": str(path), "snapshot": snapshot}


def list_snapshots(out_dir: Path, limit: int = 20) -> list[str]:
    out_dir = Path(out_dir)
    if not out_dir.exists():
        return []
    files = sorted(out_dir.glob("kpi_snapshot_*.json"), reverse=True)
    return [f.name for f in files[:limit]]