PacketStation/server/packetstation/domain/prints.py

"""Printed copies and exports: who the PDF builders stamp, the file names, and the append-only print log.

Every print and every export writes one print_log row and one audit row, in its own transaction, after the bytes
were built. Prints and exports are not packet changes: they write no packet_log row and do not bump the version.
"""

from __future__ import annotations

from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Literal, cast
from urllib.parse import quote

from .. import audit
from ..api import deps
from . import assets
from . import packets as packets_dom

if TYPE_CHECKING:
    from ..pdf.style import WhoRef

PRINTS_LIMIT = 200

# The print_log kinds of the exports (prints use the print kind itself), and the audit kind of each.
EXPORT_KINDS = {"databook-export": "databook", "traveler": "traveler", "csv": "csv", "itp-pdf": "itp"}


def who_of(p: deps.Principal) -> WhoRef:
    """The signed-in person as the builders' WhoRef: {"id", "display_name", "emp_no"}."""
    emp = p.user.get("emp_no")
    return cast(
        "WhoRef",
        {
            "id": str(p.user_id),
            "display_name": str(p.user.get("display_name") or p.user.get("sam_account") or ""),
            "emp_no": None if emp is None or str(emp).strip() == "" else str(emp),
        },
    )


def who_of_user(conn: Any, user_id: Any) -> WhoRef:
    """The same WhoRef for a user id (the ITP hook gets an actor, not a principal)."""
    r = conn.execute("SELECT display_name, emp_no FROM users WHERE id = %s", (user_id,)).fetchone()
    name = str(r[0] or "") if r else ""
    emp = r[1] if r else None
    return cast(
        "WhoRef",
        {"id": str(user_id), "display_name": name, "emp_no": None if emp is None or str(emp) == "" else str(emp)},
    )


# ---------------------------------------------------------------- file names


def job_base(read: Mapping[str, Any]) -> str:
    """{project}-Lot{lot}: the top project number and the unit's lot, as the old build named its files."""
    unit = read.get("unit") or {}
    project = (unit.get("project") or {}).get("number") or ""
    lot = unit.get("lot") or unit.get("number") or ""
    return f"{project}-Lot{lot}"


def _named(text: str, fallback: str) -> str:
    return assets.safe_name(text) or fallback


def databook_name(read: Mapping[str, Any]) -> str:
    return _named(f"{job_base(read)}_{read.get('drawing') or ''}_Manufacturing_Data_Book.pdf", "Data_Book.pdf")


def traveler_name(read: Mapping[str, Any]) -> str:
    return _named(f"{job_base(read)}_Job_Traveler.pdf", "Job_Traveler.pdf")


def print_name(read: Mapping[str, Any], what: str) -> str:
    return _named(f"{job_base(read)} {what}.pdf", "Print.pdf")


def csv_name(read: Mapping[str, Any]) -> str:
    return _named(f"{job_base(read)}_{read.get('drawing') or ''}_traceability.csv", "traceability.csv")


def itp_name(no: Any) -> str:
    return _named(f"{no or 'ITP'}.pdf", "ITP.pdf")


def disposition(filename: str, download: bool = False) -> str:
    """Content-Disposition: inline unless a download is asked for. An ASCII filename, plus filename* (RFC 5987) when
    the name has characters outside ASCII."""
    kind = "attachment" if download else "inline"
    plain = filename.encode("ascii", "replace").decode("ascii").replace("?", "_")
    plain = plain.replace('"', "").replace(";", "_").replace("\\", "_").strip() or "download"
    out = f'{kind}; filename="{plain}"'
    if plain != filename:
        out += f"; filename*=UTF-8''{quote(filename, safe='')}"
    return out


# ---------------------------------------------------------------- the log


def record(
    conn: Any,
    p: deps.Principal,
    *,
    action: Literal["print", "export", "report exported"],
    kind: str,
    what: str,
    pages: int | None,
    packet_id: int | None = None,
    unit_id: int | None = None,
    itp_id: int | None = None,
    asset_id: Any = None,
    stamp: str | None = None,
    stamp_sig: str | None = None,
    detail: Mapping[str, Any] | None = None,
) -> int:
    """One print_log row and one audit row (action `print`, `export`, or `report exported` for the PM reports, which
    belong to no packet), in one transaction. The audit detail of an export names its kind as databook, traveler,
    csv or itp."""
    audit_detail: dict[str, Any] = {
        "kind": EXPORT_KINDS.get(kind, kind) if action == "export" else kind,
        "what": what,
        "pages": pages,
        "stamp_sig": stamp_sig,
    }
    audit_detail.update({k: v for k, v in (detail or {}).items() if v is not None})
    if asset_id is not None:
        audit_detail["asset_id"] = str(asset_id)
    if itp_id is not None:
        audit_detail["itp_id"] = itp_id
    subject = "itp" if itp_id is not None and packet_id is None else "packet"
    if action == "report exported":
        subject = "report"
    with conn.transaction():
        new_id = conn.execute(
            "INSERT INTO print_log (by, by_name, session_id, packet_id, unit_id, itp_id, kind, what, pages, asset_id,"
            " stamp, stamp_sig, ip) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id",
            (
                p.user_id,
                p.user.get("display_name"),
                p.session_id,
                packet_id,
                unit_id,
                itp_id,
                kind,
                what,
                pages,
                asset_id,
                stamp,
                stamp_sig,
                p.ip,
            ),
        ).fetchone()[0]
        audit.append(
            conn,
            action=action,
            user_id=p.user_id,
            session_id=p.session_id,
            role_snapshot=p.role_snapshot,
            subject_type=subject,
            subject_id=itp_id if itp_id is not None and packet_id is None else packet_id,
            packet_id=packet_id,
            detail=audit_detail,
            ip=p.ip,
        )
    return int(new_id)


def list_prints(conn: Any, packet_id: int, limit: int = PRINTS_LIMIT) -> list[dict[str, Any]]:
    """A packet's printed copies and exports, newest first."""
    return [
        {
            "id": r["id"],
            "at": r["at"],
            "by": None if r["by"] is None else {"id": str(r["by"]), "display_name": r["by_name"]},
            "kind": r["kind"],
            "what": r["what"],
            "pages": r["pages"],
        }
        for r in packets_dom.rows(
            conn,
            "SELECT id, at, by, by_name, kind, what, pages FROM print_log WHERE packet_id = %s"
            " ORDER BY id DESC LIMIT %s",
            (packet_id, limit),
        )
    ]