#!/usr/bin/env python3
"""Rebuilds the evidence for /blog/what-a-tied-out-statement-tells-you.

  pages  every result with an http(s) url is fetched and its quote looked for in the page text
  lab    24 statements are regenerated from a fixed seed, errors are injected at every row under
         three ways of printing balances, and the balance rule is run on each case
  ts     when this file sits in the Evidilo repository and node 22 or newer is on PATH, every lab
         case is also run through the product's own TypeScript rule (packages/shared/src/chain.ts
         and packages/parsers/src/invariants/balance-chain.ts) and must give the same verdict

Exit 0 when every quote is still on its page, every measurement matches dataset.json and
dataset.csv, and the TypeScript rule agrees; 1 otherwise. `--recheck` re-reads the pages and
skips any lab step whose recorded run took longer than a minute. `--write` rewrites the lab
numbers in dataset.json and dataset.csv from a fresh run.
"""

from __future__ import annotations

import csv
import gzip
import html
import json
import random
import re
import shutil
import statistics
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.request
from pathlib import Path

HERE = Path(__file__).resolve().parent
DATASET = HERE / "dataset.json"
TABLE = HERE / "dataset.csv"
REPO = HERE.parents[4] if len(HERE.parents) > 4 else HERE

UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0 Safari/537.36"
TIMEOUT = 45
LAB_BUDGET_SECONDS = 60

# ---------------------------------------------------------------- page text


def normalise(text: str) -> str:
    text = re.sub(r"<script.*?</script>|<style.*?</style>", " ", text, flags=re.S | re.I)
    text = re.sub(r"<[^>]+>", " ", text)
    text = html.unescape(text)
    for fancy, plain in (("‘", "'"), ("’", "'"), ("“", '"'), ("”", '"'), ("\u2013", "-"), ("\u2014", "-"), (" ", " ")):
        text = text.replace(fancy, plain)
    return re.sub(r"\s+", " ", text).strip().lower()


def fetch(url: str) -> str:
    request = urllib.request.Request(url, headers={"User-Agent": UA, "Accept-Encoding": "identity"})
    try:
        response = urllib.request.urlopen(request, timeout=TIMEOUT)
    except urllib.error.HTTPError as error:
        # web.archive.org answers 429 to a second request in quick succession.
        if error.code != 429:
            raise
        time.sleep(10)
        response = urllib.request.urlopen(request, timeout=TIMEOUT)
    with response:
        body = response.read()
    if body[:2] == b"\x1f\x8b":
        body = gzip.decompress(body)
    return body.decode("utf-8", errors="replace")


def check_pages(results):
    outcomes = []
    cache: dict[str, str | Exception] = {}
    for row in results:
        if not row["url"].startswith("http"):
            continue
        if row["url"] not in cache:
            try:
                cache[row["url"]] = normalise(fetch(row["url"]))
            except Exception as error:  # noqa: BLE001 - a page that cannot be read is a failed check
                cache[row["url"]] = error
        page = cache[row["url"]]
        if isinstance(page, Exception):
            outcomes.append((row["option"], row["property"], False, f"fetch failed: {page}"))
            continue
        found = normalise(row["quote"]) in page
        outcomes.append((row["option"], row["property"], found, "quote present" if found else "quote gone"))
    return outcomes


# ---------------------------------------------------------------- the lab

# One read row: {"day": int, "amount": cents, "printed": cents | None}

SEED = 20261026
STATEMENTS = 24
DAYS_PER_STATEMENT = 18
REGIMES = ("closing_only", "daily", "per_row")


def build_corpus(seed: int = SEED, count: int = STATEMENTS):
    rng = random.Random(seed)
    corpus = []
    for _ in range(count):
        opening = rng.randrange(800_00, 2_500_000)
        balance = opening
        rows = []
        for day in range(DAYS_PER_STATEMENT):
            for _ in range(rng.choice([2, 2, 3, 3, 4])):
                amount = rng.randrange(500, 480_000)
                if rng.random() >= 0.45:
                    amount = -amount
                if balance + amount < 1_000:
                    amount = abs(amount)
                balance += amount
                rows.append({"day": day, "amount": amount, "balance": balance})
        corpus.append({"opening": opening, "closing": balance, "rows": rows})
    return corpus


