MEGA INFRA
d269c66b83
The engineer found it: '.02 The use of equipment referred to in item
110.04.01' had no rate, while the row above it had been given R764.75 - which
is '.03 Excavation by hand's rate. A wrong figure, in the wrong line, that
reads as a normal price.
The line is split across two rows - the label and the first part of the
description on one, the figures on the next:
r74 C=.02 The use of equipment referred to in item
r75 110.04.01 above L.Sum 1.00 =+G75*H75
Those halves were separate candidates, so a neighbouring line could match the
wrong one. The figures row now becomes the slot's row - which is where the
columns are - while keeping the joined description so matching still sees the
whole line.
Also found while in there: 'Carried forward' rows carry text in the item
column, so they were being treated as placeable lines and the real line above
merged down into one. They are now never placement targets.
The counts look slightly worse and the file is actually better:
written unplaced the block in question
before 196 22 '.02' row carried R764.75 - wrong
after 191 24 '.02' blank and listed; '.03' R764.75 and
'.04' R25 000 both correct
Four more lines need a click. One wrong rate in a neighbouring line is worth
more than four clicks - a blank is visible and safe, a wrong price is not.
Blueberry is unchanged at 103 written / 1 unplaced, so the simpler template is
untouched. Zero formulas damaged on both, gate not blocked.
1187 lines
49 KiB
Python
1187 lines
49 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]}")
|
||
|
||
|
||
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 [])
|
||
for r in range(first, ws.max_row + 1):
|
||
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)
|
||
# ".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 LETTER_ONLY.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
|
||
|
||
|
||
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)
|
||
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 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."""
|
||
out: list[dict] = []
|
||
|
||
def walk(node: dict) -> None:
|
||
kind = node.get("kind")
|
||
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),
|
||
})
|
||
return
|
||
for c in node.get("children", []):
|
||
walk(c)
|
||
|
||
walk(tree)
|
||
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)
|
||
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)
|
||
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")
|
||
return
|
||
|
||
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"])
|
||
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")
|