#!/usr/bin/env python3
"""The check behind /blog/ofx-file-format.

One three-row bank statement is written two ways:
  * OFX 1.0.2 (SGML) by Evidilo's production writer, packages/exports/src/ofx.ts, run through ofx-harness.mjs;
  * OFX 2.2 (XML), built by hand in this script following the 2.x structure (same aggregates, every element closed,
    an XML declaration and an <?OFX ...?> processing instruction as the header).
A third file, the 1.0.2 body with its header cut off, separates the two reasons an XML reader refuses OFX 1.x.

Each file goes through seven readers: Python's xml.etree.ElementTree, ofxtools, ofxparse, pandas.read_xml (lxml and
etree parsers), pandas.read_csv, and Evidilo's own OFX reader (packages/parsers/src/structured/ofx.ts).
Then the Financial Data Exchange OFX page is fetched and the sentences quoted in the post are looked for.

  python3 check.py            run everything, compare with dataset.json, exit 0 when it all still holds
  python3 check.py --recheck  the same (no step here takes a minute), the weekly form
  python3 check.py --write    run everything and rewrite dataset.json and dataset.csv from the results

Exit codes: 0 holds, 1 a quote is gone or a measurement differs, 2 a cited page could not be reached.
The Python readers live in a virtualenv (default /tmp/ofx-venv, or $OFX_VENV); it is created with uv and the pinned
versions below when it does not exist.
"""
import csv
import hashlib
import html
import io
import json
import os
import re
import subprocess
import sys
import urllib.error
import urllib.request
import warnings

HERE = os.path.dirname(os.path.abspath(__file__))
REPO = os.path.abspath(os.path.join(HERE, "..", "..", "..", "..", ".."))  # apps/web/public/blog/<slug> -> repo root
VENV = os.environ.get("OFX_VENV", "/tmp/ofx-venv")
PINS = ["ofxtools==1.1.1", "ofxparse==0.21", "pandas==3.1.0", "lxml==6.1.3", "beautifulsoup4==4.15.0"]
UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0 Safari/537.36"

FDX_URL = "https://financialdataexchange.org/about-fdx/ofx-work-group/"
FDX_QUOTES = [
    "OFX has evolved through the years from OFX 1.0, which used an SGML syntax, to the current OFX 2.2, which uses XML.",
    "Many servers and clients currently in production were initially developed using OFX 1.0.2 using an SGML-based OFX syntax.",
    "Many have transitioned over to XML (introduced with the OFX 2.0 specification).",
    "However, for backwards compatibility purposes, both SGML (OFX 1.x) and XML (OFX 2.x) are supported in the current marketplace by most processors.",
    "The specification is split into two parts for easier consumption",
    "OFX Banking 2.3",
]


def ensure_venv():
    """Re-run this script inside the virtualenv that holds the pinned readers."""
    try:
        import ofxtools  # noqa: F401
        import ofxparse  # noqa: F401
        import pandas  # noqa: F401
        return
    except ImportError:
        pass
    python = os.path.join(VENV, "bin", "python")
    if not os.path.exists(python):
        subprocess.run(["uv", "venv", VENV], check=True)
        subprocess.run(["uv", "pip", "install", "--python", python, *PINS], check=True)
    os.execv(python, [python, "-I", os.path.abspath(__file__), *sys.argv[1:]])


# OFX 2.2, built by hand from the 2.x structure. Same statement, same aggregates, same values as the 1.0.2 file.
OFX_220 = """<?xml version="1.0" encoding="UTF-8" standalone="no"?>
<?OFX OFXHEADER="200" VERSION="220" SECURITY="NONE" OLDFILEUID="NONE" NEWFILEUID="NONE"?>
<OFX>
<SIGNONMSGSRSV1>
<SONRS>
<STATUS>
<CODE>0</CODE>
<SEVERITY>INFO</SEVERITY>
</STATUS>
<DTSERVER>20261025120000</DTSERVER>
<LANGUAGE>ENG</LANGUAGE>
<FI>
<ORG>Wells Fargo</ORG>
</FI>
</SONRS>
</SIGNONMSGSRSV1>
<BANKMSGSRSV1>
<STMTTRNRS>
<TRNUID>0</TRNUID>
<STATUS>
<CODE>0</CODE>
<SEVERITY>INFO</SEVERITY>
</STATUS>
<STMTRS>
<CURDEF>USD</CURDEF>
<BANKACCTFROM>
<BANKID>wf</BANKID>
<ACCTID>6302</ACCTID>
<ACCTTYPE>CHECKING</ACCTTYPE>
</BANKACCTFROM>
<BANKTRANLIST>
<DTSTART>20260901</DTSTART>
<DTEND>20260930</DTEND>
<STMTTRN>
<TRNTYPE>CREDIT</TRNTYPE>
<DTPOSTED>20260902</DTPOSTED>
<TRNAMT>1874.22</TRNAMT>
<FITID>6302-202609-1</FITID>
<NAME>Square Inc</NAME>
</STMTTRN>
<STMTTRN>
<TRNTYPE>DEBIT</TRNTYPE>
<DTPOSTED>20260915</DTPOSTED>
<TRNAMT>-64.18</TRNAMT>
<FITID>6302-202609-2</FITID>
<NAME>AT&amp;T BILL PAYMENT</NAME>
</STMTTRN>
<STMTTRN>
<TRNTYPE>DEBIT</TRNTYPE>
<DTPOSTED>20260928</DTPOSTED>
<TRNAMT>-12.50</TRNAMT>
<FITID>6302-202609-3</FITID>
<NAME>MONTHLY SERVICE FEE</NAME>
</STMTTRN>
</BANKTRANLIST>
<LEDGERBAL>
<BALAMT>3797.54</BALAMT>
<DTASOF>20260930</DTASOF>
</LEDGERBAL>
</STMTRS>
</STMTTRNRS>
</BANKMSGSRSV1>
</OFX>
"""