def printed_flags(rows, regime):
    if regime == "closing_only":
        return [False] * len(rows)
    if regime == "daily":
        last = {row["day"]: index for index, row in enumerate(rows)}
        keep = set(last.values())
        return [index in keep for index in range(len(rows))]
    return [True] * len(rows)


def as_read(statement, regime):
    flags = printed_flags(statement["rows"], regime)
    return [
        {"day": row["day"], "amount": row["amount"], "printed": row["balance"] if flags[i] else None}
        for i, row in enumerate(statement["rows"])
    ]


# --- checkBalanceChain, packages/shared/src/chain.ts

def chain(rows, opening, closing):
    previous = opening
    movement = 0
    for index, row in enumerate(rows):
        moved = previous + row["amount"]
        if row["printed"] is not None:
            if row["printed"] != moved:
                return {"status": "needs_review", "break": "running_balance", "first": index + 1}
            previous = row["printed"]
        else:
            previous = moved
        movement += row["amount"]
    if closing != opening + movement:
        return {"status": "needs_review", "break": "closing", "first": len(rows) or None}
    return {"status": "balanced", "break": None, "first": None}


# --- contradictedBalanceRow, packages/parsers/src/invariants/balance-chain.ts

def contradicted_row(rows, opening, closing):
    result = chain(rows, opening, closing)
    if result["break"] != "running_balance" or result["first"] is None:
        return None
    position = result["first"]
    row = rows[position - 1]
    before = opening if position == 1 else rows[position - 2]["printed"]
    if row["printed"] is None or before is None or closing is None:
        return None
    running = before + row["amount"]
    for later in rows[position:]:
        running += later["amount"]
        if later["printed"] is not None and later["printed"] != running:
            return None
    after = rows[position] if position < len(rows) else None
    if after is not None and row["printed"] - before == after["amount"]:
        return None
    return position if running == closing else None


def verdict(rows, opening, closing):
    result = chain(rows, opening, closing)
    if result["status"] == "balanced":
        return {"flagged": False, "kind": "balanced", "row": None}
    if result["break"] == "running_balance":
        contradicted = contradicted_row(rows, opening, closing)
        if contradicted is not None:
            # The product reports this shape as balanced, with one printed figure named.
            return {"flagged": False, "kind": "contradicted", "row": contradicted}
        return {"flagged": True, "kind": "break", "row": result["first"]}
    # A closing figure names no row; the review screen marks the day of the last row.
    return {"flagged": True, "kind": "closing", "row": len(rows)}


def search_set(rows, regime, result):
    """Rows a person has to read after the check has spoken, and the day it points at."""
    if not result["flagged"]:
        return len(rows), None
    day = rows[result["row"] - 1]["day"]
    if result["kind"] == "closing":
        return len(rows), day
    if regime == "per_row":
        return 1, day
    return sum(1 for row in rows if row["day"] == day), day


# --- error injection

def mutate_amount(rows, index):
    out = [dict(row) for row in rows]
    out[index]["amount"] += 137
    return out, {index}


def mutate_sign(rows, index):
    out = [dict(row) for row in rows]
    out[index]["amount"] = -out[index]["amount"]
    return out, {index}


def mutate_skip(rows, index):
    return [dict(row) for i, row in enumerate(rows) if i != index], {index}


def mutate_duplicate(rows, index):
    copy = dict(rows[index])
    copy["printed"] = None
    out = [dict(row) for row in rows]
    out.insert(index + 1, copy)
    return out, {index}


