#!/usr/bin/env python3
"""Nightly differential-testing run (plan §18 P7): P0 GSA demo + Omni2.5 subset, every result compared with its
oracle by ``simurg compare``, one index page per night plus a history.

Steps (each isolated; a failing step does not stop the others):
  unit       pytest tests/unit (memory-capped)
  gsa_gtc    simurg genotype on the 36 GSA IDATs → compare with the gtc2vcf oracle GTCs (expect PASS)
  gsa_vcf    simurg gtc-to-vcf --unsquash-duplicates → semantic check vs gtc2vcf (scripts/validate_p6_vcf.py,
             expect 0 unexplained) and regression vs the golden VCFs of an earlier night (expect PASS)
  omni_gtc   simurg genotype --preset autocall-1.6.2 on the local Omni2.5 subset → compare with the vendor GTCs

Output: data/nightly/<YYYY-MM-DD>/{index.html, summary.json, <step>/...}, data/nightly/index.html (history).
Usage: scripts/nightly.py [--root DIR] [--keep 14] [--steps unit,gsa_gtc,gsa_vcf,omni_gtc] [--update-golden]
"""

from __future__ import annotations

import argparse
import datetime as dt
import html
import json
import shutil
import subprocess
import sys
import time
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]
PY = ROOT / ".venv" / "bin" / "python"
SIMURG = ROOT / ".venv" / "bin" / "simurg"
GSA = ROOT / "data/p0/gsa_v4_pgx"
GSA_BPM = GSA / "GSA-PGx-48v4-0_20079540_E2_bpm/GSA-PGx-48v4-0_20079540_E2.bpm"
GSA_CSV = GSA / "GSA-PGx-48v4-0_20079540_E2_csv/GSA-PGx-48v4-0_20079540_E2.csv"
GSA_EGT = GSA / "GSA-PGx-48v4-0_20079540_E2_ClusterFile/GSA-PGx-48v4-0_20079540_E2_ClusterFile.egt"
GSA_IDATS = GSA / "idats_Demo_36/idats_Demo_36"
ORACLE_GTCS = ROOT / "data/p0/oracle/gsa_gt3_nofma"
ORACLE_VCF = ROOT / "data/p6/oracle/gsa36_gtc2vcf.vcf.gz"
FASTA = ROOT / "data/p6/GRCh38.primary.fa"
BCFTOOLS = ROOT / "refs/tools/bcftools-1.22/bcftools"
OMNI = ROOT / "data/p0/omni_head"
MEM_CAP = "--as=5500000000"  # 5.5 GB address-space cap per step (the Pi has 8 GB)


def run(cmd: list[str], log: Path, timeout: int = 7200) -> tuple[int, float]:
    t = time.time()
    with log.open("a") as fh:
        fh.write(f"$ {' '.join(cmd)}\n")
        fh.flush()
        try:
            rc = subprocess.run(cmd, stdout=fh, stderr=subprocess.STDOUT, cwd=ROOT, timeout=timeout).returncode
        except subprocess.TimeoutExpired:
            fh.write("TIMEOUT\n")
            rc = 124
    return rc, round(time.time() - t, 1)


def capped(cmd: list[str], cap: str = MEM_CAP) -> list[str]:
    return ["prlimit", cap, *cmd]


# gtc-to-vcf memory-maps the whole genome cache (3.1 GB of file-backed address space that costs page cache, not RAM),
# so its address-space cap must be larger
VCF_CAP = "--as=12000000000"


def compare_verdict(folder: Path) -> dict[str, Any]:
    p = folder / "summary.json"
    if not p.exists():
        return {"verdict": "ERROR"}
    s = json.loads(p.read_text())
    return {"verdict": s["verdict"], "discrepancies": s["discrepancies"], "unknown": s["unknown"],
            "report": str((folder / "dashboard.html").relative_to(folder.parents[1]))}  # fmt: skip


def step_unit(out: Path, log: Path) -> dict[str, Any]:
    rc, sec = run(capped([str(PY), "-m", "pytest", "-q", "-m", "not slow", "tests/unit"]), log, 3600)
    return {"verdict": "PASS" if rc == 0 else "FAIL", "seconds": sec}


def step_gsa_gtc(out: Path, log: Path) -> dict[str, Any]:
    gt = out / "gt"
    rc, sec = run(capped([str(SIMURG), "genotype", "--bpm", str(GSA_BPM), "--egt", str(GSA_EGT), "--idats",
                          str(GSA_IDATS), "--out", str(gt), "--threads", "2", "--quiet", "--force"]), log)  # fmt: skip
    if rc not in (0,):
        return {"verdict": "ERROR", "seconds": sec, "exit": rc}
    _, sec2 = run(capped([str(SIMURG), "compare", str(gt / "samples"), str(ORACLE_GTCS), "--out", str(out / "compare"),
                            "--bpm", str(GSA_BPM), "--threads", "2", "--title", "GSA GTCs vs gtc2vcf oracle",
                            "--quiet", "--force"]), log)  # fmt: skip
    return {**compare_verdict(out / "compare"), "seconds": sec + sec2}


