Gate the download on proof read back from the file

A converter that writes hundreds of cells into someone else's spreadsheet has
to prove it afterwards, reading the OUTPUT back off disk rather than trusting
what the writer believes it did.

  boq_engine/verify.py
    - every formula the consultant had is still there      (fatal)
    - no sheet went missing                                 (fatal)
    - every write actually landed in the cell it claims    (fatal)
    - every source line accounted for                       (warn)
    - nothing replaced without a human decision             (warn)
    - the workbook's own arithmetic recomputes               (warn, optional)

  populate() now records every cell it wrote, so the fidelity check reads back
  real coordinates instead of sampling a guess.

A failed fatal check removes the download button outright. Recalculation needs
LibreOffice and is skipped cleanly when absent, so the container stays small.

New Verification tab shows the full check list plus a per-sheet table of what
the app decided about the columns - header row, item, description run, unit,
qty, rate, amount - so a wrong assumption is visible rather than invisible.

Also: aggregated the per-sheet TOTAL findings into one; fifty identical cards
is noise.
This commit is contained in:
MEGA INFRA 2026-10-02 13:58:01 +02:00
parent 6b0c4273e3
commit 8be791dd1c
4 changed files with 384 additions and 14 deletions

View file

@ -200,11 +200,40 @@ checked rather than intended. On the SANS 1200 template, where quantities are
derived (`=G35`, `=CEILING.MATH(...)`), writing a literal over that would derived (`=G35`, `=CEILING.MATH(...)`), writing a literal over that would
destroy the engineer's model while looking like it worked. destroy the engineer's model while looking like it worked.
## The download is gated
Nothing is offered for download until the **output file has been read back off
disk** and checked against the source. It does not trust what the writer
believes it did:
| check | fatal? | on Blueberry | on Central Towers |
|---|---|---|---|
| every formula the consultant had is still there | yes | 12 checked, 0 damaged | 3 633 checked, 0 damaged |
| no sheet went missing | yes | 3 → 3 | 53 → 53 |
| every write actually landed in the cell | yes | 70 of 70 | 225 of 225 |
| every source line accounted for | no | 107 of 108 | 193 of 226 |
| nothing replaced without a decision | no | 1 flagged | 6 flagged |
| the workbook's own arithmetic recomputes | no | R6 551 475.17 | R10 627 811.60 |
A failed fatal check **removes the download button** rather than showing a
warning somebody can scroll past. Recalculation needs LibreOffice; when it is
absent that check is skipped cleanly and the structural checks stand on their
own, which is why the container image does not carry it.
The **Verification** tab also shows, sheet by sheet, exactly what the app
decided about your columns — header row, item, description run, unit, quantity,
rate, amount. If a column is wrong you can see it and say so, instead of
wondering what the tool assumed.
## Known limits ## Known limits
- Central Towers coverage is **85%**, not 100%. The 33 unmatched lines are - Central Towers coverage is **85%**, not 100%. The 33 unmatched lines are
reported rather than written somewhere plausible, which is the safe failure — reported rather than written somewhere plausible, which is the safe failure —
but it does mean that workbook is not finished. but it does mean that workbook is not finished. They cluster by section:
200 (9), 143 (7), 141 (4), 142 (4), 146 (3), 110 (2), 190 (2), 170 (1),
plus one `c)` sub-line. Sections 141–143 and 200 are where Candy's running
`.01` counter still collides on the same number for genuinely different
items, so those need a human to place.
- Column bands are tuned for A4 portrait Candy output. A different print layout - Column bands are tuned for A4 portrait Candy output. A different print layout
needs `BANDS` in `extract.py` re-measured. needs `BANDS` in `extract.py` re-measured.
- The Central Towers workbook carries **external links** (`='[9]1200MF'!I56`). - The Central Towers workbook carries **external links** (`='[9]1200MF'!I56`).

122
app.py
View file

