Rectify sanitized statement party equivalence

scripts/fixtures/README.md

@@ -15,13 +15,13 @@ Read the banner comment at the top of each file for exactly what is real and wha
 ```bash
 pdftotext -bbox-layout -opw "$PASSWORD" statement.pdf raw.xml
 python3 scripts/sanitise-bbox-capture.py raw.xml \
-  scripts/fixtures/hdfc-bbox-capture.xml "HDFC current-account" \
+  scripts/fixtures/hdfc-bbox-capture.xml hdfc \
   0:0-800 2:200-330,700-800 3:200-300
 ```
 
 ```bash
 python3 scripts/sanitise-bbox-capture.py raw.xml \
-  scripts/fixtures/sbi-bbox-capture.xml "State Bank of India current-account" \
+  scripts/fixtures/sbi-bbox-capture.xml sbi \
   0:90-741 1:0-165
 ```
 
@@ -30,6 +30,11 @@ header, a continuation page that repeats the column header while a row is in pro
 carries the end-of-statement marker, and — for HDFC — one page *after* that marker, so the marker
 cannot be removed without a test noticing.
 
+The bank argument is a closed parser selection (`hdfc` or `sbi`). The sanitiser parses every selected
+source region and the complete generated regions, using the same retained geometry before writing the destination. It refuses empty,
+incomplete accounting rows, misaligned or party-class-changing evidence; captured geometry remains fixture evidence and does not
+qualify raw customer data.
+
 ## Adding a capture for a new bank
 
 1. Sanitise, then **diff the result against the source** and scan for surviving tokens (the script's

scripts/sanitise-bbox-capture.py

@@ -6,6 +6,8 @@
 substituted, so a value that was never anticipated is fabricated by default
 rather than kept by default.
 """
+import decimal
+import importlib.util
 import re, sys, pathlib
 
 TEMPLATE = set("""
@@ -43,6 +45,16 @@
 _dates = {}
 
 
+class EvidenceRefusal(SystemExit):
+    """A stable reason, with optional non-sensitive parser location context."""
+    def __init__(self, category, bank=None, row_index=None):
+        self.category, self.bank, self.row_index = category, bank, row_index
+        context = "" if bank is None else f" bank={bank}"
+        if row_index is not None:
+            context += f" row={row_index}"
+        super().__init__(f"sanitise: {category}{context}")
+
+
 def _fake_date(token):
     """Dates are remapped, not digit-substituted.
 
@@ -491,6 +503,89 @@ def _kept_words(pages, keep):
         yield head, words
 
 
+def _load_parser(bank_name):
+    """Load one of the parsers used to qualify a generated fixture."""
+    if bank_name not in ("hdfc", "sbi"):
+        raise EvidenceRefusal("unsupported_parser_profile")
+    path = pathlib.Path(__file__).with_name("bank_statement_import.py")
+    spec = importlib.util.spec_from_file_location("sanitise_bank_parser", path)
+    if spec is None or spec.loader is None:
+        raise EvidenceRefusal("parser_unavailable")
+    parser = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(parser)
+    return parser, parser.BANKS[bank_name]()
+
+
+def _assert_party_partition(source_keys, output_keys, bank_name):
+    """Require a two-way, one-to-one mapping of party equivalence classes."""
+    if not source_keys or len(source_keys) != len(output_keys):
+        raise EvidenceRefusal("party_evidence_empty_or_misaligned", bank_name)
+    source_to_output, output_to_source = {}, {}
+    for index, (source, output) in enumerate(zip(source_keys, output_keys)):
+        if (not source or not output
+                or (isinstance(source, str) and source.upper() in ("UNRESOLVED", "UNNAMED"))
+                or (isinstance(output, str) and output.upper() in ("UNRESOLVED", "UNNAMED"))):
+            raise EvidenceRefusal("party_evidence_underdetermined", bank_name, index)
+        old = source_to_output.setdefault(source, output)
+        reverse = output_to_source.setdefault(output, source)
+        if old != output:
+            raise EvidenceRefusal("party_partition_split", bank_name, index)
+        if reverse != source:
+            raise EvidenceRefusal("party_partition_merged", bank_name, index)
+
+
+def _validate_parser_evidence(parser, bank, source_pages, output_pages, bank_name):
+    """Parse complete page sets and compare only structure preserved by scrubbing."""
+    try:
+        source_rows = parser.parse_pages(source_pages, bank)
+        output_rows = parser.parse_pages(output_pages, bank)
+    except EvidenceRefusal:
+        raise
+    except SystemExit as error:
+        raise EvidenceRefusal("parser_evidence_invalid", bank_name) from error
+    except (KeyError, IndexError, TypeError, ValueError, decimal.InvalidOperation) as error:
+        raise EvidenceRefusal("parser_evidence_invalid", bank_name) from error
+    if not source_rows or not output_rows or len(source_rows) != len(output_rows):
+        raise EvidenceRefusal("parser_evidence_empty_or_misaligned", bank_name)
+
+    source_dates, output_dates = [], []
+    source_keys, output_keys = [], []
+    for index, (source, output) in enumerate(zip(source_rows, output_rows)):
+        try:
+            source_date = str(source[bank.date_column]).strip()
+            output_date = str(output[bank.date_column]).strip()
+            source_dates.append(bank.parse_date(source_date))
+            output_dates.append(bank.parse_date(output_date))
+            for row in (source, output):
+                for column in (bank.debit_column, bank.credit_column):
+                    value = str(row.get(column) or "").strip()
+                    if value:
+                        parser._money(value, column, index + 1)
+                balance = str(row.get(bank.balance_column) or "").strip()
+                parser._balance(balance, bank.balance_column, index + 1)
+            source_ref = bank.reference(source)
+            output_ref = bank.reference(output)
+            source_side = tuple(bool(source.get(column)) for column in (bank.debit_column, bank.credit_column))
+            output_side = tuple(bool(output.get(column)) for column in (bank.debit_column, bank.credit_column))
+            source_shape = (source_side, bool(source.get(bank.balance_column)), source_ref[0], len(str(source_ref[1])))
+            output_shape = (output_side, bool(output.get(bank.balance_column)), output_ref[0], len(str(output_ref[1])))
+            if sum(source_side) != 1 or not source_shape[1] or sum(output_side) != 1 or not output_shape[1]:
+                raise EvidenceRefusal("accounting_row_incomplete", bank_name, index)
+            if source_shape != output_shape:
+                raise EvidenceRefusal("accounting_row_shape_misaligned", bank_name, index)
+            source_keys.append(parser._key(bank.party(source)))
+            output_keys.append(parser._key(bank.party(output)))
+        except EvidenceRefusal:
+            raise
+        except SystemExit as error:
+            raise EvidenceRefusal("row_alignment_invalid", bank_name, index) from error
+        except Exception as error:
+            raise EvidenceRefusal("row_alignment_invalid", bank_name, index) from error
+
+    _assert_party_partition(source_dates, output_dates, bank_name)
+    _assert_party_partition(source_keys, output_keys, bank_name)
+
+
 def main(source, destination, keep, bank):
     """keep: [(page_index, [(y_min, y_max), ...]), ...] regions to retain."""
     # `pdftotext` emits UTF-8. `read_text()` without an encoding decodes with
@@ -499,7 +594,9 @@ def main(source, destination, keep, bank):
     # `UnicodeDecodeError` before sanitisation runs at all. Neither CI nor the
     # unit cases reach this boundary: CI is ubuntu-only, and the Unicode tests
     # call `_scrub_plain` with strings that are already decoded.
+    keep = list(keep)
     pages = pathlib.Path(source).read_text(encoding="utf-8").split("<page ")[1:]
+    parser, bank_profile = _load_parser(bank)
     regions = list(_kept_words(pages, keep))
     # Two passes, and the first one has to be complete before the second starts.
     # A replacement is only safe once the allocator knows every token the
@@ -509,15 +606,20 @@ def main(source, destination, keep, bank):
     for _, words in regions:
         for *_, body in words:
             reserve_source_tokens(body)
-    chunks = [
+    def render(transform):
+        return [
         "<page " + head + "\n"
         + "\n".join(f'<word xMin="{x0}" yMin="{y0}" xMax="{x1}" yMax="{y1}">'
-                    f'{scrub(body)}</word>' for x0, y0, x1, y1, body in words)
+                    f'{transform(body)}</word>' for x0, y0, x1, y1, body in words)
         + "\n</page>"
         for head, words in regions
-    ]
-    pathlib.Path(destination).write_text(
-        BANNER_TEMPLATE.format(bank=bank) + "\n".join(chunks) + "\n", encoding="utf-8")
+        ]
+    source_chunks = render(lambda body: body)
+    chunks = render(scrub)
+    output = BANNER_TEMPLATE.format(bank=bank_profile.name) + "\n".join(chunks) + "\n"
+    _validate_parser_evidence(parser, bank_profile, [chunk[len("<page "):] for chunk in source_chunks],
+                              output.split("<page ")[1:], bank_profile.name)
+    pathlib.Path(destination).write_text(output, encoding="utf-8")
     print(f"wrote {destination}: "
           f"{sum(chunk.count('<word') for chunk in chunks)} words, {len(chunks)} pages")
 
@@ -526,7 +628,7 @@ def main(source, destination, keep, bank):
 
   SOURCE  pdftotext -bbox-layout output from a real statement
   DEST    fixture to write
-  BANK    a description for the banner, e.g. "HDFC current-account"
+  BANK    parser profile: hdfc or sbi (closed selection)
   PAGE:.. 0-based page index and the y ranges to keep from it
 
 Always diff the result against the source before committing it, and scan the