"""Check that the Python engine (doctor/) and the JavaScript engine (web/doctor.js) agree.

    python tests/parity.py            # needs node on PATH; REAL_DATA=<dir> adds the sklearn CSVs

For each case it runs `node tests/run_js.js FILE --json` and doctor.summary() and compares rows,
score, columns (name, type, missing, unique), issues (type, column, severity, in order) and
feature strength (same order, |difference| <= 0.001). It also compares the full issue text
(title, detail, fix, evidence), the Markdown report, the cleaning script and the cleaned CSV
written by apply_fixes, which must all be identical. Besides the demo, sample and real datasets it
runs messy synthetic files from tests/edge_cases.py. Exits non-zero on any mismatch.
"""
import json
import os
import subprocess
import sys
import tempfile
import time

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, ROOT)
import doctor  # noqa: E402
sys.path.insert(0, os.path.join(ROOT, "tests"))
from edge_cases import AUTO as EDGE_AUTO, write_edge_cases  # noqa: E402

DATA = os.path.join(ROOT, "data")
REAL_DATA = os.environ.get("REAL_DATA") or os.path.join(os.path.dirname(os.path.dirname(ROOT)), "realdata")

# Full report details from the JS engine (run_js.js only writes the summary).
JS_DETAILS = r"""
const fs = require("fs"), path = require("path");
const D = require(path.join(process.argv[1], "web", "doctor.js"));
const [file, target, compare, today, filename] = process.argv.slice(2);
const opt = {};
if (target !== "@auto") opt.target = target === "@none" ? null : target;
if (compare) opt.compare = D.parseCSV(fs.readFileSync(compare, "utf8"));
if (today) opt.today = today;
const t = D.parseCSV(fs.readFileSync(file, "utf8"));
const t0 = process.hrtime.bigint();
const r = D.analyze(t, opt);
const ms = Number(process.hrtime.bigint() - t0) / 1e6;
const out = {
  ms: ms,
  issues: r.issues.map(i => ({ id: i.id, type: i.type, column: i.column, severity: i.severity, penalty: i.penalty,
    affected: i.affected, title: i.title, detail: i.detail, fix: i.fix, evidence: i.evidence || null })),
  penalties: r.score.penalties,
  strength: r.featureStrength.map(s => [s.column, s.score]),
  script: D.cleaningScript(r, null, { filename: filename }),
  markdown: D.reportMarkdown(r, { filename: filename }),
  cleaned: D.toCSV(D.applyFixes(t, r, null))
};
process.stdout.write(JSON.stringify(out));
"""


class Mismatch(Exception):
    pass


def run_js_summary(path, target, compare, today, tmpdir):
    out = os.path.join(tmpdir, "js.json")
    cmd = ["node", os.path.join(ROOT, "tests", "run_js.js"), path]
    if target is not doctor.AUTO:
        cmd += ["--target", "none" if target is None else target]
    if compare:
        cmd += ["--compare", compare]
    if today:
        cmd += ["--today", today]
    cmd += ["--json", out]
    subprocess.run(cmd, check=True, capture_output=True, cwd=ROOT)
    with open(out, encoding="utf-8") as f:
        return json.load(f)


def run_js_details(path, target, compare, today, filename):
    t = "@auto" if target is doctor.AUTO else "@none" if target is None else target
    res = subprocess.run(["node", "-e", JS_DETAILS, ROOT, path, t, compare or "", today or "", filename],
                         check=True, capture_output=True, cwd=ROOT)
    return json.loads(res.stdout.decode("utf-8"))


def diff_text(name, a, b):
    if a == b:
        return []
    al, bl = a.splitlines(), b.splitlines()
    for k in range(max(len(al), len(bl))):
        x = al[k] if k < len(al) else "<missing>"
        y = bl[k] if k < len(bl) else "<missing>"
        if x != y:
            return [f"{name} differs at line {k + 1}:\n      js: {x!r}\n      py: {y!r}"]
    return [f"{name} differs (line endings / trailing text)"]


