PacketStation/tests/server/test_pdf_databook.py

"""Phase 7 data book builder, called directly on a packet built through the API: two drawing sheets and a void
copy of sheet 1, a heat table, two welds, an MTR PDF and an MTR PNG, a short with a receipt, a packet document, a
Q.C. folder document, the closed stage and the Q.C. accept.

The index (seven sections in order, the expected rows), the page count, the cover, the index page, the void
sheet, the traceability page, the warning rows, the metadata, the stamp on every page and its signature, and the
stamp per mode. Nothing here reads VISUAL or M-Files."""

from __future__ import annotations

import io
import json
import os
import re
from collections.abc import Iterator
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

import pytest
from closehelpers import (
    HEATS,
    RECEIPT_HEAT,
    builder_conn,
    delete_asset_file,
    flat,
    page_texts,
    pdf_reader,
    rich_complete,
    who,
)
from fakedir import FakeDirectory
from pshelpers import TEST_SEAL_KEY, needs_pg
from qchelpers import g
from shophelpers import Shop, open_shop, run

from packetstation.pdf import databook, stamps
from packetstation.pdf.doc import PAD_H, Doc
from packetstation.pdf.style import L, clean, fmt_clock_hhmm, fmt_clock_hms, round_half_up
from packetstation.settings import Settings

pytestmark = needs_pg

NOW = datetime(2026, 10, 17, 21, 4, tzinfo=UTC)
SEAL = {"hash": "ab12cd34ef56" + "0" * 52, "key_id": "k1234567", "sealed_at": NOW}
SECTIONS = [1, 2, 3, 4, 5, 6, 7]
SIG_RE = re.compile(r"stamp ([0-9a-f]{16})")
DUMP_DIR = os.environ.get("DATABOOK_DUMP_DIR")  # off unless set


@pytest.fixture
def shop(pg_settings: Settings, fake_dir: FakeDirectory, dev_visual: None) -> Iterator[Shop]:
    yield from open_shop(pg_settings, fake_dir)


def build(shop: Shop, pid: int, mode: str = "preview", **kw: Any) -> Any:
    with builder_conn(shop.settings) as conn:
        return databook.build(
            conn,
            shop.settings,
            pid,
            who=who(shop),
            now=NOW,
            seal=SEAL if mode == "filed" else None,
            mode=mode,
            **kw,
        )


def dump(name: str, built: Any, pkt: dict[str, Any]) -> None:
    """With DATABOOK_DUMP_DIR set to a folder, keep the book and the packet read there (for the side by side
    comparison with the old build's data book). Off by default."""
    out = DUMP_DIR
    if not out:
        return
    Path(out).mkdir(parents=True, exist_ok=True)
    (Path(out) / f"{name}.pdf").write_bytes(built.bytes)
    (Path(out) / f"{name}.json").write_text(json.dumps({"packet": pkt, "index": built.index}, default=str, indent=1))


def index(built: Any) -> list[dict[str, Any]]:
    keys = ("section", "doc", "ref", "rev", "page", "warn", "count")
    return [{k: g(r, k) for k in keys} for r in built.index]


def docs_in(rows: list[dict[str, Any]], section: int) -> list[str]:
    return [str(r["doc"]) for r in rows if r["section"] == section]


