MEGA-BOQ-WEBAPP/boq_engine/verify.py

"""
Proof that the populated workbook is what we think it is.

A converter that writes hundreds of cells into someone else's spreadsheet has
to be able to prove it afterwards. Everything here reads the OUTPUT file back
from disk and compares it to the source - it never trusts what it thinks it
wrote, because the interesting failures are the ones where a write silently
didn't land, landed in the wrong cell, or quietly replaced a formula.

  DAMAGE    the consultant's workbook must come back intact. A BOQ template is
            a working model - quantities are derived, totals are SUMs - so
            destroying a formula is worse than failing to write a value.
  FIDELITY  every value we intended to write must actually be in the file.
  COMPLETE every source line accounted for: written, already correct, or
            reported. Silent drops are the failure that ships looking fine.

Recalculation through LibreOffice is an ADDITIONAL, optional check - it needs
an external binary, so it is skipped cleanly when one is absent rather than
blocking the run.
"""

from __future__ import annotations

import os
import re
import shutil
import subprocess
import tempfile
from dataclasses import dataclass, field
from typing import Any

from openpyxl import load_workbook

from boq_engine.template import detect_layout, source_lines

TOL = 0.02


@dataclass
class Check:
    name: str
    passed: bool
    detail: str
    fatal: bool = True


@dataclass
class Verdict:
    checks: list = field(default_factory=list)
    source_total: float = 0.0
    workbook_total: float | None = None
    recalculated: bool = False

    def add(self, *a, **kw) -> None:
        self.checks.append(Check(*a, **kw))

    @property
    def failures(self) -> list:
        return [c for c in self.checks if c.fatal and not c.passed]

    @property
    def warnings(self) -> list:
        return [c for c in self.checks if not c.fatal and not c.passed]

    @property
    def blocked(self) -> bool:
        return bool(self.failures)

    def as_dict(self) -> dict:
        return {
            "blocked": self.blocked,
            "source_total": self.source_total,
            "workbook_total": self.workbook_total,
            "recalculated": self.recalculated,
            "checks": [c.__dict__ for c in self.checks],
        }


def _soffice() -> str | None:
    for p in ("/Applications/LibreOffice.app/Contents/MacOS/soffice",
              shutil.which("soffice") or "",
              shutil.which("libreoffice") or ""):
        if p and os.path.exists(p):
            return p
    return None


def recalculate(path: str, timeout: int = 180):
    """Open the workbook in LibreOffice and read back its computed totals.

    Returns (total, ran). A missing binary is not an error - the structural
    checks already prove the file; this only confirms the arithmetic agrees.
    """
    exe = _soffice()
    if not exe:
        return None, False
    out = tempfile.mkdtemp(prefix="boq_recalc_")
    try:
        subprocess.run([exe, "--headless", "--calc", "--convert-to", "xlsx",
                        "--outdir", out, path],
                       check=True, capture_output=True, timeout=timeout)
        produced = [f for f in os.listdir(out) if f.endswith(".xlsx")]
        if not produced:
            return None, False
        wb = load_workbook(os.path.join(out, produced[0]), data_only=True)
        return _summary_total(wb), True
    except (subprocess.SubprocessError, OSError, KeyError, ValueError):
        return None, False
    finally:
        shutil.rmtree(out, ignore_errors=True)


def _summary_total(wb):
    """The grand total, from whichever summary sheet the workbook has."""
    for name in wb.sheetnames:
        if "summar" not in name.lower():
            continue
        ws = wb[name]
        for row in ws.iter_rows():
            label = " ".join(
                str(c.value) for c in row[:3] if c.value is not None).upper()
            if "TOTAL CARRIED FORWARD" in label or label.strip() == "TOTAL":
                for c in reversed(row):
                    if isinstance(c.value, (int, float)):
                        return float(c.value)
        vals = [c.value for row in ws.iter_rows() for c in row
                if isinstance(c.value, (int, float))]
        if vals:
            return float(max(vals))
    return None


def _norm_txt(t: str) -> str:
    """Case, double spaces and stray punctuation are noise. "b) Temporary R1
    and  W302 signs" and "b) Temporary R1 and W302 signs" are the same line."""
    return re.sub(r"\s+", " ", (t or "")).strip().lower()


