PacketStation/server/packetstation/pdf/parts.py

"""What the builders share: the packet read model, people with their employee numbers, departments, the packet
log, and single pages made from a raster or an image.

Everything here only reads. The read model comes from domain/packets.py `build` (outside seal mode, no principal).
"""

from __future__ import annotations

import io
import uuid
from collections.abc import Iterable, Mapping
from pathlib import Path
from typing import Any

from pypdf import PdfReader
from reportlab.lib.utils import ImageReader
from reportlab.pdfgen.canvas import Canvas

from ..domain import packets
from .style import PAGE_H, PAGE_W, PdfError

# The status texts of the frontend's lib/packets.ts STATUS_LABEL.
STATUS_LABEL = {
    "draft": "Ready to release",
    "warehouse": "With warehouse",
    "ready": "Ready to assign",
    "assigned": "Sent to station",
    "in_progress": "Working",
    "review": "Supervisor review",
    "qc": "Q.C. inspection",
    "complete": "Routed",
    "filed": "Filed in data book",
}

# The old ROLE_N texts, extended to the role kinds of auth/groups.py.
ROLE_TEXT = {
    "admin": "Admin",
    "clerical": "Clerical",
    "clerical_supervisor": "Clerical supervisor",
    "warehouse": "Warehouse",
    "warehouse_supervisor": "Warehouse supervisor",
    "qc": "Q.C.",
    "qc_supervisor": "Q.C. supervisor",
    "pm": "Project Mgmt",
    "shop_worker": "Worker",
    "shop_lead": "Shop lead",
    "shop_supervisor": "Supervisor",
    "supervisor": "Supervisor",
    "stall": "Weld Stall",
}


def load_packet(conn: Any, packet_id: int) -> dict[str, Any]:
    """The packet read model (outside seal mode, no principal). PdfError when there is no such packet."""
    row = packets.load_row(conn, int(packet_id))
    if row is None:
        raise PdfError("No such packet.")
    return packets.build(conn, row, None)


def _uuid(v: Any) -> uuid.UUID | None:
    if v is None:
        return None
    if isinstance(v, Mapping):
        v = v.get("id")
    try:
        return uuid.UUID(str(v))
    except ValueError:
        return None


def users_of(conn: Any, ids: Iterable[Any]) -> dict[str, dict[str, Any]]:
    """{user id text: {display_name, emp_no, stamp}} for the given ids (user refs or ids)."""
    want = sorted({u for u in (_uuid(i) for i in ids) if u is not None}, key=str)
    if not want:
        return {}
    out: dict[str, dict[str, Any]] = {}
    for r in conn.execute("SELECT id, display_name, emp_no, welder_stamp FROM users WHERE id = ANY(%s)", (want,)):
        out[str(r[0])] = {"display_name": r[1], "emp_no": r[2], "stamp": r[3]}
    return out


def name_emp(users: Mapping[str, Mapping[str, Any]], ref: Any, fallback: str = "") -> str:
    """'Name (emp)' for a user ref or id; the fallback name when the user is unknown."""
    u = _uuid(ref)
    info = users.get(str(u)) if u else None
    name = (info or {}).get("display_name") or (ref.get("display_name") if isinstance(ref, Mapping) else None)
    name = name or fallback
    if not name:
        return ""
    emp = (info or {}).get("emp_no") or "?"
    return f"{name} ({emp})"


def name_of(users: Mapping[str, Mapping[str, Any]], ref: Any, fallback: str = "") -> str:
    u = _uuid(ref)
    info = users.get(str(u)) if u else None
    name = (info or {}).get("display_name") or (ref.get("display_name") if isinstance(ref, Mapping) else None)
    return str(name or fallback)


def departments(conn: Any) -> dict[str, dict[str, Any]]:
    """{code: {name, label, weld}}. The label is "03 Weld Parts" for a shop department, the name otherwise."""
    out: dict[str, dict[str, Any]] = {}
    for code, name, kind, weld in conn.execute("SELECT code, name, kind::text, weld FROM departments"):
        out[code] = {"name": name, "label": packets.dept_label(code, name, kind), "weld": bool(weld)}
    return out


def role_text(role: Any) -> str:
    """The role snapshot of a log row in words."""
    parts = []
    for r in str(role or "").split(","):
        r = r.strip()
        if not r:
            continue
        bits = r.split(":")
        if bits[0] == "shop" and len(bits) == 3:  # shop:03:supervisor
            parts.append(f"{bits[1]} {ROLE_TEXT.get('shop_' + bits[2], bits[2].replace('_', ' '))}")
        else:
            parts.append(ROLE_TEXT.get(bits[0], bits[0].replace("_", " ")))
    return ", ".join(dict.fromkeys(parts))


