"""Independent reader for a Base1 export. Uses only the Python standard library and DuckDB.

    python verify_export.py EXPORT.zip [--api-result API.json] [--baseline BASELINE.json] [--json OUT.json]

Checks, in order:
  1. every file in MANIFEST.sha256 matches, and no file is missing from the manifest
  2. original files match the sha256 recorded in raw_files
  3. runs the export's own query/workflow_history.sql on the export's Parquet tables
  4. the full query result equals the API result captured for the same state
  5. for the baseline source, every baseline event is returned exactly once with identical
     content, and no other event of that source appears (checked on the export and on the API result)
Exit code 0 only if every requested check passes.
"""
import argparse
import hashlib
import json
import sys
import tempfile
import zipfile
from decimal import Decimal
from pathlib import Path

import duckdb

TABLES = ["sources", "raw_files", "events", "assignments"]


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


def canonical(row):
    return json.dumps({k: (str(v) if isinstance(v, Decimal) else v) for k, v in row.items()},
                      sort_keys=True, ensure_ascii=False, separators=(",", ":"))


def check_manifest(root):
    listed, bad = {}, []
    for line in (root / "MANIFEST.sha256").read_text().splitlines():
        digest, rel = line.split("  ", 1)
        listed[rel] = digest
    for rel, digest in listed.items():
        p = root / rel
        if not p.is_file() or sha256(p) != digest:
            bad.append(rel)
    present = {p.relative_to(root).as_posix() for p in root.rglob("*") if p.is_file()} - {"MANIFEST.sha256"}
    return len(listed), bad, sorted(present - set(listed))


def compare_history(rows, baseline):
    """Baseline events for one source versus the same source's rows in a result."""
    sid = baseline["source_id"]
    ref = {e["event_id"]: canonical(e) for e in baseline["events"]}
    got = {}
    for r in rows:
        if r["source_id"] == sid:
            got.setdefault(r["event_id"], []).append(canonical(r))
    preserved = sum(1 for eid, c in ref.items() if got.get(eid) == [c])
    missing = sorted(eid for eid in ref if eid not in got)
    duplicated = sorted(eid for eid, v in got.items() if len(v) > 1)
    changed = sorted(eid for eid, c in ref.items() if eid in got and len(got[eid]) == 1 and got[eid][0] != c)
    extra = sorted(eid for eid in got if eid not in ref)
    ok = len(ref) > 0 and preserved == len(ref) and not (missing or duplicated or changed or extra)
    return {"source_id": sid, "preserved": preserved, "baseline_events": len(ref), "missing": missing,
            "duplicated": duplicated, "changed": changed, "extra": extra, "pass": ok}


def main(argv=None):
    ap = argparse.ArgumentParser()
    ap.add_argument("export")
    ap.add_argument("--api-result")
    ap.add_argument("--baseline")
    ap.add_argument("--workflow-id")
    ap.add_argument("--json")
    a = ap.parse_args(argv)
    report = {"export": Path(a.export).name, "checks": {}}
    say = print
    with tempfile.TemporaryDirectory() as tmp:
        root = Path(tmp)
        with zipfile.ZipFile(a.export) as z:
            z.extractall(root)

        n, bad, unlisted = check_manifest(root)
        report["checks"]["manifest"] = {"files": n, "mismatched": bad, "unlisted": unlisted,
                                        "pass": not bad and not unlisted}
        say(f"Manifest            {n} files, {len(bad)} mismatched, {len(unlisted)} unlisted")

        con = duckdb.connect()
        for t in TABLES:
            con.execute(f"CREATE TABLE {t} AS SELECT * FROM read_parquet(?)", [str(root / "tables" / f"{t}.parquet")])
        orig_bad = [rel for rel, digest in con.execute("SELECT relative_path, sha256 FROM raw_files").fetchall()
                    if not (root / "original" / rel).is_file() or sha256(root / "original" / rel) != digest]
        report["checks"]["originals"] = {"mismatched": orig_bad, "pass": not orig_bad}
        say(f"Original files      {con.execute('SELECT count(*) FROM raw_files').fetchall()[0][0]} recorded, {len(orig_bad)} mismatched")

        wf = a.workflow_id or json.loads((root / "config" / "replay.json").read_text())["workflow_id"]
        cur = con.execute((root / "query" / "workflow_history.sql").read_text(), {"workflow_id": wf})
        cols = [d[0] for d in cur.description]
        rows = [dict(zip(cols, r)) for r in cur.fetchall()]
        report["export_rows"] = len(rows)
        say(f"Same SQL, export    {len(rows)} events for {wf}")

        if a.api_result:
            api = json.loads(Path(a.api_result).read_text())
            same = [canonical(r) for r in rows] == [canonical(r) for r in api]
            report["checks"]["export_equals_api"] = {"export_rows": len(rows), "api_rows": len(api), "pass": same}
            say(f"Export vs API       {len(rows)} vs {len(api)} events, {'identical' if same else 'DIFFERENT'}")

        if a.baseline:
            base = json.loads(Path(a.baseline).read_text())
            for label, data in [("export", rows)] + ([("api", api)] if a.api_result else []):
                c = compare_history(data, base)
                report["checks"][f"baseline_vs_{label}"] = c
                detail = "" if c["pass"] else (f" missing {len(c['missing'])}, changed {len(c['changed'])},"
                                               f" duplicated {len(c['duplicated'])}, extra {len(c['extra'])}")
                say(f"Baseline vs {label:<7} {c['preserved']}/{c['baseline_events']} events of {c['source_id']} "
                    f"unchanged{detail}")

    ok = all(c["pass"] for c in report["checks"].values())
    report["pass"] = ok
    say("Result              " + ("PASS" if ok else "FAIL"))
    if a.json:
        Path(a.json).write_text(json.dumps(report, indent=2) + "\n")
    return 0 if ok else 1


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