MEGA-BOQ-WEBAPP/boq_engine/profiles.py

"""
Per-consultant template profiles.

This is NOT machine learning and it is not "training". When a consultant's
workbook is first seen, we detect its layout - which row holds the headers,
which column is ITEM / UNIT / QTY / RATE / AMOUNT - and save that map under
the consultant's name. Next time their file arrives we skip detection and
apply the stored map.

Why that matters: it is deterministic and inspectable. Open a profile and you
can see exactly what the app decided, and correct it. A learned/fuzzy model
would give the same convenience with none of the auditability, and on a
tender document "the model decided" is not an acceptable answer to "where did
this number come from".
"""

from __future__ import annotations

import hashlib
import json
import os
import time
from dataclasses import asdict, dataclass, field

STORE_DIR = os.environ.get(
    "BOQ_PROFILES",
    os.path.join(os.path.expanduser("~"), ".boq_profiles"),
)
STORE_PATH = os.path.join(STORE_DIR, "profiles.json")


def file_fingerprint(path: str) -> str:
    """Content hash, so we can tell whether a file matches a known layout."""
    h = hashlib.sha256()
    with open(path, "rb") as fh:
        for chunk in iter(lambda: fh.read(65536), b""):
            h.update(chunk)
    return h.hexdigest()[:16]


@dataclass
class SheetProfile:
    sheet: str
    header_row: int
    cols: dict[str, int]
    data_rows: int
    total_rows: int
    desc_cols: list = field(default_factory=list)
    consultant: str = ""


@dataclass
class Profile:
    consultant: str
    sheets: list[SheetProfile] = field(default_factory=list)
    fingerprint: str = ""
    saved: str = ""
    uses: int = 0
    notes: list[str] = field(default_factory=list)

    def summary(self) -> str:
        parts = []
        for s in self.sheets:
            cols = ", ".join(f"{k}={v}" for k, v in sorted(
                s.cols.items(), key=lambda kv: kv[1]))
            parts.append(f"{s.sheet} (hdr row {s.header_row}): {cols}")
        return "\n".join(parts)


def _load_all() -> dict[str, dict]:
    if not os.path.exists(STORE_PATH):
        return {}
    try:
        with open(STORE_PATH, "r", encoding="utf-8") as fh:
            return json.load(fh)
    except (json.JSONDecodeError, OSError):
        return {}


def _save_all(data: dict) -> None:
    os.makedirs(STORE_DIR, exist_ok=True)
    tmp = STORE_PATH + ".tmp"
    with open(tmp, "w", encoding="utf-8") as fh:
        json.dump(data, fh, indent=1, sort_keys=True)
    os.replace(tmp, STORE_PATH)


def list_profiles() -> list[Profile]:
    out = []
    for name, raw in sorted(_load_all().items()):
        try:
            p = Profile(
                consultant=raw.get("consultant", name),
                sheets=[SheetProfile(**s) for s in raw.get("sheets", [])],
                fingerprint=raw.get("fingerprint", ""),
                saved=raw.get("saved", ""),
                uses=raw.get("uses", 0),
                notes=raw.get("notes", []),
            )
        except TypeError:
            continue
        out.append(p)
    return out


def get(consultant: str) -> Profile | None:
    raw = _load_all().get(consultant.lower().strip())
    if not raw:
        return None
    return Profile(
        consultant=raw.get("consultant", consultant),
        sheets=[SheetProfile(**s) for s in raw.get("sheets", [])],
        fingerprint=raw.get("fingerprint", ""),
        saved=raw.get("saved", ""),
        uses=raw.get("uses", 0),
        notes=raw.get("notes", []),
    )


def save(consultant: str, layouts: dict, fingerprint: str,
         notes: list[str] | None = None) -> Profile:
    consultant = consultant.strip()
    if not consultant:
        raise ValueError("consultant name is required")
    prof = Profile(
        consultant=consultant,
        sheets=[SheetProfile(
            sheet=name,
            header_row=lay.header_row,
            cols=dict(lay.cols),
            data_rows=len(lay.data_rows),
            total_rows=len(lay.total_rows),
        ) for name, lay in layouts.items()],
        fingerprint=fingerprint,
        saved=time.strftime("%Y-%m-%d %H:%M"),
        notes=notes or [],
    )
    data = _load_all()
    prev = data.get(consultant.lower(), {})
    prof.uses = int(prev.get("uses", 0))
    data[consultant.lower()] = asdict(prof)
    _save_all(data)
    return prof


def touch(consultant: str) -> None:
    data = _load_all()
    k = consultant.lower().strip()
    if k in data:
        data[k]["uses"] = int(data[k].get("uses", 0)) + 1
        _save_all(data)


def delete(consultant: str) -> None:
    data = _load_all()
    if data.pop(consultant.lower().strip(), None) is not None:
        _save_all(data)