"""Continuous data-quality monitoring (Disease #2 — Dirty Data).

Where dq-api runs Great-Expectations/Soda on *uploaded* files, this agent runs
quality checks continuously against the *live* business tables through Trino and
produces a per-dataset scorecard (0-100) across five dimensions:

  completeness  – non-null ratio across columns
  uniqueness    – distinct/key ratio (duplicate detection)
  validity      – domain rules (non-negative amounts, present timestamps)
  freshness     – age of the newest record vs a threshold
  volume        – row count + delta vs the previous cycle

A rolling score history powers trend sparklines. The loop streams the exact
SQL it runs into the Data Custodian terminal so operators can see the checks.

Endpoints:
  GET  /api/dq/scorecards          -> all datasets, dimensions, score, trend
  GET  /api/dq/scorecard/{key}     -> one dataset detail
  POST /api/dq/run                 -> run one cycle now
  POST /api/dq/config              -> {enabled, interval_s, sample}
"""

from __future__ import annotations

import threading
import time
from collections import deque
from datetime import datetime, timezone
from typing import Any

from fastapi import APIRouter, Body
from fastapi.responses import JSONResponse

from lake_meta import DATASETS, DATASET_BY_KEY, trino, trino_scalar, discover_columns

router = APIRouter(prefix="/api/dq", tags=["data-quality"])

FRESHNESS_THRESHOLD_MIN = 60.0  # newest row older than this => freshness degraded
MAX_COLS = 24

_lock = threading.Lock()
_state: dict[str, Any] = {
    "enabled": True,
    "interval_s": 180.0,
    "sample": 20000,
    "running": False,
    "cycles": 0,
    "last_cycle_ts": 0.0,
    "next_run_ts": 0.0,
    "started_at": None,
    "cards": {},                      # key -> latest scorecard
    "history": {d["key"]: deque(maxlen=60) for d in DATASETS},
    "cols_cache": {},                 # key -> [{name,type}]
    "feed": deque(maxlen=40),
}


def _now() -> datetime:
    return datetime.now(timezone.utc)


def _term(text: str, level: str = "info", phase: str = "dq") -> None:
    try:
        from agent_terminal import emit_threadsafe
        emit_threadsafe("data-custodian", text, level=level, phase=phase)
    except Exception:
        pass


def _feed(text: str, level: str = "info") -> None:
    _state["feed"].appendleft({"ts": _now().isoformat(), "text": text, "level": level})


def _score_color(score: float) -> str:
    if score >= 90:
        return "#34d399"
    if score >= 75:
        return "#fbbf24"
    if score >= 50:
        return "#fb923c"
    return "#f87171"


def _obs_volume_freshness(key: str) -> tuple[int | None, float | None]:
    """Read row count + freshness from the observability monitor (in-memory)."""
    try:
        from observability import _state, _lock
        with _lock:
            st = _state["datasets"].get(key) or {}
            return st.get("rows"), st.get("freshness_age_min")
    except Exception:
        return None, None


def _columns(ds: dict[str, Any]) -> list[dict[str, str]]:
    key = ds["key"]
    cols = _state["cols_cache"].get(key)
    if not cols:
        cols = discover_columns(ds)
        if cols:
            _state["cols_cache"][key] = cols
    return cols or []


