← Files VeraARCHIVED FILE
modules/journal-bank-reconciliation/scripts/journal_bank_core.py
248 KB · Oct 5, 2026 · 18:29 UTC
from __future__ import annotations
import argparse
import csv
import hashlib
import io
import json
import logging
import os
import re
import stat
import sys
import tempfile
import unicodedata
from bisect import bisect_left
from collections import Counter
from dataclasses import dataclass
from datetime import date, datetime, timedelta
from decimal import Decimal
from pathlib import Path
from typing import Any, Sequence
from zipfile import ZIP_DEFLATED, BadZipFile, ZipFile, ZipInfo
import fastexcel
import openpyxl
import polars as pl
from excel_sanitization import excel_safe_value
from implementation_bootstrap import (
IMPLEMENTATION_CONTRACT,
implementation_contract,
validate_implementation_tree,
)
from openpyxl.styles import Alignment, Font, PatternFill
from openpyxl.utils import get_column_letter
from openpyxl.utils.exceptions import InvalidFileException
_COMPONENT_ROOT = Path(__file__).resolve().parents[1]
_VENDOR_CANDIDATES = (
_COMPONENT_ROOT / "vendor" / "modules",
_COMPONENT_ROOT.parent.parent / "vendor" / "modules",
_COMPONENT_ROOT.parent / "_shared" / "vendor" / "modules",
)
_VERA_ASSURANCE_ROOT: Path | None = None
for _vendor_candidate in _VENDOR_CANDIDATES:
if (_vendor_candidate / "vera_assurance").is_dir():
_VERA_ASSURANCE_ROOT = _vendor_candidate / "vera_assurance"
if str(_vendor_candidate) not in sys.path:
sys.path.insert(0, str(_vendor_candidate))
break
if _VERA_ASSURANCE_ROOT is None:
raise ImportError("Cannot locate the vendored vera_assurance implementation.")
from vera_assurance import (
MoneyValidationError,
artifact_receipt,
build_allocation_ledger,
build_assurance_envelope,
build_gate_register,
build_reviewed_decision_receipt,
build_source_qualification,
canonical_json_sha256,
decimal_text,
difference_within_tolerance,
file_snapshot,
parse_canonical_decimal,
parse_localized_decimal,
validate_allocation_ledger,
validate_artifact_receipt,
validate_assurance_envelope,
validate_gate_register,
validate_reviewed_decision_receipt,
validate_source_qualification,
)
from vera_assurance import write_json as write_assurance_json # noqa: E402
try:
from .review_session import (
FINAL_ARTIFACT_EXCLUDED_NAMES,
refresh_final_artifacts,
refresh_review_execution_trace,
review_notes_copy,
write_review_session_artifacts,
write_run_intake,
)
except ImportError: # pragma: no cover - supports direct script imports
import importlib.util
_review_session_path = Path(__file__).resolve().parent / "review_session.py"
_review_session_spec = importlib.util.spec_from_file_location(
"mparanza_journal_bank_reconciliation_review_session",
_review_session_path,
)
if _review_session_spec is None or _review_session_spec.loader is None:
raise ImportError(f"Cannot load review session helper: {_review_session_path}")
_review_session = importlib.util.module_from_spec(_review_session_spec)
sys.modules[_review_session_spec.name] = _review_session
_review_session_spec.loader.exec_module(_review_session)
FINAL_ARTIFACT_EXCLUDED_NAMES = _review_session.FINAL_ARTIFACT_EXCLUDED_NAMES
review_notes_copy = _review_session.review_notes_copy
refresh_final_artifacts = _review_session.refresh_final_artifacts
refresh_review_execution_trace = _review_session.refresh_review_execution_trace
write_review_session_artifacts = _review_session.write_review_session_artifacts
write_run_intake = _review_session.write_run_intake
LOGGER = logging.getLogger(__name__)
SUPPORTED_LANGUAGES = ("it", "en", "fr", "de", "es")
SUPPORTED_INPUT_SUFFIXES = {".csv", ".xls", ".xlsx", ".xlsm", ".pdf"}
TABULAR_ADAPTER_VERSION = "6"
TABULAR_ADAPTER_ID = "journal_bank.tabular.v6"
EXTENDED_TABULAR_ADAPTER_VERSION = "7"
EXTENDED_TABULAR_ADAPTER_ID = "journal_bank.tabular.v7"
FRENCH_TABULAR_ADAPTER_ID = "journal_bank.tabular.v8"
FRENCH_TABULAR_ADAPTER_VERSION = "8"
TEXT_PDF_ADAPTER_VERSION = "2"
TEXT_PDF_ADAPTER_ID = "journal_bank.text_pdf.disabled.v2"
PDF_TABLE_ADAPTER_VERSION = "1"
PDF_TABLE_ADAPTER_ID = "journal_bank.pdf_table.v1"
RELATIONSHIP_ADAPTER_ID = "journal_bank.relationship.v3"
RELATIONSHIP_ADAPTER_VERSION = "3"
NORMALIZATION_SCHEMA_VERSION = "journal_bank.normalization.v2"
LINEAGE_SCHEMA_VERSION = "journal_bank.lineage.v1"
ASSURANCE_WORKFLOW_ID = "journal_bank_reconciliation"
ASSURANCE_WORKFLOW_VERSION = "2"
FINAL_ARTIFACT_CLOSURE_MAX_PASSES = 8
MATCH_STAGE_ORDER = (
"reference",
"reference_group",
"amount_date_unique",
"amount_date_single",
)
WORKBOOK_SHEET_ORDER = (
"matches",
"relationship_residuals",
"unmatched_bank",
"unmatched_journal",
"bank_pdf_non_movements",
"normalized_bank",
"normalized_journal",
)
NATIVE_OUTPUT_FILES = (
"normalized_bank.csv",
"normalized_journal.csv",
"reconciliation_matches.csv",
"relationship_residuals.csv",
"unmatched_bank.csv",
"unmatched_journal.csv",
"bank_pdf_non_movement_rows.csv",
"journal_bank_reconciliation.xlsx",
"reconciliation_audit.json",
"review_notes.md",
"input_receipts.json",
"source_qualifications.json",
"reviewed_decisions.json",
"lineage.json",
"relationship_ledger.json",
"material_value_ledger.json",
"assurance_gates.json",
"artifact_receipts.json",
"assurance_envelope.json",
"run_intake.json",
"review_payload.json",
"ui_decisions.json",
"final_artifacts.json",
"review_handoff.md",
"applied_decisions.json",
)
INITIAL_RUN_OUTPUT_FILES = NATIVE_OUTPUT_FILES[:-1]
POST_REVIEW_OUTPUT_FILES = ("applied_decisions.json",)
DETERMINISTIC_ARTIFACT_FILES = (
"normalized_bank.csv",
"normalized_journal.csv",
"reconciliation_matches.csv",
"unmatched_bank.csv",
"unmatched_journal.csv",
"bank_pdf_non_movement_rows.csv",
"journal_bank_reconciliation.xlsx",
"review_notes.md",
"input_receipts.json",
"source_qualifications.json",
"reviewed_decisions.json",
"lineage.json",
"relationship_ledger.json",
"assurance_gates.json",
)
MATERIAL_CLOSURE_FILES = (
"relationship_residuals.csv",
"material_value_ledger.json",
)
IMPLEMENTATION_ARTIFACT_SPECS = (
(
"implementation.plugin.codex_plugin.plugin_json",
"implementation",
".codex-plugin/plugin.json",
),
(
"implementation.plugin.app_json",
"implementation",
".app.json",
),
(
"implementation.plugin.mcp_json",
"implementation",
".mcp.json",
),
(
"implementation.plugin.assets.icon_svg",
"implementation",
"assets/icon.svg",
),
(
"implementation.plugin.assets.journal_bank_review_widget_html",
"implementation",
"assets/journal-bank-review-widget.html",
),
(
"implementation.plugin.assets.review_workbench_adapter_json",
"implementation",
"assets/review-workbench-adapter.json",
),
(
"implementation.plugin.mcp.server_cjs",
"implementation",
"mcp/server.cjs",
),
(
"implementation.plugin.scripts.apply_review_edits_py",
"implementation",
"scripts/apply_review_edits.py",
),
(
"implementation.plugin.scripts.check_dependencies_py",
"implementation",
"scripts/check_dependencies.py",
),
(
"implementation.plugin.scripts.excel_sanitization_py",
"implementation",
"scripts/excel_sanitization.py",
),
(
"implementation.plugin.scripts.implementation_bootstrap_py",
"implementation",
"scripts/implementation_bootstrap.py",
),
(
"implementation.plugin.scripts.inspect_inputs_py",
"implementation",
"scripts/inspect_inputs.py",
),
(
"implementation.plugin.scripts.journal_bank_core_py",
"implementation",
"scripts/journal_bank_core.py",
),
(
"implementation.plugin.scripts.review_session_py",
"implementation",
"scripts/review_session.py",
),
(
"implementation.plugin.scripts.run_reconciliation_py",
"implementation",
"scripts/run_reconciliation.py",
),
(
"implementation.plugin.scripts.semantic_review_py",
"implementation",
"scripts/semantic_review.py",
),
(
"implementation.shared.vera_assurance.init_py",
"shared_implementation",
"__init__.py",
),
(
"implementation.shared.vera_assurance.contracts_py",
"shared_implementation",
"contracts.py",
),
(
"implementation.shared.vera_assurance.decisions_py",
"shared_implementation",
"decisions.py",
),
(
"implementation.shared.vera_assurance.envelope_py",
"shared_implementation",
"envelope.py",
),
(
"implementation.shared.vera_assurance.jurisdiction_py",
"shared_implementation",
"jurisdiction.py",
),
(
"implementation.shared.vera_assurance.money_py",
"shared_implementation",
"money.py",
),
(
"implementation.shared.vera_assurance.relationships_py",
"shared_implementation",
"relationships.py",
),
(
"implementation.shared.vera_assurance.review_output_transaction_cjs",
"shared_implementation",
"review_output_transaction.cjs",
),
(
"implementation.shared.vera_assurance.serialization_py",
"shared_implementation",
"serialization.py",
),
)
CHATGPT_IMPLEMENTATION_ARTIFACT_SPECS = tuple(
spec
for spec in IMPLEMENTATION_ARTIFACT_SPECS
if spec[2] not in {".app.json", ".mcp.json", "mcp/server.cjs"}
) + (
(
"implementation.plugin.scripts.review_mcp_server_cjs",
"implementation",
"scripts/review_mcp_server.cjs",
),
(
"implementation.plugin.scripts.review_server_py",
"implementation",
"scripts/review_server.py",
),
)
RUN_SCOPED_ARTIFACT_FILES = (
"reconciliation_audit.json",
"artifact_receipts.json",
"assurance_envelope.json",
"run_intake.json",
"review_payload.json",
"ui_decisions.json",
"final_artifacts.json",
"review_handoff.md",
"applied_decisions.json",
)
OOXML_TIMESTAMP = b"2000-01-01T00:00:00Z"
OOXML_ZIP_TIMESTAMP = (1980, 1, 1, 0, 0, 0)
OOXML_CORE_TIMESTAMP_RE = re.compile(
rb"(<dcterms:(created|modified)\b[^>]*>).*?(</dcterms:\2>)",
flags=re.DOTALL,
)
ZERO = Decimal("0")
CANONICAL_DIRECTIONS = frozenset({"positive", "negative", "zero"})
CSV_FIELD_DELIMITERS = (",", ";", "\t", "|")
DEFAULT_CSV_FIELD_DELIMITER = ","
DATE_CONVENTIONS = ("day_first", "month_first")
DATE_LOCALES = ("it", "fr")
FRENCH_MONTH_NUMBERS = {
"janvier": 1,
"janv": 1,
"fevrier": 2,
"fevr": 2,
"mars": 3,
"avril": 4,
"avr": 4,
"mai": 5,
"juin": 6,
"juillet": 7,
"juil": 7,
"aout": 8,
"septembre": 9,
"sept": 9,
"octobre": 10,
"oct": 10,
"novembre": 11,
"nov": 11,
"decembre": 12,
"dec": 12,
}
ITALIAN_MONTH_NUMBERS = {
"gennaio": 1,
"febbraio": 2,
"marzo": 3,
"aprile": 4,
"maggio": 5,
"giugno": 6,
"luglio": 7,
"agosto": 8,
"settembre": 9,
"ottobre": 10,
"novembre": 11,
"dicembre": 12,
}
MATERIAL_VALUE_LEDGER_SCHEMA_VERSION = "journal_bank.material_value_ledger.v1"
CSV_DELIMITER_PROFILE_MAX_BYTES = 128 * 1024
CSV_DELIMITER_PROFILE_MAX_ROWS = 100
CSV_TRANSPORT_CHUNK_BYTES = 64 * 1024
_RECIPE_CONTAINER_ERROR_FIELD = "_recipe_container_error"
SOURCE_ROW_COLUMN = "__source_row__"
SOURCE_SHEET_COLUMN = "__source_sheet__"
TRANSACTION_COLUMNS = [
"side",
"transaction_id",
"transaction_date",
"amount_signed",
"amount_abs",
"description",
"beneficiary",
"reference",
"movement_number",
"account",
"currency",
"unit",
"entity_ref",
"party_ref",
"direction",
"source_file",
"source_sheet",
"source_row",
]
NON_MOVEMENT_COLUMNS = [
"side",
"source_file",
"source_sheet",
"source_row",
"classification",
"reason",
"transaction_date",
"amount_signed",
"amount_abs",
"description",
]
EXACT_HEADER_ALIASES: dict[str, frozenset[str]] = {
"date": frozenset(
{
"date",
"data",
"datum",
"booking date",
"transaction date",
"data registrazione",
}
),
"amount": frozenset(
{
"amount",
"importo",
"montant",
"betrag",
"signed amount",
"importo movimento",
}
),
"debit": frozenset({"debit", "dare", "addebito", "soll", "importo dare"}),
"credit": frozenset(
{
"credit",
"avere",
"credito",
"haben",
"accredito",
"importo avere",
}
),
"description": frozenset(
{
"description",
"descrizione",
"causale",
"libelle",
"libellé",
"beschreibung",
"narrative",
"details",
}
),
"beneficiary": frozenset(
{
"beneficiary",
"beneficiario",
"payee",
"payer",
"counterparty",
"contrepartie",
"beguenstigter",
"begünstigter",
}
),
"reference": frozenset(
{
"reference",
"riferimento",
"document reference",
"documento",
"cro",
"trn",
}
),
"movement_number": frozenset(
{
"movement",
"movement number",
"movement_number",
"movimento",
"numero movimento",
"numero registrazione",
"n. registrazione",
"nr. reg",
"n. reg",
"beleg",
}
),
"account": frozenset({"account", "conto", "konto"}),
"currency": frozenset({"currency", "valuta", "devise", "wahrung", "währung"}),
"unit": frozenset({"unit", "unita", "unità", "einheit"}),
"entity_ref": frozenset(
{
"entity",
"entity ref",
"entity_ref",
"societa",
"società",
"company",
"legal entity",
}
),
"party_ref": frozenset(
{
"party",
"party ref",
"party_ref",
"counterparty id",
"controparte",
"soggetto",
}
),
"direction": frozenset(
{"direction", "flow direction", "segno", "verso", "richtung"}
),
}
REFERENCE_GENERIC_TOKENS = frozenset(
{
"invoice",
"fattura",
"payment",
"pagamento",
"transfer",
"bonifico",
"reference",
"riferimento",
"document",
"documento",
"transaction",
"movimento",
"credit",
"debit",
"bank",
"banca",
"statement",
"estratto",
"year",
"anno",
}
)
GENERIC_PERIOD_REFERENCE_PREFIXES = (
"fy",
"fq",
"fiscal",
"financial",
"fiscalyear",
"financialyear",
"fiscalperiod",
"financialperiod",
"accountingperiod",
"reportingperiod",
"period",
"esercizio",
"exercice",
"quarter",
)
_GENERIC_PERIOD_MARKER = "(?:" + "|".join(GENERIC_PERIOD_REFERENCE_PREFIXES) + ")"
_GENERIC_PERIOD_QUALIFIER = (
r"(?:actuals?|budget(?:ed)?|forecast(?:ed)?|quarter(?:ly)?|q[1-4])"
)
_GENERIC_PERIOD_YEAR = r"(?:19|20)\d{2}(?:(?:19|20)\d{2}|\d{2})?"
GENERIC_PERIOD_REFERENCE_RE = re.compile(
rf"(?:"
rf"{_GENERIC_PERIOD_MARKER}(?:{_GENERIC_PERIOD_QUALIFIER})*"
rf"(?:fy)?{_GENERIC_PERIOD_YEAR}(?:{_GENERIC_PERIOD_QUALIFIER})*"
rf"(?:{_GENERIC_PERIOD_MARKER})?"
rf"|q[1-4](?:fy)?{_GENERIC_PERIOD_YEAR}"
rf"(?:{_GENERIC_PERIOD_QUALIFIER})*"
rf"|{_GENERIC_PERIOD_YEAR}(?:{_GENERIC_PERIOD_QUALIFIER})+"
rf"|(?:{_GENERIC_PERIOD_QUALIFIER})+(?:fy)?{_GENERIC_PERIOD_YEAR}"
rf"(?:{_GENERIC_PERIOD_MARKER})?"
rf")"
)
GENERIC_PERIOD_WORDS = frozenset(
{
*GENERIC_PERIOD_REFERENCE_PREFIXES,
"actual",
"actuals",
"budget",
"budgeted",
"forecast",
"forecasted",
"q1",
"q2",
"q3",
"q4",
"quarterly",
"month",
"monthly",
"week",
"weekly",
"half",
"semester",
"sem",
"jan",
"january",
"feb",
"february",
"mar",
"march",
"apr",
"april",
"may",
"jun",
"june",
"jul",
"july",
"aug",
"august",
"sep",
"sept",
"september",
"oct",
"october",
"nov",
"november",
"dec",
"december",
}
)
GENERIC_PERIOD_VERSION_RE = re.compile(r"(?:version|revision|draft|final|ver|rev|v)\d*")
GENERIC_PERIOD_CODE_RE = re.compile(
r"(?:"
r"(?:month|m)(?:0?[1-9]|1[0-2])"
r"|(?:week|wk|w)(?:0?[1-9]|[1-4]\d|5[0-3])"
r"|(?:half|h|semester|sem)[12]"
r"|q[1-4]"
r")"
)
MONETARY_HEADER_TOKENS = (
"amount",
"importo",
"dare",
"avere",
"debit",
"débit",
"credit",
"crédit",
"soll",
"haben",
"montant",
"betrag",
"saldo",
"balance",
"fee",
"commission",
"commissione",
"tax",
"imposta",
"ritenuta",
"total",
"totale",
)
MATCH_COLUMNS = [
"status",
"stage",
"bank_transaction_id",
"journal_transaction_id",
"bank_date",
"journal_date",
"date_diff_days",
"bank_amount",
"journal_amount",
"amount_delta",
"bank_description",
"journal_description",
"shared_references",
"review_note",
]
MATCH_MATERIAL_FIELDS = (
"status",
"stage",
"bank_transaction_id",
"journal_transaction_id",
"bank_date",
"journal_date",
"date_diff_days",
"bank_amount",
"journal_amount",
"amount_delta",
"shared_references",
)
RESIDUAL_COLUMNS = [
"side",
"record_ref",
"transaction_id",
"record_amount",
"allocated_amount",
"residual",
"currency",
"unit",
"entity_ref",
"party_ref",
]
RESIDUAL_MATERIAL_FIELDS = tuple(RESIDUAL_COLUMNS)
MATERIAL_DECIMAL_FIELDS = frozenset(
{
"bank_amount",
"journal_amount",
"amount_delta",
"record_amount",
"allocated_amount",
"residual",
}
)
MATERIAL_INTEGER_FIELDS = frozenset({"date_diff_days"})
COMPACT_ISO_DATE_RE = re.compile(r"^(?P<year>\d{4})(?P<month>\d{2})(?P<day>\d{2})$")
YEAR_FIRST_DATE_RE = re.compile(
r"^(?P<year>\d{4})(?P<separator>[./-])"
r"(?P<month>\d{1,2})(?P=separator)(?P<day>\d{1,2})$"
)
DAY_MONTH_DATE_RE = re.compile(
r"^(?P<first>\d{1,2})(?P<separator>[./-])"
r"(?P<second>\d{1,2})(?P=separator)(?P<year>\d{2}|\d{4})$"
)
TEXTUAL_MONTH_DATE_RE = re.compile(
r"^(?P<day>\d{1,2})[ \t]+(?P<month>[^\W\d_]+)\.?" r"[ \t]+(?P<year>\d{4})$",
flags=re.UNICODE,
)
AMOUNT_TOKEN_RE = re.compile(
r"(?<!\w)\(?-?\d{1,3}(?:[.,\s]\d{3})*(?:[.,]\d{2})\)?(?!\w)"
)
BANK_PDF_NON_MOVEMENT_PATTERNS = (
(
"balance",
"Bank statement balance line, not an ordinary bank movement.",
re.compile(
r"\b(?:"
r"saldo iniziale|saldo finale|saldo precedente|saldo contabile|"
r"saldo disponibile|saldo al|saldo a inizio|saldo a fine|"
r"saldo per|"
r"opening balance|closing balance|initial balance|final balance|"
r"previous balance|balance brought forward|balance carried forward|"
r"solde initial|solde final|solde au|"
r"anfangssaldo|endsaldo|anfangsbestand|schlussbestand|kontostand"
r")\b"
),
),
(
"total",
"Bank statement total line, not an ordinary bank movement.",
re.compile(
r"^(?:totale|totali|total|totaux|summe|gesamtbetrag)\s+"
r"(?:(?:de|del|della|dei|degli|des|du|der|die|of)\s+)?"
r"(?:entrate|uscite|accrediti|addebiti|versamenti|prelievi|"
r"movimenti|competenze|spese|commissioni|dare|avere|a credito|"
r"a debito|debits?|credits?|incoming|outgoing|transactions?|"
r"operations?|charges?|fees?|interests?|interets?|"
r"gutschriften|belastungen|gebuhren|gebuehren|soll|haben|zinsen)\b"
),
),
(
"scalare",
"Bank statement scalare/interest summary line, not an ordinary bank movement.",
re.compile(
r"\b(?:"
r"scalare|riassunto scalare|riepilogo scalare|numeri creditori|"
r"numeri debitori|calcolo competenze|riepilogo competenze|"
r"competenze scalari|interest scale|interest summary|"
r"echelle d interets|echelle interets|zinsstaffel|zinsenstaffel"
r")\b"
),
),
(
"conditions",
"Bank statement conditions line, not an ordinary bank movement.",
re.compile(
r"\b(?:"
r"condizioni economiche|condizioni applicate|riepilogo condizioni|"
r"condizioni del conto|conditions applied|account conditions|"
r"conditions economiques|conditions appliquees|"
r"kontokonditionen|konditionen|wirtschaftliche bedingungen"
r")\b"
),
),
)
__all__ = [
"CANONICAL_DIRECTIONS",
"CSV_FIELD_DELIMITERS",
"CSV_TRANSPORT_CHUNK_BYTES",
"DATE_CONVENTIONS",
"DATE_LOCALES",
"DEFAULT_CSV_FIELD_DELIMITER",
"DETERMINISTIC_ARTIFACT_FILES",
"IMPLEMENTATION_ARTIFACT_SPECS",
"INITIAL_RUN_OUTPUT_FILES",
"InspectionResult",
"ITALIAN_MONTH_NUMBERS",
"MAPPING_FIELDS",
"MATERIAL_CLOSURE_FILES",
"MATCH_COLUMNS",
"MATCH_MATERIAL_FIELDS",
"MATCH_STAGE_ORDER",
"NATIVE_OUTPUT_FILES",
"NON_MOVEMENT_COLUMNS",
"PDF_TABLE_ADAPTER_ID",
"PDF_TABLE_ADAPTER_VERSION",
"EXTENDED_TABULAR_ADAPTER_ID",
"EXTENDED_TABULAR_ADAPTER_VERSION",
"POST_REVIEW_OUTPUT_FILES",
"ReconciliationBlockedError",
"ReconciliationRunResult",
"RELATIONSHIP_ADAPTER_ID",
"RELATIONSHIP_ADAPTER_VERSION",
"RELATIONSHIP_POLICY_FIELDS",
"RUN_SCOPED_ARTIFACT_FILES",
"RESIDUAL_COLUMNS",
"RESIDUAL_MATERIAL_FIELDS",
"TABULAR_ADAPTER_ID",
"TABULAR_ADAPTER_VERSION",
"TRANSACTION_COLUMNS",
"WORKBOOK_SHEET_ORDER",
"add_common_args",
"build_mapping_review_receipt",
"build_implementation_artifact_receipts",
"build_relationship_review_receipt",
"configure_logging",
"excel_safe_value",
"format_workbook_sheet",
"inspect_inputs",
"implementation_artifact_roots",
"normalize_language",
"run_reconciliation",
"validate_exact_implementation_receipts",
"validate_material_value_ledger",
"write_json",
]
MAPPING_FIELDS = (
"date",
"amount",
"debit",
"credit",
"description",
"beneficiary",
"reference",
"movement_number",
"account",
"currency",
"unit",
"entity_ref",
"party_ref",
"direction",
)
RELATIONSHIP_POLICY_FIELDS = {
"relationship_shape",
"allow_evidence_reuse",
"require_same_currency",
"require_same_unit",
"require_same_entity",
"require_same_party",
"direction_policy",
"default_currency",
"default_unit",
"default_entity_ref",
"default_party_ref",
"amount_tolerance",
"date_window_days",
}
class ReconciliationBlockedError(ValueError):
"""Raised after evidence artifacts are written for a fail-closed run."""
def __init__(self, code: str, detail: str) -> None:
super().__init__(detail)
self.code = code
self.detail = detail
@dataclass(frozen=True)
class InspectionResult:
"""Deterministic inspection output for bank and journal inputs."""
bank: dict[str, Any]
journal: dict[str, Any]
sample: dict[str, Any]
suggested_recipe: dict[str, Any]
@dataclass(frozen=True)
class ReconciliationRunResult:
"""Reconciliation output plus audit metadata."""
matches: pl.DataFrame
unmatched_bank: pl.DataFrame
unmatched_journal: pl.DataFrame
audit: dict[str, Any]
def configure_logging(verbose: bool = False) -> None:
"""Configure script logging without affecting imported use."""
level = logging.DEBUG if verbose else logging.INFO
logging.basicConfig(level=level, format="%(levelname)s: %(message)s")
def normalize_language(
language: object | None,
*,
default: str = "en",
allow_auto: bool = False,
) -> str:
"""Normalize a language tag to one supported plugin locale."""
text = str(language or default).strip().lower().replace("_", "-")
code = text.split("-", 1)[0]
if allow_auto and code == "auto":
return "auto"
return code if code in SUPPORTED_LANGUAGES else default
def language_assumptions(
recipe: dict[str, Any],
*,
language: object | None = None,
document_language: object | None = None,
) -> dict[str, str]:
"""Resolve working and source-document language assumptions."""
working = normalize_language(language or recipe.get("language"), default="en")
source = normalize_language(
document_language or recipe.get("document_language") or "auto",
default=working,
allow_auto=True,
)
return {"language": working, "document_language": source}
def write_json(path: Path, payload: Any) -> None:
"""Write JSON with stable formatting."""
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
)
def read_json(path: Path | None) -> dict[str, Any]:
"""Return a JSON object or an empty mapping when no file is provided."""
if path is None:
return {}
payload = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(payload, dict):
raise ValueError(f"Recipe must be a JSON object: {path}")
return payload
def _clean_text(value: Any) -> str:
if value is None:
return ""
return str(value).replace("\u00a0", " ").replace("\u202f", " ").strip()
def _norm_label(value: Any) -> str:
text = unicodedata.normalize("NFKD", _clean_text(value))
text = "".join(ch for ch in text if not unicodedata.combining(ch))
text = text.lower()
return re.sub(r"\s+", " ", text)
def _identifier_fragment(value: object) -> str:
text = re.sub(r"[^A-Za-z0-9._-]+", "_", _clean_text(value)).strip("._-")
return text or "source"
def _excel_column_name(index: int) -> str:
idx = index + 1
letters: list[str] = []
while idx > 0:
idx, rem = divmod(idx - 1, 26)
letters.append(chr(65 + rem))
return "".join(reversed(letters))
def _unique_names(values: Sequence[Any]) -> list[str]:
names: list[str] = []
seen: dict[str, int] = {}
for idx, value in enumerate(values):
text = _clean_text(value)
base = (
text
if text and text.lower() not in {"none", "nan"}
else _excel_column_name(idx)
)
count = seen.get(base, 0)
seen[base] = count + 1
names.append(base if count == 0 else f"{base}_{count + 1}")
return names
def supported_files(input_path: Path) -> list[Path]:
"""Return supported files from a file or folder path."""
path = input_path.expanduser()
if path.is_file():
return [path] if path.suffix.lower() in SUPPORTED_INPUT_SUFFIXES else []
if not path.exists():
raise FileNotFoundError(f"Input path does not exist: {path}")
return [
candidate
for candidate in sorted(path.rglob("*"))
if candidate.is_file()
and candidate.suffix.lower() in SUPPORTED_INPUT_SUFFIXES
and not candidate.name.startswith("~$")
]
def _input_root(input_path: Path) -> Path:
path = input_path.expanduser()
return path if path.is_dir() else path.parent
def _source_identity(input_path: Path, source: Path) -> str:
"""Return a collision-safe, input-root-relative source identity."""
root = _input_root(input_path).resolve()
resolved = source.resolve()
if not resolved.is_relative_to(root):
raise ValueError(f"Source escapes its declared input root: {source}")
return resolved.relative_to(root).as_posix()
def _source_identity_fragment(source_identity: str) -> str:
path_text = Path(source_identity).with_suffix("").as_posix()
readable = _identifier_fragment(path_text)
digest = hashlib.sha256(source_identity.encode("utf-8")).hexdigest()[:12]
return f"{readable}.{digest}"
def _read_excel_raw(path: Path) -> pl.DataFrame:
if path.suffix.lower() in {".xlsx", ".xlsm"}:
workbook = openpyxl.load_workbook(path, data_only=True, read_only=False)
try:
sheet = workbook.worksheets[0]
rows = [
[
(
value.date().isoformat()
if isinstance(value, datetime)
else value.isoformat() if isinstance(value, date) else value
)
for value in row
]
for row in sheet.iter_rows(values_only=True)
]
finally:
workbook.close()
max_width = max((len(row) for row in rows), default=0)
if max_width == 0:
return pl.DataFrame()
return pl.DataFrame(
{
f"column_{idx}": [row[idx] if idx < len(row) else None for row in rows]
for idx in range(max_width)
},
strict=False,
)
try:
return pl.read_excel(
path,
has_header=False,
drop_empty_rows=False,
drop_empty_cols=False,
)
except (fastexcel.CalamineError, ValueError, RuntimeError, OSError) as exc:
raise ValueError(f"Unable to read legacy Excel source: {path}") from exc
def _bounded_csv_profile_text(path: Path) -> str:
"""Read a bounded UTF-8 prefix for mechanically reproducible delimiter checks."""
with path.open("rb") as handle:
payload = handle.read(CSV_DELIMITER_PROFILE_MAX_BYTES + 1)
if len(payload) > CSV_DELIMITER_PROFILE_MAX_BYTES:
payload = payload[:CSV_DELIMITER_PROFILE_MAX_BYTES]
boundary = max(payload.rfind(b"\n"), payload.rfind(b"\r"))
if boundary < 0:
return ""
payload = payload[: boundary + 1]
return payload.decode("utf-8-sig")
def _profile_csv_field_delimiter(path: Path) -> dict[str, Any]:
"""Profile one unique consistent delimiter without assigning review authority.
This rule is deterministic because it checks only the fixed supported byte
set, strict CSV record structure, and bounded row/byte limits.
"""
text = _bounded_csv_profile_text(path)
candidates: list[str] = []
parse_error_delimiters: list[str] = []
sampled_row_count = 0
for delimiter in CSV_FIELD_DELIMITERS:
try:
rows: list[list[str]] = []
reader = csv.reader(
io.StringIO(text, newline=""),
delimiter=delimiter,
strict=True,
)
for row in reader:
if not any(_clean_text(value) for value in row):
continue
rows.append(row)
if len(rows) >= CSV_DELIMITER_PROFILE_MAX_ROWS:
break
except csv.Error:
parse_error_delimiters.append(delimiter)
continue
sampled_row_count = max(sampled_row_count, len(rows))
widths = [len(row) for row in rows]
if len(widths) >= 2 and widths[0] > 1 and len(set(widths)) == 1:
candidates.append(delimiter)
status = (
"unique"
if len(candidates) == 1
else "ambiguous" if len(candidates) > 1 else "unsupported"
)
return {
"status": status,
"candidate_delimiters": candidates,
"parse_error_delimiters": parse_error_delimiters,
"sampled_row_count": sampled_row_count,
"max_bytes": CSV_DELIMITER_PROFILE_MAX_BYTES,
"max_rows": CSV_DELIMITER_PROFILE_MAX_ROWS,
}
def _validated_csv_field_delimiter(value: object) -> str:
if (
not isinstance(value, str)
or value not in CSV_FIELD_DELIMITERS
or len(value.encode("utf-8")) != 1
):
raise ValueError(
"csv_field_delimiter must be one reviewed byte from comma, "
"semicolon, tab, or pipe"
)
return value
def _mapping_csv_field_delimiter(
source_file: str,
value: object | None,
) -> str | None:
is_csv = Path(source_file).suffix.lower() == ".csv"
if not is_csv:
if value is not None:
raise ValueError("csv_field_delimiter is only valid for CSV sources")
return None
if value is None:
return DEFAULT_CSV_FIELD_DELIMITER
return _validated_csv_field_delimiter(value)
def _resolve_csv_field_delimiter(
path: Path,
file_recipe: dict[str, Any],
) -> dict[str, Any]:
if path.suffix.lower() != ".csv":
return {
"status": "not_applicable",
"delimiter": None,
"origin": "not_applicable",
"review_required": False,
"profile": None,
"requested_delimiter": None,
}
profile = _profile_csv_field_delimiter(path)
if (
"csv_field_delimiter" in file_recipe
and file_recipe.get("csv_field_delimiter") is not None
):
requested = file_recipe.get("csv_field_delimiter")
try:
delimiter = _validated_csv_field_delimiter(requested)
except ValueError:
return {
"status": "unsupported",
"delimiter": None,
"origin": "reviewed_recipe",
"review_required": True,
"profile": profile,
"requested_delimiter": requested,
}
profile_matches = profile["status"] == "unique" and profile[
"candidate_delimiters"
] == [delimiter]
requested_parse_failed = delimiter in profile["parse_error_delimiters"]
if (
profile["status"] in {"ambiguous", "unsupported"}
and not requested_parse_failed
and not isinstance(file_recipe.get("mapping_decision"), dict)
):
return {
"status": str(profile["status"]),
"delimiter": None,
"origin": "reviewed_recipe",
"review_required": True,
"profile": profile,
"requested_delimiter": requested,
}
return {
"status": "resolved",
"delimiter": delimiter,
"origin": "reviewed_recipe",
"review_required": (
delimiter != DEFAULT_CSV_FIELD_DELIMITER or not profile_matches
),
"profile": profile,
"requested_delimiter": requested,
}
if profile["status"] == "unique":
delimiter = str(profile["candidate_delimiters"][0])
return {
"status": "resolved",
"delimiter": delimiter,
"origin": (
"default"
if delimiter == DEFAULT_CSV_FIELD_DELIMITER
else "profiled_proposal"
),
"review_required": delimiter != DEFAULT_CSV_FIELD_DELIMITER,
"profile": profile,
"requested_delimiter": None,
}
return {
"status": str(profile["status"]),
"delimiter": None,
"origin": "profiled_proposal",
"review_required": True,
"profile": profile,
"requested_delimiter": None,
}
def _normalize_csv_record_terminators(source: Path, target: Path) -> None:
"""Normalize universal-newline byte forms without loading the file in memory."""
pending_cr = False
with source.open("rb") as source_handle, target.open("wb") as target_handle:
while chunk := source_handle.read(CSV_TRANSPORT_CHUNK_BYTES):
if pending_cr:
target_handle.write(b"\n")
if chunk.startswith(b"\n"):
chunk = chunk[1:]
pending_cr = False
if chunk.endswith(b"\r"):
chunk = chunk[:-1]
pending_cr = True
target_handle.write(chunk.replace(b"\r\n", b"\n").replace(b"\r", b"\n"))
if pending_cr:
target_handle.write(b"\n")
def _read_csv_raw(path: Path, *, field_delimiter: str) -> pl.DataFrame:
# Record terminators are transport syntax, not reviewed mapping intent.
# A private streamed copy keeps CRLF split across chunks intact and lets
# Polars own the full strict parse without duplicating a large CSV in RAM.
with tempfile.TemporaryDirectory(prefix="jbr-csv-transport-") as temp_dir:
normalized_path = Path(temp_dir) / "normalized.csv"
_normalize_csv_record_terminators(path, normalized_path)
delimiter = _validated_csv_field_delimiter(field_delimiter)
expected_width: int | None = None
try:
with normalized_path.open(
"r",
encoding="utf-8-sig",
newline="",
) as handle:
for row in csv.reader(handle, delimiter=delimiter, strict=True):
if not any(_clean_text(value) for value in row):
continue
if expected_width is None:
expected_width = len(row)
elif len(row) != expected_width:
raise ValueError("CSV strict full parse found a ragged record")
except csv.Error as exc:
raise ValueError("CSV strict full parse failed") from exc
return pl.read_csv(
normalized_path,
has_header=False,
infer_schema=False,
ignore_errors=False,
truncate_ragged_lines=False,
separator=delimiter,
)
def _read_table_raw(
path: Path,
*,
csv_field_delimiter: str | None = None,
) -> pl.DataFrame:
suffix = path.suffix.lower()
if suffix == ".csv":
return _read_csv_raw(
path,
field_delimiter=(
csv_field_delimiter
if csv_field_delimiter is not None
else DEFAULT_CSV_FIELD_DELIMITER
),
)
if suffix in {".xls", ".xlsx", ".xlsm"}:
return _read_excel_raw(path)
raise ValueError(f"Unsupported tabular file: {path}")
def _pdf_cell_text(value: object) -> str:
"""Normalize one extracted PDF table cell without assigning its meaning."""
return re.sub(r"\s+", " ", _clean_text(value)).strip()
def _pdf_page_has_unmapped_candidates(page_text: str) -> bool:
"""Detect pages whose transaction-like text was not captured as a table.
This check is mechanical: it withholds the entire source when a page has a
date and monetary token outside the bounded table extraction. It does not
decide whether the text is actually an accounting movement.
"""
date_token = re.compile(
r"\b(?:\d{4}[./-]\d{1,2}[./-]\d{1,2}|" r"\d{1,2}[./-]\d{1,2}[./-]\d{2,4})\b"
)
return any(
date_token.search(line) is not None and AMOUNT_TOKEN_RE.search(line) is not None
for line in page_text.splitlines()
)
def _pdf_header_has_required_roles(header: Sequence[str]) -> bool:
"""Require explicit date and monetary labels before proposing a PDF table."""
labels = {_norm_label(value) for value in header if _norm_label(value)}
monetary_labels = (
EXACT_HEADER_ALIASES["amount"]
| EXACT_HEADER_ALIASES["debit"]
| EXACT_HEADER_ALIASES["credit"]
| {"uscite", "entrate", "saldo", "balance"}
)
return bool(labels & EXACT_HEADER_ALIASES["date"]) and bool(
labels & monetary_labels
)
def _pdf_word_rows(
page: Any, *, top_tolerance: float = 3.0
) -> list[list[dict[str, Any]]]:
"""Group positioned PDF words into physical text lines."""
rows: list[list[dict[str, Any]]] = []
for word in sorted(
page.extract_words() or [],
key=lambda item: (float(item["top"]), float(item["x0"])),
):
if (
not rows
or abs(float(word["top"]) - float(rows[-1][0]["top"])) > top_tolerance
):
rows.append([word])
else:
rows[-1].append(word)
return rows
def _pdf_header_groups(
words: Sequence[dict[str, Any]],
*,
within_header_gap: float = 12.0,
) -> list[list[dict[str, Any]]]:
"""Group adjacent header words while keeping visibly separate columns apart."""
groups: list[list[dict[str, Any]]] = []
for word in sorted(words, key=lambda item: float(item["x0"])):
if (
not groups
or float(word["x0"]) - float(groups[-1][-1]["x1"]) > within_header_gap
):
groups.append([word])
else:
groups[-1].append(word)
return groups
def _extract_pdf_positioned_table(page: Any) -> list[list[str]] | None:
"""Recover an unruled table only from a labelled, positioned column grid."""
word_rows = _pdf_word_rows(page)
header_index: int | None = None
header_groups: list[list[dict[str, Any]]] = []
header: list[str] = []
for row_index, words in enumerate(word_rows):
groups = _pdf_header_groups(words)
labels = [" ".join(str(word["text"]) for word in group) for group in groups]
if len(labels) >= 2 and _pdf_header_has_required_roles(labels):
header_index = row_index
header_groups = groups
header = labels
break
if header_index is None:
return None
boundaries = [0.0]
for left_group, right_group in zip(header_groups, header_groups[1:]):
boundaries.append(
(float(left_group[-1]["x1"]) + float(right_group[0]["x0"])) / 2
)
boundaries.append(float(page.width))
body: list[list[str]] = []
for words in word_rows[header_index + 1 :]:
cells: list[list[str]] = [[] for _ in header]
for word in sorted(words, key=lambda item: float(item["x0"])):
center = (float(word["x0"]) + float(word["x1"])) / 2
column_index = next(
(
index
for index in range(len(header))
if boundaries[index] <= center < boundaries[index + 1]
),
len(header) - 1,
)
cells[column_index].append(str(word["text"]))
row = [" ".join(cell) for cell in cells]
if sum(bool(value) for value in row) >= 2:
body.append(row)
if not body:
return None
return [header, *body]
def _read_pdf_table_raw_unchecked(
path: Path,
) -> tuple[pl.DataFrame, list[tuple[str, int]], dict[str, Any]]:
"""Extract one repeated-header PDF table family with page-row lineage.
PDF table extraction is deterministic here because only physical cells,
exact repeated headers, and page/table coordinates are retained. Column
meaning and monetary ownership remain subject to a source-bound reviewed
mapping receipt.
"""
import pdfplumber
canonical_header: list[str] | None = None
rows: list[list[str]] = []
locators: list[tuple[str, int]] = []
page_row_counts: dict[str, int] = {}
page_count = 0
table_count = 0
candidate_pages_without_tables: list[int] = []
inconsistent_tables: list[str] = []
with pdfplumber.open(path) as document:
for page_number, page in enumerate(document.pages, start=1):
page_count = page_number
page_tables = page.extract_tables() or []
if not any(
table
and len(table) >= 2
and _pdf_header_has_required_roles(
[_pdf_cell_text(cell) for cell in table[0]]
)
for table in page_tables
):
positioned_table = _extract_pdf_positioned_table(page)
page_tables = [positioned_table] if positioned_table is not None else []
if not page_tables:
page_tables = (
page.extract_tables(
table_settings={
"vertical_strategy": "text",
"horizontal_strategy": "text",
"min_words_vertical": 2,
"min_words_horizontal": 1,
"intersection_tolerance": 5,
"text_tolerance": 3,
}
)
or []
)
usable_tables = []
for table_number, raw_table in enumerate(page_tables, start=1):
table_rows = [
[_pdf_cell_text(cell) for cell in row]
for row in (raw_table or [])
if row and any(_pdf_cell_text(cell) for cell in row)
]
if (
len(table_rows) >= 2
and len(table_rows[0]) >= 2
and _pdf_header_has_required_roles(table_rows[0])
):
usable_tables.append((table_number, table_rows))
if not usable_tables:
page_text = page.extract_text() or ""
if _pdf_page_has_unmapped_candidates(page_text):
candidate_pages_without_tables.append(page_number)
continue
for table_number, table_rows in usable_tables:
table_count += 1
header = table_rows[0]
if canonical_header is None:
canonical_header = header
if header != canonical_header:
inconsistent_tables.append(
f"page {page_number} table {table_number}"
)
continue
locator_sheet = f"PDF page {page_number} table {table_number}"
for row_number, row in enumerate(table_rows[1:], start=2):
if len(row) != len(canonical_header):
inconsistent_tables.append(
f"page {page_number} table {table_number} row {row_number}"
)
continue
rows.append(row)
locators.append((locator_sheet, row_number))
page_key = str(page_number)
page_row_counts[page_key] = page_row_counts.get(page_key, 0) + 1
failure_kind: str | None = None
limitations: list[str] = []
if candidate_pages_without_tables:
failure_kind = "unmapped_pdf_candidate_page"
limitations.append(
"Transaction-like text appeared on a page with no bounded extracted table."
)
elif canonical_header is None:
failure_kind = "pdf_table_not_found"
limitations.append(
"No bounded multi-column PDF table with a header and body rows was found."
)
elif inconsistent_tables:
failure_kind = "inconsistent_pdf_table_layout"
limitations.append(
"Every extracted table must use the exact same physical header and width."
)
elif not rows:
failure_kind = "empty_pdf_table"
limitations.append("The extracted PDF table contained no body rows.")
metadata = {
"page_count": page_count,
"table_count": table_count,
"page_row_counts": page_row_counts,
"candidate_pages_without_tables": candidate_pages_without_tables,
"inconsistent_tables": inconsistent_tables,
"failure_kind": failure_kind,
"limitations": limitations,
}
if failure_kind is not None or canonical_header is None:
return pl.DataFrame(), [], metadata
all_rows = [canonical_header, *rows]
raw = pl.DataFrame(
{
f"column_{index}": [row[index] for row in all_rows]
for index in range(len(canonical_header))
},
strict=False,
)
return raw, [("PDF header", 1), *locators], metadata
def _read_pdf_table_raw(
path: Path,
) -> tuple[pl.DataFrame, list[tuple[str, int]], dict[str, Any]]:
"""Run PDF table extraction without imposing pdfplumber on other workflows."""
from pdfplumber.utils.exceptions import PdfminerException
try:
return _read_pdf_table_raw_unchecked(path)
except PdfminerException as exc:
raise ValueError("PDF table parser could not read the source") from exc
def _drop_empty_columns(df: pl.DataFrame) -> pl.DataFrame:
if df.is_empty() or df.width == 0:
return df
keep: list[str] = []
for col in df.columns:
if col in {SOURCE_ROW_COLUMN, SOURCE_SHEET_COLUMN}:
keep.append(col)
continue
values = (
df.get_column(col)
.cast(pl.Utf8, strict=False)
.fill_null("")
.str.strip_chars()
)
if int((values != "").sum()) > 0:
keep.append(col)
return df.select(keep) if keep else pl.DataFrame()
def _drop_empty_rows(df: pl.DataFrame) -> pl.DataFrame:
if df.is_empty() or df.width == 0:
return df
data_columns = [
col for col in df.columns if col not in {SOURCE_ROW_COLUMN, SOURCE_SHEET_COLUMN}
]
if not data_columns:
return df.head(0)
exprs = [
pl.col(col).cast(pl.Utf8, strict=False).fill_null("").str.strip_chars() == ""
for col in data_columns
]
return df.filter(~pl.all_horizontal(exprs))
def _row_values(df: pl.DataFrame, idx: int) -> list[Any]:
return list(df.row(idx))
def _suggest_header_rows(df: pl.DataFrame) -> list[int]:
if df.is_empty():
return [1]
tokens = (
"data",
"date",
"datum",
"amount",
"importo",
"montant",
"betrag",
"dare",
"avere",
"debit",
"credit",
"soll",
"haben",
"descrizione",
"description",
"libelle",
"libellé",
"beschreibung",
"beneficiario",
"beneficiary",
"payee",
"iban",
"reference",
"riferimento",
"movimento",
"movement",
"conto",
"account",
)
best_idx = 0
best_score = -1
for idx in range(min(df.height, 30)):
row = _row_values(df, idx)
score = sum(1 for value in row if _clean_text(value))
for value in row:
label = _norm_label(value)
if any(token in label for token in tokens):
score += 3
elif any(ch.isalpha() for ch in label):
score += 1
if score > best_score:
best_score = score
best_idx = idx
return [best_idx + 1]
def _merge_header_rows(rows: Sequence[Sequence[Any]]) -> list[str]:
width = max((len(row) for row in rows), default=0)
labels: list[str] = []
for idx in range(width):
parts = []
for row in rows:
value = _clean_text(row[idx] if idx < len(row) else "")
if value and value.lower() not in {"none", "nan"}:
parts.append(value)
labels.append(" ".join(parts))
return _unique_names(labels)
def _apply_header(
df: pl.DataFrame,
rows_1_indexed: Sequence[int],
*,
source_sheet: str | None = None,
source_locators: Sequence[tuple[str, int]] | None = None,
) -> pl.DataFrame:
if df.is_empty():
return df
row_indexes = sorted({int(row) - 1 for row in rows_1_indexed})
if not row_indexes or min(row_indexes) < 0:
raise ValueError("Header rows must be 1-indexed positive integers.")
if max(row_indexes) >= df.height:
raise ValueError("Header row exceeds available rows.")
labels = _merge_header_rows([_row_values(df, idx) for idx in row_indexes])
body_start = max(row_indexes) + 1
body = df.slice(body_start)
if body.width != len(labels):
labels = _unique_names(labels[: body.width])
body.columns = labels
if source_locators is not None:
body_locators = list(source_locators[body_start:])
if len(body_locators) != body.height:
raise ValueError("PDF source locators do not match extracted table rows")
body = body.with_columns(
pl.Series(
SOURCE_ROW_COLUMN,
[row_number for _, row_number in body_locators],
dtype=pl.Int64,
),
pl.Series(
SOURCE_SHEET_COLUMN,
[sheet for sheet, _ in body_locators],
dtype=pl.Utf8,
),
)
else:
body = body.with_columns(
pl.Series(
SOURCE_ROW_COLUMN,
list(range(body_start + 1, body_start + 1 + body.height)),
dtype=pl.Int64,
),
pl.lit(source_sheet or "CSV").alias(SOURCE_SHEET_COLUMN),
)
# Preserve header-owned columns even when every body cell is blank. A
# debit/credit export commonly leaves one side empty for the entire file,
# but the explicit header still carries the monetary-role contract.
return _drop_empty_rows(body)
def _source_sheet_names(path: Path) -> list[str]:
if path.suffix.lower() == ".csv":
return ["CSV"]
if path.suffix.lower() == ".pdf":
return ["PDF"]
workbook = fastexcel.read_excel(path)
return [str(value) for value in workbook.sheet_names]
def _source_sheet_name(path: Path) -> str:
names = _source_sheet_names(path)
if len(names) != 1:
raise ValueError(
"Workbook must contain exactly one physical sheet for the bounded "
"tabular adapter."
)
return names[0]
def _exact_mapping_from_columns(
columns: Sequence[str],
) -> tuple[dict[str, str | None], list[str]]:
"""Resolve only exact, unique source-owned header labels.
This bounded lookup is deterministic because it never profiles cell values,
infers numeric position, or selects among duplicate semantic candidates.
"""
available = [
column
for column in columns
if column not in {SOURCE_ROW_COLUMN, SOURCE_SHEET_COLUMN}
]
normalized = {column: _norm_label(column) for column in available}
mapping: dict[str, str | None] = {}
ambiguous: list[str] = []
for field, aliases in EXACT_HEADER_ALIASES.items():
matches = [column for column, label in normalized.items() if label in aliases]
if len(matches) > 1:
ambiguous.append(field)
mapping[field] = None
else:
mapping[field] = matches[0] if matches else None
monetary_fields = [
field for field in ("amount", "debit", "credit") if mapping.get(field)
]
if mapping.get("amount") and (mapping.get("debit") or mapping.get("credit")):
ambiguous.append("monetary_role")
if not monetary_fields:
ambiguous.append("monetary_role")
if not mapping.get("date"):
ambiguous.append("date")
return mapping, sorted(set(ambiguous))
def _exact_header_contract(
raw: pl.DataFrame,
) -> tuple[list[int], dict[str, str | None]] | None:
"""Return one unambiguous bounded header contract, otherwise abstain."""
candidates: list[tuple[list[int], dict[str, str | None]]] = []
for index in range(min(raw.height, 30)):
labels = _merge_header_rows([_row_values(raw, index)])
mapping, ambiguous = _exact_mapping_from_columns(labels)
if not ambiguous:
candidates.append(([index + 1], mapping))
if len(candidates) != 1:
return None
return candidates[0]
def _calendar_date(year: int, month: int, day: int) -> date | None:
"""Return one mechanically valid Gregorian date, otherwise abstain."""
try:
return date(year, month, day)
except ValueError:
return None
def _expanded_two_digit_year(value: str) -> int:
"""Mirror the documented POSIX/Python two-digit-year boundary exactly."""
year = int(value)
if len(value) == 4:
return year
return 2000 + year if year <= 68 else 1900 + year
def _date_parse_result(
value: Any,
*,
date_convention: str | None = None,
date_locale: str | None = None,
allow_excel_serial: bool = False,
) -> tuple[str, date | None]:
"""Classify and parse a source date without list-order format guessing.
Date syntax is mechanical, so a fixed parser is more auditable than model
judgment here. Day/month strings are evaluated under both supported
interpretations; a genuinely ambiguous value needs an explicit reviewed
convention rather than whichever format happens to appear first.
"""
if date_convention not in {None, *DATE_CONVENTIONS}:
raise ValueError("date_convention must be day_first, month_first, or null")
if date_locale not in {None, *DATE_LOCALES}:
raise ValueError("date_locale must be it, fr or null")
if isinstance(value, datetime):
return "parsed", value.date()
if isinstance(value, date):
return "parsed", value
if isinstance(value, (int, float, Decimal)) and not isinstance(value, bool):
try:
serial = Decimal(str(value))
except ArithmeticError:
return "invalid", None
if not serial.is_finite() or serial != serial.to_integral_value():
return "invalid", None
compact = str(int(serial))
compact_match = COMPACT_ISO_DATE_RE.fullmatch(compact)
if compact_match is not None:
parsed = _calendar_date(
int(compact_match["year"]),
int(compact_match["month"]),
int(compact_match["day"]),
)
return ("parsed", parsed) if parsed is not None else ("invalid", None)
if allow_excel_serial and Decimal("20000") <= serial <= Decimal("60000"):
return "parsed", date(1899, 12, 30) + timedelta(days=int(serial))
return "invalid", None
token = _clean_text(value)
if not token:
return "blank", None
compact_match = COMPACT_ISO_DATE_RE.fullmatch(token)
if compact_match is not None:
parsed = _calendar_date(
int(compact_match["year"]),
int(compact_match["month"]),
int(compact_match["day"]),
)
return ("parsed", parsed) if parsed is not None else ("invalid", None)
if (
allow_excel_serial
and re.fullmatch(r"\d{5}", token) is not None
and 20000 <= int(token) <= 60000
):
return "parsed", date(1899, 12, 30) + timedelta(days=int(token))
year_first_match = YEAR_FIRST_DATE_RE.fullmatch(token)
if year_first_match is not None:
parsed = _calendar_date(
int(year_first_match["year"]),
int(year_first_match["month"]),
int(year_first_match["day"]),
)
return ("parsed", parsed) if parsed is not None else ("invalid", None)
textual_month_match = TEXTUAL_MONTH_DATE_RE.fullmatch(token)
if textual_month_match is not None:
if date_locale is None:
return "locale_required", None
if date_locale == "it" and re.search(r"\.\s+\d{4}$", token):
return "invalid", None
month_number = {"it": ITALIAN_MONTH_NUMBERS, "fr": FRENCH_MONTH_NUMBERS}[
date_locale
].get(_norm_label(textual_month_match["month"]))
if month_number is None:
return "invalid", None
parsed = _calendar_date(
int(textual_month_match["year"]),
month_number,
int(textual_month_match["day"]),
)
return ("parsed", parsed) if parsed is not None else ("invalid", None)
day_month_match = DAY_MONTH_DATE_RE.fullmatch(token)
if day_month_match is None:
return "invalid", None
year = _expanded_two_digit_year(day_month_match["year"])
first = int(day_month_match["first"])
second = int(day_month_match["second"])
interpretations = {
"day_first": _calendar_date(year, second, first),
"month_first": _calendar_date(year, first, second),
}
if date_convention is not None:
parsed = interpretations[date_convention]
return ("parsed", parsed) if parsed is not None else ("invalid", None)
valid = {parsed for parsed in interpretations.values() if parsed is not None}
if not valid:
return "invalid", None
if len(valid) == 1:
return "parsed", next(iter(valid))
return "ambiguous", None
def _parse_date(
value: Any,
*,
date_convention: str | None = None,
date_locale: str | None = None,
allow_excel_serial: bool = False,
) -> date | None:
status, parsed = _date_parse_result(
value,
date_convention=date_convention,
date_locale=date_locale,
allow_excel_serial=allow_excel_serial,
)
return parsed if status == "parsed" else None
def _parse_number(
value: Any,
*,
decimal_separator: str | None = None,
thousands_separator: str | None = None,
) -> Decimal | None:
"""Parse source money exactly, rejecting ambiguous textual separators.
Spreadsheet engines expose numeric cells as Python floats. ``allow_float`` is
therefore explicit here and converts only the engine-provided display value
through ``str``; text inputs remain subject to the stricter ambiguity rules.
"""
if value is None:
return None
if isinstance(value, str) and not _clean_text(value):
return None
try:
return parse_localized_decimal(
value,
label="source amount",
decimal_separator=decimal_separator,
thousands_separator=thousands_separator,
allow_float=True,
)
except MoneyValidationError:
return None
def _requires_numeric_separator_review(
value: Any,
*,
decimal_separator: str | None,
thousands_separator: str | None,
) -> bool:
"""Return whether two explicit separator conventions parse to different values."""
if (
decimal_separator is not None
or thousands_separator is not None
or not isinstance(value, str)
or not _clean_text(value)
or _parse_number(value) is not None
):
return False
candidates: set[Decimal] = set()
for candidate_decimal_separator in (".", ","):
try:
candidates.add(
parse_localized_decimal(
value,
label="source amount",
decimal_separator=candidate_decimal_separator,
thousands_separator=None,
allow_float=False,
)
)
except MoneyValidationError:
continue
return len(candidates) > 1
def _date_ratio(series: pl.Series) -> float:
values = [
_date_parse_result(value)[0]
for value in series.drop_nulls().head(100).to_list()
]
if not values:
return 0.0
# Ambiguous numeric and localized textual-month strings remain useful
# proposal evidence but never receive qualification authority until a
# reviewer chooses the corresponding source-bound convention.
return sum(
value in {"parsed", "ambiguous", "locale_required"} for value in values
) / len(values)
def _amount_ratio(series: pl.Series) -> float:
values = series.drop_nulls().head(100).to_list()
if not values:
return 0.0
return sum(_parse_number(value) is not None for value in values) / len(values)
def _first_matching_column(df: pl.DataFrame, tokens: Sequence[str]) -> str | None:
for col in df.columns:
label = _norm_label(col)
if any(token in label for token in tokens):
return col
return None
def infer_mapping(df: pl.DataFrame, side: str) -> dict[str, str | None]:
"""Propose a mapping for review without granting qualification authority."""
mapping: dict[str, str | None] = {
"date": _first_matching_column(df, ("data", "date", "datum", "booking")),
"amount": _first_matching_column(
df, ("amount", "importo", "montant", "betrag", "saldo", "total")
),
"debit": _first_matching_column(df, ("dare", "debit", "addebito", "soll")),
"credit": _first_matching_column(
df, ("avere", "credit", "credito", "haben", "accredito")
),
"description": _first_matching_column(
df,
(
"descrizione",
"description",
"causale",
"libelle",
"libellé",
"beschreibung",
"narrative",
"details",
),
),
"beneficiary": _first_matching_column(
df,
(
"beneficiario",
"beneficiary",
"payee",
"payer",
"cliente",
"fornitore",
"counterparty",
"contrepartie",
"beguenstigter",
"begünstigter",
),
),
"reference": _first_matching_column(
df, ("reference", "riferimento", "document", "doc", "cro", "trn", "iban")
),
"movement_number": _first_matching_column(
df,
(
"movement",
"movimento",
"nr. reg",
"n. reg",
"registrazione",
"journal",
"beleg",
),
),
"account": _first_matching_column(df, ("conto", "account", "iban", "konto")),
"currency": _first_matching_column(df, ("currency", "valuta", "devise")),
"unit": _first_matching_column(df, ("unit", "unita", "unità", "einheit")),
"entity_ref": _first_matching_column(
df, ("entity", "societa", "società", "company")
),
"party_ref": _first_matching_column(
df, ("party", "controparte", "counterparty id")
),
"direction": _first_matching_column(
df, ("direction", "segno", "verso", "richtung")
),
}
if mapping["date"] is None:
candidates = [
(col, _date_ratio(df.get_column(col)))
for col in df.columns
if col not in {SOURCE_ROW_COLUMN, SOURCE_SHEET_COLUMN}
]
if candidates and max(score for _, score in candidates) >= 0.5:
mapping["date"] = max(candidates, key=lambda item: item[1])[0]
return mapping
def _mapping_for_file(
recipe: dict[str, Any],
side: str,
path: Path,
*,
source_identity: str | None = None,
) -> dict[str, Any]:
if side not in recipe:
return {}
raw_side_recipe = recipe.get(side)
if not isinstance(raw_side_recipe, dict):
return {
_RECIPE_CONTAINER_ERROR_FIELD: (
"recipe container must use object-valued side, files, and "
"file entries"
)
}
side_recipe = raw_side_recipe
if "files" not in side_recipe:
return side_recipe
files = side_recipe.get("files")
inherited = {name: value for name, value in side_recipe.items() if name != "files"}
if not isinstance(files, dict) or any(
not isinstance(key, str) or not isinstance(value, dict)
for key, value in files.items()
):
return {
**inherited,
_RECIPE_CONTAINER_ERROR_FIELD: (
"recipe container must use object-valued side, files, and "
"file entries"
),
}
if isinstance(files, dict):
keys = [
source_identity,
path.name,
path.as_posix(),
]
for key in keys:
file_recipe = files.get(key) if key else None
if isinstance(file_recipe, dict):
merged = dict(inherited)
merged.update(file_recipe)
return merged
return inherited
def _normalized_mapping(mapping: dict[str, Any]) -> dict[str, str | None]:
return {
field: (_clean_text(mapping.get(field)) or None) for field in MAPPING_FIELDS
}
def _normalized_direction_value_mapping(
raw: object,
) -> tuple[dict[str, str], list[str]]:
"""Normalize a reviewed source vocabulary to canonical signed directions."""
if raw is None:
return {}, []
if not isinstance(raw, dict):
return {}, ["direction_value_mapping must be an object"]
normalized: dict[str, str] = {}
errors: list[str] = []
for source_value, canonical_value in raw.items():
source_label = _norm_label(source_value)
target = _norm_label(canonical_value)
if not source_label:
errors.append("direction_value_mapping contains an empty source label")
continue
if target not in CANONICAL_DIRECTIONS:
errors.append(
"direction_value_mapping targets must be positive, negative, or zero"
)
continue
if source_label in normalized:
errors.append(
"direction_value_mapping contains duplicate normalized source labels"
)
continue
normalized[source_label] = target
return dict(sorted(normalized.items())), sorted(set(errors))
def _normalized_date_convention(raw: object) -> tuple[str | None, list[str]]:
if raw is None:
return None, []
if not isinstance(raw, str) or raw not in DATE_CONVENTIONS:
return None, ["date_convention must be exactly day_first or month_first"]
return raw, []
def _normalized_date_locale(raw: object) -> tuple[str | None, list[str]]:
if raw is None:
return None, []
if not isinstance(raw, str) or raw not in DATE_LOCALES:
return None, ["date_locale must be exactly it or fr"]
return raw, []
def _normalized_non_movement_summary_labels(
raw: object,
) -> tuple[list[str], list[str]]:
if raw is None:
return [], []
if not isinstance(raw, list) or any(not isinstance(value, str) for value in raw):
return [], ["non_movement_summary_labels must be a list of strings"]
normalized = [_norm_label(value) for value in raw]
if (
any(not value for value in normalized)
or len(normalized) != len(set(normalized))
or normalized != sorted(normalized)
):
return [], [
"non_movement_summary_labels must contain unique non-empty "
"normalized labels in sorted order"
]
return normalized, []
def _tabular_adapter_binding(
date_locale: str | None,
non_movement_summary_labels: Sequence[str] = (),
*,
source_file: str | None = None,
) -> tuple[str, str]:
"""Select the source adapter bound by the reviewed mapping receipt."""
if source_file and Path(source_file).suffix.lower() == ".pdf":
return PDF_TABLE_ADAPTER_ID, PDF_TABLE_ADAPTER_VERSION
if date_locale == "fr":
return FRENCH_TABULAR_ADAPTER_ID, FRENCH_TABULAR_ADAPTER_VERSION
if date_locale is not None or non_movement_summary_labels:
return EXTENDED_TABULAR_ADAPTER_ID, EXTENDED_TABULAR_ADAPTER_VERSION
return TABULAR_ADAPTER_ID, TABULAR_ADAPTER_VERSION
def _mapping_review_content(
*,
side: str,
source_file: str,
header_rows: Sequence[int],
mapping: dict[str, Any],
direction_value_mapping: object | None,
potential_monetary_columns: Sequence[str],
excluded_monetary_columns: Sequence[str],
date_convention: object | None,
date_locale: object | None,
non_movement_summary_labels: object | None,
csv_field_delimiter: object | None,
decimal_separator: object | None,
thousands_separator: object | None,
) -> dict[str, Any]:
normalized_direction_values, direction_errors = _normalized_direction_value_mapping(
direction_value_mapping
)
if direction_errors:
raise ValueError("; ".join(direction_errors))
normalized_date_convention, date_errors = _normalized_date_convention(
date_convention
)
if date_errors:
raise ValueError("; ".join(date_errors))
normalized_date_locale, date_locale_errors = _normalized_date_locale(date_locale)
if date_locale_errors:
raise ValueError("; ".join(date_locale_errors))
normalized_summary_labels, summary_label_errors = (
_normalized_non_movement_summary_labels(non_movement_summary_labels)
)
if summary_label_errors:
raise ValueError("; ".join(summary_label_errors))
content = {
"side": side,
"source_file": source_file,
"header_rows": [int(value) for value in header_rows],
"mapping": _normalized_mapping(mapping),
"direction_value_mapping": normalized_direction_values,
"potential_monetary_columns": list(potential_monetary_columns),
"excluded_monetary_columns": list(excluded_monetary_columns),
"date_convention": normalized_date_convention,
"csv_field_delimiter": _mapping_csv_field_delimiter(
source_file,
csv_field_delimiter,
),
"decimal_separator": (
_clean_text(decimal_separator) or None
if decimal_separator is not None
else None
),
"thousands_separator": (
_clean_text(thousands_separator) or None
if thousands_separator is not None
else None
),
}
if normalized_date_locale is not None:
content["date_locale"] = normalized_date_locale
if normalized_summary_labels:
content["non_movement_summary_labels"] = normalized_summary_labels
return content
def build_mapping_review_receipt(
*,
decision_id: str,
reviewer_ref: str,
reviewed_on: str,
source_artifact_ref: str,
side: str,
source_file: str,
header_rows: Sequence[int],
mapping: dict[str, Any],
potential_monetary_columns: Sequence[str],
excluded_monetary_columns: Sequence[str],
direction_value_mapping: object | None = None,
date_convention: object | None = None,
date_locale: object | None = None,
non_movement_summary_labels: object | None = None,
csv_field_delimiter: object | None = None,
decimal_separator: object | None = None,
thousands_separator: object | None = None,
) -> dict[str, Any]:
"""Seal a professional mapping decision against source bytes and adapter."""
if isinstance(potential_monetary_columns, (str, bytes)) or isinstance(
excluded_monetary_columns,
(str, bytes),
):
raise ValueError(
"potential and excluded monetary columns must be explicit lists"
)
potential = [str(value).strip() for value in potential_monetary_columns]
excluded = [str(value).strip() for value in excluded_monetary_columns]
if (
any(not value for value in potential)
or len(potential) != len(set(potential))
or any(not value for value in excluded)
or len(excluded) != len(set(excluded))
):
raise ValueError(
"potential and excluded monetary columns must be unique non-empty names"
)
mapped_monetary = {
str(mapping[field])
for field in ("amount", "debit", "credit")
if mapping.get(field)
}
if set(excluded) - set(potential):
raise ValueError(
"excluded_monetary_columns must be a subset of "
"potential_monetary_columns"
)
if mapped_monetary & set(excluded):
raise ValueError("a mapped monetary column cannot also be explicitly excluded")
unresolved = set(potential) - mapped_monetary - set(excluded)
if unresolved or mapped_monetary - set(potential):
raise ValueError(
"potential monetary columns require a complete mapped-or-excluded "
"disposition"
)
normalized_date_locale, date_locale_errors = _normalized_date_locale(date_locale)
if date_locale_errors:
raise ValueError("; ".join(date_locale_errors))
normalized_summary_labels, summary_label_errors = (
_normalized_non_movement_summary_labels(non_movement_summary_labels)
)
if summary_label_errors:
raise ValueError("; ".join(summary_label_errors))
adapter_id, adapter_version = _tabular_adapter_binding(
normalized_date_locale,
normalized_summary_labels,
source_file=source_file,
)
return build_reviewed_decision_receipt(
decision_id=decision_id,
decision_type="journal_bank_mapping",
status="reviewed",
reviewer_ref=reviewer_ref,
reviewed_on=reviewed_on,
adapter_id=adapter_id,
adapter_version=adapter_version,
source_artifact_refs=[source_artifact_ref],
content=_mapping_review_content(
side=side,
source_file=source_file,
header_rows=header_rows,
mapping=mapping,
direction_value_mapping=direction_value_mapping,
potential_monetary_columns=potential,
excluded_monetary_columns=excluded,
date_convention=date_convention,
date_locale=normalized_date_locale,
non_movement_summary_labels=normalized_summary_labels,
csv_field_delimiter=csv_field_delimiter,
decimal_separator=decimal_separator,
thousands_separator=thousands_separator,
),
)
def _normalize_relationship_policy(policy: object) -> dict[str, Any]:
if not isinstance(policy, dict) or set(policy) != RELATIONSHIP_POLICY_FIELDS:
raise ValueError(
"relationship policy must contain the exact reviewed perimeter fields"
)
shape = _clean_text(policy["relationship_shape"])
if shape not in {"one_to_one", "one_to_many", "many_to_one", "many_to_many"}:
raise ValueError("unsupported relationship shape")
normalized: dict[str, Any] = {"relationship_shape": shape}
for field in (
"allow_evidence_reuse",
"require_same_currency",
"require_same_unit",
"require_same_entity",
"require_same_party",
):
if not isinstance(policy[field], bool):
raise ValueError(f"relationship policy {field} must be boolean")
normalized[field] = policy[field]
if normalized["allow_evidence_reuse"]:
raise ValueError("journal-bank reconciliation cannot reuse evidence")
if not normalized["require_same_currency"] or not normalized["require_same_unit"]:
raise ValueError("currency and unit equality are mandatory")
direction_policy = _clean_text(policy["direction_policy"])
if direction_policy not in {"absolute_amount", "same_sign", "opposite_sign"}:
raise ValueError("unsupported direction policy")
normalized["direction_policy"] = direction_policy
for field in (
"default_currency",
"default_unit",
"default_entity_ref",
"default_party_ref",
):
value = _clean_text(policy[field]) if policy[field] is not None else ""
normalized[field] = value or None
tolerance, tolerance_text = _canonical_tolerance(policy["amount_tolerance"])
if tolerance < ZERO: # pragma: no cover - guarded by _canonical_tolerance
raise ValueError("amount tolerance must not be negative")
normalized["amount_tolerance"] = tolerance_text
date_window = policy["date_window_days"]
if (
not isinstance(date_window, int)
or isinstance(date_window, bool)
or date_window < 0
):
raise ValueError("date_window_days must be a non-negative integer")
normalized["date_window_days"] = date_window
return normalized
def build_relationship_review_receipt(
*,
decision_id: str,
reviewer_ref: str,
reviewed_on: str,
source_artifact_refs: Sequence[str],
policy: dict[str, Any],
) -> dict[str, Any]:
"""Seal the reviewed relationship perimeter, shape, and tolerance policy."""
normalized = _normalize_relationship_policy(policy)
return build_reviewed_decision_receipt(
decision_id=decision_id,
decision_type="journal_bank_relationship",
status="reviewed",
reviewer_ref=reviewer_ref,
reviewed_on=reviewed_on,
adapter_id=RELATIONSHIP_ADAPTER_ID,
adapter_version=RELATIONSHIP_ADAPTER_VERSION,
source_artifact_refs=list(source_artifact_refs),
content={"policy": normalized},
)
def _mapped(row: dict[str, Any], mapping: dict[str, Any], key: str) -> Any:
col = mapping.get(key)
return row.get(str(col)) if col else None
def _amount_from_row(
row: dict[str, Any],
mapping: dict[str, Any],
*,
decimal_separator: str | None = None,
thousands_separator: str | None = None,
) -> Decimal | None:
amount = _parse_number(
_mapped(row, mapping, "amount"),
decimal_separator=decimal_separator,
thousands_separator=thousands_separator,
)
if amount is not None:
return amount
debit = _parse_number(
_mapped(row, mapping, "debit"),
decimal_separator=decimal_separator,
thousands_separator=thousands_separator,
)
credit = _parse_number(
_mapped(row, mapping, "credit"),
decimal_separator=decimal_separator,
thousands_separator=thousands_separator,
)
if debit is None and credit is None:
return None
return (debit or ZERO) - (credit or ZERO)
def _canonical_direction(
explicit_value: object,
amount: Decimal,
direction_value_mapping: dict[str, str],
) -> tuple[str | None, str | None]:
"""Resolve reviewed labels and prove agreement with the signed amount."""
amount_direction = (
"positive" if amount > ZERO else "negative" if amount < ZERO else "zero"
)
source_label = _norm_label(explicit_value)
if not source_label:
return amount_direction, None
direction = (
source_label
if source_label in CANONICAL_DIRECTIONS
else direction_value_mapping.get(source_label)
)
if direction is None:
return None, "unreviewed_direction_value"
if direction != amount_direction:
return None, "direction_amount_mismatch"
return direction, None
def _generic_period_component_scan(token: str) -> tuple[str, bool, bool, bool]:
"""Strip leading generic period components from one normalized token."""
remainder = token
saw_period = False
saw_year = False
consumed = False
period_words = sorted(GENERIC_PERIOD_WORDS, key=len, reverse=True)
while remainder:
code_match = GENERIC_PERIOD_CODE_RE.match(remainder)
if code_match is not None:
remainder = remainder[code_match.end() :]
saw_period = True
consumed = True
continue
word = next(
(value for value in period_words if remainder.startswith(value)),
None,
)
if word is not None:
remainder = remainder[len(word) :]
saw_period = True
consumed = True
continue
year_match = re.match(r"(?:19|20)\d{2}", remainder)
if year_match is not None:
remainder = remainder[year_match.end() :]
saw_year = True
consumed = True
continue
short_year_match = re.match(r"\d{2}", remainder)
if short_year_match is not None and (saw_period or len(remainder) == 2):
remainder = remainder[short_year_match.end() :]
saw_year = True
consumed = True
continue
version_match = GENERIC_PERIOD_VERSION_RE.match(remainder)
if version_match is not None:
remainder = remainder[version_match.end() :]
consumed = True
continue
break
return remainder, saw_period, saw_year, consumed
def _is_generic_period_fragment(token: str) -> bool:
remainder, saw_period, saw_year, consumed = _generic_period_component_scan(token)
return consumed and remainder == "" and saw_period and saw_year
def _reference_tokens(*values: Any) -> set[str]:
"""Return stable explicit identifiers, excluding generic semantic words."""
tokens: set[str] = set()
for value in values:
text = _norm_label(value)
parts = re.findall(r"[a-z0-9]+", text)
if not parts:
continue
compact = "".join(parts)
compact_remainder, compact_has_period, compact_has_year, _ = (
_generic_period_component_scan(compact)
)
part_scans = [_generic_period_component_scan(part) for part in parts]
compact_period_match = GENERIC_PERIOD_REFERENCE_RE.match(compact)
contains_year = any(
re.fullmatch(r"(?:(?:19|20)\d{2}|\d{2})", part) is not None
for part in parts
)
contains_period_word = any(
part in GENERIC_PERIOD_WORDS or part in REFERENCE_GENERIC_TOKENS or scan[1]
for part, scan in zip(parts, part_scans, strict=True)
)
period_signal = (
compact_period_match is not None
or (compact_has_period and compact_has_year)
or (
any(scan[1] for scan in part_scans)
and (any(scan[2] for scan in part_scans) or contains_year)
)
or (contains_year and contains_period_word)
)
if period_signal:
residual_parts: list[str] = []
for part, scan in zip(parts, part_scans, strict=True):
remainder, saw_period, saw_year, consumed = scan
if (
part in GENERIC_PERIOD_WORDS
or part in REFERENCE_GENERIC_TOKENS
or re.fullmatch(r"(?:(?:19|20)\d{2}|\d{2})", part) is not None
or GENERIC_PERIOD_VERSION_RE.fullmatch(part) is not None
):
continue
if consumed and (saw_period or saw_year):
if remainder:
residual_parts.append(remainder)
continue
part_period_match = GENERIC_PERIOD_REFERENCE_RE.match(part)
if part_period_match is not None:
suffix = part[part_period_match.end() :]
if suffix:
residual_parts.append(suffix)
continue
residual_parts.append(part)
residual_candidates = [
*residual_parts,
"".join(residual_parts),
]
if compact_period_match is not None:
suffix = compact[compact_period_match.end() :]
if suffix:
residual_candidates.append(suffix)
if compact_has_period and compact_has_year and compact_remainder:
residual_candidates.append(compact_remainder)
has_distinctive_suffix = any(
len(candidate) >= 5
and any(character.isdigit() for character in candidate)
and GENERIC_PERIOD_VERSION_RE.fullmatch(candidate) is None
for candidate in residual_candidates
)
if not has_distinctive_suffix:
continue
generic_period = GENERIC_PERIOD_REFERENCE_RE.fullmatch(compact) is not None
generic_year_range = (
len(parts) == 2
and re.fullmatch(r"(?:19|20)\d{2}", parts[0]) is not None
and re.fullmatch(r"(?:\d{2}|(?:19|20)\d{2})", parts[1]) is not None
)
if generic_period or generic_year_range:
continue
candidates = list(parts)
if parts[0] not in REFERENCE_GENERIC_TOKENS:
candidates.append(compact)
for token in candidates:
if token in REFERENCE_GENERIC_TOKENS or len(token) < 5:
continue
if not any(character.isdigit() for character in token):
continue
if GENERIC_PERIOD_REFERENCE_RE.fullmatch(token) is not None:
continue
if _is_generic_period_fragment(token):
continue
if token.isdigit():
if len(token) < 5:
continue
tokens.add(token)
continue
generic_prefix = next(
(
prefix
for prefix in REFERENCE_GENERIC_TOKENS
if token.startswith(prefix) and token != prefix
),
None,
)
if generic_prefix is not None:
continue
tokens.add(token)
return tokens
def _transaction_frame(records: list[dict[str, Any]]) -> pl.DataFrame:
if not records:
return pl.DataFrame(schema={col: pl.Utf8 for col in TRANSACTION_COLUMNS})
frame = pl.DataFrame(records, infer_schema_length=None)
for col in TRANSACTION_COLUMNS:
if col not in frame.columns:
frame = frame.with_columns(pl.lit(None).alias(col))
return frame.select(TRANSACTION_COLUMNS)
def _non_movement_frame(records: list[dict[str, Any]]) -> pl.DataFrame:
if not records:
return pl.DataFrame(schema={col: pl.Utf8 for col in NON_MOVEMENT_COLUMNS})
frame = pl.DataFrame(records, infer_schema_length=None)
for col in NON_MOVEMENT_COLUMNS:
if col not in frame.columns:
frame = frame.with_columns(pl.lit(None).alias(col))
return frame.select(NON_MOVEMENT_COLUMNS)
def _non_movement_records(
diagnostics: Sequence[dict[str, Any]],
) -> list[dict[str, Any]]:
records: list[dict[str, Any]] = []
for diag in diagnostics:
rows = diag.get("non_movement_rows")
if isinstance(rows, list):
records.extend(row for row in rows if isinstance(row, dict))
return records
def _mapping_contract_errors(
mapping: dict[str, Any], table_columns: Sequence[str]
) -> list[str]:
errors = _missing_mapping(mapping)
selected_columns = [
value for value in _normalized_mapping(mapping).values() if value is not None
]
missing_columns = sorted(set(selected_columns) - set(table_columns))
errors.extend(f"missing column: {column}" for column in missing_columns)
if mapping.get("amount") and (mapping.get("debit") or mapping.get("credit")):
errors.append("amount cannot be combined with debit/credit")
if mapping.get("debit") and mapping.get("credit"):
if mapping.get("debit") == mapping.get("credit"):
errors.append("debit and credit must be different columns")
return sorted(set(errors))
def _potential_monetary_columns(
table: pl.DataFrame,
mapping: dict[str, Any],
*,
decimal_separator: str | None,
thousands_separator: str | None,
) -> list[str]:
"""Return populated columns whose monetary role needs explicit disposition."""
mapped_non_monetary = {
column
for field, column in mapping.items()
if field not in {"amount", "debit", "credit"} and column
}
candidates: list[str] = []
for column in table.columns:
if column in {
SOURCE_ROW_COLUMN,
SOURCE_SHEET_COLUMN,
}:
continue
explicit_header = any(
token in _norm_label(column) for token in MONETARY_HEADER_TOKENS
)
if column in mapped_non_monetary and not explicit_header:
continue
numeric_values = any(
_parse_number(
value,
decimal_separator=decimal_separator,
thousands_separator=thousands_separator,
)
is not None
for value in table.get_column(column).drop_nulls().to_list()
if _clean_text(value)
)
if explicit_header or numeric_values:
candidates.append(column)
return candidates
def _excluded_monetary_columns(
file_recipe: dict[str, Any],
) -> tuple[list[str], bool]:
raw = file_recipe.get("excluded_monetary_columns", [])
if not isinstance(raw, list) or any(
not isinstance(value, str) or not value.strip() for value in raw
):
return [], False
normalized = [value.strip() for value in raw]
return normalized, len(normalized) == len(set(normalized))
def _declared_potential_monetary_columns(
file_recipe: dict[str, Any],
) -> tuple[list[str], bool]:
raw = file_recipe.get("potential_monetary_columns", [])
if not isinstance(raw, list) or any(
not isinstance(value, str) or not value.strip() for value in raw
):
return [], False
normalized = [value.strip() for value in raw]
return normalized, len(normalized) == len(set(normalized))
def _tabular_candidate_count(
table: pl.DataFrame,
mapping: dict[str, Any],
*,
non_movement_summary_labels: Sequence[str] = (),
) -> int:
money_columns = [
str(mapping[field])
for field in ("amount", "debit", "credit")
if mapping.get(field)
]
if not money_columns:
return 0
return sum(
(
any(_clean_text(row.get(column)) for column in money_columns)
and not _is_reviewed_non_movement_summary(
row,
mapping,
non_movement_summary_labels,
)
)
for row in table.iter_rows(named=True)
)
def _is_reviewed_non_movement_summary(
row: dict[str, Any],
mapping: dict[str, Any],
labels: Sequence[str],
) -> bool:
"""Apply an exact reviewed summary label only to structurally blank rows."""
if not labels or _clean_text(_mapped(row, mapping, "date")):
return False
stable_reference_tokens = _reference_tokens(
_clean_text(_mapped(row, mapping, "reference")),
_clean_text(_mapped(row, mapping, "movement_number")),
)
if stable_reference_tokens:
return False
description = _norm_label(_mapped(row, mapping, "description"))
return bool(description and description in labels)
def _date_contract_evidence(
table: pl.DataFrame,
mapping: dict[str, Any],
*,
date_convention: str | None,
date_locale: str | None,
non_movement_summary_labels: Sequence[str],
allow_excel_serial: bool,
) -> dict[str, Any]:
"""Describe date syntax only for populated monetary candidate rows."""
counts: Counter[str] = Counter()
rows: dict[str, list[int]] = {
"parsed": [],
"blank": [],
"ambiguous": [],
"invalid": [],
}
for row in table.iter_rows(named=True):
source_amount_values = (
_mapped(row, mapping, "amount"),
_mapped(row, mapping, "debit"),
_mapped(row, mapping, "credit"),
)
if not any(_clean_text(value) for value in source_amount_values):
continue
if _is_reviewed_non_movement_summary(
row,
mapping,
non_movement_summary_labels,
):
continue
status, _ = _date_parse_result(
_mapped(row, mapping, "date"),
date_convention=date_convention,
date_locale=date_locale,
allow_excel_serial=allow_excel_serial,
)
counts[status] += 1
rows.setdefault(status, []).append(int(row[SOURCE_ROW_COLUMN]))
evidence = {
"date_convention": date_convention,
"excel_serial_allowed": allow_excel_serial,
"status_counts": {
status: int(counts[status])
for status in ("parsed", "blank", "ambiguous", "invalid")
},
"source_rows": rows,
}
if counts["locale_required"]:
evidence["status_counts"]["locale_required"] = int(counts["locale_required"])
if date_locale is not None:
evidence["date_locale"] = date_locale
return evidence
def _validated_mapping_decision(
file_recipe: dict[str, Any],
*,
expected_content: dict[str, Any],
source_artifact_ref: str,
expected_adapter_id: str,
expected_adapter_version: str,
) -> tuple[dict[str, Any] | None, str | None]:
raw = file_recipe.get("mapping_decision")
if not isinstance(raw, dict):
return None, "A reviewed mapping receipt is required for this layout."
try:
receipt = validate_reviewed_decision_receipt(
raw,
expected_source_artifact_refs=[source_artifact_ref],
expected_adapter_id=expected_adapter_id,
expected_adapter_version=expected_adapter_version,
require_reviewed=True,
)
except ValueError as exc:
return None, f"Mapping receipt is invalid or stale: {exc}"
if receipt["decision_type"] != "journal_bank_mapping":
return None, "Mapping receipt has the wrong decision type."
if receipt["content"] != expected_content:
return None, "Mapping receipt content does not match the current recipe."
return receipt, None
def _normalize_table(
path: Path,
side: str,
recipe: dict[str, Any],
*,
source_identity: str,
source_artifact_ref: str,
) -> tuple[pl.DataFrame, dict[str, Any]]:
file_recipe = _mapping_for_file(
recipe,
side,
path,
source_identity=source_identity,
)
sheet_names = _source_sheet_names(path)
if len(sheet_names) != 1:
return _transaction_frame([]), {
"source_file": source_identity,
"source_sheet": None,
"parser": "tabular",
"adapter_id": TABULAR_ADAPTER_ID,
"source_family": "tabular.multisheet.unqualified.v1",
"qualification_status": "unsupported_source_layout",
"failure_kind": "multiple_sheets_unsupported",
"candidate_row_count": 0,
"row_count": 0,
"preview": [],
"missing_required_mapping": [],
"row_disposition_counts": {},
"row_dispositions": [],
"workbook_sheets": sheet_names,
"csv_field_delimiter": None,
"csv_field_delimiter_origin": "not_applicable",
"csv_field_delimiter_profile": None,
"limitations": [
"The bounded tabular adapter requires exactly one physical sheet; "
"no sheet was selected or silently omitted."
],
}
source_sheet = sheet_names[0]
delimiter_resolution = _resolve_csv_field_delimiter(path, file_recipe)
if delimiter_resolution["status"] not in {"resolved", "not_applicable"}:
ambiguous = delimiter_resolution["status"] == "ambiguous"
return _transaction_frame([]), {
"source_file": source_identity,
"source_sheet": source_sheet,
"parser": "tabular",
"adapter_id": TABULAR_ADAPTER_ID,
"source_family": "tabular.csv_delimiter.v1",
"qualification_status": (
"needs_review" if ambiguous else "unsupported_source_layout"
),
"failure_kind": (
"ambiguous_csv_field_delimiter"
if ambiguous
else "unsupported_csv_field_delimiter"
),
"csv_field_delimiter": None,
"requested_csv_field_delimiter": delimiter_resolution[
"requested_delimiter"
],
"csv_field_delimiter_origin": delimiter_resolution["origin"],
"csv_field_delimiter_profile": delimiter_resolution["profile"],
"candidate_row_count": 0,
"row_count": 0,
"preview": [],
"missing_required_mapping": (["csv_field_delimiter"] if ambiguous else []),
"row_disposition_counts": {},
"row_dispositions": [],
"limitations": [
(
"More than one supported CSV field delimiter produced a "
"consistent table; a reviewed delimiter is required."
if ambiguous
else "No supported one-byte CSV field delimiter was selected."
)
],
}
csv_field_delimiter = delimiter_resolution["delimiter"]
pdf_metadata: dict[str, Any] = {}
source_locators: list[tuple[str, int]] | None = None
if path.suffix.lower() == ".pdf":
raw, source_locators, pdf_metadata = _read_pdf_table_raw(path)
if pdf_metadata["failure_kind"] is not None:
frame, text_diagnostic = _normalize_text_pdf(
path,
side,
recipe,
source_identity=source_identity,
)
return frame, {
**text_diagnostic,
"parser": "pdf_table",
"adapter_id": TEXT_PDF_ADAPTER_ID,
"source_family": "pdf.table.unqualified.v1",
"qualification_status": "unsupported_source_layout",
"failure_kind": pdf_metadata["failure_kind"],
"missing_required_mapping": ["supported reviewed PDF table"],
**pdf_metadata,
"limitations": [
*pdf_metadata["limitations"],
*text_diagnostic.get("limitations", []),
],
}
pdf_metadata = {
key: value
for key, value in pdf_metadata.items()
if key not in {"failure_kind", "limitations"}
}
else:
raw = _read_table_raw(
path,
csv_field_delimiter=(
str(csv_field_delimiter) if csv_field_delimiter is not None else None
),
)
exact_contract = _exact_header_contract(raw)
raw_header_rows = file_recipe.get("header_rows")
header_rows_shape_valid = "header_rows" not in file_recipe or (
isinstance(raw_header_rows, list)
and bool(raw_header_rows)
and all(
isinstance(value, int) and not isinstance(value, bool) and value > 0
for value in raw_header_rows
)
and len(raw_header_rows) == len(set(raw_header_rows))
)
provided_header_rows = raw_header_rows if header_rows_shape_valid else None
header_rows = (
list(provided_header_rows)
if isinstance(provided_header_rows, list) and provided_header_rows
else (
exact_contract[0]
if exact_contract is not None
else _suggest_header_rows(raw)
)
)
table = _apply_header(
raw,
header_rows,
source_sheet=source_sheet,
source_locators=source_locators,
)
raw_mapping = file_recipe.get("mapping")
mapping_shape_valid = "mapping" not in file_recipe or (
isinstance(raw_mapping, dict)
and set(raw_mapping) <= set(MAPPING_FIELDS)
and all(
value is None or isinstance(value, str) for value in raw_mapping.values()
)
)
provided_mapping = raw_mapping if mapping_shape_valid else None
exact_mapping = exact_contract[1] if exact_contract is not None else None
exact_mapping_applies = (
exact_contract is not None
and header_rows == exact_contract[0]
and (
provided_mapping is None
or _normalized_mapping(provided_mapping)
== _normalized_mapping(exact_contract[1])
)
)
mapping = (
exact_mapping
if exact_mapping_applies and exact_mapping is not None
else (
provided_mapping
if provided_mapping is not None
else infer_mapping(table, side)
)
)
mapping = _normalized_mapping(mapping)
direction_value_mapping, direction_mapping_errors = (
_normalized_direction_value_mapping(file_recipe.get("direction_value_mapping"))
)
date_convention, date_convention_errors = _normalized_date_convention(
file_recipe.get("date_convention")
)
date_locale, date_locale_errors = _normalized_date_locale(
file_recipe.get("date_locale")
)
non_movement_summary_labels, summary_label_errors = (
_normalized_non_movement_summary_labels(
file_recipe.get("non_movement_summary_labels")
)
)
tabular_adapter_id, tabular_adapter_version = _tabular_adapter_binding(
date_locale,
non_movement_summary_labels,
source_file=source_identity,
)
observed_direction_values = sorted(
{
_norm_label(_mapped(row, mapping, "direction"))
for row in table.iter_rows(named=True)
if _norm_label(_mapped(row, mapping, "direction"))
}
)
noncanonical_direction_values = sorted(
value
for value in observed_direction_values
if value not in CANONICAL_DIRECTIONS
)
if not mapping.get("direction") and direction_value_mapping:
direction_mapping_errors.append(
"direction_value_mapping requires a mapped direction column"
)
elif mapping.get("direction") and set(direction_value_mapping) != set(
noncanonical_direction_values
):
direction_mapping_errors.append(
"direction_value_mapping must exactly cover every observed "
"non-canonical direction label"
)
direction_mapping_errors = sorted(set(direction_mapping_errors))
decimal_separator = file_recipe.get("decimal_separator")
thousands_separator = file_recipe.get("thousands_separator")
execution_decimal_separator = (
_clean_text(decimal_separator) or None
if decimal_separator is not None
else None
)
execution_thousands_separator = (
_clean_text(thousands_separator) or None
if thousands_separator is not None
else None
)
ambiguous_numeric_rows = sorted(
{
int(row[SOURCE_ROW_COLUMN])
for row in table.iter_rows(named=True)
if any(
_requires_numeric_separator_review(
_mapped(row, mapping, field),
decimal_separator=execution_decimal_separator,
thousands_separator=execution_thousands_separator,
)
for field in ("amount", "debit", "credit")
if mapping.get(field)
)
}
)
declared_potential_columns, potential_declaration_valid = (
_declared_potential_monetary_columns(file_recipe)
)
excluded_monetary_columns, exclusions_valid = _excluded_monetary_columns(
file_recipe
)
potential_monetary_columns = _potential_monetary_columns(
table,
mapping,
decimal_separator=execution_decimal_separator,
thousands_separator=execution_thousands_separator,
)
mapped_monetary_columns = {
str(mapping[field])
for field in ("amount", "debit", "credit")
if mapping.get(field)
}
excluded_set = set(excluded_monetary_columns)
exclusions_exist = excluded_set <= set(potential_monetary_columns)
mapped_monetary_columns_have_evidence = mapped_monetary_columns <= set(
potential_monetary_columns
)
mapped_and_excluded_columns = sorted(mapped_monetary_columns & excluded_set)
unresolved_monetary_columns = sorted(
set(potential_monetary_columns) - mapped_monetary_columns - excluded_set
)
date_evidence = _date_contract_evidence(
table,
mapping,
date_convention=date_convention,
date_locale=date_locale,
non_movement_summary_labels=non_movement_summary_labels,
allow_excel_serial=path.suffix.lower() in {".xls", ".xlsx", ".xlsm"},
)
ambiguous_date_rows = date_evidence["source_rows"]["ambiguous"]
locale_required_date_rows = date_evidence["source_rows"].get(
"locale_required",
[],
)
review_content = _mapping_review_content(
side=side,
source_file=source_identity,
header_rows=header_rows,
mapping=mapping,
direction_value_mapping=direction_value_mapping,
potential_monetary_columns=potential_monetary_columns,
excluded_monetary_columns=excluded_monetary_columns,
date_convention=date_convention,
date_locale=date_locale,
non_movement_summary_labels=non_movement_summary_labels,
csv_field_delimiter=csv_field_delimiter,
decimal_separator=decimal_separator,
thousands_separator=thousands_separator,
)
mapping_receipt: dict[str, Any] | None = None
mapping_review_error: str | None = None
monetary_declarations_match_automatic_evidence = (
"potential_monetary_columns" not in file_recipe
and "excluded_monetary_columns" not in file_recipe
) or (
"potential_monetary_columns" in file_recipe
and "excluded_monetary_columns" in file_recipe
and potential_declaration_valid
and exclusions_valid
and declared_potential_columns == potential_monetary_columns
)
automatic_contract_applies = (
exact_mapping_applies
and path.suffix.lower() != ".pdf"
and not direction_value_mapping
and date_convention is None
and date_locale is None
and not non_movement_summary_labels
and not ambiguous_date_rows
and not locale_required_date_rows
and not ambiguous_numeric_rows
and not excluded_monetary_columns
and not unresolved_monetary_columns
and not delimiter_resolution["review_required"]
and decimal_separator is None
and thousands_separator is None
and monetary_declarations_match_automatic_evidence
and file_recipe.get("mapping_decision") is None
)
if not automatic_contract_applies:
mapping_receipt, mapping_review_error = _validated_mapping_decision(
file_recipe,
expected_content=review_content,
source_artifact_ref=source_artifact_ref,
expected_adapter_id=tabular_adapter_id,
expected_adapter_version=tabular_adapter_version,
)
contract_errors = _mapping_contract_errors(mapping, table.columns)
recipe_container_error = file_recipe.get(_RECIPE_CONTAINER_ERROR_FIELD)
if recipe_container_error:
contract_errors.append(str(recipe_container_error))
if not header_rows_shape_valid:
contract_errors.append(
"header_rows must be a non-empty list of unique positive integers"
)
if not mapping_shape_valid:
contract_errors.append(
"mapping must be an object containing only supported fields with "
"string or null column names"
)
contract_errors.extend(direction_mapping_errors)
contract_errors.extend(date_convention_errors)
contract_errors.extend(date_locale_errors)
contract_errors.extend(summary_label_errors)
if ambiguous_date_rows and date_convention is None:
contract_errors.append(
"ambiguous day/month dates require a reviewed date_convention"
)
if locale_required_date_rows and date_locale is None:
contract_errors.append(
"localized textual-month dates require a reviewed date_locale"
)
if ambiguous_numeric_rows:
contract_errors.append(
"ambiguous numeric separators require reviewed decimal_separator "
"and thousands_separator values"
)
if not automatic_contract_applies:
if "potential_monetary_columns" not in file_recipe:
contract_errors.append(
"potential_monetary_columns must be explicitly reviewed"
)
elif not potential_declaration_valid:
contract_errors.append(
"potential_monetary_columns must contain unique non-empty "
"column names"
)
elif declared_potential_columns != potential_monetary_columns:
contract_errors.append(
"potential_monetary_columns does not match current source evidence"
)
if "excluded_monetary_columns" not in file_recipe:
contract_errors.append(
"excluded_monetary_columns must be explicitly reviewed, even "
"when empty"
)
if not exclusions_valid:
contract_errors.append(
"excluded_monetary_columns must contain unique non-empty column names"
)
if not exclusions_exist:
contract_errors.append(
"excluded_monetary_columns contains a column without monetary evidence"
)
if not mapped_monetary_columns_have_evidence:
contract_errors.append(
"a mapped monetary column lacks current monetary evidence"
)
contract_errors.extend(
f"mapped monetary column is also excluded: {column}"
for column in mapped_and_excluded_columns
)
contract_errors.extend(
f"unresolved monetary column: {column}"
for column in unresolved_monetary_columns
)
contract_errors = sorted(set(contract_errors))
if mapping_review_error is not None or contract_errors:
limitations = [
value
for value in (
mapping_review_error,
*contract_errors,
)
if value
]
return _transaction_frame([]), {
"source_file": source_identity,
"source_sheet": source_sheet,
"parser": "pdf_table" if path.suffix.lower() == ".pdf" else "tabular",
"adapter_id": tabular_adapter_id,
"source_family": (
"pdf.table.reviewed.v1"
if path.suffix.lower() == ".pdf"
else "tabular.explicit_columns.v1"
),
"qualification_status": "needs_review",
"failure_kind": "mapping_review_required",
"header_rows": header_rows,
"mapping": mapping,
"direction_value_mapping": direction_value_mapping,
"observed_direction_values": observed_direction_values,
"date_convention": date_convention,
**({"date_locale": date_locale} if date_locale is not None else {}),
**(
{"non_movement_summary_labels": non_movement_summary_labels}
if non_movement_summary_labels
else {}
),
"date_interpretation_evidence": date_evidence,
"ambiguous_numeric_rows": ambiguous_numeric_rows,
"mapping_origin": (
"reviewed_recipe" if provided_mapping is not None else "proposal"
),
"mapping_review_content": review_content,
"mapping_review_content_sha256": canonical_json_sha256(review_content),
"reviewed_mapping_ref": None,
"mapping_decision": None,
"csv_field_delimiter": csv_field_delimiter,
"csv_field_delimiter_origin": delimiter_resolution["origin"],
"csv_field_delimiter_profile": delimiter_resolution["profile"],
"raw_columns": [
column
for column in table.columns
if column not in {SOURCE_ROW_COLUMN, SOURCE_SHEET_COLUMN}
],
"potential_monetary_columns": potential_monetary_columns,
"excluded_monetary_columns": excluded_monetary_columns,
"unresolved_monetary_columns": unresolved_monetary_columns,
"candidate_row_count": _tabular_candidate_count(
table,
mapping,
non_movement_summary_labels=non_movement_summary_labels,
),
"row_count": 0,
"preview": [],
"missing_required_mapping": contract_errors,
"row_disposition_counts": {},
"row_dispositions": [],
"limitations": limitations,
**pdf_metadata,
}
records: list[dict[str, Any]] = []
row_dispositions: list[dict[str, Any]] = []
disposition_counts: Counter[str] = Counter()
invalid_candidate_rows: list[int] = []
monetary_candidate_count = 0
for row in table.iter_rows(named=True):
source_row = int(row[SOURCE_ROW_COLUMN])
row_source_sheet = str(row[SOURCE_SHEET_COLUMN])
amount = _amount_from_row(
row,
mapping,
decimal_separator=execution_decimal_separator,
thousands_separator=execution_thousands_separator,
)
date_status, parsed_date = _date_parse_result(
_mapped(row, mapping, "date"),
date_convention=date_convention,
date_locale=date_locale,
allow_excel_serial=path.suffix.lower() in {".xls", ".xlsx", ".xlsm"},
)
description = _clean_text(_mapped(row, mapping, "description"))
beneficiary = _clean_text(_mapped(row, mapping, "beneficiary"))
reference = _clean_text(_mapped(row, mapping, "reference"))
movement = _clean_text(_mapped(row, mapping, "movement_number"))
stable_reference_tokens = _reference_tokens(reference, movement)
account = _clean_text(_mapped(row, mapping, "account"))
currency = _clean_text(_mapped(row, mapping, "currency"))
unit = _clean_text(_mapped(row, mapping, "unit"))
entity_ref = _clean_text(_mapped(row, mapping, "entity_ref"))
party_ref = _clean_text(_mapped(row, mapping, "party_ref"))
explicit_direction = _clean_text(_mapped(row, mapping, "direction"))
source_amount_values = (
_mapped(row, mapping, "amount"),
_mapped(row, mapping, "debit"),
_mapped(row, mapping, "credit"),
)
has_monetary_value = any(_clean_text(value) for value in source_amount_values)
locator = {"source_sheet": row_source_sheet, "source_row": source_row}
if not has_monetary_value:
disposition_counts["excluded_non_monetary"] += 1
row_dispositions.append(
{
**locator,
"status": "excluded_non_monetary",
"reason": "No mapped monetary cell is populated.",
}
)
continue
if _is_reviewed_non_movement_summary(
row,
mapping,
non_movement_summary_labels,
):
disposition_counts["excluded_reviewed_summary"] += 1
row_dispositions.append(
{
**locator,
"status": "excluded_reviewed_summary",
"reason": (
"A blank-date row with no stable reference matched an "
"exact reviewed non-movement summary label."
),
}
)
continue
monetary_candidate_count += 1
if amount is None:
invalid_candidate_rows.append(source_row)
disposition_counts["invalid_monetary_value"] += 1
row_dispositions.append(
{
**locator,
"status": "invalid_monetary_value",
"reason": "A populated mapped monetary cell did not parse exactly.",
}
)
continue
direction, direction_error = _canonical_direction(
explicit_direction,
amount,
direction_value_mapping,
)
if direction_error is not None:
invalid_candidate_rows.append(source_row)
disposition_counts[direction_error] += 1
row_dispositions.append(
{
**locator,
"status": direction_error,
"reason": (
"The reviewed source direction does not agree with the "
"exact signed amount."
if direction_error == "direction_amount_mismatch"
else "A source direction label lacks a reviewed canonical mapping."
),
}
)
continue
if date_status in {"ambiguous", "invalid", "locale_required"}:
invalid_candidate_rows.append(source_row)
date_disposition = {
"ambiguous": "ambiguous_date_without_review",
"locale_required": "localized_date_without_review",
"invalid": "invalid_date_value",
}[date_status]
disposition_counts[date_disposition] += 1
row_dispositions.append(
{
**locator,
"status": date_disposition,
"reason": (
"A populated day/month date has more than one valid "
"interpretation and lacks a reviewed convention."
if date_status == "ambiguous"
else (
"A populated textual-month date lacks a reviewed locale."
if date_status == "locale_required"
else "A populated mapped date is not a valid date under "
"the current mechanical or reviewed convention."
)
),
}
)
continue
if date_status == "blank" and not stable_reference_tokens:
invalid_candidate_rows.append(source_row)
disposition_counts["missing_date_without_stable_reference"] += 1
row_dispositions.append(
{
**locator,
"status": "missing_date_without_stable_reference",
"reason": (
"A monetary row without an actual date requires a stable "
"non-generic explicit identifier."
),
}
)
continue
disposition = "emitted_reference_only" if date_status == "blank" else "emitted"
disposition_counts[disposition] += 1
row_dispositions.append(
{
**locator,
"status": disposition,
"reason": (
"Missing date; eligible only for explicit-reference matching."
if date_status == "blank"
else "Mapped monetary row emitted."
),
}
)
transaction_id = (
f"{side}:{_source_identity_fragment(source_identity)}:"
f"{_identifier_fragment(row_source_sheet)}:{source_row}"
)
records.append(
{
"side": side,
"transaction_id": transaction_id,
"transaction_date": parsed_date.isoformat() if parsed_date else None,
"amount_signed": decimal_text(amount) if amount is not None else None,
"amount_abs": (
decimal_text(abs(amount)) if amount is not None else None
),
"description": description or None,
"beneficiary": beneficiary or None,
"reference": reference or None,
"movement_number": movement or None,
"account": account or None,
"currency": currency or None,
"unit": unit or None,
"entity_ref": entity_ref or None,
"party_ref": party_ref or None,
"direction": direction,
"source_file": source_identity,
"source_sheet": row_source_sheet,
"source_row": source_row,
}
)
if invalid_candidate_rows:
return _transaction_frame([]), {
"source_file": source_identity,
"source_sheet": source_sheet,
"parser": "pdf_table" if path.suffix.lower() == ".pdf" else "tabular",
"adapter_id": tabular_adapter_id,
"source_family": (
"pdf.table.reviewed.v1"
if path.suffix.lower() == ".pdf"
else "tabular.explicit_columns.v1"
),
"qualification_status": "unsupported_source_layout",
"failure_kind": "candidate_row_contract_failed",
"header_rows": header_rows,
"mapping": mapping,
"direction_value_mapping": direction_value_mapping,
"observed_direction_values": observed_direction_values,
"date_convention": date_convention,
**({"date_locale": date_locale} if date_locale is not None else {}),
**(
{"non_movement_summary_labels": non_movement_summary_labels}
if non_movement_summary_labels
else {}
),
"date_interpretation_evidence": date_evidence,
"ambiguous_numeric_rows": ambiguous_numeric_rows,
"mapping_origin": (
"bounded_exact_headers"
if automatic_contract_applies
else "reviewed_recipe"
),
"mapping_review_content": review_content,
"mapping_review_content_sha256": canonical_json_sha256(review_content),
"reviewed_mapping_ref": (
str(mapping_receipt["decision_id"]) if mapping_receipt else None
),
"mapping_decision": mapping_receipt,
"csv_field_delimiter": csv_field_delimiter,
"csv_field_delimiter_origin": delimiter_resolution["origin"],
"csv_field_delimiter_profile": delimiter_resolution["profile"],
"raw_columns": [
column
for column in table.columns
if column not in {SOURCE_ROW_COLUMN, SOURCE_SHEET_COLUMN}
],
"potential_monetary_columns": potential_monetary_columns,
"excluded_monetary_columns": excluded_monetary_columns,
"unresolved_monetary_columns": unresolved_monetary_columns,
"candidate_row_count": monetary_candidate_count,
"row_count": 0,
"preview": [],
"missing_required_mapping": [],
"invalid_candidate_rows": invalid_candidate_rows,
"row_disposition_counts": dict(sorted(disposition_counts.items())),
"row_dispositions": row_dispositions,
"limitations": [
"At least one monetary candidate failed exact amount, date, or "
"reviewed-direction disposition; no rows were emitted."
],
**pdf_metadata,
}
frame = _transaction_frame(records)
diagnostics = {
"source_file": source_identity,
"source_sheet": source_sheet,
"parser": "pdf_table" if path.suffix.lower() == ".pdf" else "tabular",
"adapter_id": tabular_adapter_id,
"source_family": (
"pdf.table.reviewed.v1"
if path.suffix.lower() == ".pdf"
else "tabular.explicit_columns.v1"
),
"qualification_status": "qualified",
"failure_kind": None,
"header_rows": header_rows,
"mapping": mapping,
"direction_value_mapping": direction_value_mapping,
"observed_direction_values": observed_direction_values,
"date_convention": date_convention,
**({"date_locale": date_locale} if date_locale is not None else {}),
**(
{"non_movement_summary_labels": non_movement_summary_labels}
if non_movement_summary_labels
else {}
),
"date_interpretation_evidence": date_evidence,
"ambiguous_numeric_rows": ambiguous_numeric_rows,
"mapping_origin": (
"bounded_exact_headers" if automatic_contract_applies else "reviewed_recipe"
),
"mapping_review_content": review_content,
"mapping_review_content_sha256": canonical_json_sha256(review_content),
"reviewed_mapping_ref": (
str(mapping_receipt["decision_id"]) if mapping_receipt else None
),
"mapping_decision": mapping_receipt,
"csv_field_delimiter": csv_field_delimiter,
"csv_field_delimiter_origin": delimiter_resolution["origin"],
"csv_field_delimiter_profile": delimiter_resolution["profile"],
"raw_columns": [
column
for column in table.columns
if column not in {SOURCE_ROW_COLUMN, SOURCE_SHEET_COLUMN}
],
"potential_monetary_columns": potential_monetary_columns,
"excluded_monetary_columns": excluded_monetary_columns,
"unresolved_monetary_columns": unresolved_monetary_columns,
"candidate_row_count": monetary_candidate_count,
"row_count": frame.height,
"preview": frame.head(20).to_dicts(),
"missing_required_mapping": [],
"row_disposition_counts": dict(sorted(disposition_counts.items())),
"row_dispositions": row_dispositions,
"limitations": (
[
"Rows without actual dates are emitted only for explicit-reference matching."
]
if disposition_counts["emitted_reference_only"]
else []
),
**pdf_metadata,
}
return frame, diagnostics
def _extract_pdf_text(path: Path) -> str:
import pdfplumber
lines: list[str] = []
with pdfplumber.open(path) as pdf:
for page in pdf.pages:
text = page.extract_text() or ""
lines.extend(line.strip() for line in text.splitlines() if line.strip())
return "\n".join(lines)
def _classify_bank_pdf_non_movement_line(line: str) -> tuple[str, str] | None:
"""Classify mechanically identifiable bank statement summary rows.
These deterministic rules are intentionally narrow: they remove only explicit
balance, total, scalare, and conditions lines that are statement metadata,
while leaving ambiguous fee/payment descriptions in the matching population.
"""
label = _norm_label(line)
for classification, reason, pattern in BANK_PDF_NON_MOVEMENT_PATTERNS:
if pattern.search(label):
return classification, reason
return None
def _count_classifications(rows: Sequence[dict[str, Any]]) -> dict[str, int]:
counts: dict[str, int] = {}
for row in rows:
classification = _clean_text(row.get("classification"))
if classification:
counts[classification] = counts.get(classification, 0) + 1
return counts
def _normalize_text_pdf(
path: Path,
side: str,
recipe: dict[str, Any],
*,
source_identity: str,
) -> tuple[pl.DataFrame, dict[str, Any]]:
"""Inspect PDF text without emitting movements from an unbounded layout.
A line containing a date and amount is not enough evidence of a transaction
boundary, amount side, or statement layout. No generic PDF movement adapter
is registered; future adapters must be source-family-specific and reviewed.
The narrow non-movement classifications remain useful review evidence.
"""
text = _extract_pdf_text(path)
non_movement_rows: list[dict[str, Any]] = []
candidate_movement_rows = 0
for line_idx, line in enumerate(text.splitlines(), start=1):
parsed_date = _parse_date(line)
amounts = [
_parse_number(match.group(0)) for match in AMOUNT_TOKEN_RE.finditer(line)
]
amounts = [value for value in amounts if value is not None]
if not amounts:
continue
amount = amounts[-1]
classification = (
_classify_bank_pdf_non_movement_line(line) if side == "bank" else None
)
if classification is not None:
class_name, reason = classification
non_movement_rows.append(
{
"side": side,
"source_file": source_identity,
"source_sheet": "PDF",
"source_row": line_idx,
"classification": class_name,
"reason": reason,
"transaction_date": (
parsed_date.isoformat() if parsed_date else None
),
"amount_signed": decimal_text(amount),
"amount_abs": decimal_text(abs(amount)),
"description": line,
}
)
continue
candidate_movement_rows += 1
frame = _transaction_frame([])
file_recipe = _mapping_for_file(
recipe,
side,
path,
source_identity=source_identity,
)
requested_adapter = _clean_text(file_recipe.get("pdf_adapter"))
return frame, {
"source_file": source_identity,
"source_sheet": "PDF",
"parser": "text_pdf_disabled",
"adapter_id": TEXT_PDF_ADAPTER_ID,
"source_family": "text_pdf.unqualified.v1",
"qualification_status": "unsupported_source_layout",
"candidate_row_count": candidate_movement_rows,
"row_count": 0,
"excluded_non_movement_row_count": len(non_movement_rows),
"non_movement_classifications": _count_classifications(non_movement_rows),
"non_movement_rows": non_movement_rows,
"preview": [],
"missing_required_mapping": ["supported reviewed PDF layout adapter"],
"requested_pdf_adapter": requested_adapter or None,
"limitations": [
"Generic text-PDF movement extraction is disabled because row boundaries and amount-side meaning are unproven.",
(
f"Requested adapter {requested_adapter!r} is not registered."
if requested_adapter
else "No supported reviewed PDF layout adapter was supplied."
),
],
}
def _parser_failure_diagnostic(
path: Path,
*,
source_identity: str,
error: BaseException,
) -> dict[str, Any]:
return {
"source_file": source_identity,
"source_sheet": None,
"parser": "failed",
"adapter_id": TABULAR_ADAPTER_ID,
"source_family": "parser_failure.v1",
"qualification_status": "unsupported_source_layout",
"failure_kind": "parser_failure",
"candidate_row_count": 0,
"row_count": 0,
"preview": [],
"missing_required_mapping": [],
"row_disposition_counts": {},
"row_dispositions": [],
"csv_field_delimiter": None,
"csv_field_delimiter_origin": "unresolved",
"csv_field_delimiter_profile": None,
"parser_error": f"{type(error).__name__}: {error}",
"limitations": [
"The source could not be parsed; parser failure is distinct from an unsupported readable layout."
],
}
def _normalize_files(
input_path: Path,
side: str,
recipe: dict[str, Any],
*,
source_refs: dict[tuple[str, str], str],
) -> tuple[pl.DataFrame, list[dict[str, Any]]]:
frames: list[pl.DataFrame] = []
diagnostics: list[dict[str, Any]] = []
for file_path in supported_files(input_path):
source_identity = _source_identity(input_path, file_path)
try:
frame, diag = _normalize_table(
file_path,
side,
recipe,
source_identity=source_identity,
source_artifact_ref=source_refs[(side, source_identity)],
)
except (
fastexcel.CalamineError,
BadZipFile,
InvalidFileException,
pl.exceptions.PolarsError,
UnicodeError,
ValueError,
RuntimeError,
OSError,
) as exc:
frame = _transaction_frame([])
diag = _parser_failure_diagnostic(
file_path,
source_identity=source_identity,
error=exc,
)
frames.append(frame)
diagnostics.append(diag)
if not frames:
return _transaction_frame([]), diagnostics
return pl.concat(frames, how="diagonal_relaxed"), diagnostics
def _source_artifact_receipts(
inputs: Sequence[tuple[str, Path | None]],
) -> tuple[list[dict[str, Any]], dict[tuple[str, str], str]]:
"""Create stable source-byte receipts used by qualifications and lineage."""
receipts: list[dict[str, Any]] = []
refs: dict[tuple[str, str], str] = {}
for side, input_path in inputs:
if input_path is None:
continue
files = supported_files(input_path)
root = _input_root(input_path)
for index, source in enumerate(files, start=1):
source_identity = _source_identity(input_path, source)
_, source_sha256 = file_snapshot(source)
artifact_id = f"source.{side}.{index}.{source_sha256}"
receipt = artifact_receipt(
root,
source,
artifact_id=artifact_id,
root_id=f"source_{side}",
role="source",
)
if receipt["sha256"] != source_sha256:
raise ValueError(f"Source changed while receipt was built: {source}")
receipts.append(receipt)
refs[(side, source_identity)] = artifact_id
return receipts, refs
def _source_snapshot_changes(
before: Sequence[dict[str, Any]],
after: Sequence[dict[str, Any]],
) -> set[tuple[str, str]]:
def indexed(
receipts: Sequence[dict[str, Any]],
) -> dict[tuple[str, str], tuple[int, str]]:
return {
(str(item["root_id"]), str(item["path"])): (
int(item["byte_count"]),
str(item["sha256"]),
)
for item in receipts
}
before_by_source = indexed(before)
after_by_source = indexed(after)
keys = set(before_by_source) | set(after_by_source)
return {
key for key in keys if before_by_source.get(key) != after_by_source.get(key)
}
def _mark_changed_source_diagnostics(
side: str,
diagnostics: Sequence[dict[str, Any]],
changed_sources: set[tuple[str, str]],
) -> list[dict[str, Any]]:
source_root_changed = any(
root_id == f"source_{side}" for root_id, _ in changed_sources
)
if not source_root_changed:
return list(diagnostics)
updated: list[dict[str, Any]] = []
for diagnostic in diagnostics:
updated.append(
{
**diagnostic,
"qualification_status": "unsupported_source_layout",
"failure_kind": "source_changed_during_run",
"row_count": 0,
"preview": [],
"mapping_decision": None,
"reviewed_mapping_ref": None,
"limitations": [
*list(diagnostic.get("limitations") or []),
"Source bytes or membership within the affected source root "
"changed between the pre-parser and post-parser snapshots; "
"all prepared rows from that root were discarded.",
],
}
)
return updated
def _mark_changed_sample_diagnostics(
diagnostics: Sequence[dict[str, Any]],
changed_sources: set[tuple[str, str]],
) -> list[dict[str, Any]]:
changed_paths = {
source_path
for root_id, source_path in changed_sources
if root_id == "source_sample"
}
updated: list[dict[str, Any]] = []
for diagnostic in diagnostics:
source_file = str(diagnostic["source_file"])
if source_file not in changed_paths:
updated.append(diagnostic)
continue
updated.append(
{
**diagnostic,
"status": "invalid",
"failure_kind": "source_changed_during_run",
"movement_count": 0,
"limitations": [
"Sample bytes or sample membership changed between the "
"pre-parser and post-parser snapshots."
],
}
)
return updated
def _control(
control_id: str,
status: str,
detail: str,
*,
evidence_refs: Sequence[str],
) -> dict[str, Any]:
return {
"control_id": control_id,
"required": True,
"status": status,
"evidence_refs": list(evidence_refs),
"detail": detail,
}
def _source_qualifications(
side: str,
diagnostics: Sequence[dict[str, Any]],
source_refs: dict[tuple[str, str], str],
) -> list[dict[str, Any]]:
qualifications: list[dict[str, Any]] = []
for index, diagnostic in enumerate(diagnostics, start=1):
source_file = str(diagnostic["source_file"])
artifact_ref = source_refs.get((side, source_file))
if artifact_ref is None:
continue
status = str(
diagnostic.get("qualification_status") or "unsupported_source_layout"
)
adapter_id = str(diagnostic.get("adapter_id") or TABULAR_ADAPTER_ID)
if adapter_id == TABULAR_ADAPTER_ID:
adapter_version = TABULAR_ADAPTER_VERSION
elif adapter_id == EXTENDED_TABULAR_ADAPTER_ID:
adapter_version = EXTENDED_TABULAR_ADAPTER_VERSION
elif adapter_id == FRENCH_TABULAR_ADAPTER_ID:
adapter_version = FRENCH_TABULAR_ADAPTER_VERSION
elif adapter_id == TEXT_PDF_ADAPTER_ID:
adapter_version = TEXT_PDF_ADAPTER_VERSION
elif adapter_id == PDF_TABLE_ADAPTER_ID:
adapter_version = PDF_TABLE_ADAPTER_VERSION
else:
raise ValueError(f"Unsupported Journal-Bank adapter: {adapter_id}")
supported = status == "qualified"
needs_review = status == "needs_review"
control_status = (
"passed" if supported else "not_assessed" if needs_review else "failed"
)
qualification = build_source_qualification(
qualification_id=f"qualification.{side}.{index}",
adapter_id=adapter_id,
adapter_version=adapter_version,
source_family=str(
diagnostic.get("source_family") or "tabular.explicit_columns.v1"
),
status=status,
source_artifact_refs=[artifact_ref],
candidate_row_count=int(diagnostic.get("candidate_row_count") or 0),
emitted_row_count=int(diagnostic.get("row_count") or 0),
controls=[
_control(
"bounded_adapter",
control_status,
(
"A bounded tabular adapter identified explicit transaction columns."
if supported
else (
"A reviewed mapping is required before the bounded adapter can run."
if needs_review
else "No supported bounded adapter can emit movement rows from this layout."
)
),
evidence_refs=[artifact_ref],
),
_control(
"required_fields",
control_status,
(
"Date and exact monetary source fields are mapped."
if supported
else (
"Date and monetary ownership remain subject to mapping review."
if needs_review
else "Required transaction fields are not proven for this layout."
)
),
evidence_refs=[artifact_ref],
),
_control(
"monetary_field_disposition",
control_status,
(
"Every populated monetary candidate column is mapped or "
"explicitly excluded by the reviewed mapping."
if supported
else (
"One or more populated monetary candidate columns still "
"need an explicit mapped or excluded disposition."
if needs_review
else "Monetary column ownership is not proven for this layout."
)
),
evidence_refs=[artifact_ref],
),
],
reviewed_mapping_ref=(
str(diagnostic["reviewed_mapping_ref"])
if diagnostic.get("reviewed_mapping_ref")
else None
),
limitations=list(diagnostic.get("limitations") or []),
)
qualifications.append(validate_source_qualification(qualification))
return qualifications
def _aggregate_qualification_status(
qualifications: Sequence[dict[str, Any]],
) -> str:
statuses = {str(item["status"]) for item in qualifications}
if not qualifications or "unsupported_source_layout" in statuses:
return "unsupported_source_layout"
if "needs_review" in statuses:
return "needs_review"
return "qualified"
def _source_outcomes(
bank_diagnostics: Sequence[dict[str, Any]],
journal_diagnostics: Sequence[dict[str, Any]],
) -> list[dict[str, Any]]:
"""Project the frozen source-outcome matrix without altering shared records."""
outcomes: list[dict[str, Any]] = []
for side, diagnostics in (
("bank", bank_diagnostics),
("journal", journal_diagnostics),
):
statuses = [
str(diagnostic.get("qualification_status") or "unsupported_source_layout")
for diagnostic in diagnostics
]
if "unsupported_source_layout" in statuses:
qualification_status = "unsupported_source_layout"
elif "needs_review" in statuses:
qualification_status = "needs_review"
else:
qualification_status = "qualified"
failure_kinds = sorted(
{
str(diagnostic["failure_kind"])
for diagnostic in diagnostics
if diagnostic.get("failure_kind")
}
)
outcomes.append(
{
"side": side,
"qualification_status": qualification_status,
"failure_kind": (
"parser_failure"
if "parser_failure" in failure_kinds
else failure_kinds[0] if failure_kinds else None
),
"emitted_row_count": sum(
int(diagnostic.get("row_count") or 0) for diagnostic in diagnostics
),
}
)
return outcomes
def _mapping_decisions(
diagnostics: Sequence[dict[str, Any]],
) -> list[dict[str, Any]]:
decisions = [
item["mapping_decision"]
for item in diagnostics
if isinstance(item.get("mapping_decision"), dict)
]
decision_ids = [str(item["decision_id"]) for item in decisions]
if len(decision_ids) != len(set(decision_ids)):
raise ValueError("mapping decision IDs must be unique")
return decisions
def _lineage_payload(
bank: pl.DataFrame,
journal: pl.DataFrame,
source_refs: dict[tuple[str, str], str],
) -> dict[str, Any]:
entries: list[dict[str, Any]] = []
for side, frame in (("bank", bank), ("journal", journal)):
for row in frame.to_dicts():
source_file = str(row["source_file"])
source_sheet = _clean_text(row.get("source_sheet"))
locator = (
f"sheet:{source_sheet};row:{row['source_row']}"
if source_sheet and source_sheet != "CSV"
else f"row:{row['source_row']}"
)
entries.append(
{
"prepared_id": str(row["transaction_id"]),
"side": side,
"source_artifact_ref": source_refs[(side, source_file)],
"source_locator": locator,
}
)
return {
"schema_version": LINEAGE_SCHEMA_VERSION,
"normalization_schema_version": NORMALIZATION_SCHEMA_VERSION,
"entries": entries,
}
def _gate_payload(
*,
source_status: str,
source_refs: Sequence[str],
preparation_status: str,
reconciliation_status: str,
reconciliation_ref: str | None = None,
semantic_status: str = "not_assessed",
reporting_status: str = "blocked",
limitation: str | None = None,
) -> dict[str, Any]:
limits = [limitation] if limitation else []
return build_gate_register(
{
"source": {
"status": source_status,
"evidence_refs": list(source_refs) if source_status == "passed" else [],
"limitations": [] if source_status == "passed" else limits,
},
"preparation": {
"status": preparation_status,
"evidence_refs": (
["output.lineage_json"] if preparation_status == "passed" else []
),
"limitations": [] if preparation_status == "passed" else limits,
},
"reconciliation": {
"status": reconciliation_status,
"evidence_refs": (
[reconciliation_ref or "output.relationship_ledger_json"]
if reconciliation_status == "passed"
else []
),
"limitations": [] if reconciliation_status == "passed" else limits,
},
"semantic_review": {
"status": semantic_status,
"evidence_refs": (
["review.applied"] if semantic_status == "passed" else []
),
"limitations": [
"Professional review remains outside deterministic matching."
],
},
"reporting": {
"status": reporting_status,
"evidence_refs": (
["artifact.receipts"] if reporting_status == "passed" else []
),
"limitations": (
[]
if reporting_status == "passed"
else ["Reporting is withheld pending semantic review."]
),
},
"publication": {
"status": "blocked",
"evidence_refs": [],
"limitations": ["Publication is outside this component."],
},
}
)
def _missing_mapping(mapping: dict[str, Any]) -> list[str]:
missing = []
if not mapping.get("date"):
missing.append("date")
if not (mapping.get("amount") or mapping.get("debit") or mapping.get("credit")):
missing.append("amount or debit/credit")
return missing
def _recipe_side(diagnostics: list[dict[str, Any]]) -> dict[str, Any]:
files = {}
for diag in diagnostics:
entry: dict[str, Any] = {"parser": diag.get("parser")}
if diag.get("header_rows"):
entry["header_rows"] = diag["header_rows"]
if diag.get("mapping"):
entry["mapping"] = diag["mapping"]
entry["direction_value_mapping"] = diag.get("direction_value_mapping", {})
entry["observed_direction_values"] = diag.get("observed_direction_values", [])
entry["date_convention"] = diag.get("date_convention")
date_evidence = diag.get(
"date_interpretation_evidence",
{
"date_convention": None,
"status_counts": {},
"source_rows": {},
},
)
entry["date_interpretation_evidence"] = date_evidence
if diag.get("date_locale") is not None or (
isinstance(date_evidence, dict)
and int(
(date_evidence.get("status_counts") or {}).get(
"locale_required",
0,
)
)
> 0
):
entry["date_locale"] = diag.get("date_locale")
if diag.get("non_movement_summary_labels"):
entry["non_movement_summary_labels"] = diag["non_movement_summary_labels"]
entry["mapping_origin"] = diag.get("mapping_origin")
entry["mapping_review_content"] = diag.get("mapping_review_content")
entry["mapping_review_content_sha256"] = diag.get(
"mapping_review_content_sha256"
)
entry["mapping_decision"] = diag.get("mapping_decision")
entry["csv_field_delimiter"] = diag.get("csv_field_delimiter")
entry["csv_field_delimiter_origin"] = diag.get("csv_field_delimiter_origin")
entry["csv_field_delimiter_profile"] = diag.get("csv_field_delimiter_profile")
entry["potential_monetary_columns"] = diag.get("potential_monetary_columns", [])
entry["excluded_monetary_columns"] = diag.get("excluded_monetary_columns", [])
entry["unresolved_monetary_columns"] = diag.get(
"unresolved_monetary_columns", []
)
files[str(diag["source_file"])] = entry
return {"files": files}
def _read_sample_movements(
sample_path: Path | None,
) -> tuple[set[str], list[dict[str, Any]]]:
if sample_path is None:
return set(), []
values: set[str] = set()
diagnostics: list[dict[str, Any]] = []
for source in supported_files(sample_path):
source_identity = _source_identity(sample_path, source)
if source.suffix.lower() == ".pdf":
diagnostics.append(
{
"source_file": source_identity,
"status": "invalid",
"failure_kind": "unsupported_sample_layout",
"movement_count": 0,
"csv_field_delimiter": None,
"csv_field_delimiter_profile": None,
}
)
continue
delimiter_resolution: dict[str, Any] | None = None
try:
sheet_names = _source_sheet_names(source)
if len(sheet_names) != 1:
diagnostics.append(
{
"source_file": source_identity,
"status": "invalid",
"failure_kind": "multiple_sheets_unsupported",
"movement_count": 0,
"workbook_sheets": sheet_names,
"csv_field_delimiter": None,
"csv_field_delimiter_profile": None,
}
)
continue
delimiter_resolution = _resolve_csv_field_delimiter(source, {})
# A one-column sample has no field boundary to interpret. Using the
# comma default is therefore transport-only and cannot promote
# transaction rows or infer a professional mapping.
single_column_default = (
source.suffix.lower() == ".csv"
and delimiter_resolution["status"] == "unsupported"
and delimiter_resolution["profile"]["candidate_delimiters"] == []
)
if (
delimiter_resolution["status"]
not in {
"resolved",
"not_applicable",
}
and not single_column_default
):
diagnostics.append(
{
"source_file": source_identity,
"status": "invalid",
"failure_kind": (
"ambiguous_csv_field_delimiter"
if delimiter_resolution["status"] == "ambiguous"
else "unsupported_csv_field_delimiter"
),
"movement_count": 0,
"csv_field_delimiter": None,
"csv_field_delimiter_profile": delimiter_resolution["profile"],
}
)
continue
sample_delimiter = (
DEFAULT_CSV_FIELD_DELIMITER
if single_column_default
else delimiter_resolution["delimiter"]
)
raw = _read_table_raw(
source,
csv_field_delimiter=(
str(sample_delimiter) if sample_delimiter is not None else None
),
)
header_candidates: list[tuple[int, str]] = []
sample_aliases = (
EXACT_HEADER_ALIASES["movement_number"]
| EXACT_HEADER_ALIASES["reference"]
)
for row_index in range(min(raw.height, 30)):
labels = _merge_header_rows([_row_values(raw, row_index)])
matches = [
label for label in labels if _norm_label(label) in sample_aliases
]
if len(matches) == 1:
header_candidates.append((row_index + 1, matches[0]))
if len(header_candidates) != 1:
diagnostics.append(
{
"source_file": source_identity,
"status": "invalid",
"failure_kind": "ambiguous_sample_identifier_column",
"movement_count": 0,
"csv_field_delimiter": sample_delimiter,
"csv_field_delimiter_profile": delimiter_resolution["profile"],
}
)
continue
header_row, identifier_column = header_candidates[0]
table = _apply_header(
raw,
[header_row],
source_sheet=_source_sheet_name(source),
)
file_values = {
_clean_text(value)
for value in table.get_column(identifier_column)
.cast(pl.Utf8, strict=False)
.to_list()
if _clean_text(value)
}
values.update(file_values)
diagnostics.append(
{
"source_file": source_identity,
"status": "qualified" if file_values else "invalid",
"failure_kind": None if file_values else "empty_sample",
"movement_count": len(file_values),
"header_row": header_row,
"identifier_column": identifier_column,
"csv_field_delimiter": sample_delimiter,
"csv_field_delimiter_profile": delimiter_resolution["profile"],
}
)
except (
fastexcel.CalamineError,
BadZipFile,
InvalidFileException,
pl.exceptions.PolarsError,
UnicodeError,
ValueError,
RuntimeError,
OSError,
) as exc:
diagnostics.append(
{
"source_file": source_identity,
"status": "invalid",
"failure_kind": "parser_failure",
"movement_count": 0,
"parser_error": f"{type(exc).__name__}: {exc}",
"csv_field_delimiter": (
delimiter_resolution["delimiter"]
if delimiter_resolution is not None
else None
),
"csv_field_delimiter_profile": (
delimiter_resolution["profile"]
if delimiter_resolution is not None
else None
),
}
)
if diagnostics and any(item["status"] != "qualified" for item in diagnostics):
return set(), diagnostics
return values, diagnostics
def inspect_inputs(
bank_path: Path,
journal_path: Path,
output_dir: Path,
recipe_path: Path | None = None,
*,
sample_path: Path | None = None,
language: object | None = None,
document_language: object | None = None,
) -> InspectionResult:
"""Inspect input files and write deterministic recipe artifacts."""
recipe = read_json(recipe_path)
languages = language_assumptions(
recipe, language=language, document_language=document_language
)
initial_source_receipts, source_refs = _source_artifact_receipts(
(("bank", bank_path), ("journal", journal_path), ("sample", sample_path))
)
bank_frame, bank_diag = _normalize_files(
bank_path,
"bank",
recipe,
source_refs=source_refs,
)
journal_frame, journal_diag = _normalize_files(
journal_path,
"journal",
recipe,
source_refs=source_refs,
)
sample_movements, sample_diagnostics = _read_sample_movements(sample_path)
source_receipts, current_source_refs = _source_artifact_receipts(
(("bank", bank_path), ("journal", journal_path), ("sample", sample_path))
)
changed_sources = _source_snapshot_changes(
initial_source_receipts,
source_receipts,
)
source_refs = current_source_refs
if changed_sources:
bank_diag = _mark_changed_source_diagnostics("bank", bank_diag, changed_sources)
journal_diag = _mark_changed_source_diagnostics(
"journal", journal_diag, changed_sources
)
sample_diagnostics = _mark_changed_sample_diagnostics(
sample_diagnostics,
changed_sources,
)
if any(root_id == "source_bank" for root_id, _ in changed_sources):
bank_frame = _transaction_frame([])
if any(root_id == "source_journal" for root_id, _ in changed_sources):
journal_frame = _transaction_frame([])
if any(root_id == "source_sample" for root_id, _ in changed_sources):
sample_movements = set()
qualifications = [
*_source_qualifications("bank", bank_diag, source_refs),
*_source_qualifications("journal", journal_diag, source_refs),
]
qualification_status = _aggregate_qualification_status(qualifications)
if changed_sources:
qualification_status = "unsupported_source_layout"
sample_status = (
"not_supplied"
if sample_path is None
else "qualified" if sample_movements else "invalid_or_empty"
)
proposed_relationship_policy = {
"relationship_shape": "many_to_many",
"allow_evidence_reuse": False,
"require_same_currency": True,
"require_same_unit": True,
"require_same_entity": True,
"require_same_party": False,
"direction_policy": "absolute_amount",
"default_currency": "EUR",
"default_unit": "currency",
"default_entity_ref": "entity.case",
"default_party_ref": None,
"amount_tolerance": "1",
"date_window_days": 7,
}
normalized_relationship_policy = _normalize_relationship_policy(
proposed_relationship_policy
)
suggested_recipe = {
"version": 2,
"description": "Bounded journal-bank reconciliation recipe for professional review.",
**languages,
"bank": _recipe_side(bank_diag),
"journal": _recipe_side(journal_diag),
"matching": {
"amount_tolerance": "1",
"date_window_days": 7,
"use_absolute_amounts": True,
"stages": list(MATCH_STAGE_ORDER),
},
"relationship": {
"policy": normalized_relationship_policy,
"review_content_sha256": canonical_json_sha256(
{"policy": normalized_relationship_policy}
),
"decision": None,
},
}
output_dir.mkdir(parents=True, exist_ok=True)
write_assurance_json(
output_dir / "input_receipts.json",
{
"schema_version": "journal_bank.input_receipts.v1",
"receipts": source_receipts,
},
)
write_assurance_json(
output_dir / "source_qualifications.json",
{
"schema_version": "journal_bank.source_qualifications.v1",
"status": qualification_status,
"source_outcomes": _source_outcomes(bank_diag, journal_diag),
"qualifications": qualifications,
},
)
write_assurance_json(
output_dir / "reviewed_decisions.json",
{
"schema_version": "journal_bank.reviewed_decisions.v1",
"decisions": _mapping_decisions([*bank_diag, *journal_diag]),
},
)
write_assurance_json(
output_dir / "lineage.json",
_lineage_payload(bank_frame, journal_frame, source_refs),
)
write_json(
output_dir / "inspection.json",
{
**languages,
"qualification_status": qualification_status,
"source_snapshot_changed": bool(changed_sources),
"changed_sources": [
{"root_id": root_id, "path": source_path}
for root_id, source_path in sorted(changed_sources)
],
"bank": {"row_count": bank_frame.height, "files": bank_diag},
"bank_pdf_non_movement_row_count": len(_non_movement_records(bank_diag)),
"journal": {"row_count": journal_frame.height, "files": journal_diag},
"sample": {
"path": sample_path.as_posix() if sample_path else None,
"status": sample_status,
"movement_count": len(sample_movements),
"movements": sorted(sample_movements)[:100],
"diagnostics": sample_diagnostics,
},
},
)
write_json(output_dir / "suggested_recipe.json", suggested_recipe)
return InspectionResult(
bank={"row_count": bank_frame.height, "files": bank_diag},
journal={"row_count": journal_frame.height, "files": journal_diag},
sample={"movement_count": len(sample_movements), "status": sample_status},
suggested_recipe=suggested_recipe,
)
def _date_diff_days(left: Any, right: Any) -> int | None:
left_date = _parse_date(left)
right_date = _parse_date(right)
if left_date is None or right_date is None:
return None
return abs((left_date - right_date).days)
def _canonical_tolerance(value: object) -> tuple[Decimal, str]:
"""Return an exact non-negative tolerance and its canonical text.
A reviewed relationship tolerance is authoritative policy, not localized
source data. Accepting only canonical text, ``Decimal``, or integers avoids
hidden locale inference and binary-float rounding at that boundary.
"""
if isinstance(value, bool):
raise MoneyValidationError(
"tolerance must be canonical Decimal text, int, or Decimal"
)
if isinstance(value, str):
tolerance = parse_canonical_decimal(value, label="tolerance")
elif isinstance(value, Decimal):
tolerance = value
elif isinstance(value, int):
tolerance = Decimal(value)
else:
raise MoneyValidationError(
"tolerance must be canonical Decimal text, int, or Decimal"
)
if not tolerance.is_finite():
raise MoneyValidationError("tolerance must be finite")
if tolerance < ZERO:
raise MoneyValidationError("tolerance must not be negative")
return tolerance, decimal_text(tolerance)
def _validated_relationship_decision(
recipe: dict[str, Any],
*,
source_artifact_refs: Sequence[str],
tolerance_text: str,
date_window_days: int,
) -> tuple[dict[str, Any] | None, dict[str, Any] | None, str | None]:
relationship = recipe.get("relationship")
if not isinstance(relationship, dict):
return None, None, "A reviewed relationship and perimeter policy is required."
try:
policy = _normalize_relationship_policy(relationship.get("policy"))
except ValueError as exc:
return None, None, str(exc)
if policy["amount_tolerance"] != tolerance_text:
return None, None, "Execution tolerance differs from the reviewed policy."
if policy["date_window_days"] != date_window_days:
return None, None, "Execution date window differs from the reviewed policy."
raw_decision = relationship.get("decision")
if not isinstance(raw_decision, dict):
return None, None, "A reviewed relationship decision receipt is required."
try:
decision = validate_reviewed_decision_receipt(
raw_decision,
expected_source_artifact_refs=list(source_artifact_refs),
expected_adapter_id=RELATIONSHIP_ADAPTER_ID,
expected_adapter_version=RELATIONSHIP_ADAPTER_VERSION,
require_reviewed=True,
)
except ValueError as exc:
return None, None, f"Relationship decision is invalid or stale: {exc}"
if decision["decision_type"] != "journal_bank_relationship":
return None, None, "Relationship decision has the wrong decision type."
if decision["content"] != {"policy": policy}:
return None, None, "Relationship decision content does not match the policy."
return policy, decision, None
def _with_policy_defaults(frame: pl.DataFrame, policy: dict[str, Any]) -> pl.DataFrame:
result = frame
defaults = {
"currency": policy["default_currency"],
"unit": policy["default_unit"],
"entity_ref": policy["default_entity_ref"],
"party_ref": policy["default_party_ref"],
}
for field, default in defaults.items():
if default is None:
continue
result = result.with_columns(
pl.when(
pl.col(field).is_null()
| (pl.col(field).cast(pl.Utf8, strict=False).str.strip_chars() == "")
)
.then(pl.lit(default))
.otherwise(pl.col(field))
.alias(field)
)
return result
def _relationship_perimeter_error(
bank: pl.DataFrame,
journal: pl.DataFrame,
policy: dict[str, Any],
) -> str | None:
required = ["currency", "unit"]
if policy["require_same_entity"]:
required.append("entity_ref")
if policy["require_same_party"]:
required.append("party_ref")
for side, frame in (("bank", bank), ("journal", journal)):
for field in required:
missing = frame.filter(
pl.col(field).is_null()
| (pl.col(field).cast(pl.Utf8, strict=False).str.strip_chars() == "")
)
if not missing.is_empty():
return (
f"{side} rows lack required {field}; provide a source field or "
"a reviewed policy default."
)
return None
def _direction_compatible(
bank_row: dict[str, Any],
journal_row: dict[str, Any],
policy: dict[str, Any],
) -> bool:
direction_policy = str(policy["direction_policy"])
if direction_policy == "absolute_amount":
return True
bank_direction = _clean_text(bank_row.get("direction"))
journal_direction = _clean_text(journal_row.get("direction"))
if bank_direction not in {"positive", "negative", "zero"}:
return False
if journal_direction not in {"positive", "negative", "zero"}:
return False
if direction_policy == "same_sign":
return bank_direction == journal_direction
return {bank_direction, journal_direction} == {"positive", "negative"}
def _same_required_perimeter(
bank_row: dict[str, Any],
journal_row: dict[str, Any],
policy: dict[str, Any],
) -> bool:
comparisons = [
("currency", policy["require_same_currency"]),
("unit", policy["require_same_unit"]),
("entity_ref", policy["require_same_entity"]),
("party_ref", policy["require_same_party"]),
]
return all(
not required
or (
_clean_text(bank_row.get(field))
and _clean_text(bank_row.get(field)) == _clean_text(journal_row.get(field))
)
for field, required in comparisons
) and _direction_compatible(bank_row, journal_row, policy)
@dataclass(frozen=True)
class _JournalAmountIndex:
"""Exact Decimal range index for mechanically eligible journal amounts.
The index only removes rows outside the authoritative reviewed-tolerance
predicate. Existing perimeter, date, reference, and non-reuse checks remain
authoritative for every row returned by the index.
"""
amounts: tuple[Decimal, ...]
positioned_rows: tuple[tuple[int, dict[str, Any]], ...]
@classmethod
def from_rows(cls, journal_rows: Sequence[dict[str, Any]]) -> _JournalAmountIndex:
indexed_rows: list[tuple[Decimal, int, dict[str, Any]]] = []
for position, journal_row in enumerate(journal_rows):
journal_value = journal_row.get("amount_abs")
if not isinstance(journal_value, str):
continue
indexed_rows.append(
(
parse_canonical_decimal(journal_value, label="journal amount"),
position,
journal_row,
)
)
indexed_rows.sort(key=lambda item: (item[0], item[1]))
return cls(
amounts=tuple(item[0] for item in indexed_rows),
positioned_rows=tuple((item[1], item[2]) for item in indexed_rows),
)
def rows_within_tolerance(
self, amount: Decimal, tolerance: Decimal
) -> list[dict[str, Any]]:
"""Return predicate-approved candidates in original journal order.
Searching outward from the insertion point avoids calculating Decimal
bounds that the ambient context could round inward. The authoritative
tolerance predicate decides both inclusion and where each scan stops.
"""
pivot = bisect_left(self.amounts, amount)
positioned_candidates: list[tuple[int, dict[str, Any]]] = []
left = pivot - 1
while left >= 0:
_, within_tolerance = difference_within_tolerance(
amount, self.amounts[left], tolerance
)
if not within_tolerance:
break
positioned_candidates.append(self.positioned_rows[left])
left -= 1
right = pivot
while right < len(self.amounts):
_, within_tolerance = difference_within_tolerance(
amount, self.amounts[right], tolerance
)
if not within_tolerance:
break
positioned_candidates.append(self.positioned_rows[right])
right += 1
positioned_candidates.sort(key=lambda item: item[0])
return [row for _, row in positioned_candidates]
def _candidate_rows(
bank_row: dict[str, Any],
journal_index: _JournalAmountIndex,
used_journal: set[str],
*,
tolerance: Decimal,
date_window_days: int,
relationship_policy: dict[str, Any],
) -> list[dict[str, Any]]:
candidates = []
bank_value = bank_row.get("amount_abs")
bank_amount = (
parse_canonical_decimal(bank_value, label="bank amount")
if isinstance(bank_value, str)
else None
)
if bank_amount is None:
return candidates
for journal_row in journal_index.rows_within_tolerance(bank_amount, tolerance):
journal_id = str(journal_row["transaction_id"])
if journal_id in used_journal:
continue
journal_value = journal_row.get("amount_abs")
journal_amount = (
parse_canonical_decimal(journal_value, label="journal amount")
if isinstance(journal_value, str)
else None
)
if journal_amount is None:
continue
if not _same_required_perimeter(bank_row, journal_row, relationship_policy):
continue
signed_delta, within_tolerance = difference_within_tolerance(
bank_amount, journal_amount, tolerance
)
if not within_tolerance:
continue
date_diff = _date_diff_days(
bank_row.get("transaction_date"), journal_row.get("transaction_date")
)
if date_diff is not None and date_diff > date_window_days:
continue
candidates.append(
{
"row": journal_row,
"amount_delta": abs(signed_delta),
"date_diff_days": date_diff,
"shared_references": sorted(
_reference_tokens(
bank_row.get("reference"),
bank_row.get("movement_number"),
)
& _reference_tokens(
journal_row.get("reference"),
journal_row.get("movement_number"),
)
),
}
)
return candidates
def _make_match_record(
bank_row: dict[str, Any],
journal_row: dict[str, Any],
*,
stage: str,
amount_delta: Decimal,
date_diff_days: int | None,
shared_references: Sequence[str],
) -> dict[str, Any]:
return {
"status": "matched",
"stage": stage,
"bank_transaction_id": bank_row["transaction_id"],
"journal_transaction_id": journal_row["transaction_id"],
"bank_date": bank_row.get("transaction_date"),
"journal_date": journal_row.get("transaction_date"),
"date_diff_days": date_diff_days,
"bank_amount": bank_row.get("amount_signed"),
"journal_amount": journal_row.get("amount_signed"),
"amount_delta": decimal_text(amount_delta),
"bank_description": bank_row.get("description"),
"journal_description": journal_row.get("description"),
"shared_references": ",".join(shared_references),
"review_note": "",
}
def _unconflicted_singleton_batch(
bank_rows: Sequence[dict[str, Any]],
journal_index: _JournalAmountIndex,
used_bank: set[str],
used_journal: set[str],
*,
tolerance: Decimal,
date_window_days: int,
relationship_policy: dict[str, Any],
require_shared_reference: bool,
) -> list[tuple[dict[str, Any], dict[str, Any]]]:
"""Return one order-independent batch of conflict-free singleton matches.
Matching is a mechanical one-to-one allocation. Evaluating the complete
batch before accepting any row prevents source order from granting a
journal candidate to the first competing bank row.
"""
singleton_candidates: list[tuple[dict[str, Any], dict[str, Any]]] = []
for bank_row in bank_rows:
if str(bank_row["transaction_id"]) in used_bank:
continue
candidates = _candidate_rows(
bank_row,
journal_index,
used_journal,
tolerance=tolerance,
date_window_days=date_window_days,
relationship_policy=relationship_policy,
)
if require_shared_reference:
candidates = [
candidate for candidate in candidates if candidate["shared_references"]
]
else:
candidates = [
candidate
for candidate in candidates
if candidate["date_diff_days"] is not None
]
if len(candidates) == 1:
singleton_candidates.append((bank_row, candidates[0]))
target_counts = Counter(
str(candidate["row"]["transaction_id"]) for _, candidate in singleton_candidates
)
return sorted(
(
(bank_row, candidate)
for bank_row, candidate in singleton_candidates
if target_counts[str(candidate["row"]["transaction_id"])] == 1
),
key=lambda item: (
str(item[0]["transaction_id"]),
str(item[1]["row"]["transaction_id"]),
),
)
def _unconflicted_reference_group_batch(
bank_rows: Sequence[dict[str, Any]],
journal_rows: Sequence[dict[str, Any]],
used_bank: set[str],
used_journal: set[str],
*,
tolerance: Decimal,
date_window_days: int,
relationship_policy: dict[str, Any],
) -> list[dict[str, Any]]:
"""Return exact-sum reference groups with order-independent membership.
Grouping is deterministic because a shared explicit identifier defines the
population and exact Decimal conservation decides acceptance. Descriptions
and beneficiary similarity never create a group.
"""
shape = str(relationship_policy["relationship_shape"])
if shape == "one_to_one":
return []
available_bank = [
row for row in bank_rows if str(row["transaction_id"]) not in used_bank
]
available_journal = [
row for row in journal_rows if str(row["transaction_id"]) not in used_journal
]
bank_by_token: dict[str, list[dict[str, Any]]] = {}
journal_by_token: dict[str, list[dict[str, Any]]] = {}
for row in available_bank:
for token in _reference_tokens(
row.get("reference"), row.get("movement_number")
):
bank_by_token.setdefault(token, []).append(row)
for row in available_journal:
for token in _reference_tokens(
row.get("reference"), row.get("movement_number")
):
journal_by_token.setdefault(token, []).append(row)
reference_groups = [
(token, bank_by_token[token], journal_by_token[token])
for token in sorted(set(bank_by_token) & set(journal_by_token))
]
# An explicit list can name distinct invoice identifiers without a shared
# batch identifier. Preserve punctuation inside each identifier: splitting
# INV-1 into generic words/numbers would destroy the source identity.
def listed_references(row: dict[str, Any]) -> tuple[str, ...]:
value = str(row.get("reference") or "").strip().casefold()
parts = tuple(re.split(r"[\s,;|]+", value)) if value else ()
if not parts or len(parts) > 100 or len(set(parts)) != len(parts):
return ()
if any(
re.fullmatch(r"[a-z0-9]+(?:[-/.][a-z0-9]+)*", part) is None
or not any(char.isdigit() for char in part)
or not any(char.isalpha() for char in part)
or _is_generic_period_fragment(re.sub(r"[-/.]", "", part))
or GENERIC_PERIOD_REFERENCE_RE.fullmatch(re.sub(r"[-/.]", "", part))
for part in parts
):
return ()
return parts
for anchors, counterparts, reverse in (
(available_bank, available_journal, False),
(available_journal, available_bank, True),
):
by_reference: dict[str, list[dict[str, Any]]] = {}
for row in counterparts:
refs = listed_references(row)
if len(refs) == 1:
by_reference.setdefault(refs[0], []).append(row)
for anchor in anchors:
refs = listed_references(anchor)
if len(refs) < 2 or any(
len(by_reference.get(ref, [])) != 1 for ref in refs
):
continue
members = [by_reference[ref][0] for ref in refs]
reference_groups.append(
(
"explicit_list:" + "|".join(sorted(refs)),
members if reverse else [anchor],
[anchor] if reverse else members,
)
)
grouped: dict[tuple[tuple[str, ...], tuple[str, ...]], dict[str, Any]] = {}
for token, bank_group, journal_group in reference_groups:
bank_count = len(bank_group)
journal_count = len(journal_group)
one_to_many = bank_count == 1 and journal_count > 1
many_to_one = bank_count > 1 and journal_count == 1
if not (
(one_to_many and shape in {"one_to_many", "many_to_many"})
or (many_to_one and shape in {"many_to_one", "many_to_many"})
):
continue
if not all(
_same_required_perimeter(bank_row, journal_row, relationship_policy)
for bank_row in bank_group
for journal_row in journal_group
):
continue
date_differences = [
_date_diff_days(
bank_row.get("transaction_date"),
journal_row.get("transaction_date"),
)
for bank_row in bank_group
for journal_row in journal_group
]
if any(
difference is not None and difference > date_window_days
for difference in date_differences
):
continue
bank_total = sum(
(parse_canonical_decimal(str(row["amount_abs"])) for row in bank_group),
ZERO,
)
journal_total = sum(
(parse_canonical_decimal(str(row["amount_abs"])) for row in journal_group),
ZERO,
)
signed_delta, within_tolerance = difference_within_tolerance(
bank_total,
journal_total,
tolerance,
)
if not within_tolerance:
continue
bank_ids = tuple(sorted(str(row["transaction_id"]) for row in bank_group))
journal_ids = tuple(sorted(str(row["transaction_id"]) for row in journal_group))
key = (bank_ids, journal_ids)
candidate = grouped.setdefault(
key,
{
"bank_rows": bank_group,
"journal_rows": journal_group,
"amount_delta": abs(signed_delta),
"shared_references": [],
},
)
candidate["shared_references"].append(token)
bank_membership: Counter[str] = Counter()
journal_membership: Counter[str] = Counter()
for bank_ids, journal_ids in grouped:
bank_membership.update(bank_ids)
journal_membership.update(journal_ids)
accepted = [
candidate
for (bank_ids, journal_ids), candidate in grouped.items()
if all(bank_membership[item] == 1 for item in bank_ids)
and all(journal_membership[item] == 1 for item in journal_ids)
]
return sorted(
accepted,
key=lambda item: (
tuple(str(row["transaction_id"]) for row in item["bank_rows"]),
tuple(str(row["transaction_id"]) for row in item["journal_rows"]),
),
)
def _match_transactions(
bank: pl.DataFrame,
journal: pl.DataFrame,
*,
tolerance: Decimal,
date_window_days: int,
relationship_policy: dict[str, Any],
) -> tuple[pl.DataFrame, pl.DataFrame, pl.DataFrame, dict[str, int]]:
bank_rows = bank.to_dicts()
journal_rows = journal.to_dicts()
journal_index = _JournalAmountIndex.from_rows(journal_rows)
used_bank: set[str] = set()
used_journal: set[str] = set()
matches: list[dict[str, Any]] = []
stage_counts: dict[str, int] = {}
(
reference_stage,
reference_group_stage,
first_amount_date_stage,
later_amount_date_stage,
) = MATCH_STAGE_ORDER
def accept(bank_row: dict[str, Any], candidate: dict[str, Any], stage: str) -> None:
matches.append(
_make_match_record(
bank_row,
candidate["row"],
stage=stage,
amount_delta=candidate["amount_delta"],
date_diff_days=candidate["date_diff_days"],
shared_references=candidate["shared_references"],
)
)
used_bank.add(str(bank_row["transaction_id"]))
used_journal.add(str(candidate["row"]["transaction_id"]))
stage_counts[stage] = stage_counts.get(stage, 0) + 1
while True:
reference_batch = _unconflicted_singleton_batch(
bank_rows,
journal_index,
used_bank,
used_journal,
tolerance=tolerance,
date_window_days=date_window_days,
relationship_policy=relationship_policy,
require_shared_reference=True,
)
if not reference_batch:
break
for bank_row, candidate in reference_batch:
accept(bank_row, candidate, reference_stage)
while True:
reference_groups = _unconflicted_reference_group_batch(
bank_rows,
journal_rows,
used_bank,
used_journal,
tolerance=tolerance,
date_window_days=date_window_days,
relationship_policy=relationship_policy,
)
if not reference_groups:
break
for group in reference_groups:
bank_group = group["bank_rows"]
journal_group = group["journal_rows"]
if len(bank_group) == 1:
pairs = [(bank_group[0], journal_row) for journal_row in journal_group]
else:
pairs = [(bank_row, journal_group[0]) for bank_row in bank_group]
for bank_row, journal_row in pairs:
matches.append(
_make_match_record(
bank_row,
journal_row,
stage=reference_group_stage,
amount_delta=group["amount_delta"],
date_diff_days=_date_diff_days(
bank_row.get("transaction_date"),
journal_row.get("transaction_date"),
),
shared_references=group["shared_references"],
)
)
used_bank.update(str(row["transaction_id"]) for row in bank_group)
used_journal.update(str(row["transaction_id"]) for row in journal_group)
stage_counts[reference_group_stage] = stage_counts.get(
reference_group_stage, 0
) + len(pairs)
first_amount_date_batch = _unconflicted_singleton_batch(
bank_rows,
journal_index,
used_bank,
used_journal,
tolerance=tolerance,
date_window_days=date_window_days,
relationship_policy=relationship_policy,
require_shared_reference=False,
)
for bank_row, candidate in first_amount_date_batch:
accept(bank_row, candidate, first_amount_date_stage)
while first_amount_date_batch:
later_amount_date_batch = _unconflicted_singleton_batch(
bank_rows,
journal_index,
used_bank,
used_journal,
tolerance=tolerance,
date_window_days=date_window_days,
relationship_policy=relationship_policy,
require_shared_reference=False,
)
if not later_amount_date_batch:
break
for bank_row, candidate in later_amount_date_batch:
accept(bank_row, candidate, later_amount_date_stage)
first_amount_date_batch = later_amount_date_batch
unmatched_bank = _transaction_frame(
[row for row in bank_rows if str(row["transaction_id"]) not in used_bank]
)
unmatched_journal = _transaction_frame(
[row for row in journal_rows if str(row["transaction_id"]) not in used_journal]
)
match_frame = (
pl.DataFrame(matches, infer_schema_length=None)
if matches
else pl.DataFrame(schema={col: pl.Utf8 for col in MATCH_COLUMNS})
)
for col in MATCH_COLUMNS:
if col not in match_frame.columns:
match_frame = match_frame.with_columns(pl.lit(None).alias(col))
return (
match_frame.select(MATCH_COLUMNS),
unmatched_bank,
unmatched_journal,
stage_counts,
)
def _relationship_ledger(
bank: pl.DataFrame,
journal: pl.DataFrame,
matches: pl.DataFrame,
*,
policy: dict[str, Any],
) -> dict[str, Any]:
bank_rows = bank.to_dicts()
journal_rows = journal.to_dicts()
bank_ids = {
str(row["transaction_id"]): f"bank.{index}"
for index, row in enumerate(bank_rows, start=1)
}
journal_ids = {
str(row["transaction_id"]): f"journal.{index}"
for index, row in enumerate(journal_rows, start=1)
}
def record(row: dict[str, Any], record_id: str) -> dict[str, str | None]:
return {
"record_id": record_id,
"amount": str(row["amount_abs"]),
"currency": _clean_text(row["currency"]),
"unit": _clean_text(row["unit"]),
"entity_ref": _clean_text(row.get("entity_ref")) or None,
"party_ref": _clean_text(row.get("party_ref")) or None,
}
bank_by_id = {str(row["transaction_id"]): row for row in bank_rows}
journal_by_id = {str(row["transaction_id"]): row for row in journal_rows}
allocations: list[dict[str, Any]] = []
for index, match in enumerate(matches.to_dicts(), start=1):
bank_transaction_id = str(match["bank_transaction_id"])
journal_transaction_id = str(match["journal_transaction_id"])
bank_row = bank_by_id[bank_transaction_id]
journal_row = journal_by_id[journal_transaction_id]
amount = min(
parse_canonical_decimal(str(bank_row["amount_abs"])),
parse_canonical_decimal(str(journal_row["amount_abs"])),
)
allocations.append(
{
"allocation_id": f"allocation.{index}",
"source_record_ref": bank_ids[bank_transaction_id],
"target_record_ref": journal_ids[journal_transaction_id],
"amount": decimal_text(amount),
"currency": str(bank_row["currency"]),
"unit": str(bank_row["unit"]),
"evidence_refs": [f"match.{index}"],
}
)
return build_allocation_ledger(
ledger_id="journal_bank.relationship",
policy={
"relationship_shape": policy["relationship_shape"],
"require_same_currency": policy["require_same_currency"],
"require_same_unit": policy["require_same_unit"],
"require_same_entity": policy["require_same_entity"],
"require_same_party": policy["require_same_party"],
"allow_evidence_reuse": policy["allow_evidence_reuse"],
"tolerance": policy["amount_tolerance"],
},
source_records=[
record(row, bank_ids[str(row["transaction_id"])]) for row in bank_rows
],
target_records=[
record(row, journal_ids[str(row["transaction_id"])]) for row in journal_rows
],
allocations=allocations,
)
def _blocked_relationship_ledger(
*,
block_code: str,
block_detail: str,
) -> dict[str, Any]:
"""Return explicit zero-allocation evidence when reconciliation never ran."""
content = {
"schema_version": "journal_bank.blocked_relationship_ledger.v1",
"ledger_id": "journal_bank.relationship",
"status": "not_assessed",
"block_code": block_code,
"block_detail": block_detail,
"source_records": [],
"target_records": [],
"allocations": [],
"source_residuals": [],
"target_residuals": [],
"balanced": False,
}
return {**content, "content_sha256": canonical_json_sha256(content)}
def _relationship_residual_frame(
ledger: dict[str, Any],
bank: pl.DataFrame,
journal: pl.DataFrame,
) -> pl.DataFrame:
"""Project allocation residuals without classifying or forcing them to zero."""
validated = validate_allocation_ledger(ledger)
records: list[dict[str, Any]] = []
for side, population, ledger_records, residuals in (
(
"bank",
bank,
validated["source_records"],
validated["source_residuals"],
),
(
"journal",
journal,
validated["target_records"],
validated["target_residuals"],
),
):
population_rows = population.to_dicts()
if not (len(population_rows) == len(ledger_records) == len(residuals)):
raise ValueError("Relationship residual population cardinality is stale.")
for index, (population_row, record, residual_row) in enumerate(
zip(population_rows, ledger_records, residuals, strict=True),
start=1,
):
expected_record_ref = f"{side}.{index}"
record_ref = str(record["record_id"])
if (
record_ref != expected_record_ref
or residual_row["record_ref"] != record_ref
or str(record["amount"]) != str(population_row["amount_abs"])
):
raise ValueError("Relationship residual row identity is stale.")
record_amount = parse_canonical_decimal(
str(record["amount"]),
label=f"{record_ref}.record_amount",
)
residual = parse_canonical_decimal(
str(residual_row["residual"]),
label=f"{record_ref}.residual",
)
allocated_amount = record_amount - residual
records.append(
{
"side": side,
"record_ref": record_ref,
"transaction_id": str(population_row["transaction_id"]),
"record_amount": decimal_text(record_amount),
"allocated_amount": decimal_text(allocated_amount),
"residual": decimal_text(residual),
"currency": str(record["currency"]),
"unit": str(record["unit"]),
"entity_ref": record["entity_ref"],
"party_ref": record["party_ref"],
}
)
if not records:
return pl.DataFrame(schema={field: pl.Utf8 for field in RESIDUAL_COLUMNS})
return pl.DataFrame(records, infer_schema_length=None).select(RESIDUAL_COLUMNS)
def _material_value(value: object, field: str) -> dict[str, str]:
"""Canonicalize one mechanically verifiable match or residual value."""
if value is None or value == "":
return {"kind": "empty", "value": ""}
text = str(value)
if field in MATERIAL_DECIMAL_FIELDS:
parsed = parse_canonical_decimal(text, label=field)
if decimal_text(parsed) != text:
raise ValueError(f"{field} is not canonical Decimal text.")
return {"kind": "decimal", "value": text}
if field in MATERIAL_INTEGER_FIELDS:
try:
parsed_integer = int(text)
except ValueError as exc:
raise ValueError(f"{field} is not canonical integer text.") from exc
if str(parsed_integer) != text:
raise ValueError(f"{field} is not canonical integer text.")
return {"kind": "integer", "value": text}
return {"kind": "text", "value": text}
def _excel_cell(columns: Sequence[str], field: str, row_number: int) -> str:
column_number = list(columns).index(field) + 1
letters = ""
while column_number:
column_number, remainder = divmod(column_number - 1, 26)
letters = chr(65 + remainder) + letters
return f"{letters}{row_number}"
def _canonical_material_rows(
frame: pl.DataFrame,
fields: Sequence[str],
) -> list[dict[str, str]]:
return [
{field: _material_value(row.get(field), field)["value"] for field in fields}
for row in frame.iter_rows(named=True)
]
def _read_material_csv(path: Path, columns: Sequence[str]) -> pl.DataFrame:
frame = pl.read_csv(path, infer_schema=False)
if frame.columns != list(columns):
raise ValueError(f"{path.name} columns do not close to the material contract.")
return frame
def _verify_material_native_outputs(
output_dir: Path,
matches: pl.DataFrame,
residuals: pl.DataFrame,
) -> None:
"""Verify every declared material value in its CSV and XLSX addresses."""
datasets = (
(
"matches",
output_dir / "reconciliation_matches.csv",
matches,
MATCH_COLUMNS,
MATCH_MATERIAL_FIELDS,
),
(
"relationship_residuals",
output_dir / "relationship_residuals.csv",
residuals,
RESIDUAL_COLUMNS,
RESIDUAL_MATERIAL_FIELDS,
),
)
csv_frames: dict[str, pl.DataFrame] = {}
for worksheet, csv_path, prepared, columns, material_fields in datasets:
csv_frame = _read_material_csv(csv_path, columns)
csv_frames[worksheet] = csv_frame
if _canonical_material_rows(csv_frame, material_fields) != (
_canonical_material_rows(prepared, material_fields)
):
raise ValueError(
f"{csv_path.name} material values do not close to prepared {worksheet}."
)
workbook_path = output_dir / "journal_bank_reconciliation.xlsx"
# A duplicate OOXML member makes a declared native address non-unique.
# Rejecting it is a mechanically verifiable package-integrity check.
with ZipFile(workbook_path, "r") as package:
member_names = package.namelist()
if len(member_names) != len(set(member_names)):
raise ValueError("OOXML workbook contains duplicate member names.")
workbook = openpyxl.load_workbook(
workbook_path,
read_only=True,
data_only=False,
)
try:
if workbook.sheetnames != list(WORKBOOK_SHEET_ORDER):
raise ValueError("Reconciliation XLSX worksheet set does not close.")
for worksheet, _, prepared, columns, material_fields in datasets:
sheet = workbook[worksheet]
header_values = next(
sheet.iter_rows(
min_row=1,
max_row=1,
max_col=len(columns),
values_only=True,
),
(),
)
headers = ["" if value is None else str(value) for value in header_values]
if headers != list(columns):
raise ValueError(
f"Reconciliation XLSX headers do not close for {worksheet}."
)
if sheet.max_row != prepared.height + 1 or sheet.max_column != len(columns):
raise ValueError(
f"Reconciliation XLSX dimensions do not close for {worksheet}."
)
expected_rows = _canonical_material_rows(prepared, material_fields)
workbook_rows = sheet.iter_rows(
min_row=2,
max_row=prepared.height + 1,
max_col=len(columns),
values_only=True,
)
field_positions = {
field: list(columns).index(field) for field in material_fields
}
for row_number, (workbook_row, expected_row) in enumerate(
zip(workbook_rows, expected_rows, strict=True),
start=2,
):
for field in material_fields:
actual = _material_value(
workbook_row[field_positions[field]],
field,
)["value"]
if actual != expected_row[field]:
raise ValueError(
"Reconciliation XLSX material value does not close at "
f"{worksheet}!{_excel_cell(columns, field, row_number)}."
)
if (
_canonical_material_rows(
csv_frames[worksheet],
material_fields,
)
!= expected_rows
):
raise ValueError(
f"Reconciliation CSV/XLSX material values diverge for {worksheet}."
)
finally:
workbook.close()
def _build_material_value_ledger(
output_dir: Path,
matches: pl.DataFrame,
residuals: pl.DataFrame,
) -> dict[str, Any]:
"""Build exact all-row prepared-to-CSV/XLSX material-value addresses."""
_verify_material_native_outputs(output_dir, matches, residuals)
entries: list[dict[str, Any]] = []
dataset_specs = (
{
"dataset_id": "matches",
"worksheet": "matches",
"csv_path": "reconciliation_matches.csv",
"prepared_artifact_ref": "prepared.reconciliation_matches",
"csv_artifact_ref": "output.reconciliation_matches_csv",
"frame": matches,
"columns": MATCH_COLUMNS,
"material_fields": MATCH_MATERIAL_FIELDS,
"identity_fields": (
"bank_transaction_id",
"journal_transaction_id",
),
},
{
"dataset_id": "relationship_residuals",
"worksheet": "relationship_residuals",
"csv_path": "relationship_residuals.csv",
"prepared_artifact_ref": "prepared.relationship_residuals",
"csv_artifact_ref": "output.relationship_residuals_csv",
"frame": residuals,
"columns": RESIDUAL_COLUMNS,
"material_fields": RESIDUAL_MATERIAL_FIELDS,
"identity_fields": ("side", "record_ref", "transaction_id"),
},
)
datasets: list[dict[str, Any]] = []
for spec in dataset_specs:
frame = spec["frame"]
material_fields = spec["material_fields"]
datasets.append(
{
"dataset_id": spec["dataset_id"],
"worksheet": spec["worksheet"],
"csv_path": spec["csv_path"],
"columns": list(spec["columns"]),
"material_fields": list(material_fields),
"identity_fields": list(spec["identity_fields"]),
"row_count": frame.height,
"entry_count": frame.height * len(material_fields),
}
)
for prepared_row_number, row in enumerate(
frame.iter_rows(named=True),
start=1,
):
output_row_number = prepared_row_number + 1
identity = {
field: _material_value(row.get(field), field)["value"]
for field in spec["identity_fields"]
}
for field in material_fields:
canonical = _material_value(row.get(field), field)
entries.append(
{
"evidence_id": (
f"{spec['dataset_id']}.{prepared_row_number}.{field}"
),
"dataset_id": spec["dataset_id"],
"prepared_row_number": prepared_row_number,
"field": field,
"value_kind": canonical["kind"],
"canonical_value": canonical["value"],
"row_identity": identity,
"prepared_locator": (
f"row={prepared_row_number};column={field}"
),
"prepared": {
"artifact_ref": spec["prepared_artifact_ref"],
"locator": (f"row={prepared_row_number};column={field}"),
"value": canonical["value"],
},
"outputs": [
{
"artifact_ref": spec["csv_artifact_ref"],
"locator": (f"row={output_row_number};column={field}"),
"value": canonical["value"],
},
{
"artifact_ref": "output.workbook_xlsx",
"locator": (
f"{spec['worksheet']}!"
f"{_excel_cell(spec['columns'], field, output_row_number)}"
),
"value": canonical["value"],
},
],
}
)
content = {
"schema_version": MATERIAL_VALUE_LEDGER_SCHEMA_VERSION,
"ledger_id": "journal_bank.material_values",
"datasets": datasets,
"entry_count": len(entries),
"entries": entries,
}
return {**content, "content_sha256": canonical_json_sha256(content)}
def _current_relationship_policy(output_dir: Path) -> dict[str, Any]:
reviewed = read_json(output_dir / "reviewed_decisions.json")
decisions = reviewed.get("decisions")
if not isinstance(decisions, list):
raise ValueError("Reviewed decisions do not contain a relationship policy.")
relationship_decisions = [
decision
for decision in decisions
if isinstance(decision, dict)
and decision.get("decision_type") == "journal_bank_relationship"
]
if len(relationship_decisions) != 1:
raise ValueError("Exactly one reviewed relationship policy is required.")
content = relationship_decisions[0].get("content")
if not isinstance(content, dict):
raise ValueError("Reviewed relationship policy content is malformed.")
return _normalize_relationship_policy(content.get("policy"))
def validate_material_value_ledger(output_dir: Path) -> dict[str, Any]:
"""Freshly replay match/residual preparation and every native value address."""
unresolved = output_dir.expanduser()
if unresolved.is_symlink():
raise ValueError("Reconciliation output directory cannot be a symlink.")
output_dir = unresolved.resolve()
payload = read_json(output_dir / "material_value_ledger.json")
recorded_digest = payload.pop("content_sha256", None)
if not isinstance(recorded_digest, str) or recorded_digest != canonical_json_sha256(
payload
):
raise ValueError("Material-value ledger content hash is stale.")
if payload.get("schema_version") != MATERIAL_VALUE_LEDGER_SCHEMA_VERSION:
raise ValueError("Unsupported material-value ledger schema.")
bank = _read_material_csv(
output_dir / "normalized_bank.csv",
TRANSACTION_COLUMNS,
)
journal = _read_material_csv(
output_dir / "normalized_journal.csv",
TRANSACTION_COLUMNS,
)
policy = _current_relationship_policy(output_dir)
audit = read_json(output_dir / "reconciliation_audit.json")
tolerance, tolerance_text = _canonical_tolerance(audit.get("tolerance"))
date_window_days = audit.get("date_window_days")
if (
not isinstance(date_window_days, int)
or isinstance(date_window_days, bool)
or date_window_days < 0
or policy["amount_tolerance"] != tolerance_text
or policy["date_window_days"] != date_window_days
):
raise ValueError("Material replay relationship parameters are stale.")
matches, _, _, _ = _match_transactions(
bank,
journal,
tolerance=tolerance,
date_window_days=date_window_days,
relationship_policy=policy,
)
current_matches = _read_material_csv(
output_dir / "reconciliation_matches.csv",
MATCH_COLUMNS,
)
if _canonical_material_rows(
current_matches,
MATCH_MATERIAL_FIELDS,
) != _canonical_material_rows(matches, MATCH_MATERIAL_FIELDS):
raise ValueError("Reconciliation match material values do not replay exactly.")
ledger = validate_allocation_ledger(
read_json(output_dir / "relationship_ledger.json")
)
replayed_ledger = _relationship_ledger(bank, journal, matches, policy=policy)
if ledger != replayed_ledger:
raise ValueError("Relationship ledger does not replay exactly.")
residuals = _relationship_residual_frame(ledger, bank, journal)
expected = _build_material_value_ledger(output_dir, matches, residuals)
actual = {**payload, "content_sha256": recorded_digest}
if actual != expected:
raise ValueError("Material-value ledger does not replay exactly.")
return actual
def _ledger_has_zero_residuals(ledger: dict[str, Any]) -> bool:
return all(
parse_canonical_decimal(str(item["residual"]), label="relationship residual")
== ZERO
for item in (
*ledger["source_residuals"],
*ledger["target_residuals"],
)
)
def _filter_journal_by_sample(
journal: pl.DataFrame, movements: set[str]
) -> pl.DataFrame:
if journal.is_empty():
return journal
if not movements:
return journal.head(0)
if "movement_number" not in journal.columns:
return journal.head(0)
return journal.filter(
pl.col("movement_number").cast(pl.Utf8, strict=False).is_in(sorted(movements))
)
def _stable_ooxml_member(name: str, payload: bytes) -> bytes:
"""Normalize package timestamps that do not describe workbook facts."""
if name == "docProps/core.xml":
return OOXML_CORE_TIMESTAMP_RE.sub(
lambda match: match.group(1) + OOXML_TIMESTAMP + match.group(3),
payload,
)
return payload
def _stabilize_ooxml_package(path: Path) -> None:
"""Canonicalize mechanical XLSX metadata and reject duplicate members."""
original_mode = path.stat().st_mode & 0o7777
with tempfile.NamedTemporaryFile(
prefix=f".{path.name}.",
suffix=".tmp",
dir=path.parent,
delete=False,
) as temporary:
temporary_path = Path(temporary.name)
try:
with ZipFile(path, "r") as source:
members = source.infolist()
names = [member.filename for member in members]
if len(names) != len(set(names)):
raise ValueError("OOXML workbook contains duplicate member names.")
with ZipFile(
temporary_path,
"w",
compression=ZIP_DEFLATED,
compresslevel=9,
) as target:
for member in sorted(members, key=lambda value: value.filename):
info = ZipInfo(member.filename, date_time=OOXML_ZIP_TIMESTAMP)
info.compress_type = ZIP_DEFLATED
info.create_system = 0
info.external_attr = 0
info.flag_bits = 0
target.writestr(
info,
_stable_ooxml_member(
member.filename,
source.read(member),
),
)
temporary_path.chmod(original_mode)
temporary_path.replace(path)
finally:
temporary_path.unlink(missing_ok=True)
def format_workbook_sheet(sheet: Any) -> None:
"""Apply the native readable layout without changing cells or sheet structure."""
sheet.sheet_view.showGridLines = False
sheet.freeze_panes = "B2"
sheet.auto_filter.ref = sheet.dimensions
for column in sheet.columns:
width = min(52, max(16, max(len(str(cell.value or "")) for cell in column) + 2))
sheet.column_dimensions[get_column_letter(column[0].column)].width = width
for row in sheet:
lines = 1
for cell in row:
content_width = int(sheet.column_dimensions[cell.column_letter].width) - 2
text = str(cell.value if cell.value is not None else "")
lines = max(
lines,
max(
(len(line) + content_width - 1) // content_width
for line in text.split("\n")
),
)
cell.font = Font(name="Arial", size=11, bold=cell.row == 1, color="1C2935")
cell.alignment = Alignment(vertical="center", wrap_text=True)
if cell.row == 1:
cell.fill = PatternFill("solid", fgColor="E8EEF5")
sheet.row_dimensions[row[0].row].height = max(24, 16 * lines + 8)
def _write_workbook(path: Path, sheets: dict[str, pl.DataFrame]) -> None:
workbook = openpyxl.Workbook()
default = workbook.active
workbook.remove(default)
for title, frame in sheets.items():
sheet = workbook.create_sheet(title[:31])
sheet.append([excel_safe_value(value) for value in frame.columns])
for row in frame.iter_rows():
sheet.append([excel_safe_value(value) for value in row])
format_workbook_sheet(sheet)
path.parent.mkdir(parents=True, exist_ok=True)
workbook.save(path)
_stabilize_ooxml_package(path)
def _write_review_notes(path: Path, audit: dict[str, Any]) -> None:
words = review_notes_copy(audit.get("language"))
lines = [f"# {words['title']}", ""]
for key in (
"language",
"bank_row_count",
"journal_row_count",
"matched_count",
"unmatched_bank_count",
"unmatched_journal_count",
):
lines.append(f"- {words[key]}: {audit[key]}")
lines.extend(["", f"## {words['stages']}"])
counts = audit.get("stage_counts", {})
if counts:
for stage, count in sorted(counts.items()):
lines.append(f"- {stage}: {count}")
else:
lines.append(f"- {words['none']}")
lines.extend(["", f"## {words['review']}", words["guidance"], ""])
path.write_text("\n".join(lines), encoding="utf-8")
def _write_output_receipts(
output_dir: Path,
paths: dict[str, Path],
*,
source_receipts: Sequence[dict[str, Any]],
) -> dict[str, Any]:
covered = {
"normalized_bank_csv",
"normalized_journal_csv",
"reconciliation_matches_csv",
"relationship_residuals_csv",
"unmatched_bank_csv",
"unmatched_journal_csv",
"bank_pdf_non_movement_rows_csv",
"workbook_xlsx",
"audit_json",
"review_notes_md",
"source_qualifications_json",
"reviewed_decisions_json",
"lineage_json",
"relationship_ledger_json",
"material_value_ledger_json",
"assurance_gates_json",
"review_payload_json",
"ui_decisions_json",
"final_artifacts_json",
"run_intake_json",
"assurance_envelope_json",
}
output_receipts = [
artifact_receipt(
output_dir,
path,
artifact_id=f"output.{key}",
root_id="run",
role="output",
)
for key, path in paths.items()
if key in covered and path.is_file()
]
payload = {
"schema_version": "journal_bank.artifact_receipts.v1",
"source_receipts": list(source_receipts),
"output_receipts": output_receipts,
}
write_assurance_json(paths["artifact_receipts_json"], payload)
return payload
def _final_artifact_closure_is_current(
output_dir: Path,
paths: dict[str, Path],
receipt_bundle: dict[str, Any],
) -> bool:
payload = read_json(paths["final_artifacts_json"])
outputs = payload.get("outputs")
if not isinstance(outputs, list):
return False
declared_paths: list[str] = []
for output in outputs:
if not isinstance(output, dict):
return False
relative = output.get("path")
size_bytes = output.get("size_bytes")
if not isinstance(relative, str) or not isinstance(size_bytes, int):
return False
relative_path = Path(relative)
if (
relative_path.is_absolute()
or relative_path.as_posix() != relative
or ".." in relative_path.parts
):
return False
declared_paths.append(relative)
path = output_dir / relative
if not path.is_file() or path.stat().st_size != size_bytes:
return False
current_paths = {
path.relative_to(output_dir).as_posix()
for path in output_dir.rglob("*")
if path.is_file() and path.name not in FINAL_ARTIFACT_EXCLUDED_NAMES
}
if len(declared_paths) != len(set(declared_paths)):
return False
if set(declared_paths) != current_paths:
return False
final_receipt = next(
(
receipt
for receipt in receipt_bundle.get("output_receipts", [])
if isinstance(receipt, dict)
and receipt.get("path") == paths["final_artifacts_json"].name
),
None,
)
if final_receipt is None:
return False
try:
validate_artifact_receipt(output_dir, final_receipt)
except ValueError:
return False
return True
def _close_final_artifacts(
output_dir: Path,
paths: dict[str, Path],
*,
source_receipts: Sequence[dict[str, Any]],
) -> dict[str, Any]:
"""Reach a bounded size/receipt fixed point after all late output writes."""
for _ in range(FINAL_ARTIFACT_CLOSURE_MAX_PASSES):
refresh_final_artifacts(paths["final_artifacts_json"])
refresh_review_execution_trace(
paths["run_intake_json"],
paths["final_artifacts_json"],
)
receipt_bundle = _write_output_receipts(
output_dir,
paths,
source_receipts=source_receipts,
)
if _final_artifact_closure_is_current(output_dir, paths, receipt_bundle):
return receipt_bundle
raise RuntimeError("Final artifact manifest and receipts did not converge.")
def _artifact_roots(
*,
bank_path: Path,
journal_path: Path,
sample_path: Path | None,
output_dir: Path,
) -> dict[str, Path]:
roots = {
"source_bank": _input_root(bank_path),
"source_journal": _input_root(journal_path),
"run": output_dir,
**implementation_artifact_roots(),
}
if sample_path is not None:
roots["source_sample"] = _input_root(sample_path)
return roots
def implementation_artifact_roots() -> dict[str, Path]:
"""Return the two immutable code roots bound into every assurance envelope."""
if _VERA_ASSURANCE_ROOT is None: # pragma: no cover - import gate above
raise RuntimeError("Vendored vera_assurance implementation is unavailable.")
return {
"implementation": _COMPONENT_ROOT,
"shared_implementation": _VERA_ASSURANCE_ROOT,
}
def _ordinary_single_link_implementation_file(
root: Path,
relative_path: str,
) -> Path:
"""Resolve one implementation leaf without accepting links or aliases."""
relative = Path(relative_path)
if relative.is_absolute() or ".." in relative.parts or "\\" in relative_path:
raise ValueError("implementation receipt path must be canonical")
absolute_root = Path(os.path.abspath(root))
root_stat = absolute_root.lstat()
if stat.S_ISLNK(root_stat.st_mode) or not stat.S_ISDIR(root_stat.st_mode):
raise ValueError("implementation root must be an ordinary directory")
current = absolute_root
for component in relative.parts[:-1]:
current /= component
current_stat = current.lstat()
if stat.S_ISLNK(current_stat.st_mode) or not stat.S_ISDIR(current_stat.st_mode):
raise ValueError(
"implementation receipt path must have ordinary directories"
)
candidate = absolute_root / relative
candidate_stat = candidate.lstat()
if stat.S_ISLNK(candidate_stat.st_mode) or not stat.S_ISREG(candidate_stat.st_mode):
raise ValueError("implementation receipt path must be an ordinary file")
if candidate_stat.st_nlink != 1:
raise ValueError("implementation receipt path cannot have hardlink aliases")
return candidate
def _validate_implementation_spec_coverage(roots: dict[str, Path]) -> None:
"""Keep the declared transitive implementation set complete and closed."""
contract = implementation_contract(str(roots["implementation"]))
specs = (
IMPLEMENTATION_ARTIFACT_SPECS
if contract == IMPLEMENTATION_CONTRACT
else CHATGPT_IMPLEMENTATION_ARTIFACT_SPECS
)
declared_contract = tuple(
(
"plugin" if root_id == "implementation" else "shared_assurance",
relative_path,
)
for _, root_id, relative_path in specs
)
if declared_contract != contract:
raise ValueError("implementation receipt specification is incomplete")
validate_implementation_tree(
str(roots["implementation"]),
shared_assurance_root=str(roots["shared_implementation"]),
)
def build_implementation_artifact_receipts() -> list[dict[str, Any]]:
"""Snapshot the exact ordered transitive implementation set."""
roots = implementation_artifact_roots()
_validate_implementation_spec_coverage(roots)
specs = (
IMPLEMENTATION_ARTIFACT_SPECS
if implementation_contract(str(roots["implementation"]))
== IMPLEMENTATION_CONTRACT
else CHATGPT_IMPLEMENTATION_ARTIFACT_SPECS
)
receipts: list[dict[str, Any]] = []
for artifact_id, root_id, relative_path in specs:
artifact_path = _ordinary_single_link_implementation_file(
roots[root_id],
relative_path,
)
receipts.append(
artifact_receipt(
roots[root_id],
artifact_path,
artifact_id=artifact_id,
root_id=root_id,
role="implementation",
)
)
return receipts
def validate_exact_implementation_receipts(
envelope: dict[str, Any],
*,
artifact_roots: dict[str, Path] | None = None,
) -> list[dict[str, Any]]:
"""Replay the exact ordered code receipts against ordinary current bytes."""
expected = build_implementation_artifact_receipts()
expected_refs = [receipt["artifact_id"] for receipt in expected]
raw_refs = envelope.get("implementation_artifact_refs")
raw_artifacts = envelope.get("artifact_receipts")
if not isinstance(raw_refs, list) or raw_refs != expected_refs:
raise ValueError("assurance implementation receipt set or order is invalid")
if not isinstance(raw_artifacts, list):
raise ValueError("assurance implementation receipts are unavailable")
implementation_receipts = [
receipt
for receipt in raw_artifacts
if isinstance(receipt, dict) and receipt.get("role") == "implementation"
]
if implementation_receipts != expected:
raise ValueError("assurance implementation receipts are not exact")
roots = artifact_roots or implementation_artifact_roots()
for receipt in implementation_receipts:
validate_artifact_receipt(roots, receipt)
return implementation_receipts
def _write_assurance_envelope(
*,
run_id: str,
bank_path: Path,
journal_path: Path,
sample_path: Path | None,
output_dir: Path,
path: Path,
receipt_bundle: dict[str, Any],
reviewed_decisions: Sequence[dict[str, Any]],
qualifications: Sequence[dict[str, Any]],
gates: dict[str, Any],
limitations: Sequence[str],
) -> dict[str, Any]:
implementation_receipts = build_implementation_artifact_receipts()
mutable_control_paths = {
"artifact_receipts.json",
"run_intake.json",
"ui_decisions.json",
"applied_decisions.json",
"final_artifacts.json",
}
output_receipts = [
receipt
for receipt in receipt_bundle["output_receipts"]
if str(receipt["path"]) not in mutable_control_paths
]
roots = _artifact_roots(
bank_path=bank_path,
journal_path=journal_path,
sample_path=sample_path,
output_dir=output_dir,
)
envelope = build_assurance_envelope(
run_id=run_id,
workflow_id=ASSURANCE_WORKFLOW_ID,
workflow_version=ASSURANCE_WORKFLOW_VERSION,
artifact_receipts=[
*receipt_bundle["source_receipts"],
*implementation_receipts,
*output_receipts,
],
implementation_artifact_refs=[
receipt["artifact_id"] for receipt in implementation_receipts
],
reviewed_decisions=reviewed_decisions,
source_qualifications=qualifications,
allocation_ledgers=[],
numeric_evidence_ledgers=[],
gate_register=gates,
limitations=list(limitations),
artifact_roots=roots,
)
validate_exact_implementation_receipts(envelope, artifact_roots=roots)
validate_assurance_envelope(envelope, artifact_roots=roots)
write_assurance_json(path, envelope)
return envelope
def run_reconciliation(
bank_path: Path,
journal_path: Path,
output_dir: Path,
recipe_path: Path | None = None,
*,
sample_path: Path | None = None,
tolerance: object = "1",
date_window_days: int = 7,
language: object | None = None,
document_language: object | None = None,
client_run_id: str | None = None,
client_run_root: Path | None = None,
) -> ReconciliationRunResult:
"""Run exact, policy-bound journal-to-bank reconciliation."""
recipe = read_json(recipe_path)
tolerance_value, tolerance_text = _canonical_tolerance(tolerance)
if (
not isinstance(date_window_days, int)
or isinstance(date_window_days, bool)
or date_window_days < 0
):
raise ValueError("date_window_days must be a non-negative integer")
languages = language_assumptions(
recipe, language=language, document_language=document_language
)
output_dir.mkdir(parents=True, exist_ok=True)
managed_run_root = (
client_run_root.expanduser().resolve() if client_run_root is not None else None
)
def output_reference(path: Path) -> str:
if managed_run_root is None:
return path.as_posix()
try:
relative = path.expanduser().resolve().relative_to(managed_run_root)
except ValueError as exc:
raise ValueError(
"Journal-Bank output leaves the managed customer run."
) from exc
if not relative.parts:
raise ValueError("Journal-Bank output must identify a run artifact.")
return relative.as_posix()
run_intake = write_run_intake(
output_dir,
bank_path=bank_path,
journal_path=journal_path,
recipe_path=recipe_path,
sample_path=sample_path,
language=languages["language"],
document_language=languages["document_language"],
tolerance=tolerance_text,
date_window_days=date_window_days,
client_run_id=client_run_id,
client_run_root=client_run_root,
)
recorded_intake = read_json(run_intake.path)
recorded_assumptions = recorded_intake.get("assumptions")
if not isinstance(recorded_assumptions, dict):
raise ValueError("Journal-Bank run intake assumptions are unavailable")
recorded_bank_path = str(recorded_assumptions["bank_path"])
recorded_journal_path = str(recorded_assumptions["journal_path"])
recorded_sample_path = recorded_assumptions.get("sample_path")
initial_source_receipts, source_refs = _source_artifact_receipts(
(("bank", bank_path), ("journal", journal_path), ("sample", sample_path))
)
bank, bank_diag = _normalize_files(
bank_path, "bank", recipe, source_refs=source_refs
)
journal, journal_diag = _normalize_files(
journal_path, "journal", recipe, source_refs=source_refs
)
sample_movements, sample_diagnostics = _read_sample_movements(sample_path)
source_receipts, current_source_refs = _source_artifact_receipts(
(("bank", bank_path), ("journal", journal_path), ("sample", sample_path))
)
changed_sources = _source_snapshot_changes(
initial_source_receipts,
source_receipts,
)
source_refs = current_source_refs
if changed_sources:
bank_diag = _mark_changed_source_diagnostics("bank", bank_diag, changed_sources)
journal_diag = _mark_changed_source_diagnostics(
"journal", journal_diag, changed_sources
)
sample_diagnostics = _mark_changed_sample_diagnostics(
sample_diagnostics,
changed_sources,
)
if any(root_id == "source_bank" for root_id, _ in changed_sources):
bank = _transaction_frame([])
if any(root_id == "source_journal" for root_id, _ in changed_sources):
journal = _transaction_frame([])
if any(root_id == "source_sample" for root_id, _ in changed_sources):
sample_movements = set()
bank_pdf_non_movements = _non_movement_frame(_non_movement_records(bank_diag))
qualifications = [
*_source_qualifications("bank", bank_diag, source_refs),
*_source_qualifications("journal", journal_diag, source_refs),
]
qualification_status = _aggregate_qualification_status(qualifications)
if changed_sources:
qualification_status = "unsupported_source_layout"
qualification_refs = [str(item["qualification_id"]) for item in qualifications]
bank_journal_source_refs = [
str(receipt["artifact_id"])
for receipt in source_receipts
if str(receipt["artifact_id"]).startswith(("source.bank.", "source.journal."))
]
relationship_policy, relationship_decision, relationship_error = (
_validated_relationship_decision(
recipe,
source_artifact_refs=bank_journal_source_refs,
tolerance_text=tolerance_text,
date_window_days=date_window_days,
)
)
reviewed_decisions = _mapping_decisions([*bank_diag, *journal_diag])
if relationship_decision is not None:
reviewed_decisions.append(relationship_decision)
paths = {
"normalized_bank_csv": output_dir / "normalized_bank.csv",
"normalized_journal_csv": output_dir / "normalized_journal.csv",
"reconciliation_matches_csv": output_dir / "reconciliation_matches.csv",
"relationship_residuals_csv": output_dir / "relationship_residuals.csv",
"unmatched_bank_csv": output_dir / "unmatched_bank.csv",
"unmatched_journal_csv": output_dir / "unmatched_journal.csv",
"bank_pdf_non_movement_rows_csv": output_dir / "bank_pdf_non_movement_rows.csv",
"workbook_xlsx": output_dir / "journal_bank_reconciliation.xlsx",
"audit_json": output_dir / "reconciliation_audit.json",
"review_notes_md": output_dir / "review_notes.md",
"input_receipts_json": output_dir / "input_receipts.json",
"source_qualifications_json": output_dir / "source_qualifications.json",
"reviewed_decisions_json": output_dir / "reviewed_decisions.json",
"lineage_json": output_dir / "lineage.json",
"relationship_ledger_json": output_dir / "relationship_ledger.json",
"material_value_ledger_json": output_dir / "material_value_ledger.json",
"assurance_gates_json": output_dir / "assurance_gates.json",
"artifact_receipts_json": output_dir / "artifact_receipts.json",
"assurance_envelope_json": output_dir / "assurance_envelope.json",
"run_intake_json": output_dir / "run_intake.json",
"review_payload_json": output_dir / "review_payload.json",
"ui_decisions_json": output_dir / "ui_decisions.json",
"final_artifacts_json": output_dir / "final_artifacts.json",
}
write_assurance_json(
paths["input_receipts_json"],
{
"schema_version": "journal_bank.input_receipts.v1",
"receipts": source_receipts,
},
)
write_assurance_json(
paths["source_qualifications_json"],
{
"schema_version": "journal_bank.source_qualifications.v1",
"status": qualification_status,
"source_outcomes": _source_outcomes(bank_diag, journal_diag),
"qualifications": qualifications,
},
)
write_assurance_json(
paths["reviewed_decisions_json"],
{
"schema_version": "journal_bank.reviewed_decisions.v1",
"decisions": reviewed_decisions,
},
)
bank_pdf_non_movements.write_csv(paths["bank_pdf_non_movement_rows_csv"])
blocked_code: str | None = None
blocked_detail: str | None = None
source_gate_status = "passed"
preparation_gate_status = "passed"
if changed_sources:
blocked_code = "source_changed_during_run"
blocked_detail = (
"Source bytes or source membership changed between the pre-parser and "
"post-parser snapshots; all prepared rows from affected inputs were "
"discarded."
)
source_gate_status = "failed"
preparation_gate_status = "blocked"
elif qualification_status == "unsupported_source_layout":
parser_failed = any(
item.get("failure_kind") == "parser_failure"
for item in (*bank_diag, *journal_diag)
)
blocked_code = (
"parser_failure" if parser_failed else "unsupported_source_layout"
)
blocked_detail = (
"At least one source could not be parsed."
if parser_failed
else "At least one source lacks a supported bounded layout adapter."
)
source_gate_status = "failed"
preparation_gate_status = "blocked"
elif qualification_status == "needs_review":
blocked_code = "mapping_review_required"
blocked_detail = "At least one source requires a current reviewed mapping."
source_gate_status = "blocked"
preparation_gate_status = "blocked"
elif sample_path is not None and not sample_movements:
blocked_code = "invalid_or_empty_sample"
blocked_detail = (
"A sample was supplied but its exact identifier adapter did not yield "
"a complete non-empty movement set."
)
preparation_gate_status = "blocked"
elif bank.is_empty() or journal.is_empty():
blocked_code = "empty_prepared_population"
blocked_detail = (
"A qualified source emitted no bank or journal monetary movements."
)
preparation_gate_status = "failed"
elif relationship_error is not None:
blocked_code = "relationship_review_required"
blocked_detail = relationship_error
filtered_journal = journal
if blocked_code is None and sample_path is not None:
filtered_journal = _filter_journal_by_sample(journal, sample_movements)
if filtered_journal.is_empty():
blocked_code = "sample_not_found_in_journal"
blocked_detail = (
"The supplied sample identifiers do not select any journal movement."
)
preparation_gate_status = "failed"
if blocked_code is None and relationship_policy is not None:
bank = _with_policy_defaults(bank, relationship_policy)
filtered_journal = _with_policy_defaults(filtered_journal, relationship_policy)
perimeter_error = _relationship_perimeter_error(
bank, filtered_journal, relationship_policy
)
if perimeter_error is not None:
blocked_code = "relationship_perimeter_incomplete"
blocked_detail = perimeter_error
write_assurance_json(
paths["lineage_json"],
_lineage_payload(bank, filtered_journal, source_refs),
)
if blocked_code is not None and blocked_detail is not None:
gates = _gate_payload(
source_status=source_gate_status,
source_refs=qualification_refs,
preparation_status=preparation_gate_status,
reconciliation_status="blocked",
limitation=blocked_detail,
)
write_assurance_json(paths["assurance_gates_json"], gates)
blocked_matches = pl.DataFrame(
schema={column: pl.Utf8 for column in MATCH_COLUMNS}
)
blocked_residuals = pl.DataFrame(
schema={column: pl.Utf8 for column in RESIDUAL_COLUMNS}
)
unmatched_bank = bank
unmatched_journal = filtered_journal
bank.write_csv(paths["normalized_bank_csv"])
filtered_journal.write_csv(paths["normalized_journal_csv"])
blocked_matches.write_csv(paths["reconciliation_matches_csv"])
blocked_residuals.write_csv(paths["relationship_residuals_csv"])
unmatched_bank.write_csv(paths["unmatched_bank_csv"])
unmatched_journal.write_csv(paths["unmatched_journal_csv"])
_write_workbook(
paths["workbook_xlsx"],
{
"matches": blocked_matches,
"relationship_residuals": blocked_residuals,
"unmatched_bank": unmatched_bank,
"unmatched_journal": unmatched_journal,
"bank_pdf_non_movements": bank_pdf_non_movements,
"normalized_bank": bank,
"normalized_journal": filtered_journal,
},
)
write_assurance_json(
paths["relationship_ledger_json"],
_blocked_relationship_ledger(
block_code=blocked_code,
block_detail=blocked_detail,
),
)
blocked_audit = {
**languages,
"schema_version": "journal_bank.reconciliation_audit.v3",
"status": "blocked",
"block_code": blocked_code,
"block_detail": blocked_detail,
"bank_path": recorded_bank_path,
"journal_path": recorded_journal_path,
"sample_path": recorded_sample_path,
"sample_movement_count": len(sample_movements),
"sample_diagnostics": sample_diagnostics,
"source_snapshot_changed": bool(changed_sources),
"changed_sources": [
{"root_id": root_id, "path": source_path}
for root_id, source_path in sorted(changed_sources)
],
"bank_row_count": bank.height,
"journal_row_count": filtered_journal.height,
"matched_count": 0,
"unmatched_bank_count": unmatched_bank.height,
"unmatched_journal_count": unmatched_journal.height,
"bank_pdf_non_movement_row_count": bank_pdf_non_movements.height,
"bank_pdf_non_movement_classifications": _count_classifications(
bank_pdf_non_movements.to_dicts()
),
"stage_counts": {},
"tolerance": tolerance_text,
"date_window_days": date_window_days,
"source_qualification_status": qualification_status,
"relationship_decision_ref": (
str(relationship_decision["decision_id"])
if relationship_decision is not None
else None
),
"relationship_balanced": False,
"relationship_within_policy_tolerance": False,
"relationship_residual_row_count": 0,
"diagnostics": {"bank": bank_diag, "journal": journal_diag},
"outputs": {
key: output_reference(value)
for key, value in paths.items()
if key != "material_value_ledger_json"
},
}
write_assurance_json(paths["audit_json"], blocked_audit)
_write_review_notes(paths["review_notes_md"], blocked_audit)
_write_output_receipts(
output_dir,
paths,
source_receipts=source_receipts,
)
review_session = write_review_session_artifacts(
output_dir,
run_id=run_intake.run_id,
run_intake_path=run_intake.path,
matches=blocked_matches,
unmatched_bank=unmatched_bank,
unmatched_journal=unmatched_journal,
bank_pdf_non_movements=bank_pdf_non_movements,
relationship_residuals=blocked_residuals,
audit=blocked_audit,
)
blocked_audit["review_session"] = {
"run_id": review_session.run_id,
"run_intake_path": output_reference(review_session.run_intake_path),
"review_payload_path": output_reference(review_session.review_payload_path),
"ui_decisions_path": output_reference(review_session.ui_decisions_path),
"final_artifacts_path": output_reference(
review_session.final_artifacts_path
),
"review_item_count": review_session.review_item_count,
}
write_assurance_json(paths["audit_json"], blocked_audit)
receipt_bundle = _write_output_receipts(
output_dir,
paths,
source_receipts=source_receipts,
)
_write_assurance_envelope(
run_id=run_intake.run_id,
bank_path=bank_path,
journal_path=journal_path,
sample_path=sample_path,
output_dir=output_dir,
path=paths["assurance_envelope_json"],
receipt_bundle=receipt_bundle,
reviewed_decisions=reviewed_decisions,
qualifications=qualifications,
gates=gates,
limitations=[blocked_detail],
)
_close_final_artifacts(
output_dir,
paths,
source_receipts=source_receipts,
)
raise ReconciliationBlockedError(blocked_code, blocked_detail)
if relationship_policy is None: # pragma: no cover - guarded by block handling
raise RuntimeError("Validated relationship policy unexpectedly unavailable.")
matches, unmatched_bank, unmatched_journal, stage_counts = _match_transactions(
bank,
filtered_journal,
tolerance=tolerance_value,
date_window_days=date_window_days,
relationship_policy=relationship_policy,
)
relationship_ledger = _relationship_ledger(
bank,
filtered_journal,
matches,
policy=relationship_policy,
)
write_assurance_json(paths["relationship_ledger_json"], relationship_ledger)
relationship_residuals = _relationship_residual_frame(
relationship_ledger,
bank,
filtered_journal,
)
relationship_exactly_balanced = _ledger_has_zero_residuals(relationship_ledger)
reconciliation_status = (
"passed"
if relationship_ledger["balanced"]
and relationship_exactly_balanced
and unmatched_bank.is_empty()
and unmatched_journal.is_empty()
else "withheld"
)
unresolved_detail = (
None
if reconciliation_status == "passed"
else "Unmatched rows or exact residuals remain unresolved."
)
gates = _gate_payload(
source_status="passed",
source_refs=qualification_refs,
preparation_status="passed",
reconciliation_status=reconciliation_status,
reconciliation_ref="output.relationship_ledger_json",
limitation=unresolved_detail,
)
write_assurance_json(paths["assurance_gates_json"], gates)
bank.write_csv(paths["normalized_bank_csv"])
filtered_journal.write_csv(paths["normalized_journal_csv"])
matches.write_csv(paths["reconciliation_matches_csv"])
relationship_residuals.write_csv(paths["relationship_residuals_csv"])
unmatched_bank.write_csv(paths["unmatched_bank_csv"])
unmatched_journal.write_csv(paths["unmatched_journal_csv"])
workbook_frames = {
"matches": matches,
"relationship_residuals": relationship_residuals,
"unmatched_bank": unmatched_bank,
"unmatched_journal": unmatched_journal,
"bank_pdf_non_movements": bank_pdf_non_movements,
"normalized_bank": bank,
"normalized_journal": filtered_journal,
}
_write_workbook(
paths["workbook_xlsx"],
{name: workbook_frames[name] for name in WORKBOOK_SHEET_ORDER},
)
audit = {
**languages,
"schema_version": "journal_bank.reconciliation_audit.v3",
"status": (
"completed_pending_review"
if reconciliation_status == "passed"
else "completed_with_unresolved_reconciliation"
),
"bank_path": recorded_bank_path,
"journal_path": recorded_journal_path,
"sample_path": recorded_sample_path,
"sample_movement_count": len(sample_movements),
"sample_diagnostics": sample_diagnostics,
"source_snapshot_changed": bool(changed_sources),
"changed_sources": [
{"root_id": root_id, "path": source_path}
for root_id, source_path in sorted(changed_sources)
],
"bank_row_count": bank.height,
"journal_row_count": filtered_journal.height,
"matched_count": matches.height,
"unmatched_bank_count": unmatched_bank.height,
"unmatched_journal_count": unmatched_journal.height,
"bank_pdf_non_movement_row_count": bank_pdf_non_movements.height,
"bank_pdf_non_movement_classifications": _count_classifications(
bank_pdf_non_movements.to_dicts()
),
"stage_counts": stage_counts,
"tolerance": tolerance_text,
"date_window_days": date_window_days,
"source_qualification_status": qualification_status,
"relationship_decision_ref": str(relationship_decision["decision_id"]),
"relationship_balanced": relationship_exactly_balanced,
"relationship_within_policy_tolerance": bool(relationship_ledger["balanced"]),
"relationship_residual_row_count": relationship_residuals.height,
"diagnostics": {"bank": bank_diag, "journal": journal_diag},
"outputs": {key: output_reference(value) for key, value in paths.items()},
}
write_json(paths["audit_json"], audit)
_write_review_notes(paths["review_notes_md"], audit)
material_value_ledger = _build_material_value_ledger(
output_dir,
matches,
relationship_residuals,
)
write_json(paths["material_value_ledger_json"], material_value_ledger)
validate_material_value_ledger(output_dir)
_write_output_receipts(
output_dir,
paths,
source_receipts=source_receipts,
)
review_session = write_review_session_artifacts(
output_dir,
run_id=run_intake.run_id,
run_intake_path=run_intake.path,
matches=matches,
unmatched_bank=unmatched_bank,
unmatched_journal=unmatched_journal,
bank_pdf_non_movements=bank_pdf_non_movements,
relationship_residuals=relationship_residuals,
audit=audit,
)
audit["review_session"] = {
"run_id": review_session.run_id,
"run_intake_path": output_reference(review_session.run_intake_path),
"review_payload_path": output_reference(review_session.review_payload_path),
"ui_decisions_path": output_reference(review_session.ui_decisions_path),
"final_artifacts_path": output_reference(review_session.final_artifacts_path),
"review_item_count": review_session.review_item_count,
}
write_json(paths["audit_json"], audit)
receipt_bundle = _write_output_receipts(
output_dir,
paths,
source_receipts=source_receipts,
)
_write_assurance_envelope(
run_id=run_intake.run_id,
bank_path=bank_path,
journal_path=journal_path,
sample_path=sample_path,
output_dir=output_dir,
path=paths["assurance_envelope_json"],
receipt_bundle=receipt_bundle,
reviewed_decisions=reviewed_decisions,
qualifications=qualifications,
gates=gates,
limitations=(
[]
if reconciliation_status == "passed"
else [unresolved_detail or "Reconciliation assurance remains withheld."]
),
)
_close_final_artifacts(
output_dir,
paths,
source_receipts=source_receipts,
)
return ReconciliationRunResult(
matches=matches,
unmatched_bank=unmatched_bank,
unmatched_journal=unmatched_journal,
audit=audit,
)
def add_common_args(parser: argparse.ArgumentParser) -> None:
parser.add_argument(
"--language",
help="Working/output language locale: it, en, fr, de, or es. Defaults to recipe or en.",
)
parser.add_argument(
"--document-language",
help="Source-document language locale: it, en, fr, de, es, or auto. Defaults to recipe or auto.",
)
parser.add_argument("--verbose", action="store_true", help="Enable debug logging.")
SHA-256: 363f569a3c27cf2b7bc01cf7d7b9a2cfac8c0b4d852dad48e9577c5852ad09b2