Dave
c831e2d4a9
The MEGA CHECK sheet was meant to answer "which section do I look at". It was quietly failing to, in four separate ways, each of which made the file look more finished than it was. A line with no row to go to was dropped on the floor. reconcile() collected it, tagged it with sheet "-", and then only attached rows to a section whose SHEET matched - so "-" matched nothing and the line vanished. It was the one thing most worth telling a person, and it was the only thing not shown. Every unplaced line now appears with its qty, unit and rate. Sections were paired on a bare number, and Candy restarts its numbering inside every schedule. Blueberry has a "1" in Schedule A, another "1" under Schedule B and a "4" under a third. Keying on that merged unrelated sections: 6 of Blueberry's 8 matched nothing, and the sheet showed 2 clean rows that read as "this book is checked". Now keyed on the section title - 8 of 8 and 18 of 18. "WORKBOOK SAYS" summed the whole sheet. Blueberry's SCHEDULE B holds sections 1 to 6, so "1 SITE CLEARANCE" was reported as R4,422,018 against a printed R160,057.99 - the right section with a figure that is not that section's. Each section is now summed from its own rows only, and a row that cannot be attributed counts for nothing rather than for every section at once. That last one had 141, 142 and 143 all reporting the same R100,722.77. Untested sections are now stated as untested. A table of clean sections implies the rest balance, so the sheet says plainly which ones could not be matched. A section the PDF prints as 0.00 is a fill-in for the tenderer, not a disagreement. Blueberry's "4 CABLE DUCTS" is twelve lines that all carry qty 0.0, and the consultant then fills in their own quantities; reporting that as a R348,746.25 difference points at our work when the source simply has nothing to compare against. It now reads "no figure in PDF". "every write landed" only checked that a cell was not empty, then stopped after 600. A rate overwritten with a wrong number passed silently - proved by planting one and watching the gate stay green. It now compares every value, and the 600-cell cap is gone: a check that covers part of the file and reports as if it covered all of it is worse than no check. Damage still blocks the download. Incomplete or misplaced work is handed over with the punch list, and needs_review now fires on unplaced lines too - it used to stay quiet on Blueberry, which had one. Verified by fault injection, not by reading the code: corrupting a written cell is now caught, two rates swapped between rows are flagged without blocking, and no unplaced line is ever dropped. All four behave.
1304 lines
55 KiB
Python
1304 lines
55 KiB
Python
"""
|
||
Populate a CONSULTANT's BOQ workbook from our own Candy PDF export.
|
||
|
||
The direction matters and is easy to get backwards: Candy is *our* system, so
|
||
the PDF is the source of truth and is always the same shape. The consultant's
|
||
Excel is *theirs*, and changes from consultant to consultant. So the hard part
|
||
is not reading the document - it is writing into a workbook we did not design
|
||
without destroying what is already in it.
|
||
|
||
Three jobs:
|
||
|
||
1. LAYOUT DETECTION
|
||
Find the header row and work out which column is ITEM / DESCRIPTION /
|
||
UNIT / QTY / RATE / AMOUNT. A consultant's PAYMENT column (SABS / PSA
|
||
clause references) is detected too and then never touched.
|
||
|
||
2. MAPPING
|
||
Line the PDF's lines up with the workbook's rows. The two documents are
|
||
near 1:1 but not identical: the workbook truncates descriptions, inserts
|
||
spacer rows and adds "TOTAL CARRIED FORWARD" blocks, while the PDF wraps
|
||
descriptions across several lines. A sequence alignment over
|
||
(item number, description similarity) survives all of that.
|
||
|
||
3. PROOF
|
||
A population pass that cannot be checked is a guess. Every source line is
|
||
accounted for - written, already-matching, or reported as an exception -
|
||
and section totals are written as live formulas that must reproduce the
|
||
PDF's own printed subtotal.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
import shutil
|
||
from dataclasses import dataclass, field
|
||
from difflib import SequenceMatcher
|
||
from typing import Any
|
||
|
||
from openpyxl import load_workbook
|
||
from openpyxl.utils import get_column_letter
|
||
|
||
# --------------------------------------------------------------------------
|
||
# header synonyms
|
||
# --------------------------------------------------------------------------
|
||
HEADER_PATTERNS: dict[str, tuple[str, ...]] = {
|
||
"item": ("item", "item no", "item no.", "item number", "no", "no."),
|
||
"description": ("description", "bill description", "desc", "particulars",
|
||
"description of work"),
|
||
"unit": ("unit", "units", "uom"),
|
||
"qty": ("qty", "quantity", "bill quantity", "quantities"),
|
||
"rate": ("rate", "unit rate", "unit price", "price"),
|
||
"amount": ("amount", "total", "value", "sum"),
|
||
"payment": ("payment", "payment clause", "clause", "spec", "specification"),
|
||
}
|
||
|
||
TOTAL_RE = re.compile(r"^\s*total\b", re.I)
|
||
# "Carried forward" / "Brought forward" / "B/F" rows are section continuation
|
||
# markers, not BOQ lines. They sit in the ITEM column, so a naive data-row test
|
||
# made them placeable - and the real line above merged down into one, putting
|
||
# its rate on a row that belongs to nothing.
|
||
CARRY_RE = re.compile(r"^\s*(carried|brought|b/f)\b|\bforward\b|"
|
||
r"^\s*to\s+carry\b|^\s*cf\b", re.I)
|
||
BROUGHT_RE = re.compile(r"brought\s+forward", re.I)
|
||
SUMMARY_RE = re.compile(r"to\s+(combined\s+)?summary", re.I)
|
||
|
||
|
||
def _norm(s: Any) -> str:
|
||
return re.sub(r"\s+", " ", str(s or "")).strip().lower()
|
||
|
||
|
||
def _cellsafe(v: Any) -> str:
|
||
"""Normalise a cell value to something comparable across both documents."""
|
||
s = _norm(v)
|
||
s = s.replace("ℓ", "l").replace("³", "3").replace("²", "2")
|
||
s = re.sub(r"[^a-z0-9./-]+", " ", s)
|
||
return re.sub(r"\s+", " ", s).strip()
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# layout
|
||
# --------------------------------------------------------------------------
|
||
@dataclass
|
||
class Layout:
|
||
sheet: str
|
||
header_row: int
|
||
cols: dict[str, int] = field(default_factory=dict)
|
||
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 _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
|
||
if label in pats or any(p in label for p in pats):
|
||
cols[name] = c
|
||
break
|
||
return cols
|
||
|
||
|
||
def detect_layout(ws, hint: "SheetProfile | None" = None) -> 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.
|
||
|
||
`hint` is a previously saved layout for this consultant. It is APPLIED, not
|
||
trusted: a cached column map is only ever used when the columns it names
|
||
still hold the kind of data they were saved for. If the file has moved on -
|
||
a column inserted, a section retitled - the detection runs as normal and
|
||
says so. A silently-wrong profile is worse than no profile.
|
||
"""
|
||
if hint is not None and _hint_holds(ws, hint):
|
||
cols = {k: int(v) for k, v in hint.cols.items()}
|
||
lay = Layout(sheet=ws.title, header_row=hint.header_row, cols=cols)
|
||
lay.notes.append(f"column map reused from {hint.consultant}'s saved "
|
||
f"profile (verified against this file)")
|
||
lay.desc_cols = list(hint.desc_cols or [])
|
||
# Re-run the same data verification and description-run derivation the
|
||
# fresh path does. Without this a warm run and a cold run can differ,
|
||
# and a converter that gives two answers to the same pair of files is
|
||
# not one anybody should trust on a tender.
|
||
_verify_against_data(ws, lay)
|
||
_derive_desc_run(lay)
|
||
_rebuild_rows(ws, lay)
|
||
return lay
|
||
|
||
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(
|
||
f"Could not find a header row in sheet '{ws.title}' - "
|
||
f"expected 4+ of {', '.join(HEADER_PATTERNS)} in the first 25 rows"
|
||
)
|
||
|
||
_, 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}")
|
||
|
||
_verify_against_data(ws, lay)
|
||
_derive_desc_run(lay)
|
||
_rebuild_rows(ws, lay, start=header_row + span)
|
||
|
||
for name in ("item", "description", "unit", "qty", "rate", "amount"):
|
||
if name not in cols:
|
||
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")
|
||
|
||
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]}")
|
||
|
||
|
||
# Markers that only ever appear in a header block. These sheets repeat their
|
||
# headers partway down a long sheet so a printout stays readable, and the
|
||
# repeat sits in the middle of the data - a naive scan treats it as a line and
|
||
# will happily write a rate onto "DESCRIPTION: CENTRAL TOWERS".
|
||
HEADER_MARKERS = ("description:", "payment", "refers", "item no",
|
||
"bill description", "section 1200", "bill quantity")
|
||
|
||
|
||
def _is_header_row(ws, r: int, ncols: int) -> bool:
|
||
txt = " ".join(str(ws.cell(r, c).value or "")
|
||
for c in range(1, ncols + 1)).lower()
|
||
if any(m in txt for m in HEADER_MARKERS):
|
||
return True
|
||
return len(_score_header_row(ws, r, ncols, 1)) >= 4
|
||
|
||
|
||
def _rebuild_rows(ws, lay: Layout, start: int | None = None) -> None:
|
||
"""Re-scan which rows are data and which are totals, from a column map.
|
||
|
||
Tests the description RUN, not the single column the header claimed. On
|
||
the SANS 1200 template the header says DESCRIPTION is in E while the text
|
||
is in D, so testing E alone classified real line rows as blank - and
|
||
".04 Appointment of Community Liaison Officer" never became a candidate at
|
||
all.
|
||
"""
|
||
first = lay.header_row + 1 if start is None else start
|
||
c_item = lay.cols.get("item")
|
||
c_qty = lay.cols.get("qty")
|
||
run = lay.desc_cols or ([lay.cols["description"]]
|
||
if "description" in lay.cols else [])
|
||
ncols = min(ws.max_column, 30)
|
||
for r in range(first, ws.max_row + 1):
|
||
# skip a repeated header block, and the two title rows above it
|
||
if _is_header_row(ws, r, ncols):
|
||
for rr in (r, r - 1, r - 2):
|
||
if rr >= first and rr not in lay.data_rows:
|
||
lay.header_rows_found = getattr(
|
||
lay, "header_rows_found", set())
|
||
lay.header_rows_found.add(rr)
|
||
continue
|
||
if r in getattr(lay, "header_rows_found", set()):
|
||
continue
|
||
a = ws.cell(r, c_item).value if c_item else None
|
||
d = " ".join(str(ws.cell(r, c).value or "").strip()
|
||
for c in run).strip() or None
|
||
q = ws.cell(r, c_qty).value if c_qty else None
|
||
label = f"{a or ''} {d or ''}".strip()
|
||
if label and CARRY_RE.match(label):
|
||
continue # a continuation marker, not a line
|
||
if label and TOTAL_RE.match(label):
|
||
lay.total_rows.append(r)
|
||
continue
|
||
if any(x not in (None, "") for x in (a, d)) or q is not None:
|
||
lay.data_rows.append(r)
|
||
|
||
|
||
def _hint_holds(ws, hint) -> bool:
|
||
"""Does this saved column map still describe THIS file?
|
||
|
||
A cheap, decisive check: the item, unit, quantity and amount columns must
|
||
exist and must not be absurdly wide for the sheet. A template that gained
|
||
or lost columns fails here and falls back to fresh detection.
|
||
"""
|
||
for name in ("item", "unit", "qty", "amount"):
|
||
c = hint.cols.get(name)
|
||
if c is None:
|
||
return False
|
||
if not (1 <= int(c) <= max(ws.max_column, 1)):
|
||
return False
|
||
if hint.header_row < 1 or hint.header_row > min(ws.max_row, 25):
|
||
return False
|
||
return True
|
||
|
||
|
||
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
|
||
# --------------------------------------------------------------------------
|
||
@dataclass
|
||
class Slot:
|
||
row: int
|
||
number: str = ""
|
||
desc: str = ""
|
||
indent: int = 0
|
||
unit: Any = None
|
||
qty: Any = None
|
||
rate: Any = None
|
||
amount: Any = None
|
||
_sheet: str = ""
|
||
|
||
@property
|
||
def key(self) -> str:
|
||
return _cellsafe(f"{self.number} {self.desc}")
|
||
|
||
|
||
# a description opening with a lettered sub-item label (a), b), c)) starts a
|
||
# NEW line rather than continuing the one above
|
||
LETTER_ONLY = re.compile(r"^\(?[a-z]\)\s", re.I)
|
||
# a description that OPENS with its own reference - ".01 ", "1.2 ", "c) " -
|
||
# starts a NEW line. Only bare continuation text may be folded into the row
|
||
# above. Excluding only "a)" left ".01 0.5 - 1.0m Deep" being swallowed by the
|
||
# sub-group heading above it, which is how a whole depth band went missing from
|
||
# the pool and its rate drifted onto the next group's rows.
|
||
STARTS_LINE = re.compile(r"^\(?[a-z]\)\s|^\.?\d+(?:\.\d+)*[\s.]", re.I)
|
||
# ".01", "1.2", "a)" - a description that OPENS with a reference starts a new
|
||
# line; only bare continuation text may be folded into the one above
|
||
STARTS_LINE = re.compile(r"^\(?[a-z]\)\s|^\.?\d+(?:\.\d+)*[\s.]", re.I)
|
||
|
||
|
||
def read_slots(ws, lay: Layout) -> list[Slot]:
|
||
"""One slot per real line.
|
||
|
||
A BOQ row in these templates is often TWO rows: the one carrying the item
|
||
number, unit and quantity, and a following row holding the wrapped tail of
|
||
the description - "Appointment of Community Liaison Officer" on one row and
|
||
"for the duration of the contract period" on the next.
|
||
|
||
Left as separate slots, those tails become decoys. The aligner matched
|
||
".04 Appointment of Community Liaison Officer" to the PREVIOUS line's tail
|
||
row, and the values landed one row above the label they belonged to. A row
|
||
that has no item number of its own and nothing in the priced columns is a
|
||
continuation, so its text is joined onto the line above.
|
||
"""
|
||
ci = lay.col("item")
|
||
cu, cq, cr = lay.col("unit"), lay.col("qty"), lay.col("rate")
|
||
run = lay.desc_cols or ([lay.col("description")]
|
||
if lay.col("description") else [])
|
||
out: list[Slot] = []
|
||
carried = ""
|
||
|
||
for r in lay.data_rows:
|
||
raw = str(ws.cell(r, ci).value or "").strip() if ci else ""
|
||
# Only a real reference may be carried down a blank item cell. The
|
||
# wrapped header "ITEM" / "NO" leaves a literal "NO" in that column,
|
||
# and propagating it made every row beneath look like a mismatch.
|
||
if REF_TOKEN.match(raw):
|
||
carried = raw
|
||
elif raw:
|
||
carried = ""
|
||
num = carried
|
||
|
||
desc = _row_desc(ws, r, lay)
|
||
if CARRY_RE.match(f"{raw} {desc}".strip()):
|
||
continue # never a placement target
|
||
priced = any(ws.cell(r, c).value not in (None, "")
|
||
for c in (cu, cq, cr) if c)
|
||
|
||
if not desc and not priced and not raw:
|
||
continue # wholly empty spacer row
|
||
|
||
# A row carrying the FIGURES but no item number and no label of its own
|
||
# is the tail of the line above: the description wrapped onto this row
|
||
# while the unit and quantity stayed. They are one BOQ line, so the
|
||
# slot becomes this row - that is where the columns are - and keeps the
|
||
# joined description so matching still sees the whole line.
|
||
#
|
||
# A description opening "a)" / "b)" / "c)" is a NEW line and must never
|
||
# be absorbed, or sub-items get folded into the heading above them.
|
||
if (out and priced and not raw and desc and out[-1]._bare
|
||
and not STARTS_LINE.match(desc)):
|
||
prev = out[-1]
|
||
prev.desc = f"{prev.desc} {desc}".strip() if prev.desc else desc
|
||
prev.row = r
|
||
prev._bare = False
|
||
prev._advanced = True
|
||
continue
|
||
|
||
# bare continuation text only folds in when it is plainly a tail and
|
||
# the slot above has not already advanced to a rows-with-figures
|
||
if (out and not priced and not raw and desc
|
||
and not STARTS_LINE.match(desc) and not out[-1]._advanced):
|
||
if out[-1].desc:
|
||
out[-1].desc = f"{out[-1].desc} {desc}".strip()
|
||
else:
|
||
out[-1].desc = desc
|
||
continue
|
||
|
||
raw_desc = 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_desc) - len(raw_desc.lstrip())
|
||
if raw_desc != raw_desc.lstrip() else 0)
|
||
slot = Slot(row=r, number=num, desc=desc, indent=indent)
|
||
slot._bare = not priced # does this row carry figures of its own?
|
||
slot._advanced = False
|
||
out.append(slot)
|
||
return out
|
||
|
||
|
||
def _sim(a: Slot, num: str, desc: str) -> float:
|
||
"""How well does a source line match a template slot?
|
||
|
||
Item number is the strong signal, but most of a BOQ is unnumbered a)/b)/c)
|
||
sub-lines where the number is blank on BOTH sides. Scoring those as "no
|
||
evidence" pushed every sub-line under the threshold, so description has to
|
||
carry full weight when neither side has a number.
|
||
"""
|
||
tnum, snum = _cellsafe(a.number), _cellsafe(num)
|
||
tdesc = f"{tnum} {_cellsafe(a.desc)}".strip()
|
||
sdesc = f"{snum} {_cellsafe(desc)}".strip()
|
||
|
||
if tnum and snum:
|
||
if tnum == snum:
|
||
num_score = 1.0
|
||
elif tnum in snum or snum in tnum:
|
||
num_score = 0.75
|
||
else:
|
||
num_score = 0.0 # two different numbers: real evidence against
|
||
else:
|
||
# 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)
|
||
|
||
|
||
def align(slots: list[Slot], lines: list[dict], threshold: float = 0.45
|
||
) -> list[tuple[dict, Slot | None, float]]:
|
||
"""Sequence-align source lines onto template slots.
|
||
|
||
A Needleman-Wunsch style DP rather than a greedy pass: a single
|
||
mis-detected row would otherwise shift every line after it, which is
|
||
exactly the silent-corruption failure this whole tool exists to prevent.
|
||
"""
|
||
n, m = len(lines), len(slots)
|
||
GAP = -0.35
|
||
|
||
def sc(i: int, j: int) -> float:
|
||
ln = lines[i]
|
||
return _sim(slots[j], ln.get("number", ""), ln.get("desc", ""))
|
||
|
||
dp = [[0.0] * (m + 1) for _ in range(n + 1)]
|
||
bt = [[0] * (m + 1) for _ in range(n + 1)]
|
||
for i in range(1, n + 1):
|
||
dp[i][0] = dp[i - 1][0] + GAP
|
||
bt[i][0] = 1
|
||
for j in range(1, m + 1):
|
||
dp[0][j] = dp[0][j - 1] + GAP
|
||
bt[0][j] = 2
|
||
for i in range(1, n + 1):
|
||
for j in range(1, m + 1):
|
||
diag = dp[i - 1][j - 1] + sc(i - 1, j - 1)
|
||
up = dp[i - 1][j] + GAP
|
||
left = dp[i][j - 1] + GAP
|
||
best = max(diag, up, left)
|
||
dp[i][j] = best
|
||
bt[i][j] = 0 if best == diag else (1 if best == up else 2)
|
||
|
||
out: list[tuple[dict, Slot | None, float]] = []
|
||
i, j = n, m
|
||
while i > 0 or j > 0:
|
||
mv = bt[i][j]
|
||
if mv == 0 and i > 0 and j > 0:
|
||
s = sc(i - 1, j - 1)
|
||
out.append((lines[i - 1], slots[j - 1] if s >= threshold else None, s))
|
||
i, j = i - 1, j - 1
|
||
elif mv == 1 and i > 0:
|
||
out.append((lines[i - 1], None, 0.0))
|
||
i -= 1
|
||
else:
|
||
j -= 1
|
||
out.reverse()
|
||
return out
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# the population pass
|
||
# --------------------------------------------------------------------------
|
||
@dataclass
|
||
class Finding:
|
||
kind: str # 'written' | 'kept' | 'conflict' | 'unmapped' | 'total' | 'info'
|
||
sheet: str
|
||
row: int
|
||
label: str
|
||
detail: str
|
||
source: str = ""
|
||
# How much a human needs to care. This is data, not presentation - the UI
|
||
# maps it to a colour, so the classification lives here where it can be
|
||
# asserted in tests.
|
||
# 'decision' the two documents genuinely disagree; a person must pick
|
||
# 'source_gap' OUR pdf has no figure here, theirs is better
|
||
# 'source_defect' OUR pdf printed something corrupt
|
||
# 'note' worth knowing, no action implied
|
||
severity: str = "note"
|
||
# For a conflict the UI needs enough to offer a real choice: which cell,
|
||
# what is in it now, and what the source wants to put there.
|
||
field: str = "" # 'qty' | 'rate' | 'unit'
|
||
template_value: Any = None # what the consultant's file holds today
|
||
source_value: Any = None # what our Candy PDF holds
|
||
index: int = -1 # position in source_lines(); -1 = not a line
|
||
# The top-level section this line came from, as a bare number ("140").
|
||
# A line the aligner could not place has no row, so without this there is
|
||
# nothing to hang it off in the MEGA CHECK punch list and it simply
|
||
# disappears - which is the one case a person most needs to be told about.
|
||
section: str = ""
|
||
|
||
|
||
class _FormulaProtected(Exception):
|
||
"""Raised internally when a cell must not be overwritten."""
|
||
|
||
|
||
@dataclass
|
||
class Result:
|
||
written: 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)
|
||
# (sheet, row, line) for every line the writer placed - the ground truth
|
||
# the placement check verifies against
|
||
placements: list = field(default_factory=list)
|
||
findings: list[Finding] = field(default_factory=list)
|
||
sheets: 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 placed(self, sheet: str, row: int, ln: dict, slot_desc: str = "") -> None:
|
||
"""Record that this line was placed on this row.
|
||
|
||
The writer is the only thing that knows this for certain, so it says so
|
||
explicitly rather than the verifier trying to reverse-engineer it from
|
||
the cell values afterwards.
|
||
"""
|
||
self.placements.append({
|
||
"sheet": sheet, "row": row,
|
||
"number": ln.get("number", ""), "desc": ln.get("desc", ""),
|
||
"unit": ln.get("unit", ""), "qty": ln.get("qty"),
|
||
"rate": ln.get("rate"),
|
||
# the slot's own merged description - for a line split across two
|
||
# rows this is the full text, where the source line may be cut off
|
||
# by the PDF's own wrapping
|
||
"slot_desc": slot_desc,
|
||
})
|
||
|
||
def add(self, *a, **kw) -> None:
|
||
self.findings.append(Finding(*a, **kw))
|
||
|
||
@property
|
||
def problems(self) -> list[Finding]:
|
||
return [f for f in self.findings
|
||
if f.kind in ("conflict", "unmapped", "info")]
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# source-side flattening: PDF tree -> the list we try to place
|
||
# --------------------------------------------------------------------------
|
||
def source_lines(tree: dict, *, with_prices: bool = True) -> list[dict]:
|
||
"""Every priced line in the Candy PDF, in document order.
|
||
|
||
Each line also carries `sec`: the label of the SUMMARY section it sits
|
||
under, and `path`: every heading above it.
|
||
|
||
The ancestry is not decoration. Candy RESTARTS its numbering inside each
|
||
schedule, so Blueberry has a section "1" in Schedule A, another "1" under
|
||
Schedule B, and a "4" under a third schedule. Keying anything on the bare
|
||
number therefore merges unrelated sections - which is how the reconciliation
|
||
silently lost 6 of Blueberry's 8 sections and left a check sheet that
|
||
implied the untested ones were fine.
|
||
|
||
`sec` tracks the DIRECT child of a schedule, not the deepest heading: a
|
||
summary sheet lists "2 ROADS" and never its children "2.1 EARTHWORKS", so
|
||
recording the deepest one would put the line in a section the summary has
|
||
no row for and match nothing.
|
||
"""
|
||
out: list[dict] = []
|
||
|
||
def walk(node: dict, path: list[str], sec: str, top: bool) -> None:
|
||
kind = node.get("kind")
|
||
if kind == "root":
|
||
for c in node.get("children", []):
|
||
walk(c, path, sec, True)
|
||
return
|
||
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 "",
|
||
"tariff_note": (node.get("tariff_note") or "").strip(),
|
||
"desc": (node.get("desc") or "").strip(),
|
||
"unit": (node.get("unit") or "").strip(),
|
||
"qty": node.get("qty"),
|
||
"rate": node.get("rate"),
|
||
"amount": node.get("amount"),
|
||
"rate_only": bool(node.get("rate_only")),
|
||
"page": node.get("page", 0),
|
||
"sec": sec,
|
||
"path": " > ".join(path),
|
||
})
|
||
return
|
||
here, new_sec, new_top = path, sec, False
|
||
if kind in ("section", "schedule", "group"):
|
||
label = f"{node.get('number','')} {node.get('title','')}".strip()
|
||
here = path + [label]
|
||
if top:
|
||
# This node is itself an entry the summary sheet lists.
|
||
new_sec = label
|
||
# Only a SCHEDULE promotes its children to summary entries. A
|
||
# top-level group contributes its own label alone - its children
|
||
# ("1.1 Fixed-Charge Items") are detail the summary never names,
|
||
# and recording them as the section is what left 26 phantom
|
||
# sections and matched none of the 8 that exist.
|
||
new_top = (kind == "schedule")
|
||
for c in node.get("children", []):
|
||
walk(c, here, new_sec, new_top)
|
||
|
||
walk(tree, [], "", False)
|
||
return out
|
||
|
||
|
||
def section_totals(tree: dict) -> dict[str, float]:
|
||
"""Total of the priced lines beneath each section / schedule.
|
||
|
||
Deliberately a LEAF roll-up rather than the printed subtotals: the source
|
||
prints totals that swallow a neighbouring block (1.6 Temporary Works
|
||
includes 1.5 Protection of Services), so summing printed subtotals
|
||
double counts. Summing leaves reproduces the printed grand total exactly.
|
||
"""
|
||
totals: dict[str, float] = {}
|
||
|
||
def leaf_sum(node: dict) -> float:
|
||
if node.get("kind") == "line":
|
||
if node.get("rate_only") or node.get("amount") is None:
|
||
return 0.0
|
||
return float(node["amount"])
|
||
return sum(leaf_sum(c) for c in node.get("children", []))
|
||
|
||
def walk(node: dict) -> None:
|
||
if node.get("kind") in ("section", "schedule", "root"):
|
||
if node.get("kind") in ("section", "schedule"):
|
||
label = f"{node.get('number','')} {node.get('title','')}".strip()
|
||
totals[label] = round(leaf_sum(node), 2)
|
||
for c in node.get("children", []):
|
||
walk(c)
|
||
|
||
walk(tree)
|
||
return totals
|
||
|
||
|
||
def _norm_unit(u: str) -> str:
|
||
s = _norm(u).replace("ℓ", "l").replace("³", "3").replace("²", "2")
|
||
return s.rstrip(".")
|
||
|
||
|
||
UNIT_ALIASES = {
|
||
"l": "l", "litre": "l", "litres": "l", "l": "l",
|
||
"m2": "m2", "m 2": "m2", "sqm": "m2",
|
||
"m3": "m3", "m 3": "m3", "cbm": "m3",
|
||
"no": "no", "no.": "no", "nr": "no", "number": "no",
|
||
"sum": "sum", "sum.": "sum", "ls": "sum",
|
||
"p.sum": "psum", "psum": "psum", "ps.sum": "psum", "pc.sum": "pcsum",
|
||
"month": "month", "months": "month", "mth": "month",
|
||
"t": "t", "tonne": "t", "tonnes": "t",
|
||
"m": "m", "m ": "m",
|
||
"%": "%", "pct": "%", "?": "?", "": "",
|
||
}
|
||
|
||
|
||
def _unit_key(u: str) -> str:
|
||
"""Collapse a unit to a comparable key: case, spacing and dots are noise.
|
||
|
||
"PC Sum", "PCSum" and "PC.Sum" are the same unit written three ways, and
|
||
flagging that as a disagreement buried the real findings under 12 false
|
||
alarms. Genuinely different units (m vs m3, Sum vs %) still differ.
|
||
"""
|
||
s = (u or "").strip().lower()
|
||
s = s.replace("ℓ", "l").replace("³", "3").replace("²", "2")
|
||
s = re.sub(r"[.\s]+", "", s)
|
||
return s
|
||
|
||
|
||
def units_compatible(a: str, b: str) -> bool:
|
||
na, nb = _norm_unit(a), _norm_unit(b)
|
||
if na == nb:
|
||
return True
|
||
ka, kb = _unit_key(a), _unit_key(b)
|
||
if ka == kb:
|
||
return True
|
||
ca, cb = UNIT_ALIASES.get(na), UNIT_ALIASES.get(nb)
|
||
if ca is not None and ca == cb:
|
||
return True
|
||
return UNIT_ALIASES.get(ka) is not None and UNIT_ALIASES.get(ka) == UNIT_ALIASES.get(kb)
|
||
|
||
|
||
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 hints_from_profile(profile) -> dict:
|
||
"""Turn a saved consultant profile into a {sheet: SheetProfile} hint map."""
|
||
if profile is None:
|
||
return {}
|
||
return {sh.sheet: sh for sh in getattr(profile, "sheets", []) if sh.cols}
|
||
|
||
|
||
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 _row_desc(ws, row: int, lay: Layout) -> str:
|
||
run = lay.desc_cols or ([lay.col("description")] if lay.col("description") else [])
|
||
return " ".join(str(ws.cell(row, c).value or "").strip()
|
||
for c in run if ws.cell(row, c).value not in (None, "")).strip()
|
||
|
||
|
||
def propose_placements(tree: dict, consultant_xlsx: str, lines: list[dict],
|
||
top_n: int = 5,
|
||
hints: dict | None = None) -> list[dict]:
|
||
"""For each source line, the template rows it could plausibly belong to.
|
||
|
||
The aligner leaves some lines unplaced because nothing scored high enough,
|
||
and that is the wrong moment to guess: these are the lines a person should
|
||
decide. So rather than only reporting "no row", offer the best few
|
||
candidates by the same similarity the aligner uses, already scored and
|
||
ready to accept.
|
||
|
||
Returns one record per line:
|
||
{index, line, sheet, candidates: [{row, number, desc, score}]}
|
||
"""
|
||
wb = load_workbook(consultant_xlsx)
|
||
lines_all = source_lines(tree)
|
||
layouts, slots_by_sheet = {}, {}
|
||
for name in wb.sheetnames:
|
||
try:
|
||
lay = detect_layout(wb[name], (hints or {}).get(name))
|
||
except ValueError:
|
||
continue
|
||
layouts[name] = lay
|
||
slots = read_slots(wb[name], lay)
|
||
for sl in slots:
|
||
sl._sheet = name
|
||
slots_by_sheet[name] = slots
|
||
|
||
pool, _ = _relevant_pool(lines_all, slots_by_sheet)
|
||
# rank every row against every line; the documents are small enough
|
||
ranked = []
|
||
for ln in lines_all:
|
||
scored = []
|
||
for sl in pool:
|
||
sc = _sim(sl, ln.get("number", ""), ln.get("desc", ""))
|
||
if sc > 0.30:
|
||
scored.append({"sheet": sl._sheet, "row": sl.row,
|
||
"number": sl.number, "desc": sl.desc,
|
||
"score": round(sc, 3)})
|
||
scored.sort(key=lambda c: -c["score"])
|
||
ranked.append({"line": ln, "candidates": scored[:top_n]})
|
||
return ranked
|
||
|
||
|
||
def populate(consultant_xlsx: str, tree: dict, out_path: str, *,
|
||
sheet_hint: str | None = None,
|
||
overrides: dict | None = None,
|
||
placements: dict | None = None,
|
||
hints: dict | None = None) -> Result:
|
||
"""Write our Candy BOQ into a consultant's workbook, in place-safe fashion.
|
||
|
||
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()
|
||
original = load_workbook(consultant_xlsx) # pristine copy for comparison
|
||
wb = load_workbook(consultant_xlsx) # formulas preserved
|
||
lines = source_lines(tree)
|
||
if not lines:
|
||
raise ValueError("no priced lines found in the source tree")
|
||
|
||
names = [sheet_hint] if sheet_hint else list(wb.sheetnames)
|
||
layouts: dict[str, Layout] = {}
|
||
slots_by_sheet: dict[str, list[Slot]] = {}
|
||
|
||
for name in names:
|
||
if name not in wb.sheetnames:
|
||
continue
|
||
ws = wb[name]
|
||
try:
|
||
lay = detect_layout(ws, (hints or {}).get(name))
|
||
except ValueError as exc:
|
||
res.add("info", name, 0, "layout", str(exc))
|
||
continue
|
||
layouts[name] = lay
|
||
res.layout_notes.extend(f"[{name}] {n}" for n in lay.notes)
|
||
slots = read_slots(ws, lay)
|
||
for sl in slots:
|
||
sl._sheet = name
|
||
slots_by_sheet[name] = slots
|
||
|
||
if not slots_by_sheet:
|
||
raise ValueError("no BOQ-shaped sheet found in the workbook")
|
||
|
||
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.
|
||
#
|
||
# Two alternatives were tried and both are worse. Routing lines to sheets
|
||
# by section prefix fails because one sheet can span several sections
|
||
# (Blueberry's Schedule B holds sections 1-6). Aligning each section
|
||
# separately fails the same way, and a global pass also beat it: Central
|
||
# Towers went 23 unmatched against 34, Blueberry 1 against 14.
|
||
#
|
||
# The pool is already restricted to sheets that own a section named in the
|
||
# source, which is what keeps the candidate count sane. The residual
|
||
# unmatched lines are genuine ambiguities, and the review screen exists to
|
||
# put a person on those rather than to keep guessing here.
|
||
lay_of = {name: lay for name, lay in layouts.items()}
|
||
if placements:
|
||
# a person has decided these; drop the lines they claim from the
|
||
# automatic pass so the same row is not filled twice
|
||
claimed = {p[0] for p in placements.values()}
|
||
lines = [ln for i, ln in enumerate(lines) if i not in claimed]
|
||
free = [sl for sl in pool
|
||
if (sl._sheet, sl.row) not in set(placements.values())]
|
||
pool = free
|
||
_fill_pool(wb, lay_of, pool, lines, overrides, res)
|
||
|
||
if placements:
|
||
all_lines = source_lines(tree)
|
||
for idx, (sheet, row) in placements.items():
|
||
if idx >= len(all_lines) or sheet not in lay_of:
|
||
continue
|
||
ws = wb[sheet]
|
||
slot = Slot(row=row,
|
||
number=str(ws.cell(row, lay_of[sheet].col("item") or 1)
|
||
.value or "").strip(),
|
||
desc=_row_desc(ws, row, lay_of[sheet]))
|
||
slot._sheet = sheet
|
||
try:
|
||
_write_one(wb, sheet, lay_of[sheet], all_lines[idx], slot,
|
||
overrides, res)
|
||
except _FormulaProtected:
|
||
pass
|
||
|
||
_write_totals(wb, layouts, tree, res)
|
||
_restore_formulas(original, wb, res)
|
||
|
||
# A "MEGA CHECK" sheet at the front: every section the PDF priced against
|
||
# what this workbook now says, plus the rows that still need a person. The
|
||
# person at the spreadsheet works from the totals, so the totals are where
|
||
# the punch list belongs.
|
||
try:
|
||
from boq_engine.reconcile import reconcile, write_check_sheet
|
||
rec = reconcile(tree, wb, res, summary_values(tree))
|
||
n_bad, n_rows = write_check_sheet(wb, rec, out_name=out_path)
|
||
res.layout_notes.append(
|
||
f"MEGA CHECK written: {len(rec.checks)} of "
|
||
f"{len(rec.checks) + len(rec.unmatched)} section(s) compared, "
|
||
f"{n_bad} do not balance, {n_rows} line(s) flagged for a person")
|
||
except Exception as exc: # never block on this
|
||
res.layout_notes.append(f"MEGA CHECK not written: {exc}")
|
||
|
||
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 more will be added. 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.
|
||
"""
|
||
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 i, (ln, slot, score) in enumerate(pairs):
|
||
if slot is None:
|
||
res.add("unmapped", "-", 0,
|
||
f"{ln['number']} {ln['desc']}"[:70],
|
||
f"no row in the template for this line (p{ln['page']})",
|
||
source=f"{ln['qty']} {ln['unit']} @ {ln['rate']}",
|
||
severity="decision", index=i,
|
||
section=ln.get("sec", ""))
|
||
continue
|
||
name = slot._sheet
|
||
try:
|
||
_write_one(wb, name, layouts[name], ln, slot, overrides, res)
|
||
except _FormulaProtected:
|
||
continue # already recorded; nothing was written
|
||
|
||
|
||
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",
|
||
section=ln.get("sec", ""))
|
||
return
|
||
|
||
ws = wb[name]
|
||
r = slot.row
|
||
# The writer is the only thing that knows which line went on which row, so
|
||
# it records that explicitly. The verification then re-reads the row and
|
||
# asks whether it really is that line - which is the question the writer
|
||
# cannot answer about itself.
|
||
res.placed(name, r, ln, slot.desc or "")
|
||
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"])
|
||
res.wrote(name, r, c_unit, ln["unit"])
|
||
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:
|
||
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"])
|
||
|
||
# --- a tariff carried inside a sum ("INCL", "incl VAT") -----------------
|
||
# Written as the literal text the source printed. A blank cell here reads
|
||
# as a failure to the consultant; the source's own wording is the honest
|
||
# answer, and the amount column is left alone because there is no figure.
|
||
note = (ln.get("tariff_note") or "").strip()
|
||
if note and ln["rate"] is None and c_rate and \
|
||
ws.cell(r, c_rate).value in (None, ""):
|
||
if _writable(ws, r, c_rate, res, name, label, "rate"):
|
||
ws.cell(r, c_rate, note)
|
||
res.wrote(name, r, c_rate, note)
|
||
res.written += 1
|
||
res.add("info", name, r, label,
|
||
f"carried in the sum - the source prints \"{note}\" rather "
|
||
f"than a rate, so that wording was kept. There is no amount "
|
||
f"for this line.",
|
||
source=note)
|
||
|
||
# --- 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"])
|
||
res.wrote(name, 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"])
|
||
res.wrote(name, r, c_rate, ln["rate"])
|
||
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
|
||
|
||
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))
|
||
res.wrote(name, r, c_amt, amt_formula.format(r=r))
|
||
elif c_amt and ln["rate_only"]:
|
||
ws.cell(r, c_amt, "Rate only")
|
||
|
||
|
||
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("="):
|
||
# Not a disagreement and not a decision for the user - it is the tool
|
||
# working. Reported separately so it cannot dilute the findings that
|
||
# actually need a person.
|
||
res.add("formula", sheet, r, label,
|
||
f"{field} left as your own formula ({cur[:36]})",
|
||
severity="note", 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()
|
||
|
||
|
||
# A token that could plausibly be an item reference: digits and dots only.
|
||
REF_TOKEN = re.compile(r"^[\d.]+$")
|
||
|
||
|
||
SUMMARY_REF = re.compile(
|
||
r"^=\s*'?([^'!]+)'?!\$?([A-Z]{1,2})\$?(\d+)\s*$")
|
||
|
||
|
||
def summary_values(tree: dict) -> list[tuple[str, float]]:
|
||
"""The totals a COMBINED SUMMARY row expects, in document order.
|
||
|
||
A consultant template's summary sheet points at specific cells such as
|
||
='SCHEDULE B - ROADS'!G66, and those cells are blank in the template - so
|
||
without this the consultant opens the file and sees zero. The values come
|
||
from the source in the same order the summary lists them: each schedule,
|
||
then each of its sections.
|
||
"""
|
||
def leaf(node: dict) -> float:
|
||
if node.get("kind") == "line":
|
||
if node.get("rate_only") or node.get("amount") is None:
|
||
return 0.0
|
||
return float(node["amount"])
|
||
return sum(leaf(c) for c in node.get("children", []))
|
||
|
||
out: list[tuple[str, float]] = []
|
||
for branch in tree.get("children", []):
|
||
if branch.get("kind") not in ("group", "schedule", "section"):
|
||
continue
|
||
if branch.get("kind") != "schedule":
|
||
label = f"{branch.get('number','')} {branch.get('title','')}".strip()
|
||
val = branch.get("subtotal")
|
||
out.append((label, round(float(val), 2) if val is not None
|
||
else round(leaf(branch), 2)))
|
||
for sec in branch.get("children", []):
|
||
if sec.get("kind") != "section":
|
||
continue
|
||
label = f"{sec.get('number','')} {sec.get('title','')}".strip()
|
||
val = sec.get("subtotal")
|
||
out.append((label, round(float(val), 2) if val is not None
|
||
else round(leaf(sec), 2)))
|
||
return out
|
||
|
||
|
||
def _write_totals(wb, layouts: dict[str, Layout], tree: dict,
|
||
res: Result) -> None:
|
||
"""Fill the cells the workbook's own summary depends on."""
|
||
n_sheets_with_totals = 0
|
||
values = summary_values(tree)
|
||
vi = 0
|
||
|
||
# 1. the summary sheet, if there is one
|
||
for name in wb.sheetnames:
|
||
if "summary" not in name.lower():
|
||
continue
|
||
ws = wb[name]
|
||
for r in range(1, min(ws.max_row, 120) + 1):
|
||
for c in range(1, min(ws.max_column, 12) + 1):
|
||
m = SUMMARY_REF.match(str(ws.cell(r, c).value or ""))
|
||
if not m or vi >= len(values):
|
||
continue
|
||
target_sheet, col_letter, row_no = m.group(1), m.group(2), m.group(3)
|
||
if target_sheet not in wb.sheetnames:
|
||
continue
|
||
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,
|
||
f"summary row -> {target_sheet}!{col_letter}{row_no} "
|
||
f"= R{val:,.2f}")
|
||
|
||
# 2. any remaining TOTAL CARRIED FORWARD rows get a live SUM over the data
|
||
# rows in their own block, so the workbook still adds up on its own
|
||
for name, lay in layouts.items():
|
||
ws = wb[name]
|
||
c_amt = lay.col("amount")
|
||
if not c_amt or not lay.total_rows:
|
||
continue
|
||
letter_amt = get_column_letter(c_amt)
|
||
for tr in lay.total_rows:
|
||
if ws.cell(tr, c_amt).value is not None:
|
||
continue # already written by the summary pass
|
||
prev = lay.header_row
|
||
for r2 in lay.total_rows:
|
||
if r2 >= tr:
|
||
break
|
||
prev = r2
|
||
rows = [r for r in lay.data_rows if prev < r < tr]
|
||
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"
|
||
n_sheets_with_totals += 1
|
||
|
||
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")
|