MEGA INFRA
51e46000ec
Automatic coverage is now 203 of 226 on Central Towers (90%), up from 65. The Unplaced tab lists every remaining line with its five best candidate rows, ranked and scored, because where a description repeats in the document only a person can say which occurrence is which. Pick one or leave it out; chosen lines are written on the next run through the same verification gate. Two engine fixes found while building it: - only a real reference may be carried down a blank item cell. The wrapped header 'ITEM'/'NO' left a literal NO in that column and every row beneath inherited it, so all of 1200LE rows 73-120 read as a mismatch. - the '.NN' is not part of a line's identity. The source reuses one number space for a parent heading and its first child, and prints '.01' twice in a row; the consultant's template copies that duplication while its group column stays sequential. Also fixed: two tab registrations that had silently never applied, and a NameError in the Verification tab. Tried and reverted: per-section alignment. Blueberry 1 unmatched -> 14, Central 23 -> 34. One sheet can span several sections, so splitting is worse.
1025 lines
41 KiB
Python
1025 lines
41 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)
|
||
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) -> Layout:
|
||
"""Find the header row and map the columns.
|
||
|
||
Each candidate row is scored twice - read alone, and joined with the row
|
||
below it - because a wrapped label only reads correctly when the two halves
|
||
are put back together.
|
||
"""
|
||
ncols = min(ws.max_column, 30)
|
||
cands: list[tuple[int, int, int, dict[str, int]]] = []
|
||
for r in range(1, min(ws.max_row, 25) + 1):
|
||
for span in (1, 2):
|
||
if r + span - 1 > ws.max_row:
|
||
continue
|
||
cols = _score_header_row(ws, r, ncols, span)
|
||
if len(cols) >= 4:
|
||
cands.append((len(cols), r, span, cols))
|
||
# most roles found wins; a tie goes to the LOWER row, because the genuine
|
||
# header block is the last one before the data starts. Getting this wrong
|
||
# cost the quantity column on 49 of 51 sheets.
|
||
best = max(cands, key=lambda t: (t[0], t[1], t[2])) if cands else None
|
||
|
||
if best is None:
|
||
raise ValueError(
|
||
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}")
|
||
|
||
c_item = cols.get("item")
|
||
c_desc = cols.get("description")
|
||
c_qty = cols.get("qty")
|
||
|
||
for r in range(header_row + span, ws.max_row + 1):
|
||
a = ws.cell(r, c_item).value if c_item else None
|
||
d = ws.cell(r, c_desc).value if c_desc else None
|
||
q = ws.cell(r, c_qty).value if c_qty else None
|
||
label = f"{a or ''} {d or ''}".strip()
|
||
|
||
if label and TOTAL_RE.match(label):
|
||
lay.total_rows.append(r)
|
||
continue
|
||
# a real data row: something in item, description or qty
|
||
if any(x not in (None, "") for x in (a, d)) or q is not None:
|
||
lay.data_rows.append(r)
|
||
|
||
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")
|
||
|
||
_verify_against_data(ws, lay)
|
||
_derive_desc_run(lay)
|
||
return lay
|
||
|
||
|
||
def _derive_desc_run(lay: Layout) -> None:
|
||
"""Everything between the item column and the unit column is description."""
|
||
lo = lay.cols.get("item")
|
||
hi = lay.cols.get("unit")
|
||
if not lo or not hi or hi <= lo:
|
||
lay.desc_cols = [lay.cols["description"]] if "description" in lay.cols else []
|
||
return
|
||
run = [c for c in range(lo + 1, hi) if c not in lay.cols.values()]
|
||
# plus the claimed description column even if it sits outside the run
|
||
d = lay.cols.get("description")
|
||
if d and d not in run and d not in (lo, hi):
|
||
run.append(d)
|
||
lay.desc_cols = sorted(set(run))
|
||
if len(lay.desc_cols) > 1:
|
||
lay.notes.append(
|
||
f"description spans columns "
|
||
f"{[get_column_letter(c) for c in lay.desc_cols]}")
|
||
|
||
|
||
def _verify_against_data(ws, lay: Layout) -> None:
|
||
"""Correct the header-derived column map using what the data actually holds.
|
||
|
||
A header row is a claim; the cells below it are the fact. On the SANS 1200
|
||
template the header advertises DESCRIPTION in column E while the text sits
|
||
in D, and a wrong description column quietly wrecks every line match.
|
||
"""
|
||
sample = lay.data_rows[3:60]
|
||
if len(sample) < 5:
|
||
return
|
||
|
||
def longest_text(col: int) -> tuple[int, int]:
|
||
"""(count of substantial strings, longest) in a column over the sample."""
|
||
n = longest = 0
|
||
for r in sample:
|
||
v = ws.cell(r, col).value
|
||
if isinstance(v, str) and len(v.strip()) > 6:
|
||
n += 1
|
||
longest = max(longest, len(v.strip()))
|
||
return n, longest
|
||
|
||
# --- description: the header may be off by one --------------------------
|
||
d = lay.cols.get("description")
|
||
if d:
|
||
have, _ = longest_text(d)
|
||
best_col, best_score = d, 0.0
|
||
for cand in (d - 1, d + 1):
|
||
if cand < 1 or cand in lay.cols.values():
|
||
continue
|
||
cnt, longest = longest_text(cand)
|
||
# a real description column: nearly every sampled row carries long
|
||
# text. Score it, and only switch if it clearly beats the claim.
|
||
score = (cnt / max(1, len(sample))) * (1 if longest > 25 else 0.3)
|
||
if score > best_score:
|
||
best_col, best_score = cand, score
|
||
if best_col != d and best_score > 0.6 and have < len(sample) * 0.2:
|
||
lay.notes.append(
|
||
f"header claims DESCRIPTION in col {d} but the text is in "
|
||
f"col {best_col} - using {get_column_letter(best_col)}")
|
||
lay.cols["description"] = best_col
|
||
|
||
# --- unit: short, mostly-textual values sitting beside the quantity -----
|
||
u = lay.cols.get("unit")
|
||
if u:
|
||
def unitish(col: int) -> int:
|
||
n = 0
|
||
for r in sample:
|
||
v = ws.cell(r, col).value
|
||
if isinstance(v, str) and 1 < len(v.strip()) < 12:
|
||
n += 1
|
||
return n
|
||
|
||
have = unitish(u)
|
||
if have < len(sample) * 0.3:
|
||
for cand in (u - 1, u + 1):
|
||
if cand < 1 or cand in lay.cols.values():
|
||
continue
|
||
if unitish(cand) >= len(sample) * 0.5:
|
||
lay.notes.append(
|
||
f"unit column shifted {u} -> {cand} to match the data")
|
||
lay.cols["unit"] = cand
|
||
break
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# matching
|
||
# --------------------------------------------------------------------------
|
||
@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}")
|
||
|
||
|
||
def read_slots(ws, lay: Layout) -> list[Slot]:
|
||
ci = lay.col("item")
|
||
run = lay.desc_cols or ([lay.col("description")] if lay.col("description") else [])
|
||
out: list[Slot] = []
|
||
# A spreadsheet implies its hierarchy: once "110.01" is printed, the blank
|
||
# cells below it still belong to 110.01. Without carrying that down, two
|
||
# thirds of the rows read as unnumbered and nothing can match them.
|
||
carried = ""
|
||
for r in lay.data_rows:
|
||
raw = str(ws.cell(r, ci).value or "").strip() if ci else ""
|
||
# Only a real reference may be carried down. The wrapped header
|
||
# "ITEM" / "NO" leaves a literal "NO" in that column, and propagating
|
||
# it as an item number made every row beneath look like a mismatch -
|
||
# 1200LE rows 73-120 all inherited "NO" and became unmatchable.
|
||
if REF_TOKEN.match(raw):
|
||
carried = raw
|
||
elif raw:
|
||
carried = "" # a real value, but not a reference
|
||
num = carried
|
||
parts = [str(ws.cell(r, c).value or "").strip() for c in run]
|
||
desc = " ".join(p for p in parts if p).strip()
|
||
if not num and not desc:
|
||
continue
|
||
# indent level from the leading whitespace the consultant preserved
|
||
raw = next((str(ws.cell(r, c).value or "") for c in run
|
||
if str(ws.cell(r, c).value or "") !=
|
||
str(ws.cell(r, c).value or "").lstrip()), "")
|
||
indent = len(raw) - len(raw.lstrip()) if raw else 0
|
||
out.append(Slot(row=r, number=num, desc=desc, indent=indent))
|
||
return out
|
||
|
||
|
||
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 "",
|
||
"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 units_compatible(a: str, b: str) -> bool:
|
||
na, nb = _norm_unit(a), _norm_unit(b)
|
||
if na == nb:
|
||
return True
|
||
ca, cb = UNIT_ALIASES.get(na), UNIT_ALIASES.get(nb)
|
||
return ca is not None and ca == cb
|
||
|
||
|
||
def _section_of(number: str) -> str:
|
||
"""The leading segment of a reference: "110.01.01" -> "110".
|
||
|
||
SANS 1200 sheets are named for the section they cover, so this is what
|
||
locates a sheet. It is NOT a unique key on its own - the Candy PDF restarts
|
||
".01" under every repeated "110...." heading, so "110.01" names three
|
||
different lines in one document.
|
||
"""
|
||
n = _cellsafe(number)
|
||
return n.split(".")[0] if n else ""
|
||
|
||
|
||
def _relevant_pool(lines: list[dict],
|
||
slots_by_sheet: dict[str, list[Slot]]
|
||
) -> tuple[list[Slot], dict[str, int]]:
|
||
"""The slots worth aligning against, in workbook order.
|
||
|
||
Aligning against every sheet in the workbook looks fine on a 3-sheet
|
||
template and falls apart on a 53-sheet one: with 1314 candidate rows for
|
||
226 lines, the aligner starts cherry-picking scattered high scores instead
|
||
of finding the one coherent run. So the pool is cut down to the sheets that
|
||
actually own a section named in the source, which leaves 12 candidates
|
||
instead of 1314 and lets the sequence aligner work as designed.
|
||
"""
|
||
wanted = {_section_of(ln.get("number", "")) for ln in lines}
|
||
wanted.discard("")
|
||
|
||
pool: list[Slot] = []
|
||
counts: dict[str, int] = {}
|
||
for sheet, slots in slots_by_sheet.items():
|
||
secs = {_section_of(sl.number) for sl in slots}
|
||
secs.discard("")
|
||
if wanted and not (secs & wanted):
|
||
continue
|
||
pool.extend(slots)
|
||
counts[sheet] = len(slots)
|
||
return pool, counts
|
||
|
||
|
||
def _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) -> 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])
|
||
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) -> 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)
|
||
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"])
|
||
|
||
# --- 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("="):
|
||
res.add("conflict", sheet, r, label,
|
||
f"{field}: that cell holds a formula ({cur[:40]}) - your model "
|
||
f"was left alone and the source value was not written",
|
||
severity="decision", field=field, template_value=cur)
|
||
return False
|
||
return True
|
||
|
||
|
||
def _guard_formula(ws, r: int, col: int, res: Result, sheet: str,
|
||
label: str) -> None:
|
||
"""Refuse to clobber a cell that holds a formula.
|
||
|
||
On a SANS 1200 template the quantities are DERIVED - "=G35",
|
||
"=CEILING.MATH(414.065+302.817,10)" - from a Measurement sheet. Writing
|
||
Candy's number over one of those destroys the engineer's model, and it
|
||
looks like it worked.
|
||
"""
|
||
cur = ws.cell(r, col).value
|
||
if isinstance(cur, str) and cur.startswith("="):
|
||
res.add("conflict", sheet, r, label,
|
||
f"that cell holds a FORMULA ({cur[:44]}) - your model's value "
|
||
f"was left in place, and the source quantity was not written",
|
||
severity="decision", field="qty",
|
||
template_value=cur, source_value=None)
|
||
raise _FormulaProtected()
|
||
|
||
|
||
# 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")
|