David/MEGA-BOQ-WEBAPP
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Handle a structurally different consultant template

Browse source 
Second real pair (Central Towers: 20pp Candy PDF into a 53-sheet SANS 1200
workbook) exposed four failures. All fixed and verified against both pairs.

- item numbering: accept 1.1.1, 110...., .01 and rebuild the running counter
  Candy restarts under each repeated heading. 0% -> 86% of lines numbered.
- wrapped headers: join 'QUAN-' + 'TITY'. qty was missing on 49 of 51 sheets.
- description is a RUN between ITEM and UNIT, not one column: the header
  claims E, the data sits in D, and it moves within a single sheet.
- carry item numbers down blank cells, and score a missing number as neutral
  rather than as disagreement.

Mapping: 1/108 unmatched (Blueberry), 33/226 (Central Towers).

Safety: the consultant's workbook is never damaged. Per-write guard plus a
final pass that restores any formula disturbed - 0 of 3633 formulas clobbered.
Writing a literal over a derived quantity would have destroyed their model
while looking like it worked.
This commit is contained in:
MEGA INFRA 2026-10-02 13:35:29 +02:00
parent 0ab958ff6a
commit 6b0c4273e3
3 changed files with 602 additions and 159 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

50
README.md
View file

@ -159,13 +159,55 @@ app.py Streamlit UI
run.py CLI: PDF -> xlsx + csv
```
## How it handles a different template
Two real consultant workbooks have been run through it end to end:
| | Blueberry Square | Central Towers |
|---|---|---|
| source PDF | 11pp, 5 sections | 20pp, 18 sections |
| consultant workbook | 3 sheets, Candy-derived | **53 sheets, SANS 1200** |
| source total reproduced | R5,696,934.93 exact | R38,250,252.60 exact |
| lines written | 83 | 182 |
| already correct | 94 | 159 |
| **lines with no home** | **1 of 108** | **33 of 226** |
| consultant formulas destroyed | **0** | **0** |
What it takes to cope with a template nobody has seen before:
- **Item numbering.** Candy prints references three ways - `1.1.1`, `110....`,
and a leading-dot `.01`. All are recognised. A running counter is rebuilt per
section, because Candy restarts `.01` under every repeated heading while the
consultant's template numbers them sequentially.
- **Wrapped headers.** A narrow column splits `QUANTITY` into `QUAN-` / `TITY`.
Both header rows are read and joined, hyphen stripped, or the quantity column
is simply never found.
- **The header lies.** On the SANS 1200 template the header advertises
DESCRIPTION in column E while the text sits in D - and moves *within* one
sheet. So the description is treated as whatever run of columns sits between
ITEM and UNIT, not a single column.
- **Numbers are implied, not repeated.** A blank item cell still belongs to the
number printed above it, so that is carried down. Without it two thirds of
the rows read as unnumbered and nothing can match them.
- **The pool is cut to what matters.** Aligning against all 1314 rows of a
53-sheet workbook makes the aligner cherry-pick; it is restricted to the
sheets that own a section named in the source, leaving 499.
**The hard guarantee: the consultant's workbook is never damaged.** Every cell
that held a formula still holds it. A guard stops each write, and a final pass
re-reads the original and restores anything disturbed, so the promise is
checked rather than intended. On the SANS 1200 template, where quantities are
derived (`=G35`, `=CEILING.MATH(...)`), writing a literal over that would
destroy the engineer's model while looking like it worked.
## Known limits
- Central Towers coverage is **85%**, not 100%. The 33 unmatched lines are
reported rather than written somewhere plausible, which is the safe failure —
but it does mean that workbook is not finished.
- Column bands are tuned for A4 portrait Candy output. A different print layout
needs `BANDS` in `extract.py` re-measured.
- Only one consultant's file has been seen. Alignment is near 1:1 for this one;
a structurally different template may need its own profile tuning.
- Multi-sheet templates are handled one BOQ-shaped sheet at a time; a template
that splits one section across two sheets would need a different strategy.
- The Central Towers workbook carries **external links** (`='[9]1200MF'!I56`).
Those are preserved but will not resolve without the linked file.
- Description joins can drop a space where Candy's print engine split a word
(`Fixed-ChargeItems`). Cosmetic.

146
boq_engine/extract.py
View file

@ -40,7 +40,11 @@ BANDS = {
"amount": (500.0, 575.0),
}
ITEM_X_MAX = 60.0
ITEM_X_MAX = 68.0
# A second, narrower band just right of the item column. Some Candy layouts
# print the sub-item number indented - ".01", ".02" - which is otherwise
# indistinguishable from the start of a description and gets glued onto it.
SUBITEM_X_MAX = 84.0
DESC_X_MIN, DESC_X_MAX = 60.0, 296.0
UNIT_X_MIN, UNIT_X_MAX = 296.0, 380.0
@ -61,6 +65,53 @@ HEADER_TOKENS = {"item", "unit", "rate", "amount", "bill", "quantity"}
SECTION_RE = re.compile(r"^SECTION\s+(\d+)\s*[:\-]\s*(.+)$", re.I)
ITEM_RE = re.compile(r"^\d{1,2}(?:\.{1,2}\d{1,2})*\.?$")
# The other numbering style Candy emits, used by the Bill 1200 BOQs:
# "110..." a section heading, trailing dots
# "110...." the same, deeper heading
# ".01" an item, LEADING dot, subordinate to the section above it
# "110.03.01" a fully qualified item
# Anything made only of digits and dots, short, is treated as a reference.
REFISH = re.compile(r"^[\d.]{1,12}$")
# A sub-item reference, strictly shaped. A leading dot (".01", ".1.2") or a
# fully qualified three-part code ("110.03.01"). A bare "1.5" does NOT match,
# so a description that opens with a measurement is not mistaken for an item.
SUBREF = re.compile(r"^\.[\d.]+$|^\d{2,3}(?:\.\d+){2,}$")
def _is_ref(tok: str) -> bool:
"""A bare section/bill reference: digits and dots only, at least 2 digits.
Strict on purpose. A loose version accepted "2.05p" - the version string
from the "Candy 2.05p" page footer - and it then propagated through the
whole document as if it were a section number.
"""
t = (tok or "").strip()
# A single digit is a valid section number ("1 SCHEDULE A : GENERAL"), and
# the item column is the only thing left of x=68, so there is no risk of
# picking up a quantity. Letters are what disqualify a token: "2.05p" from
# the "Candy 2.05p" footer is digits-and-dots-or-not, and REFISH is the
# rule that rejects it.
return bool(REFISH.match(t))
def _is_subref(tok: str) -> bool:
return bool(SUBREF.match((tok or "").strip()))
def _norm_ref(tok: str) -> str:
"""Normalise any of Candy's reference styles to a dotted, digit-anchored key.
"110...." -> "110" ".01" -> "01" "4..2.5" -> "4.2.5"
"""
t = (tok or "").strip().strip(".")
if not t:
return ""
# collapse runs of dots, then drop a leading dot
while ".." in t:
t = t.replace("..", ".")
return t.strip(".")
LETTER_RE = re.compile(r"^\(?([a-z])\)\s*(.*)$", re.S)
PLACEHOLDER_RE = re.compile(r"^[.\s]{6,}$")
RATE_ONLY_RE = re.compile(r"rate\s*only", re.I)
@ -90,6 +141,7 @@ def _classify(x1: float) -> str | None:
@dataclass
class Row:
item: str = ""
subitem: str = ""
desc: str = ""
unit: str = ""
qty: float | None = None
@ -117,13 +169,27 @@ def _page_rows(page, pno: int) -> list[Row]:
for key in sorted(buckets):
words = sorted(buckets[key], key=lambda z: z["x0"])
r = Row(top=words[0]["top"], page=pno)
# --- pre-pass: lift the reference columns out before anything else.
# Doing this first means a reference can never be swallowed by a wider
# text band further down, which is what left every Bill 1200 line
# unnumbered.
rest: list = []
for w in words:
x0, txt = w["x0"], w["text"]
if x0 < ITEM_X_MAX and _is_ref(txt):
r.item = (r.item + " " + txt).strip()
elif (ITEM_X_MAX <= x0 < SUBITEM_X_MAX and not r.subitem
and _is_subref(txt)):
r.subitem = txt
else:
rest.append(w)
# --- pass 2: the priced columns, then the text
desc_parts: list[str] = []
last_desc_x1 = None
for w in words:
for w in rest:
x0, x1, txt = w["x0"], w["x1"], w["text"]
if x1 <= ITEM_X_MAX:
r.item = (r.item + " " + txt).strip()
continue
col = _classify(x1)
if col == "qty" and r.qty is None:
r.qty = _to_num(txt)
@ -155,9 +221,70 @@ def _page_rows(page, pno: int) -> list[Row]:
return rows
# ==========================================================================
# pass 2 - strip noise, merge continuations
# ==========================================================================
def _resolve_refs(rows: list[Row]) -> list[Row]:
"""Turn Candy's several reference styles into one comparable key.
This is the fiddliest rule in the whole reader, and getting it wrong is
why a Bill 1200 BOQ used to come out with no usable numbers at all.
Candy prints, for section 110:
110.... GENERAL <- the section title
110.... Scheduled fixed-charge ... <- a group, numbered 110.01
.01 Fixed preliminary ... an item of that group
.02 Value-related ...
110.... Scheduled time-related ... <- the NEXT group, 110.02
.01 Time-related ... ".01" AGAIN
So ".01" is NOT unique - it restarts under every group, and three different
lines in one document are all ".01". The consultant's template numbers
those groups sequentially (110.01, 110.02, 110.03 ...), so that running
counter is what has to be reconstructed here. Without it, 110.01 names
three different lines and cannot be used as a key.
"""
section = ""
group_no = 0
seen_section_title = False
for r in rows:
head = _norm_ref(r.item)
sub = _norm_ref(r.subitem)
if head and not sub and "." not in head:
# a BARE heading - "110...." or "1" or "200". Only these carry the
# section/group logic. A dotted reference like "1.1.1" is a real
# item number and must be left exactly as printed, or the whole
# hierarchy above it is invented.
if head == section:
group_no += 1 # a group inside the section
r.item = f"{section}.{group_no:02d}"
seen_section_title = True
else:
if not seen_section_title and not section:
section = head # first heading = the title
group_no = 0
seen_section_title = True
else:
section = head # a new section entirely
group_no = 0
seen_section_title = False
r.item = head
r.subitem = ""
continue
if sub:
if "." not in sub and group_no:
r.item = f"{section}.{group_no:02d}.{sub}"
elif "." not in sub and section:
r.item = f"{section}.{sub}"
else:
r.item = sub
elif head:
r.item = head
r.subitem = ""
return rows
def _is_noise(r: Row) -> bool:
blob = f"{r.item} {r.desc} {r.unit}".strip()
if r.has_price:
@ -404,7 +531,7 @@ def _build(content: list[Row]) -> Group:
continue
# ---- numbered heading, optionally carrying a subtotal ------------
if num and ITEM_RE.match(num):
if num and (ITEM_RE.match(num) or re.fullmatch(r"[\d.]+", num)):
depth = t.depth(num)
subtotal = _to_num(r.amount_txt) if r.amount_txt else None
@ -479,6 +606,7 @@ def extract(pdf_path: str) -> dict:
rows.extend(_page_rows(page, pno))
rows = _drop_summary_page(rows)
rows = _resolve_refs(rows)
clean = [r for r in rows if not r.is_empty and not _is_noise(r)]
merged = _merge_continuations(clean)
return asdict(_build(merged))

565
boq_engine/template.py
View file

@ -81,28 +81,63 @@ class Layout:
data_rows: list[int] = field(default_factory=list)
total_rows: list[int] = field(default_factory=list)
notes: list[str] = field(default_factory=list)
# The description is not always one column. It occupies whatever sits
# between ITEM and UNIT, and on the SANS 1200 template it moves *within*
# a single sheet - the label "a)" lands in one column and the text in the
# next. So the description is a RUN, and a row's text is the join of them.
desc_cols: list[int] = field(default_factory=list)
def col(self, name: str) -> int | None:
return self.cols.get(name)
def detect_layout(ws) -> Layout:
"""Find the header row and map the columns."""
best: tuple[int, dict[str, int], int] | None = None
for r in range(1, min(ws.max_row, 25) + 1):
cols: dict[str, int] = {}
for c in range(1, min(ws.max_column, 30) + 1):
v = _cellsafe(ws.cell(r, c).value)
if not v or v in cols.values():
def _score_header_row(ws, r: int, ncols: int, span: int = 1) -> dict[str, int]:
"""Map columns to names using a header that may wrap over `span` rows.
Templates split a label across two lines to fit a narrow column -
"QUAN-" over "TITY". Reading either row alone finds no QUANTITY at all, so
the quantity column is simply missed and never written. Joining the rows
recovers the word.
"""
cols: dict[str, int] = {}
for c in range(1, ncols + 1):
parts = [str(ws.cell(r + k, c).value or "").strip()
for k in range(span)]
# A hyphen here is a column-wrap artefact ("QUAN-" / "TITY"), never part
# of the word. Left in, the joined label reads "quan-tity" and matches
# nothing - which is how the quantity column went missing on 49 sheets.
label = _cellsafe("".join(parts).replace("-", ""))
if not label:
continue
for name, pats in HEADER_PATTERNS.items():
if name in cols:
continue
for name, pats in HEADER_PATTERNS.items():
if name in cols:
continue
if v in pats or any(p in v for p in pats):
cols[name] = c
break
if len(cols) >= 4 and (best is None or len(cols) > len(best[1])):
best = (r, cols, len(cols))
if label in pats or any(p in label for p in pats):
cols[name] = c
break
return cols
def detect_layout(ws) -> Layout:
"""Find the header row and map the columns.
Each candidate row is scored twice - read alone, and joined with the row
below it - because a wrapped label only reads correctly when the two halves
are put back together.
"""
ncols = min(ws.max_column, 30)
cands: list[tuple[int, int, int, dict[str, int]]] = []
for r in range(1, min(ws.max_row, 25) + 1):
for span in (1, 2):
if r + span - 1 > ws.max_row:
continue
cols = _score_header_row(ws, r, ncols, span)
if len(cols) >= 4:
cands.append((len(cols), r, span, cols))
# most roles found wins; a tie goes to the LOWER row, because the genuine
# header block is the last one before the data starts. Getting this wrong
# cost the quantity column on 49 of 51 sheets.
best = max(cands, key=lambda t: (t[0], t[1], t[2])) if cands else None
if best is None:
raise ValueError(
@ -110,14 +145,17 @@ def detect_layout(ws) -> Layout:
f"expected 4+ of {', '.join(HEADER_PATTERNS)} in the first 25 rows"
)
header_row, cols, _ = best
_, header_row, span, cols = best
lay = Layout(sheet=ws.title, header_row=header_row, cols=cols)
if span > 1:
lay.notes.append(f"header wraps over rows {header_row}"
f"-{header_row + span - 1}")
c_item = cols.get("item")
c_desc = cols.get("description")
c_qty = cols.get("qty")
for r in range(header_row + 1, ws.max_row + 1):
for r in range(header_row + span, ws.max_row + 1):
a = ws.cell(r, c_item).value if c_item else None
d = ws.cell(r, c_desc).value if c_desc else None
q = ws.cell(r, c_qty).value if c_qty else None
@ -135,9 +173,95 @@ def detect_layout(ws) -> Layout:
lay.notes.append(f"template has no '{name}' column - that value will not be written")
if "payment" in cols:
lay.notes.append("PAYMENT column detected and will be left untouched")
_verify_against_data(ws, lay)
_derive_desc_run(lay)
return lay
def _derive_desc_run(lay: Layout) -> None:
"""Everything between the item column and the unit column is description."""
lo = lay.cols.get("item")
hi = lay.cols.get("unit")
if not lo or not hi or hi <= lo:
lay.desc_cols = [lay.cols["description"]] if "description" in lay.cols else []
return
run = [c for c in range(lo + 1, hi) if c not in lay.cols.values()]
# plus the claimed description column even if it sits outside the run
d = lay.cols.get("description")
if d and d not in run and d not in (lo, hi):
run.append(d)
lay.desc_cols = sorted(set(run))
if len(lay.desc_cols) > 1:
lay.notes.append(
f"description spans columns "
f"{[get_column_letter(c) for c in lay.desc_cols]}")
def _verify_against_data(ws, lay: Layout) -> None:
"""Correct the header-derived column map using what the data actually holds.
A header row is a claim; the cells below it are the fact. On the SANS 1200
template the header advertises DESCRIPTION in column E while the text sits
in D, and a wrong description column quietly wrecks every line match.
"""
sample = lay.data_rows[3:60]
if len(sample) < 5:
return
def longest_text(col: int) -> tuple[int, int]:
"""(count of substantial strings, longest) in a column over the sample."""
n = longest = 0
for r in sample:
v = ws.cell(r, col).value
if isinstance(v, str) and len(v.strip()) > 6:
n += 1
longest = max(longest, len(v.strip()))
return n, longest
# --- description: the header may be off by one --------------------------
d = lay.cols.get("description")
if d:
have, _ = longest_text(d)
best_col, best_score = d, 0.0
for cand in (d - 1, d + 1):
if cand < 1 or cand in lay.cols.values():
continue
cnt, longest = longest_text(cand)
# a real description column: nearly every sampled row carries long
# text. Score it, and only switch if it clearly beats the claim.
score = (cnt / max(1, len(sample))) * (1 if longest > 25 else 0.3)
if score > best_score:
best_col, best_score = cand, score
if best_col != d and best_score > 0.6 and have < len(sample) * 0.2:
lay.notes.append(
f"header claims DESCRIPTION in col {d} but the text is in "
f"col {best_col} - using {get_column_letter(best_col)}")
lay.cols["description"] = best_col
# --- unit: short, mostly-textual values sitting beside the quantity -----
u = lay.cols.get("unit")
if u:
def unitish(col: int) -> int:
n = 0
for r in sample:
v = ws.cell(r, col).value
if isinstance(v, str) and 1 < len(v.strip()) < 12:
n += 1
return n
have = unitish(u)
if have < len(sample) * 0.3:
for cand in (u - 1, u + 1):
if cand < 1 or cand in lay.cols.values():
continue
if unitish(cand) >= len(sample) * 0.5:
lay.notes.append(
f"unit column shifted {u} -> {cand} to match the data")
lay.cols["unit"] = cand
break
# --------------------------------------------------------------------------
# matching
# --------------------------------------------------------------------------
@ -159,16 +283,28 @@ class Slot:
def read_slots(ws, lay: Layout) -> list[Slot]:
ci, cd = lay.col("item"), lay.col("description")
ci = lay.col("item")
run = lay.desc_cols or ([lay.col("description")] if lay.col("description") else [])
out: list[Slot] = []
# A spreadsheet implies its hierarchy: once "110.01" is printed, the blank
# cells below it still belong to 110.01. Without carrying that down, two
# thirds of the rows read as unnumbered and nothing can match them.
carried = ""
for r in lay.data_rows:
num = str(ws.cell(r, ci).value or "").strip() if ci else ""
desc = str(ws.cell(r, cd).value or "").strip() if cd else ""
if num:
carried = num
else:
num = carried
parts = [str(ws.cell(r, c).value or "").strip() for c in run]
desc = " ".join(p for p in parts if p).strip()
if not num and not desc:
continue
# indent level from the leading whitespace the consultant preserved
raw = str(ws.cell(r, cd).value or "") if cd else ""
indent = len(raw) - len(raw.lstrip()) if raw != raw.lstrip() else 0
raw = next((str(ws.cell(r, c).value or "") for c in run
if str(ws.cell(r, c).value or "") !=
str(ws.cell(r, c).value or "").lstrip()), "")
indent = len(raw) - len(raw.lstrip()) if raw else 0
out.append(Slot(row=r, number=num, desc=desc, indent=indent))
return out
@ -191,11 +327,12 @@ def _sim(a: Slot, num: str, desc: str) -> float:
elif tnum in snum or snum in tnum:
num_score = 0.75
else:
num_score = 0.0
elif not tnum and not snum:
num_score = 0.5 # neither numbered: description decides
num_score = 0.0 # two different numbers: real evidence against
else:
num_score = 0.0 # one side numbered, the other not
# A missing number is NEUTRAL, not a mismatch. Most of a BOQ is
# unnumbered a)/b) sub-lines, and a template carries a number down into
# them - scoring that as disagreement made every sub-line unmatchable.
num_score = 0.5
d = SequenceMatcher(None, tdesc, sdesc).ratio()
return round(num_score * 0.6 + d * 0.4, 4)
@ -276,6 +413,10 @@ class Finding:
source_value: Any = None # what our Candy PDF holds
class _FormulaProtected(Exception):
"""Raised internally when a cell must not be overwritten."""
@dataclass
class Result:
written: int = 0
@ -305,6 +446,10 @@ def source_lines(tree: dict, *, with_prices: bool = True) -> list[dict]:
if kind == "line":
if node.get("qty") is None and node.get("rate") is None:
return
# a row of dots is Candy's blank-for-the-tenderer placeholder, not
# a priced line
if re.fullmatch(r"[.\s]+", (node.get("desc") or "").strip()):
return
out.append({
"number": node.get("number", "") or "",
"desc": (node.get("desc") or "").strip(),
@ -379,24 +524,57 @@ def units_compatible(a: str, b: str) -> bool:
return ca is not None and ca == cb
def _section_of(number: str) -> str:
"""The leading segment of a reference: "110.01.01" -> "110".
SANS 1200 sheets are named for the section they cover, so this is what
locates a sheet. It is NOT a unique key on its own - the Candy PDF restarts
".01" under every repeated "110...." heading, so "110.01" names three
different lines in one document.
"""
n = _cellsafe(number)
return n.split(".")[0] if n else ""
def _relevant_pool(lines: list[dict],
slots_by_sheet: dict[str, list[Slot]]
) -> tuple[list[Slot], dict[str, int]]:
"""The slots worth aligning against, in workbook order.
Aligning against every sheet in the workbook looks fine on a 3-sheet
template and falls apart on a 53-sheet one: with 1314 candidate rows for
226 lines, the aligner starts cherry-picking scattered high scores instead
of finding the one coherent run. So the pool is cut down to the sheets that
actually own a section named in the source, which leaves 12 candidates
instead of 1314 and lets the sequence aligner work as designed.
"""
wanted = {_section_of(ln.get("number", "")) for ln in lines}
wanted.discard("")
pool: list[Slot] = []
counts: dict[str, int] = {}
for sheet, slots in slots_by_sheet.items():
secs = {_section_of(sl.number) for sl in slots}
secs.discard("")
if wanted and not (secs & wanted):
continue
pool.extend(slots)
counts[sheet] = len(slots)
return pool, counts
def populate(consultant_xlsx: str, tree: dict, out_path: str, *,
sheet_hint: str | None = None,
overrides: dict | None = None) -> Result:
"""Write our Candy BOQ into a consultant's workbook, in place-safe fashion.
ONE alignment is run across every sheet, in workbook order, rather than
per sheet. The Candy PDF runs Schedule A then Schedule B; the workbook
does too. Aligning each sheet separately against the whole line list
scatters Schedule B's lines into Schedule A's rows.
`overrides` lets the consultant settle a disagreement from inside the app:
keys are (sheet, row, field) and the value is what should be written.
A cell with an override is written rather than left alone, and the
conflict is downgraded to 'resolved' so it stops being flagged.
Lines are routed to a sheet by item number, then aligned WITHIN that sheet.
Nothing is ever written over a value the consultant already had, unless
they explicitly asked for it via `overrides`.
"""
overrides = overrides or {}
res = Result()
res = Result()
original = load_workbook(consultant_xlsx) # pristine copy for comparison
wb = load_workbook(consultant_xlsx) # formulas preserved
lines = source_lines(tree)
if not lines:
@ -404,7 +582,7 @@ def populate(consultant_xlsx: str, tree: dict, out_path: str, *,
names = [sheet_hint] if sheet_hint else list(wb.sheetnames)
layouts: dict[str, Layout] = {}
combined: list[tuple[str, Slot]] = []
slots_by_sheet: dict[str, list[Slot]] = {}
for name in names:
if name not in wb.sheetnames:
@ -417,136 +595,223 @@ def populate(consultant_xlsx: str, tree: dict, out_path: str, *,
continue
layouts[name] = lay
res.layout_notes.extend(f"[{name}] {n}" for n in lay.notes)
for slot in read_slots(ws, lay):
slot._sheet = name
combined.append((name, slot))
slots = read_slots(ws, lay)
for sl in slots:
sl._sheet = name
slots_by_sheet[name] = slots
if not combined:
if not slots_by_sheet:
raise ValueError("no BOQ-shaped sheet found in the workbook")
pairs = align([sl for _, sl in combined], lines)
res.sheets = list(layouts)
pool, counts = _relevant_pool(lines, slots_by_sheet)
res.layout_notes.append(
f"aligned against {len(pool)} candidate rows on "
f"{len(counts)} of {len(slots_by_sheet)} sheets")
if not pool:
raise ValueError("no template sheet shares a section with the source BOQ")
# one sequential alignment over the whole pool, exactly as the aligner is
# designed to work: the document order and the workbook order agree, and a
# single mis-scored row no longer shifts the rest
lay_of = {name: lay for name, lay in layouts.items()}
_fill_pool(wb, lay_of, pool, lines, overrides, res)
_write_totals(wb, layouts, tree, res)
_restore_formulas(original, wb, res)
wb.save(out_path)
return res
def _restore_formulas(original, out, res: Result) -> int:
"""Put back any formula the write pass disturbed. Returns how many.
A guard on each write is the right first line of defence, but there are a
dozen write paths and a new one will be added eventually. This is the
backstop that makes the promise checkable rather than merely intended: if
the consultant's workbook had a formula in a cell, it still has it in the
output. Everything else is rebuilt around that.
"""
restored = 0
for name in out.sheetnames:
if name not in original.sheetnames:
continue
src, dst = original[name], out[name]
for row in src.iter_rows():
for cell in row:
v = cell.value
if isinstance(v, str) and v.startswith("="):
if dst[cell.coordinate].value != v:
dst[cell.coordinate] = v
restored += 1
if restored:
res.add("info", "-", 0, "formulas restored",
f"{restored} cell(s) held a formula in your workbook and have "
f"been put back exactly as they were - your model is intact")
return restored
def _fill_pool(wb, layouts: dict[str, Layout], pool: list[Slot],
lines: list[dict], overrides: dict, res: Result) -> None:
"""Align the whole source against the whole candidate pool, and write."""
pairs = align(pool, lines)
for ln, slot, score in pairs:
if slot is None:
res.add("unmapped", "-", 0,
f"{ln['number']} {ln['desc']}"[:70],
f"source line (p{ln['page']}) has no row in any template sheet",
f"no row in the template for this line (p{ln['page']})",
source=f"{ln['qty']} {ln['unit']} @ {ln['rate']}",
severity="decision")
continue
name = slot._sheet
try:
_write_one(wb, name, layouts[name], ln, slot, overrides, res)
except _FormulaProtected:
continue # already recorded; nothing was written
name, r = slot._sheet, slot.row
ws = wb[name]
lay = layouts[name]
c_unit, c_qty = lay.col("unit"), lay.col("qty")
c_rate, c_amt = lay.col("rate"), lay.col("amount")
label = f"{slot.number} {slot.desc}"[:70]
# build the amount formula from the columns we actually detected - a
# consultant template puts them wherever it likes, and assuming D/E
# silently multiplies UNIT x QTY
if c_qty and c_rate:
ql, rl = get_column_letter(c_qty), get_column_letter(c_rate)
amt_formula = (f'=IF(OR({ql}{{r}}="",{rl}{{r}}=""),"",'
f'{ql}{{r}}*{rl}{{r}})')
else:
amt_formula = None
def _write_one(wb, name: str, lay: Layout, ln: dict,
slot: Slot | None, overrides: dict, res: Result) -> None:
"""Write one aligned line. Returns rather than continuing."""
if slot is None:
res.add("unmapped", name, 0,
f"{ln['number']} {ln['desc']}"[:70],
f"no row in '{name}' for this line (p{ln['page']})",
source=f"{ln['qty']} {ln['unit']} @ {ln['rate']}",
severity="decision")
return
# --- unit ------------------------------------------------------
if c_unit and ln["unit"]:
cur = ws.cell(r, c_unit).value
if cur in (None, ""):
ws = wb[name]
r = slot.row
c_unit, c_qty = lay.col("unit"), lay.col("qty")
c_rate, c_amt = lay.col("rate"), lay.col("amount")
label = f"{slot.number} {slot.desc}"[:70]
if c_qty and c_rate:
ql, rl = get_column_letter(c_qty), get_column_letter(c_rate)
amt_formula = (f'=IF(OR({ql}{{r}}="",{rl}{{r}}=""),"",'
f'{ql}{{r}}*{rl}{{r}})')
else:
amt_formula = None
# --- unit ---------------------------------------------------------------
if c_unit and ln["unit"]:
cur = ws.cell(r, c_unit).value
if cur in (None, ""):
if _writable(ws, r, c_unit, res, name, label, "unit"):
ws.cell(r, c_unit, ln["unit"])
elif not units_compatible(cur, ln["unit"]):
hint = ""
if _norm_unit(ln["unit"]) == "?":
hint = (" -- Candy printed a corrupt unit character; the "
"template's real unit was kept, not overwritten")
ws.cell(r, c_unit, cur)
elif _norm_unit(cur) == "?":
hint = (" -- template unit is unreadable, source unit "
"written instead")
else:
hint = " -- left as the template has it"
okey = (name, r, "unit")
if okey in overrides:
ws.cell(r, c_unit, overrides[okey])
res.add("conflict", name, r, label,
f"unit: set to {overrides[okey]!r} by the consultant",
source=str(ln["unit"]), severity="resolved",
field="unit", template_value=cur,
source_value=ln["unit"])
else:
res.add("conflict", name, r, label,
f"unit: template {cur!r} vs source {ln['unit']!r}{hint}",
source=str(ln["unit"]),
severity="source_defect",
field="unit", template_value=cur,
source_value=ln["unit"])
# --- quantity ---------------------------------------------------
if c_qty and ln["qty"] is not None:
cur = ws.cell(r, c_qty).value
if cur in (None, ""):
ws.cell(r, c_qty, ln["qty"])
elif isinstance(cur, (int, float)) and abs(float(cur) - ln["qty"]) > 1e-6:
okey = (name, r, "qty")
if okey in overrides:
ws.cell(r, c_qty, overrides[okey])
res.add("conflict", name, r, label,
f"quantity: set to {overrides[okey]} by the consultant",
source=str(ln["qty"]), severity="resolved",
field="qty", template_value=cur,
source_value=ln["qty"])
else:
if ln["rate_only"] or ln["qty"] == 0:
reason = ("source carries no quantity here (printed 'Rate "
"only'), so the template's own figure was kept")
sev = "source_gap"
else:
reason = ("the two documents disagree; NEITHER was "
"overwritten")
sev = "decision"
res.add("conflict", name, r, label,
f"quantity: template {cur} vs source {ln['qty']} - {reason}",
source=str(ln["qty"]), severity=sev,
field="qty", template_value=cur,
source_value=ln["qty"])
res.written += 1
elif not units_compatible(cur, ln["unit"]):
okey = (name, r, "unit")
if okey in overrides:
ws.cell(r, c_unit, overrides[okey])
res.add("conflict", name, r, label,
f"unit: set to {overrides[okey]!r} by the consultant",
source=str(ln["unit"]), severity="resolved",
field="unit", template_value=cur,
source_value=ln["unit"])
else:
res.kept += 1
# already correct - still make sure a rate landed
if c_rate and ln["rate"] is not None and not ln["rate_only"]:
if ws.cell(r, c_rate).value in (None, ""):
note = (" -- Candy printed a corrupt unit character, so the "
"template's real unit was kept"
if _norm_unit(ln["unit"]) == "?" else
" -- left as the template has it")
res.add("conflict", name, r, label,
f"unit: template {cur!r} vs source {ln['unit']!r}{note}",
source=str(ln["unit"]), severity="source_defect",
field="unit", template_value=cur,
source_value=ln["unit"])
# --- quantity ------------------------------------------------------------
stop = False
if c_qty and ln["qty"] is not None:
cur = ws.cell(r, c_qty).value
okey = (name, r, "qty")
if okey in overrides:
ws.cell(r, c_qty, overrides[okey])
res.add("conflict", name, r, label,
f"quantity: set to {overrides[okey]} by the consultant",
source=str(ln["qty"]), severity="resolved", field="qty",
template_value=cur, source_value=ln["qty"])
elif cur in (None, ""):
if _writable(ws, r, c_qty, res, name, label, "qty"):
ws.cell(r, c_qty, ln["qty"])
elif isinstance(cur, (int, float)) and abs(float(cur) - ln["qty"]) > 1e-6:
if ln["rate_only"] or ln["qty"] == 0:
reason = ("source carries no quantity here (printed 'Rate "
"only'), so the template's own figure was kept")
sev = "source_gap"
else:
reason = "the two documents disagree; NEITHER was overwritten"
sev = "decision"
res.add("conflict", name, r, label,
f"quantity: template {cur} vs source {ln['qty']} - {reason}",
source=str(ln["qty"]), severity=sev, field="qty",
template_value=cur, source_value=ln["qty"])
else:
res.kept += 1
stop = True
if c_rate and ln["rate"] is not None and not ln["rate_only"]:
if ws.cell(r, c_rate).value in (None, ""):
if _writable(ws, r, c_rate, res, name, label, "rate"):
ws.cell(r, c_rate, ln["rate"])
if c_amt:
if c_amt and amt_formula and _writable(
ws, r, c_amt, res, name, label, "amount"):
ws.cell(r, c_amt, amt_formula.format(r=r))
res.written += 1
continue
# --- rate + amount ----------------------------------------------
if c_rate and ln["rate"] is not None:
ws.cell(r, c_rate, ln["rate"])
if c_amt:
if ln["rate_only"]:
ws.cell(r, c_amt, "Rate only")
res.add("info", name, r, label,
"'Rate only' in source - rate written, amount as text",
source=str(ln["rate"]))
else:
# live formula so the consultant can re-rate and watch the
# total follow. Never a frozen value.
if amt_formula:
ws.cell(r, c_amt, amt_formula.format(r=r))
res.written += 1
elif c_rate and ln["rate_only"] and ln["amount"] is None:
if c_amt:
if stop:
return
# --- rate and amount -----------------------------------------------------
res.written += 1
if c_rate and ln["rate"] is not None:
ws.cell(r, c_rate, ln["rate"])
if c_amt:
if ln["rate_only"]:
ws.cell(r, c_amt, "Rate only")
res.add("info", name, r, label,
"'Rate only' in source - rate written, amount as text",
source=str(ln["rate"]))
elif amt_formula:
ws.cell(r, c_amt, amt_formula.format(r=r))
elif c_amt and ln["rate_only"]:
ws.cell(r, c_amt, "Rate only")
_write_totals(wb, layouts, tree, res)
wb.save(out_path)
return res
def _writable(ws, r: int, col: int, res: Result, sheet: str, label: str,
field: str) -> bool:
"""Refuse to overwrite a cell that holds a formula.
A consultant's workbook is a working model, not a data dump. Quantities
are derived ("=G35", "=CEILING.MATH(...)"), line amounts are computed, and
section totals are SUMs over the rows above. Writing a literal over any of
those destroys their work while looking like it succeeded.
"""
cur = ws.cell(r, col).value
if isinstance(cur, str) and cur.startswith("="):
res.add("conflict", sheet, r, label,
f"{field}: that cell holds a formula ({cur[:40]}) - your model "
f"was left alone and the source value was not written",
severity="decision", field=field, template_value=cur)
return False
return True
def _guard_formula(ws, r: int, col: int, res: Result, sheet: str,
label: str) -> None:
"""Refuse to clobber a cell that holds a formula.
On a SANS 1200 template the quantities are DERIVED - "=G35",
"=CEILING.MATH(414.065+302.817,10)" - from a Measurement sheet. Writing
Candy's number over one of those destroys the engineer's model, and it
looks like it worked.
"""
cur = ws.cell(r, col).value
if isinstance(cur, str) and cur.startswith("="):
res.add("conflict", sheet, r, label,
f"that cell holds a FORMULA ({cur[:44]}) - your model's value "
f"was left in place, and the source quantity was not written",
severity="decision", field="qty",
template_value=cur, source_value=None)
raise _FormulaProtected()
SUMMARY_REF = re.compile(
@ -610,6 +875,13 @@ def _write_totals(wb, layouts: dict[str, Layout], tree: dict,
label, val = values[vi]
vi += 1
tgt = wb[target_sheet]
existing = tgt[f"{col_letter}{row_no}"].value
if isinstance(existing, str) and existing.startswith("="):
res.add("info", target_sheet, int(row_no), label,
f"summary cell {col_letter}{row_no} already holds a "
f"formula ({existing[:34]}) - left as it is",
severity="note")
continue
tgt[f"{col_letter}{row_no}"] = val
tgt[f"{col_letter}{row_no}"].number_format = "#,##0.00"
res.add("total", name, r, label,
@ -633,7 +905,8 @@ def _write_totals(wb, layouts: dict[str, Layout], tree: dict,
break
prev = r2
rows = [r for r in lay.data_rows if prev < r < tr]
if rows:
cur = ws.cell(tr, c_amt).value
if rows and not (isinstance(cur, str) and cur.startswith("=")):
ws.cell(tr, c_amt,
f"=SUM({letter_amt}{min(rows)}:{letter_amt}{max(rows)})"
).number_format = "#,##0.00"