#!/usr/bin/env python3 """Re-runs every check behind the bank-statement-date-once-a-day post. Two kinds of row live in dataset.json: git the quote must still appear in the product's source at origin/main, read with `git show` and never by touching the working tree; lab the measurement is recomputed and compared with what the row records. The lab copies the product's own writers and its balance-chain rule out of origin/main into a temporary directory, rewrites the import specifiers so Node can run them as TypeScript (`@sc/shared` to a local barrel, `./x.js` to `./x.ts`), and measures what those functions produce. The claim it tests is byte identity, not a size: a writer that changes its spacing still passes while the carried mark stays out of the file, and the run prints the sizes it measured. Excel's writer imports SheetJS, which the lab loads from packages/exports/node_modules, so the repository has to be installed (`pnpm install`) before the Excel row can run. Exit 0 when every row holds and dataset.csv matches dataset.json, 1 otherwise. python3 check.py every row python3 check.py --recheck the same, without the lab when it runs longer than a minute """ import base64 import csv import hashlib import io import html import json import os import re import subprocess import sys import tempfile import time import unicodedata import urllib.error import urllib.parse import urllib.request import zipfile HERE = os.path.dirname(os.path.abspath(__file__)) DATASET = os.path.join(HERE, "dataset.json") CSV_PATH = os.path.join(HERE, "dataset.csv") REPO = os.environ.get("EVIDILO_REPO", os.path.normpath(os.path.join(HERE, "..", "..", "..", "..", ".."))) REF = os.environ.get("EVIDILO_REF", "origin/main") UA = ( "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/130.0 Safari/537.36" ) LAB_BUDGET_S = 60 # --- normalising a page --------------------------------------------------------------- QUOTE_MAP = { "‘": "'", "’": "'", "‚": "'", "‛": "'", "“": '"', "”": '"', "„": '"', "‟": '"', "«": '"', "»": '"', "‐": "-", "‑": "-", "‒": "-", "–": "-", "—": "-", "―": "-", "−": "-", " ": " ", " ": " ", " ": " ", " ": " ", " ": " ", "​": "", "": "", } QUOTE_RE = re.compile("|".join(re.escape(key) for key in QUOTE_MAP)) DROP = re.compile(r"(?is)<(script|style|noscript|svg|template)\b.*?") COMMENT = re.compile(r"(?is)") TAG = re.compile(r"(?s)<[^>]+>") SPACE = re.compile(r"\s+") def normalise(text): text = QUOTE_RE.sub(lambda match: QUOTE_MAP[match.group(0)], text) text = unicodedata.normalize("NFKC", text) return SPACE.sub(" ", text).strip() def page_variants(raw): """Two readings: tags become a space, and tags vanish so words split by markup still join.""" body = COMMENT.sub(" ", DROP.sub(" ", raw)) return [normalise(html.unescape(TAG.sub(" ", body))), normalise(html.unescape(TAG.sub("", body)))] def normalise_code(text): """A comment line's `//` and `*` are stripped, so a sentence wrapped over lines reads as one.""" text = re.sub(r"(?m)^\s*(?://|\*/|\*)\s?", " ", text) return normalise(text) def fetch(url, hops=4): request = urllib.request.Request(url, headers={"User-Agent": UA, "Accept": "text/html,*/*"}) try: with urllib.request.urlopen(request, timeout=45) as response: charset = response.headers.get_content_charset() or "utf-8" return response.read().decode(charset, "replace") except urllib.error.HTTPError as error: target = error.headers.get("Location") if error.headers and error.code in (301, 302, 303, 307, 308) else None if not target or hops == 0: raise return fetch(urllib.parse.urljoin(url, target), hops - 1) def git_show(path): result = subprocess.run( ["git", "-C", REPO, "show", "%s:%s" % (REF, path)], capture_output=True, text=True, check=False, ) if result.returncode != 0: raise RuntimeError("git show %s failed: %s" % (path, result.stderr.strip())) return result.stdout # --- the lab: the product's own writers and its own chain rule -------------------------- SOURCES = { "exports/csv.ts": "packages/exports/src/csv.ts", "exports/iif.ts": "packages/exports/src/iif.ts", "exports/ofx.ts": "packages/exports/src/ofx.ts", "exports/qbo.ts": "packages/exports/src/qbo.ts", "exports/qfx.ts": "packages/exports/src/qfx.ts", "exports/qif.ts": "packages/exports/src/qif.ts", "exports/xlsx.ts": "packages/exports/src/xlsx.ts", "exports/fields.ts": "packages/exports/src/fields.ts", "exports/clock.ts": "packages/exports/src/clock.ts", "exports/history.ts": "packages/exports/src/history.ts", "shared/money.ts": "packages/shared/src/money.ts", "shared/errors.ts": "packages/shared/src/errors.ts", "shared/types.ts": "packages/shared/src/types.ts", "shared/banks.ts": "packages/shared/src/banks.ts", "shared/dates.ts": "packages/shared/src/dates.ts", "shared/layouts.ts": "packages/shared/src/layouts.ts", "shared/chain.ts": "packages/shared/src/chain.ts", } BARREL = """ export * from "./banks.ts"; export * from "./chain.ts"; export * from "./dates.ts"; export * from "./errors.ts"; export * from "./layouts.ts"; export * from "./money.ts"; export * from "./types.ts"; """ SPECIFIER = re.compile(r'(from\s+")\./([A-Za-z0-9_-]+)\.js(")') PACKAGE = re.compile(r'(from\s+")@sc/shared(")') SHEETJS = os.path.join(REPO, "packages", "exports", "node_modules", "xlsx", "xlsx.mjs") SHEETJS_IMPORT = re.compile(r'(from\s+")xlsx(")') def rewrite(text): text = SHEETJS_IMPORT.sub(lambda match: match.group(1) + "file://" + SHEETJS + match.group(2), text) text = PACKAGE.sub(lambda match: match.group(1) + "../shared/all.ts" + match.group(2), text) return SPECIFIER.sub(lambda match: match.group(1) + "./" + match.group(2) + ".ts" + match.group(3), text) DRIVER = r''' import { renderCsv } from "./exports/csv.ts"; import { DEFAULT_CSV_OPTIONS } from "./exports/fields.ts"; import { renderIif } from "./exports/iif.ts"; import { renderOfx } from "./exports/ofx.ts"; import { renderQbo } from "./exports/qbo.ts"; import { renderQfx } from "./exports/qfx.ts"; import { renderQif } from "./exports/qif.ts"; import { renderXlsx } from "./exports/xlsx.ts"; import { checkBalanceChain } from "./shared/all.ts"; const cents = (value) => ({ minorUnits: BigInt(value), currency: "USD" }); // The post's own six rows: three printed dates, six transactions, a balance beside every row. const ROWS = [ { date: "2026-10-02", description: "ACH DEBIT ACME SUPPLY", counterparty: "Acme Supply", cents: -12000, balance: 88000 }, { date: "2026-10-02", description: "CARD PURCHASE HARBOR FUEL", counterparty: "Harbor Fuel", cents: -4550, balance: 83450 }, { date: "2026-10-02", description: "DEPOSIT CLIENT PAYMENT", counterparty: "Client Payment", cents: 30000, balance: 113450 }, { date: "2026-10-05", description: "CHECK 1043", counterparty: "Ridge Property", cents: -20000, balance: 93450 }, { date: "2026-10-05", description: "SERVICE FEE", counterparty: "JPMorgan Chase", cents: -1200, balance: 92250 }, { date: "2026-10-09", description: "ACH CREDIT REFUND", counterparty: "Refund Co", cents: 7525, balance: 99775 }, ]; function statement(marks) { return { bankId: "chase", layoutFamily: "digital-table", accountType: "checking", accountId: "000123456789", period: { from: "2026-10-01", to: "2026-10-31" }, openingBalance: cents(100000), closingBalance: cents(99775), pageCount: 1, rows: ROWS.map((row, index) => ({ ordinal: index + 1, date: row.date, ...(marks[index] ? { dateCarried: true } : {}), description: row.description, counterparty: row.counterparty, amount: cents(row.cents), direction: row.cents < 0 ? "debit" : "credit", runningBalance: cents(row.balance), symbols: { variable: null, constant: null, specific: null }, page: 1, note: null, })), }; } const clock = () => new Date("2026-10-26T00:00:00Z"); const id = (row) => "r" + String(row.ordinal) + "-" + row.date; const VARIANTS = { printed: [false, false, false, false, false, false], carried3: [false, true, true, false, true, false], carriedAll: [true, true, true, true, true, true], }; const WRITERS = { QBO: (s) => renderQbo(s, clock, id), OFX: (s) => renderOfx(s, clock, id), QFX: (s) => renderQfx(s, clock, id), QIF: (s) => renderQif(s), IIF: (s) => renderIif(s, { accountName: null }), CSV: (s) => renderCsv(s, DEFAULT_CSV_OPTIONS), XLSX: (s) => renderXlsx(s), }; const exported = {}; for (const [format, write] of Object.entries(WRITERS)) { exported[format] = {}; for (const [name, marks] of Object.entries(VARIANTS)) { exported[format][name] = Buffer.from(write(statement(marks))).toString("base64"); } } // The balance check, over the same six rows. A row's date never reaches it: the input it is // handed is the amount and the balance the statement printed beside the row. const chain = (rows) => checkBalanceChain({ openingBalance: cents(100000), closingBalance: cents(99775), rows: rows.map((row) => ({ amount: cents(row.cents), runningBalance: row.balance === null ? null : cents(row.balance) })), }); const perRow = ROWS.map((row) => ({ ...row })); // One balance at the end of each day: 10/02 ends on row 3, 10/05 on row 5, 10/09 on row 6. const perDay = ROWS.map((row, index) => ({ ...row, balance: index === 2 || index === 4 || index === 5 ? row.balance : null })); // The deposit printed under 10/05 instead of 10/02, in the place it held on the page. const movedDate = ROWS.map((row, index) => ({ ...row, date: index === 2 ? "2026-10-05" : row.date })); const movedPerDay = perDay.map((row, index) => ({ ...row, date: index === 2 ? "2026-10-05" : row.date })); // A control: the deposit moved to the end, which is a change to the order the check does read. const orderChanged = [ROWS[0], ROWS[1], ROWS[3], ROWS[4], ROWS[5], ROWS[2]]; const verdicts = { perRow_printed: chain(perRow), perDay_printed: chain(perDay), perRow_dateMoved: chain(movedDate), perDay_dateMoved: chain(movedPerDay), control_orderChanged: chain(orderChanged), }; console.log( JSON.stringify({ exported, chain: verdicts }, (_key, value) => (typeof value === "bigint" ? value.toString() : value)), ); ''' def run_lab(timeout=None): """Builds the sources in a temporary directory, runs Node, and returns what was measured.""" started = time.time() work = tempfile.mkdtemp(prefix="evidilo-carry-lab-") for destination, source in SOURCES.items(): path = os.path.join(work, destination) os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w", encoding="utf-8") as handle: handle.write(rewrite(git_show(source))) with open(os.path.join(work, "shared", "all.ts"), "w", encoding="utf-8") as handle: handle.write(BARREL) driver = os.path.join(work, "driver.ts") with open(driver, "w", encoding="utf-8") as handle: handle.write(DRIVER) node = subprocess.run(["node", "--version"], capture_output=True, text=True) result = subprocess.run( ["node", "--experimental-transform-types", "--no-warnings", driver], capture_output=True, text=True, check=False, timeout=timeout, ) if result.returncode != 0: raise RuntimeError("node failed: %s" % result.stderr.strip()[:2000]) measured = json.loads(result.stdout) formats = {} for format_name, variants in measured["exported"].items(): blobs = {name: base64.b64decode(body) for name, body in variants.items()} first = blobs["printed"] formats[format_name] = { "identical": all(body == first for body in blobs.values()), "variants": len(blobs), "bytes": len(first), "sha256": hashlib.sha256(first).hexdigest(), "carry_mark_in_bytes": any(b"carried" in readable(body).lower() for body in blobs.values()), } # Two formats that render the same document are one file under two names, which the post's # "seven formats" does not say. pairs = {} names = sorted(formats) for left_index, left in enumerate(names): for right in names[left_index + 1:]: pairs["%s_%s" % (left, right)] = { "identical": formats[left]["sha256"] == formats[right]["sha256"], } return {"formats": formats, "pairs": pairs, "chain": measured["chain"], "seconds": round(time.time() - started, 2), "node": node.stdout.strip() or "unknown"} def readable(body): """A workbook is a zip, so the mark is looked for in its unpacked parts rather than its deflated bytes.""" if not body.startswith(b"PK"): return body with zipfile.ZipFile(io.BytesIO(body)) as archive: return b"".join(archive.read(name) for name in archive.namelist()) def lookup(lab, key): node = lab for part in key.split("."): node = node[part] return node # --- reporting -------------------------------------------------------------------------- def check_csv(dataset): if not os.path.exists(CSV_PATH): return ["dataset.csv is missing"] problems = [] with open(CSV_PATH, newline="", encoding="utf-8") as handle: rows = list(csv.DictReader(handle)) expected = dataset["results"] if len(rows) != len(expected): problems.append("dataset.csv has %d rows, dataset.json has %d" % (len(rows), len(expected))) for index, (row, want) in enumerate(zip(rows, expected)): for key in ("option", "property", "value", "quote", "url", "method", "source", "verdict"): if row.get(key) != (want.get(key) or ""): problems.append("dataset.csv row %d, %s differs from dataset.json" % (index + 2, key)) return problems def main(argv): recheck = "--recheck" in argv with open(DATASET, encoding="utf-8") as handle: dataset = json.load(handle) print("check: %s, method %s, checkedAt %s" % (dataset["slug"], dataset["method"], dataset["checkedAt"])) print("mode: %s" % ("recheck" if recheck else "full")) failures = [] pages = {} for url in dict.fromkeys(row["url"] for row in dataset["results"] if row["kind"] == "page"): try: pages[url] = page_variants(fetch(url)) except Exception as error: # network, DNS, timeout, decode, HTTP pages[url] = None failures.append("%s -> %s: %s" % (url, type(error).__name__, error)) sources = {} for path in dict.fromkeys(row["url"].split(":", 2)[2] for row in dataset["results"] if row["kind"] == "git"): try: sources[path] = normalise_code(git_show(path)) except Exception as error: sources[path] = None failures.append("%s -> %s" % (path, error)) lab = None try: lab = run_lab(LAB_BUDGET_S if recheck else None) print("lab: %d formats and %d chain verdicts in %ss on node %s" % (len(lab["formats"]), len(lab["chain"]), lab["seconds"], lab["node"])) for name, measured in lab["formats"].items(): print(" %-4s %s bytes, sha256 %s, carry mark in bytes: %s" % (name, measured["bytes"], measured["sha256"][:16], measured["carry_mark_in_bytes"])) if lab["seconds"] > LAB_BUDGET_S: failures.append("the lab took %ss, over its %ss budget" % (lab["seconds"], LAB_BUDGET_S)) except subprocess.TimeoutExpired: print("lab: skipped, longer than %ss" % LAB_BUDGET_S) except Exception as error: failures.append("lab did not run: %s" % error) for row in dataset["results"]: label = "%-10s %-52s" % (row["option"][:10], row["property"][:52]) if row["kind"] == "page": haystack = pages.get(row["url"]) if haystack is None: print("FETCH-FAIL %s %s" % (label, row["url"])) continue needle = normalise(row["quote"]) found = any(needle in text for text in haystack) print("%s %s %s" % ("PASS" if found else "FAIL", label, row["url"])) if not found: failures.append("quote gone from %s: %r" % (row["url"], row["quote"][:90])) continue if row["kind"] == "git": haystack = sources.get(row["url"].split(":", 2)[2]) if haystack is None: print("GIT-FAIL %s %s" % (label, row["url"])) continue needle = normalise_code(row["quote"]) found = needle in haystack print("%s %s %s" % ("PASS" if found else "FAIL", label, row["url"])) if not found: failures.append("quote gone from %s: %r" % (row["url"], row["quote"][:90])) continue if lab is None: print("LAB-SKIP %s" % label) continue measured = lookup(lab, row["lab"]["key"]) want = row["lab"]["expect"] wrong = {key: (measured.get(key), value) for key, value in want.items() if measured.get(key) != value} if row["lab"]["key"].startswith("formats."): print("%s %s (%d bytes)" % ("PASS" if not wrong else "FAIL", label, measured["bytes"])) elif row["lab"]["key"].startswith("pairs."): print("%s %s (%s)" % ("PASS" if not wrong else "FAIL", label, measured.get("identical"))) else: print("%s %s (%s)" % ("PASS" if not wrong else "FAIL", label, measured.get("status"))) if wrong: failures.append("%s: %s" % (row["lab"]["key"], wrong)) for problem in check_csv(dataset): print("FAIL csv: %s" % problem) failures.append(problem) print("\n%d of %d rows hold, %d problem(s)" % (len(dataset["results"]) - len(failures), len(dataset["results"]), len(failures))) for failure in failures: print(" - %s" % failure) return 0 if not failures else 1 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))