Finance-Accounts/ar-aging-app/backend/app/core/compare.py

126 lines
4.7 KiB
Python

"""
Compare the engine's output against the sample Accounts Receivable Aging workbook.
Reads the sample's *cached* cell values (Excel stores the last-computed result next to
each formula) without evaluating anything, and diffs them against the engine result to
produce an expected/generated/difference/status report.
"""
from __future__ import annotations
import re
import zipfile
from dataclasses import dataclass
from xml.etree import ElementTree as ET
from .pipeline import ProcessResult
NS = "{http://schemas.openxmlformats.org/spreadsheetml/2006/main}"
_COL_RE = re.compile(r"^([A-Z]+)")
@dataclass
class DiffRow:
sheet: str
cell: str
label: str
expected: float | str | None # from sample workbook
generated: float | str | None # from engine
difference: float | None
status: str # MATCH | DIFF | INFO
def _sheet_part(z: zipfile.ZipFile, sheet_name: str) -> str | None:
wb = z.read("xl/workbook.xml").decode("utf-8", "replace")
rels = z.read("xl/_rels/workbook.xml.rels").decode("utf-8", "replace")
rid_target: dict[str, str] = {}
for rel in re.findall(r"<Relationship\b[^>]*/?>", rels):
i = re.search(r'Id="([^"]+)"', rel)
t = re.search(r'Target="([^"]+)"', rel)
if i and t:
rid_target[i.group(1)] = t.group(1)
for s in re.findall(r"<sheet\b[^>]*/?>", wb):
nm = re.search(r'name="([^"]*)"', s)
rid = re.search(r'r:id="([^"]*)"', s)
if nm and rid and nm.group(1) == sheet_name:
tgt = rid_target.get(rid.group(1), "")
if not tgt:
return None
return tgt.lstrip("/") if tgt.startswith("/") else "xl/" + tgt
return None
def read_cached_cells(path: str, sheet_name: str, refs: set[str]) -> dict[str, float | str | None]:
"""Return {cell_ref: cached numeric/text value} for the requested cells."""
z = zipfile.ZipFile(path)
part = _sheet_part(z, sheet_name)
out: dict[str, float | str | None] = {r: None for r in refs}
if part is None:
return out
want_rows = {int(_row_of(r)) for r in refs}
max_row = max(want_rows)
with z.open(part) as fh:
for _ev, el in ET.iterparse(fh, events=("end",)):
if el.tag != NS + "row":
continue
rn = el.get("r")
if rn and int(rn) in want_rows:
for c in el:
ref = c.get("r")
if ref in refs:
v = c.find(NS + "v")
if v is not None and v.text is not None:
try:
out[ref] = float(v.text)
except ValueError:
out[ref] = v.text
el.clear()
if rn and int(rn) > max_row:
break
return out
def _row_of(ref: str) -> str:
return re.sub(r"^[A-Z]+", "", ref)
def list_sheets(path: str) -> list[str]:
z = zipfile.ZipFile(path)
wb = z.read("xl/workbook.xml").decode("utf-8", "replace")
return [re.search(r'name="([^"]*)"', s).group(1) for s in re.findall(r"<sheet\b[^>]*/?>", wb)]
# Sample workbook benchmark cells (USA block on the Detail sheet).
SAMPLE_DETAIL_CELLS = {
"D11": "USA receivable (local, rounded)",
"C10": "USA Standard additional sales",
"B10": "USA Invoiced additional sales",
"C9": "USA Standard Net Closing Balance (reserve)",
"D24": "USA subtotal ($)",
}
def compare_to_sample(result: ProcessResult, sample_path: str,
tolerance: float = 1.0) -> list[DiffRow]:
"""Diff the engine's USA figures against the sample workbook's cached Detail cells."""
cells = read_cached_cells(sample_path, "Detail", set(SAMPLE_DETAIL_CELLS))
rec = result.receivable
usa = rec.marketplaces.get("USA")
gen = {
"D11": float(usa.receivable_local) if usa else None,
"C10": usa.accounts.get("Standard Orders").additional_sales if usa and "Standard Orders" in usa.accounts else None,
"B10": usa.accounts.get("Invoiced Orders").additional_sales if usa and "Invoiced Orders" in usa.accounts else None,
"C9": usa.accounts.get("Standard Orders").reserve if usa and "Standard Orders" in usa.accounts else None,
"D24": float(usa.receivable_usd) if usa else None,
}
rows: list[DiffRow] = []
for ref, label in SAMPLE_DETAIL_CELLS.items():
exp = cells.get(ref)
g = gen.get(ref)
diff = None
status = "INFO"
if isinstance(exp, (int, float)) and isinstance(g, (int, float)):
diff = g - exp
status = "MATCH" if abs(diff) <= tolerance else "DIFF"
rows.append(DiffRow("Detail", ref, label, exp, g, diff, status))
return rows