""" 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"]*/?>", 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"]*/?>", 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"]*/?>", 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