FILES = [
    ("ofx-1.0.2", "OFX 1.0.2 (SGML), written by Evidilo's ofx.ts"),
    ("ofx-2.2", "OFX 2.2 (XML), built by hand"),
    ("ofx-1.0.2-no-header", "The OFX 1.0.2 body with its header cut off"),
]


def first_line(message):
    return str(message).strip().splitlines()[0].strip()


def build_files(workdir):
    out = subprocess.run(
        ["node", os.path.join(HERE, "ofx-harness.mjs"), REPO, "write"], capture_output=True, check=True
    ).stdout
    body = out[out.index(b"<OFX>"):]
    data = {"ofx-1.0.2": out, "ofx-2.2": OFX_220.encode("utf-8"), "ofx-1.0.2-no-header": body}
    paths = {}
    for key, raw in data.items():
        path = os.path.join(workdir, key + ".ofx")
        with open(path, "wb") as handle:
            handle.write(raw)
        paths[key] = path
    return data, paths


def read_all(data, paths):
    import xml.etree.ElementTree as ET

    import pandas as pd
    from ofxtools.Parser import OFXTree

    warnings.filterwarnings("ignore")
    import ofxparse

    def result(ok, rows=None, closing=None, name=None, error=None):
        return {"ok": ok, "rows": rows, "closing": closing, "row2Name": name, "error": error}

    def etree(key):
        try:
            root = ET.fromstring(data[key])
            trn = root.findall(".//STMTTRN")
            return result(True, len(trn), root.findtext(".//LEDGERBAL/BALAMT"), trn[1].findtext("NAME"))
        except Exception as e:  # noqa: BLE001
            return result(False, error=f"{type(e).__name__}: {first_line(e)}")

    def ofxtools_(key):
        try:
            tree = OFXTree()
            tree.parse(paths[key])
            s = tree.convert().statements[0]
            return result(True, len(s.transactions), str(s.balance.balamt), s.transactions[1].name)
        except Exception as e:  # noqa: BLE001
            return result(False, error=f"{type(e).__name__}: {first_line(e)}")

    def ofxparse_(key):
        try:
            with open(paths[key], "rb") as handle:
                s = ofxparse.OfxParser.parse(handle).account.statement
            return result(True, len(s.transactions), str(s.balance), s.transactions[1].payee)
        except Exception as e:  # noqa: BLE001
            return result(False, error=f"{type(e).__name__}: {first_line(e)}")

    def read_xml(parser):
        def run(key):
            try:
                df = pd.read_xml(io.BytesIO(data[key]), xpath=".//STMTTRN", parser=parser)
                note = f"{df.shape[0]} rows x {df.shape[1]} columns: {', '.join(map(str, df.columns))}"
                return result(True, int(df.shape[0]), None, str(df["NAME"].iloc[1]), note)
            except Exception as e:  # noqa: BLE001
                return result(False, error=f"{type(e).__name__}: {first_line(e)}")

        return run

    def read_csv(key):
        try:
            df = pd.read_csv(io.BytesIO(data[key]))
            return result(
                True,
                int(df.shape[0]),
                None,
                None,
                f"{df.shape[0]} rows x {df.shape[1]} column; header cell {df.columns[0]!r}",
            )
        except Exception as e:  # noqa: BLE001
            return result(False, error=f"{type(e).__name__}: {first_line(e)}")

    proc = subprocess.run(
        ["node", os.path.join(HERE, "ofx-harness.mjs"), REPO, "read", *[paths[k] for k, _ in FILES]],
        capture_output=True,
        check=True,
    )
    evidilo_out = json.loads(proc.stdout)

    def evidilo(key):
        r = evidilo_out[os.path.basename(paths[key])]
        if not r["ok"]:
            return result(False, error=r["error"])
        s = r["statement"]
        closing = s["closingBalance"]
        cents = int(closing["minorUnits"])
        return result(
            True,
            len(s["rows"]),
            f"{cents // 100}.{cents % 100:02d}",
            s["rows"][1]["cells"]["description"]["value"],
        )

    readers = [
        ("xml.etree.ElementTree", etree),
        ("ofxtools", ofxtools_),
        ("ofxparse", ofxparse_),
        ("pandas.read_xml (lxml)", read_xml("lxml")),
        ("pandas.read_xml (etree)", read_xml("etree")),
        ("pandas.read_csv", read_csv),
        ("Evidilo OFX reader", evidilo),
    ]
    results = []
    for key, label in FILES:
        for reader, fn in readers:
            r = fn(key)
            results.append({"file": key, "fileLabel": label, "reader": reader, **r})
    return results