def mutate_pair(rows, index, same_day):
    day = rows[index]["day"]
    if same_day:
        mates = [i for i, row in enumerate(rows) if row["day"] == day and i != index]
    else:
        mates = [i for i, row in enumerate(rows) if row["day"] > day]
    if not mates:
        return None
    mate = mates[-1]
    out = [dict(row) for row in rows]
    out[index]["amount"] += 211
    out[mate]["amount"] -= 211
    return out, {index, mate}


SINGLE = {"amount": mutate_amount, "sign": mutate_sign, "skip": mutate_skip, "duplicate": mutate_duplicate}
PAIRS = {
    "pair_same_day": lambda rows, i: mutate_pair(rows, i, True),
    "pair_across_days": lambda rows, i: mutate_pair(rows, i, False),
}


def measure_cases(corpus):
    cases = []
    for statement in corpus:
        for regime in REGIMES:
            base = as_read(statement, regime)
            runs = [(kind, mutate, False) for kind, mutate in SINGLE.items()]
            runs += [(kind, mutate, True) for kind, mutate in PAIRS.items()]
            for kind, mutate, is_pair in runs:
                for index in range(len(base)):
                    changed = mutate(base, index)
                    if changed is None:
                        continue
                    rows, injected = changed
                    result = verdict(rows, statement["opening"], statement["closing"])
                    size, day = search_set(rows, regime, result)
                    error_days = {base[i]["day"] for i in injected}
                    cases.append({
                        "regime": regime, "kind": kind, "pair": is_pair,
                        "rows": rows, "opening": statement["opening"], "closing": statement["closing"],
                        "flagged": result["flagged"], "flag_kind": result["kind"],
                        "flag_day_holds_error": day is not None and day in error_days,
                        "error_on_day_before": day is not None and day not in error_days
                        and (day - 1) in error_days,
                        "search_rows": size, "statement_rows": len(rows),
                    })
    return cases


def pct(part, whole):
    return round(100 * part / whole, 1) if whole else None


def summarise(cases):
    out = {}
    for regime in REGIMES:
        mine = [c for c in cases if c["regime"] == regime]
        single = [c for c in mine if not c["pair"]]
        misread = [c for c in single if c["kind"] in ("amount", "sign")]
        moved = [c for c in single if c["kind"] in ("skip", "duplicate")]
        moved_missed = [c for c in moved if not c["flag_day_holds_error"]]
        same = [c for c in mine if c["kind"] == "pair_same_day"]
        across = [c for c in mine if c["kind"] == "pair_across_days"]
        out[regime] = {
            "single_cases": len(single),
            "single_detected_pct": pct(sum(c["flagged"] for c in single), len(single)),
            "mean_statement_rows": round(statistics.mean(c["statement_rows"] for c in single), 2),
            "mean_search_rows": round(statistics.mean(c["search_rows"] for c in single), 2),
            "median_search_rows": statistics.median(c["search_rows"] for c in single),
            "search_rows_share_pct": round(100 * statistics.mean(c["search_rows"] / c["statement_rows"] for c in single), 1),
            "misread_on_flagged_day_pct": pct(sum(c["flag_day_holds_error"] for c in misread), len(misread)),
            "skipped_or_doubled_on_flagged_day_pct": pct(sum(c["flag_day_holds_error"] for c in moved), len(moved)),
            "skipped_or_doubled_elsewhere_on_day_before_pct": pct(
                sum(c["error_on_day_before"] for c in moved_missed), len(moved_missed)
            ),
            "contradicted_row_cases": sum(c["flag_kind"] == "contradicted" for c in mine),
            "pair_same_day_cases": len(same),
            "pair_same_day_detected_pct": pct(sum(c["flagged"] for c in same), len(same)),
            "pair_across_days_cases": len(across),
            "pair_across_days_detected_pct": pct(sum(c["flagged"] for c in across), len(across)),
        }
    return out


def run_lab():
    started = time.time()
    corpus = build_corpus()
    cases = measure_cases(corpus)
    summary = summarise(cases)
    return {
        "seed": SEED,
        "statements": len(corpus),
        "rows": sum(len(s["rows"]) for s in corpus),
        "cases": len(cases),
        "seconds": round(time.time() - started, 2),
        "summary": summary,
    }, cases