def _assess(ds: dict[str, Any], sample: int) -> dict[str, Any]:
    key = ds["key"]
    fq = ds["fqtn"]
    cols = _columns(ds)
    col_names = [c["name"] for c in cols][:MAX_COLS]
    dims: dict[str, Any] = {}
    issues: list[str] = []
    col_profiles: list[dict[str, Any]] = []

    # ── one bounded query: completeness + uniqueness + validity over a sample ──
    has_unique_key = ds.get("unique_key", True)
    key_col = ds.get("key_col") if ds.get("key_col") in col_names else (col_names[0] if col_names else None)
    if not has_unique_key:
        key_col = None  # no single-column unique key (e.g. composite PK) → skip dedup
    ts_col = ds.get("ts_col") if ds.get("ts_col") in col_names else None
    amt_col = ds.get("amount_col") if ds.get("amount_col") in col_names else None

    selects = ["count(*) AS n"]
    for i, c in enumerate(col_names):
        selects.append(f'count("{c}") AS c{i}')
    if key_col:
        selects.append(f'count(DISTINCT "{key_col}") AS dk')
    if amt_col:
        selects.append(f'count_if("{amt_col}" >= 0) AS amt_ok')
    if ts_col:
        selects.append(f'count_if("{ts_col}" IS NOT NULL) AS ts_ok')

    sql = f"SELECT {', '.join(selects)} FROM (SELECT * FROM {fq} LIMIT {sample}) s"
    _term(f"$ trino: profile {key}  ({len(col_names)} cols · sample {sample:,})", level="cmd", phase="dq")
    cols_out, rows = trino(sql, timeout=40.0)
    rec = dict(zip(cols_out, rows[0])) if rows else {}
    n = int(rec.get("n") or 0)

    if n > 0:
        # completeness
        non_null_ratios = []
        for i, c in enumerate(col_names):
            cnt = int(rec.get(f"c{i}") or 0)
            ratio = cnt / n
            non_null_ratios.append(ratio)
            col_profiles.append({"name": c, "completeness": round(ratio * 100, 1),
                                 "nulls": n - cnt})
        completeness = round(100.0 * sum(non_null_ratios) / max(1, len(non_null_ratios)), 1)
        dims["completeness"] = completeness
        worst = sorted(col_profiles, key=lambda x: x["completeness"])[:3]
        for w in worst:
            if w["completeness"] < 95:
                issues.append(f'{w["name"]} {100 - w["completeness"]:.0f}% null')

        # uniqueness / dedup
        if key_col and rec.get("dk") is not None:
            dk = int(rec["dk"])
            uniq = round(100.0 * dk / n, 1)
            dims["uniqueness"] = uniq
            dups = n - dk
            if dups > 0:
                issues.append(f"{dups:,} duplicate {key_col} in sample")

        # validity
        valid_parts = []
        if amt_col and rec.get("amt_ok") is not None:
            valid_parts.append(int(rec["amt_ok"]) / n)
        if ts_col and rec.get("ts_ok") is not None:
            valid_parts.append(int(rec["ts_ok"]) / n)
        if valid_parts:
            validity = round(100.0 * sum(valid_parts) / len(valid_parts), 1)
            dims["validity"] = validity
            if validity < 99 and amt_col:
                issues.append(f"negative/invalid {amt_col}")

    # ── volume + freshness: reuse the observability monitor's in-memory probes
    # (it already polls count(*) and max(ts)). We deliberately do NOT issue our
    # own count/max here — those are expensive full scans on 25-54M-row tables —
    # so a DQ cycle stays fast and the score is driven by the cheap sample. ──
    volume, fresh_age_min = _obs_volume_freshness(key)
    if volume is None:
        volume = n  # sample size until observability reports the real count
    if fresh_age_min is not None:
        fresh_score = 100.0 if fresh_age_min <= FRESHNESS_THRESHOLD_MIN else max(
            0.0, 100.0 - (fresh_age_min - FRESHNESS_THRESHOLD_MIN) / 5.0)
        dims["freshness"] = round(fresh_score, 1)
        if fresh_age_min > FRESHNESS_THRESHOLD_MIN and not ds.get("curated"):
            issues.append(f"stale {fresh_age_min:.0f}m")

    score = round(sum(dims.values()) / len(dims), 1) if dims else 0.0
    prev = _state["cards"].get(key, {})
    prev_vol = prev.get("volume")
    delta = (volume - prev_vol) if (volume is not None and prev_vol is not None) else None

    card = {
        "key": key, "label": ds["label"], "engine": ds["engine"], "color": ds["color"],
        "table": fq, "domain": ds.get("domain"),
        "score": score, "score_color": _score_color(score),
        "dimensions": dims, "volume": volume, "volume_delta": delta,
        "freshness_age_min": round(fresh_age_min, 1) if fresh_age_min is not None else None,
        "issues": issues[:5], "columns": len(col_names),
        "worst_columns": sorted(col_profiles, key=lambda x: x["completeness"])[:5],
        "ts": _now().isoformat(),
    }
    lvl = "ok" if score >= 90 else ("warn" if score >= 60 else "err")
    _term(f"  ← {key}: score {score} · " + " · ".join(f"{k} {v}" for k, v in dims.items())
          + (f" · {len(issues)} issue(s)" if issues else ""), level=lvl, phase="dq")
    return card


