PacketStation/server/packetstation/policy/preconditions.py

"""Named preconditions. Each takes (state, extra) and returns the reason when it fails, else None.

They are checked on the server and never bypassed, including by Admin. Some read request input (extra) rather
than packet state; `can()` skips those, because they are inputs the user gives with the step.
"""

from __future__ import annotations

from collections.abc import Callable, Mapping
from typing import Any

from ..text import clean_text, reason_ok
from .actions import SUPPORT_CODES
from .principal import PacketState

Precondition = Callable[[PacketState, Mapping[str, Any]], str | None]

MESSAGES: dict[str, str] = {
    "heat_map_answered": "Say whether this packet needs a heat map.",
    "drawing_attached": "Attach the drawing first.",
    "bom_all_pulled_or_short": "Pull every material line or mark it short first.",
    "short_open": "No part is short on this packet.",
    "welder_set": "Pick a station and a welder.",
    "checklist_complete": "Finish the checklist first.",
    "welds_recorded": "Record at least one weld first.",
    "filler_set": "Enter the filler metal first.",
    "qc_book_spool_location": "Confirm the book and the spool, and give the location.",
    "qc_missing_empty": "Q.C. is not finished.",
    "reason_given": "Give a reason of at least 8 characters.",
    "vclosed_if_databook": "Close the job in VISUAL first.",
    "co_staged_if_dwg_change": "Stage the new drawing for the change order first.",
    "route_given": "Choose Data Book or Shipping.",
    "target_dept_given": "Choose a different department.",
}

# Preconditions that read request input, not packet state. can() skips them.
INPUT_ONLY: frozenset[str] = frozenset(
    {
        "welder_set",
        "target_dept_given",
        "route_given",
        "reason_given",
        "qc_book_spool_location",
        "co_staged_if_dwg_change",
    }
)

ROUTES = ("databook", "shipping")


def _fail(name: str, ok: bool) -> str | None:
    return None if ok else MESSAGES[name]


def heat_map_answered(state: PacketState, extra: Mapping[str, Any]) -> str | None:
    return _fail("heat_map_answered", state.heat_map is not None)


def drawing_attached(state: PacketState, extra: Mapping[str, Any]) -> str | None:
    return _fail("drawing_attached", state.drawing_attached)


def bom_all_pulled_or_short(state: PacketState, extra: Mapping[str, Any]) -> str | None:
    return _fail("bom_all_pulled_or_short", state.bom_ready >= state.bom_total)


def short_open(state: PacketState, extra: Mapping[str, Any]) -> str | None:
    """The old whCanReceive: the warehouse may only receive into a packet with an open short."""
    return _fail("short_open", state.open_shorts > 0)


def welder_set(state: PacketState, extra: Mapping[str, Any]) -> str | None:
    return _fail("welder_set", extra.get("station_id") not in (None, "") and extra.get("welder_id") not in (None, ""))


def checklist_complete(state: PacketState, extra: Mapping[str, Any]) -> str | None:
    return _fail("checklist_complete", state.checklist_complete)


def welds_recorded(state: PacketState, extra: Mapping[str, Any]) -> str | None:
    return _fail("welds_recorded", not state.weld_dept or state.welds > 0)


def filler_set(state: PacketState, extra: Mapping[str, Any]) -> str | None:
    return _fail("filler_set", not state.weld_dept or state.filler_set)


def qc_book_spool_location(state: PacketState, extra: Mapping[str, Any]) -> str | None:
    ok = extra.get("book") is True and extra.get("spool") is True and bool(clean_text(extra.get("location"), 120))
    return _fail("qc_book_spool_location", ok)


def qc_missing_empty(state: PacketState, extra: Mapping[str, Any]) -> str | None:
    if not state.qc_missing:
        return None
    return "Q.C. still needs: " + ", ".join(state.qc_missing) + "."


def reason_given(state: PacketState, extra: Mapping[str, Any]) -> str | None:
    return _fail("reason_given", reason_ok(extra.get("reason")) is not None)


def vclosed_if_databook(state: PacketState, extra: Mapping[str, Any]) -> str | None:
    return _fail("vclosed_if_databook", state.routed_to != "databook" or state.vclosed)


def co_staged_if_dwg_change(state: PacketState, extra: Mapping[str, Any]) -> str | None:
    return _fail("co_staged_if_dwg_change", extra.get("dwg_change") is not True or extra.get("staged") is True)


def route_given(state: PacketState, extra: Mapping[str, Any]) -> str | None:
    return _fail("route_given", extra.get("routed_to") in ROUTES)


def is_shop_code(value: Any) -> bool:
    """A non-empty department code that is not a support code. The API checks it exists."""
    return isinstance(value, str) and bool(value.strip()) and value not in SUPPORT_CODES


def target_dept_given(state: PacketState, extra: Mapping[str, Any]) -> str | None:
    """A shop department different from the packet's. For a draft (RELEASE_TO_SHOP) any shop department."""
    dept = extra.get("dept")
    if not is_shop_code(dept):
        return "Choose a department."
    if state.status != "draft" and dept == state.dept:
        return MESSAGES["target_dept_given"]
    return None


PRECONDITIONS: dict[str, Precondition] = {
    "heat_map_answered": heat_map_answered,
    "drawing_attached": drawing_attached,
    "bom_all_pulled_or_short": bom_all_pulled_or_short,
    "short_open": short_open,
    "welder_set": welder_set,
    "checklist_complete": checklist_complete,
    "welds_recorded": welds_recorded,
    "filler_set": filler_set,
    "qc_book_spool_location": qc_book_spool_location,
    "qc_missing_empty": qc_missing_empty,
    "reason_given": reason_given,
    "vclosed_if_databook": vclosed_if_databook,
    "co_staged_if_dwg_change": co_staged_if_dwg_change,
    "route_given": route_given,
    "target_dept_given": target_dept_given,
}