← Files ClaraARCHIVED FILE
scripts/parse_commercial_general_journal.py
103 KB · Oct 2, 2026 · 00:29 UTC
#!/usr/bin/env python3
"""Parse one explicitly reviewed paginated commercial general journal.
This parser is deterministic because byte binding, OpenXML extraction, fixed
layout matching, exact arithmetic, line-ID uniqueness, date ordering, and
control-total reconciliation are mechanically verifiable. It does not decide
that a workbook is a general journal, choose an entity, currency, unit, account
mapping, sign convention, or accounting interpretation. The caller must review
and supply those decisions outside this parser.
The parser writes nothing. Raw descriptions and entity labels are deliberately
excluded from the returned in-memory model.
"""
from __future__ import annotations
import hashlib
import io
import json
import os
import posixpath
import re
import stat
import zipfile
from dataclasses import dataclass
from datetime import date, datetime
from decimal import (
Decimal,
DecimalException,
Inexact,
InvalidOperation,
Rounded,
localcontext,
)
from pathlib import Path
from types import MappingProxyType
from typing import Any, Mapping, NoReturn, Pattern
# Every XML payload is bounded and screened by _safe_xml_member_bytes first.
from xml.etree import ElementTree # nosec B405
__all__ = [
"DatePattern",
"GeneralJournalLayoutContract",
"GeneralJournalParseError",
"JournalMovement",
"LogicalMovementPattern",
"PageLayout",
"ParsedGeneralJournal",
"PhysicalEmbeddedAmountPattern",
"ReviewedAmountLocator",
"ReviewedAmountlessExclusion",
"ReviewedAmountPair",
"ReviewedAmountPairMember",
"general_journal_layout_contract_from_mapping",
"load_general_journal_layout_contract",
"parse_commercial_general_journal",
]
CONTRACT_VERSION = "clara.commercial_general_journal_layout.v4"
REVIEW_STATUS = "reviewed"
SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$")
ACCOUNTING_AMOUNT_FORMATS = frozenset(
{
"canonical_dot",
"italian_grouped_2",
}
)
AMOUNT_SIGN_POLICIES = frozenset({"nonnegative", "signed"})
MAX_SOURCE_BYTES = 512 * 1024 * 1024
MAX_CONTRACT_BYTES = 1024 * 1024
MAX_PATTERN_LENGTH = 2_048
MAX_CELL_TEXT_LENGTH = 1_000_000
MAX_XLSX_MEMBERS = 2_048
MAX_XLSX_MEMBER_UNCOMPRESSED_BYTES = 128 * 1024 * 1024
MAX_XLSX_TOTAL_UNCOMPRESSED_BYTES = 256 * 1024 * 1024
MAX_XLSX_COMPRESSION_RATIO = 200
MAX_WORKSHEET_ROWS = 1_000_000
MAX_WORKSHEET_CELLS = 10_000_000
MAX_ROW_CELLS = 20_000
MAX_SHARED_STRINGS = 2_000_000
MAX_SHARED_STRING_CHARACTERS = 128 * 1024 * 1024
MAX_XLSX_COLUMN = 16_384
_MAIN_NS = "http://schemas.openxmlformats.org/spreadsheetml/2006/main"
_DOCUMENT_REL_NS = "http://schemas.openxmlformats.org/officeDocument/2006/relationships"
_PACKAGE_REL_NS = "http://schemas.openxmlformats.org/package/2006/relationships"
_RELATIONSHIP_ID = f"{{{_DOCUMENT_REL_NS}}}id"
_CELL_REFERENCE = re.compile(r"^([A-Z]+)([1-9][0-9]*)$")
_CANONICAL_DOT_DECIMAL = re.compile(r"^[+-]?(?:0|[1-9][0-9]*)(?:\.[0-9]+)?$")
_ITALIAN_GROUPED_2_DECIMAL = re.compile(
r"^[+-]?(?:0|[1-9][0-9]{0,2}(?:\.[0-9]{3})*),[0-9]{2}$"
)
_XML_DECLARATION = re.compile(
rb"\A(?:\xef\xbb\xbf)?[ \t\r\n]*<\?xml\b.*?\?>",
re.IGNORECASE | re.DOTALL,
)
_XML_ENCODING = re.compile(
rb"\bencoding\s*=\s*(['\"])(?P<encoding>[A-Za-z0-9._-]+)\1",
re.IGNORECASE,
)
_SUPPORTED_XML_ENCODINGS = frozenset({"ascii", "us-ascii", "utf-8"})
_CONTRACT_FIELDS = frozenset(
{
"contract_version",
"review_status",
"sheet_name",
"date_header_label",
"line_header_label",
"account_header_label",
"debit_header_label",
"credit_header_label",
"page_layouts",
"date_patterns",
"account_code_pattern",
"logical_candidate_pattern",
"logical_movement_patterns",
"physical_embedded_amount_patterns",
"reviewed_amount_pairs",
"reviewed_amountless_exclusions",
"reviewed_zero_amount_line_ids",
"physical_amount_format",
"amount_sign_policy",
"control_pattern",
"control_amount_format",
"reviewed_final_debit_total",
"reviewed_final_credit_total",
}
)
_PAGE_LAYOUT_FIELDS = frozenset(
{
"layout_id",
"date_header_column",
"line_header_column",
"account_header_column",
"debit_header_column",
"credit_header_column",
"date_columns",
"line_id_columns",
"account_columns",
"debit_amount_columns",
"credit_amount_columns",
"physical_first_line_columns",
}
)
_DATE_PATTERN_FIELDS = frozenset({"pattern", "strptime_format"})
_LOGICAL_PATTERN_FIELDS = frozenset({"layout_ids", "pattern", "amount_format"})
_PHYSICAL_EMBEDDED_AMOUNT_PATTERN_FIELDS = frozenset(
{"layout_ids", "column", "pattern", "amount_format"}
)
_REVIEWED_AMOUNT_LOCATOR_FIELDS = frozenset(
{
"layout_id",
"row_number",
"column",
"line_index",
"pattern",
"amount_format",
}
)
_REVIEWED_AMOUNT_PAIR_MEMBER_FIELDS = frozenset(
{
"movement_layout_id",
"movement_row_number",
"movement_line_id",
"amount_locator",
}
)
_REVIEWED_AMOUNT_PAIR_FIELDS = frozenset({"debit", "credit"})
_REVIEWED_AMOUNTLESS_EXCLUSION_FIELDS = frozenset(
{
"layout_id",
"row_number",
"line_id",
"nonempty_columns",
"residual_columns",
}
)
class GeneralJournalParseError(ValueError):
"""Raised when an exact parser or reviewed-contract invariant fails."""
@dataclass(frozen=True)
class DatePattern:
"""One caller-reviewed registration-date syntax."""
pattern: str
strptime_format: str
@dataclass(frozen=True)
class LogicalMovementPattern:
"""One caller-reviewed fixed-width logical movement-line syntax."""
layout_ids: tuple[str, ...]
pattern: str
amount_format: str
@dataclass(frozen=True)
class PhysicalEmbeddedAmountPattern:
"""One reviewed amount embedded in a physical text cell."""
layout_ids: tuple[str, ...]
column: int
pattern: str
amount_format: str
@dataclass(frozen=True)
class ReviewedAmountLocator:
"""One exact source line containing a reviewer-owned pair amount."""
layout_id: str
row_number: int
column: int
line_index: int
pattern: str
amount_format: str
@dataclass(frozen=True)
class ReviewedAmountPairMember:
"""One exact movement row bound to one exact amount locator."""
movement_layout_id: str
movement_row_number: int
movement_line_id: int
amount_locator: ReviewedAmountLocator
@dataclass(frozen=True)
class ReviewedAmountPair:
"""One reviewer-owned equal debit/credit pair."""
debit: ReviewedAmountPairMember
credit: ReviewedAmountPairMember
@dataclass(frozen=True)
class ReviewedAmountlessExclusion:
"""One exact row with no reviewed physical amount signal."""
layout_id: str
row_number: int
line_id: int
nonempty_columns: tuple[int, ...]
residual_columns: tuple[int, ...]
@dataclass(frozen=True)
class PageLayout:
"""One exact page-header signature and its physical extraction columns."""
layout_id: str
date_header_column: int
line_header_column: int
account_header_column: int
debit_header_column: int
credit_header_column: int
date_columns: tuple[int, ...]
line_id_columns: tuple[int, ...]
account_columns: tuple[int, ...]
debit_amount_columns: tuple[int, ...]
credit_amount_columns: tuple[int, ...]
physical_first_line_columns: tuple[int, ...]
@dataclass(frozen=True)
class GeneralJournalLayoutContract:
"""Caller-reviewed mechanics; no semantic source decisions are inferred."""
contract_version: str
review_status: str
sheet_name: str
date_header_label: str
line_header_label: str
account_header_label: str
debit_header_label: str
credit_header_label: str
page_layouts: tuple[PageLayout, ...]
date_patterns: tuple[DatePattern, ...]
account_code_pattern: str
logical_candidate_pattern: str
logical_movement_patterns: tuple[LogicalMovementPattern, ...]
physical_embedded_amount_patterns: tuple[
PhysicalEmbeddedAmountPattern,
...,
]
reviewed_amount_pairs: tuple[ReviewedAmountPair, ...]
reviewed_amountless_exclusions: tuple[
ReviewedAmountlessExclusion,
...,
]
reviewed_zero_amount_line_ids: tuple[int, ...]
physical_amount_format: str
amount_sign_policy: str
control_pattern: str
control_amount_format: str
reviewed_final_debit_total: str
reviewed_final_credit_total: str
@dataclass(frozen=True)
class JournalMovement:
"""One mechanically parsed movement without descriptions or entity data."""
line_id: int
posting_date: date
account_code: str
debit: Decimal
credit: Decimal
source_form: str
@dataclass(frozen=True)
class ParsedGeneralJournal:
"""Deterministic numeric model; row-level completeness is not established."""
source_sha256: str
movements: tuple[JournalMovement, ...]
debit_total: Decimal
credit_total: Decimal
source_control_debit_total: Decimal
source_control_credit_total: Decimal
first_posting_date: date
last_posting_date: date
page_header_count: int
physical_movement_count: int
logical_movement_count: int
excluded_amountless_count: int
line_id_gap_count: int
layout_page_counts: Mapping[str, int]
def sanitized_counts(self) -> dict[str, int | str]:
"""Return aggregate mechanics only; no row-level or identifying values."""
return {
"movement_count": len(self.movements),
"physical_movement_count": self.physical_movement_count,
"logical_movement_count": self.logical_movement_count,
"excluded_amountless_count": self.excluded_amountless_count,
"page_header_count": self.page_header_count,
"line_id_gap_count": self.line_id_gap_count,
"layout_variant_count": len(self.layout_page_counts),
"first_posting_date": self.first_posting_date.isoformat(),
"last_posting_date": self.last_posting_date.isoformat(),
}
@dataclass(frozen=True)
class _SheetRow:
row_number: int
cells: Mapping[int, str]
@dataclass(frozen=True)
class _CompiledDatePattern:
pattern: Pattern[str]
strptime_format: str
@dataclass(frozen=True)
class _CompiledLogicalMovementPattern:
layout_ids: frozenset[str]
pattern: Pattern[str]
amount_format: str
@dataclass(frozen=True)
class _CompiledPhysicalEmbeddedAmountPattern:
layout_ids: frozenset[str]
column: int
pattern: Pattern[str]
amount_format: str
@dataclass(frozen=True)
class _CompiledReviewedAmountLocator:
value: ReviewedAmountLocator
pattern: Pattern[str]
@dataclass(frozen=True)
class _CompiledReviewedAmountPairMember:
value: ReviewedAmountPairMember
role: str
amount_locator: _CompiledReviewedAmountLocator
@dataclass(frozen=True)
class _CompiledReviewedAmountPair:
debit: _CompiledReviewedAmountPairMember
credit: _CompiledReviewedAmountPairMember
@dataclass(frozen=True)
class _CompiledContract:
value: GeneralJournalLayoutContract
account_code_pattern: Pattern[str]
logical_candidate_pattern: Pattern[str]
logical_movement_patterns: tuple[_CompiledLogicalMovementPattern, ...]
physical_embedded_amount_patterns: tuple[
_CompiledPhysicalEmbeddedAmountPattern,
...,
]
reviewed_amount_pairs: tuple[_CompiledReviewedAmountPair, ...]
reviewed_amountless_exclusions: tuple[
ReviewedAmountlessExclusion,
...,
]
reviewed_zero_amount_line_ids: frozenset[int]
date_patterns: tuple[_CompiledDatePattern, ...]
control_pattern: Pattern[str]
reviewed_debit_total: Decimal
reviewed_credit_total: Decimal
@dataclass(frozen=True)
class _LogicalLine:
row_number: int
column: int
line_index: int
text: str
@dataclass(frozen=True)
class _ResolvedReviewedAmountPairMember:
value: ReviewedAmountPairMember
role: str
amount: Decimal
@dataclass(frozen=True)
class _ResolvedReviewedAmountPairs:
members_by_row: Mapping[int, _ResolvedReviewedAmountPairMember]
movement_rows_by_line_id: Mapping[int, int]
locator_keys: frozenset[tuple[int, int, int]]
@dataclass(frozen=True)
class _ResolvedReviewedAmountlessExclusions:
by_row: Mapping[int, ReviewedAmountlessExclusion]
rows_by_line_id: Mapping[int, int]
def _fail(message: str) -> NoReturn:
raise GeneralJournalParseError(message)
def _canonical_text(value: object, *, label: str) -> str:
if not isinstance(value, str) or not value or value != value.strip():
_fail(f"{label} must be canonical non-empty text")
return value
def _mapping(value: object, *, label: str) -> Mapping[str, object]:
if not isinstance(value, Mapping):
_fail(f"{label} must be an object")
if any(not isinstance(key, str) for key in value):
_fail(f"{label} keys must be strings")
return value
def _exact_fields(
value: Mapping[str, object],
*,
expected: frozenset[str],
label: str,
) -> None:
fields = set(value)
missing = sorted(expected - fields)
unexpected = sorted(fields - expected)
if missing:
_fail(f"{label} is missing fields: {missing}")
if unexpected:
_fail(f"{label} contains unexpected fields: {unexpected}")
def _json_list(value: object, *, label: str) -> list[object]:
if not isinstance(value, list):
_fail(f"{label} must be a JSON array")
return value
def _json_integer(value: object, *, label: str) -> int:
if type(value) is not int:
_fail(f"{label} must be an integer")
return value
def _json_integer_tuple(value: object, *, label: str) -> tuple[int, ...]:
items = _json_list(value, label=label)
return tuple(
_json_integer(item, label=f"{label}[{index}]")
for index, item in enumerate(items)
)
def _json_text_tuple(value: object, *, label: str) -> tuple[str, ...]:
items = _json_list(value, label=label)
return tuple(
_canonical_text(item, label=f"{label}[{index}]")
for index, item in enumerate(items)
)
def _normalized_label(value: str) -> str:
return " ".join(value.casefold().split())
def _positive_columns(values: tuple[int, ...], *, label: str) -> tuple[int, ...]:
if (
not values
or any(type(value) is not int or value <= 0 for value in values)
or tuple(sorted(set(values))) != values
):
_fail(f"{label} must be sorted unique positive integer columns")
return values
def _compile_pattern(value: str, *, label: str) -> Pattern[str]:
text = _canonical_text(value, label=label)
if len(text) > MAX_PATTERN_LENGTH:
_fail(f"{label} is too long")
try:
return re.compile(text)
except re.error as exc:
raise GeneralJournalParseError(f"{label} must be a valid regex") from exc
def _decimal(
text: str,
*,
decimal_format: str,
sign_policy: str,
label: str,
) -> Decimal:
if decimal_format not in ACCOUNTING_AMOUNT_FORMATS:
_fail(f"{label} uses an unsupported declared decimal format")
if sign_policy not in AMOUNT_SIGN_POLICIES:
_fail("amount_sign_policy is unsupported")
raw = _canonical_text(text, label=label)
if decimal_format == "canonical_dot":
if _CANONICAL_DOT_DECIMAL.fullmatch(raw) is None:
_fail(f"{label} is not an exact canonical-dot Decimal")
normalized = raw
else:
if _ITALIAN_GROUPED_2_DECIMAL.fullmatch(raw) is None:
_fail(f"{label} is not an exact Italian-grouped Decimal")
normalized = raw.replace(".", "").replace(",", ".")
try:
result = Decimal(normalized)
except InvalidOperation as exc:
raise GeneralJournalParseError(f"{label} is not a finite Decimal") from exc
if not result.is_finite():
_fail(f"{label} is not a finite Decimal")
if sign_policy == "nonnegative" and result < 0:
_fail(f"{label} violates the reviewed nonnegative amount policy")
return result
def _decimal_pattern(decimal_format: str) -> Pattern[str]:
if decimal_format == "canonical_dot":
return _CANONICAL_DOT_DECIMAL
if decimal_format == "italian_grouped_2":
return _ITALIAN_GROUPED_2_DECIMAL
_fail("declared decimal format is unsupported")
def _exact_decimal_sum(
values: tuple[Decimal, ...],
*,
label: str,
) -> Decimal:
"""Sum finite Decimals exactly, independent of ambient precision."""
nonzero_values = tuple(value for value in values if not value.is_zero())
if not nonzero_values:
return Decimal(0)
exponents: list[int] = []
for value in nonzero_values:
exponent = value.as_tuple().exponent
if not isinstance(exponent, int) or not value.is_finite():
_fail(f"{label} contains a non-finite Decimal")
exponents.append(exponent)
minimum_exponent = min(exponents)
maximum_adjusted = max(value.adjusted() for value in nonzero_values)
carry_digits = len(str(len(nonzero_values)))
exact_precision = max(
1,
maximum_adjusted - minimum_exponent + 1 + carry_digits,
)
try:
with localcontext() as context:
context.prec = exact_precision
context.Emax = max(
context.Emax,
maximum_adjusted + carry_digits,
)
context.Emin = min(
context.Emin,
minimum_exponent - carry_digits,
)
context.traps[Inexact] = True
context.traps[Rounded] = True
return sum(nonzero_values, Decimal(0))
except DecimalException as exc:
raise GeneralJournalParseError(
f"{label} could not be accumulated exactly"
) from exc
def _validate_layout(layout: PageLayout) -> None:
_canonical_text(layout.layout_id, label="page layout ID")
header_columns = (
layout.date_header_column,
layout.line_header_column,
layout.account_header_column,
layout.debit_header_column,
layout.credit_header_column,
)
if any(type(column) is not int or column <= 0 for column in header_columns):
_fail("page layout header columns must be positive integers")
if not (
layout.date_header_column < layout.account_header_column
and layout.line_header_column < layout.account_header_column
and layout.account_header_column < layout.debit_header_column
and layout.debit_header_column < layout.credit_header_column
):
_fail("page layout header columns are not ordered")
date_columns = _positive_columns(
layout.date_columns,
label=f"{layout.layout_id} date_columns",
)
line_columns = _positive_columns(
layout.line_id_columns,
label=f"{layout.layout_id} line_id_columns",
)
account_columns = _positive_columns(
layout.account_columns,
label=f"{layout.layout_id} account_columns",
)
debit_columns = _positive_columns(
layout.debit_amount_columns,
label=f"{layout.layout_id} debit_amount_columns",
)
credit_columns = _positive_columns(
layout.credit_amount_columns,
label=f"{layout.layout_id} credit_amount_columns",
)
physical_first_line_columns = (
_positive_columns(
layout.physical_first_line_columns,
label=f"{layout.layout_id} physical_first_line_columns",
)
if layout.physical_first_line_columns
else ()
)
if any(column >= layout.account_header_column for column in date_columns):
_fail("date extraction columns must precede the account header")
if any(column >= layout.account_header_column for column in line_columns):
_fail("line-ID extraction columns must precede the account header")
if layout.account_header_column not in account_columns:
_fail("account extraction columns must include the account header column")
if set(debit_columns) & set(credit_columns):
_fail("debit and credit extraction columns must be disjoint")
if any(column <= layout.account_header_column for column in debit_columns):
_fail("debit extraction columns must follow the account header")
if any(column <= layout.account_header_column for column in credit_columns):
_fail("credit extraction columns must follow the account header")
extraction_columns = (
set(date_columns)
| set(line_columns)
| set(account_columns)
| set(debit_columns)
| set(credit_columns)
)
if not set(physical_first_line_columns).issubset(extraction_columns):
_fail("physical-first-line columns must be reviewed extraction columns")
def _reviewed_amount_locator_key(
locator: ReviewedAmountLocator,
) -> tuple[int, int, int]:
return (locator.row_number, locator.column, locator.line_index)
def _reviewed_amount_pair_order_key(
pair: ReviewedAmountPair,
) -> tuple[int, int, int, int]:
return (
min(
pair.debit.movement_row_number,
pair.credit.movement_row_number,
),
pair.debit.movement_row_number,
pair.credit.movement_row_number,
pair.debit.movement_line_id,
)
def _reviewed_amountless_exclusion_order_key(
exclusion: ReviewedAmountlessExclusion,
) -> tuple[int, int, str, tuple[int, ...], tuple[int, ...]]:
return (
exclusion.row_number,
exclusion.line_id,
exclusion.layout_id,
exclusion.nonempty_columns,
exclusion.residual_columns,
)
def _compile_contract(
contract: GeneralJournalLayoutContract,
) -> _CompiledContract:
if not isinstance(contract, GeneralJournalLayoutContract):
_fail("layout_contract must be a GeneralJournalLayoutContract")
if contract.contract_version != CONTRACT_VERSION:
_fail("layout contract version is not supported")
if contract.review_status != REVIEW_STATUS:
_fail("layout contract must be explicitly reviewed")
_canonical_text(contract.sheet_name, label="sheet_name")
for label, value in (
("date_header_label", contract.date_header_label),
("line_header_label", contract.line_header_label),
("account_header_label", contract.account_header_label),
("debit_header_label", contract.debit_header_label),
("credit_header_label", contract.credit_header_label),
):
_canonical_text(value, label=label)
if not contract.page_layouts:
_fail("layout contract must declare page layouts")
for layout in contract.page_layouts:
_validate_layout(layout)
layout_ids = [layout.layout_id for layout in contract.page_layouts]
if len(layout_ids) != len(set(layout_ids)):
_fail("page layout IDs must be unique")
layout_id_set = frozenset(layout_ids)
header_signatures = [
(
layout.date_header_column,
layout.line_header_column,
layout.account_header_column,
layout.debit_header_column,
layout.credit_header_column,
)
for layout in contract.page_layouts
]
if len(header_signatures) != len(set(header_signatures)):
_fail("page layout header signatures must be unique")
date_patterns: list[_CompiledDatePattern] = []
if not contract.date_patterns:
_fail("layout contract must declare date patterns")
for index, item in enumerate(contract.date_patterns):
compiled = _compile_pattern(
item.pattern,
label=f"date_patterns[{index}].pattern",
)
if "date" not in compiled.groupindex:
_fail("each date pattern must provide a named date group")
if item.strptime_format not in {"%Y-%m-%d", "%d/%m/%Y"}:
_fail("date pattern uses an unsupported reviewed date format")
date_patterns.append(
_CompiledDatePattern(
pattern=compiled,
strptime_format=item.strptime_format,
)
)
account_code_pattern = _compile_pattern(
contract.account_code_pattern,
label="account_code_pattern",
)
logical_candidate_pattern = _compile_pattern(
contract.logical_candidate_pattern,
label="logical_candidate_pattern",
)
if not contract.logical_movement_patterns:
_fail("layout contract must declare logical movement patterns")
logical_patterns: list[_CompiledLogicalMovementPattern] = []
for index, item in enumerate(contract.logical_movement_patterns):
if (
not item.layout_ids
or tuple(sorted(set(item.layout_ids))) != item.layout_ids
or any(
not isinstance(layout_id, str)
or not layout_id
or layout_id != layout_id.strip()
for layout_id in item.layout_ids
)
):
_fail(
"logical movement pattern layout IDs must be sorted unique "
"canonical text"
)
if not set(item.layout_ids).issubset(layout_id_set):
_fail("logical movement pattern references an unknown layout ID")
compiled = _compile_pattern(
item.pattern,
label=f"logical_movement_patterns[{index}].pattern",
)
if not {"line_id", "account", "debit", "credit"}.issubset(compiled.groupindex):
_fail(
"each logical movement pattern must provide named line_id, "
"account, debit, and credit groups"
)
if item.amount_format not in ACCOUNTING_AMOUNT_FORMATS:
_fail("logical movement pattern has an unsupported amount format")
logical_patterns.append(
_CompiledLogicalMovementPattern(
layout_ids=frozenset(item.layout_ids),
pattern=compiled,
amount_format=item.amount_format,
)
)
physical_embedded_patterns: list[_CompiledPhysicalEmbeddedAmountPattern] = []
for index, item in enumerate(contract.physical_embedded_amount_patterns):
if (
not item.layout_ids
or tuple(sorted(set(item.layout_ids))) != item.layout_ids
or any(
not isinstance(layout_id, str)
or not layout_id
or layout_id != layout_id.strip()
for layout_id in item.layout_ids
)
):
_fail(
"physical embedded amount pattern layout IDs must be "
"sorted unique canonical text"
)
if not set(item.layout_ids).issubset(layout_id_set):
_fail("physical embedded amount pattern references an unknown " "layout ID")
if type(item.column) is not int or item.column <= 0:
_fail("physical embedded amount pattern column must be positive")
compiled = _compile_pattern(
item.pattern,
label=f"physical_embedded_amount_patterns[{index}].pattern",
)
if not {"debit", "credit"}.issubset(compiled.groupindex):
_fail(
"each physical embedded amount pattern must provide named "
"debit and credit groups"
)
if item.amount_format not in ACCOUNTING_AMOUNT_FORMATS:
_fail(
"physical embedded amount pattern has an unsupported " "amount format"
)
physical_embedded_patterns.append(
_CompiledPhysicalEmbeddedAmountPattern(
layout_ids=frozenset(item.layout_ids),
column=item.column,
pattern=compiled,
amount_format=item.amount_format,
)
)
if not isinstance(contract.reviewed_amount_pairs, tuple):
_fail("reviewed_amount_pairs must be a tuple")
if any(
not isinstance(pair, ReviewedAmountPair)
for pair in contract.reviewed_amount_pairs
):
_fail("reviewed amount pair must use the reviewed pair type")
if (
tuple(
sorted(
contract.reviewed_amount_pairs,
key=_reviewed_amount_pair_order_key,
)
)
!= contract.reviewed_amount_pairs
):
_fail("reviewed amount pairs must be in canonical movement-row order")
reviewed_amount_pairs: list[_CompiledReviewedAmountPair] = []
reviewed_movement_rows: set[int] = set()
reviewed_movement_line_ids: set[int] = set()
reviewed_locator_keys: set[tuple[int, int, int]] = set()
reviewed_locator_rows: set[int] = set()
for pair_index, pair in enumerate(contract.reviewed_amount_pairs):
if (
pair.debit.movement_row_number == pair.credit.movement_row_number
or pair.debit.movement_line_id == pair.credit.movement_line_id
):
_fail("reviewed amount pair members must be distinct")
compiled_members: list[_CompiledReviewedAmountPairMember] = []
for role, member in (("debit", pair.debit), ("credit", pair.credit)):
if not isinstance(member, ReviewedAmountPairMember):
_fail("reviewed amount pair member has an invalid type")
if member.movement_layout_id not in layout_id_set:
_fail(
"reviewed amount pair member references an unknown "
"movement layout"
)
if (
type(member.movement_row_number) is not int
or member.movement_row_number <= 0
or type(member.movement_line_id) is not int
or member.movement_line_id <= 0
):
_fail(
"reviewed amount pair movement row and line ID must be "
"positive integers"
)
if member.movement_row_number in reviewed_movement_rows:
_fail("reviewed amount pair movement rows must be unique")
if member.movement_line_id in reviewed_movement_line_ids:
_fail("reviewed amount pair movement line IDs must be unique")
reviewed_movement_rows.add(member.movement_row_number)
reviewed_movement_line_ids.add(member.movement_line_id)
locator = member.amount_locator
if not isinstance(locator, ReviewedAmountLocator):
_fail("reviewed amount locator has an invalid type")
if locator.layout_id not in layout_id_set:
_fail("reviewed amount locator references an unknown layout ID")
if (
type(locator.row_number) is not int
or locator.row_number <= 0
or type(locator.column) is not int
or locator.column <= 0
or type(locator.line_index) is not int
or locator.line_index < 0
):
_fail(
"reviewed amount locator coordinates must be positive "
"rows/columns and a nonnegative line index"
)
locator_key = _reviewed_amount_locator_key(locator)
if locator_key in reviewed_locator_keys:
_fail("reviewed amount locators must be globally unique")
reviewed_locator_keys.add(locator_key)
reviewed_locator_rows.add(locator.row_number)
compiled_pattern = _compile_pattern(
locator.pattern,
label=(
f"reviewed_amount_pairs[{pair_index}].{role}."
"amount_locator.pattern"
),
)
if "amount" not in compiled_pattern.groupindex:
_fail(
"reviewed amount locator pattern must provide a named "
"amount group"
)
if locator.amount_format not in ACCOUNTING_AMOUNT_FORMATS:
_fail("reviewed amount locator has an unsupported amount format")
compiled_members.append(
_CompiledReviewedAmountPairMember(
value=member,
role=role,
amount_locator=_CompiledReviewedAmountLocator(
value=locator,
pattern=compiled_pattern,
),
)
)
reviewed_amount_pairs.append(
_CompiledReviewedAmountPair(
debit=compiled_members[0],
credit=compiled_members[1],
)
)
if not isinstance(contract.reviewed_amountless_exclusions, tuple):
_fail("reviewed_amountless_exclusions must be a tuple")
if any(
not isinstance(exclusion, ReviewedAmountlessExclusion)
for exclusion in contract.reviewed_amountless_exclusions
):
_fail("reviewed amountless exclusion must use the reviewed exclusion type")
if (
tuple(
sorted(
contract.reviewed_amountless_exclusions,
key=_reviewed_amountless_exclusion_order_key,
)
)
!= contract.reviewed_amountless_exclusions
):
_fail("reviewed amountless exclusions must be in canonical row order")
reviewed_exclusion_rows: set[int] = set()
reviewed_exclusion_line_ids: set[int] = set()
for exclusion in contract.reviewed_amountless_exclusions:
_canonical_text(
exclusion.layout_id,
label="reviewed amountless exclusion layout ID",
)
if exclusion.layout_id not in layout_id_set:
_fail("reviewed amountless exclusion references an unknown layout")
if (
type(exclusion.row_number) is not int
or exclusion.row_number <= 0
or type(exclusion.line_id) is not int
or exclusion.line_id <= 0
):
_fail(
"reviewed amountless exclusion row and line ID must be "
"positive integers"
)
if exclusion.row_number in reviewed_exclusion_rows:
_fail("reviewed amountless exclusion rows must be unique")
if exclusion.line_id in reviewed_exclusion_line_ids:
_fail("reviewed amountless exclusion line IDs must be unique")
if exclusion.row_number in reviewed_movement_rows:
_fail(
"reviewed amountless exclusion rows must not overlap "
"reviewed amount pair rows"
)
if exclusion.row_number in reviewed_locator_rows:
_fail(
"reviewed amountless exclusion rows must not overlap "
"reviewed amount locator rows"
)
if exclusion.line_id in reviewed_movement_line_ids:
_fail(
"reviewed amountless exclusion line IDs must not overlap "
"reviewed amount pair line IDs"
)
if not isinstance(exclusion.nonempty_columns, tuple):
_fail("reviewed amountless exclusion nonempty columns must be a tuple")
_positive_columns(
exclusion.nonempty_columns,
label="reviewed amountless exclusion nonempty columns",
)
if any(column > MAX_XLSX_COLUMN for column in exclusion.nonempty_columns):
_fail(
"reviewed amountless exclusion nonempty columns exceed XLSX " "limits"
)
if not isinstance(exclusion.residual_columns, tuple):
_fail("reviewed amountless exclusion residual columns must be a tuple")
if (
any(
type(column) is not int or column <= 0 or column > MAX_XLSX_COLUMN
for column in exclusion.residual_columns
)
or tuple(sorted(set(exclusion.residual_columns)))
!= exclusion.residual_columns
):
_fail(
"reviewed amountless exclusion residual columns must be "
"sorted unique positive XLSX columns"
)
if not set(exclusion.residual_columns).issubset(exclusion.nonempty_columns):
_fail(
"reviewed amountless exclusion residual columns must be a "
"subset of nonempty columns"
)
if len(set(exclusion.nonempty_columns) - set(exclusion.residual_columns)) != 2:
_fail(
"reviewed amountless exclusion must reserve exactly two "
"line/account signal columns"
)
reviewed_exclusion_rows.add(exclusion.row_number)
reviewed_exclusion_line_ids.add(exclusion.line_id)
if contract.physical_amount_format not in ACCOUNTING_AMOUNT_FORMATS:
_fail("physical_amount_format is unsupported")
if (
any(
type(line_id) is not int or line_id <= 0
for line_id in contract.reviewed_zero_amount_line_ids
)
or tuple(sorted(set(contract.reviewed_zero_amount_line_ids)))
!= contract.reviewed_zero_amount_line_ids
):
_fail(
"reviewed zero-amount line IDs must be sorted unique positive " "integers"
)
if reviewed_movement_line_ids & set(contract.reviewed_zero_amount_line_ids):
_fail(
"reviewed amount pair line IDs must not overlap reviewed "
"zero-amount line IDs"
)
if reviewed_exclusion_line_ids & set(contract.reviewed_zero_amount_line_ids):
_fail(
"reviewed amountless exclusion line IDs must not overlap reviewed "
"zero-amount line IDs"
)
if contract.control_amount_format not in ACCOUNTING_AMOUNT_FORMATS:
_fail("control_amount_format is unsupported")
if contract.amount_sign_policy not in AMOUNT_SIGN_POLICIES:
_fail("amount_sign_policy is unsupported")
control_pattern = _compile_pattern(
contract.control_pattern,
label="control_pattern",
)
if not {"debit", "credit"}.issubset(control_pattern.groupindex):
_fail("control_pattern must provide named debit and credit groups")
reviewed_debit = _decimal(
contract.reviewed_final_debit_total,
decimal_format="canonical_dot",
sign_policy=contract.amount_sign_policy,
label="reviewed_final_debit_total",
)
reviewed_credit = _decimal(
contract.reviewed_final_credit_total,
decimal_format="canonical_dot",
sign_policy=contract.amount_sign_policy,
label="reviewed_final_credit_total",
)
return _CompiledContract(
value=contract,
account_code_pattern=account_code_pattern,
logical_candidate_pattern=logical_candidate_pattern,
logical_movement_patterns=tuple(logical_patterns),
physical_embedded_amount_patterns=tuple(physical_embedded_patterns),
reviewed_amount_pairs=tuple(reviewed_amount_pairs),
reviewed_amountless_exclusions=(contract.reviewed_amountless_exclusions),
reviewed_zero_amount_line_ids=frozenset(contract.reviewed_zero_amount_line_ids),
date_patterns=tuple(date_patterns),
control_pattern=control_pattern,
reviewed_debit_total=reviewed_debit,
reviewed_credit_total=reviewed_credit,
)
def general_journal_layout_contract_from_mapping(
value: Mapping[str, object],
) -> GeneralJournalLayoutContract:
"""Build and validate a contract from an exact JSON-compatible object."""
source = _mapping(value, label="layout contract")
_exact_fields(source, expected=_CONTRACT_FIELDS, label="layout contract")
layouts: list[PageLayout] = []
for index, raw_layout in enumerate(
_json_list(source["page_layouts"], label="page_layouts")
):
layout = _mapping(raw_layout, label=f"page_layouts[{index}]")
_exact_fields(
layout,
expected=_PAGE_LAYOUT_FIELDS,
label=f"page_layouts[{index}]",
)
layouts.append(
PageLayout(
layout_id=_canonical_text(
layout["layout_id"],
label=f"page_layouts[{index}].layout_id",
),
date_header_column=_json_integer(
layout["date_header_column"],
label=f"page_layouts[{index}].date_header_column",
),
line_header_column=_json_integer(
layout["line_header_column"],
label=f"page_layouts[{index}].line_header_column",
),
account_header_column=_json_integer(
layout["account_header_column"],
label=f"page_layouts[{index}].account_header_column",
),
debit_header_column=_json_integer(
layout["debit_header_column"],
label=f"page_layouts[{index}].debit_header_column",
),
credit_header_column=_json_integer(
layout["credit_header_column"],
label=f"page_layouts[{index}].credit_header_column",
),
date_columns=_json_integer_tuple(
layout["date_columns"],
label=f"page_layouts[{index}].date_columns",
),
line_id_columns=_json_integer_tuple(
layout["line_id_columns"],
label=f"page_layouts[{index}].line_id_columns",
),
account_columns=_json_integer_tuple(
layout["account_columns"],
label=f"page_layouts[{index}].account_columns",
),
debit_amount_columns=_json_integer_tuple(
layout["debit_amount_columns"],
label=f"page_layouts[{index}].debit_amount_columns",
),
credit_amount_columns=_json_integer_tuple(
layout["credit_amount_columns"],
label=f"page_layouts[{index}].credit_amount_columns",
),
physical_first_line_columns=_json_integer_tuple(
layout["physical_first_line_columns"],
label=(f"page_layouts[{index}].physical_first_line_columns"),
),
)
)
dates: list[DatePattern] = []
for index, raw_date in enumerate(
_json_list(source["date_patterns"], label="date_patterns")
):
item = _mapping(raw_date, label=f"date_patterns[{index}]")
_exact_fields(
item,
expected=_DATE_PATTERN_FIELDS,
label=f"date_patterns[{index}]",
)
dates.append(
DatePattern(
pattern=_canonical_text(
item["pattern"],
label=f"date_patterns[{index}].pattern",
),
strptime_format=_canonical_text(
item["strptime_format"],
label=f"date_patterns[{index}].strptime_format",
),
)
)
logical_patterns: list[LogicalMovementPattern] = []
for index, raw_pattern in enumerate(
_json_list(
source["logical_movement_patterns"],
label="logical_movement_patterns",
)
):
item = _mapping(
raw_pattern,
label=f"logical_movement_patterns[{index}]",
)
_exact_fields(
item,
expected=_LOGICAL_PATTERN_FIELDS,
label=f"logical_movement_patterns[{index}]",
)
logical_patterns.append(
LogicalMovementPattern(
layout_ids=_json_text_tuple(
item["layout_ids"],
label=f"logical_movement_patterns[{index}].layout_ids",
),
pattern=_canonical_text(
item["pattern"],
label=f"logical_movement_patterns[{index}].pattern",
),
amount_format=_canonical_text(
item["amount_format"],
label=f"logical_movement_patterns[{index}].amount_format",
),
)
)
physical_embedded_patterns: list[PhysicalEmbeddedAmountPattern] = []
for index, raw_pattern in enumerate(
_json_list(
source["physical_embedded_amount_patterns"],
label="physical_embedded_amount_patterns",
)
):
item = _mapping(
raw_pattern,
label=f"physical_embedded_amount_patterns[{index}]",
)
_exact_fields(
item,
expected=_PHYSICAL_EMBEDDED_AMOUNT_PATTERN_FIELDS,
label=f"physical_embedded_amount_patterns[{index}]",
)
physical_embedded_patterns.append(
PhysicalEmbeddedAmountPattern(
layout_ids=_json_text_tuple(
item["layout_ids"],
label=(f"physical_embedded_amount_patterns[{index}]" ".layout_ids"),
),
column=_json_integer(
item["column"],
label=(f"physical_embedded_amount_patterns[{index}].column"),
),
pattern=_canonical_text(
item["pattern"],
label=(f"physical_embedded_amount_patterns[{index}].pattern"),
),
amount_format=_canonical_text(
item["amount_format"],
label=(
f"physical_embedded_amount_patterns[{index}]" ".amount_format"
),
),
)
)
reviewed_amount_pairs: list[ReviewedAmountPair] = []
for pair_index, raw_pair in enumerate(
_json_list(
source["reviewed_amount_pairs"],
label="reviewed_amount_pairs",
)
):
pair = _mapping(
raw_pair,
label=f"reviewed_amount_pairs[{pair_index}]",
)
_exact_fields(
pair,
expected=_REVIEWED_AMOUNT_PAIR_FIELDS,
label=f"reviewed_amount_pairs[{pair_index}]",
)
members: dict[str, ReviewedAmountPairMember] = {}
for role in ("debit", "credit"):
member_label = f"reviewed_amount_pairs[{pair_index}].{role}"
member = _mapping(pair[role], label=member_label)
_exact_fields(
member,
expected=_REVIEWED_AMOUNT_PAIR_MEMBER_FIELDS,
label=member_label,
)
locator_label = f"{member_label}.amount_locator"
locator = _mapping(
member["amount_locator"],
label=locator_label,
)
_exact_fields(
locator,
expected=_REVIEWED_AMOUNT_LOCATOR_FIELDS,
label=locator_label,
)
members[role] = ReviewedAmountPairMember(
movement_layout_id=_canonical_text(
member["movement_layout_id"],
label=f"{member_label}.movement_layout_id",
),
movement_row_number=_json_integer(
member["movement_row_number"],
label=f"{member_label}.movement_row_number",
),
movement_line_id=_json_integer(
member["movement_line_id"],
label=f"{member_label}.movement_line_id",
),
amount_locator=ReviewedAmountLocator(
layout_id=_canonical_text(
locator["layout_id"],
label=f"{locator_label}.layout_id",
),
row_number=_json_integer(
locator["row_number"],
label=f"{locator_label}.row_number",
),
column=_json_integer(
locator["column"],
label=f"{locator_label}.column",
),
line_index=_json_integer(
locator["line_index"],
label=f"{locator_label}.line_index",
),
pattern=_canonical_text(
locator["pattern"],
label=f"{locator_label}.pattern",
),
amount_format=_canonical_text(
locator["amount_format"],
label=f"{locator_label}.amount_format",
),
),
)
reviewed_amount_pairs.append(
ReviewedAmountPair(
debit=members["debit"],
credit=members["credit"],
)
)
reviewed_amountless_exclusions: list[ReviewedAmountlessExclusion] = []
for exclusion_index, raw_exclusion in enumerate(
_json_list(
source["reviewed_amountless_exclusions"],
label="reviewed_amountless_exclusions",
)
):
exclusion_label = f"reviewed_amountless_exclusions[{exclusion_index}]"
exclusion = _mapping(raw_exclusion, label=exclusion_label)
_exact_fields(
exclusion,
expected=_REVIEWED_AMOUNTLESS_EXCLUSION_FIELDS,
label=exclusion_label,
)
reviewed_amountless_exclusions.append(
ReviewedAmountlessExclusion(
layout_id=_canonical_text(
exclusion["layout_id"],
label=f"{exclusion_label}.layout_id",
),
row_number=_json_integer(
exclusion["row_number"],
label=f"{exclusion_label}.row_number",
),
line_id=_json_integer(
exclusion["line_id"],
label=f"{exclusion_label}.line_id",
),
nonempty_columns=_json_integer_tuple(
exclusion["nonempty_columns"],
label=f"{exclusion_label}.nonempty_columns",
),
residual_columns=_json_integer_tuple(
exclusion["residual_columns"],
label=f"{exclusion_label}.residual_columns",
),
)
)
contract = GeneralJournalLayoutContract(
contract_version=_canonical_text(
source["contract_version"],
label="contract_version",
),
review_status=_canonical_text(
source["review_status"],
label="review_status",
),
sheet_name=_canonical_text(source["sheet_name"], label="sheet_name"),
date_header_label=_canonical_text(
source["date_header_label"],
label="date_header_label",
),
line_header_label=_canonical_text(
source["line_header_label"],
label="line_header_label",
),
account_header_label=_canonical_text(
source["account_header_label"],
label="account_header_label",
),
debit_header_label=_canonical_text(
source["debit_header_label"],
label="debit_header_label",
),
credit_header_label=_canonical_text(
source["credit_header_label"],
label="credit_header_label",
),
page_layouts=tuple(layouts),
date_patterns=tuple(dates),
account_code_pattern=_canonical_text(
source["account_code_pattern"],
label="account_code_pattern",
),
logical_candidate_pattern=_canonical_text(
source["logical_candidate_pattern"],
label="logical_candidate_pattern",
),
logical_movement_patterns=tuple(logical_patterns),
physical_embedded_amount_patterns=tuple(physical_embedded_patterns),
reviewed_amount_pairs=tuple(reviewed_amount_pairs),
reviewed_amountless_exclusions=tuple(reviewed_amountless_exclusions),
reviewed_zero_amount_line_ids=_json_integer_tuple(
source["reviewed_zero_amount_line_ids"],
label="reviewed_zero_amount_line_ids",
),
physical_amount_format=_canonical_text(
source["physical_amount_format"],
label="physical_amount_format",
),
amount_sign_policy=_canonical_text(
source["amount_sign_policy"],
label="amount_sign_policy",
),
control_pattern=_canonical_text(
source["control_pattern"],
label="control_pattern",
),
control_amount_format=_canonical_text(
source["control_amount_format"],
label="control_amount_format",
),
reviewed_final_debit_total=_canonical_text(
source["reviewed_final_debit_total"],
label="reviewed_final_debit_total",
),
reviewed_final_credit_total=_canonical_text(
source["reviewed_final_credit_total"],
label="reviewed_final_credit_total",
),
)
_compile_contract(contract)
return contract
def _strict_object_pairs(
pairs: list[tuple[str, object]],
) -> dict[str, object]:
result: dict[str, object] = {}
for key, item in pairs:
if key in result:
_fail(f"layout contract JSON contains duplicate field: {key}")
result[key] = item
return result
def _reject_json_constant(value: str) -> None:
_fail(f"layout contract JSON contains unsupported constant: {value}")
def load_general_journal_layout_contract(
path: Path,
*,
expected_contract_sha256: str,
) -> GeneralJournalLayoutContract:
"""Load one exact, duplicate-key-safe, digest-bound private contract."""
payload = _stable_file_bytes(
Path(path),
expected_sha256=expected_contract_sha256,
max_bytes=MAX_CONTRACT_BYTES,
label="layout contract",
)
if payload.startswith(b"\xef\xbb\xbf"):
_fail("layout contract JSON must not contain a UTF-8 BOM")
try:
text = payload.decode("utf-8")
except UnicodeDecodeError as exc:
raise GeneralJournalParseError("layout contract JSON must be UTF-8") from exc
try:
raw = json.loads(
text,
object_pairs_hook=_strict_object_pairs,
parse_constant=_reject_json_constant,
)
except json.JSONDecodeError as exc:
raise GeneralJournalParseError("layout contract JSON is invalid") from exc
return general_journal_layout_contract_from_mapping(
_mapping(raw, label="layout contract")
)
def _stable_file_bytes(
path: Path,
*,
expected_sha256: str,
max_bytes: int,
label: str,
) -> bytes:
if SHA256_PATTERN.fullmatch(expected_sha256) is None:
_fail(f"expected {label} SHA-256 must be a lowercase digest")
file_path = Path(path)
try:
lexical_status = file_path.lstat()
if stat.S_ISLNK(lexical_status.st_mode):
_fail(f"{label} path must not be a symlink")
if not stat.S_ISREG(lexical_status.st_mode):
_fail(f"{label} path must identify one regular file")
if lexical_status.st_nlink != 1:
_fail(f"{label} file must not be hard linked")
with file_path.open("rb") as handle:
before = os.fstat(handle.fileno())
if not stat.S_ISREG(before.st_mode):
_fail(f"{label} path must identify one regular file")
if before.st_nlink != 1:
_fail(f"{label} file must not be hard linked")
if before.st_size > max_bytes:
_fail(f"{label} exceeds the parser byte limit")
payload = handle.read()
after = os.fstat(handle.fileno())
except OSError as exc:
raise GeneralJournalParseError(f"{label} could not be read") from exc
identity_before = (
before.st_dev,
before.st_ino,
before.st_size,
before.st_mtime_ns,
before.st_ctime_ns,
)
identity_lexical = (
lexical_status.st_dev,
lexical_status.st_ino,
lexical_status.st_size,
lexical_status.st_mtime_ns,
lexical_status.st_ctime_ns,
)
identity_after = (
after.st_dev,
after.st_ino,
after.st_size,
after.st_mtime_ns,
after.st_ctime_ns,
)
if (
identity_lexical != identity_before
or identity_before != identity_after
or len(payload) != after.st_size
):
_fail(f"{label} changed while it was read")
try:
current = file_path.lstat()
except OSError as exc:
raise GeneralJournalParseError(
f"{label} path changed while it was read"
) from exc
if stat.S_ISLNK(current.st_mode):
_fail(f"{label} path changed to a symlink while it was read")
if not stat.S_ISREG(current.st_mode) or current.st_nlink != 1:
_fail(f"{label} path changed while it was read")
identity_current = (
current.st_dev,
current.st_ino,
current.st_size,
current.st_mtime_ns,
current.st_ctime_ns,
)
if identity_after != identity_current:
_fail(f"{label} path changed while it was read")
if hashlib.sha256(payload).hexdigest() != expected_sha256:
_fail(f"{label} does not match the exact reviewed digest")
return payload
def _stable_source_bytes(path: Path, *, expected_sha256: str) -> bytes:
return _stable_file_bytes(
path,
expected_sha256=expected_sha256,
max_bytes=MAX_SOURCE_BYTES,
label="source workbook",
)
def _safe_zip_member(target: str, *, base: str) -> str:
if (
not target
or "\\" in target
or "\x00" in target
or "?" in target
or "#" in target
or ":" in target
):
_fail("workbook relationship target is unsafe")
package_absolute = target.startswith("/")
relative_target = target[1:] if package_absolute else target
parts = relative_target.split("/")
if any(part in {"", ".", ".."} for part in parts):
_fail("workbook relationship target is unsafe")
resolved = posixpath.normpath(
relative_target if package_absolute else posixpath.join(base, target)
)
if (
not resolved
or resolved == ".."
or resolved.startswith("../")
or resolved.startswith("/")
):
_fail("workbook relationship target escapes the package")
return resolved
def _validate_archive_limits(archive: zipfile.ZipFile) -> None:
infos = archive.infolist()
if len(infos) > MAX_XLSX_MEMBERS:
_fail("XLSX package contains too many members")
names: set[str] = set()
total_uncompressed = 0
for info in infos:
name = info.filename
if (
not name
or "\\" in name
or name.startswith("/")
or posixpath.normpath(name) != name.rstrip("/")
or name == ".."
or name.startswith("../")
):
_fail("XLSX package contains an unsafe member path")
if name in names:
_fail("XLSX package contains duplicate member names")
names.add(name)
if info.flag_bits & 0x1:
_fail("XLSX package must not contain encrypted members")
if info.file_size > MAX_XLSX_MEMBER_UNCOMPRESSED_BYTES:
_fail("XLSX package member exceeds the decompressed byte limit")
total_uncompressed += info.file_size
if total_uncompressed > MAX_XLSX_TOTAL_UNCOMPRESSED_BYTES:
_fail("XLSX package exceeds the total decompressed byte limit")
if (
info.file_size
and info.file_size / max(info.compress_size, 1) > MAX_XLSX_COMPRESSION_RATIO
):
_fail("XLSX package member exceeds the compression-ratio limit")
def _safe_xml_member_bytes(
archive: zipfile.ZipFile,
path: str,
*,
label: str,
) -> bytes:
"""Read bounded package XML after rejecting entity-capable syntax."""
try:
payload = archive.read(path)
except (KeyError, OSError) as exc:
raise GeneralJournalParseError(f"{label} is missing or unreadable") from exc
if len(payload) > MAX_XLSX_MEMBER_UNCOMPRESSED_BYTES:
_fail(f"{label} exceeds the decompressed byte limit")
if b"\x00" in payload:
_fail(f"{label} uses an unsupported XML encoding")
declaration = _XML_DECLARATION.match(payload)
if declaration is not None:
encoding = _XML_ENCODING.search(declaration.group(0))
if (
encoding is not None
and encoding.group("encoding").decode("ascii").casefold()
not in _SUPPORTED_XML_ENCODINGS
):
_fail(f"{label} uses an unsupported XML encoding")
lowered = payload.lower()
if b"<!doctype" in lowered or b"<!entity" in lowered:
_fail(f"{label} contains a forbidden DTD or entity declaration")
return payload
def _shared_strings(archive: zipfile.ZipFile) -> tuple[str, ...]:
path = "xl/sharedStrings.xml"
if path not in archive.namelist():
return ()
payload = _safe_xml_member_bytes(
archive,
path,
label="shared strings XML",
)
try:
# DTDs and entity declarations were rejected above.
root = ElementTree.fromstring(payload) # nosec B314
except ElementTree.ParseError as exc:
raise GeneralJournalParseError("shared strings XML is invalid") from exc
values: list[str] = []
character_count = 0
for item in root.findall(f"{{{_MAIN_NS}}}si"):
if len(values) >= MAX_SHARED_STRINGS:
_fail("shared-string table exceeds the item-count limit")
value = "".join(node.text or "" for node in item.iter(f"{{{_MAIN_NS}}}t"))
character_count += len(value)
if character_count > MAX_SHARED_STRING_CHARACTERS:
_fail("shared-string table exceeds the character-count limit")
values.append(value)
return tuple(values)
def _worksheet_path(archive: zipfile.ZipFile, *, sheet_name: str) -> str:
workbook_payload = _safe_xml_member_bytes(
archive,
"xl/workbook.xml",
label="workbook XML",
)
relationships_payload = _safe_xml_member_bytes(
archive,
"xl/_rels/workbook.xml.rels",
label="workbook relationships XML",
)
try:
# DTDs and entity declarations were rejected above.
workbook = ElementTree.fromstring(workbook_payload) # nosec B314
relationships = ElementTree.fromstring(relationships_payload) # nosec B314
except ElementTree.ParseError as exc:
raise GeneralJournalParseError("workbook metadata XML is invalid") from exc
relationship_targets: dict[str, str] = {}
for relationship in relationships.findall(f"{{{_PACKAGE_REL_NS}}}Relationship"):
relationship_id = relationship.get("Id")
target = relationship.get("Target")
if relationship.get("TargetMode") == "External":
continue
if relationship_id and target:
relationship_targets[relationship_id] = target
matches = []
for sheet in workbook.findall(f".//{{{_MAIN_NS}}}sheet"):
if sheet.get("name") == sheet_name:
relationship_id = sheet.get(_RELATIONSHIP_ID)
if not relationship_id or relationship_id not in relationship_targets:
_fail("reviewed sheet has no internal worksheet relationship")
matches.append(
_safe_zip_member(
relationship_targets[relationship_id],
base="xl",
)
)
if len(matches) != 1:
_fail("reviewed sheet name must resolve to exactly one worksheet")
if matches[0] not in archive.namelist():
_fail("reviewed worksheet part is missing")
return matches[0]
def _column_number(cell_reference: str, *, row_number: int) -> int:
match = _CELL_REFERENCE.fullmatch(cell_reference)
if match is None or int(match.group(2)) != row_number:
_fail("worksheet contains an invalid cell reference")
result = 0
for character in match.group(1):
result = result * 26 + ord(character) - ord("A") + 1
if result > MAX_XLSX_COLUMN:
_fail("worksheet cell exceeds the XLSX column limit")
return result
def _cell_text(
cell: ElementTree.Element,
*,
row_number: int,
shared_strings: tuple[str, ...],
) -> tuple[int, str] | None:
reference = cell.get("r")
if reference is None:
_fail("worksheet cell is missing its reference")
column = _column_number(reference, row_number=row_number)
if cell.find(f"{{{_MAIN_NS}}}f") is not None:
_fail("reviewed worksheet must not contain formulas")
cell_type = cell.get("t")
if cell_type == "e":
_fail("reviewed worksheet must not contain error cells")
if cell_type == "inlineStr":
value = "".join(node.text or "" for node in cell.iter(f"{{{_MAIN_NS}}}t"))
else:
value_node = cell.find(f"{{{_MAIN_NS}}}v")
if value_node is None:
return None
raw = value_node.text or ""
if cell_type == "s":
try:
index = int(raw)
value = shared_strings[index]
except (ValueError, IndexError) as exc:
raise GeneralJournalParseError(
"worksheet shared-string reference is invalid"
) from exc
else:
value = raw
if len(value) > MAX_CELL_TEXT_LENGTH:
_fail("worksheet cell exceeds the parser text limit")
if value == "":
return None
return column, value
def _worksheet_rows(
archive: zipfile.ZipFile,
*,
worksheet_path: str,
shared_strings: tuple[str, ...],
) -> tuple[_SheetRow, ...]:
rows: list[_SheetRow] = []
previous_row_number = 0
parsed_row_count = 0
parsed_cell_count = 0
payload = _safe_xml_member_bytes(
archive,
worksheet_path,
label="worksheet XML",
)
try:
with io.BytesIO(payload) as handle:
# DTDs and entity declarations were rejected above.
for _, element in ElementTree.iterparse( # nosec B314
handle,
events=("end",),
):
if element.tag != f"{{{_MAIN_NS}}}row":
continue
parsed_row_count += 1
if parsed_row_count > MAX_WORKSHEET_ROWS:
_fail("worksheet exceeds the row-count limit")
raw_row_number = element.get("r")
try:
row_number = int(raw_row_number or "")
except ValueError as exc:
raise GeneralJournalParseError(
"worksheet row number is invalid"
) from exc
if row_number <= previous_row_number:
_fail("worksheet row numbers must be strictly increasing")
previous_row_number = row_number
cells: dict[int, str] = {}
cell_elements = element.findall(f"{{{_MAIN_NS}}}c")
if len(cell_elements) > MAX_ROW_CELLS:
_fail("worksheet row exceeds the cell-count limit")
parsed_cell_count += len(cell_elements)
if parsed_cell_count > MAX_WORKSHEET_CELLS:
_fail("worksheet exceeds the total cell-count limit")
for cell in cell_elements:
parsed = _cell_text(
cell,
row_number=row_number,
shared_strings=shared_strings,
)
if parsed is None:
continue
column, value = parsed
if column in cells:
_fail("worksheet row contains duplicate cell columns")
cells[column] = value
if cells:
rows.append(
_SheetRow(
row_number=row_number,
cells=MappingProxyType(cells),
)
)
element.clear()
except (ElementTree.ParseError, KeyError, OSError) as exc:
raise GeneralJournalParseError("worksheet XML is invalid") from exc
return tuple(rows)
def _read_rows(payload: bytes, *, sheet_name: str) -> tuple[_SheetRow, ...]:
try:
with zipfile.ZipFile(io.BytesIO(payload)) as archive:
_validate_archive_limits(archive)
worksheet_path = _worksheet_path(archive, sheet_name=sheet_name)
strings = _shared_strings(archive)
return _worksheet_rows(
archive,
worksheet_path=worksheet_path,
shared_strings=strings,
)
except zipfile.BadZipFile as exc:
raise GeneralJournalParseError(
"source workbook is not a valid XLSX package"
) from exc
def _match_layout(
row: _SheetRow,
next_row: _SheetRow | None,
*,
contract: GeneralJournalLayoutContract,
) -> PageLayout | None:
date_label = _normalized_label(contract.date_header_label)
if not any(_normalized_label(value) == date_label for value in row.cells.values()):
return None
if next_row is None or next_row.row_number != row.row_number + 1:
_fail("page header is missing its consecutive second header row")
matches: list[PageLayout] = []
labels = {
"line": _normalized_label(contract.line_header_label),
"account": _normalized_label(contract.account_header_label),
"debit": _normalized_label(contract.debit_header_label),
"credit": _normalized_label(contract.credit_header_label),
}
for layout in contract.page_layouts:
if (
_normalized_label(row.cells.get(layout.date_header_column, ""))
== date_label
and _normalized_label(next_row.cells.get(layout.line_header_column, ""))
== labels["line"]
and _normalized_label(next_row.cells.get(layout.account_header_column, ""))
== labels["account"]
and _normalized_label(next_row.cells.get(layout.debit_header_column, ""))
== labels["debit"]
and _normalized_label(next_row.cells.get(layout.credit_header_column, ""))
== labels["credit"]
):
matches.append(layout)
if len(matches) != 1:
_fail("page header does not match exactly one reviewed layout")
return matches[0]
def _date_from_text(
text: str,
*,
patterns: tuple[_CompiledDatePattern, ...],
label: str,
) -> date | None:
matches: list[date] = []
for item in patterns:
match = item.pattern.match(text)
if match is None:
continue
try:
parsed = datetime.strptime(
match.group("date"),
item.strptime_format,
).date()
except (ValueError, IndexError) as exc:
raise GeneralJournalParseError(f"{label} is invalid") from exc
matches.append(parsed)
if len(matches) > 1:
_fail(f"{label} matches more than one reviewed date pattern")
return matches[0] if matches else None
def _line_id(text: str, *, label: str) -> int:
if re.fullmatch(r"[1-9][0-9]*", text) is None:
_fail(f"{label} must be a positive base-10 integer")
return int(text)
def _account_code(
text: str,
*,
pattern: Pattern[str],
label: str,
) -> str:
matches = list(pattern.finditer(text))
if len(matches) != 1:
_fail(f"{label} must contain exactly one reviewed account code")
return re.sub(r"\s+", "", matches[0].group(0))
def _active_layout_ids_by_row(
rows: tuple[_SheetRow, ...],
*,
contract: GeneralJournalLayoutContract,
) -> Mapping[int, str]:
current_layout: PageLayout | None = None
skip_row_number: int | None = None
result: dict[int, str] = {}
for index, row in enumerate(rows):
if skip_row_number == row.row_number:
skip_row_number = None
continue
next_row = rows[index + 1] if index + 1 < len(rows) else None
layout = _match_layout(row, next_row, contract=contract)
if layout is not None:
current_layout = layout
if next_row is None:
_fail("page header is missing its second header row")
skip_row_number = next_row.row_number
continue
if current_layout is not None:
result[row.row_number] = current_layout.layout_id
return MappingProxyType(result)
def _resolve_reviewed_amount_pairs(
rows: tuple[_SheetRow, ...],
*,
contract: _CompiledContract,
) -> _ResolvedReviewedAmountPairs:
rows_by_number = {row.row_number: row for row in rows}
active_layouts = _active_layout_ids_by_row(
rows,
contract=contract.value,
)
members_by_row: dict[int, _ResolvedReviewedAmountPairMember] = {}
movement_rows_by_line_id: dict[int, int] = {}
locator_keys: set[tuple[int, int, int]] = set()
for pair in contract.reviewed_amount_pairs:
resolved_members: list[_ResolvedReviewedAmountPairMember] = []
for member in (pair.debit, pair.credit):
value = member.value
movement_layout = active_layouts.get(value.movement_row_number)
if movement_layout is None:
_fail("reviewed amount pair movement row is missing")
if movement_layout != value.movement_layout_id:
_fail("reviewed amount pair movement layout does not match")
locator = member.amount_locator
locator_value = locator.value
locator_layout = active_layouts.get(locator_value.row_number)
if locator_layout is None:
_fail("reviewed amount locator row is missing")
if locator_layout != locator_value.layout_id:
_fail("reviewed amount locator layout does not match")
locator_row = rows_by_number.get(locator_value.row_number)
if locator_row is None:
_fail("reviewed amount locator row is missing")
cell = locator_row.cells.get(locator_value.column)
if cell is None:
_fail("reviewed amount locator column is missing")
lines = cell.splitlines()
if locator_value.line_index >= len(lines):
_fail("reviewed amount locator line is missing")
line = lines[locator_value.line_index]
match = locator.pattern.fullmatch(line)
if match is None:
_fail("reviewed amount locator content changed")
if (
contract.logical_candidate_pattern.search(line) is not None
or contract.control_pattern.fullmatch(line) is not None
):
_fail("reviewed amount locator collides with ordinary extraction")
amount = _decimal(
match.group("amount"),
decimal_format=locator_value.amount_format,
sign_policy=contract.value.amount_sign_policy,
label="reviewed amount locator",
)
resolved = _ResolvedReviewedAmountPairMember(
value=value,
role=member.role,
amount=amount,
)
members_by_row[value.movement_row_number] = resolved
movement_rows_by_line_id[value.movement_line_id] = value.movement_row_number
locator_keys.add(_reviewed_amount_locator_key(locator_value))
resolved_members.append(resolved)
if resolved_members[0].amount != resolved_members[1].amount:
_fail("reviewed amount pair values must be exactly equal")
return _ResolvedReviewedAmountPairs(
members_by_row=MappingProxyType(members_by_row),
movement_rows_by_line_id=MappingProxyType(movement_rows_by_line_id),
locator_keys=frozenset(locator_keys),
)
def _resolve_reviewed_amountless_exclusions(
rows: tuple[_SheetRow, ...],
*,
contract: _CompiledContract,
) -> _ResolvedReviewedAmountlessExclusions:
rows_by_number = {row.row_number: row for row in rows}
active_layouts = _active_layout_ids_by_row(
rows,
contract=contract.value,
)
by_row: dict[int, ReviewedAmountlessExclusion] = {}
rows_by_line_id: dict[int, int] = {}
for exclusion in contract.reviewed_amountless_exclusions:
row = rows_by_number.get(exclusion.row_number)
if row is None:
_fail("reviewed amountless exclusion row is missing")
layout_id = active_layouts.get(exclusion.row_number)
if layout_id is None:
_fail("reviewed amountless exclusion row has no active layout")
if layout_id != exclusion.layout_id:
_fail("reviewed amountless exclusion layout does not match")
if tuple(sorted(row.cells)) != exclusion.nonempty_columns or any(
not value.strip() for value in row.cells.values()
):
_fail("reviewed amountless exclusion nonempty columns changed")
if any("\n" in value or "\r" in value for value in row.cells.values()):
_fail("reviewed amountless exclusion contains multiline text")
by_row[exclusion.row_number] = exclusion
rows_by_line_id[exclusion.line_id] = exclusion.row_number
return _ResolvedReviewedAmountlessExclusions(
by_row=MappingProxyType(by_row),
rows_by_line_id=MappingProxyType(rows_by_line_id),
)
class _ParserState:
def __init__(
self,
contract: _CompiledContract,
reviewed_amount_pairs: _ResolvedReviewedAmountPairs,
reviewed_amountless_exclusions: _ResolvedReviewedAmountlessExclusions,
) -> None:
self.contract = contract
self.reviewed_amount_pairs = reviewed_amount_pairs
self.reviewed_amountless_exclusions = reviewed_amountless_exclusions
self.current_layout: PageLayout | None = None
self.current_date: date | None = None
self.movements: list[JournalMovement] = []
self.line_ids: set[int] = set()
self.last_line_id: int | None = None
self.line_id_gap_count = 0
self.page_header_count = 0
self.physical_movement_count = 0
self.logical_movement_count = 0
self.reviewed_amountless_exclusion_rows_seen: set[int] = set()
self.reviewed_zero_amount_line_ids_seen: set[int] = set()
self.reviewed_amount_pair_rows_seen: set[int] = set()
self.reviewed_amount_locator_keys_consumed: set[tuple[int, int, int]] = set()
self.layout_page_counts: dict[str, int] = {}
self.control_totals: tuple[Decimal, Decimal] | None = None
self.control_seen = False
def update_date(self, value: date) -> None:
if self.current_date is not None and value < self.current_date:
_fail("posting dates must be nondecreasing")
self.current_date = value
def add_movement(
self,
*,
line_id: int,
account_code: str,
debit: Decimal,
credit: Decimal,
source_form: str,
) -> None:
if self.current_date is None:
_fail("movement appears before a parseable carried posting date")
self.observe_line_id(line_id)
self.movements.append(
JournalMovement(
line_id=line_id,
posting_date=self.current_date,
account_code=account_code,
debit=debit,
credit=credit,
source_form=source_form,
)
)
if source_form == "physical_row":
self.physical_movement_count += 1
else:
self.logical_movement_count += 1
def observe_line_id(self, line_id: int) -> None:
if self.control_seen:
_fail("movement appears after the reviewed final control line")
if self.current_layout is None:
_fail("movement appears before a reviewed page layout")
if line_id in self.line_ids:
_fail("duplicate global line ID")
if self.last_line_id is not None:
if line_id < self.last_line_id:
_fail("global line IDs must be strictly increasing")
self.line_id_gap_count += line_id - self.last_line_id - 1
self.line_ids.add(line_id)
self.last_line_id = line_id
def exclude_reviewed_amountless_row(
self,
*,
row_number: int,
line_id: int,
) -> None:
if row_number not in self.reviewed_amountless_exclusions.by_row:
_fail("amountless exclusion is not explicitly reviewed")
if row_number in self.reviewed_amountless_exclusion_rows_seen:
_fail("reviewed amountless exclusion row was consumed twice")
self.observe_line_id(line_id)
self.reviewed_amountless_exclusion_rows_seen.add(row_number)
def record_control(self, *, debit: Decimal, credit: Decimal) -> None:
if self.control_totals is not None:
_fail("reviewed final control line must occur exactly once")
self.control_totals = (debit, credit)
self.control_seen = True
def add_reviewed_zero_amount_movement(
self,
*,
line_id: int,
account_code: str,
) -> None:
if line_id not in self.contract.reviewed_zero_amount_line_ids:
_fail("zero-amount movement is not explicitly reviewed")
if line_id in self.reviewed_zero_amount_line_ids_seen:
_fail("reviewed zero-amount line ID was encountered more than once")
self.reviewed_zero_amount_line_ids_seen.add(line_id)
self.add_movement(
line_id=line_id,
account_code=account_code,
debit=Decimal(0),
credit=Decimal(0),
source_form="physical_row",
)
def _record_control_line(
line: str,
*,
state: _ParserState,
) -> bool:
match = state.contract.control_pattern.fullmatch(line)
if match is None:
return False
debit = _decimal(
match.group("debit"),
decimal_format=state.contract.value.control_amount_format,
sign_policy=state.contract.value.amount_sign_policy,
label="source final debit control",
)
credit = _decimal(
match.group("credit"),
decimal_format=state.contract.value.control_amount_format,
sign_policy=state.contract.value.amount_sign_policy,
label="source final credit control",
)
state.record_control(debit=debit, credit=credit)
return True
def _parse_logical_line(line: _LogicalLine, *, state: _ParserState) -> None:
locator_key = (line.row_number, line.column, line.line_index)
if locator_key in state.reviewed_amount_pairs.locator_keys:
return
text = line.text
if not text.strip():
return
if _record_control_line(text, state=state):
return
parsed_date = _date_from_text(
text,
patterns=state.contract.date_patterns,
label="logical-line posting date",
)
if parsed_date is not None:
state.update_date(parsed_date)
if state.contract.logical_candidate_pattern.search(text) is None:
return
if state.current_layout is None:
_fail("logical movement candidate appears before a reviewed page layout")
matches = [
(item, item.pattern.fullmatch(text))
for item in state.contract.logical_movement_patterns
if state.current_layout.layout_id in item.layout_ids
]
matched = [(item, match) for item, match in matches if match is not None]
if len(matched) != 1:
_fail("logical movement candidate does not match exactly one reviewed pattern")
item, match = matched[0]
if match is None:
_fail("logical movement pattern selection is internally inconsistent")
line_id = _line_id(
match.group("line_id"),
label="logical movement line ID",
)
if line_id in state.reviewed_amount_pairs.movement_rows_by_line_id:
_fail("reviewed amount pair line ID appeared in a logical line")
if line_id in state.reviewed_amountless_exclusions.rows_by_line_id:
_fail("reviewed amountless exclusion line ID appeared in a logical line")
if line_id in state.contract.reviewed_zero_amount_line_ids:
_fail("reviewed zero-amount line ID contains a logical amount")
account = _account_code(
match.group("account"),
pattern=state.contract.account_code_pattern,
label="logical movement account code",
)
debit_text = match.group("debit")
credit_text = match.group("credit")
if (debit_text is None) == (credit_text is None):
_fail("logical movement must contain exactly one debit or credit amount")
zero = Decimal(0)
if debit_text is not None:
debit = _decimal(
debit_text,
decimal_format=item.amount_format,
sign_policy=state.contract.value.amount_sign_policy,
label="logical movement debit",
)
credit = zero
else:
debit = zero
credit = _decimal(
credit_text,
decimal_format=item.amount_format,
sign_policy=state.contract.value.amount_sign_policy,
label="logical movement credit",
)
state.add_movement(
line_id=line_id,
account_code=account,
debit=debit,
credit=credit,
source_form="logical_line",
)
def _partition_multiline_cells(
row: _SheetRow,
*,
state: _ParserState,
) -> tuple[
_SheetRow | None,
tuple[_LogicalLine, ...],
tuple[_LogicalLine, ...],
]:
multiline_cells = [
(column, value)
for column, value in row.cells.items()
if "\n" in value or "\r" in value
]
if not multiline_cells:
return row, (), ()
candidate_cell_count = sum(
any(
state.contract.logical_candidate_pattern.search(line) is not None
for line in value.splitlines()
)
for _, value in multiline_cells
)
if candidate_cell_count > 1:
_fail("multiple multiline cells contain movement candidates in one row")
physical_first_columns = (
set(state.current_layout.physical_first_line_columns)
if state.current_layout is not None
else set()
)
remaining = {
column: value
for column, value in row.cells.items()
if "\n" not in value and "\r" not in value
}
before_physical: list[_LogicalLine] = []
after_physical: list[_LogicalLine] = []
for column, value in sorted(multiline_cells):
lines = value.splitlines()
if column in physical_first_columns:
if not lines or not lines[0].strip():
_fail("reviewed physical-first multiline cell has no first line")
remaining[column] = lines[0]
after_physical.extend(
_LogicalLine(
row_number=row.row_number,
column=column,
line_index=line_index,
text=line,
)
for line_index, line in enumerate(lines[1:], start=1)
)
else:
before_physical.extend(
_LogicalLine(
row_number=row.row_number,
column=column,
line_index=line_index,
text=line,
)
for line_index, line in enumerate(lines)
)
if not remaining:
physical_row = None
else:
physical_row = _SheetRow(
row_number=row.row_number,
cells=MappingProxyType(remaining),
)
return physical_row, tuple(before_physical), tuple(after_physical)
def _physical_date(row: _SheetRow, *, state: _ParserState) -> date | None:
layout = state.current_layout
if layout is None:
return None
matches: list[date] = []
for column in layout.date_columns:
value = row.cells.get(column)
if value is None:
continue
parsed = _date_from_text(
value,
patterns=state.contract.date_patterns,
label=f"physical posting date at row {row.row_number}",
)
if parsed is not None:
matches.append(parsed)
if len(matches) > 1:
_fail("physical row contains multiple parseable posting dates")
return matches[0] if matches else None
def _physical_movement(row: _SheetRow, *, state: _ParserState) -> None:
layout = state.current_layout
if layout is None:
return
account_signals = [
(column, row.cells[column])
for column in layout.account_columns
if column in row.cells
and state.contract.account_code_pattern.search(row.cells[column]) is not None
]
line_id_signals = [
(column, row.cells[column])
for column in layout.line_id_columns
if column in row.cells
and re.fullmatch(r"[1-9][0-9]*", row.cells[column]) is not None
]
amount_pattern = _decimal_pattern(state.contract.value.physical_amount_format)
debit_signals = [
(column, row.cells[column])
for column in layout.debit_amount_columns
if column in row.cells
and amount_pattern.fullmatch(row.cells[column]) is not None
]
credit_signals = [
(column, row.cells[column])
for column in layout.credit_amount_columns
if column in row.cells
and amount_pattern.fullmatch(row.cells[column]) is not None
]
embedded_matches = []
for item in state.contract.physical_embedded_amount_patterns:
if layout.layout_id not in item.layout_ids or item.column not in row.cells:
continue
match = item.pattern.fullmatch(row.cells[item.column])
if match is not None:
embedded_matches.append((item, match))
amount_signal_count = (
len(debit_signals) + len(credit_signals) + len(embedded_matches)
)
ordinary_amount_locator_keys = [
(row.row_number, column, 0) for column, _ in (*debit_signals, *credit_signals)
] + [(row.row_number, item.column, 0) for item, _ in embedded_matches]
reviewed_member = state.reviewed_amount_pairs.members_by_row.get(row.row_number)
reviewed_exclusion = state.reviewed_amountless_exclusions.by_row.get(row.row_number)
for _, line_id_text in line_id_signals:
signaled_line_id = int(line_id_text)
expected_row = state.reviewed_amount_pairs.movement_rows_by_line_id.get(
signaled_line_id
)
if expected_row is not None and expected_row != row.row_number:
_fail("reviewed amount pair line ID appeared on the wrong row")
expected_exclusion_row = (
state.reviewed_amountless_exclusions.rows_by_line_id.get(signaled_line_id)
)
if (
expected_exclusion_row is not None
and expected_exclusion_row != row.row_number
):
_fail("reviewed amountless exclusion line ID appeared on the " "wrong row")
if reviewed_member is not None:
if layout.layout_id != reviewed_member.value.movement_layout_id:
_fail("reviewed amount pair movement layout does not match")
if len(line_id_signals) != 1:
_fail("reviewed amount pair movement line ID is missing")
line_id = _line_id(
line_id_signals[0][1],
label="reviewed amount pair movement line ID",
)
if line_id != reviewed_member.value.movement_line_id:
_fail("reviewed amount pair movement line ID changed")
if len(account_signals) != 1:
_fail("reviewed amount pair movement account is ambiguous or missing")
if amount_signal_count > 1:
_fail("reviewed amount pair movement has ordinary amount ambiguity")
locator_key = _reviewed_amount_locator_key(reviewed_member.value.amount_locator)
if ordinary_amount_locator_keys and ordinary_amount_locator_keys != [
locator_key
]:
_fail("reviewed amount pair movement has an undeclared ordinary " "amount")
if row.row_number in state.reviewed_amount_pair_rows_seen:
_fail("reviewed amount pair movement row was consumed twice")
if locator_key in state.reviewed_amount_locator_keys_consumed:
_fail("reviewed amount locator was consumed twice")
account = _account_code(
account_signals[0][1],
pattern=state.contract.account_code_pattern,
label="reviewed amount pair movement account code",
)
zero = Decimal(0)
state.reviewed_amount_pair_rows_seen.add(row.row_number)
state.reviewed_amount_locator_keys_consumed.add(locator_key)
state.add_movement(
line_id=line_id,
account_code=account,
debit=reviewed_member.amount if reviewed_member.role == "debit" else zero,
credit=(
reviewed_member.amount if reviewed_member.role == "credit" else zero
),
source_form="physical_row",
)
return
if reviewed_exclusion is not None:
if layout.layout_id != reviewed_exclusion.layout_id:
_fail("reviewed amountless exclusion layout does not match")
if tuple(sorted(row.cells)) != reviewed_exclusion.nonempty_columns:
_fail("reviewed amountless exclusion nonempty columns changed")
if any("\n" in value or "\r" in value for value in row.cells.values()):
_fail("reviewed amountless exclusion contains multiline text")
if len(line_id_signals) != 1:
_fail("reviewed amountless exclusion line ID is ambiguous or missing")
line_id = _line_id(
line_id_signals[0][1],
label="reviewed amountless exclusion line ID",
)
if line_id != reviewed_exclusion.line_id:
_fail("reviewed amountless exclusion line ID changed")
if len(account_signals) != 1:
_fail("reviewed amountless exclusion account is ambiguous or missing")
if amount_signal_count != 0:
_fail("reviewed amountless exclusion contains an amount")
if any(
state.contract.logical_candidate_pattern.search(value) is not None
for value in row.cells.values()
):
_fail("reviewed amountless exclusion contains a logical amount")
signal_columns = {
line_id_signals[0][0],
account_signals[0][0],
}
expected_signal_columns = set(reviewed_exclusion.nonempty_columns) - set(
reviewed_exclusion.residual_columns
)
if signal_columns != expected_signal_columns:
_fail("reviewed amountless exclusion residual columns changed")
_account_code(
account_signals[0][1],
pattern=state.contract.account_code_pattern,
label="reviewed amountless exclusion account code",
)
state.exclude_reviewed_amountless_row(
row_number=row.row_number,
line_id=line_id,
)
return
if any(
key in state.reviewed_amount_pairs.locator_keys
for key in ordinary_amount_locator_keys
):
_fail("reviewed amount locator reached ordinary physical extraction")
reviewed_zero_line_signals = [
signal
for signal in line_id_signals
if int(signal[1]) in state.contract.reviewed_zero_amount_line_ids
]
is_candidate = (
bool(account_signals)
or bool(line_id_signals and amount_signal_count)
or bool(reviewed_zero_line_signals)
)
if not is_candidate:
return
if len(line_id_signals) != 1:
_fail(
"physical movement candidate has an ambiguous or missing line ID "
f"at row {row.row_number}"
)
line_id = _line_id(
line_id_signals[0][1],
label=f"physical movement line ID at row {row.row_number}",
)
if len(account_signals) != 1:
_fail(
"physical movement candidate has an ambiguous or missing account "
f"at row {row.row_number}"
)
account_column, account_text = account_signals[0]
if account_column not in layout.account_columns:
_fail("physical movement account lies outside the reviewed columns")
account = _account_code(
account_text,
pattern=state.contract.account_code_pattern,
label=f"physical movement account code at row {row.row_number}",
)
if line_id in state.contract.reviewed_zero_amount_line_ids:
if amount_signal_count != 0:
_fail("reviewed zero-amount line ID contains an amount")
state.add_reviewed_zero_amount_movement(
line_id=line_id,
account_code=account,
)
return
if amount_signal_count != 1:
_fail(
"physical movement candidate has an ambiguous or missing amount "
f"at row {row.row_number}"
)
zero = Decimal(0)
if embedded_matches:
item, match = embedded_matches[0]
debit_text = match.group("debit")
credit_text = match.group("credit")
if (debit_text is None) == (credit_text is None):
_fail(
"reviewed physical embedded amount must contain exactly one "
"debit or credit amount"
)
if debit_text is not None:
debit = _decimal(
debit_text,
decimal_format=item.amount_format,
sign_policy=state.contract.value.amount_sign_policy,
label=("physical embedded debit amount at row " f"{row.row_number}"),
)
credit = zero
else:
debit = zero
credit = _decimal(
credit_text,
decimal_format=item.amount_format,
sign_policy=state.contract.value.amount_sign_policy,
label=("physical embedded credit amount at row " f"{row.row_number}"),
)
else:
amount_text = debit_signals[0][1] if debit_signals else credit_signals[0][1]
amount = _decimal(
amount_text,
decimal_format=state.contract.value.physical_amount_format,
sign_policy=state.contract.value.amount_sign_policy,
label=f"physical movement amount at row {row.row_number}",
)
if debit_signals:
debit = amount
credit = zero
else:
debit = zero
credit = amount
state.add_movement(
line_id=line_id,
account_code=account,
debit=debit,
credit=credit,
source_form="physical_row",
)
def _parse_rows(
rows: tuple[_SheetRow, ...],
*,
contract: _CompiledContract,
) -> _ParserState:
reviewed_amount_pairs = _resolve_reviewed_amount_pairs(
rows,
contract=contract,
)
reviewed_amountless_exclusions = _resolve_reviewed_amountless_exclusions(
rows,
contract=contract,
)
state = _ParserState(
contract,
reviewed_amount_pairs,
reviewed_amountless_exclusions,
)
skip_row_number: int | None = None
for index, row in enumerate(rows):
if skip_row_number == row.row_number:
skip_row_number = None
continue
next_row = rows[index + 1] if index + 1 < len(rows) else None
layout = _match_layout(
row,
next_row,
contract=contract.value,
)
if layout is not None:
state.current_layout = layout
state.page_header_count += 1
state.layout_page_counts[layout.layout_id] = (
state.layout_page_counts.get(layout.layout_id, 0) + 1
)
if next_row is None:
_fail("page header is missing its second header row")
skip_row_number = next_row.row_number
continue
physical_row, before_physical, after_physical = _partition_multiline_cells(
row, state=state
)
for line in before_physical:
_parse_logical_line(line, state=state)
if physical_row is not None:
for line in physical_row.cells.values():
if _record_control_line(line, state=state):
break
else:
parsed_date = _physical_date(physical_row, state=state)
if parsed_date is not None:
state.update_date(parsed_date)
_physical_movement(physical_row, state=state)
for line in after_physical:
_parse_logical_line(line, state=state)
return state
def _finish(
state: _ParserState,
*,
source_sha256: str,
) -> ParsedGeneralJournal:
if state.page_header_count == 0:
_fail("reviewed worksheet contains no recognized page headers")
if not state.movements:
_fail("reviewed worksheet contains no parsed movements")
if state.current_date is None:
_fail("reviewed worksheet contains no parsed posting date")
if state.control_totals is None:
_fail("reviewed final control line was not found")
if (
state.reviewed_amount_locator_keys_consumed
!= state.reviewed_amount_pairs.locator_keys
):
_fail("reviewed amount locators were not consumed exactly once")
if state.reviewed_amount_pair_rows_seen != set(
state.reviewed_amount_pairs.members_by_row
):
_fail("reviewed amount pair movement rows were not consumed exactly once")
if state.reviewed_amountless_exclusion_rows_seen != set(
state.reviewed_amountless_exclusions.by_row
):
_fail("reviewed amountless exclusion rows were not consumed exactly once")
if (
state.reviewed_zero_amount_line_ids_seen
!= state.contract.reviewed_zero_amount_line_ids
):
_fail("reviewed zero-amount line IDs were not encountered exactly once")
source_debit, source_credit = state.control_totals
if (
source_debit != state.contract.reviewed_debit_total
or source_credit != state.contract.reviewed_credit_total
):
_fail("source final controls do not match the exact reviewed controls")
debit_total = _exact_decimal_sum(
tuple(movement.debit for movement in state.movements),
label="parsed debit total",
)
credit_total = _exact_decimal_sum(
tuple(movement.credit for movement in state.movements),
label="parsed credit total",
)
if (
debit_total != state.contract.reviewed_debit_total
or credit_total != state.contract.reviewed_credit_total
):
_fail("parsed movements do not reconcile to the exact reviewed controls")
dates = [movement.posting_date for movement in state.movements]
return ParsedGeneralJournal(
source_sha256=source_sha256,
movements=tuple(state.movements),
debit_total=debit_total,
credit_total=credit_total,
source_control_debit_total=source_debit,
source_control_credit_total=source_credit,
first_posting_date=min(dates),
last_posting_date=max(dates),
page_header_count=state.page_header_count,
physical_movement_count=state.physical_movement_count,
logical_movement_count=state.logical_movement_count,
excluded_amountless_count=len(state.reviewed_amountless_exclusion_rows_seen),
line_id_gap_count=state.line_id_gap_count,
layout_page_counts=MappingProxyType(dict(state.layout_page_counts)),
)
def parse_commercial_general_journal(
source_path: Path,
*,
expected_source_sha256: str,
layout_contract: GeneralJournalLayoutContract,
) -> ParsedGeneralJournal:
"""Parse one digest-bound workbook under an explicit reviewed layout.
No source kind, entity, currency, unit, sign convention, or account mapping
is selected here. A successful result proves only the declared parser
mechanics and exact control reconciliation.
"""
compiled = _compile_contract(layout_contract)
payload = _stable_source_bytes(
Path(source_path),
expected_sha256=expected_source_sha256,
)
rows = _read_rows(payload, sheet_name=compiled.value.sheet_name)
state = _parse_rows(rows, contract=compiled)
return _finish(state, source_sha256=expected_source_sha256)
SHA-256: 9ee7b702f494bca21193216afdc889f181a87cff6becd6124d29e2b5c3427b49