PacketStation/server/packetstation/pdf/itp.py

"""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))