# ---------------------------------------------------------------- the product's own rule

TS_RUNNER = r"""
import { readFileSync } from "node:fs";
const { checkBalanceChain } = await import(process.argv[2] + "/chain.ts");
const { contradictedBalanceRow } = await import(process.argv[2] + "/balance-chain.ts");
const usd = (cents) => (cents === null ? null : { minorUnits: BigInt(cents), currency: "USD" });
const out = [];
for (const line of readFileSync(process.argv[3], "utf8").split("\n")) {
  if (!line) continue;
  const c = JSON.parse(line);
  const input = {
    openingBalance: usd(c.opening),
    closingBalance: usd(c.closing),
    rows: c.rows.map(([amount, printed]) => ({ amount: usd(amount), runningBalance: usd(printed) })),
  };
  const v = checkBalanceChain(input);
  out.push([v.status, v.status === "balanced" ? null : v.break, v.status === "balanced" ? null : v.firstBrokenRow,
    contradictedBalanceRow(input)]);
}
process.stdout.write(JSON.stringify(out));
"""


def ts_sources():
    shared = REPO / "packages" / "shared" / "src"
    parsers = REPO / "packages" / "parsers" / "src" / "invariants"
    files = [shared / "chain.ts", shared / "money.ts", shared / "errors.ts", parsers / "balance-chain.ts"]
    return files if all(f.exists() for f in files) else None


def run_ts(cases):
    """The verdict of the product's TypeScript rule on every case, or a reason it could not run."""
    files = ts_sources()
    if files is None:
        return None, "the Evidilo repository sources are not beside this file"
    if shutil.which("node") is None:
        return None, "node is not on PATH"
    with tempfile.TemporaryDirectory() as tmp:
        work = Path(tmp)
        for source in files:
            text = source.read_text()
            text = text.replace('"./money.js"', '"./money.ts"').replace('"./errors.js"', '"./errors.ts"')
            text = text.replace(
                'import { addMoney, checkBalanceChain, subtractMoney } from "@sc/shared";',
                'import { checkBalanceChain } from "./chain.ts";\nimport { addMoney, subtractMoney } from "./money.ts";',
            )
            (work / source.name).write_text(text)
        (work / "run.mjs").write_text(TS_RUNNER)
        with (work / "cases.jsonl").open("w") as handle:
            for c in cases:
                handle.write(json.dumps({
                    "opening": c["opening"], "closing": c["closing"],
                    "rows": [[r["amount"], r["printed"]] for r in c["rows"]],
                }) + "\n")
        done = subprocess.run(
            ["node", "--experimental-transform-types", "--no-warnings", str(work / "run.mjs"), str(work), str(work / "cases.jsonl")],
            capture_output=True, text=True,
        )
        if done.returncode != 0:
            return None, f"node failed: {done.stderr.strip().splitlines()[-1] if done.stderr.strip() else done.returncode}"
        return json.loads(done.stdout), None


def compare_ts(cases, ts):
    mismatches = 0
    for case, (status, brk, first, contradicted) in zip(cases, ts):
        mine = chain(case["rows"], case["opening"], case["closing"])
        if (mine["status"], mine["break"], mine["first"]) != (status, brk, first):
            mismatches += 1
        elif contradicted_row(case["rows"], case["opening"], case["closing"]) != contradicted:
            mismatches += 1
    return mismatches


# ---------------------------------------------------------------- the table

LABEL = {
    "closing_only": "A closing balance only",
    "daily": "A balance at the end of each day",
    "per_row": "A balance beside every row",
}
COLUMNS = (
    "printed_balances", "single_errors", "single_detected_pct", "rows_to_read_mean", "rows_to_read_median",
    "share_of_statement_pct", "misread_on_flagged_day_pct", "skipped_or_doubled_on_flagged_day_pct",
    "cancelling_pairs_different_days", "pair_different_days_detected_pct", "cancelling_pairs_same_day",
    "pair_same_day_detected_pct", "checked_at",
)