def tools():
    import importlib.metadata as md
    import platform

    node = subprocess.run(["node", "--version"], capture_output=True, text=True).stdout.strip()
    pkgs = {name: md.version(name) for name in ["ofxtools", "ofxparse", "pandas", "lxml", "beautifulsoup4"]}
    return {"python": platform.python_version(), "node": node, **pkgs}


def fetch_text(url):
    req = urllib.request.Request(url, headers={"User-Agent": UA})
    with urllib.request.urlopen(req, timeout=40) as resp:
        raw = resp.read().decode("utf-8", "replace")
    raw = re.sub(r"(?is)<script.*?</script>|<style.*?</style>", " ", raw)
    return re.sub(r"\s+", " ", html.unescape(re.sub(r"<[^>]+>", " ", raw)))


def check_quotes():
    try:
        text = fetch_text(FDX_URL)
    except (urllib.error.URLError, TimeoutError, OSError) as e:
        return None, f"{FDX_URL} could not be reached: {e}"
    return [{"url": FDX_URL, "quote": q, "found": q in text} for q in FDX_QUOTES], None


def main():
    ensure_venv()
    write = "--write" in sys.argv
    workdir = os.path.join("/tmp", "ofx-file-format-check")
    os.makedirs(workdir, exist_ok=True)
    data, paths = build_files(workdir)
    results = read_all(data, paths)
    hashes = {k: hashlib.sha256(v).hexdigest() for k, v in data.items()}
    quotes, unreachable = check_quotes()

    if write:
        if unreachable:
            print(unreachable)
            return 2
        with open(os.path.join(HERE, "dataset.json"), encoding="utf-8") as handle:
            dataset = json.load(handle)
        dataset["tools"] = tools()
        dataset["results"] = {"readers": results, "fileSha256": hashes, "quotes": quotes}
        dataset["files"] = {k: v.decode("utf-8") for k, v in data.items() if k != "ofx-1.0.2-no-header"}
        with open(os.path.join(HERE, "dataset.json"), "w", encoding="utf-8") as handle:
            json.dump(dataset, handle, indent=2, ensure_ascii=False)
            handle.write("\n")
        with open(os.path.join(HERE, "dataset.csv"), "w", newline="", encoding="utf-8") as handle:
            w = csv.writer(handle)
            w.writerow(["file", "reader", "read", "rows", "closing_balance", "row2_name", "error_or_note"])
            for r in results:
                w.writerow([r["file"], r["reader"], "yes" if r["ok"] else "no", r["rows"] if r["rows"] is not None else "",
                            r["closing"] or "", r["row2Name"] or "", r["error"] or ""])
        print("dataset.json and dataset.csv written")

    with open(os.path.join(HERE, "dataset.json"), encoding="utf-8") as handle:
        expected = json.load(handle)["results"]

    failures = []
    exp_by = {(r["file"], r["reader"]): r for r in expected["readers"]}
    for r in results:
        e = exp_by.get((r["file"], r["reader"]))
        if e is None or any(e[k] != r[k] for k in ("ok", "rows", "closing", "row2Name", "error")):
            failures.append(f"{r['file']} / {r['reader']}: expected {e}, got {r}")
        status = "read" if r["ok"] else "refused"
        print(f"{r['file']:<22} {r['reader']:<26} {status:<8} {r['error'] or ''}")
    for k, h in hashes.items():
        if expected["fileSha256"].get(k) != h:
            failures.append(f"{k}: file bytes changed (sha256 {h})")

    if unreachable:
        print(unreachable)
        for f in failures:
            print("DIFFERS:", f)
        return 2
    for q in quotes:
        print(("found   " if q["found"] else "MISSING ") + q["quote"])
        if not q["found"]:
            failures.append(f"quote gone from {q['url']}: {q['quote']}")

    for f in failures:
        print("DIFFERS:", f)
    print("OK" if not failures else f"{len(failures)} difference(s)")
    return 1 if failures else 0


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