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.
491 lines
18 KiB
Python
491 lines
18 KiB
Python
"""
|
|
BOQ PDF -> structured tree extractor.
|
|
|
|
Built for South African municipal / civil tender BOQs produced by Candy
|
|
"Continuous Listing" print layouts. Those PDFs are *text* (printed from Excel),
|
|
not scans, so the approach is deterministic column detection by x-position
|
|
rather than OCR or an LLM.
|
|
|
|
Column geometry observed on Candy 2.05p output (A4 portrait):
|
|
Item left edge ~ x0 = 32
|
|
Description left edge ~ x0 = 70
|
|
Unit left edge ~ x0 = 302
|
|
Qty RIGHT aligned, x1 ~ 400-411
|
|
Rate RIGHT aligned, x1 ~ 453-484
|
|
Amount RIGHT aligned, x1 ~ 512-565
|
|
|
|
Numeric columns are right-aligned, so RIGHT edges cluster far more tightly than
|
|
left edges and are the reliable signal. We therefore bucket by x1.
|
|
|
|
Three passes:
|
|
1. rows - words grouped by y, split into columns
|
|
2. content - noise stripped, description continuations merged
|
|
3. tree - content rows assembled into the numbered hierarchy
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from dataclasses import dataclass, field, asdict
|
|
from typing import Any
|
|
|
|
import pdfplumber
|
|
|
|
# --------------------------------------------------------------------------
|
|
# column geometry
|
|
# --------------------------------------------------------------------------
|
|
BANDS = {
|
|
"qty": (380.0, 436.0),
|
|
"rate": (436.0, 500.0),
|
|
"amount": (500.0, 575.0),
|
|
}
|
|
|
|
ITEM_X_MAX = 60.0
|
|
DESC_X_MIN, DESC_X_MAX = 60.0, 296.0
|
|
UNIT_X_MIN, UNIT_X_MAX = 296.0, 380.0
|
|
|
|
# running header / footer / column-heading chrome
|
|
NOISE_PATTERNS = [
|
|
re.compile(r"\d{1,2}:\d{2}:\d{2}"), # "Printed 08:21:20 ..."
|
|
re.compile(r"page\s+\d+", re.I),
|
|
re.compile(r"^\d{1,2}\s+\w+\s+\d{4}\s*$"), # "1 August 2026"
|
|
re.compile(r"^candy\b", re.I),
|
|
re.compile(r"continuous\s+listing", re.I),
|
|
re.compile(r"\bbill\s+description\b", re.I),
|
|
re.compile(r"^\d{2}-\d{3}\b"), # "26-048 - Blueberry Square"
|
|
re.compile(r"^boq\s*-", re.I),
|
|
re.compile(r"^selling$", re.I),
|
|
]
|
|
# the repeated column-heading row: item | bill description | unit | ...
|
|
HEADER_TOKENS = {"item", "unit", "rate", "amount", "bill", "quantity"}
|
|
|
|
SECTION_RE = re.compile(r"^SECTION\s+(\d+)\s*[:\-]\s*(.+)$", re.I)
|
|
ITEM_RE = re.compile(r"^\d{1,2}(?:\.{1,2}\d{1,2})*\.?$")
|
|
LETTER_RE = re.compile(r"^\(?([a-z])\)\s*(.*)$", re.S)
|
|
PLACEHOLDER_RE = re.compile(r"^[.\s]{6,}$")
|
|
RATE_ONLY_RE = re.compile(r"rate\s*only", re.I)
|
|
SCHEDULE_RE = re.compile(r"^SCHEDULE\s+([A-Z])\b", re.I)
|
|
|
|
|
|
def _to_num(tok: str) -> float | None:
|
|
t = (tok or "").replace(",", "").replace(" ", "")
|
|
if not t or t in {"-", "."}:
|
|
return None
|
|
try:
|
|
return float(t)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _classify(x1: float) -> str | None:
|
|
for name, (lo, hi) in BANDS.items():
|
|
if lo <= x1 < hi:
|
|
return name
|
|
return None
|
|
|
|
|
|
# ==========================================================================
|
|
# pass 1 - rows
|
|
# ==========================================================================
|
|
@dataclass
|
|
class Row:
|
|
item: str = ""
|
|
desc: str = ""
|
|
unit: str = ""
|
|
qty: float | None = None
|
|
rate: float | None = None
|
|
amount_txt: str = ""
|
|
top: float = 0.0
|
|
page: int = 0
|
|
|
|
@property
|
|
def has_price(self) -> bool:
|
|
return self.qty is not None or self.rate is not None
|
|
|
|
@property
|
|
def is_empty(self) -> bool:
|
|
return not (self.item or self.desc or self.unit
|
|
or self.has_price or self.amount_txt)
|
|
|
|
|
|
def _page_rows(page, pno: int) -> list[Row]:
|
|
buckets: dict[int, list] = {}
|
|
for w in page.extract_words():
|
|
buckets.setdefault(round(w["top"] / 3.0), []).append(w)
|
|
|
|
rows: list[Row] = []
|
|
for key in sorted(buckets):
|
|
words = sorted(buckets[key], key=lambda z: z["x0"])
|
|
r = Row(top=words[0]["top"], page=pno)
|
|
desc_parts: list[str] = []
|
|
last_desc_x1 = None
|
|
for w in words:
|
|
x0, x1, txt = w["x0"], w["x1"], w["text"]
|
|
if x1 <= ITEM_X_MAX:
|
|
r.item = (r.item + " " + txt).strip()
|
|
continue
|
|
col = _classify(x1)
|
|
if col == "qty" and r.qty is None:
|
|
r.qty = _to_num(txt)
|
|
continue
|
|
if col == "rate" and r.rate is None:
|
|
r.rate = _to_num(txt)
|
|
continue
|
|
if col == "amount":
|
|
r.amount_txt = (r.amount_txt + " " + txt).strip()
|
|
continue
|
|
if UNIT_X_MIN <= x0 < UNIT_X_MAX:
|
|
r.unit = (r.unit + txt).strip()
|
|
continue
|
|
if DESC_X_MIN <= x0 < DESC_X_MAX:
|
|
# Candy's print engine sometimes splits a word across two text
|
|
# runs ("S" + "ITE CLEARANCE"). Only heal a genuine fragment:
|
|
# a tiny orphan sitting flush against the previous word.
|
|
if (desc_parts and last_desc_x1 is not None
|
|
and len(desc_parts[-1]) <= 2
|
|
and x0 - last_desc_x1 < 0.9
|
|
and desc_parts[-1].isalpha()
|
|
and txt.isalpha()):
|
|
desc_parts[-1] += txt
|
|
else:
|
|
desc_parts.append(txt)
|
|
last_desc_x1 = x1
|
|
r.desc = " ".join(desc_parts).strip()
|
|
rows.append(r)
|
|
return rows
|
|
|
|
|
|
# ==========================================================================
|
|
# pass 2 - strip noise, merge continuations
|
|
# ==========================================================================
|
|
def _is_noise(r: Row) -> bool:
|
|
blob = f"{r.item} {r.desc} {r.unit}".strip()
|
|
if r.has_price:
|
|
return False
|
|
if any(p.search(blob) for p in NOISE_PATTERNS):
|
|
return True
|
|
toks = {t.lower().strip(".,:") for t in blob.split()}
|
|
if toks and toks <= HEADER_TOKENS:
|
|
return True
|
|
return False
|
|
|
|
|
|
def _looks_like_heading(text: str) -> bool:
|
|
"""True for text that starts a new block rather than continuing one."""
|
|
t = text.strip()
|
|
if not t:
|
|
return False
|
|
if SECTION_RE.match(t) or SCHEDULE_RE.match(t):
|
|
return True
|
|
# an all-caps run of words is a heading ("TEMPORARY WORKS", "SUBBASE")
|
|
letters = [c for c in t if c.isalpha()]
|
|
if letters and all(c.isupper() for c in letters) and len(letters) > 3:
|
|
return True
|
|
return False
|
|
|
|
|
|
def _merge_continuations(rows: list[Row]) -> list[Row]:
|
|
"""A row with no number, no unit, no price and no amount is a wrapped
|
|
description belonging to the row above it -- unless it opens a new heading.
|
|
"""
|
|
out: list[Row] = []
|
|
for r in rows:
|
|
is_cont = (
|
|
not r.item
|
|
and not r.unit
|
|
and not r.has_price
|
|
and not r.amount_txt
|
|
and r.desc
|
|
and not _looks_like_heading(r.desc)
|
|
)
|
|
if is_cont and out:
|
|
prev = out[-1]
|
|
if prev.desc and not prev.is_empty:
|
|
prev.desc = f"{prev.desc} {r.desc}".strip()
|
|
continue
|
|
out.append(r)
|
|
return out
|
|
|
|
|
|
# ==========================================================================
|
|
# pass 3 - build the tree
|
|
# ==========================================================================
|
|
@dataclass
|
|
class Line:
|
|
kind: str = "line"
|
|
number: str = ""
|
|
desc: str = ""
|
|
unit: str = ""
|
|
qty: float | None = None
|
|
rate: float | None = None
|
|
amount: float | None = None
|
|
rate_only: bool = False
|
|
children: list = field(default_factory=list)
|
|
page: int = 0
|
|
flags: list = field(default_factory=list)
|
|
|
|
|
|
@dataclass
|
|
class Group:
|
|
kind: str = "group" # 'root' | 'section' | 'schedule' | 'group' | 'item'
|
|
number: str = ""
|
|
title: str = ""
|
|
subtotal: float | None = None
|
|
children: list = field(default_factory=list)
|
|
page: int = 0
|
|
flags: list = field(default_factory=list)
|
|
|
|
|
|
class _Tree:
|
|
def __init__(self) -> None:
|
|
self.root = Group(kind="root", title="BOQ")
|
|
self.stack: list[Group] = [self.root]
|
|
self.item_holder: Group | None = None
|
|
# Candy prints a heading when the block opens and the SAME number
|
|
# again with its subtotal when the block closes. Track them so the
|
|
# closing line attaches to the opening node instead of duplicating it.
|
|
self.by_number: dict[str, Group] = {}
|
|
self.seen_sections: set[str] = set()
|
|
|
|
@property
|
|
def cur(self) -> Group:
|
|
return self.stack[-1]
|
|
|
|
@staticmethod
|
|
def depth(number: str) -> int:
|
|
return number.count(".") + 1 if number else 1
|
|
|
|
def pop_to(self, depth: int) -> None:
|
|
"""Close open nodes until the top of stack is shallower than `depth`."""
|
|
while len(self.stack) > 1:
|
|
top = self.stack[-1]
|
|
if top.kind in ("root", "section", "schedule"):
|
|
break
|
|
if self.depth(top.number) < depth:
|
|
break
|
|
self.stack.pop()
|
|
|
|
def section_base(self) -> Group:
|
|
"""Where a new SECTION attaches: inside the current schedule if there
|
|
is one, otherwise at the root."""
|
|
for node in reversed(self.stack):
|
|
if node.kind == "schedule":
|
|
return node
|
|
return self.root
|
|
|
|
def reopen(self, number: str, depth: int) -> Group | None:
|
|
"""If `number` is already open, make it current again."""
|
|
node = self.by_number.get(number)
|
|
if node is None:
|
|
return None
|
|
while self.stack and self.stack[-1] is not node:
|
|
self.stack.pop()
|
|
return node
|
|
|
|
def add(self, node) -> None:
|
|
self.cur.children.append(node)
|
|
|
|
|
|
def _same_heading(a: str, b: str) -> bool:
|
|
"""Do these two labels refer to the same block?"""
|
|
na = re.sub(r"[^a-z0-9]", "", (a or "").lower())
|
|
nb = re.sub(r"[^a-z0-9]", "", (b or "").lower())
|
|
if not na or not nb:
|
|
return True
|
|
return na.startswith(nb[:24]) or nb.startswith(na[:24])
|
|
|
|
|
|
def _build(content: list[Row]) -> Group:
|
|
t = _Tree()
|
|
seen_totals: set[str] = set()
|
|
|
|
for r in content:
|
|
desc = r.desc
|
|
num = r.item.strip()
|
|
|
|
# ---- page-2+ running totals: "BLUEBERRY SQUARE 5,696,934.93" ----
|
|
if desc and r.amount_txt and not r.has_price and not num:
|
|
key = f"{desc[:40].upper()}|{r.amount_txt}"
|
|
if desc.upper() in {"TOTAL", "BLUEBERRY SQUARE"} or key in seen_totals:
|
|
if key in seen_totals:
|
|
continue
|
|
seen_totals.add(key)
|
|
|
|
# ---- section / schedule headings ---------------------------------
|
|
# Both start a fresh top-level branch: a section is never nested
|
|
# inside the previous section's last item.
|
|
sm = SECTION_RE.match(desc)
|
|
if sm and not r.has_price:
|
|
t.item_holder = None
|
|
# key on number + TITLE. The document genuinely has two sections
|
|
# numbered 6 (STORMWATER and Traffic Lights), so the number alone
|
|
# is not an identity.
|
|
key = f"sec:{sm.group(1)}:{re.sub(r'[^a-z0-9]', '', sm.group(2).lower())}"
|
|
if key in t.seen_sections:
|
|
continue
|
|
t.seen_sections.add(key)
|
|
sec = Group(kind="section", number=sm.group(1),
|
|
title=sm.group(2).strip(), page=r.page,
|
|
# Candy prints "SECTION 1: SITE CLEARANCE 160,057.99"
|
|
# on the heading row - keep that figure, it is what the
|
|
# consultant's own summary sheet expects to reconcile to
|
|
subtotal=_to_num(r.amount_txt) if r.amount_txt else None)
|
|
# reset FIRST: a section is a sibling of every other section, not
|
|
# a child of whatever the previous section happened to leave open
|
|
base = t.section_base()
|
|
t.stack = [t.root, base] if base is not t.root else [t.root]
|
|
t.add(sec)
|
|
t.stack.append(sec)
|
|
t.by_number.clear()
|
|
continue
|
|
|
|
if SCHEDULE_RE.match(desc) and not r.has_price and not num:
|
|
t.item_holder = None
|
|
if desc.upper() in t.seen_sections:
|
|
continue
|
|
t.seen_sections.add(desc.upper())
|
|
sch = Group(kind="schedule", number="", title=desc, page=r.page)
|
|
t.stack = [t.root]
|
|
t.add(sch)
|
|
t.stack.append(sch)
|
|
t.by_number.clear()
|
|
continue
|
|
|
|
# ---- priced leaf line -------------------------------------------
|
|
if r.has_price and (r.unit or r.qty is not None):
|
|
rate_only = bool(RATE_ONLY_RE.search(r.amount_txt))
|
|
amount = None if rate_only else _to_num(r.amount_txt)
|
|
|
|
sub, body = None, desc
|
|
lm = LETTER_RE.match(desc)
|
|
if lm and not num:
|
|
sub, body = lm.group(1), lm.group(2).strip()
|
|
|
|
# Candy often prices a NUMBERED item directly on its own heading
|
|
# row ("2.1.3 Cut to spoil m3 1,400 259.34 363,076.00").
|
|
# It carries its own price, so it belongs to the current group -
|
|
# not to whichever item happened to precede it. Only an unnumbered
|
|
# a)/b) sub-line should attach to the open item.
|
|
numbered_item = bool(num) and ITEM_RE.match(num) \
|
|
and t.depth(num) >= 3
|
|
|
|
line = Line(
|
|
number=num,
|
|
desc=f"{sub}) {body}" if sub else body,
|
|
unit=r.unit.strip(),
|
|
qty=r.qty,
|
|
rate=r.rate,
|
|
amount=amount,
|
|
rate_only=rate_only,
|
|
page=r.page,
|
|
)
|
|
if not line.unit:
|
|
line.flags.append("no_unit")
|
|
if line.rate is None and line.qty is not None and amount is None \
|
|
and not rate_only:
|
|
line.flags.append("amount_missing")
|
|
|
|
if sub and t.item_holder is not None:
|
|
t.item_holder.children.append(line)
|
|
else:
|
|
if numbered_item:
|
|
# close the previously open item so this one lands beside
|
|
# it in the enclosing group, not inside it
|
|
t.pop_to(t.depth(num))
|
|
t.item_holder = None
|
|
t.add(line)
|
|
if not numbered_item:
|
|
t.item_holder = None
|
|
continue
|
|
|
|
if PLACEHOLDER_RE.match(desc or "x"):
|
|
t.add(Line(number=num, desc=desc, page=r.page))
|
|
t.item_holder = None
|
|
continue
|
|
|
|
# ---- numbered heading, optionally carrying a subtotal ------------
|
|
if num and ITEM_RE.match(num):
|
|
depth = t.depth(num)
|
|
subtotal = _to_num(r.amount_txt) if r.amount_txt else None
|
|
|
|
existing = t.by_number.get(num)
|
|
if existing is not None:
|
|
same_block = _same_heading(existing.title, desc)
|
|
if same_block and subtotal is not None and existing.subtotal is None:
|
|
# closing subtotal line for a block already opened
|
|
existing.subtotal = subtotal
|
|
t.item_holder = None
|
|
continue
|
|
if same_block and subtotal is None:
|
|
# plain repeat (the closing summary page lists every
|
|
# subtotal again) - nothing new to record
|
|
t.item_holder = None
|
|
continue
|
|
# same number, DIFFERENT block. Real Candy output does this
|
|
# ("2.3 SUBBASE" and "2.3 BASE"). Open a fresh node and leave a
|
|
# marker so validation can raise it.
|
|
existing.flags.append(f"duplicate_number:{num}")
|
|
del t.by_number[num]
|
|
t.pop_to(depth)
|
|
node = Group(
|
|
kind="item" if depth >= 3 else "group",
|
|
number=num, title=desc, subtotal=subtotal, page=r.page,
|
|
)
|
|
t.add(node)
|
|
t.by_number[num] = node
|
|
t.stack.append(node)
|
|
# a depth>=3 item heading owns its a)/b)/c) sub-lines
|
|
t.item_holder = node if depth >= 3 else None
|
|
continue
|
|
|
|
# ---- continuation of a heading that wrapped ---------------------
|
|
if desc and t.item_holder is not None and not r.unit and not r.has_price:
|
|
t.item_holder.title = f"{t.item_holder.title} {desc}".strip()
|
|
continue
|
|
|
|
if desc:
|
|
t.add(Line(desc=desc, page=r.page))
|
|
t.item_holder = None
|
|
|
|
return t.root
|
|
|
|
|
|
def _drop_summary_page(rows: list[Row]) -> list[Row]:
|
|
"""Candy closes a BOQ with a page that just re-lists every subtotal.
|
|
|
|
That page contributes no new line items, only duplicates, so it is
|
|
identified by carrying several numbered subtotal rows and no priced line,
|
|
and dropped wholesale.
|
|
"""
|
|
by_page: dict[int, list[Row]] = {}
|
|
for r in rows:
|
|
by_page.setdefault(r.page, []).append(r)
|
|
|
|
for page, prows in by_page.items():
|
|
if len(prows) < 4:
|
|
continue
|
|
subtotals = [r for r in prows
|
|
if r.amount_txt and not r.has_price and r.item]
|
|
priced = [r for r in prows if r.has_price]
|
|
if len(subtotals) >= 4 and not priced:
|
|
rows = [r for r in rows if r.page != page]
|
|
return rows
|
|
|
|
|
|
def extract(pdf_path: str) -> dict:
|
|
rows: list[Row] = []
|
|
with pdfplumber.open(pdf_path) as pdf:
|
|
for pno, page in enumerate(pdf.pages, start=1):
|
|
rows.extend(_page_rows(page, pno))
|
|
|
|
rows = _drop_summary_page(rows)
|
|
clean = [r for r in rows if not r.is_empty and not _is_noise(r)]
|
|
merged = _merge_continuations(clean)
|
|
return asdict(_build(merged))
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover
|
|
import json
|
|
import sys
|
|
|
|
print(json.dumps(extract(sys.argv[1]), indent=2, default=str))
|