def table_rows(summary, checked_at):
    rows = []
    for regime in REGIMES:
        s = summary[regime]
        rows.append({
            "printed_balances": LABEL[regime],
            "single_errors": str(s["single_cases"]),
            "single_detected_pct": str(s["single_detected_pct"]),
            "rows_to_read_mean": str(s["mean_search_rows"]),
            "rows_to_read_median": str(s["median_search_rows"]),
            "share_of_statement_pct": str(s["search_rows_share_pct"]),
            "misread_on_flagged_day_pct": "no day named" if regime == "closing_only" else str(s["misread_on_flagged_day_pct"]),
            "skipped_or_doubled_on_flagged_day_pct": "no day named" if regime == "closing_only" else str(s["skipped_or_doubled_on_flagged_day_pct"]),
            "cancelling_pairs_different_days": str(s["pair_across_days_cases"]),
            "pair_different_days_detected_pct": str(s["pair_across_days_detected_pct"]),
            "cancelling_pairs_same_day": str(s["pair_same_day_cases"]),
            "pair_same_day_detected_pct": str(s["pair_same_day_detected_pct"]),
            "checked_at": checked_at,
        })
    return rows


# ---------------------------------------------------------------- main


def main(argv):
    recheck = "--recheck" in argv
    write = "--write" in argv
    dataset = json.loads(DATASET.read_text())
    failures = []

    for option, prop, ok, detail in check_pages(dataset["results"]):
        print(f"{'ok  ' if ok else 'FAIL'} page {option} / {prop}: {detail}")
        if not ok:
            failures.append(f"{option} / {prop}")

    recorded = dataset["lab"]
    if recheck and recorded["seconds"] > LAB_BUDGET_SECONDS:
        print(f"skip lab: recorded run took {recorded['seconds']}s, over the one-minute budget")
    else:
        lab, cases = run_lab()
        print(f"lab  {lab['statements']} statements, {lab['rows']} rows, {lab['cases']} cases, {lab['seconds']}s")

        started = time.time()
        ts, reason = run_ts(cases)
        if ts is None:
            print(f"skip ts: {reason}")
            lab["typescript"] = recorded.get("typescript")
        else:
            mismatches = compare_ts(cases, ts)
            print(f"{'ok  ' if mismatches == 0 else 'FAIL'} ts   the product's rule agrees on {len(cases) - mismatches} of {len(cases)} cases, {round(time.time() - started, 2)}s")
            if mismatches:
                failures.append(f"ts mismatches {mismatches}")
            lab["typescript"] = {"cases": len(cases), "agree": len(cases) - mismatches}

        if write:
            dataset["lab"] = lab
            DATASET.write_text(json.dumps(dataset, indent=2, ensure_ascii=False) + "\n")
            with TABLE.open("w", newline="") as handle:
                writer = csv.DictWriter(handle, fieldnames=COLUMNS, lineterminator="\n")
                writer.writeheader()
                writer.writerows(table_rows(lab["summary"], dataset["checkedAt"]))
            print("wrote dataset.json and dataset.csv")
            recorded = lab

        for regime, values in recorded["summary"].items():
            for key, value in values.items():
                got = lab["summary"].get(regime, {}).get(key)
                if got != value:
                    failures.append(f"lab {regime}.{key}")
                    print(f"FAIL lab {regime}.{key}: dataset {value!r}, measured {got!r}")
        with TABLE.open(newline="") as handle:
            on_disk = list(csv.DictReader(handle))
        expected = table_rows(lab["summary"], dataset["checkedAt"])
        if on_disk != expected:
            failures.append("dataset.csv")
            print("FAIL dataset.csv differs from the measurement")
        else:
            print(f"ok   dataset.csv matches the measurement, {len(expected)} rows")

    if failures:
        print(f"\n{len(failures)} failed")
        return 1
    print("\nall quotes and every lab number hold")
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))