def log_rows(conn: Any, packet_id: int) -> list[dict[str, Any]]:
    """Every packet_log row oldest first, with the signer's name and employee number on signed rows (the log
    endpoint's columns, no limit)."""
    return packets.rows(
        conn,
        "SELECT l.id, l.at, l.by, l.by_name, l.role, l.action, l.note, l.it_reason, l.signature_id,"
        " u.emp_no, s.display_name AS sig_name, su.emp_no AS sig_emp, su.welder_stamp AS sig_stamp"
        " FROM packet_log l LEFT JOIN users u ON u.id = l.by"
        " LEFT JOIN signatures s ON s.id = l.signature_id LEFT JOIN users su ON su.id = s.user_id"
        " WHERE l.packet_id = %s ORDER BY l.id",
        (int(packet_id),),
    )


def shorts_of(conn: Any, packet_id: int) -> list[dict[str, Any]]:
    """Every short of the packet, open and closed, with its BOM item and its receipts."""
    found = packets.rows(
        conn,
        "SELECT s.*, b.item FROM bom_shorts s JOIN bom_lines b ON b.id = s.bom_line_id"
        " WHERE s.packet_id = %s ORDER BY b.item, s.id",
        (int(packet_id),),
    )
    if not found:
        return []
    rec: dict[int, list[dict[str, Any]]] = {}
    for r in packets.rows(
        conn, "SELECT * FROM short_receipts WHERE short_id = ANY(%s) ORDER BY id", ([s["id"] for s in found],)
    ):
        rec.setdefault(r["short_id"], []).append(r)
    for s in found:
        s["receipts"] = rec.get(s["id"], [])
    return found


def project_of(p: Mapping[str, Any]) -> dict[str, Any]:
    return dict(((p.get("unit") or {}).get("project")) or {})


def tag_of(p: Mapping[str, Any]) -> tuple[str, str]:
    """(project number, lot) as text."""
    unit = p.get("unit") or {}
    return str(project_of(p).get("number") or ""), str(unit.get("lot") or "")


# ---------------------------------------------------------------- pages


def drawing_pages(p: Mapping[str, Any]) -> list[dict[str, Any]]:
    """The live drawing pages in page order."""
    return [pg for pg in p.get("pages") or [] if pg.get("type") == "drawing" and not pg.get("void")]


def void_pages(p: Mapping[str, Any]) -> list[dict[str, Any]]:
    return [pg for pg in p.get("pages") or [] if pg.get("type") == "drawing" and pg.get("void")]


def page_co(p: Mapping[str, Any], n: int) -> int | None:
    """The change order a live page came with, from the change order items (new_page)."""
    for co in p.get("change_orders") or []:
        for it in co.get("items") or []:
            if it.get("new_page") == n:
                return co.get("co_no")
    return None


def open_pdf(path: Path) -> PdfReader:
    """A PdfReader for a stored file. An encrypted file is opened with an empty password or refused."""
    reader = PdfReader(str(path))
    if reader.is_encrypted and not reader.decrypt(""):
        raise PdfError("The PDF is encrypted.")
    if len(reader.pages) < 1:
        raise PdfError("The PDF has no pages.")
    return reader


def raster_page_pdf(jpeg: bytes, w: float, h: float) -> bytes:
    """One page of the sheet's size with the JPEG filling it."""
    buf = io.BytesIO()
    c = Canvas(buf, pagesize=(w, h), pageCompression=1)
    c.drawImage(ImageReader(io.BytesIO(jpeg)), 0, 0, w, h)
    c.showPage()
    c.save()
    return buf.getvalue()


def image_page_pdf(path: Path) -> bytes:
    """A PNG or JPEG on a letter page, fit inside 540 x 700 and centred."""
    img = ImageReader(str(path))
    iw, ih = img.getSize()
    if iw <= 0 or ih <= 0:
        raise PdfError("The image is empty.")
    sc = min(540 / iw, 700 / ih)
    buf = io.BytesIO()
    c = Canvas(buf, pagesize=(PAGE_W, PAGE_H), pageCompression=1)
    c.drawImage(img, (PAGE_W - iw * sc) / 2, (PAGE_H - ih * sc) / 2, iw * sc, ih * sc, mask="auto")
    c.showPage()
    c.save()
    return buf.getvalue()


def sniff(path: Path) -> str | None:
    """pdf, png or jpeg from the first bytes, or None."""
    with path.open("rb") as f:
        head = f.read(8)
    if head.startswith(b"%PDF"):
        return "pdf"
    if head.startswith(b"\x89PNG"):
        return "png"
    if head.startswith(b"\xff\xd8"):
        return "jpeg"
    return None