PacketStation/server/packetstation/domain/checklists.py

"""The department checklists and the Q.C. checks, ported from the old checklistFor(), qcChecksFor() and qcMissing().

Weld departments get the welding list, the others the plain list. The drawing, rev and WPS are filled into the text.
The lists live in the settings row `checklist_templates` (migration 0006); the defaults here are the fallback.
The keys are the old keys, so a checklist stored as {key: {done, at, by, stamp}} reads the same.
"""

from __future__ import annotations

import logging
import re
from collections.abc import Iterable, Mapping
from typing import Any

log = logging.getLogger(__name__)

NO_REV = "-"
NDE_NONE = ("", "none", "none required")
NDE_OK = ("acceptable", "accept", "accepted")
REJECTS = ("reject", "rejected")


def _item(group: str, key: str, text: str) -> dict[str, str]:
    return {"group": group, "key": key, "text": text}


DWG_TEXT = "Drawing {drawing} Rev {rev} matches the packet"

# The defaults. The settings row `checklist_templates` holds the same lists (migration 0006); these are the fallback
# when that row is missing or does not parse.
DEFAULT_TEMPLATES: dict[str, list[dict[str, str]]] = {
    "weld": [
        _item("Before welding", "dwg", DWG_TEXT),
        _item("Before welding", "heat", "Heat numbers on material match the Materials tab"),
        _item("Before welding", "wps", "WPS reviewed: {wps}"),
        _item("Before welding", "fit", "Fit-up and dimensions checked against the drawing"),
        _item("Before welding", "pre", "Preheat and interpass temperature per WPS"),
        _item("After welding", "vt", "Visual inspection of all welds acceptable"),
        _item("After welding", "stamp", "Welds stamped with welder ID"),
        _item("After welding", "clean", "Slag and spatter removed"),
        _item("After welding", "qty", "Completed quantity matches each work order"),
    ],
    "plain": [
        _item("Before starting", "dwg", DWG_TEXT),
        _item("Before starting", "prev", "Previous department's work complete and released"),
        _item("Before starting", "mat", "Material and parts on hand match the packet"),
        _item("When done", "work", "Work completed per drawing and procedure"),
        _item("When done", "insp", "Self-inspection done; no damage to finished surfaces"),
        _item("When done", "qty", "Completed quantity matches each work order"),
    ],
}
SETTINGS_KEY = "checklist_templates"
RX_KEY = re.compile(r"^[a-z][a-z0-9_]{0,29}$")
RX_PLACEHOLDER = re.compile(r"\{(drawing|rev|wps)\}")
TEXT_MAX = 200
ITEMS_MAX = 40


def _valid_list(value: Any) -> list[dict[str, str]] | None:
    """A template list from settings, or None when it does not have the right shape."""
    if not isinstance(value, list) or not value or len(value) > ITEMS_MAX:
        return None
    out: list[dict[str, str]] = []
    seen: set[str] = set()
    for it in value:
        if not isinstance(it, Mapping):
            return None
        group, key, text = it.get("group"), it.get("key"), it.get("text")
        if not (isinstance(group, str) and isinstance(key, str) and isinstance(text, str)):
            return None
        if not RX_KEY.match(key) or key in seen or not group.strip() or not text.strip():
            return None
        seen.add(key)
        out.append(_item(group.strip()[:TEXT_MAX], key, text.strip()[:TEXT_MAX]))
    return out


def parse_templates(value: Any) -> dict[str, list[dict[str, str]]] | None:
    """The `checklist_templates` settings value, checked. None when it is malformed."""
    if not isinstance(value, Mapping):
        return None
    weld, plain = _valid_list(value.get("weld")), _valid_list(value.get("plain"))
    if weld is None or plain is None:
        return None
    return {"weld": weld, "plain": plain}


def load_templates(conn: Any) -> dict[str, list[dict[str, str]]]:
    """The department checklists from settings, else the defaults in code (with a warning when the row is bad)."""
    row = conn.execute("SELECT value FROM settings WHERE key = %s", (SETTINGS_KEY,)).fetchone()
    if row is None:
        return DEFAULT_TEMPLATES
    parsed = parse_templates(row[0])
    if parsed is None:
        log.warning("settings %s does not parse; using the default checklists", SETTINGS_KEY)
        return DEFAULT_TEMPLATES
    return parsed