def run_cycle() -> dict[str, Any]:
    if _state["running"]:
        return {"ok": True, "skipped": "already running"}
    _state["running"] = True
    sample = int(_state["sample"])
    n_ok = 0
    try:
        _term(f"═══ DQ monitor cycle {_state['cycles'] + 1} — live tables via Trino ═══",
              level="info", phase="cycle")
        for ds in DATASETS:
            if not _state["enabled"]:
                break
            try:
                card = _assess(ds, sample)
            except Exception as exc:
                card = {"key": ds["key"], "label": ds["label"], "engine": ds["engine"],
                        "color": ds["color"], "table": ds["fqtn"], "score": None,
                        "score_color": "#64748b", "dimensions": {}, "error": str(exc)[:160],
                        "issues": [f"check failed: {str(exc)[:80]}"], "ts": _now().isoformat()}
                _term(f"  ✗ {ds['key']}: {str(exc)[:120]}", level="err", phase="dq")
            with _lock:
                _state["cards"][ds["key"]] = card
                if card.get("score") is not None:
                    _state["history"][ds["key"]].append({"t": _now().strftime("%H:%M"), "score": card["score"]})
                    n_ok += 1
        _state["cycles"] += 1
        _state["last_cycle_ts"] = time.time()
        _state["next_run_ts"] = time.time() + float(_state["interval_s"])
        avg = _overall_score()
        _feed(f"cycle {_state['cycles']} — {n_ok}/{len(DATASETS)} datasets scored · platform DQ {avg}")
        _term(f"═══ cycle {_state['cycles']} done — platform DQ score {avg} ═══", level="ok", phase="cycle")
    finally:
        _state["running"] = False
    return {"ok": True, "scored": n_ok, "platform_score": _overall_score()}


def _overall_score() -> float | None:
    vals = [c["score"] for c in _state["cards"].values() if c.get("score") is not None]
    return round(sum(vals) / len(vals), 1) if vals else None


def _loop() -> None:
    _state["started_at"] = _now().isoformat()
    time.sleep(20)
    while True:
        try:
            if _state["enabled"]:
                run_cycle()
        except Exception:
            pass
        time.sleep(max(30.0, float(_state["interval_s"])))


threading.Thread(target=_loop, daemon=True, name="dq-monitor").start()


def scorecards_view() -> dict[str, Any]:
    with _lock:
        cards = []
        for ds in DATASETS:
            c = _state["cards"].get(ds["key"])
            if c:
                cards.append({**c, "trend": list(_state["history"][ds["key"]])})
            else:
                cards.append({"key": ds["key"], "label": ds["label"], "engine": ds["engine"],
                              "color": ds["color"], "table": ds["fqtn"], "score": None,
                              "score_color": "#64748b", "dimensions": {}, "issues": [],
                              "trend": [], "pending": True})
        # platform dimension averages
        dim_avg: dict[str, list[float]] = {}
        for c in cards:
            for k, v in (c.get("dimensions") or {}).items():
                dim_avg.setdefault(k, []).append(v)
        return {
            "ok": True,
            "enabled": _state["enabled"],
            "interval_s": _state["interval_s"],
            "sample": _state["sample"],
            "running": _state["running"],
            "cycles": _state["cycles"],
            "last_cycle_ts": _state["last_cycle_ts"],
            "next_run_ts": _state["next_run_ts"],
            "platform_score": _overall_score(),
            "dimension_averages": {k: round(sum(v) / len(v), 1) for k, v in dim_avg.items()},
            "cards": cards,
            "feed": list(_state["feed"])[:20],
        }


def summary_for_llm() -> dict[str, Any]:
    with _lock:
        return {
            "platform_dq_score": _overall_score(),
            "datasets": [{"key": c["key"], "score": c.get("score"),
                          "issues": c.get("issues", []), "freshness_age_min": c.get("freshness_age_min")}
                         for c in _state["cards"].values()],
        }


@router.get("/scorecards")
async def get_scorecards() -> JSONResponse:
    return JSONResponse(scorecards_view())


@router.get("/scorecard/{key}")
async def get_scorecard(key: str) -> JSONResponse:
    with _lock:
        c = _state["cards"].get(key)
        if not c:
            return JSONResponse({"ok": False, "error": "unknown or not yet scored"}, status_code=404)
        return JSONResponse({"ok": True, "card": {**c, "trend": list(_state["history"].get(key, []))}})


@router.post("/run")
async def post_run() -> JSONResponse:
    from starlette.concurrency import run_in_threadpool
    res = await run_in_threadpool(run_cycle)
    return JSONResponse({**res, "scorecards": scorecards_view()})


@router.post("/config")
async def post_config(body: dict = Body(default={})) -> JSONResponse:
    if "enabled" in body:
        _state["enabled"] = bool(body["enabled"])
    if "interval_s" in body:
        try:
            _state["interval_s"] = max(30.0, min(1800.0, float(body["interval_s"])))
        except Exception:
            pass
    if "sample" in body:
        try:
            _state["sample"] = max(1000, min(200000, int(body["sample"])))
        except Exception:
            pass
    return JSONResponse({"ok": True, "enabled": _state["enabled"],
                         "interval_s": _state["interval_s"], "sample": _state["sample"]})