#!/usr/bin/env python3
"""
BOQ converter - local web app.
Drop in the Candy PDF (ours) and the consultant's Excel (theirs); get back the
consultant's workbook populated, plus a proof report of exactly what was
written, what was already correct, and what the two documents disagree about.
streamlit run app.py --server.port 8501
Runs entirely on this machine. Nothing is uploaded anywhere.
"""
from __future__ import annotations
import os
import sys
import html as _html
import re
import tempfile
import streamlit as st
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from openpyxl import load_workbook # noqa: E402
from openpyxl.utils import get_column_letter # noqa: E402
from boq_engine import profiles, theme # noqa: E402
from boq_engine.extract import extract # noqa: E402
from boq_engine.template import ( # noqa: E402
detect_layout, hints_from_profile, populate, propose_placements,
source_lines, summary_values,
)
from boq_engine.validate import validate # noqa: E402
from boq_engine.verify import verify # noqa: E402
def _fmt(v) -> str:
if isinstance(v, float):
return f"{v:,.6g}"
return str(v)
def _strip_md(t: str) -> str:
"""Cell text can carry markdown bold from a wrapped description."""
return re.sub(r"\*\*(.+?)\*\*", r"\1", t or "").strip()
def _cl(lay, name: str) -> str:
c = lay.col(name)
return get_column_letter(c) if c else "—"
def _desc_run(lay) -> str:
if not lay.desc_cols:
return "—"
return "+".join(get_column_letter(c) for c in lay.desc_cols)
def _all_layouts(wb):
out = {}
for name in wb.sheetnames:
try:
out[name] = detect_layout(wb[name])
except ValueError:
continue
return out
def _n_formula(path: str) -> int:
try:
wb = load_workbook(path)
return sum(1 for nm in wb.sheetnames for row in wb[nm].iter_rows()
for c in row if isinstance(c.value, str)
and c.value.startswith("="))
except Exception:
return 0
def _coerce(text: str):
"""A typed value should land as a number when it looks like one."""
t = text.strip().replace(",", "")
if not t:
return text
try:
f = float(t)
return int(f) if f.is_integer() else f
except ValueError:
return text
st.set_page_config(page_title="MEGA INFRA · BOQ Converter",
page_icon="M", layout="wide")
theme.inject()
@st.cache_data(show_spinner="Reading the Candy PDF…")
def read_pdf(data: bytes):
tmp = tempfile.NamedTemporaryFile(suffix=".pdf", delete=False)
tmp.write(data)
tmp.close()
try:
return extract(tmp.name)
finally:
os.unlink(tmp.name)
def _fp(data: bytes) -> str:
import hashlib
return hashlib.sha256(data).hexdigest()[:16]
def money(v) -> str:
if isinstance(v, (int, float)):
return f"R{v:,.2f}"
return str(v) if v not in (None, "") else "—"
# ---------------------------------------------------------------- header
st.markdown(
''
f'{theme.icon("building", 30)}'
''
'MEGA INFRA'
'BOQ Converter · '
'Candy PDF (ours) → consultant\'s Excel (theirs) '
' · nothing leaves this machine'
'
',
unsafe_allow_html=True)
known = profiles.list_profiles()
with st.sidebar:
st.caption(f"build {os.environ.get('BOQ_BUILD_SHA', 'dev')}")
st.subheader("Consultant profile")
if known:
names = ["— new / not set —"] + [p.consultant for p in known]
pick = st.selectbox("Known consultants", names, index=0)
else:
pick = "— new / not set —"
st.caption("No profiles saved yet.")
consultant = st.text_input(
"Consultant name",
value="" if pick.startswith("—") else pick,
placeholder="e.g. Civil Matters",
help="Saved so next time their layout is recognised automatically.",
)
if consultant and not pick.startswith("—"):
prof = profiles.get(pick)
if prof:
st.caption(f"Saved {prof.saved} · used {prof.uses}× · "
f"{len(prof.sheets)} sheet(s)")
st.checkbox("Re-detect from scratch", key="redetect",
help="Ignore the saved column map for this run and "
"work it out again from the file in front of you.")
with st.expander("What we learned about their layout"):
st.code(prof.summary(), language="text")
# ---------------------------------------------------------------- inputs
c1, c2 = st.columns(2)
with c1:
st.markdown(
f'{theme.icon("file", 15)}'
f'1 Candy PDF (ours)
',
unsafe_allow_html=True)
pdf_file = st.file_uploader(" ", type=["pdf"], label_visibility="collapsed")
with c2:
st.markdown(
f'{theme.icon("table", 15)}'
f'2 Consultant Excel (theirs)
',
unsafe_allow_html=True)
xls_file = st.file_uploader(" ", type=["xlsx", "xlsm"],
label_visibility="collapsed")
st.markdown("---")
if not (pdf_file and xls_file):
st.markdown(theme.pill(
"Drop in both files to begin — the Candy PDF and the "
"consultant's Excel.", "info", "info"), unsafe_allow_html=True)
st.stop()
# ---------------------------------------------------------------- run
out_dir = tempfile.mkdtemp(prefix="boq_out_")
base = os.path.splitext(os.path.splitext(pdf_file.name)[0])[0]
out_xlsx = os.path.join(out_dir, f"{base}_POPULATED.xlsx")
with st.spinner("Populating…"):
tree = read_pdf(pdf_file.getvalue())
src_rep = validate(tree)
xls_tmp = os.path.join(out_dir, "in.xlsx")
with open(xls_tmp, "wb") as fh:
fh.write(xls_file.getvalue())
# Re-read the staged choices from the previous run's widgets. This happens
# BEFORE populate() so a regenerate writes the consultant's decisions
# straight into the workbook - no second manual step.
# The dispute catalogue is built on the first pass and then kept. Once a
# cell is settled it drops out of the findings list, so re-deriving it each
# run would lose the very rows the consultant already decided about.
# Reuse this consultant's saved column map when there is one. It is
# verified against the file in use, never trusted blindly - a stale profile
# silently applied would be worse than no profile at all.
_prof = profiles.get(consultant) if consultant else None
hints = ({} if st.session_state.get("redetect") else hints_from_profile(_prof))
if _prof and hints:
st.session_state["layout_notes_lede"] = (
f"using {_prof.consultant}'s saved column map "
f"(saved {getattr(_prof, 'saved', '?')})")
catalogue = st.session_state.get("disputes") or {}
if not catalogue:
probe = populate(xls_tmp, tree, os.path.join(out_dir, "probe.xlsx"))
catalogue = {
(f.sheet, f.row, f.field): {
"label": f.label, "detail": f.detail,
"source_value": f.source_value,
"template_value": f.template_value,
"severity": f.severity,
}
for f in probe.findings if f.kind == "conflict"
}
st.session_state["disputes"] = catalogue
overrides: dict = {}
if st.session_state.get("dirty"):
for key, val in list(st.session_state.items()):
if not key.startswith("choice::") or key.endswith("::custom"):
continue
parts = key.split("::")
if len(parts) != 4:
continue
cell = (parts[1], int(parts[2]), parts[3])
if cell not in catalogue:
continue
custom = st.session_state.get(f"{key}::custom", "").strip()
if custom:
overrides[cell] = _coerce(custom)
elif str(val).startswith("Use our"):
overrides[cell] = catalogue[cell]["source_value"]
st.session_state["dirty"] = False
st.session_state["applied"] = len(overrides)
placements = {}
if st.session_state.get("dirty"):
for k, v in st.session_state.items():
if k.startswith("place::") and v and not v.endswith("|"):
try:
idx, sheet, row = v.split("|")
placements[int(idx)] = (sheet, int(row))
except ValueError:
pass
st.session_state["applied_places"] = len(placements)
st.session_state["dirty"] = False
result = populate(xls_tmp, tree, out_xlsx, overrides=overrides,
placements=placements, hints=hints)
# Prove the file before offering it. Reads the OUTPUT back off disk and
# compares it to the source - it does not trust what populate() believes
# it wrote. A failed check removes the download entirely rather than
# showing a warning somebody can scroll past.
with st.spinner("Proving the output…"):
verdict = verify(xls_tmp, out_xlsx, tree, result)
# remember this consultant's layout for next time
if consultant:
wb = load_workbook(xls_tmp)
lays = {}
for nm in wb.sheetnames:
try:
lays[nm] = detect_layout(wb[nm], hints.get(nm))
except ValueError:
pass
if lays:
profiles.save(consultant, lays, _fp(xls_file.getvalue()))
profiles.touch(consultant)
# ---------------------------------------------------------------- results
def kpi(icon_name: str, label: str, value, tone: str = "orange") -> str:
"""A KPI tile. Hand-built so it can carry a real icon - st.metric's label
is plain text, and a row of four plain numbers is the generic-looking part
of this page."""
cls = f"boq-kpi boq-kpi-{tone}"
return (
f''
f'
'
f'{theme.icon(icon_name, 16)}'
f'{_html.escape(label)}'
f"
"
f'
{value}
'
f"
{theme.icon("clipboard", 20)}'
f'Proof — what happened, line by line
',
unsafe_allow_html=True)
conflicts = [f for f in result.findings if f.kind == "conflict"]
unmapped = [f for f in result.findings if f.kind == "unmapped"]
totals = [f for f in result.findings if f.kind == "total"]
notes = [f for f in result.findings if f.kind == "info"]
SEVERITY_SWATCH = {
"decision": ("#8C363B", "needs a decision"),
"source_gap": ("#DF8441", "our PDF has no figure - theirs kept"),
"source_defect": ("#FEA944", "our PDF printed something corrupt"),
"resolved": ("#00CD8A", "settled by the consultant"),
"note": ("#00CD8A", "no action needed"),
}
def _legend() -> str:
seen, parts = set(), []
for sev, (col, label) in SEVERITY_SWATCH.items():
if sev == "note" or label in seen:
continue
seen.add(label)
parts.append(
f''
f"{label}")
return f'{"".join(parts)}
'
# NOTE: st.tabs / st.button / st.download_button render their labels as PLAIN
# TEXT. Inline SVG there leaks to the user as literal markup, and emoji read as
# emoticons. So those stay typographic; the real icons live where HTML is
# actually rendered - the brand mark, KPI tiles, pills and section headings.
_unmapped = [f for f in result.findings if f.kind == "unmapped"]
tab_conf, tab_place, tab_tot, tab_src, tab_ver = st.tabs([
f"Disagreements ({len(conflicts)})",
f"Unplaced ({len(_unmapped)})",
f"Totals written ({len(totals)})",
f"Source BOQ checks ({len(src_rep.errors)})",
f"Verification ({len(verdict.checks)})",
])
with tab_conf:
if not conflicts and not unmapped:
st.markdown(theme.pill("The two documents agree everywhere.",
"check", "ok"), unsafe_allow_html=True)
if unmapped:
st.error(f"**{len(unmapped)} source line(s) had no row in the "
f"template.** These are NOT in the consultant's file:")
st.markdown("".join(
theme.card(f.label.strip(),
f"{f.detail} (source: {f.source})",
meta="no row in template", severity="decision")
for f in unmapped), unsafe_allow_html=True)
settled = [f for f in conflicts if f.severity == "resolved"]
open_f = [f for f in conflicts if f.severity != "resolved"]
if settled:
st.markdown(theme.pill(
f"{len(settled)} decision(s) applied and written into the "
f"workbook.", "check", "ok"), unsafe_allow_html=True)
if open_f:
st.markdown(theme.pill(
"Nothing was overwritten. Where the two documents differ, "
"your own value was left alone. Open anything you want changed, "
"then regenerate.",
"shield", "ok"), unsafe_allow_html=True)
st.markdown(_legend(), unsafe_allow_html=True)
# Real Streamlit widgets, so the choice actually reaches Python. Keys
# are derived from the cell, so a choice survives a rerun and can be
# read back to build the override map.
# one collapsed row per dispute: 16 open boxes is a wall of scrolling,
# and the point is to scan for the one that needs a human
for f in open_f:
key = f"choice::{f.sheet}::{f.row}::{f.field}"
chosen = st.session_state.get(key, "Keep yours (as it is)")
staged = st.session_state.get(f"{key}::custom", "").strip()
flag = "✎ " if (staged or chosen.startswith("Use our")) else ""
with st.expander(
f"{flag}{f.label.strip()[:70]} · {f.sheet} r{f.row} · "
f"{f.field or 'value'}",
expanded=bool(flag)):
st.caption(_strip_md(f.detail))
c_a, c_b = st.columns([1, 1])
with c_a:
st.selectbox(
"which value stands",
options=["Keep yours (as it is)",
f"Use our PDF's value ({_fmt(f.source_value)})"],
key=key, label_visibility="collapsed")
with c_b:
st.text_input(
"or type a different value",
key=f"{key}::custom", label_visibility="collapsed",
placeholder="or type a value")
st.markdown("---")
c1, c2 = st.columns([3, 1])
with c1:
n_custom = len([
1 for f in open_f
if st.session_state.get(f"choice::{f.sheet}::{f.row}::{f.field}::custom", "").strip()
])
n_take = len([
1 for f in open_f
if st.session_state.get(f"choice::{f.sheet}::{f.row}::{f.field}",
"Keep yours (as it is)").startswith("Use our")
])
if n_custom or n_take:
st.markdown(theme.pill(
f"{n_take + n_custom} change(s) staged. Press "
f"regenerate to write them into the workbook.",
"refresh", "warn"), unsafe_allow_html=True)
else:
st.caption("Your PAYMENT column, formulas and formatting are "
"never touched.")
with c2:
if st.button("Apply & regenerate", type="primary",
use_container_width=True):
st.session_state["dirty"] = True
st.rerun()
# ---------------------------------------------------------------- place tab
with tab_place:
if not _unmapped:
st.markdown(theme.pill(
"Every line in the Candy PDF found a row in your workbook. "
"Nothing to decide.", "check", "ok"), unsafe_allow_html=True)
else:
st.markdown(theme.pill(
f"{len(_unmapped)} line(s) had no obvious row. These are the "
f"ones worth a human's eye — the tool ranked the best few "
f"candidates, but where a description repeats in the document only "
f"you can say which is which. Leave one on \u201cleave unplaced\u201d "
f"to keep it out of the workbook.",
"alert", "warn"), unsafe_allow_html=True)
_lines_all = source_lines(tree)
_props = propose_placements(tree, xls_tmp, _lines_all, top_n=5)
for f in _unmapped:
# populate() records the source line index, so the row here is the
# exact line that had no home - no fuzzy label matching needed
i = f.index
p = _props[i] if 0 <= i < len(_props) else None
if not p or not p["candidates"]:
continue
opts = ["— leave unplaced —"] + [
f"{c['sheet']} · row {c['row']} · "
f"{(c['number'] or '—')} {(c['desc'] or '')[:44]} "
f"[{c['score']:.0%}]"
for c in p["candidates"]]
pick = st.selectbox(
f"{f.label.strip()[:68]} — source: {f.source}",
opts, key=f"place::{i}", label_visibility="collapsed")
if pick != opts[0]:
sel = p["candidates"][opts.index(pick) - 1]
st.session_state[f"_place__{i}"] = (sel["sheet"], sel["row"])
c1, c2 = st.columns([3, 1])
with c1:
n = len([k for k in st.session_state if k.startswith("_place__")])
if n:
st.markdown(theme.pill(
f"{n} line(s) placed by hand. Press regenerate to "
f"write them in and rebuild the workbook.",
"check", "ok"), unsafe_allow_html=True)
else:
st.caption("Nothing chosen yet.")
with c2:
if st.button("Apply placements & regenerate",
type="primary", use_container_width=True):
st.session_state["dirty"] = True
st.rerun()
with tab_tot:
st.caption("The cells your consultant's summary sheet reads were blank in "
"the template. Each now carries its section's total.")
# Streamlit's dataframe has no text-wrapping option (no wrap_text in
# column_config), and this column is prose, so it gets cards like the
# findings above rather than a table that truncates.
st.markdown("".join(
theme.card(f.label.strip(), f.detail, meta="written to",
severity="note")
for f in totals), unsafe_allow_html=True)
vals = summary_values(tree)
st.markdown(
f'{theme.icon("layers", 15)}'
f'Totals carried across, in order
',
unsafe_allow_html=True)
st.dataframe([{"#": i + 1, "Schedule / Section": l, "Amount": money(v)}
for i, (l, v) in enumerate(vals)],
use_container_width=True, hide_index=True)
if vals:
st.markdown(theme.pill(
f"Sum of all sections: {money(sum(v for _, v in vals))}",
"calculator", "info"), unsafe_allow_html=True)
with tab_src:
if not src_rep.errors:
st.markdown(theme.pill("The source BOQ adds up on its own.",
"check", "ok"), unsafe_allow_html=True)
for i in src_rep.errors:
st.error(f"**{i.where}**\n\n{i.detail}")
if notes:
with st.expander(f"Notes ({len(notes)})"):
for f in notes:
st.write(f"· `{f.sheet}` r{f.row} — {f.label}: {f.detail}")
if result.layout_notes:
with st.expander("Layout detection notes"):
for n in result.layout_notes:
st.write(f"· {n}")
# ---------------------------------------------------------------- verify tab
with tab_ver:
st.caption("Each check reads the populated file back off disk and compares "
"it to the source. It does not trust what the writer believes it "
"did.")
tone = {"decision": "", "source_gap": "info", "source_defect": "warn",
"resolved": "ok", "note": "ok"}
st.markdown("".join(
theme.card(
c.name.replace("_", " ").capitalize(),
c.detail,
meta="passed" if c.passed else ("warning" if not c.fatal
else "FAILED"),
severity="note" if c.passed else
("source_defect" if not c.fatal else "decision"),
)
for c in verdict.checks), unsafe_allow_html=True)
if verdict.warnings:
st.markdown(theme.pill(
f"{len(verdict.warnings)} check(s) raised a warning. Warnings do "
f"not block the download, but they are worth reading before you "
f"send this on.", "alert", "warn"), unsafe_allow_html=True)
st.markdown(f''
f'{theme.icon("clipboard", 15)}'
f'What your workbook was saved with
',
unsafe_allow_html=True)
st.dataframe(
[{"Sheet": name,
"Header row": lay.header_row,
"Item": _cl(lay, "item"), "Description": _desc_run(lay),
"Unit": _cl(lay, "unit"), "Qty": _cl(lay, "qty"),
"Rate": _cl(lay, "rate"), "Amount": _cl(lay, "amount"),
"Rows": len(lay.data_rows)}
for name, lay in _all_layouts(load_workbook(xls_tmp)).items()],
use_container_width=True, hide_index=True)
st.caption("This is exactly what the app decided about your file. If a "
"column looks wrong, tell us and we will correct the profile — "
"it is saved under your name, so the fix applies next time.")