1. Bridge

"""
parquet marts  <->  database rows.

Write direction (`sync_marts`) runs on a machine that has pandas and the raw
tape. Read direction (`get_marts`) runs on every page load and must be cheap.
"""
from __future__ import annotations

import hashlib
import json
import time
from pathlib import Path

import pandas as pd
from django.db import transaction

from lending_analytics import config as C
from lending_analytics import marts as M

from .models import MartRow, QualityCheck, ReportRun, Snapshot

# Which columns of each mart are keys rather than payload.
KEY_MAP: dict[str, dict[str, str]] = {
    "portfolio_monthly":   {"period": "period", "dim_name": "dim_name", "dim_value": "dim_value"},
    "origination_monthly": {"period": "period", "dim_name": "dim_name", "dim_value": "dim_value"},
    "pd_observed":         {"period": "period", "dim_name": "dim_name", "dim_value": "dim_value"},
    "vintages":            {"period": "vintage", "dim_name": "dim_name", "dim_value": "dim_value",
                            "k1": "mob"},
    "vintage_marks":       {"period": "vintage", "dim_name": "dim_name", "dim_value": "dim_value",
                            "k1": "mob", "k2": "metric"},
    "transitions":         {"period": "period", "k1": "from_bucket", "k2": "to_bucket"},
    "cure_rates":          {"period": "period", "k1": "from_bucket", "k2": "outcome"},
    "npl_bridge":          {"period": "period"},
    "client_burden":       {"period": "period", "k1": "burden"},
    "concentration":       {"period": "period"},
    "realised_term":       {"period": "period", "k1": "closure_type"},
    "cost_of_credit":      {"period": "vintage"},
    "insurance_effect":    {"period": "period", "dim_value": "department"},
}

MART_NAMES = tuple(KEY_MAP)


def sha256(path: Path, chunk: int = 1 << 20) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for block in iter(lambda: f.read(chunk), b""):
            h.update(block)
    return h.hexdigest()


def _json_safe(v):
    if v is None or (isinstance(v, float) and pd.isna(v)):
        return None
    if isinstance(v, (pd.Timestamp,)):
        return v.isoformat()
    if hasattr(v, "item"):
        try:
            v = v.item()
        except (ValueError, AttributeError):
            return str(v)
    if isinstance(v, (bool, int, float, str)):
        return v
    if pd.isna(v):
        return None
    return str(v)


# --------------------------------------------------------------------------
# Write
# --------------------------------------------------------------------------
@transaction.atomic
def sync_mart(name: str, df: pd.DataFrame, batch_size: int = 2000) -> int:
    """
    Replace one mart wholesale.

    Delete-then-insert rather than upsert: marts are derived, so a partial
    rebuild is never something you want. One transaction means a failed sync
    leaves the previous mart intact and the site keeps serving.
    """
    keys = KEY_MAP[name]
    payload_cols = [c for c in df.columns if c not in keys.values()]

    rows = []
    for rec in df.to_dict("records"):
        rows.append(MartRow(
            mart=name,
            period=int(rec[keys["period"]]) if pd.notna(rec.get(keys["period"])) else 0,
            dim_name=str(rec.get(keys.get("dim_name", ""), "__all__") or "__all__"),
            dim_value=str(rec.get(keys.get("dim_value", ""), "__all__") or "__all__"),
            k1=("" if "k1" not in keys or pd.isna(rec.get(keys["k1"]))
                else str(_json_safe(rec[keys["k1"]]))),
            k2=("" if "k2" not in keys or pd.isna(rec.get(keys["k2"]))
                else str(_json_safe(rec[keys["k2"]]))),
            data={c: _json_safe(rec[c]) for c in payload_cols},
        ))

    MartRow.objects.filter(mart=name).delete()
    MartRow.objects.bulk_create(rows, batch_size=batch_size)
    return len(rows)


