spencermandClaude Fable 5.1
904437153d
Clerical close and file: the VISUAL observed state beside the close confirm, a 409 with a second confirm when the feed still shows work orders open, filing with a signature and the HMAC seal, the data book built on the server and stored on the seal in the filing transaction, admin reopen keeping the broken seal with its data book. Server PDFs with reportlab, pypdf and pypdfium2: the manufacturing data book page for page after the old build (cover, index, drawings with VOID stamps, traceability and heat map rasters, weld maps and welding records, Q.C. inspections, Q.C. and test documents, MTRs, record history), the job traveler with work order QR codes, the ITP PDF filed automatically at sign-off, every print kind with the PRINTED COPY stamp, the controlled-copy stamp signed with the seal key, PDF metadata, the traceability CSV with the formula guard, and the append-only print_log with an audit row for every print and export (migration 0008). Frontend: the close and file panel, the print menu and its entry points, the printed copies list on the History tab, the ITP PDF buttons. Tests: server (close, file, reopen, seal, the builders, prints, CSV, stamps), frontend, and the Playwright close and print scenarios. Docs updated; SECURITY closes M2, S6, S7 and S8. Co-Authored-By: Claude Fable 5.1 <[email protected]>
838 lines
33 KiB
Python
838 lines
33 KiB
Python
"""The digital ITP: read, start, header, lines, sections, blocks, sign-off and reopen.
|
|
|
|
Reads need any session. Every write runs policy.engine.check_scoped with its action, then locks the ITP row; the
|
|
answer is the full ITP read. ITP writes never look at a packet: a filed packet in the unit does not lock the ITP.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import date
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, Query, Request, Response
|
|
from psycopg.types.json import Jsonb
|
|
from pydantic import BaseModel, Field
|
|
|
|
from .. import policy
|
|
from ..domain import itp as dom
|
|
from ..domain import ncr as ncrs
|
|
from ..domain import packets
|
|
from ..domain import prints as prints_dom
|
|
from ..domain.itp_template import ITP_TEMPLATE
|
|
from ..domain.pipeline import SignBody, apply_field_cas, jsonable, parse_if_match
|
|
from ..text import LIMITS, clean_opt, clean_text
|
|
from .deps import Db, Principal, current_principal, get_db, settings_of
|
|
from .errors import ApiError
|
|
from .packets import WRITE
|
|
from .prints import file_response, now_utc, pdf_failed, stamp_of
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(tags=["itp"])
|
|
|
|
A = policy.Action
|
|
|
|
HDR_MAX = 120
|
|
CUST_BY_MAX = 80
|
|
NOTE_MAX = 500
|
|
QN_MAX = 40
|
|
REMARKS_MAX = 400
|
|
CUST_NAME_MAX = 80
|
|
HDR_FIELDS = ("no", *dom.HDR_KEYS)
|
|
|
|
MSG_NO_LINE = "No such ITP line."
|
|
MSG_NO_SECTION = "No such ITP section."
|
|
MSG_NO_BLOCK = "No such ITP block."
|
|
|
|
|
|
# ---------------------------------------------------------------- bodies
|
|
|
|
|
|
class It(BaseModel):
|
|
it_reason: str | None = Field(default=None, max_length=1000)
|
|
|
|
|
|
class HeaderBody(It):
|
|
set: dict[str, str | None] = Field(max_length=10)
|
|
base: dict[str, str | None] = Field(default_factory=dict, max_length=10)
|
|
|
|
|
|
class LineBody(It):
|
|
na: bool | None = None
|
|
cust_code: str | None = Field(default=None, max_length=20)
|
|
cust_by: str | None = Field(default=None, max_length=500)
|
|
cust_at: date | None = None
|
|
note: str | None = Field(default=None, max_length=2000)
|
|
|
|
|
|
class SectionNaBody(It):
|
|
na: bool
|
|
|
|
|
|
class RejectBody(It):
|
|
ncr: str | None = Field(default=None, max_length=100)
|
|
note: str | None = Field(default=None, max_length=2000)
|
|
|
|
|
|
class NcrBody(It):
|
|
ncr: str = Field(max_length=100)
|
|
|
|
|
|
class BlockBody(It):
|
|
qn: str | None = Field(default=None, max_length=500)
|
|
remarks: str | None = Field(default=None, max_length=2000)
|
|
cust_name: str | None = Field(default=None, max_length=500)
|
|
cust_at: date | None = None
|
|
cust_clear: bool | None = None
|
|
|
|
|
|
class SignBodyOnly(It):
|
|
sign: SignBody | None = None
|
|
|
|
|
|
class ReopenBody(It):
|
|
reason: str | None = Field(default=None, max_length=2000)
|
|
|
|
|
|
# ---------------------------------------------------------------- helpers
|
|
|
|
|
|
def _pp(d: Db, p: Principal) -> policy.Principal:
|
|
return packets.policy_principal_for(d.conn, p)
|
|
|
|
|
|
def _answer(d: Db, p: Principal, itp_id: int) -> dict[str, Any]:
|
|
row = dom.load_row(d.conn, itp_id)
|
|
if row is None:
|
|
raise ApiError(404, "not_found", dom.MSG_NO_ITP)
|
|
return dom.read(d.conn, row, _pp(d, p))
|
|
|
|
|
|
def _item(w: dom.ItpWrite, key: str) -> dict[str, Any]:
|
|
it = next((i for i in dom.items(w.itp["template"] or {}) if str(i["n"]) == key), None)
|
|
if it is None:
|
|
raise ApiError(404, "not_found", MSG_NO_LINE)
|
|
return it
|
|
|
|
|
|
def _line(w: dom.ItpWrite, key: str) -> dict[str, Any]:
|
|
_item(w, key)
|
|
row = packets.one(
|
|
w.conn, "SELECT * FROM itp_lines WHERE itp_id = %s AND key = %s FOR UPDATE", (w.id, key)
|
|
) or packets.one(
|
|
w.conn,
|
|
"INSERT INTO itp_lines (itp_id, key) VALUES (%s, %s) RETURNING *",
|
|
(w.id, key),
|
|
)
|
|
assert row is not None
|
|
return row
|
|
|
|
|
|
def _update_line(w: dom.ItpWrite, line_id: int, values: dict[str, Any]) -> None:
|
|
sets = ", ".join(f"{k} = %s" for k in values)
|
|
w.conn.execute(
|
|
f"UPDATE itp_lines SET {sets}, row_version = row_version + 1 WHERE id = %s", # noqa: S608 (fixed field names)
|
|
(*values.values(), line_id),
|
|
)
|
|
|
|
|
|
def _block(w: dom.ItpWrite, key: str) -> tuple[dict[str, Any], dict[str, Any]]:
|
|
b = next((x for x in dom.blocks(w.itp["template"] or {}) if x["key"] == key), None)
|
|
if b is None:
|
|
raise ApiError(404, "not_found", MSG_NO_BLOCK)
|
|
row = packets.one(
|
|
w.conn, "SELECT * FROM itp_blocks WHERE itp_id = %s AND block_key = %s FOR UPDATE", (w.id, key)
|
|
) or packets.one(
|
|
w.conn,
|
|
"INSERT INTO itp_blocks (itp_id, block_key) VALUES (%s, %s) RETURNING *",
|
|
(w.id, key),
|
|
)
|
|
assert row is not None
|
|
return b, row
|
|
|
|
|
|
def _update_block(w: dom.ItpWrite, block_id: int, values: dict[str, Any]) -> None:
|
|
sets = ", ".join(f"{k} = %s" for k in values)
|
|
w.conn.execute(
|
|
f"UPDATE itp_blocks SET {sets}, row_version = row_version + 1 WHERE id = %s", # noqa: S608 (fixed field names)
|
|
(*values.values(), block_id),
|
|
)
|
|
|
|
|
|
def _lines(w: dom.ItpWrite) -> dict[str, dict[str, Any]]:
|
|
return dom.load_lines(w.conn, w.id)
|
|
|
|
|
|
def _is_admin(p: Principal) -> bool:
|
|
return bool(p.is_admin)
|
|
|
|
|
|
def _same_user(a: Any, b: Any) -> bool:
|
|
return a is not None and b is not None and str(a) == str(b)
|
|
|
|
|
|
# ---------------------------------------------------------------- read and start
|
|
|
|
|
|
@router.get("/api/units/{unit_id}/itp")
|
|
def get_itp(
|
|
unit_id: int,
|
|
p: Principal = Depends(current_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
) -> dict[str, Any]:
|
|
"""The unit's ITP, or {"itp": null, "unit": {id, number}} when it has none. 404 for an unknown unit."""
|
|
conn = d.conn
|
|
unit = packets.one(conn, "SELECT id, number FROM v_units WHERE id = %s AND removed_at IS NULL", (unit_id,))
|
|
if unit is None:
|
|
raise ApiError(404, "not_found", "No such unit.")
|
|
row = dom.row_for_unit(conn, unit_id)
|
|
if row is None:
|
|
return {"itp": None, "unit": {"id": unit["id"], "number": unit["number"]}}
|
|
return dom.read(conn, row, _pp(d, p))
|
|
|
|
|
|
@router.post("/api/units/{unit_id}/itp", dependencies=WRITE)
|
|
def start_itp(
|
|
unit_id: int,
|
|
request: Request,
|
|
body: It | None = None,
|
|
p: Principal = Depends(current_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
) -> dict[str, Any]:
|
|
"""Start the unit's ITP from the template: the template is frozen, the header filled from the unit and project."""
|
|
body = body or It()
|
|
conn = d.conn
|
|
pp = _pp(d, p)
|
|
decision, it_kept = dom.scoped_decision(p, pp, A.ITP_START, it_reason=body.it_reason)
|
|
with conn.transaction():
|
|
unit = packets.one(
|
|
conn,
|
|
"SELECT u.id, u.po, u.lot, p.number AS pno, p.name AS pname, p.customer, p.po AS project_po"
|
|
" FROM units u JOIN projects p ON p.id = u.project_id"
|
|
" WHERE u.id = %s AND u.removed_at IS NULL FOR UPDATE OF u",
|
|
(unit_id,),
|
|
)
|
|
if unit is None:
|
|
raise ApiError(404, "not_found", "No such unit.")
|
|
if dom.row_for_unit(conn, unit_id) is not None:
|
|
raise ApiError(409, "itp_exists", "This unit already has an ITP.")
|
|
number = f"{unit['pno']}{unit['lot']}"
|
|
no = f"{number}-ITP"
|
|
hdr = {
|
|
"client": unit["customer"] or "",
|
|
"pname": unit["pname"] or "",
|
|
"pno": unit["pno"] or "",
|
|
"po": unit["po"] or unit["project_po"] or "",
|
|
"unit": number,
|
|
}
|
|
template = jsonable(ITP_TEMPLATE)
|
|
itp_id = int(
|
|
conn.execute(
|
|
"INSERT INTO itps (unit_id, template, status, no, hdr, created_by) VALUES (%s, %s, 'open', %s, %s, %s)"
|
|
" RETURNING id",
|
|
(unit_id, Jsonb(template), no, Jsonb(hdr), p.user_id),
|
|
).fetchone()[0]
|
|
)
|
|
keys = [str(i["n"]) for i in dom.items(template)]
|
|
with conn.cursor() as cur:
|
|
cur.executemany("INSERT INTO itp_lines (itp_id, key) VALUES (%s, %s)", [(itp_id, k) for k in keys])
|
|
cur.executemany(
|
|
"INSERT INTO itp_blocks (itp_id, block_key) VALUES (%s, %s)",
|
|
[(itp_id, b["key"]) for b in dom.blocks(template)],
|
|
)
|
|
row = dom.load_row(conn, itp_id, lock=True)
|
|
assert row is not None
|
|
w = dom.ItpWrite(
|
|
conn=conn,
|
|
principal=p,
|
|
policy_principal=pp,
|
|
itp=row,
|
|
action=A.ITP_START,
|
|
decision=decision,
|
|
request=request,
|
|
it_reason=it_kept,
|
|
version=int(row["version"]),
|
|
)
|
|
w.log("ITP started", f"{no} · template {template.get('ver')}")
|
|
dom._finish(w)
|
|
return _answer(d, p, itp_id)
|
|
|
|
|
|
# ---------------------------------------------------------------- header
|
|
|
|
|
|
@router.patch("/api/itps/{itp_id}", dependencies=WRITE)
|
|
def patch_header(
|
|
itp_id: int,
|
|
body: HeaderBody,
|
|
request: Request,
|
|
p: Principal = Depends(current_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
) -> dict[str, Any]:
|
|
"""The plan number and header fields, with field-level compare-and-set (409 field_conflict)."""
|
|
bad = sorted(k for k in body.set if k not in HDR_FIELDS)
|
|
if bad:
|
|
raise ApiError(422, "invalid_field", f"Not an ITP header field: {', '.join(bad)}.")
|
|
values = {k: clean_text(v, HDR_MAX) for k, v in body.set.items()}
|
|
base = {k: clean_text(v, HDR_MAX) for k, v in body.base.items() if k in HDR_FIELDS}
|
|
with dom.itp_write(d, p, itp_id, A.ITP_EDIT_LINE, request=request, it_reason=body.it_reason) as w:
|
|
hdr = dict(w.itp["hdr"] or {})
|
|
current = {"no": w.itp["no"] or "", **{k: hdr.get(k) or "" for k in dom.HDR_KEYS}}
|
|
changes = apply_field_cas(current, values, base)
|
|
if "no" in changes and not changes["no"]:
|
|
raise ApiError(422, "invalid_field", "The plan number cannot be empty.")
|
|
if not changes:
|
|
return _answer(d, p, itp_id)
|
|
for k in HDR_FIELDS:
|
|
if k not in changes:
|
|
continue
|
|
if k == "no":
|
|
w.log("ITP plan no. changed", changes[k])
|
|
else:
|
|
hdr[k] = changes[k]
|
|
w.log("ITP header changed", f"{k}: {changes[k]}")
|
|
w.conn.execute(
|
|
"UPDATE itps SET no = %s, hdr = %s WHERE id = %s", (changes.get("no", w.itp["no"]), Jsonb(hdr), itp_id)
|
|
)
|
|
return _answer(d, p, itp_id)
|
|
|
|
|
|
# ---------------------------------------------------------------- lines
|
|
|
|
|
|
@router.post("/api/itps/{itp_id}/lines/{key}/sign", dependencies=WRITE)
|
|
def sign_line(
|
|
itp_id: int,
|
|
key: str,
|
|
request: Request,
|
|
body: It | None = None,
|
|
p: Principal = Depends(current_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
) -> dict[str, Any]:
|
|
"""Approve a line (Q.C. initials). Optional If-Match is the line's row_version."""
|
|
body = body or It()
|
|
if_match = parse_if_match(request)
|
|
with dom.itp_write(d, p, itp_id, A.ITP_LINE_SIGN, request=request, it_reason=body.it_reason) as w:
|
|
line = _line(w, key)
|
|
if if_match is not None and int(line["row_version"]) != if_match:
|
|
raise ApiError(
|
|
409,
|
|
"row_conflict",
|
|
"Someone else changed this line. Reload and try again.",
|
|
extra={"current": {"row_version": line["row_version"], "result": line["result"]}},
|
|
)
|
|
if line["result"] == "approve":
|
|
raise ApiError(409, "already_signed", "This line is already approved.")
|
|
if line["result"] == "na":
|
|
raise ApiError(409, "line_na", "This line is marked N/A.")
|
|
rejected = line["result"] == "reject"
|
|
if rejected and not line["ncr"]:
|
|
raise ApiError(409, "ncr_required", "Enter the NCR # before approving this line.")
|
|
name = p.user["display_name"]
|
|
w.conn.execute(
|
|
"UPDATE itp_lines SET result = 'approve', by = %s, at = now(), initials = %s,"
|
|
" row_version = row_version + 1 WHERE id = %s",
|
|
(p.user_id, dom.initials(name), line["id"]),
|
|
)
|
|
w.log(f"ITP line {key} approved", f"after NCR {line['ncr']}" if rejected else None)
|
|
return _answer(d, p, itp_id)
|
|
|
|
|
|
@router.post("/api/itps/{itp_id}/lines/{key}/unsign", dependencies=WRITE)
|
|
def unsign_line(
|
|
itp_id: int,
|
|
key: str,
|
|
request: Request,
|
|
body: It | None = None,
|
|
p: Principal = Depends(current_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
) -> dict[str, Any]:
|
|
"""Remove an approval: the inspector who initialed it, or an admin, while the line's block is not signed."""
|
|
body = body or It()
|
|
with dom.itp_write(d, p, itp_id, A.ITP_LINE_UNSIGN, request=request, it_reason=body.it_reason) as w:
|
|
line = _line(w, key)
|
|
if line["result"] != "approve":
|
|
raise ApiError(409, "not_signed", "This line is not approved.")
|
|
if not (_same_user(line["by"], p.user_id) or _is_admin(p)):
|
|
raise ApiError(403, "forbidden", "Only the inspector who initialed this line, or an admin, can remove it.")
|
|
blk = dom.block_of_line(w.itp["template"] or {}, key)
|
|
if blk is not None:
|
|
brow = packets.one(
|
|
w.conn, "SELECT signed_at FROM itp_blocks WHERE itp_id = %s AND block_key = %s", (w.id, blk["key"])
|
|
)
|
|
if brow is not None and brow["signed_at"] is not None:
|
|
raise ApiError(409, "block_signed", "Undo the section signature first.")
|
|
who = packets.People(w.conn, [line["by"]]).name(line["by"]) or ""
|
|
was = f"was {who} {dom.date_text(line['at'], w.tz)}".strip()
|
|
w.conn.execute(
|
|
"UPDATE itp_lines SET result = NULL, by = NULL, at = NULL, initials = NULL,"
|
|
" row_version = row_version + 1 WHERE id = %s",
|
|
(line["id"],),
|
|
)
|
|
w.log(f"ITP line {key} initial removed", was)
|
|
return _answer(d, p, itp_id)
|
|
|
|
|
|
@router.patch("/api/itps/{itp_id}/lines/{key}", dependencies=WRITE)
|
|
def patch_line(
|
|
itp_id: int,
|
|
key: str,
|
|
body: LineBody,
|
|
request: Request,
|
|
p: Principal = Depends(current_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
) -> dict[str, Any]:
|
|
"""N/A, the customer code and witness, and the note of a line."""
|
|
given = body.model_fields_set
|
|
cc: str | None = None
|
|
if "cust_code" in given:
|
|
cc = clean_text(body.cust_code, 20).upper()
|
|
if cc and cc not in dom.CUST_CODES:
|
|
raise ApiError(422, "invalid_field", "Pick a customer code from the list.")
|
|
with dom.itp_write(d, p, itp_id, A.ITP_EDIT_LINE, request=request, it_reason=body.it_reason) as w:
|
|
line = _line(w, key)
|
|
values: dict[str, Any] = {}
|
|
parts: list[str] = []
|
|
if "na" in given and body.na is not None and body.na != (line["result"] == "na"):
|
|
if body.na:
|
|
if line["result"] == "approve":
|
|
raise ApiError(409, "already_signed", "Undo the approval first.")
|
|
if line["result"] == "reject":
|
|
raise ApiError(409, "line_rejected", "This line is rejected.")
|
|
values["result"] = "na"
|
|
parts.append("marked N/A")
|
|
else:
|
|
values["result"] = None
|
|
parts.append("applies again")
|
|
if "cust_code" in given and (cc or None) != (line["cust_code"] or None):
|
|
values["cust_code"] = cc or None
|
|
parts.append(f"customer code {cc}" if cc else "customer code as template")
|
|
if "cust_by" in given:
|
|
who = clean_opt(body.cust_by, CUST_BY_MAX)
|
|
if who != (line["cust_by"] or None):
|
|
values["cust_by"] = who
|
|
parts.append(f"customer {who}" if who else "customer cleared")
|
|
if who:
|
|
values["cust_at"] = body.cust_at or dom.today(w.tz)
|
|
values["cust_rec"] = p.user_id
|
|
values["cust_rec_at"] = w.conn.execute("SELECT now()").fetchone()[0]
|
|
else:
|
|
values.update({"cust_at": None, "cust_rec": None, "cust_rec_at": None})
|
|
elif who and "cust_at" in given and body.cust_at and body.cust_at != line["cust_at"]:
|
|
values["cust_at"] = body.cust_at
|
|
parts.append(f"customer date {dom.date_text(body.cust_at)}")
|
|
elif "cust_at" in given and line["cust_by"] and body.cust_at and body.cust_at != line["cust_at"]:
|
|
values["cust_at"] = body.cust_at
|
|
parts.append(f"customer date {dom.date_text(body.cust_at)}")
|
|
if "note" in given:
|
|
note = clean_opt(body.note, NOTE_MAX)
|
|
if note != (line["note"] or None):
|
|
values["note"] = note
|
|
parts.append("note")
|
|
if not values:
|
|
return _answer(d, p, itp_id)
|
|
_update_line(w, line["id"], values)
|
|
w.log(f"ITP line {key} updated", ", ".join(parts))
|
|
return _answer(d, p, itp_id)
|
|
|
|
|
|
@router.post("/api/itps/{itp_id}/sections/{n}/na", dependencies=WRITE)
|
|
def section_na(
|
|
itp_id: int,
|
|
n: str,
|
|
body: SectionNaBody,
|
|
request: Request,
|
|
p: Principal = Depends(current_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
) -> dict[str, Any]:
|
|
"""Mark every line of a section N/A, or make it apply again. Refused once a line of it is approved."""
|
|
with dom.itp_write(d, p, itp_id, A.ITP_EDIT_LINE, request=request, it_reason=body.it_reason) as w:
|
|
sec = next((s for s in (w.itp["template"] or {}).get("sections") or [] if str(s.get("n")) == n), None)
|
|
if sec is None:
|
|
raise ApiError(404, "not_found", MSG_NO_SECTION)
|
|
rows = [_line(w, str(i["n"])) for i in sec.get("items") or []]
|
|
if any(r["result"] == "approve" for r in rows):
|
|
raise ApiError(409, "already_signed", "A line of this section is approved. Undo it first.")
|
|
if body.na and any(r["result"] == "reject" for r in rows):
|
|
raise ApiError(409, "line_rejected", "A line of this section is rejected.")
|
|
want = "na" if body.na else None
|
|
todo = [r for r in rows if (r["result"] == "na") != body.na]
|
|
if not todo:
|
|
return _answer(d, p, itp_id)
|
|
for r in todo:
|
|
_update_line(w, r["id"], {"result": want})
|
|
w.log(f"ITP section {n} {'marked N/A' if body.na else 'applies again'}", str(sec.get("t") or ""))
|
|
return _answer(d, p, itp_id)
|
|
|
|
|
|
@router.post("/api/itps/{itp_id}/lines/{key}/reject", dependencies=WRITE)
|
|
def reject_line(
|
|
itp_id: int,
|
|
key: str,
|
|
body: RejectBody,
|
|
request: Request,
|
|
p: Principal = Depends(current_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
) -> dict[str, Any]:
|
|
"""Reject a line, with its NCR number now or later. The line cannot be approved until the NCR is entered."""
|
|
ncr = ncrs.validate_ncr(body.ncr) if clean_text(body.ncr, 100) else None
|
|
note = clean_opt(body.note, NOTE_MAX)
|
|
with dom.itp_write(d, p, itp_id, A.ITP_EDIT_LINE, request=request, it_reason=body.it_reason) as w:
|
|
line = _line(w, key)
|
|
if line["result"] == "approve":
|
|
raise ApiError(409, "already_signed", "This line is approved. Undo the approval first.")
|
|
if line["result"] == "na":
|
|
raise ApiError(409, "line_na", "This line is marked N/A.")
|
|
if line["result"] == "reject":
|
|
raise ApiError(409, "line_rejected", "This line is already rejected.")
|
|
_update_line(w, line["id"], {"result": "reject", "ncr": ncr})
|
|
w.conn.execute(
|
|
"INSERT INTO itp_rejects (itp_id, line_key, ncr, note, by, kind) VALUES (%s, %s, %s, %s, %s, 'reject')",
|
|
(w.id, key, ncr, note, p.user_id),
|
|
)
|
|
head = f"NCR {ncr}" if ncr else "NCR # pending"
|
|
w.log(f"ITP line {key} rejected", f"{head} · {note}" if note else head)
|
|
return _answer(d, p, itp_id)
|
|
|
|
|
|
@router.post("/api/itps/{itp_id}/lines/{key}/ncr", dependencies=WRITE)
|
|
def enter_ncr(
|
|
itp_id: int,
|
|
key: str,
|
|
body: NcrBody,
|
|
request: Request,
|
|
p: Principal = Depends(current_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
) -> dict[str, Any]:
|
|
"""The NCR number of an open reject that has none yet."""
|
|
ncr = ncrs.validate_ncr(body.ncr)
|
|
with dom.itp_write(d, p, itp_id, A.ITP_EDIT_LINE, request=request, it_reason=body.it_reason) as w:
|
|
line = _line(w, key)
|
|
if line["result"] != "reject" or line["ncr"]:
|
|
raise ApiError(409, "no_open_reject", "This line has no open reject waiting for an NCR #.")
|
|
_update_line(w, line["id"], {"ncr": ncr})
|
|
w.conn.execute(
|
|
"INSERT INTO itp_rejects (itp_id, line_key, ncr, by, kind) VALUES (%s, %s, %s, %s, 'ncr')",
|
|
(w.id, key, ncr, p.user_id),
|
|
)
|
|
w.log(f"ITP line {key} NCR number entered", f"NCR {ncr}")
|
|
return _answer(d, p, itp_id)
|
|
|
|
|
|
# ---------------------------------------------------------------- blocks
|
|
|
|
|
|
@router.patch("/api/itps/{itp_id}/blocks/{key}", dependencies=WRITE)
|
|
def patch_block(
|
|
itp_id: int,
|
|
key: str,
|
|
body: BlockBody,
|
|
request: Request,
|
|
p: Principal = Depends(current_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
) -> dict[str, Any]:
|
|
"""QN #, remarks and the customer complete record of a block. No signature."""
|
|
given = body.model_fields_set
|
|
with dom.itp_write(d, p, itp_id, A.ITP_EDIT_LINE, request=request, it_reason=body.it_reason) as w:
|
|
_b, row = _block(w, key)
|
|
values: dict[str, Any] = {}
|
|
if "qn" in given:
|
|
qn = clean_opt(body.qn, QN_MAX)
|
|
if qn != (row["qn"] or None):
|
|
values["qn"] = qn
|
|
w.log(f"ITP block {key} QN # changed", qn or "")
|
|
if "remarks" in given:
|
|
rem = clean_opt(body.remarks, REMARKS_MAX)
|
|
if rem != (row["remarks"] or None):
|
|
values["remarks"] = rem
|
|
w.log(f"ITP block {key} remarks changed", rem or "")
|
|
if body.cust_clear:
|
|
if row["cust_name"] or row["cust_at"]:
|
|
values.update({"cust_name": None, "cust_at": None, "cust_by": None, "cust_rec_at": None})
|
|
w.log(f"ITP block {key} customer complete cleared", row["cust_name"] or "")
|
|
elif "cust_name" in given:
|
|
name = clean_opt(body.cust_name, CUST_NAME_MAX)
|
|
if name is None:
|
|
raise ApiError(422, "invalid_field", "Enter the customer rep's name.")
|
|
at = body.cust_at or dom.today(w.tz)
|
|
if name != row["cust_name"] or at != row["cust_at"]:
|
|
values.update(
|
|
{
|
|
"cust_name": name,
|
|
"cust_at": at,
|
|
"cust_by": p.user_id,
|
|
"cust_rec_at": w.conn.execute("SELECT now()").fetchone()[0],
|
|
}
|
|
)
|
|
w.log(f"ITP block {key} customer complete recorded", name)
|
|
if not values:
|
|
return _answer(d, p, itp_id)
|
|
_update_block(w, row["id"], values)
|
|
return _answer(d, p, itp_id)
|
|
|
|
|
|
@router.post("/api/itps/{itp_id}/blocks/{key}/sign", dependencies=WRITE)
|
|
def sign_block(
|
|
itp_id: int,
|
|
key: str,
|
|
body: SignBodyOnly,
|
|
request: Request,
|
|
p: Principal = Depends(current_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
) -> dict[str, Any]:
|
|
"""Sign a block of sections. Every line in it must be approved or N/A."""
|
|
conn = d.conn
|
|
row0 = dom.load_row(conn, itp_id)
|
|
if row0 is None:
|
|
raise ApiError(404, "not_found", dom.MSG_NO_ITP)
|
|
# Who first, then the ready and unsigned checks before the signature, so a refused step costs no PIN attempt.
|
|
# Both run again under the lock below.
|
|
dom.scoped_decision(p, _pp(d, p), A.ITP_BLOCK_SIGN, it_reason=body.it_reason)
|
|
_block_gate(conn, row0, key)
|
|
with dom.itp_write(
|
|
d,
|
|
p,
|
|
itp_id,
|
|
A.ITP_BLOCK_SIGN,
|
|
request=request,
|
|
it_reason=body.it_reason,
|
|
sign=body.sign,
|
|
sign_extra={"block_key": key},
|
|
) as w:
|
|
b, row = _block(w, key)
|
|
_block_gate(conn, w.itp, key)
|
|
_update_block(
|
|
w,
|
|
row["id"],
|
|
{
|
|
"signed_at": conn.execute("SELECT now()").fetchone()[0],
|
|
"signed_by": p.user_id,
|
|
"signature_id": w.signature.id if w.signature else None,
|
|
},
|
|
)
|
|
w.log(f"ITP block {key} signed", dom.block_scope(b))
|
|
return _answer(d, p, itp_id)
|
|
|
|
|
|
def _block_gate(conn: Any, itp_row: dict[str, Any], key: str) -> None:
|
|
if str(itp_row["status"]) == "complete":
|
|
raise ApiError(409, "itp_complete", dom.MSG_COMPLETE)
|
|
b = next((x for x in dom.blocks(itp_row["template"] or {}) if x["key"] == key), None)
|
|
if b is None:
|
|
raise ApiError(404, "not_found", MSG_NO_BLOCK)
|
|
brow = packets.one(
|
|
conn, "SELECT signed_at FROM itp_blocks WHERE itp_id = %s AND block_key = %s", (itp_row["id"], key)
|
|
)
|
|
if brow is not None and brow["signed_at"] is not None:
|
|
raise ApiError(409, "already_signed", "This block is already signed.")
|
|
if not dom.block_ready(b, dom.load_lines(conn, int(itp_row["id"]))):
|
|
raise ApiError(409, "block_not_ready", "Every line must be initialed or N/A first.")
|
|
|
|
|
|
@router.delete("/api/itps/{itp_id}/blocks/{key}/sign", dependencies=WRITE)
|
|
def unsign_block(
|
|
itp_id: int,
|
|
key: str,
|
|
request: Request,
|
|
body: It | None = None,
|
|
p: Principal = Depends(current_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
) -> dict[str, Any]:
|
|
"""Remove a block signature: the signer or an admin, while the ITP is open. The block's fields stay."""
|
|
body = body or It()
|
|
with dom.itp_write(d, p, itp_id, A.ITP_BLOCK_UNSIGN, request=request, it_reason=body.it_reason) as w:
|
|
_b, row = _block(w, key)
|
|
if row["signed_at"] is None:
|
|
raise ApiError(409, "not_signed", "This block is not signed.")
|
|
if not (_same_user(row["signed_by"], p.user_id) or _is_admin(p)):
|
|
raise ApiError(403, "forbidden", "Only the inspector who signed this block, or an admin, can remove it.")
|
|
_update_block(w, row["id"], {"signed_at": None, "signed_by": None, "signature_id": None})
|
|
w.log(f"ITP block {key} signature removed")
|
|
return _answer(d, p, itp_id)
|
|
|
|
|
|
# ---------------------------------------------------------------- sign-off and reopen
|
|
|
|
|
|
@router.post("/api/itps/{itp_id}/signoff", dependencies=WRITE)
|
|
def sign_off(
|
|
itp_id: int,
|
|
body: SignBodyOnly,
|
|
request: Request,
|
|
p: Principal = Depends(current_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
) -> dict[str, Any]:
|
|
"""Sign the ITP off: every block signed. The ITP locks and the signed PDF is filed in the unit's Q.C. folder.
|
|
A PDF that cannot be built answers 500 itp_pdf_failed and nothing is signed off."""
|
|
from ..pdf import itp as itp_pdf
|
|
from ..pdf.style import PdfError
|
|
|
|
conn = d.conn
|
|
row0 = dom.load_row(conn, itp_id)
|
|
if row0 is None:
|
|
raise ApiError(404, "not_found", dom.MSG_NO_ITP)
|
|
dom.scoped_decision(p, _pp(d, p), A.ITP_SIGN_OFF, it_reason=body.it_reason)
|
|
_signoff_gate(conn, row0)
|
|
with dom.itp_write(d, p, itp_id, A.ITP_SIGN_OFF, request=request, it_reason=body.it_reason, sign=body.sign) as w:
|
|
_signoff_gate(conn, w.itp)
|
|
conn.execute(
|
|
"UPDATE itps SET status = 'complete', signoff_at = now(), signoff_by = %s, signoff_signature_id = %s"
|
|
" WHERE id = %s",
|
|
(p.user_id, w.signature.id if w.signature else None, itp_id),
|
|
)
|
|
row = dom.load_row(conn, itp_id)
|
|
assert row is not None
|
|
try:
|
|
asset_id = itp_pdf.file_signed_itp_pdf(conn, row, p.actor, settings=settings_of(request))
|
|
except PdfError as e:
|
|
log.warning("signed ITP PDF for ITP %s failed: %s", itp_id, e)
|
|
text = " ".join(str(e).split()).rstrip(".")[:300] or "an unknown error"
|
|
raise ApiError(
|
|
500, "itp_pdf_failed", f"The signed ITP PDF could not be built: {text}. The ITP was not signed off."
|
|
) from None
|
|
if asset_id is not None:
|
|
_file_auto_doc(w, row, asset_id)
|
|
w.log("ITP signed off", "PDF filed")
|
|
else:
|
|
w.log("ITP signed off")
|
|
return _answer(d, p, itp_id)
|
|
|
|
|
|
def _signoff_gate(conn: Any, itp_row: dict[str, Any]) -> None:
|
|
if str(itp_row["status"]) == "complete":
|
|
raise ApiError(409, "itp_complete", dom.MSG_COMPLETE)
|
|
keys = [b["key"] for b in dom.blocks(itp_row["template"] or {})]
|
|
signed = {
|
|
r[0]
|
|
for r in conn.execute(
|
|
"SELECT block_key FROM itp_blocks WHERE itp_id = %s AND signed_at IS NOT NULL", (itp_row["id"],)
|
|
)
|
|
}
|
|
if not keys or any(k not in signed for k in keys):
|
|
raise ApiError(409, "blocks_unsigned", "Every section must be signed first.")
|
|
|
|
|
|
AUTO_NOTE = "Filed automatically when the digital ITP was signed off."
|
|
|
|
|
|
def _file_auto_doc(w: dom.ItpWrite, row: dict[str, Any], asset_id: Any) -> None:
|
|
"""Store the signed PDF on the ITP and file it in the unit's Q.C. folder, superseding the earlier automatic copy."""
|
|
conn = w.conn
|
|
conn.execute("UPDATE itps SET signoff_asset_id = %s WHERE id = %s", (asset_id, row["id"]))
|
|
new_id = conn.execute(
|
|
"INSERT INTO qc_docs (unit_id, type, title, rev, note, asset_id, auto, added_by)"
|
|
" VALUES (%s, 'ITP', %s, NULL, %s, %s, true, %s) RETURNING id",
|
|
(row["unit_id"], f"{row['no']} — signed off", AUTO_NOTE, asset_id, w.user_id),
|
|
).fetchone()[0]
|
|
conn.execute(
|
|
"UPDATE qc_docs SET superseded_by = %s, superseded_at = now(), superseded_by_user = %s"
|
|
" WHERE unit_id = %s AND type = 'ITP' AND auto AND superseded_by IS NULL AND id <> %s",
|
|
(new_id, w.user_id, row["unit_id"], new_id),
|
|
)
|
|
|
|
|
|
@router.post("/api/itps/{itp_id}/reopen", dependencies=WRITE)
|
|
def reopen(
|
|
itp_id: int,
|
|
body: ReopenBody,
|
|
request: Request,
|
|
p: Principal = Depends(current_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
) -> dict[str, Any]:
|
|
"""Reopen a signed-off ITP: a Q.C. supervisor, or an admin (the reason serves as the IT reason when none is given).
|
|
The blocks and lines keep their signatures."""
|
|
reason = clean_opt(body.reason, LIMITS["reason"])
|
|
it_reason = body.it_reason or (body.reason if p.is_admin else None)
|
|
extra = {"reason": reason} if reason else {}
|
|
with dom.itp_write(
|
|
d, p, itp_id, A.ITP_REOPEN, request=request, extra=extra, it_reason=it_reason, allow_complete=True
|
|
) as w:
|
|
if str(w.itp["status"]) != "complete":
|
|
raise ApiError(409, "itp_open", "The ITP is open already.")
|
|
w.conn.execute(
|
|
"UPDATE itps SET status = 'open', reopen_at = now(), reopen_by = %s, reopen_reason = %s WHERE id = %s",
|
|
(p.user_id, reason, itp_id),
|
|
)
|
|
w.log("ITP reopened", reason)
|
|
return _answer(d, p, itp_id)
|
|
|
|
|
|
# ---------------------------------------------------------------- the ITP PDF and its printed copy
|
|
|
|
|
|
def _itp_row(d: Db, itp_id: int) -> dict[str, Any]:
|
|
row = dom.load_row(d.conn, itp_id)
|
|
if row is None:
|
|
raise ApiError(404, "not_found", dom.MSG_NO_ITP)
|
|
return row
|
|
|
|
|
|
@router.get("/api/itps/{itp_id}/pdf")
|
|
def itp_pdf_file(
|
|
itp_id: int,
|
|
request: Request,
|
|
download: int = Query(0, ge=0, le=1),
|
|
p: Principal = Depends(current_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
) -> Response:
|
|
"""The ITP PDF built now, open or complete; the stamp says which. Logged as an export."""
|
|
from ..pdf import itp as itp_pdf
|
|
from ..pdf.style import PdfError
|
|
|
|
conn = d.conn
|
|
row = _itp_row(d, itp_id)
|
|
try:
|
|
built = itp_pdf.build(conn, settings_of(request), row, who=prints_dom.who_of(p), now=now_utc())
|
|
except PdfError as e:
|
|
raise pdf_failed(e) from None
|
|
stamp, sig = stamp_of(built)
|
|
prints_dom.record(
|
|
conn,
|
|
p,
|
|
action="export",
|
|
kind="itp-pdf",
|
|
what=f"ITP {row['no'] or ''}".strip(),
|
|
pages=built.pages,
|
|
unit_id=row["unit_id"],
|
|
itp_id=itp_id,
|
|
stamp=stamp,
|
|
stamp_sig=sig,
|
|
)
|
|
return file_response(built.bytes, prints_dom.itp_name(row["no"]), download=bool(download))
|
|
|
|
|
|
@router.get("/api/itps/{itp_id}/print")
|
|
def itp_print(
|
|
itp_id: int,
|
|
request: Request,
|
|
download: int = Query(0, ge=0, le=1),
|
|
p: Principal = Depends(current_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
) -> Response:
|
|
"""The ITP PDF stamped PRINTED COPY. Logged as a print."""
|
|
from ..pdf import prints as print_pdf
|
|
from ..pdf.style import PdfError
|
|
|
|
conn = d.conn
|
|
row = _itp_row(d, itp_id)
|
|
try:
|
|
built = print_pdf.print_itp(conn, settings_of(request), row, who=prints_dom.who_of(p), now=now_utc())
|
|
except PdfError as e:
|
|
raise pdf_failed(e) from None
|
|
stamp, sig = stamp_of(built)
|
|
what = str(getattr(built, "what", None) or f"ITP {row['no'] or ''}".strip())
|
|
prints_dom.record(
|
|
conn,
|
|
p,
|
|
action="print",
|
|
kind="itp",
|
|
what=what,
|
|
pages=built.pages,
|
|
unit_id=row["unit_id"],
|
|
itp_id=itp_id,
|
|
stamp=stamp,
|
|
stamp_sig=sig,
|
|
)
|
|
return file_response(built.bytes, prints_dom.itp_name(row["no"]), download=bool(download))
|