def compare_case(label, path, target=doctor.AUTO, compare=None, today=None, tmpdir=None):
    errors = []
    js = run_js_summary(path, target, compare, today, tmpdir)
    t0 = time.perf_counter()
    rep = doctor.analyze(path, target=target, compare=compare, today=today)
    py_ms = (time.perf_counter() - t0) * 1000
    py = json.loads(json.dumps(doctor.summary(rep)))

    for key in ("rows", "score"):
        if js[key] != py[key]:
            errors.append(f"{key}: js {js[key]} vs py {py[key]}")
    if js["columns"] != py["columns"]:
        for k in range(max(len(js["columns"]), len(py["columns"]))):
            a = js["columns"][k] if k < len(js["columns"]) else None
            b = py["columns"][k] if k < len(py["columns"]) else None
            if a != b:
                errors.append(f"column {k}: js {a} vs py {b}")
    if js["issues"] != py["issues"]:
        for k in range(max(len(js["issues"]), len(py["issues"]))):
            a = js["issues"][k] if k < len(js["issues"]) else None
            b = py["issues"][k] if k < len(py["issues"]) else None
            if a != b:
                errors.append(f"issue {k}: js {a} vs py {b}")
    js_fs, py_fs = js["featureStrength"], py["featureStrength"]
    if [s["column"] for s in js_fs] != [s["column"] for s in py_fs]:
        errors.append(f"featureStrength order: js {[s['column'] for s in js_fs]} vs py {[s['column'] for s in py_fs]}")
    else:
        for a, b in zip(js_fs, py_fs):
            if (a["score"] is None) != (b["score"] is None) or (a["score"] is not None and abs(a["score"] - b["score"]) > 0.001):
                errors.append(f"featureStrength {a['column']}: js {a['score']} vs py {b['score']}")

    # full text, script, markdown and cleaned CSV
    fname = os.path.basename(path)
    det = run_js_details(path, target, compare, today, fname)
    pyi = rep["issues"]
    for k, a in enumerate(det["issues"]):
        if k >= len(pyi):
            break
        b = pyi[k]
        for field in ("id", "title", "detail", "fix", "penalty"):
            if a[field] != b.get(field):
                errors.append(f"issue {k} ({a['type']} {a['column']}) {field}:\n      js: {a[field]!r}\n      py: {b.get(field)!r}")
        if a["evidence"] != b.get("evidence") and not (a["evidence"] is None and b.get("evidence") is None):
            errors.append(f"issue {k} ({a['type']}) evidence: js {a['evidence']} vs py {b.get('evidence')}")
        if abs(a["affected"] - b["affected"]) > 1e-12:
            errors.append(f"issue {k} ({a['type']}) affected: js {a['affected']} vs py {b['affected']}")
    for c, v in det["penalties"].items():
        if abs(v - rep["score"]["penalties"].get(c, 0)) > 1e-9:
            errors.append(f"penalty {c}: js {v} vs py {rep['score']['penalties'].get(c)}")
    for (cj, sj), s in zip(det["strength"], rep["featureStrength"]):
        if sj is not None and s["score"] == s["score"] and abs(sj - s["score"]) > 1e-9:
            errors.append(f"raw strength {cj}: js {sj!r} vs py {s['score']!r}")
    errors += diff_text("cleaning script", det["script"], doctor.cleaning_script(rep, filename=fname))
    errors += diff_text("markdown report", det["markdown"], doctor.report_markdown(rep, filename=fname))
    cleaned = doctor.to_csv(doctor.apply_fixes(doctor.read_csv(path), rep))
    if cleaned != det["cleaned"]:
        errors += diff_text("apply_fixes CSV", det["cleaned"], cleaned) or ["apply_fixes CSV differs"]
    return errors, det["ms"], py_ms, cleaned


def main():
    try:
        subprocess.run(["node", "--version"], check=True, capture_output=True)
    except (OSError, subprocess.CalledProcessError):
        print("node is not available; can't run the parity check")
        return 2
    train, test = os.path.join(DATA, "demo_train.csv"), os.path.join(DATA, "demo_test.csv")
    cases = [
        ("demo_train (target churned)", train, "churned", None, "2026-10-14"),
        ("demo_train vs demo_test", train, "churned", test, None),
        ("demo_train (auto target, no today)", train, doctor.AUTO, None, None),
        ("demo_train (no target)", train, None, None, None),
        ("sample.csv (auto target)", os.path.join(DATA, "sample.csv"), doctor.AUTO, None, None),
    ]
    if os.path.isdir(REAL_DATA):
        for f in ["iris", "wine", "breast_cancer", "diabetes"]:
            p = os.path.join(REAL_DATA, f + ".csv")
            if os.path.exists(p):
                cases.append((f + ".csv (target target)", p, "target", None, None))
    failed = 0
    with tempfile.TemporaryDirectory() as tmp:
        for label, path, target, compare in write_edge_cases(os.path.join(tmp, "edge")):
            cases.append((label, path, doctor.AUTO if target == EDGE_AUTO else target, compare, "2026-10-14"))
        cleaned_path = os.path.join(tmp, "demo_train_clean.csv")
        k = 0
        while k < len(cases):
            label, path, target, compare, today = cases[k]
            errors, js_ms, py_ms, cleaned = compare_case(label, path, target, compare, today, tmp)
            if k == 0:
                with open(cleaned_path, "w", encoding="utf-8", newline="") as f:
                    f.write(cleaned)
                cases.append(("demo_train cleaned by apply_fixes (target churned)", cleaned_path, "churned", None, None))
            status = "ok  " if not errors else "FAIL"
            print(f"{status} {label}   (js {js_ms:.0f} ms, py {py_ms:.0f} ms)")
            for e in errors[:25]:
                print("     " + e)
            failed += bool(errors)
            k += 1
    print(f"{len(cases) - failed}/{len(cases)} cases identical")
    return 1 if failed else 0


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