spencermandClaude Fable 5.1
39a4eb5213
The Warehouse home with the numbers, the Open shorts table, the find bar and the five-column task board (waiting, being pulled, on hold, pulled, released in the last 3 days) with buttons, hold reasons and drag. Every role sees the warehouse stage on a packet as the WH pill and the line under it. The pull table on the Materials tab: Pulled with auto-start, issued, location, heat with Enter to the next line, shorts with receipts, N/A, MTR per line with a viewer. Send back to Clerical with a reason. Release to supervisor stays blocked until every line is pulled or short and every pulled line has a heat. Server: the heats_recorded precondition on WH_RELEASE, receive by custody, server-validated hold and short reasons, the old packet history texts, GET /api/warehouse/board, cards with the warehouse stage and pull counts. Uploads typed by content per kind (drawing, mtr, document, bom with XLSX and CSV), limits per kind, optional ClamAV behind the clamav compose profile, a storage quota alert on the new notification_events table (migration 0005) shown on the Admin Storage tab. Tests: every warehouse action as every role, the release gate, the short-receive exception, the heat pattern, upload sniffing and limits, the ClamAV path with a fake scanner, the quota alert; frontend board and pull table tests; Playwright scenarios for the board and the pull flow. Verified on docker-1 in the CI stack: 3557 server tests, 106 frontend tests, 17 Playwright scenarios. Docs updated; D5 closed. Co-Authored-By: Claude Fable 5.1 <[email protected]>
101 lines
3.8 KiB
Python
101 lines
3.8 KiB
Python
"""The policy's view of a person (Principal) and of a packet (PacketState)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Iterable, Mapping
|
|
from dataclasses import dataclass, field
|
|
from types import MappingProxyType
|
|
|
|
from ..auth.groups import SHOP_LEVELS, parse_role
|
|
|
|
SUPPORT = ("CL", "WH", "QC")
|
|
LEVELS: dict[str, int] = {level: i + 1 for i, level in enumerate(SHOP_LEVELS)} # worker 1, lead 2, supervisor 3
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Principal:
|
|
user_id: str | None # uuid as text; None for system
|
|
admin: bool = False
|
|
support: frozenset[str] = frozenset() # codes among CL, WH, QC the user is a member of (any level)
|
|
support_sup: frozenset[str] = frozenset() # codes among CL, WH, QC where the user is Supervisor
|
|
shop: Mapping[str, int] = field(default_factory=lambda: MappingProxyType({})) # code -> 1 worker, 2 lead, 3 sup
|
|
station_limits: frozenset[int] = frozenset() # station ids a worker is limited to; empty = any
|
|
pm: bool = False
|
|
session_kind: str = "password" # password | pin
|
|
system: str | None = None # "visual-feed" for the worker; never from a session
|
|
|
|
def level(self, dept: str) -> int:
|
|
"""Shop level in a department: 0 none, 1 worker, 2 lead, 3 supervisor."""
|
|
return int(self.shop.get(dept, 0))
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PacketState:
|
|
id: int
|
|
status: str # packet_status value
|
|
dept: str # dept_code
|
|
station_id: int | None
|
|
welder_id: str | None # uuid text of the assigned welder
|
|
wh_skipped: bool
|
|
routed_to: str | None # databook | shipping | None
|
|
vclosed: bool
|
|
removed: bool
|
|
source_kind: str # visual | hand
|
|
weld_dept: bool # departments.weld
|
|
co_hold: bool # an approved CO with pending verifications, or a pending CO with hold=true
|
|
co_hold_why: str = "" # text for the error
|
|
open_shorts: int = 0
|
|
drawing_attached: bool = False
|
|
heat_map: bool | None = None
|
|
bom_total: int = 0 # lines not marked n/a
|
|
bom_ready: int = 0 # lines pulled, or with an open short
|
|
bom_missing_heat: int = 0 # pulled lines with pieces issued and no heat (not n/a, not a sub-assembly)
|
|
welds: int = 0
|
|
checklist_complete: bool = False # every key of the department checklist is done (empty list is complete)
|
|
filler_set: bool = False
|
|
qc_missing: tuple[str, ...] = () # the old qcMissing() list, computed by the read model
|
|
has_qc_sent: bool = False # qc_inspections row exists for the current stage
|
|
|
|
|
|
SYSTEM_FEED = Principal(user_id=None, system="visual-feed")
|
|
|
|
|
|
def principal_from_roles(
|
|
user_id: str | None,
|
|
roles: Iterable[str],
|
|
station_limits: Iterable[int] = (),
|
|
session_kind: str = "password",
|
|
) -> Principal:
|
|
"""Build the policy Principal from phase 1 role strings. A role that does not parse is ignored."""
|
|
admin = False
|
|
pm = False
|
|
support: set[str] = set()
|
|
support_sup: set[str] = set()
|
|
shop: dict[str, int] = {}
|
|
for text in roles:
|
|
try:
|
|
role = parse_role(text)
|
|
except ValueError:
|
|
continue
|
|
if role.kind == "admin":
|
|
admin = True
|
|
elif role.kind == "pm":
|
|
pm = True
|
|
elif role.kind.startswith("shop_") and role.dept_code:
|
|
level = LEVELS[role.kind.removeprefix("shop_")]
|
|
shop[role.dept_code] = max(level, shop.get(role.dept_code, 0))
|
|
elif role.dept_code in SUPPORT:
|
|
support.add(role.dept_code)
|
|
if role.kind.endswith("_supervisor"):
|
|
support_sup.add(role.dept_code)
|
|
return Principal(
|
|
user_id=user_id,
|
|
admin=admin,
|
|
support=frozenset(support),
|
|
support_sup=frozenset(support_sup),
|
|
shop=MappingProxyType(shop),
|
|
station_limits=frozenset(int(s) for s in station_limits),
|
|
pm=pm,
|
|
session_kind=session_kind,
|
|
)
|