def step_gsa_vcf(out: Path, log: Path, golden: Path, update_golden: bool) -> dict[str, Any]:
    gt = out.parent / "gsa_gtc" / "gt" / "samples"
    if not gt.exists():
        return {"verdict": "SKIPPED", "reason": "gsa_gtc did not run"}
    vcf = out / "vcf"
    rc, sec = run(capped([str(SIMURG), "gtc-to-vcf", "--bpm", str(GSA_BPM), "--csv", str(GSA_CSV), "--fasta",
                          str(FASTA), "--gtcs", str(gt), "--out", str(vcf), "--unsquash-duplicates", "--threads", "2",
                          "--quiet", "--force"], VCF_CAP), log)  # fmt: skip
    if rc != 0:
        return {"verdict": "ERROR", "seconds": sec, "exit": rc}
    res: dict[str, Any] = {"seconds": sec}
    if ORACLE_VCF.exists() and BCFTOOLS.exists():
        _, sec2 = run(capped([str(PY), "scripts/validate_p6_vcf.py", "--ours", str(vcf), "--oracle", str(ORACLE_VCF),
                                "--fasta", str(FASTA), "--bcftools", str(BCFTOOLS), "--bpm", str(GSA_BPM),
                                "--out", str(out / "vs_gtc2vcf.json")]), log)  # fmt: skip
        d = json.loads((out / "vs_gtc2vcf.json").read_text()) if (out / "vs_gtc2vcf.json").exists() else {}
        res["vs_gtc2vcf_unexplained"] = d.get("unexplained", "ERROR")
        res["seconds"] += sec2
    if golden.exists() and not update_golden:
        run(capped([str(SIMURG), "compare", str(golden), str(vcf), "--out", str(out / "regression"), "--threads", "2",
                    "--title", "GSA VCFs vs golden", "--quiet", "--force"]), log)  # fmt: skip
        res["regression"] = compare_verdict(out / "regression")
    else:
        if golden.exists():
            shutil.rmtree(golden)
        golden.mkdir(parents=True)
        for f in vcf.glob("*.snv.vcf.gz"):
            shutil.copy2(f, golden / f.name)
        res["regression"] = {"verdict": "GOLDEN_CREATED"}
    ok = res.get("vs_gtc2vcf_unexplained", 0) == 0 and res["regression"]["verdict"] in ("PASS", "GOLDEN_CREATED")
    res["verdict"] = "PASS" if ok else "FAIL"
    return res


def step_omni_gtc(out: Path, log: Path) -> dict[str, Any]:
    bpm, egt = OMNI / "HumanOmni2.5-4v1-Multi_B.bpm", OMNI / "HumanOmni2.5-4v1-Multi_B.egt"
    if not bpm.exists():
        return {"verdict": "SKIPPED", "reason": "no local Omni subset"}
    idats, vendor = out / "idats", out / "vendor"
    for d in (idats, vendor):
        d.mkdir(parents=True, exist_ok=True)
    for f in sorted(OMNI.rglob("*.idat")):
        (idats / f.name).unlink(missing_ok=True)
        (idats / f.name).symlink_to(f)
    for f in sorted(OMNI.rglob("*.gtc")):
        (vendor / f.name).unlink(missing_ok=True)
        (vendor / f.name).symlink_to(f)
    gt = out / "gt"
    rc, sec = run(capped([str(SIMURG), "genotype", "--bpm", str(bpm), "--egt", str(egt), "--idats", str(idats),
                          "--out", str(gt), "--preset", "autocall-1.6.2", "--threads", "1", "--no-cohort-qc",
                          "--quiet", "--force"]), log)  # fmt: skip
    if rc != 0:
        return {"verdict": "ERROR", "seconds": sec, "exit": rc}
    run(capped([str(SIMURG), "compare", str(gt / "samples"), str(vendor), "--out", str(out / "compare"), "--bpm",
                str(bpm), "--threads", "1", "--title", "Omni2.5 subset vs vendor GTCs (AutoCall 1.6.2.2)",
                "--quiet", "--force"]), log)  # fmt: skip
    return {**compare_verdict(out / "compare"), "seconds": sec}


CSS = (
    "body{font:14px system-ui;margin:16px} td,th{border:1px solid #ccc;padding:3px 8px}"
    " table{border-collapse:collapse} .PASS td:nth-child(2){background:#d6e6f5}"
    " .FAIL td:nth-child(2),.ERROR td:nth-child(2),.DISCREPANCIES td:nth-child(2){background:#f9d3ac}"
)


def render(night: dict[str, Any]) -> str:
    esc = html.escape
    rows = []
    for k, r in night["steps"].items():
        details = json.dumps({x: y for x, y in r.items() if x not in ("verdict", "seconds", "report")})
        link = f'<a href="{esc(r["report"])}">dashboard</a>' if r.get("report") else ""
        verdict = esc(str(r.get("verdict")))
        rows.append(
            f"<tr class='{verdict}'><td>{esc(k)}</td><td>{verdict}</td><td>{r.get('seconds', '')}</td>"
            f"<td>{esc(details)}</td><td>{link}</td></tr>"
        )
    return (
        "<!doctype html><meta charset=utf-8><meta name=viewport content='width=device-width,initial-scale=1'>"
        f"<title>Nightly {night['date']}</title><style>{CSS}</style>"
        f"<h1>Nightly {night['date']} — {esc(night['overall'])}</h1>"
        f"<p>{esc(night['version'])} · {night['seconds']} s</p>"
        "<table><tr><th>step</th><th>verdict</th><th>seconds</th><th>details</th><th></th></tr>"
        + "".join(rows)
        + "</table>"
        + _triage_html(night.get("triage"))
    )