def sync_all(mart_dir: Path | None = None, verbose=print) -> dict[str, int]:
    mart_dir = Path(mart_dir or C.MART_DIR)
    counts = {}
    for name in MART_NAMES:
        p = mart_dir / f"{name}.parquet"
        if not p.exists():
            verbose(f"    skip {name} (not built)")
            continue
        counts[name] = sync_mart(name, pd.read_parquet(p))
        verbose(f"    {name}: {counts[name]:,} rows")

    q = mart_dir / "quality_report.parquet"
    if q.exists():
        qdf = pd.read_parquet(q)
        period = int(pd.read_parquet(mart_dir / "portfolio_monthly.parquet")["period"].max())
        QualityCheck.objects.filter(run_period=period).delete()
        QualityCheck.objects.bulk_create([
            QualityCheck(run_period=period, check_name=r.check, severity=r.severity,
                         status=r.status, detail=str(r.detail))
            for r in qdf.itertuples()])
        counts["quality_report"] = len(qdf)

    bump_version()
    return counts


@transaction.atomic
def register_snapshots(panel: pd.DataFrame) -> int:
    """Record what is in the warehouse, straight from the panel."""
    g = panel.groupby("period", observed=True)
    stats = pd.DataFrame({
        "row_count": g["anketa"].size(),
        "loan_count": g["anketa"].nunique(),
        "exposure_total": g["exposure"].sum(),
        "npl_exposure": g["npl_exposure"].sum(),
    }).reset_index()

    for r in stats.itertuples():
        y, m = divmod(int(r.period), 100)
        report_date = (pd.Timestamp(year=y, month=m, day=1) + pd.offsets.MonthBegin(1)).date()
        Snapshot.objects.update_or_create(
            report_date=report_date,
            defaults=dict(
                period=int(r.period),
                row_count=int(r.row_count),
                loan_count=int(r.loan_count),
                exposure_total=float(r.exposure_total),
                npl_ratio=(float(r.npl_exposure / r.exposure_total)
                           if r.exposure_total else None),
                status="ok",
            ))
    return len(stats)


# --------------------------------------------------------------------------
# Read
# --------------------------------------------------------------------------
_VERSION_FILE = Path(C.MART_DIR) / ".version"


def bump_version() -> str:
    """Cache key for everything derived from the marts."""
    _VERSION_FILE.parent.mkdir(parents=True, exist_ok=True)
    v = str(int(time.time()))
    _VERSION_FILE.write_text(v)
    _read_mart.cache_clear()
    return v


def mart_version() -> str:
    try:
        return _VERSION_FILE.read_text().strip()
    except OSError:
        return "0"


from functools import lru_cache  # noqa: E402


@lru_cache(maxsize=32)
def _read_mart(name: str, version: str) -> pd.DataFrame:
    keys = KEY_MAP[name]
    qs = MartRow.objects.filter(mart=name).values(
        "period", "dim_name", "dim_value", "k1", "k2", "data")
    if not qs:
        return pd.DataFrame()

    rows = []
    for r in qs:
        rec = dict(r.pop("data") or {})
        rec[keys["period"]] = r["period"]
        if "dim_name" in keys:
            rec["dim_name"] = r["dim_name"]
        if "dim_value" in keys:
            rec[keys["dim_value"]] = r["dim_value"]
        for k in ("k1", "k2"):
            if k in keys and r[k] != "":
                rec[keys[k]] = r[k]
        rows.append(rec)

    df = pd.DataFrame(rows)
    # k1/k2 come back as text; restore the numeric ones the charts rely on.
    for col in ("mob", "vintage", "period"):
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors="coerce")
    return df


def get_mart(name: str) -> pd.DataFrame:
    return _read_mart(name, mart_version())


def get_marts() -> dict[str, pd.DataFrame]:
    """Everything the report layer expects, read from the database."""
    return {n: get_mart(n) for n in MART_NAMES}


def latest_period() -> int | None:
    row = MartRow.objects.filter(mart="portfolio_monthly").order_by("-period").first()
    return row.period if row else None


def available_periods() -> list[int]:
    return sorted(MartRow.objects.filter(mart="portfolio_monthly")
                  .values_list("period", flat=True).distinct())