MEGA-BOQ-WEBAPP/app.py

#!/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:
    """Fingerprint of the source in this image, shown in the sidebar.

    Computed at image build time over the .py/.css/.toml actually copied in, so
    it identifies the running code by construction. Falls back through several
    paths because Streamlit execs this file where __file__ is not guaranteed -
    a bare reference to it would take the whole app down - 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 "unknown"


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.")