def test_data_book_preview(shop: Shop) -> None:
    pkt = rich_complete(shop)
    pid = pkt["id"]
    built = build(shop, pid)
    dump("preview", built, pkt)
    data: bytes = built.bytes
    assert data[:5] == b"%PDF-"
    reader = pdf_reader(data)
    assert len(reader.pages) == built.pages
    rows = index(built)

    # the seven sections, in order, each present
    seen = [r["section"] for r in rows]
    assert seen == sorted(seen) and sorted(set(seen)) == SECTIONS, seen

    s1 = docs_in(rows, 1)
    assert sum("approved drawing" in d for d in s1) == 2, s1
    assert any(d.startswith("SUPERSEDED - VOID by CO-1") for d in s1), s1
    drawing_rows = [r for r in rows if r["section"] == 1 and "approved drawing" in str(r["doc"])]
    assert all(r["ref"] == "D-5001" and r["rev"] == "RB" for r in drawing_rows), drawing_rows

    s2 = docs_in(rows, 2)
    assert s2[0].startswith("Material traceability record"), s2
    assert any(d.startswith("Heat map") and "heat number table" in d for d in s2), s2

    s3 = docs_in(rows, 3)
    assert any(d.startswith("Weld map") and "W1, W2" in d for d in s3), s3
    assert any(d.startswith("Weld log, welding record and checklist") for d in s3), s3

    s4 = docs_in(rows, 4)
    assert len(s4) == 1 and s4[0].startswith("Q.C. inspection") and "ACCEPTED" in s4[0], s4

    s5 = docs_in(rows, 5)
    assert any("WPS / PQR" in d and "Unit folder doc" in d for d in s5), s5
    assert any("RT report W1-W2" in d for d in s5), s5
    itp_warn = [r for r in rows if r["section"] == 5 and r["doc"] == "No signed ITP filed for this unit"]
    assert len(itp_warn) == 1 and itp_warn[0]["warn"] is True

    s6 = docs_in(rows, 6)
    assert len(s6) == 2 and all(d.startswith("MTR") for d in s6), s6
    assert "item 1" in s6[0] and "item 2" in s6[1]
    mtr_rows = [r for r in rows if r["section"] == 6]
    assert mtr_rows[0]["ref"] == HEATS[1]

    assert docs_in(rows, 7) == ["Packet record history"]

    # pages: the parts plus the front matter (cover and index); page numbers point into the book
    parts = sum(int(r["count"] or 0) for r in rows)
    front = built.pages - parts
    assert front == 2, (built.pages, parts)  # the cover and one index page
    numbered = [r for r in rows if r["page"] is not None]
    assert all(front < int(r["page"]) <= built.pages for r in numbered), numbered
    assert all(r["page"] is None and not r["count"] for r in rows if r["warn"] and "could not" in str(r["doc"]))

    texts = page_texts(data)
    assert "MANUFACTURING" in texts[0] and "DATA BOOK" in flat(texts[0])
    assert pkt["unit"]["project"]["number"] in flat(texts[0])
    index_text = flat(" ".join(texts[1:front]))
    assert "INDEX" in index_text and "TABLE OF CONTENTS" in index_text
    for r in rows:
        words = flat(clean(str(r["doc"])))[:30]
        assert words in index_text, (words, index_text[:2000])

    # the void sheet carries the void text, the traceability page the heats
    void_page = next(r for r in rows if str(r["doc"]).startswith("SUPERSEDED"))["page"]
    assert "VOID - superseded by CO-1" in flat(texts[int(void_page) - 1])
    trace_page = next(r for r in rows if str(r["doc"]).startswith("Material traceability"))["page"]
    trace = flat(" ".join(texts[int(trace_page) - 1 : int(trace_page) + 1]))
    for heat in (HEATS[1], HEATS[2], RECEIPT_HEAT):
        assert heat in trace, heat
    assert "open items" in trace.lower() and "item 3" in trace

    # phase 8 comparison with the old build's book (each line one fix)
    assert '(9" x 11")' in index_text  # a letter sheet is 8.5 in wide: Math.round gives 9
    assert re.search(r"\b\d\d:\d\d WELD TIME", flat(texts[0])), flat(texts[0])  # HH:MM as the old cover
    weld_page = next(r for r in rows if str(r["doc"]).startswith("Weld log"))["page"]
    assert re.search(r"Time recorded \d\d:\d\d:\d\d \(target 4\.5 h\)", flat(texts[int(weld_page) - 1]))
    qc_page = next(r for r in rows if r["section"] == 4)["page"]
    qc_lines = texts[int(qc_page) - 1].splitlines()
    assert "AM" not in qc_lines and "PM" not in qc_lines, qc_lines  # the date and time stay on one line
    hist_page = next(r for r in rows if r["section"] == 7)["page"]
    assert "CO-1 " in flat(texts[int(hist_page) - 1]) and "documents changed" in flat(texts[int(hist_page) - 1])

    # metadata
    meta = reader.metadata
    assert meta.title == f"{pkt['unit']['project']['number']} Lot 001 Manufacturing Data Book - Spool 7 outlet"
    assert meta.producer == "Packet Station"
    assert meta.get("/Author") == "PESCO - Process Equipment & Service Company, Inc."

    # the stamp on every page, preview mode, and its signature verifies
    sigs = set()
    for i, t in enumerate(texts):
        ft = flat(t)
        assert "Controlled copy - exported by Clara Clerk (4001)" in ft, (i, ft[-300:])
        assert "not yet filed" in ft, i
        m = SIG_RE.search(ft)
        assert m, (i, ft[-300:])
        sigs.add(m.group(1))
    assert len(sigs) == 1
    sig = sigs.pop()
    user = who(shop)["id"]
    key = TEST_SEAL_KEY.encode()
    assert stamps.verify_stamp(key, pid, "databook", "", user, NOW.isoformat(), sig)
    assert not stamps.verify_stamp(key, pid, "traveler", "", user, NOW.isoformat(), sig)


