MEGA INFRA
301d0fff44
Reads our Candy PDF (deterministically, no OCR or model needed) and populates the consultant's own workbook using their formulas, column layout and PAYMENT column. Nothing is ever silently overwritten: every disagreement is reported, can be settled inside the app, and regenerated. - extract Candy text PDF -> tree (x-position column detection) - template layout detect, sequence align, write, totals, overrides - validate qty x rate and children-vs-subtotal reconciliation - profiles per-consultant layout profiles, stored on a volume Shipped with a Dockerfile + docker-compose for Coolify.
642 lines
25 KiB
Python
642 lines
25 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)
|
||
|
||
def col(self, name: str) -> int | None:
|
||
return self.cols.get(name)
|
||
|
||
|
||
def detect_layout(ws) -> Layout:
|
||
"""Find the header row and map the columns."""
|
||
best: tuple[int, dict[str, int], int] | None = None
|
||
for r in range(1, min(ws.max_row, 25) + 1):
|
||
cols: dict[str, int] = {}
|
||
for c in range(1, min(ws.max_column, 30) + 1):
|
||
v = _cellsafe(ws.cell(r, c).value)
|
||
if not v or v in cols.values():
|
||
continue
|
||
for name, pats in HEADER_PATTERNS.items():
|
||
if name in cols:
|
||
continue
|
||
if v in pats or any(p in v for p in pats):
|
||
cols[name] = c
|
||
break
|
||
if len(cols) >= 4 and (best is None or len(cols) > len(best[1])):
|
||
best = (r, cols, len(cols))
|
||
|
||
if 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, cols, _ = best
|
||
lay = Layout(sheet=ws.title, header_row=header_row, cols=cols)
|
||
|
||
c_item = cols.get("item")
|
||
c_desc = cols.get("description")
|
||
c_qty = cols.get("qty")
|
||
|
||
for r in range(header_row + 1, ws.max_row + 1):
|
||
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")
|
||
return lay
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 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, cd = lay.col("item"), lay.col("description")
|
||
out: list[Slot] = []
|
||
for r in lay.data_rows:
|
||
num = str(ws.cell(r, ci).value or "").strip() if ci else ""
|
||
desc = str(ws.cell(r, cd).value or "").strip() if cd else ""
|
||
if not num and not desc:
|
||
continue
|
||
# indent level from the leading whitespace the consultant preserved
|
||
raw = str(ws.cell(r, cd).value or "") if cd else ""
|
||
indent = len(raw) - len(raw.lstrip()) if raw != raw.lstrip() else 0
|
||
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
|
||
elif not tnum and not snum:
|
||
num_score = 0.5 # neither numbered: description decides
|
||
else:
|
||
num_score = 0.0 # one side numbered, the other not
|
||
|
||
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
|
||
|
||
|
||
@dataclass
|
||
class Result:
|
||
written: int = 0
|
||
kept: int = 0
|
||
findings: list[Finding] = field(default_factory=list)
|
||
sheets: list[str] = field(default_factory=list)
|
||
layout_notes: list[str] = field(default_factory=list)
|
||
|
||
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
|
||
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 populate(consultant_xlsx: str, tree: dict, out_path: str, *,
|
||
sheet_hint: str | None = None,
|
||
overrides: dict | None = None) -> Result:
|
||
"""Write our Candy BOQ into a consultant's workbook, in place-safe fashion.
|
||
|
||
ONE alignment is run across every sheet, in workbook order, rather than
|
||
per sheet. The Candy PDF runs Schedule A then Schedule B; the workbook
|
||
does too. Aligning each sheet separately against the whole line list
|
||
scatters Schedule B's lines into Schedule A's rows.
|
||
|
||
`overrides` lets the consultant settle a disagreement from inside the app:
|
||
keys are (sheet, row, field) and the value is what should be written.
|
||
A cell with an override is written rather than left alone, and the
|
||
conflict is downgraded to 'resolved' so it stops being flagged.
|
||
"""
|
||
overrides = overrides or {}
|
||
res = Result()
|
||
res = Result()
|
||
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] = {}
|
||
combined: list[tuple[str, 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)
|
||
for slot in read_slots(ws, lay):
|
||
slot._sheet = name
|
||
combined.append((name, slot))
|
||
|
||
if not combined:
|
||
raise ValueError("no BOQ-shaped sheet found in the workbook")
|
||
|
||
pairs = align([sl for _, sl in combined], lines)
|
||
res.sheets = list(layouts)
|
||
|
||
for ln, slot, score in pairs:
|
||
if slot is None:
|
||
res.add("unmapped", "-", 0,
|
||
f"{ln['number']} {ln['desc']}"[:70],
|
||
f"source line (p{ln['page']}) has no row in any template sheet",
|
||
source=f"{ln['qty']} {ln['unit']} @ {ln['rate']}",
|
||
severity="decision")
|
||
continue
|
||
|
||
name, r = slot._sheet, slot.row
|
||
ws = wb[name]
|
||
lay = layouts[name]
|
||
c_unit, c_qty = lay.col("unit"), lay.col("qty")
|
||
c_rate, c_amt = lay.col("rate"), lay.col("amount")
|
||
label = f"{slot.number} {slot.desc}"[:70]
|
||
|
||
# build the amount formula from the columns we actually detected - a
|
||
# consultant template puts them wherever it likes, and assuming D/E
|
||
# silently multiplies UNIT x QTY
|
||
if c_qty and c_rate:
|
||
ql, rl = get_column_letter(c_qty), get_column_letter(c_rate)
|
||
amt_formula = (f'=IF(OR({ql}{{r}}="",{rl}{{r}}=""),"",'
|
||
f'{ql}{{r}}*{rl}{{r}})')
|
||
else:
|
||
amt_formula = None
|
||
|
||
# --- unit ------------------------------------------------------
|
||
if c_unit and ln["unit"]:
|
||
cur = ws.cell(r, c_unit).value
|
||
if cur in (None, ""):
|
||
ws.cell(r, c_unit, ln["unit"])
|
||
elif not units_compatible(cur, ln["unit"]):
|
||
hint = ""
|
||
if _norm_unit(ln["unit"]) == "?":
|
||
hint = (" -- Candy printed a corrupt unit character; the "
|
||
"template's real unit was kept, not overwritten")
|
||
ws.cell(r, c_unit, cur)
|
||
elif _norm_unit(cur) == "?":
|
||
hint = (" -- template unit is unreadable, source unit "
|
||
"written instead")
|
||
else:
|
||
hint = " -- left as the template has it"
|
||
okey = (name, r, "unit")
|
||
if okey in overrides:
|
||
ws.cell(r, c_unit, overrides[okey])
|
||
res.add("conflict", name, r, label,
|
||
f"unit: set to {overrides[okey]!r} by the consultant",
|
||
source=str(ln["unit"]), severity="resolved",
|
||
field="unit", template_value=cur,
|
||
source_value=ln["unit"])
|
||
else:
|
||
res.add("conflict", name, r, label,
|
||
f"unit: template {cur!r} vs source {ln['unit']!r}{hint}",
|
||
source=str(ln["unit"]),
|
||
severity="source_defect",
|
||
field="unit", template_value=cur,
|
||
source_value=ln["unit"])
|
||
|
||
# --- quantity ---------------------------------------------------
|
||
if c_qty and ln["qty"] is not None:
|
||
cur = ws.cell(r, c_qty).value
|
||
if cur in (None, ""):
|
||
ws.cell(r, c_qty, ln["qty"])
|
||
elif isinstance(cur, (int, float)) and abs(float(cur) - ln["qty"]) > 1e-6:
|
||
okey = (name, r, "qty")
|
||
if okey in overrides:
|
||
ws.cell(r, c_qty, overrides[okey])
|
||
res.add("conflict", name, r, label,
|
||
f"quantity: set to {overrides[okey]} by the consultant",
|
||
source=str(ln["qty"]), severity="resolved",
|
||
field="qty", template_value=cur,
|
||
source_value=ln["qty"])
|
||
else:
|
||
if ln["rate_only"] or ln["qty"] == 0:
|
||
reason = ("source carries no quantity here (printed 'Rate "
|
||
"only'), so the template's own figure was kept")
|
||
sev = "source_gap"
|
||
else:
|
||
reason = ("the two documents disagree; NEITHER was "
|
||
"overwritten")
|
||
sev = "decision"
|
||
res.add("conflict", name, r, label,
|
||
f"quantity: template {cur} vs source {ln['qty']} - {reason}",
|
||
source=str(ln["qty"]), severity=sev,
|
||
field="qty", template_value=cur,
|
||
source_value=ln["qty"])
|
||
res.written += 1
|
||
else:
|
||
res.kept += 1
|
||
# already correct - still make sure a rate landed
|
||
if c_rate and ln["rate"] is not None and not ln["rate_only"]:
|
||
if ws.cell(r, c_rate).value in (None, ""):
|
||
ws.cell(r, c_rate, ln["rate"])
|
||
if c_amt:
|
||
ws.cell(r, c_amt, amt_formula.format(r=r))
|
||
res.written += 1
|
||
continue
|
||
|
||
# --- rate + amount ----------------------------------------------
|
||
if c_rate and ln["rate"] is not None:
|
||
ws.cell(r, c_rate, ln["rate"])
|
||
if c_amt:
|
||
if ln["rate_only"]:
|
||
ws.cell(r, c_amt, "Rate only")
|
||
res.add("info", name, r, label,
|
||
"'Rate only' in source - rate written, amount as text",
|
||
source=str(ln["rate"]))
|
||
else:
|
||
# live formula so the consultant can re-rate and watch the
|
||
# total follow. Never a frozen value.
|
||
if amt_formula:
|
||
ws.cell(r, c_amt, amt_formula.format(r=r))
|
||
res.written += 1
|
||
elif c_rate and ln["rate_only"] and ln["amount"] is None:
|
||
if c_amt:
|
||
ws.cell(r, c_amt, "Rate only")
|
||
|
||
_write_totals(wb, layouts, tree, res)
|
||
wb.save(out_path)
|
||
return res
|
||
|
||
|
||
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."""
|
||
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]
|
||
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]
|
||
if rows:
|
||
ws.cell(tr, c_amt,
|
||
f"=SUM({letter_amt}{min(rows)}:{letter_amt}{max(rows)})"
|
||
).number_format = "#,##0.00"
|
||
res.add("total", name, lay.total_rows[0] if lay.total_rows else 0,
|
||
f"{len(lay.total_rows)} TOTAL row(s)",
|
||
"remaining TOTAL rows given a live SUM over their own block")
|