"""The ITP PDF, after the old itpPdfBytes: one run of pages per ITP page group (WP, C, FA), each with the header table, the code legend, a table per section and the block signature table after a section that closes a block. `file_signed_itp_pdf` is called inside the sign-off transaction, after the ITP row is set complete. It builds the PDF, stores it as an asset and returns the asset id. Without settings it files nothing and returns None. """ from __future__ import annotations import contextlib import io import os import tempfile import uuid from collections.abc import Mapping from datetime import datetime from pathlib import Path from typing import Any from pypdf import PdfReader, PdfWriter from .. import policy from ..domain import assets from ..domain import itp as dom from . import parts, stamps from .doc import Doc from .style import ( CW, GREEN, GREY, PANEL, REMARK, Built, L, PdfError, R, WhoRef, ascii_only, clean, fmt_d, fmt_dt, site_tz, utc_now, utc_stamp, who_text, ) def read_itp(conn: Any, itp_row: Mapping[str, Any]) -> dict[str, Any]: """The full ITP read (domain/itp.py read) with a principal that may do nothing; `can` is not used here.""" pp = policy.principal_from_roles(None, []) return dom.read(conn, itp_row, pp) def _sig_text(users: Mapping[str, Mapping[str, Any]], ref: Any) -> str: return parts.name_emp(users, ref) def _ids(r: Mapping[str, Any]) -> list[Any]: ids: list[Any] = [] for ln in (r.get("lines") or {}).values(): ids.append(ln.get("by")) for b in r.get("blocks") or []: ids += [b.get("signed_by"), b.get("cust_by")] if r.get("signoff"): ids.append(r["signoff"].get("by")) return ids def pages_of(conn: Any, settings: Any, itp_row: Mapping[str, Any]) -> tuple[bytes, int, dict[str, Any]]: """The ITP pages without page numbers or stamp, their count, and the read.""" try: r = read_itp(conn, itp_row) d = _draw(r, parts.users_of(conn, _ids(r)), site_tz(settings)) return d.finish(), d.pages, r except PdfError: raise except Exception as e: raise PdfError(f"The ITP PDF could not be built: {type(e).__name__}.") from e def reject_lines(rows: list[Mapping[str, Any]], tz: str) -> list[str]: """'Rejected M/D/YYYY - NCR n' per reject. An NCR entered later (kind ncr) fills the reject before it.""" out: list[tuple[Any, str | None]] = [] for x in rows: if x.get("kind") == "ncr": if out and not out[-1][1]: out[-1] = (out[-1][0], x.get("ncr")) continue out.append((x.get("at"), x.get("ncr"))) return [f"Rejected {fmt_d(at, tz)} - NCR {ncr or 'pending'}" for at, ncr in out] def _draw(r: Mapping[str, Any], users: Mapping[str, Mapping[str, Any]], tz: str) -> Doc: template = r.get("template") or {} hdr = r.get("hdr") or {} lines = r.get("lines") or {} blocks = {str(b["key"]): b for b in r.get("blocks") or []} no = str(r.get("no") or "") d: Doc | None = None for pi, code in enumerate(dom.PAGE_ORDER): secs = [s for s in template.get("sections") or [] if s.get("page") == code] if not secs: continue header = ("Inspection & Test Plan", f"{no} • {pi + 1} of 3: {code} - {dom.PAGE_NAMES[code]}") if d is None: d = Doc(header) else: d.set_header(*header) d.add_page() y = 126.0 y = d.table( None, [ ["Inspection plan no.", no or "—", "Client", hdr.get("client") or "—"], ["Project name", hdr.get("pname") or "—", "Project no.", hdr.get("pno") or "—"], ["Purchase order", hdr.get("po") or "—", "Unit no.", hdr.get("unit") or "—"], ], y, {0: {"w": 86, "color": GREY}, 1: {"w": 172, "bold": True}, 2: {"w": 70, "color": GREY}, 3: {"bold": True}}, size=8, ) y += 4 n = d.tx("Codes: " + str(template.get("legend") or ""), L, y + 8, size=6.4, color=GREY, max_width=CW) y += 20 + (n - 1) * 7.4 for s in secs: body: list[list[Any]] = [ [ { "content": f"{s.get('n')} {s.get('t') or ''}", "colspan": 7, "bold": True, "fill": PANEL, "size": 7.6, } ] ] for it in s.get("items") or []: ln = lines.get(str(it.get("n"))) or {} result = ln.get("result") rejects = "\n".join(reject_lines(ln.get("rejects") or [], tz)) if result == "na": appr = "N/A" elif result == "approve": by = ln.get("by") or {} ini = ln.get("initials") or dom.initials(parts.name_of(users, by)) emp = (users.get(str(by.get("id"))) or {}).get("emp_no") if isinstance(by, Mapping) else None appr = f"Approved {ini} ({emp or ''})" else: appr = "" if rejects: appr = f"{appr}\n{rejects}" if appr else rejects cust = dom.cust_code(it, ln) if ln.get("cust_by"): cust += f"\n{ln['cust_by']} {fmt_d(ln.get('cust_at'), tz, empty='')}".rstrip() crit = str(it.get("crit") or "") + (f"\nNote: {ln['note']}" if ln.get("note") else "") body.append( [ str(it.get("n") or ""), it.get("t") or "", it.get("code") or "", appr, fmt_d(ln.get("at"), tz, empty="") if result == "approve" else "", cust, crit, ] ) y = d.table( ["NO", "INSPECTION ITEM", "CODE", "QC APPROVAL", "DATE", "CUSTOMER", "REMARKS / ACCEPTANCE"], body, y, { 0: {"w": 24}, 1: {"w": 108, "bold": True}, 2: {"w": 30, "align": "center"}, 3: {"w": 74}, 4: {"w": 44}, 5: {"w": 48}, 6: {"size": 6.3, "color": REMARK}, }, size=6.8, ) b = blocks.get(str(s.get("n"))) if b is not None: signed = ( f"{_sig_text(users, b.get('signed_by'))} {fmt_d(b.get('signed_at'), tz)} - signed with PIN" if b.get("signed") else "" ) cust = ( f"{b.get('cust_name') or ''} {fmt_d(b.get('cust_at'), tz)} (recorded by" f" {parts.name_of(users, b.get('cust_by'))})" if b.get("cust_at") else "" ) y = d.table( None, [ ["QN #", b.get("qn") or "", "PESCO QC signature", signed], [ "Remarks / notes", b.get("remarks") or "", "COP complete" if b.get("final") else "Customer complete", cust, ], ], y + 2, {0: {"w": 70, "color": GREY}, 1: {"w": 150}, 2: {"w": 96, "color": GREY}, 3: {"bold": True}}, size=7.4, ) y += 10 if d is None: d = Doc(("Inspection & Test Plan", no)) d.tx("The ITP template has no sections.", L, 130, size=9, color=GREY) if r.get("status") == "complete" and r.get("signoff"): so = r["signoff"] d.tx( f"ITP signed off {fmt_dt(so.get('at'), tz)} by {_sig_text(users, so.get('by'))} - PIN", L, 748, size=7.5, bold=True, color=GREEN, ) return d def finish(data: bytes, r: Mapping[str, Any], stamp: str, now: datetime, *, box: bool) -> bytes: writer = PdfWriter(clone_from=PdfReader(io.BytesIO(data))) n = len(writer.pages) def draw(c: Any, i: int, w: float, h: float) -> None: del h c.setFont("Helvetica", 7) c.setFillColorRGB(GREY[0] / 255, GREY[1] / 255, GREY[2] / 255) c.drawRightString(R, 20, f"Page {i + 1} of {n}") if not stamp: return if box: stamps.draw_box_stamp(c, w, stamp) else: stamps.draw_line_stamp(c, w, stamp, own=True) stamps.overlay_writer(writer, draw) hdr = r.get("hdr") or {} meta = stamps.metadata( f"ITP {r.get('no') or ''}", f"Inspection and test plan {r.get('no') or ''}" + (" - signed off" if r.get("status") == "complete" else ""), ", ".join(str(x) for x in (r.get("no"), hdr.get("pno"), hdr.get("unit"), hdr.get("client"), "ITP") if x), "Packet Station - controlled copy", now=now, ) return stamps.write(writer, meta) def itp_stamp(who: WhoRef | None, now: datetime, status: Any) -> str: """Controlled copy - by Name (emp) UTC - signed off (or OPEN, not signed off).""" state = "signed off" if status == "complete" else "OPEN, not signed off" return ascii_only(f"Controlled copy - by {who_text(who)} {utc_stamp(now)} - {state}") def build(conn: Any, settings: Any, itp_row: Mapping[str, Any], *, who: WhoRef, now: datetime | None = None) -> Built: """The ITP PDF as it is now (open or complete; the stamp says which).""" now = utc_now(now) data, pages, r = pages_of(conn, settings, itp_row) warnings: list[str] = [] stamp, sig = stamps.signed_stamp( settings, itp_stamp(who, now, r.get("status")), packet_id=f"itp-{itp_row['id']}", kind="itp-pdf", seal_hash_hex=None, who=who, now=now, warnings=warnings, ) data = finish(data, r, stamp, now, box=False) return Built( bytes=data, pages=pages, warnings=warnings, stamp=stamp, stamp_sig=sig, what=f"ITP {clean(r.get('no'))}" ) def who_of_actor(conn: Any, actor: Any) -> WhoRef: """The WhoRef of a session actor (sessions.Actor, or a mapping with id or user_id and display_name).""" if isinstance(actor, Mapping): uid = actor.get("user_id") or actor.get("id") name = actor.get("display_name") else: uid = getattr(actor, "user_id", None) name = getattr(actor, "display_name", None) info = parts.users_of(conn, [uid]).get(str(uid)) if uid else None return { "id": None if uid is None else str(uid), "display_name": name or (info or {}).get("display_name"), "emp_no": (info or {}).get("emp_no"), } def file_signed_itp_pdf(conn: Any, itp_row: Mapping[str, Any], actor: Any, settings: Any = None) -> uuid.UUID | None: """Build the signed ITP PDF, store it as an asset and return its id. Without settings it files nothing (None). A build failure raises PdfError; the sign-off then answers 500 and nothing is signed off.""" if settings is None: return None who = who_of_actor(conn, actor) built = build(conn, settings, itp_row, who=who) tmp_dir = Path(settings.assets_dir) / "tmp" tmp_dir.mkdir(parents=True, exist_ok=True) fd, name = tempfile.mkstemp(prefix="itp-", suffix=".pdf", dir=tmp_dir) with os.fdopen(fd, "wb") as f: f.write(built.bytes) try: row = assets.store_file( conn, settings, Path(name), mime=assets.PDF, original_name=f"{itp_row['no']}_signed.pdf", source="itp_signoff", source_ref={"itp_id": int(itp_row["id"]), "no": str(itp_row["no"])}, uploaded_by=who["id"], ) except Exception as e: raise PdfError(f"The signed ITP PDF could not be stored: {type(e).__name__}.") from e finally: with contextlib.suppress(OSError): Path(name).unlink() aid = row["id"] return aid if isinstance(aid, uuid.UUID) else uuid.UUID(str(aid))