def _fill(text: str, drawing: str | None, rev: str | None, wps: str | None) -> str:
    values = {"drawing": drawing or "", "rev": rev or NO_REV, "wps": wps or "per drawing"}
    return " ".join(RX_PLACEHOLDER.sub(lambda m: values[m.group(1)], text).split())


def checklist_for(
    weld: bool,
    drawing: str | None,
    rev: str | None,
    wps: str | None,
    templates: Mapping[str, list[dict[str, str]]] | None = None,
) -> list[dict[str, str]]:
    """The stage checklist for a department, as a flat list of {group, key, text} in display order."""
    t = templates or DEFAULT_TEMPLATES
    items = t.get("weld" if weld else "plain") or DEFAULT_TEMPLATES["weld" if weld else "plain"]
    return [_item(i["group"], i["key"], _fill(i["text"], drawing, rev, wps)) for i in items]


def qc_items_for(
    drawing: str | None, rev: str | None, wps: str | None, welder_stamp: str | None
) -> list[dict[str, str]]:
    """The Q.C. inspection items, as a flat list of {group, key, text}."""
    g1, g2 = "Data book package", "Physical spool"
    stamp = f"[{welder_stamp}] " if welder_stamp else ""
    return [
        _item(g1, "db_heats", "Heat number recorded for every BOM item"),
        _item(g1, "db_mtr", "MTRs on file and match the heat numbers"),
        _item(g1, "db_weldmap", "Weld map complete: every weld identified"),
        _item(g1, "db_wps", f"Welds made to {wps or 'the WPS'}; welder {stamp}qualified"),
        _item(g1, "db_ck", "Stall checklist and weld record complete"),
        _item(g2, "ph_dim", f"Dimensions verified against {drawing or 'the drawing'} Rev {rev or NO_REV}"),
        _item(g2, "ph_heat", "Heat numbers on material match the heat map"),
        _item(g2, "ph_stamp", "Welds stamped with welder ID"),
        _item(g2, "ph_clean", "Clean: no slag, spatter or arc strikes"),
        _item(g2, "ph_qty", "Quantity matches the work orders"),
    ]


def checklist_keys(weld: bool, templates: Mapping[str, list[dict[str, str]]] | None = None) -> list[str]:
    return [i["key"] for i in checklist_for(weld, None, None, None, templates)]


def qc_keys() -> list[str]:
    return [i["key"] for i in qc_items_for(None, None, None, None)]


def is_done(value: Any) -> bool:
    """A checklist or Q.C. check entry: `true`, or `{"done": true, ...}`."""
    if value is True:
        return True
    return isinstance(value, Mapping) and value.get("done") is True


def checklist_complete(
    checklist: Mapping[str, Any] | None, weld: bool, templates: Mapping[str, list[dict[str, str]]] | None = None
) -> bool:
    """Every key of the department checklist is done. An empty list counts as complete."""
    ck = checklist or {}
    return all(is_done(ck.get(k)) for k in checklist_keys(weld, templates))


def _plural(n: int, word: str) -> str:
    return f"{n} {word}{'s' if n > 1 else ''}"


def qc_missing(
    packet: Mapping[str, Any], qc: Mapping[str, Any] | None, welds: Iterable[Mapping[str, Any]]
) -> list[str]:
    """What Q.C. still has to record before an accept. The old qcMissing().

    `packet` is unused today (the item list does not depend on it for the keys) but kept for the old signature.
    `qc` is the open inspection row (checks, weld_exam, nde_method, nde_result, inspector) or None.
    `welds` are the live weld rows (weld_no).
    """
    del packet
    q = qc or {}
    checks = q.get("checks") or {}
    missing: list[str] = []
    open_items = sum(1 for k in qc_keys() if not is_done(checks.get(k)))
    if open_items:
        missing.append(_plural(open_items, "inspection item"))
    exam = q.get("weld_exam") or {}
    results = []
    for w in welds:
        entry = exam.get(str(w.get("weld_no")))
        result = entry.get("result") if isinstance(entry, Mapping) else None
        results.append(str(result).lower() if result else "")
    no_result = sum(1 for r in results if not r)
    if no_result:
        missing.append(_plural(no_result, "weld") + " without a visual result")
    if any(r in REJECTS for r in results):
        missing.append("a rejected weld (reject the packet instead)")
    method = str(q.get("nde_method") or "").strip().lower()
    if method not in NDE_NONE and str(q.get("nde_result") or "").strip().lower() not in NDE_OK:
        missing.append("NDE result")
    if not str(q.get("inspector") or "").strip():
        missing.append("inspector name")
    return missing