#!/usr/bin/env python3 """Check for the post /blog/qif-file-format. Lab: write one statement with Evidilo's production QIF and OFX writers (packages/exports/src/qif.ts, ofx.ts, and the row ids of packages/parsers/src/txid.ts), then read the QIF dates four ways: a strict month-first reader, a strict day-first reader, and quiffen with day_first False and True. Docs: re-read every page cited in dataset.json and look for its quotes. python3 check.py lab + pages; compares with dataset.csv python3 check.py --write lab only; rewrites dataset.csv from this run python3 check.py --recheck pages, plus the lab steps that need no install Exit 0: everything holds. 1: a quote is gone or a measurement differs. 2: a page could not be reached. """ import csv import datetime as dt import html import json import os import re import subprocess import sys import tempfile import urllib.error import urllib.request from pathlib import Path HERE = Path(__file__).resolve().parent ROOT = HERE.parents[4] # -> blog -> public -> web -> apps -> repo root UA = "Mozilla/5.0 (X11; Linux x86_64; rv:131.0) Gecko/20100101 Firefox/131.0" VENV = Path(tempfile.gettempdir()) / "qif-file-format-check-venv" # One March statement. Two coffees on the same day for the same amount are two purchases. ROWS = [ ["2026-03-03", "ACH DEPOSIT PAYROLL", 245000], ["2026-03-04", "ONLINE TRANSFER TO SAVINGS", -50000], ["2026-03-09", "RENT MARCH", -180000], ["2026-03-12", "CORNER CAFE", -450], ["2026-03-12", "CORNER CAFE", -450], ["2026-03-13", "CITY ELECTRIC", -9620], ["2026-03-27", "CARD PAYMENT", -31215], ] GEN_JS = r""" import { register } from 'node:module'; import { pathToFileURL } from 'node:url'; const root = process.argv[2]; // The package imports './x.js' and the source is './x.ts': try the .ts when the .js is absent. register('data:text/javascript,' + encodeURIComponent( "export async function resolve(s,c,n){try{return await n(s,c)}catch(e){" + "if(s.startsWith('.')&&s.endsWith('.js'))return n(s.slice(0,-3)+'.ts',c);throw e}}")); const imp = (p) => import(pathToFileURL(root + '/' + p).href); const { renderQif } = await imp('packages/exports/src/qif.ts'); const { renderOfx } = await imp('packages/exports/src/ofx.ts'); const { txids } = await imp('packages/parsers/src/txid.ts'); const { money } = await imp('packages/exports/node_modules/@sc/shared/dist/index.js'); const rowsIn = JSON.parse(process.argv[3]); let bal = 400000n; const rows = rowsIn.map(([date, desc, cents], i) => { const amt = BigInt(cents); bal += amt; return { ordinal: i + 1, date, description: desc, counterparty: desc, amount: money(amt, 'USD'), direction: amt < 0n ? 'debit' : 'credit', runningBalance: money(bal, 'USD'), symbols: { variable: null, constant: null, specific: null }, page: 1, note: null }; }); const statement = { bankId: 'wf', layoutFamily: 'digital-table', accountId: '6302', period: { from: '2026-03-01', to: '2026-03-31' }, openingBalance: money(400000n, 'USD'), closingBalance: money(bal, 'USD'), pageCount: 1, rows }; // As apps/api/src/conversions/conversions.service.ts does: ids from txids, looked up by ordinal. const ids = txids({ bankId: 'wf', accountId: '6302' }, rows); const byOrd = new Map(rows.map((r, i) => [r.ordinal, ids[i]])); const td = new TextDecoder(); const qif = td.decode(renderQif(statement)); const ofx = td.decode(renderOfx(statement, () => new Date('2026-10-25T12:00:00Z'), (r) => byOrd.get(r.ordinal))); process.stdout.write(JSON.stringify({ qif, ofx })); """ def write_files(): with tempfile.TemporaryDirectory() as tmp: script = Path(tmp) / "gen.mjs" script.write_text(GEN_JS) out = subprocess.run( ["node", "--no-warnings", str(script), str(ROOT), json.dumps(ROWS)], capture_output=True, text=True, check=True, cwd=tmp, ) return json.loads(out.stdout) def qif_records(qif): records, cur = [], {} for line in qif.split("\r\n"): if not line or line.startswith("!"): continue if line == "^": records.append(cur) cur = {} else: cur[line[0]] = line[1:] return records def strict(value, fmt): try: return dt.datetime.strptime(value, fmt).date().isoformat() except ValueError: return "invalid" def quiffen_dates(qif_text): try: import quiffen # noqa: F401 except ImportError: return None import importlib.metadata as md import quiffen out = {"version": md.version("quiffen")} with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "statement.qif" path.write_bytes(qif_text.encode()) for day_first in (False, True): q = quiffen.Qif.parse(str(path), day_first=day_first) txs = [t for acc in q.accounts.values() for ts in acc.transactions.values() for t in ts] out[day_first] = [t.date.date().isoformat() for t in txs] return out def ensure_quiffen(): """Re-run under a venv holding quiffen when this interpreter lacks it. Returns only when present.""" try: import quiffen # noqa: F401 return except ImportError: pass if os.environ.get("QIF_CHECK_REEXEC"): print("quiffen could not be installed") sys.exit(1) py = VENV / "bin" / "python" if not py.exists(): subprocess.run([sys.executable, "-m", "venv", str(VENV)], check=True) subprocess.run([str(py), "-m", "pip", "install", "-q", "quiffen==4.0.1"], check=True) env = dict(os.environ, QIF_CHECK_REEXEC="1") sys.exit(subprocess.run([str(py), "-I", str(Path(__file__).resolve()), *sys.argv[1:]], env=env, cwd=HERE).returncode) def lab(need_quiffen): first, second = write_files(), write_files() qif, ofx = first["qif"], first["ofx"] fitids = re.findall(r"(\S+)", ofx) posted = re.findall(r"(\S+)", ofx) recs = qif_records(qif) qz = quiffen_dates(qif) if need_quiffen and qz is None: raise SystemExit("quiffen missing") rows = [] for i, (rec, (iso, _desc, _cents)) in enumerate(zip(recs, ROWS)): d = rec["D"] rows.append({ "row": str(i + 1), "statement_date": iso, "qif_date_line": "D" + d, "qif_fields": "".join(sorted(rec.keys())), "amount": rec["T"], "payee": rec["P"], "month_first": strict(d, "%m/%d'%Y"), "day_first_strict": strict(d, "%d/%m'%Y"), "quiffen_day_first_false": qz[False][i] if qz else "", "quiffen_day_first_true": qz[True][i] if qz else "", "ofx_dtposted": posted[i], "ofx_fitid": fitids[i], }) summary = { "qif_records": len(recs), "qif_distinct_records": len({tuple(sorted(r.items())) for r in recs}), "ofx_fitids": len(fitids), "ofx_distinct_fitids": len(set(fitids)), "qif_identical_on_second_run": first["qif"] == second["qif"], "fitids_identical_on_second_run": fitids == re.findall(r"(\S+)", second["ofx"]), "day_first_strict_changed": sum(r["day_first_strict"] not in ("invalid", r["statement_date"]) for r in rows), "day_first_strict_invalid": sum(r["day_first_strict"] == "invalid" for r in rows), "quiffen_day_first_changed": sum(bool(qz) and r["quiffen_day_first_true"] != r["statement_date"] for r in rows), "quiffen_version": qz["version"] if qz else None, } return rows, summary, qif def page_text(url): req = urllib.request.Request(url, headers={"User-Agent": UA, "Accept": "text/html,*/*", "Accept-Language": "en-US"}) body = urllib.request.urlopen(req, timeout=45).read().decode("utf-8", "replace") body = re.sub(r"(?s)<(script|style)[^>]*>.*?", " ", body) text = html.unescape(re.sub(r"<[^>]+>", " ", body)) text = text.replace("’", "'").replace("‘", "'").replace("\xa0", " ") return re.sub(r"\s+", " ", text) def pages(dataset): failed, unreachable = 0, 0 for src in dataset["inputs"]["sources"]: try: text = page_text(src["url"]) except (urllib.error.URLError, TimeoutError, OSError) as exc: print(f"UNREACHABLE {src['url']}: {exc}") unreachable += 1 continue for quote in src.get("quotes", []): ok = quote in text print(("ok " if ok else "MISSING") + f" {src['id']}: {quote}") failed += not ok for absent in src.get("absent", []): ok = absent not in text print(("ok " if ok else "PRESENT") + f" {src['id']}: no '{absent}' on the page") failed += not ok return failed, unreachable def main(): args = set(sys.argv[1:]) recheck, write = "--recheck" in args, "--write" in args if not recheck: ensure_quiffen() dataset = json.loads((HERE / "dataset.json").read_text()) rows, summary, qif = lab(need_quiffen=not recheck) if write: with open(HERE / "dataset.csv", "w", newline="") as f: w = csv.DictWriter(f, fieldnames=list(rows[0].keys())) w.writeheader() w.writerows(rows) print(qif) print(json.dumps(summary, indent=1)) return 0 failed = 0 with open(HERE / "dataset.csv", newline="") as f: expected = list(csv.DictReader(f)) for got, want in zip(rows, expected): for key, value in got.items(): if value == "" and recheck: continue # quiffen columns when it is not installed if want.get(key) != value: print(f"DIFFERS row {got['row']} {key}: dataset {want.get(key)!r}, now {value!r}") failed += 1 if len(rows) != len(expected): print("DIFFERS row count") failed += 1 for key, value in dataset["results"]["summary"].items(): if key == "quiffen_version" and summary[key] is None: continue if key in ("quiffen_day_first_changed",) and summary["quiffen_version"] is None: continue if summary.get(key) != value: print(f"DIFFERS {key}: dataset {value!r}, now {summary.get(key)!r}") failed += 1 print(f"lab: {len(rows)} rows compared, {failed} differences") page_failed, unreachable = pages(dataset) failed += page_failed if failed: return 1 if unreachable: print(f"{unreachable} page(s) unreachable; nothing else differs") return 2 print("everything holds") return 0 if __name__ == "__main__": sys.exit(main())