def test_data_book_filed_mode_and_warnings(shop: Shop) -> None:
    pkt = rich_complete(shop)
    pid = pkt["id"]
    filed = build(shop, pid, mode="filed")
    texts = page_texts(filed.bytes)
    assert all("seal ab12cd34ef56" in flat(t) for t in texts)
    assert not any("not yet filed" in flat(t) for t in texts)
    m = SIG_RE.search(flat(texts[0]))
    assert m and filed.stamp_sig == m.group(1)
    assert stamps.verify_stamp(TEST_SEAL_KEY.encode(), pid, "databook", SEAL["hash"], who(shop)["id"], NOW, m.group(1))
    meta = pdf_reader(filed.bytes).metadata
    assert "seal" in (meta.creator or "") and "4001" in (meta.creator or "")

    # no MTRs: the MTR warning row; a deleted packet document file: could not be read
    run(shop.settings, "UPDATE bom_lines SET mtr_asset_id = NULL WHERE packet_id = %s", (pid,))
    doc = next(d for d in pkt["docs"] if d.get("removed") is None)
    asset_id = doc.get("asset_id") or doc["url"].rsplit("/", 1)[-1]
    delete_asset_file(shop.settings, asset_id)
    built = build(shop, pid)
    rows = index(built)
    mtr = [r for r in rows if r["section"] == 6]
    assert [r["doc"] for r in mtr] == ["No MTRs attached in Packet Station — add from the warehouse files"]
    assert mtr[0]["warn"] is True and mtr[0]["page"] is None
    bad = [r for r in rows if r["section"] == 5 and str(r["doc"]).endswith("could not be read")]
    assert len(bad) == 1 and "RT report W1-W2" in bad[0]["doc"] and bad[0]["warn"] is True
    # the index still fits on one page: the cover, the index and the parts
    assert built.pages == 2 + sum(int(r["count"] or 0) for r in rows)


def test_data_book_with_an_itp_document_has_no_itp_warning(shop: Shop) -> None:
    pkt = rich_complete(shop, itp_doc=True, mtrs=False)
    rows = index(build(shop, pkt["id"]))
    s5 = [r for r in rows if r["section"] == 5]
    assert s5[0]["doc"].startswith("ITP"), s5
    assert not any(r["doc"] == "No signed ITP filed for this unit" for r in s5)


# ---------------------------------------------------------------- phase 8: against the old build's book


def test_co_summary_of_a_manual_change_without_sheets_reads_documents_changed() -> None:
    co: dict[str, Any] = {"origin": "manual", "review": "approved", "reason": "Nozzle moved", "items": []}
    assert databook.co_summary(co) == "documents changed"  # the reason has its own column, as in coSummary()
    co["items"] = [{"old_dwg": "D-5001", "old_rev": "A", "new_rev": "B"}]
    assert databook.co_summary(co) == "D-5001 Rev A -> Rev B"


def test_clock_and_sheet_size_forms_of_the_old_build() -> None:
    assert fmt_clock_hms(48_112) == "00:00:48" and fmt_clock_hms(3_725_000) == "01:02:05"
    assert fmt_clock_hhmm(48_112) == "00:00" and fmt_clock_hhmm(37_800_000) == "10:30"
    assert fmt_clock_hhmm(None) == "00:00"
    assert round_half_up(612 / 72) == 9 and round_half_up(792 / 72) == 11 and round_half_up(2448 / 72) == 34


def _rows_height(rows: list[list[Any]], cols: dict[int, dict[str, Any]]) -> float:
    d = Doc(("T", "S"))
    return d.table(None, rows, 200.0, cols) - 200.0


def test_table_cells_wrap_where_autotable_did() -> None:
    one_line = 8.3 * 1.15 + 2 * 4.2
    # 79.4 pt of text in an 86 pt column: one line in the old book (autotable's 1 pt slack, jsPDF's widths)
    assert abs(_rows_height([["10/18/2026 11:32 AM"]], {0: {"w": 86}}) - one_line) < 0.01
    # clearly too long still wraps
    assert _rows_height([["10/18/2026 11:32 AM and more"]], {0: {"w": 86}}) > one_line * 1.5


def test_table_heads_sit_left_over_centred_columns() -> None:
    from pypdf import PdfReader

    d = Doc(("T", "S"))
    d.table(["ITEM", "QTY", "TEXT"], [["1", "2", "x"]], 200.0, {0: {"w": 30, "align": "center"}, 1: {"w": 30}})
    xs: dict[str, float] = {}

    def visit(text: str, cm: Any, tm: Any, font: Any, size: Any) -> None:
        if text.strip() in ("ITEM", "QTY"):
            xs[text.strip()] = float(tm[4])

    PdfReader(io.BytesIO(d.finish())).pages[0].extract_text(visitor_text=visit)
    assert xs["ITEM"] == pytest.approx(L + PAD_H, abs=0.5), xs  # autotable column styles style the body only
    assert xs["QTY"] == pytest.approx(L + 30 + PAD_H, abs=0.5), xs