def _triage_html(tri: dict[str, Any] | None) -> str:
    if not tri:
        return ""
    if "error" in tri:
        return f"<p>Triage failed: {html.escape(str(tri['error']))}</p>"
    rows = "".join(
        f"<tr><td>{html.escape(k)}</td><td>{html.escape(str(t['match']))}</td><td>{t['match_confidence']}</td>"
        f"<td>{t['regression']}</td><td>{t['severity']}</td><td>{'yes' if t['needs_person'] else 'no'}</td></tr>"
        for k, t in tri.items()
    )
    return (
        "<h2>Triage</h2><table><tr><th>step</th><th>explained by</th><th>confidence</th><th>regression p</th>"
        f"<th>severity (0 to 3)</th><th>needs a person</th></tr>{rows}</table>"
    )


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--root", type=Path, default=ROOT / "data" / "nightly")
    ap.add_argument("--keep", type=int, default=14, help="nights to keep")
    ap.add_argument("--steps", default="unit,gsa_gtc,gsa_vcf,omni_gtc")
    ap.add_argument("--update-golden", action="store_true")
    a = ap.parse_args()
    day = dt.date.today().isoformat()
    base = a.root / day
    base.mkdir(parents=True, exist_ok=True)
    log = base / "run.log"
    t0 = time.time()
    version = subprocess.run([str(SIMURG), "version"], capture_output=True, text=True, cwd=ROOT).stdout.strip()
    steps: dict[str, Any] = {}
    for name in a.steps.split(","):
        out = base / name
        out.mkdir(exist_ok=True)
        try:
            if name == "unit":
                steps[name] = step_unit(out, log)
            elif name == "gsa_gtc":
                steps[name] = step_gsa_gtc(out, log)
            elif name == "gsa_vcf":
                steps[name] = step_gsa_vcf(out, log, a.root / "golden" / "gsa_vcf", a.update_golden)
            elif name == "omni_gtc":
                steps[name] = step_omni_gtc(out, log)
        except Exception as e:  # a crashing step must not stop the others
            steps[name] = {"verdict": "ERROR", "error": f"{type(e).__name__}: {e}"}
    overall = "PASS" if all(s.get("verdict") in ("PASS", "SKIPPED") for s in steps.values()) else "FAIL"
    night = {"date": day, "version": version, "overall": overall, "seconds": round(time.time() - t0), "steps": steps}
    (base / "summary.json").write_text(json.dumps(night, indent=2) + "\n")
    if overall != "PASS":  # which known finding explains each failing step (scripts/jev, needs the TypeSafe key)
        try:
            sys.path.insert(0, str(ROOT / "scripts" / "jev"))
            from client import api_key
            from nightly_triage import triage

            if api_key():
                night["triage"] = triage(base)
                (base / "summary.json").write_text(json.dumps(night, indent=2) + "\n")
        except Exception as e:  # triage is advice; it must never break the nightly run
            night["triage"] = {"error": f"{type(e).__name__}: {e}"}
    (base / "index.html").write_text(render(night))
    # large intermediates are not kept (reports and summaries are)
    for heavy in ("gsa_gtc/gt/samples", "gsa_vcf/vcf", "omni_gtc/gt/samples"):
        shutil.rmtree(base / heavy, ignore_errors=True)
    nights = sorted(p for p in a.root.iterdir() if p.is_dir() and len(p.name) == 10 and p.name[4] == "-")
    for old in nights[: max(0, len(nights) - a.keep)]:
        shutil.rmtree(old, ignore_errors=True)
    hist = []
    for p in sorted(a.root.glob("????-??-??/summary.json"), reverse=True)[:30]:
        s = json.loads(p.read_text())
        cells = "".join(f"<td>{html.escape(str(s['steps'].get(k, {}).get('verdict', '')))}</td>" for k in steps)
        hist.append(f"<tr><td><a href='{s['date']}/index.html'>{s['date']}</a></td><td>{s['overall']}</td>{cells}</tr>")
    head = "".join(f"<th>{html.escape(k)}</th>" for k in steps)
    (a.root / "index.html").write_text(
        f"<!doctype html><meta charset=utf-8><title>SimurgArray nightly</title><style>{CSS}</style>"
        "<h1>SimurgArray nightly history</h1>"
        f"<table><tr><th>night</th><th>overall</th>{head}</tr>{''.join(hist)}</table>"
    )
    print(json.dumps({"overall": overall, **{k: v.get("verdict") for k, v in steps.items()}}))
    return 0 if overall == "PASS" else 1


if __name__ == "__main__":
    sys.exit(main())