MEGA-BOQ-WEBAPP/app.py
MEGA INFRA 7fd221288a Make the build marker actually report the build
It read 'dev' on the live site, so it was decoration rather than evidence.
Cause: Coolify does not pass a commit SHA into a Docker Compose build, so my
${BUILD_SHA:-dev} fallback always won.

Read from build.txt in the repo instead, which cannot drift from what is
deployed. Two things that took a try each:
- Docker's ENV does not evaluate $(cat file), so the file is copied into the
  image and read at startup.
- Streamlit executes app.py in an exec context where __file__ is not
  guaranteed; a bare reference to it inside build_sha() would have taken the
  whole app down. It now walks a list of candidate paths and never raises.

Verified: docker build ok; build.txt inside the image reads e1f4127;
build_sha() resolves to e1f4127 in the container; the sidebar renders it.
2026-10-02 21:57:13 +02:00

629 lines
25 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/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 build_sha() -> str:
"""Which build is this? Shown in the sidebar so a deploy can be confirmed
without digging through Coolify's log.
Falls back through several routes because Streamlit executes this file in
an exec context where __file__ is not guaranteed - a bare reference to it
would take the whole app down. Falls back to the env var, then to a
literal, and never raises.
"""
candidates = []
try:
candidates.append(os.path.join(
os.path.dirname(os.path.abspath(__file__)), "build.txt"))
except NameError:
pass
candidates.append(os.path.join(os.getcwd(), "build.txt"))
candidates.append("/app/build.txt")
for path in candidates:
try:
with open(path, encoding="utf-8") as fh:
val = fh.read().strip()
if val and val != "dev":
return val
except OSError:
continue
return os.environ.get("BOQ_BUILD_SHA") or "dev"
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(
'<div class="boq-brand">'
f'<span class="boq-brand-mark">{theme.icon("building", 30)}</span>'
'<span class="boq-brand-text">'
'<span class="boq-brand-name">MEGA INFRA</span>'
'<span class="boq-brand-sub">BOQ Converter &nbsp;·&nbsp; '
'Candy PDF (ours) &nbsp;→&nbsp; consultant\'s Excel (theirs) '
'&nbsp;·&nbsp; nothing leaves this machine</span>'
'</span></div>',
unsafe_allow_html=True)
known = profiles.list_profiles()
with st.sidebar:
st.caption(f"build {build_sha()}")
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'<div class="boq-field-label">{theme.icon("file", 15)}'
f'<span><b>1</b> &nbsp;Candy PDF <em>(ours)</em></span></div>',
unsafe_allow_html=True)
pdf_file = st.file_uploader(" ", type=["pdf"], label_visibility="collapsed")
with c2:
st.markdown(
f'<div class="boq-field-label">{theme.icon("table", 15)}'
f'<span><b>2</b> &nbsp;Consultant Excel <em>(theirs)</em></span></div>',
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 <b>both</b> 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'<div class="{cls}">'
f'<div class="boq-kpi-head">'
f'<span class="boq-kpi-icon">{theme.icon(icon_name, 16)}</span>'
f'<span class="boq-kpi-label">{_html.escape(label)}</span>'
f"</div>"
f'<div class="boq-kpi-value">{value}</div>'
f"</div>"
)
k1, k2, k3, k4 = st.columns(4)
_open_n = len([f for f in result.findings
if f.kind == "conflict" and f.severity != "resolved"])
_unmapped_n = len([f for f in result.findings if f.kind == "unmapped"])
k1.markdown(kpi("check", "Values written", result.written), unsafe_allow_html=True)
k2.markdown(kpi("shield", "Already correct", result.kept), unsafe_allow_html=True)
k3.markdown(kpi("alert", "Open disagreements", _open_n,
"warn" if _open_n else "ok"), unsafe_allow_html=True)
k4.markdown(kpi("info", "Unmapped source lines", _unmapped_n,
"warn" if _unmapped_n else "ok"), unsafe_allow_html=True)
applied = st.session_state.get("applied", 0)
_tail = (f"with <b>{applied}</b> of your decisions written in. "
if applied else "")
if verdict.blocked:
st.error(
"**This workbook is not safe to send.** "
f"{len(verdict.failures)} check(s) failed, so the download has been "
f"removed. Nothing has been lost — your original file is untouched."
)
for c in verdict.failures:
st.write(f"• **{c.name}** — {c.detail}")
st.caption("Send these to us; this is a bug in the converter, not in "
"your file.")
else:
st.markdown(theme.pill(
f"<b>Verified safe to send.</b> {_tail}Your file was populated using "
f"its own formulas and layout, and every write was read back and "
f"checked. Download it and send it with the Candy PDF.",
"shield", "ok"), unsafe_allow_html=True)
g1, g2 = st.columns([1, 1])
with g1:
with open(out_xlsx, "rb") as fh:
st.download_button("Download populated workbook",
fh.read(),
file_name=os.path.basename(out_xlsx),
mime="application/vnd.openxmlformats-"
"officedocument.spreadsheetml.sheet")
with g2:
st.markdown(theme.pill(
f"Source PDF totals <b>R{verdict.source_total:,.2f}</b> before "
f"VAT. The populated workbook keeps every one of the "
f"{_n_formula(out_xlsx)} formulas you had.",
"check", "info"), unsafe_allow_html=True)
# ---------------------------------------------------------------- proof
st.markdown("---")
st.markdown(
f'<div class="boq-section-title">{theme.icon("clipboard", 20)}'
f'<span>Proof &mdash; what happened, line by line</span></div>',
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'<span><span class="boq-swatch" style="background:{col}"></span>'
f"{label}</span>")
return f'<div class="boq-legend">{"".join(parts)}</div>'
# 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"<b>{len(settled)} decision(s) applied</b> and written into the "
f"workbook.", "check", "ok"), unsafe_allow_html=True)
if open_f:
st.markdown(theme.pill(
"<b>Nothing was overwritten.</b> 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"<b>{n_take + n_custom} change(s) staged.</b> 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"<b>{len(_unmapped)} line(s) had no obvious row.</b> 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"<b>{n} line(s) placed by hand.</b> 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'<div class="boq-section-sub">{theme.icon("layers", 15)}'
f'<span>Totals carried across, in order</span></div>',
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: <b>{money(sum(v for _, v in vals))}</b>",
"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'<div class="boq-section-sub">'
f'{theme.icon("clipboard", 15)}'
f'<span>What your workbook was saved with</span></div>',
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.")