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.
219 lines
7.3 KiB
Python
219 lines
7.3 KiB
Python
"""
|
|
Validation + reconciliation for an extracted BOQ.
|
|
|
|
This is the part that actually earns its keep. A BOQ is a priced legal
|
|
document: the rates go into a Form of Offer and a contract sum. Silently
|
|
mistranscribing one line is a real financial exposure, so every extraction is
|
|
re-checked against the document's own arithmetic before it is allowed out.
|
|
|
|
Three classes of check:
|
|
1. ARITHMETIC - qty x rate == printed amount, and children sum to the
|
|
group's printed subtotal.
|
|
2. STRUCTURE - duplicate / out-of-order / malformed item numbers, and
|
|
text that landed in the wrong column.
|
|
3. COMPLETENESS- sections present in the doc but empty, rate-only items,
|
|
provisional sums.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
TOL = 0.02 # half a cent
|
|
|
|
|
|
@dataclass
|
|
class Issue:
|
|
severity: str # 'error' | 'warn' | 'info'
|
|
code: str
|
|
where: str
|
|
detail: str
|
|
suggested: str = ""
|
|
|
|
|
|
@dataclass
|
|
class Report:
|
|
issues: list[Issue] = field(default_factory=list)
|
|
checked_lines: int = 0
|
|
checked_groups: int = 0
|
|
|
|
@property
|
|
def errors(self) -> list[Issue]:
|
|
return [i for i in self.issues if i.severity == "error"]
|
|
|
|
@property
|
|
def warnings(self) -> list[Issue]:
|
|
return [i for i in self.issues if i.severity == "warn"]
|
|
|
|
def as_dict(self) -> dict:
|
|
return {
|
|
"checked_lines": self.checked_lines,
|
|
"checked_groups": self.checked_groups,
|
|
"errors": len(self.errors),
|
|
"warnings": len(self.warnings),
|
|
"issues": [i.__dict__ for i in self.issues],
|
|
}
|
|
|
|
|
|
def _label(node: dict) -> str:
|
|
n = node.get("number") or ""
|
|
t = node.get("title") or node.get("desc") or ""
|
|
return f"{n} {t}".strip()[:60]
|
|
|
|
|
|
def _leaf_rows(node: dict) -> list[dict]:
|
|
out = []
|
|
for c in node.get("children", []):
|
|
if c.get("kind") in ("line", "item") and (
|
|
c.get("qty") is not None or c.get("rate") is not None
|
|
):
|
|
out.append(c)
|
|
out.extend(_leaf_rows(c))
|
|
return out
|
|
|
|
|
|
def _line_amount(line: dict) -> float:
|
|
"""What the line contributes to its group total."""
|
|
if line.get("rate_only") or line.get("amount") is None:
|
|
return 0.0
|
|
return float(line["amount"])
|
|
|
|
|
|
def _computed_amount(line: dict) -> float | None:
|
|
q, r = line.get("qty"), line.get("rate")
|
|
if q is None or r is None:
|
|
return None
|
|
return round(q * r, 2)
|
|
|
|
|
|
def _direct_leaves(node: dict) -> list[dict]:
|
|
"""Priced leaf lines that roll up into THIS group's printed subtotal.
|
|
|
|
Recurses through numbered *item* nodes (2.6 -> 2.6.1 -> a)) because a
|
|
Candy subtotal covers the whole branch, but stops at nested *groups* so a
|
|
section total is never double counted.
|
|
"""
|
|
out: list[dict] = []
|
|
|
|
def rec(n: dict) -> None:
|
|
for c in n.get("children", []):
|
|
k = c.get("kind")
|
|
if k == "group":
|
|
continue # nested group: its own subtotal
|
|
if k in ("item", "line"):
|
|
kids = [g for g in c.get("children", [])
|
|
if g.get("kind") in ("item", "line")]
|
|
if kids:
|
|
rec(c)
|
|
elif c.get("qty") is not None or c.get("rate") is not None:
|
|
out.append(c)
|
|
continue
|
|
if k in ("section", "schedule"):
|
|
continue
|
|
|
|
rec(node)
|
|
return out
|
|
|
|
|
|
def validate(tree: dict) -> Report:
|
|
rep = Report()
|
|
_walk(tree, rep, path="BOQ")
|
|
return rep
|
|
|
|
|
|
def _child_group_subtotals(node: dict) -> list[dict]:
|
|
out = []
|
|
for c in node.get("children", []):
|
|
if c.get("kind") == "group" and c.get("subtotal") is not None:
|
|
out.append(c)
|
|
return out
|
|
|
|
|
|
def _reconcile(node: dict) -> tuple[float | None, str, int]:
|
|
"""Work out what this node's printed subtotal *should* be.
|
|
|
|
Candy rolls a total up in one of two ways, so try both and report which
|
|
one actually applied:
|
|
- 'leaves' : the total covers the priced lines directly beneath it
|
|
- 'groups' : the total covers nested numbered groups instead
|
|
"""
|
|
direct = _direct_leaves(node)
|
|
if direct:
|
|
return (round(sum(_line_amount(l) for l in direct), 2),
|
|
"leaves", len(direct))
|
|
kids = _child_group_subtotals(node)
|
|
if kids:
|
|
return (round(sum(float(k["subtotal"]) for k in kids), 2),
|
|
"groups", len(kids))
|
|
return None, "none", 0
|
|
|
|
|
|
def _walk(node: dict, rep: Report, path: str) -> None:
|
|
label = _label(node) or node.get("kind", "")
|
|
here = f"{path} > {label}" if label else path
|
|
|
|
# ---- group subtotal reconciliation ---------------------------------
|
|
if node.get("kind") in ("group", "section", "schedule", "item") \
|
|
and node.get("subtotal") is not None:
|
|
rep.checked_groups += 1
|
|
expected, basis, n = _reconcile(node)
|
|
if expected is not None:
|
|
printed = round(float(node["subtotal"]), 2)
|
|
if abs(expected - printed) > TOL:
|
|
rep.issues.append(Issue(
|
|
"error", "SUBTOTAL_MISMATCH", here,
|
|
f"printed {printed:,.2f} but its own {basis} "
|
|
f"({'lines' if basis == 'leaves' else 'sub-groups'}) "
|
|
f"sum to {expected:,.2f} — difference {printed - expected:,.2f}",
|
|
"Confirm against the source before this goes into a Form "
|
|
"of Offer; the error propagates to the contract sum.",
|
|
))
|
|
|
|
# ---- leaf lines -----------------------------------------------------
|
|
for child in node.get("children", []):
|
|
if child.get("kind") == "line":
|
|
_check_line(child, rep, here)
|
|
else:
|
|
_walk(child, rep, here)
|
|
|
|
|
|
def _check_line(line: dict, rep: Report, where: str) -> None:
|
|
if line.get("qty") is None and line.get("rate") is None:
|
|
return
|
|
rep.checked_lines += 1
|
|
lab = f"{line.get('number','')} {line.get('desc','')}".strip()[:60]
|
|
|
|
calc = _computed_amount(line)
|
|
if calc is not None and line.get("amount") is not None and not line.get("rate_only"):
|
|
if abs(calc - float(line["amount"])) > TOL:
|
|
rep.issues.append(Issue(
|
|
"error", "LINE_MISMATCH", where,
|
|
f"{lab}: qty {line['qty']} x rate {line['rate']} = {calc:,.2f} "
|
|
f"but printed amount is {float(line['amount']):,.2f}",
|
|
))
|
|
|
|
if not line.get("unit"):
|
|
rep.issues.append(Issue(
|
|
"warn", "NO_UNIT", where,
|
|
f"{lab}: has a price but no unit - column may have shifted",
|
|
))
|
|
|
|
if line.get("rate") is None and line.get("qty") is not None and not line.get("rate_only"):
|
|
rep.issues.append(Issue(
|
|
"info", "PERCENT_LINE", where,
|
|
f"{lab}: qty {line['qty']:,.0f} with no rate (percentage/markup line)",
|
|
))
|
|
|
|
if line.get("rate_only"):
|
|
rep.issues.append(Issue(
|
|
"info", "RATE_ONLY", where,
|
|
f"{lab}: priced 'Rate only' (qty {line.get('qty')}) - rate is "
|
|
f"required but contributes R0.00 to the total",
|
|
))
|
|
|
|
for f in line.get("flags", []):
|
|
if f == "empty":
|
|
rep.issues.append(Issue(
|
|
"warn", "EMPTY_LINE", where, f"{lab}: row has no qty, rate or amount",
|
|
))
|