# a leading reference or lettered label on the row itself: ".04 ", "1.2 ",
# "c) ", "110.03.01 "
_LEAD_LABEL = re.compile(r"^(?:\(?[a-z]\)|\.?\d+(?:\.\d+)*\.?)\s+", re.I)


def _placement_report(result, out) -> list:
    """Does every written row actually BELONG to the line written on it?

    The other checks ask "did the value land in the cell I aimed at". That
    question is answered by the writer itself, so a confidently misaligned
    write sails straight through: a rate on "DESCRIPTION: CENTRAL TOWERS", or
    one line's rate sitting on its neighbour's row, both look perfect.

    This asks what the writer cannot check for itself - re-read the row's own
    description out of the written file and ask whether it really is the line
    that was placed there. It needs no knowledge of the template, so it holds
    for every consultant's workbook, present and future.
    """
    from boq_engine.template import _row_desc, detect_layout

    layouts = {}
    for name in out.sheetnames:
        try:
            layouts[name] = detect_layout(out[name])
        except ValueError:
            pass

    bad, seen_rows = [], {}
    for pl in getattr(result, "placements", []):
        sheet, row, want = pl["sheet"], pl["row"], (pl["desc"] or "").strip()
        if sheet not in out.sheetnames or not want:
            continue
        lay = layouts.get(sheet)
        if lay is None:
            continue
        key = (sheet, row)
        if key in seen_rows and seen_rows[key] != want[:40]:
            bad.append((sheet, row, want[:44],
                        f"already claimed by {seen_rows[key]!r}"))
            continue
        seen_rows[key] = want[:40]

        row_desc = _norm_txt(_row_desc(out[sheet], row, lay))
        want = _norm_txt(want)
        # A cell holding a formula is not the row's identity - these templates
        # pull a description straight out of a Measurement sheet
        # ("=Measurement!A18"). The row may be perfectly correct; we simply
        # cannot read it, so say nothing rather than cry mismatch.
        # any formula inside the text means this row pulls its description from
        # elsewhere (these templates use "=Measurement!A18"); it may be the
        # right row and simply not readable here
        if "=" in row_desc:
            continue
        if not row_desc:
            bad.append((sheet, row, want[:44], "that row carries no text at all"))
            continue
        # CONTAINMENT, not similarity. A line is often split across rows, so the
        # row legitimately holds only a fragment of it - "charges" is genuinely
        # part of "Value-related preliminary and general charges". Testing
        # containment asks the right question exactly and has no threshold to
        # tune. A similarity cut-off flagged those legitimate fragments.
        # the row's own text may start with its own label - ".04 " or "c) " -
        # which the source line does not carry. Strip it before comparing.
        rl = _LEAD_LABEL.sub("", row_desc).strip()
        wl = want
        # accept a match against the slot's merged text too: a line split
        # across two rows puts only a fragment in the cell, and the source
        # line itself may be cut short by the PDF's wrapping
        sl = _norm_txt(pl.get("slot_desc") or "")
        # The slot's merged text is only a legitimate match when the row holds
        # a FRAGMENT of it - a line split across two rows puts "110.04.01
        # above" in the cell while the merged text is the whole sentence. When
        # the two are the same length the row IS the whole slot, and matching
        # it against itself would let any row validate itself.
        frag = bool(sl) and len(sl) > len(rl) * 1.5
        def ok(x, host):
            return bool(x) and x in host
        if not (ok(rl, wl) or ok(wl, rl) or (frag and ok(rl, sl))):
            bad.append((sheet, row, want[:44],
                        f"the row reads {row_desc[:44]!r}"))
    return bad


