spencermandClaude Opus 5.5
a82e056b6a
Admin: Departments (name and stations; add shows the three groups IT must create), Profiles (welder stamp, station limits, active) with a CSV import that previews and applies only when clean, Checklists (weld and plain), Security (session, lockout and rate-limit settings with bounds, alerts for the last 24 hours, the integrity result), Access log with filters and a guarded CSV, Who is in, Stamp check, Notifications, and the employeeID flags. Every Admin change is audited. Worker: ad-reconcile (nightly and on "refresh from AD now"), session-refresh (15 minutes), integrity (seals, asset hashes, missing and orphan files, the audit chain), backup-check, notify-send and notify-digest. Notifications route per event type to AD groups, users, mailboxes or the department Supervisor group, immediately or in a daily digest, over STARTTLS verified against the PESCO CA, with dry run on the test stack. Backups: scripts/backup/ps-backup.sh and ps-restore.sh with a systemd timer at 02:00 America/Denver. Restore rehearsed on the test stack 2026-10-21. CI: pip-audit and npm audit. pypdf 6.21.1 and certifi 2026.10.19. Migration 0010. Closes the app part of I6, M3, O1, O2 and S3. Co-Authored-By: Claude Opus 5.5 <[email protected]>
632 lines
24 KiB
Python
632 lines
24 KiB
Python
"""The digital ITP (Inspection and Test Plan), one per unit.
|
||
|
||
Starting an ITP freezes a copy of the Q.C. template (domain/itp_template.py) into itps.template, so a later template
|
||
change never alters an ITP in use. Q.C. initials each line (approve), marks lines N/A, rejects a line with an NCR
|
||
number (now or later), records the customer's code and witness, signs each block of sections with a signature, and
|
||
signs the ITP off with a signature. Open hold (H) and witness (W) points, and rejected lines, show as a warning on the
|
||
unit's packets for the department's pages. Nothing is blocked by them.
|
||
|
||
The pure functions here (items, is_done, is_hold_code, open_holds, blocks, block_ready, progress, dept_pages,
|
||
holds_for_packet) port the old itp.js helpers. They take the frozen template and the lines as a mapping of line key to
|
||
a row with `result` ("approve", "reject", "na" or None), `cust_code` and `ncr`.
|
||
|
||
Writes go through `itp_write`: check_scoped with the action, then the ITP row locked FOR UPDATE, the complete lock,
|
||
the signature when the rule needs one, then the version bump, the itp_log rows, the audit rows and the NOTIFY. ITP
|
||
writes never look at a packet's status: a filed packet in the unit does not lock the unit's ITP.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from collections.abc import Iterable, Iterator, Mapping, Sequence
|
||
from contextlib import contextmanager
|
||
from dataclasses import dataclass, field
|
||
from datetime import date, datetime
|
||
from typing import Any
|
||
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
||
|
||
from .. import audit, events, policy
|
||
from ..api import deps
|
||
from ..api.errors import ApiError
|
||
from ..auth import signatures
|
||
from ..auth.signatures import Signature
|
||
from ..notify import raise_event
|
||
from ..policy.actions import SCOPED_LABELS
|
||
from ..policy.engine import check_scoped
|
||
from ..text import LIMITS, clean_opt, reason_ok
|
||
from . import packets
|
||
|
||
A = policy.Action
|
||
|
||
PAGE_NAMES: dict[str, str] = {"WP": "Weld Parts", "C": "Code", "FA": "Final Assembly"}
|
||
PAGE_ORDER: tuple[str, ...] = ("WP", "C", "FA")
|
||
# The pages each shop department works to (the old ITP_DEPT_PAGES). A support department or an unknown code sees all.
|
||
DEPT_PAGES: dict[str, tuple[str, ...]] = {
|
||
"01": ("WP",),
|
||
"02": ("WP",),
|
||
"03": ("WP",),
|
||
"04": ("WP",),
|
||
"05": ("WP",),
|
||
"10": ("WP",),
|
||
"06": ("C",),
|
||
"07": ("C",),
|
||
"13": ("C",),
|
||
"09": ("WP", "C"),
|
||
"08": ("FA",),
|
||
"11": ("FA",),
|
||
"12": ("FA",),
|
||
"14": ("FA",),
|
||
}
|
||
CUST_CODES: tuple[str, ...] = ("R", "R/AU", "AU", "W", "H", "NA")
|
||
RX_HOLD = re.compile(r"(^|/)(H|W)$")
|
||
LOG_MAX = 600
|
||
|
||
MSG_COMPLETE = "The ITP is signed off and locked. An admin can reopen it."
|
||
MSG_NO_ITP = "No such ITP."
|
||
|
||
# ---------------------------------------------------------------- pure functions over the template and lines
|
||
|
||
|
||
def items(template: Mapping[str, Any]) -> list[dict[str, Any]]:
|
||
"""Every line of the template in order, with its section number (`sec`) and page."""
|
||
out: list[dict[str, Any]] = []
|
||
for s in template.get("sections") or []:
|
||
for i in s.get("items") or []:
|
||
out.append({**i, "sec": s.get("n"), "page": s.get("page")})
|
||
return out
|
||
|
||
|
||
def _line(lines: Mapping[str, Any], key: str) -> Mapping[str, Any]:
|
||
v = lines.get(key)
|
||
return v if isinstance(v, Mapping) else {}
|
||
|
||
|
||
def is_done(line: Mapping[str, Any] | None) -> bool:
|
||
"""A line is done when it is N/A or approved."""
|
||
return bool(line) and (line or {}).get("result") in ("approve", "na")
|
||
|
||
|
||
def is_rejected(line: Mapping[str, Any] | None) -> bool:
|
||
"""An open reject: the line is rejected and not approved since."""
|
||
return bool(line) and (line or {}).get("result") == "reject"
|
||
|
||
|
||
def is_hold_code(code: Any) -> bool:
|
||
"""H or W, alone or after a slash (R/H). The old isHoldCode."""
|
||
return bool(RX_HOLD.search(str(code or "")))
|
||
|
||
|
||
def cust_code(item: Mapping[str, Any], line: Mapping[str, Any] | None) -> str:
|
||
"""The customer code of a line: its own when set, else the template's."""
|
||
own = (line or {}).get("cust_code")
|
||
return str(own) if own else str(item.get("cust") or "")
|
||
|
||
|
||
def hold_codes(item: Mapping[str, Any], line: Mapping[str, Any] | None) -> list[str]:
|
||
return [c for c in (str(item.get("code") or ""), cust_code(item, line)) if is_hold_code(c)]
|
||
|
||
|
||
def open_holds(
|
||
template: Mapping[str, Any], lines: Mapping[str, Any], pages: Sequence[str] | None = None
|
||
) -> list[dict[str, Any]]:
|
||
"""The hold, witness or rejected points not done yet, on the given pages (None: every page)."""
|
||
out: list[dict[str, Any]] = []
|
||
for i in items(template):
|
||
if pages is not None and i["page"] not in pages:
|
||
continue
|
||
line = _line(lines, str(i["n"]))
|
||
codes = hold_codes(i, line)
|
||
rej = is_rejected(line)
|
||
if (codes or rej) and not is_done(line):
|
||
out.append(
|
||
{
|
||
"n": i["n"],
|
||
"t": i["t"],
|
||
"page": i["page"],
|
||
"codes": codes,
|
||
"rejected": rej,
|
||
"ncr": line.get("ncr") if rej else None,
|
||
}
|
||
)
|
||
return out
|
||
|
||
|
||
def blocks(template: Mapping[str, Any]) -> list[dict[str, Any]]:
|
||
"""The signature blocks: one closes at every section with `sig` and covers the sections since the last block."""
|
||
out: list[dict[str, Any]] = []
|
||
acc: list[Mapping[str, Any]] = []
|
||
for s in template.get("sections") or []:
|
||
acc.append(s)
|
||
if s.get("sig"):
|
||
out.append({"key": str(s["n"]), "sections": acc, "final": bool(s.get("final"))})
|
||
acc = []
|
||
if acc:
|
||
out.append({"key": str(acc[-1]["n"]), "sections": acc, "final": True})
|
||
return out
|
||
|
||
|
||
def block_scope(block: Mapping[str, Any]) -> str:
|
||
secs = block["sections"]
|
||
if len(secs) > 1:
|
||
return f"Sections {secs[0]['n']}–{secs[-1]['n']}"
|
||
return f"Section {block['key']}"
|
||
|
||
|
||
def block_of_line(template: Mapping[str, Any], key: str) -> dict[str, Any] | None:
|
||
for b in blocks(template):
|
||
if any(str(i.get("n")) == key for s in b["sections"] for i in s.get("items") or []):
|
||
return b
|
||
return None
|
||
|
||
|
||
def _block_lines(block: Mapping[str, Any]) -> list[str]:
|
||
return [str(i.get("n")) for s in block["sections"] for i in s.get("items") or []]
|
||
|
||
|
||
def block_ready(block: Mapping[str, Any], lines: Mapping[str, Any]) -> bool:
|
||
"""Every line of the block is done (approved or N/A)."""
|
||
return all(is_done(_line(lines, k)) for k in _block_lines(block))
|
||
|
||
|
||
def block_all_na(block: Mapping[str, Any], lines: Mapping[str, Any]) -> bool:
|
||
return all(_line(lines, k).get("result") == "na" for k in _block_lines(block))
|
||
|
||
|
||
def progress(template: Mapping[str, Any], lines: Mapping[str, Any]) -> dict[str, int]:
|
||
it = items(template)
|
||
return {
|
||
"done": sum(1 for i in it if is_done(_line(lines, str(i["n"])))),
|
||
"total": len(it),
|
||
"holds": len(open_holds(template, lines)),
|
||
}
|
||
|
||
|
||
def dept_pages(code: Any) -> list[str] | None:
|
||
"""The pages a department works to, or None for a support department or an unknown code (every page)."""
|
||
pages = DEPT_PAGES.get(str(code or ""))
|
||
return list(pages) if pages is not None else None
|
||
|
||
|
||
def holds_for_packet(
|
||
itp_row: Mapping[str, Any] | None, lines: Mapping[str, Any], dept_code: Any
|
||
) -> dict[str, Any] | None:
|
||
"""The packet's ITP warning: open hold, witness or rejected points on its department's pages, while the unit's
|
||
ITP is open. None when there is none."""
|
||
if itp_row is None or str(itp_row.get("status")) == "complete":
|
||
return None
|
||
pages = dept_pages(dept_code)
|
||
hs = open_holds(itp_row["template"] or {}, lines, pages)
|
||
if not hs:
|
||
return None
|
||
return {
|
||
"no": itp_row.get("no"),
|
||
"count": len(hs),
|
||
"rejected": any(h["rejected"] for h in hs),
|
||
"pages": pages,
|
||
"items": [{k: h[k] for k in ("n", "t", "codes", "rejected", "ncr")} for h in hs],
|
||
}
|
||
|
||
|
||
def initials(name: Any) -> str:
|
||
"""The first letter of each word of a name, upper case, at most three. The old initials()."""
|
||
return "".join(w[0].upper() for w in str(name or "").split() if w)[:3]
|
||
|
||
|
||
def site_zone(tz: str | None) -> ZoneInfo:
|
||
try:
|
||
return ZoneInfo(tz or "America/Denver")
|
||
except (ZoneInfoNotFoundError, ValueError):
|
||
return ZoneInfo("America/Denver")
|
||
|
||
|
||
def date_text(value: Any, tz: str | None = None) -> str:
|
||
"""M/D/YYYY in the site time zone (10/17/2026)."""
|
||
if value is None:
|
||
return ""
|
||
if isinstance(value, datetime):
|
||
d = value.astimezone(site_zone(tz)).date() if value.tzinfo else value.date()
|
||
elif isinstance(value, date):
|
||
d = value
|
||
else:
|
||
return str(value)
|
||
return f"{d.month}/{d.day}/{d.year}"
|
||
|
||
|
||
def today(tz: str | None = None) -> date:
|
||
return datetime.now(site_zone(tz)).date()
|
||
|
||
|
||
# ---------------------------------------------------------------- loading
|
||
|
||
|
||
def load_row(conn: Any, itp_id: int, *, lock: bool = False) -> dict[str, Any] | None:
|
||
sql = "SELECT *, status::text AS status FROM itps WHERE id = %s"
|
||
if lock:
|
||
sql += " FOR UPDATE"
|
||
return packets.one(conn, sql, (itp_id,))
|
||
|
||
|
||
def row_for_unit(conn: Any, unit_id: int) -> dict[str, Any] | None:
|
||
return packets.one(conn, "SELECT *, status::text AS status FROM itps WHERE unit_id = %s", (unit_id,))
|
||
|
||
|
||
def load_lines(conn: Any, itp_id: int) -> dict[str, dict[str, Any]]:
|
||
return {r["key"]: r for r in packets.rows(conn, "SELECT * FROM itp_lines WHERE itp_id = %s", (itp_id,))}
|
||
|
||
|
||
def load_blocks(conn: Any, itp_id: int) -> dict[str, dict[str, Any]]:
|
||
return {r["block_key"]: r for r in packets.rows(conn, "SELECT * FROM itp_blocks WHERE itp_id = %s", (itp_id,))}
|
||
|
||
|
||
def summary(itp_row: Mapping[str, Any], lines: Mapping[str, Any]) -> dict[str, Any]:
|
||
pr = progress(itp_row["template"] or {}, lines)
|
||
return {
|
||
"id": itp_row["id"],
|
||
"no": itp_row["no"],
|
||
"status": str(itp_row["status"]),
|
||
"done": pr["done"],
|
||
"total": pr["total"],
|
||
"holds": pr["holds"],
|
||
}
|
||
|
||
|
||
def unit_itps(conn: Any, unit_ids: Iterable[int]) -> dict[int, tuple[dict[str, Any], dict[str, dict[str, Any]]]]:
|
||
"""The ITP row and its lines per unit, for the given units, in two queries."""
|
||
ids = sorted({int(u) for u in unit_ids})
|
||
if not ids:
|
||
return {}
|
||
itps = packets.rows(conn, "SELECT *, status::text AS status FROM itps WHERE unit_id = ANY(%s)", (ids,))
|
||
if not itps:
|
||
return {}
|
||
lines: dict[int, dict[str, dict[str, Any]]] = {r["id"]: {} for r in itps}
|
||
for ln in packets.rows(
|
||
conn,
|
||
"SELECT itp_id, key, result, cust_code, ncr FROM itp_lines WHERE itp_id = ANY(%s)",
|
||
([r["id"] for r in itps],),
|
||
):
|
||
lines[ln["itp_id"]][ln["key"]] = ln
|
||
return {int(r["unit_id"]): (r, lines[r["id"]]) for r in itps}
|
||
|
||
|
||
def summaries(conn: Any, unit_ids: Iterable[int]) -> dict[int, dict[str, Any]]:
|
||
return {u: summary(row, lines) for u, (row, lines) in unit_itps(conn, unit_ids).items()}
|
||
|
||
|
||
# ---------------------------------------------------------------- the read
|
||
|
||
|
||
def can_of(pp: policy.Principal, itp_row: Mapping[str, Any], blocks_signed: bool) -> dict[str, bool]:
|
||
open_ = str(itp_row["status"]) != "complete"
|
||
edit = open_ and check_scoped(pp, A.ITP_EDIT_LINE, assume_it_reason=True).ok
|
||
sign_off = open_ and blocks_signed and check_scoped(pp, A.ITP_SIGN_OFF, assume_it_reason=True).ok
|
||
reopen = (not open_) and check_scoped(pp, A.ITP_REOPEN, extra={"reason": "x" * 8}, assume_it_reason=True).ok
|
||
return {"edit": edit, "sign_off": sign_off, "reopen": reopen}
|
||
|
||
|
||
def read(conn: Any, itp_row: Mapping[str, Any], pp: policy.Principal) -> dict[str, Any]:
|
||
"""The full ITP answer (GET /api/units/{unit_id}/itp and every ITP write)."""
|
||
iid = itp_row["id"]
|
||
template = itp_row["template"] or {}
|
||
lines = load_lines(conn, iid)
|
||
brows = load_blocks(conn, iid)
|
||
rejects: dict[str, list[dict[str, Any]]] = {}
|
||
for r in packets.rows(conn, "SELECT * FROM itp_rejects WHERE itp_id = %s ORDER BY id", (iid,)):
|
||
rejects.setdefault(r["line_key"], []).append(r)
|
||
log_rows = packets.rows(conn, "SELECT * FROM itp_log WHERE itp_id = %s ORDER BY id DESC LIMIT %s", (iid, LOG_MAX))
|
||
ids: list[Any] = [itp_row["signoff_by"], itp_row["reopen_by"], itp_row["created_by"]]
|
||
for ln in lines.values():
|
||
ids += [ln["by"], ln["cust_rec"]]
|
||
for b in brows.values():
|
||
ids += [b["signed_by"], b["cust_by"]]
|
||
for rs in rejects.values():
|
||
ids += [r["by"] for r in rs]
|
||
ids += [r["by"] for r in log_rows]
|
||
people = packets.People(conn, ids)
|
||
unit = packets.one(
|
||
conn,
|
||
"SELECT u.id, u.number, u.project_id, u.project_number FROM v_units u WHERE u.id = %s",
|
||
(itp_row["unit_id"],),
|
||
)
|
||
|
||
out_lines: dict[str, Any] = {}
|
||
for i in items(template):
|
||
key = str(i["n"])
|
||
ln = lines.get(key) or {}
|
||
result = ln.get("result")
|
||
out_lines[key] = {
|
||
"key": key,
|
||
"result": result,
|
||
"na": result == "na",
|
||
"by": people.ref(ln.get("by")) if result == "approve" else None,
|
||
"initials": ln.get("initials") if result == "approve" else None,
|
||
"at": ln.get("at") if result == "approve" else None,
|
||
"ncr": ln.get("ncr"),
|
||
"ncr_missing": result == "reject" and not ln.get("ncr"),
|
||
"note": ln.get("note"),
|
||
"cust_code": ln.get("cust_code"),
|
||
"cust_by": ln.get("cust_by"),
|
||
"cust_at": ln.get("cust_at"),
|
||
"cust_rec": people.ref(ln.get("cust_rec")),
|
||
"rejects": [
|
||
{"at": r["at"], "by": people.ref(r["by"]), "ncr": r["ncr"], "note": r["note"], "kind": r["kind"]}
|
||
for r in rejects.get(key, [])
|
||
],
|
||
"row_version": ln.get("row_version", 0),
|
||
}
|
||
out_blocks: list[dict[str, Any]] = []
|
||
for b in blocks(template):
|
||
br = brows.get(b["key"]) or {}
|
||
out_blocks.append(
|
||
{
|
||
"key": b["key"],
|
||
"sections": [str(s["n"]) for s in b["sections"]],
|
||
"scope": block_scope(b),
|
||
"final": b["final"],
|
||
"ready": block_ready(b, lines),
|
||
"all_na": block_all_na(b, lines),
|
||
"signed": br.get("signed_at") is not None,
|
||
"signed_at": br.get("signed_at"),
|
||
"signed_by": people.ref(br.get("signed_by")) if br.get("signed_at") is not None else None,
|
||
"qn": br.get("qn"),
|
||
"remarks": br.get("remarks"),
|
||
"cust_name": br.get("cust_name"),
|
||
"cust_at": br.get("cust_at"),
|
||
"cust_by": people.ref(br.get("cust_by")),
|
||
"row_version": br.get("row_version", 0),
|
||
}
|
||
)
|
||
all_signed = bool(out_blocks) and all(b["signed"] for b in out_blocks)
|
||
pages_present = {s.get("page") for s in template.get("sections") or []}
|
||
status = str(itp_row["status"])
|
||
return {
|
||
"id": iid,
|
||
"unit_id": itp_row["unit_id"],
|
||
"unit": None
|
||
if unit is None
|
||
else {
|
||
"id": unit["id"],
|
||
"number": unit["number"],
|
||
"project": {"id": unit["project_id"], "number": unit["project_number"]},
|
||
},
|
||
"no": itp_row["no"],
|
||
"hdr": {k: (itp_row["hdr"] or {}).get(k) for k in HDR_KEYS},
|
||
"status": "complete" if status == "complete" else "open",
|
||
"version": itp_row["version"],
|
||
"template": template,
|
||
"pages": [{"code": c, "name": PAGE_NAMES[c]} for c in PAGE_ORDER if c in pages_present],
|
||
"lines": out_lines,
|
||
"blocks": out_blocks,
|
||
"progress": progress(template, lines),
|
||
"holds": open_holds(template, lines),
|
||
"signoff": None
|
||
if itp_row["signoff_at"] is None
|
||
else {
|
||
"at": itp_row["signoff_at"],
|
||
"by": people.ref(itp_row["signoff_by"]),
|
||
"asset_id": packets.uid_text(itp_row["signoff_asset_id"]),
|
||
},
|
||
"reopen": None
|
||
if itp_row["reopen_at"] is None
|
||
else {"at": itp_row["reopen_at"], "by": people.ref(itp_row["reopen_by"]), "reason": itp_row["reopen_reason"]},
|
||
"created_at": itp_row["created_at"],
|
||
"created_by": people.ref(itp_row["created_by"]),
|
||
"can": can_of(pp, itp_row, all_signed),
|
||
"log": [
|
||
{
|
||
"id": r["id"],
|
||
"at": r["at"],
|
||
"by": people.ref(r["by"]),
|
||
"by_name": r["by_name"],
|
||
"action": r["action"],
|
||
"note": r["note"],
|
||
"signed": r["signature_id"] is not None,
|
||
}
|
||
for r in log_rows
|
||
],
|
||
}
|
||
|
||
|
||
HDR_KEYS: tuple[str, ...] = ("client", "pname", "pno", "po", "unit")
|
||
|
||
|
||
# ---------------------------------------------------------------- the write
|
||
|
||
|
||
@dataclass
|
||
class ItpWrite:
|
||
conn: Any
|
||
principal: deps.Principal
|
||
policy_principal: policy.Principal
|
||
itp: dict[str, Any]
|
||
action: policy.Action
|
||
decision: policy.Decision
|
||
request: Any
|
||
it_reason: str | None # the cleaned IT reason, kept only when the admin bypass was used
|
||
signature: Signature | None = None
|
||
logs: list[tuple[str, str | None]] = field(default_factory=list)
|
||
version: int = 0
|
||
|
||
@property
|
||
def id(self) -> int:
|
||
return int(self.itp["id"])
|
||
|
||
@property
|
||
def user_id(self) -> Any:
|
||
return self.principal.user_id
|
||
|
||
@property
|
||
def tz(self) -> str:
|
||
return str(deps.settings_of(self.request).site_tz)
|
||
|
||
def log(self, action: str, note: str | None = None) -> None:
|
||
"""One itp_log row (and audit row) for this write. A write that logs nothing changes nothing."""
|
||
self.logs.append((action, note))
|
||
|
||
|
||
def refuse(decision: policy.Decision) -> None:
|
||
if decision.ok:
|
||
return
|
||
status = 400 if decision.code in ("it_reason_required", "reason_required") else 403
|
||
raise ApiError(status, decision.code, decision.why)
|
||
|
||
|
||
def scoped_decision(
|
||
p: deps.Principal,
|
||
pp: policy.Principal,
|
||
action: policy.Action,
|
||
*,
|
||
extra: Mapping[str, Any] | None = None,
|
||
it_reason: str | None = None,
|
||
) -> tuple[policy.Decision, str | None]:
|
||
"""check_scoped, raising the refusal. Returns the decision and the IT reason to record (bypass only)."""
|
||
del p
|
||
decision = check_scoped(pp, action, extra=extra, it_reason=it_reason)
|
||
refuse(decision)
|
||
return decision, (reason_ok(it_reason) if decision.admin_bypass else None)
|
||
|
||
|
||
@contextmanager
|
||
def itp_write(
|
||
d: deps.Db,
|
||
p: deps.Principal,
|
||
itp_id: int,
|
||
action: policy.Action,
|
||
*,
|
||
request: Any,
|
||
extra: Mapping[str, Any] | None = None,
|
||
it_reason: str | None = None,
|
||
sign: Any = None,
|
||
allow_complete: bool = False,
|
||
sign_extra: Mapping[str, Any] | None = None,
|
||
) -> Iterator[ItpWrite]:
|
||
"""One ITP write, in one transaction. See the module docstring for the order of steps.
|
||
|
||
A wrong signature secret commits the step-up failure records and then raises, like the packet pipeline.
|
||
"""
|
||
conn = d.conn
|
||
pp = packets.policy_principal_for(conn, p)
|
||
decision, it_kept = scoped_decision(p, pp, action, extra=extra, it_reason=it_reason)
|
||
failure: ApiError | None = None
|
||
with conn.transaction():
|
||
row = load_row(conn, itp_id, lock=True)
|
||
if row is None:
|
||
raise ApiError(404, "not_found", MSG_NO_ITP)
|
||
if str(row["status"]) == "complete" and not allow_complete:
|
||
raise ApiError(409, "itp_complete", MSG_COMPLETE)
|
||
w = ItpWrite(
|
||
conn=conn,
|
||
principal=p,
|
||
policy_principal=pp,
|
||
itp=row,
|
||
action=action,
|
||
decision=decision,
|
||
request=request,
|
||
it_reason=it_kept,
|
||
version=int(row["version"]),
|
||
)
|
||
if decision.signature:
|
||
if sign is None:
|
||
raise ApiError(
|
||
400, "signature_required", "This step needs your signature.", extra={"meaning": decision.signature}
|
||
)
|
||
try:
|
||
w.signature = signatures.sign(
|
||
conn,
|
||
deps.directory_of(request),
|
||
p.actor,
|
||
factor=sign.factor,
|
||
secret=sign.secret,
|
||
action=f"itp.{action}",
|
||
meaning=decision.signature,
|
||
packet_id=None,
|
||
subject_type="itp",
|
||
subject_id=itp_id,
|
||
content={
|
||
"action": str(action),
|
||
"itp_id": itp_id,
|
||
"version": int(row["version"]),
|
||
**(sign_extra or {}),
|
||
},
|
||
ip=p.ip,
|
||
settings=deps.settings_of(request),
|
||
)
|
||
except ApiError as e:
|
||
failure = e
|
||
if failure is None:
|
||
yield w
|
||
_finish(w)
|
||
if failure is not None:
|
||
raise failure
|
||
|
||
|
||
def _finish(w: ItpWrite) -> None:
|
||
if not w.logs:
|
||
return
|
||
conn = w.conn
|
||
w.version = int(
|
||
conn.execute("UPDATE itps SET version = version + 1 WHERE id = %s RETURNING version", (w.id,)).fetchone()[0]
|
||
)
|
||
sig_id = w.signature.id if w.signature else None
|
||
name = w.principal.user["display_name"]
|
||
for n, (action, note) in enumerate(w.logs):
|
||
text = clean_opt(note, LIMITS["note"]) if note else None
|
||
sid = sig_id if n == 0 else None
|
||
conn.execute(
|
||
"INSERT INTO itp_log (itp_id, by, by_name, action, note, signature_id, it_reason)"
|
||
" VALUES (%s, %s, %s, %s, %s, %s, %s)",
|
||
(w.id, w.user_id, name, action, text, sid, w.it_reason),
|
||
)
|
||
detail: dict[str, Any] = {"text": action, "note": text, "unit_id": w.itp["unit_id"]}
|
||
if sid:
|
||
detail["signature_id"] = sid
|
||
audit.append(
|
||
conn,
|
||
action=f"itp.{w.action}",
|
||
user_id=w.user_id,
|
||
session_id=w.principal.session_id,
|
||
role_snapshot=w.principal.role_snapshot,
|
||
subject_type="itp",
|
||
subject_id=w.id,
|
||
detail=detail,
|
||
it_reason=w.it_reason,
|
||
ip=w.principal.ip,
|
||
)
|
||
events.publish(conn, {"event": "changed", "kind": "itp", "unit_id": w.itp["unit_id"], "itp_id": w.id})
|
||
|
||
|
||
def reject_dept(conn: Any, unit_id: Any, page: Any) -> str | None:
|
||
"""The department of a rejected line: the one department among the unit's live packets that works to the
|
||
line's page, or None when there is none or more than one."""
|
||
codes = {
|
||
str(r[0])
|
||
for r in conn.execute(
|
||
"SELECT DISTINCT dept_code FROM packets WHERE unit_id = %s AND removed_at IS NULL", (unit_id,)
|
||
).fetchall()
|
||
}
|
||
hits = sorted(c for c in codes if page and page in DEPT_PAGES.get(c, ()))
|
||
return hits[0] if len(hits) == 1 else None
|
||
|
||
|
||
def notify_reject(w: ItpWrite, key: str, ncr: str | None, note: str | None) -> int | None:
|
||
"""qc_reject for an ITP line reject, in the write's transaction (notify-send mails it)."""
|
||
item = next((i for i in items(w.itp.get("template") or {}) if str(i.get("n")) == key), {})
|
||
page = item.get("page")
|
||
dept = reject_dept(w.conn, w.itp["unit_id"], page)
|
||
who = (w.principal.user or {}).get("display_name") or ""
|
||
head = f"NCR {ncr}" if ncr else "NCR # pending"
|
||
lines = [str(item.get("t") or ""), head, note or "", f"Rejected by {who}." if who else ""]
|
||
where = f" ({PAGE_NAMES.get(str(page), page)})" if page else ""
|
||
return raise_event(
|
||
w.conn,
|
||
"qc_reject",
|
||
subject=f"itp:{w.id}:{key}",
|
||
title=f"ITP {w.itp.get('no') or w.id} line {key} rejected{where}",
|
||
body="\n".join(x for x in lines if x),
|
||
detail={"itp_id": w.id, "unit_id": w.itp["unit_id"], "line": key, "page": page, "ncr": ncr},
|
||
dept_code=dept,
|
||
)
|
||
|
||
|
||
def label(action: policy.Action) -> str:
|
||
return SCOPED_LABELS.get(action, str(action))
|