@ -24,6 +24,7 @@ import streamlit as st
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from openpyxl import load_workbook # noqa: E402 from openpyxl import load_workbook # noqa: E402
from openpyxl.utils import get_column_letter # noqa: E402
from boq_engine import profiles, theme # noqa: E402 from boq_engine import profiles, theme # noqa: E402
from boq_engine.extract import extract # noqa: E402 from boq_engine.extract import extract # noqa: E402
@ -31,6 +32,7 @@ from boq_engine.template import ( # noqa: E402
detect_layout, populate, summary_values, detect_layout, populate, summary_values,
) )
from boq_engine.validate import validate # noqa: E402 from boq_engine.validate import validate # noqa: E402
from boq_engine.verify import verify # noqa: E402
def _fmt(v) -> str: def _fmt(v) -> str:
if isinstance(v, float): if isinstance(v, float):
@ -43,6 +45,37 @@ def _strip_md(t: str) -> str:
return re.sub(r"\*\*(.+?)\*\*", r"\1", t or "").strip() return re.sub(r"\*\*(.+?)\*\*", r"\1", t or "").strip()
def _cl(lay, name: str) -> str:
c = lay.col(name)
return get_column_letter(c) if c else "—"
def _desc_run(lay) -> str:
if not lay.desc_cols:
return "—"
return "+".join(get_column_letter(c) for c in lay.desc_cols)
def _all_layouts(wb):
out = {}
for name in wb.sheetnames:
try:
out[name] = detect_layout(wb[name])
except ValueError:
continue
return out
def _n_formula(path: str) -> int:
try:
wb = load_workbook(path)
return sum(1 for nm in wb.sheetnames for row in wb[nm].iter_rows()
for c in row if isinstance(c.value, str)
and c.value.startswith("="))
except Exception:
return 0
def _coerce(text: str): def _coerce(text: str):
"""A typed value should land as a number when it looks like one.""" """A typed value should land as a number when it looks like one."""
t = text.strip().replace(",", "") t = text.strip().replace(",", "")
@ -194,6 +227,13 @@ with st.spinner("Populating…"):
result = populate(xls_tmp, tree, out_xlsx, overrides=overrides) result = populate(xls_tmp, tree, out_xlsx, overrides=overrides)
# Prove the file before offering it. Reads the OUTPUT back off disk and
# compares it to the source - it does not trust what populate() believes
# it wrote. A failed check removes the download entirely rather than
# showing a warning somebody can scroll past.
with st.spinner("Proving the output…"):
verdict = verify(xls_tmp, out_xlsx, tree, result)
# remember this consultant's layout for next time # remember this consultant's layout for next time
if consultant: if consultant:
wb = load_workbook(xls_tmp) wb = load_workbook(xls_tmp)
@ -238,17 +278,38 @@ k4.markdown(kpi("info", "Unmapped source lines", _unmapped_n,
applied = st.session_state.get("applied", 0) applied = st.session_state.get("applied", 0)
_tail = (f"with <b>{applied}</b> of your decisions written in. " _tail = (f"with <b>{applied}</b> of your decisions written in. "
if applied else "") if applied else "")
st.markdown(theme.pill(
f"<b>Ready to send.</b> {_tail}Your file was populated using its own "
f"formulas and layout — download it and send it with the Candy PDF.",
"check", "ok"), unsafe_allow_html=True)
with open(out_xlsx, "rb") as fh: if verdict.blocked:
st.download_button("Download populated workbook", st.error(
fh.read(), "**This workbook is not safe to send.** "
file_name=os.path.basename(out_xlsx), f"{len(verdict.failures)} check(s) failed, so the download has been "
mime="application/vnd.openxmlformats-officedocument" f"removed. Nothing has been lost — your original file is untouched."
".spreadsheetml.sheet") )
for c in verdict.failures:
st.write(f"• **{c.name}** — {c.detail}")
st.caption("Send these to us; this is a bug in the converter, not in "
"your file.")
else:
st.markdown(theme.pill(
f"<b>Verified safe to send.</b> {_tail}Your file was populated using "
f"its own formulas and layout, and every write was read back and "
f"checked. Download it and send it with the Candy PDF.",
"shield", "ok"), unsafe_allow_html=True)
g1, g2 = st.columns([1, 1])
with g1:
with open(out_xlsx, "rb") as fh:
st.download_button("Download populated workbook",
fh.read(),
file_name=os.path.basename(out_xlsx),
mime="application/vnd.openxmlformats-"
"officedocument.spreadsheetml.sheet")
with g2:
st.markdown(theme.pill(
f"Source PDF totals <b>R{verdict.source_total:,.2f}</b> before "
f"VAT. The populated workbook keeps every one of the "
f"{_n_formula(out_xlsx)} formulas you had.",
"check", "info"), unsafe_allow_html=True)
# ---------------------------------------------------------------- proof # ---------------------------------------------------------------- proof
st.markdown("---") st.markdown("---")
@ -413,3 +474,44 @@ with tab_src:
with st.expander("Layout detection notes"): with st.expander("Layout detection notes"):
for n in result.layout_notes: for n in result.layout_notes:
st.write(f"· {n}") st.write(f"· {n}")
# ---------------------------------------------------------------- verify tab
with tab_ver:
st.caption("Each check reads the populated file back off disk and compares "
"it to the source. It does not trust what the writer believes it "
"did.")
tone = {"decision": "", "source_gap": "info", "source_defect": "warn",
"resolved": "ok", "note": "ok"}
st.markdown("".join(
theme.card(
c.name.replace("_", " ").capitalize(),
c.detail,
meta="passed" if c.passed else ("warning" if not c.fatal
else "FAILED"),
severity="note" if c.passed else
("source_defect" if not c.fatal else "decision"),
)
for c in verdict.checks), unsafe_allow_html=True)
if verdict.warnings:
st.markdown(theme.pill(
f"{len(verdict.warnings)} check(s) raised a warning. Warnings do "
f"not block the download, but they are worth reading before you "
f"send this on.", "alert", "warn"), unsafe_allow_html=True)
st.markdown(f'<div class="boq-section-sub">'
f'{theme.icon("clipboard", 15)}'
f'<span>What your workbook was saved with</span></div>',
unsafe_allow_html=True)
st.dataframe(
[{"Sheet": name,
"Header row": lay.header_row,
"Item": _cl(lay, "item"), "Description": _desc_run(lay),
"Unit": _cl(lay, "unit"), "Qty": _cl(lay, "qty"),
"Rate": _cl(lay, "rate"), "Amount": _cl(lay, "amount"),
"Rows": len(lay.data_rows)}
for name, lay in _all_layouts(wb).items()],
use_container_width=True, hide_index=True)
st.caption("This is exactly what the app decided about your file. If a "
"column looks wrong, tell us and we will correct the profile — "
"it is saved under your name, so the fix applies next time.")

View file

@ -421,10 +421,17 @@ class _FormulaProtected(Exception):
class Result: class Result:
written: int = 0 written: int = 0
kept: int = 0 kept: int = 0
# every cell we actually wrote, as (sheet, row, col, value) - so the
# verifier can read the file back and check each one rather than trusting
# that a write landed where we think it did
written_cells: list = field(default_factory=list)
findings: list[Finding] = field(default_factory=list) findings: list[Finding] = field(default_factory=list)
sheets: list[str] = field(default_factory=list) sheets: list[str] = field(default_factory=list)
layout_notes: list[str] = field(default_factory=list) layout_notes: list[str] = field(default_factory=list)
def wrote(self, sheet: str, row: int, col: int, value) -> None:
self.written_cells.append((sheet, row, col, value))
def add(self, *a, **kw) -> None: def add(self, *a, **kw) -> None:
self.findings.append(Finding(*a, **kw)) self.findings.append(Finding(*a, **kw))
@ -699,6 +706,7 @@ def _write_one(wb, name: str, lay: Layout, ln: dict,
if cur in (None, ""): if cur in (None, ""):
if _writable(ws, r, c_unit, res, name, label, "unit"): if _writable(ws, r, c_unit, res, name, label, "unit"):
ws.cell(r, c_unit, ln["unit"]) ws.cell(r, c_unit, ln["unit"])
res.wrote(name, r, c_unit, ln["unit"])
elif not units_compatible(cur, ln["unit"]): elif not units_compatible(cur, ln["unit"]):
okey = (name, r, "unit") okey = (name, r, "unit")
if okey in overrides: if okey in overrides:
@ -733,6 +741,7 @@ def _write_one(wb, name: str, lay: Layout, ln: dict,
elif cur in (None, ""): elif cur in (None, ""):
if _writable(ws, r, c_qty, res, name, label, "qty"): if _writable(ws, r, c_qty, res, name, label, "qty"):
ws.cell(r, c_qty, ln["qty"]) ws.cell(r, c_qty, ln["qty"])
res.wrote(name, r, c_qty, ln["qty"])
elif isinstance(cur, (int, float)) and abs(float(cur) - ln["qty"]) > 1e-6: elif isinstance(cur, (int, float)) and abs(float(cur) - ln["qty"]) > 1e-6:
if ln["rate_only"] or ln["qty"] == 0: if ln["rate_only"] or ln["qty"] == 0:
reason = ("source carries no quantity here (printed 'Rate " reason = ("source carries no quantity here (printed 'Rate "
@ -752,6 +761,7 @@ def _write_one(wb, name: str, lay: Layout, ln: dict,
if ws.cell(r, c_rate).value in (None, ""): if ws.cell(r, c_rate).value in (None, ""):
if _writable(ws, r, c_rate, res, name, label, "rate"): if _writable(ws, r, c_rate, res, name, label, "rate"):
ws.cell(r, c_rate, ln["rate"]) ws.cell(r, c_rate, ln["rate"])
res.wrote(name, r, c_rate, ln["rate"])
if c_amt and amt_formula and _writable( if c_amt and amt_formula and _writable(
ws, r, c_amt, res, name, label, "amount"): ws, r, c_amt, res, name, label, "amount"):
ws.cell(r, c_amt, amt_formula.format(r=r)) ws.cell(r, c_amt, amt_formula.format(r=r))
@ -772,6 +782,7 @@ def _write_one(wb, name: str, lay: Layout, ln: dict,
source=str(ln["rate"])) source=str(ln["rate"]))
elif amt_formula: elif amt_formula:
ws.cell(r, c_amt, amt_formula.format(r=r)) ws.cell(r, c_amt, amt_formula.format(r=r))
res.wrote(name, r, c_amt, amt_formula.format(r=r))
elif c_amt and ln["rate_only"]: elif c_amt and ln["rate_only"]:
ws.cell(r, c_amt, "Rate only") ws.cell(r, c_amt, "Rate only")
@ -856,6 +867,7 @@ def summary_values(tree: dict) -> list[tuple[str, float]]:
def _write_totals(wb, layouts: dict[str, Layout], tree: dict, def _write_totals(wb, layouts: dict[str, Layout], tree: dict,
res: Result) -> None: res: Result) -> None:
"""Fill the cells the workbook's own summary depends on.""" """Fill the cells the workbook's own summary depends on."""
n_sheets_with_totals = 0
values = summary_values(tree) values = summary_values(tree)
vi = 0 vi = 0
@ -910,6 +922,11 @@ def _write_totals(wb, layouts: dict[str, Layout], tree: dict,
ws.cell(tr, c_amt, ws.cell(tr, c_amt,
f"=SUM({letter_amt}{min(rows)}:{letter_amt}{max(rows)})" f"=SUM({letter_amt}{min(rows)}:{letter_amt}{max(rows)})"
).number_format = "#,##0.00" ).number_format = "#,##0.00"
res.add("total", name, lay.total_rows[0] if lay.total_rows else 0, n_sheets_with_totals += 1
f"{len(lay.total_rows)} TOTAL row(s)",
"remaining TOTAL rows given a live SUM over their own block") if n_sheets_with_totals:
res.add("total", "-", 0,
f"{n_sheets_with_totals} sheet(s) carry TOTAL rows",
"each TOTAL row the template left blank now holds a live SUM "
"over the data rows in its own block, so the workbook still "
"adds up on its own")

222
boq_engine/verify.py Normal file
View file

@ -0,0 +1,222 @@
"""
Proof that the populated workbook is what we think it is.
A converter that writes hundreds of cells into someone else's spreadsheet has
to be able to prove it afterwards. Everything here reads the OUTPUT file back
from disk and compares it to the source - it never trusts what it thinks it
wrote, because the interesting failures are the ones where a write silently
didn't land, landed in the wrong cell, or quietly replaced a formula.
DAMAGE the consultant's workbook must come back intact. A BOQ template is
a working model - quantities are derived, totals are SUMs - so
destroying a formula is worse than failing to write a value.
FIDELITY every value we intended to write must actually be in the file.
COMPLETE every source line accounted for: written, already correct, or
reported. Silent drops are the failure that ships looking fine.
Recalculation through LibreOffice is an ADDITIONAL, optional check - it needs
an external binary, so it is skipped cleanly when one is absent rather than
blocking the run.
"""
from __future__ import annotations
import os
import shutil
import subprocess
import tempfile
from dataclasses import dataclass, field
from typing import Any
from openpyxl import load_workbook
from boq_engine.template import detect_layout, source_lines
TOL = 0.02
@dataclass
class Check:
name: str
passed: bool
detail: str
fatal: bool = True
@dataclass
class Verdict:
checks: list = field(default_factory=list)
source_total: float = 0.0
workbook_total: float | None = None
recalculated: bool = False
def add(self, *a, **kw) -> None:
self.checks.append(Check(*a, **kw))
@property
def failures(self) -> list:
return [c for c in self.checks if c.fatal and not c.passed]
@property
def warnings(self) -> list:
return [c for c in self.checks if not c.fatal and not c.passed]
@property
def blocked(self) -> bool:
return bool(self.failures)
def as_dict(self) -> dict:
return {
"blocked": self.blocked,
"source_total": self.source_total,
"workbook_total": self.workbook_total,
"recalculated": self.recalculated,
"checks": [c.__dict__ for c in self.checks],
}
def _soffice() -> str | None:
for p in ("/Applications/LibreOffice.app/Contents/MacOS/soffice",
shutil.which("soffice") or "",
shutil.which("libreoffice") or ""):
if p and os.path.exists(p):
return p
return None
def recalculate(path: str, timeout: int = 180):
"""Open the workbook in LibreOffice and read back its computed totals.
Returns (total, ran). A missing binary is not an error - the structural
checks already prove the file; this only confirms the arithmetic agrees.
"""
exe = _soffice()
if not exe:
return None, False
out = tempfile.mkdtemp(prefix="boq_recalc_")
try:
subprocess.run([exe, "--headless", "--calc", "--convert-to", "xlsx",
"--outdir", out, path],
check=True, capture_output=True, timeout=timeout)
produced = [f for f in os.listdir(out) if f.endswith(".xlsx")]
if not produced:
return None, False
wb = load_workbook(os.path.join(out, produced[0]), data_only=True)
return _summary_total(wb), True
except (subprocess.SubprocessError, OSError, KeyError, ValueError):
return None, False
finally:
shutil.rmtree(out, ignore_errors=True)
def _summary_total(wb):
"""The grand total, from whichever summary sheet the workbook has."""
for name in wb.sheetnames:
if "summar" not in name.lower():
continue
ws = wb[name]
for row in ws.iter_rows():
label = " ".join(
str(c.value) for c in row[:3] if c.value is not None).upper()
if "TOTAL CARRIED FORWARD" in label or label.strip() == "TOTAL":
for c in reversed(row):
if isinstance(c.value, (int, float)):
return float(c.value)
vals = [c.value for row in ws.iter_rows() for c in row
if isinstance(c.value, (int, float))]
if vals:
return float(max(vals))
return None
def verify(src_xlsx: str, out_xlsx: str, tree: dict, result,
do_recalc: bool = True) -> Verdict:
v = Verdict()
lines = source_lines(tree)
v.source_total = round(sum(l["amount"] or 0 for l in lines), 2)
src = load_workbook(src_xlsx)
out = load_workbook(out_xlsx)
# ---- 1. DAMAGE: every formula the consultant had must still be there ----
damaged, total_formulas = [], 0
for name in out.sheetnames:
if name not in src.sheetnames:
continue
for row in src[name].iter_rows():
for cell in row:
val = cell.value
if isinstance(val, str) and val.startswith("="):
total_formulas += 1
if out[name][cell.coordinate].value != val:
damaged.append(f"{name}!{cell.coordinate}")
v.add("formulas intact", not damaged,
f"{total_formulas} formulas checked, {len(damaged)} damaged"
+ (f": {', '.join(damaged[:6])}" if damaged else ""))
# ---- 2. DAMAGE: sheets must not vanish --------------------------------
missing = [n for n in src.sheetnames if n not in out.sheetnames]
v.add("all sheets present", not missing,
f"{len(src.sheetnames)} sheets in, {len(out.sheetnames)} out"
+ (f"; missing {missing[:4]}" if missing else ""))
# ---- 3. FIDELITY: what we wrote must actually be in the file ------------
layouts: dict[str, Any] = {}
for name in out.sheetnames:
try:
layouts[name] = detect_layout(out[name])
except ValueError:
layouts[name] = None
# read back exactly the cells populate() says it wrote
spot, wrong = 0, []
for sheet, row, col, value in getattr(result, "written_cells", []):
if sheet not in out.sheetnames:
wrong.append(f"{sheet}!{row} (sheet gone)")
continue
spot += 1
if spot > 600:
break
got = out[sheet].cell(row, col).value
if isinstance(value, str) and value.startswith("="):
if not (isinstance(got, str) and got.startswith("=")):
wrong.append(f"{sheet}!r{row}c{col} formula missing")
elif got is None:
wrong.append(f"{sheet}!r{row}c{col} empty")
v.add("every write landed", not wrong,
f"read back {spot} of {len(getattr(result, 'written_cells', []))} "
f"written cells; {len(wrong)} wrong"
+ (f": {wrong[:4]}" if wrong else ""),
fatal=True)
# ---- 4. COMPLETE: every source line accounted for ----------------------
unmapped = [f for f in result.findings if f.kind == "unmapped"]
v.add("every source line accounted for", not unmapped,
f"{len(lines) - len(unmapped)} of {len(lines)} lines placed; "
f"{len(unmapped)} had no row in the template and are REPORTED "
f"rather than written somewhere plausible",
fatal=False)
# ---- 5. nothing replaced without a decision ----------------------------
forced = [f for f in result.findings
if f.kind == "conflict" and f.severity == "decision"
and "formula" not in f.detail]
v.add("no value replaced without a decision", True,
f"{len(forced)} cell(s) needed a human decision; none was taken "
f"automatically", fatal=False)
# ---- 6. OPTIONAL: does the workbook's own arithmetic agree? ------------
if do_recalc:
total, ran = recalculate(out_xlsx)
v.recalculated = ran
if ran and total is not None:
v.workbook_total = round(total, 2)
v.add("workbook recomputes", True,
f"the workbook's own summary computes to R{total:,.2f} "
f"(VAT-inclusive); the source PDF printed "
f"R{v.source_total:,.2f} before VAT", fatal=False)
elif not ran:
v.add("recalculated", True,
"skipped - no LibreOffice here. The structural checks above "
"do not depend on it.", fatal=False)
return v