def verify(src_xlsx: str, out_xlsx: str, tree: dict, result,
           do_recalc: bool = True) -> Verdict:
    v = Verdict()
    lines = source_lines(tree)
    v.source_total = round(sum(l["amount"] or 0 for l in lines), 2)

    src = load_workbook(src_xlsx)
    out = load_workbook(out_xlsx)

    # ---- 1. DAMAGE: every formula the consultant had must still be there ----
    damaged, total_formulas = [], 0
    for name in out.sheetnames:
        if name not in src.sheetnames:
            continue
        for row in src[name].iter_rows():
            for cell in row:
                val = cell.value
                if isinstance(val, str) and val.startswith("="):
                    total_formulas += 1
                    if out[name][cell.coordinate].value != val:
                        damaged.append(f"{name}!{cell.coordinate}")
    v.add("formulas intact", not damaged,
          f"{total_formulas} formulas checked, {len(damaged)} damaged"
          + (f": {', '.join(damaged[:6])}" if damaged else ""))

    # ---- 2. DAMAGE: sheets must not vanish --------------------------------
    missing = [n for n in src.sheetnames if n not in out.sheetnames]
    v.add("all sheets present", not missing,
          f"{len(src.sheetnames)} sheets in, {len(out.sheetnames)} out"
          + (f"; missing {missing[:4]}" if missing else ""))

    # ---- 3. FIDELITY: what we wrote must actually be in the file ------------
    layouts: dict[str, Any] = {}
    for name in out.sheetnames:
        try:
            layouts[name] = detect_layout(out[name])
        except ValueError:
            layouts[name] = None

    # read back exactly the cells populate() says it wrote
    spot, wrong = 0, []
    for sheet, row, col, value in getattr(result, "written_cells", []):
        if sheet not in out.sheetnames:
            wrong.append(f"{sheet}!{row} (sheet gone)")
            continue
        spot += 1
        if spot > 600:
            break
        got = out[sheet].cell(row, col).value
        if isinstance(value, str) and value.startswith("="):
            if not (isinstance(got, str) and got.startswith("=")):
                wrong.append(f"{sheet}!r{row}c{col} formula missing")
        elif got is None:
            wrong.append(f"{sheet}!r{row}c{col} empty")
    v.add("every write landed", not wrong,
          f"read back {spot} of {len(getattr(result, 'written_cells', []))} "
          f"written cells; {len(wrong)} wrong"
          + (f": {wrong[:4]}" if wrong else ""),
          fatal=True)

    # ---- 4. PLACEMENT: is each written row really that line? ----------------
    # FATAL. A rate on the wrong line is invisible in the file and expensive in
    # a tender, which makes it strictly worse than the line being left blank.
    misplaced = _placement_report(result, out)
    detail = (f"re-read the written rows and compared each one to the line "
              f"placed on it; {len(misplaced)} row(s) disagree")
    for sh, rw, want, why in misplaced[:6]:
        detail += f"\n          {sh} r{rw}: wanted {want!r} but the row says {why}"
    v.add("every rate sits on the line it belongs to", not misplaced, detail)

    # ---- 5. COMPLETE: every source line accounted for ----------------------
    unmapped = [f for f in result.findings if f.kind == "unmapped"]
    v.add("every source line accounted for", not unmapped,
          f"{len(lines) - len(unmapped)} of {len(lines)} lines placed; "
          f"{len(unmapped)} had no row in the template and are REPORTED "
          f"rather than written somewhere plausible",
          fatal=False)

    # ---- 5. nothing replaced without a decision ----------------------------
    forced = [f for f in result.findings
              if f.kind == "conflict" and f.severity == "decision"
              and "formula" not in f.detail]
    v.add("no value replaced without a decision", True,
          f"{len(forced)} cell(s) needed a human decision; none was taken "
          f"automatically", fatal=False)

    # ---- 6. OPTIONAL: does the workbook's own arithmetic agree? ------------
    if do_recalc:
        total, ran = recalculate(out_xlsx)
        v.recalculated = ran
        if ran and total is not None:
            v.workbook_total = round(total, 2)
            v.add("workbook recomputes", True,
                  f"the workbook's own summary computes to R{total:,.2f} "
                  f"(VAT-inclusive); the source PDF printed "
                  f"R{v.source_total:,.2f} before VAT", fatal=False)
        elif not ran:
            v.add("recalculated", True,
                  "skipped - no LibreOffice here. The structural checks above "
                  "do not depend on it.", fatal=False)
    return v