Aislo/common_util/common_util_print_pdf.py

"""공용 PDF 엔진 — HTML → PDF(헤드리스 Edge · 없으면 Chrome · Playwright) · PDF 합치기(pypdf) ·
엑셀 인쇄 설정.

    render_html_to_pdf(html, *, page_size, landscape, margins, page_numbers) → bytes
    render_htmls_to_pdfs([{html, page_size, …}]) → [bytes]   브라우저 한 번으로 여러 벌
    merge_pdfs([bytes]) → bytes
    sheet_title · number_format · print_setup            엑셀(openpyxl) 시트 이름 · 숫자 서식 · 인쇄

쪽 크기 = page_size — "A4" 등 이름 · (가로 mm, 세로 mm) · None(HTML 의 CSS @page 가 정함).
PDF 는 자식 파이썬 프로세스(이 파일을 스크립트로 · 입력 JSON stdin · 결과 JSON 한 줄) —
서버 안 스레드에선 Playwright 가 드라이버를 못 띄움(`B08_Quantity_Milestone_Capture` 와 같은 까닭).
헤드리스 쪽은 스크립트 끔 · 바깥 요청 모두 막음(그림 = data: 주소만).
"""

from __future__ import annotations

import io
import json
import re
import subprocess
import sys
import tempfile
import threading
from pathlib import Path
from typing import Any

TIMEOUT_S = 90
CHANNELS = ("msedge", "chrome")  # 시스템 브라우저 — 번들 chromium 은 안 받아 둠
NO_BROWSER = "서버에서 브라우저(Edge · Chrome)를 못 띄움"
FOOTER = (
    '<div style="width:100%;font-size:9px;text-align:center;font-family:sans-serif">'
    '<span class="pageNumber"></span> / <span class="totalPages"></span></div>'
)

_lock = threading.Lock()  # ponytail: 서버 전체 한 줄 — 브라우저 한 번에 하나


class PrintError(Exception):
    """까닭 한 줄(예외 원문 없음)."""


def _pdf_options(job: dict[str, Any]) -> dict[str, Any]:
    """job → Playwright page.pdf 인자."""
    size = job.get("page_size")
    options: dict[str, Any] = {"print_background": True}
    if size is None:
        options["prefer_css_page_size"] = True
    elif isinstance(size, str):
        options.update(format=size, landscape=bool(job.get("landscape")))
    else:
        width, height = size
        if job.get("landscape"):
            width, height = max(width, height), min(width, height)
        options.update(width=f"{width}mm", height=f"{height}mm")
    if job.get("margins"):
        options["margin"] = job["margins"]
    if job.get("page_numbers"):
        options.update(
            display_header_footer=True, header_template="<span></span>", footer_template=FOOTER
        )
    return options


def _render(jobs: list[dict[str, Any]]) -> list[bytes]:
    """자식 프로세스 안 — 브라우저 한 번 띄워 job 마다 PDF. 실패 = PrintError."""
    try:
        from playwright.sync_api import Error as PwError
        from playwright.sync_api import sync_playwright
    except ImportError as error:
        raise PrintError("서버에 Playwright 가 없습니다(pip install playwright)") from error
    try:
        with sync_playwright() as p:
            browser = None
            for channel in CHANNELS:
                try:
                    browser = p.chromium.launch(channel=channel, headless=True)
                    break
                except PwError:
                    continue
            if browser is None:
                raise PrintError(NO_BROWSER)
            try:
                context = browser.new_context(java_script_enabled=False)
                context.route("**/*", lambda route: route.abort())  # 바깥 요청 없음(data: 만)
                page = context.new_page()
                out = []
                for job in jobs:
                    page.set_content(job["html"], wait_until="load")
                    if job.get("prepare_js"):
                        page.evaluate(job["prepare_js"], job.get("prepare_arg"))
                    out.append(page.pdf(**_pdf_options(job)))
                return out
            finally:
                browser.close()
    except PrintError:
        raise
    except Exception as error:  # 드라이버 못 뜸 등 — 예외 글은 내지 않음
        raise PrintError(NO_BROWSER) from error


def render_htmls_to_pdfs(jobs: list[dict[str, Any]], timeout_s: float = TIMEOUT_S) -> list[bytes]:
    """HTML 여러 벌 → PDF 여러 벌(차례 그대로 · 자식 프로세스 한 번).

    job = {html, page_size="A4", landscape=False, margins=None, page_numbers=False,
           prepare_js=None, prepare_arg=None} — prepare_js = PDF 직전 page.evaluate(줄임 등).
    """
    jobs = [{"page_size": "A4", **job} for job in jobs]
    with _lock, tempfile.TemporaryDirectory() as tmp:
        outs = [str(Path(tmp) / f"{i}.pdf") for i in range(len(jobs))]
        try:
            done = subprocess.run(
                [sys.executable, __file__],
                input=json.dumps({"jobs": jobs, "outs": outs}).encode(),
                capture_output=True,
                timeout=timeout_s,
            )
        except subprocess.TimeoutExpired as error:
            raise PrintError(f"PDF 만들기가 {timeout_s:.0f}초 안에 안 끝남") from error
        except OSError as error:
            raise PrintError(NO_BROWSER) from error
        try:
            result = json.loads(done.stdout.decode().strip().splitlines()[-1])
        except (ValueError, IndexError):  # 자식이 죽음 — stderr(예외 글)는 내지 않음
            result = {"reason": NO_BROWSER}
        if result.get("reason"):
            raise PrintError(result["reason"])
        return [Path(out).read_bytes() for out in outs]


def render_html_to_pdf(
    html: str,
    *,
    page_size: str | tuple[float, float] | None = "A4",
    landscape: bool = False,
    margins: dict[str, str] | None = None,
    page_numbers: bool = False,
    prepare_js: str | None = None,
    prepare_arg: Any = None,
    timeout_s: float = TIMEOUT_S,
) -> bytes:
    """HTML 한 벌 → PDF 바이트. margins = {"top": "10mm", …}(None = CSS · 0)."""
    job = {
        "html": html,
        "page_size": page_size,
        "landscape": landscape,
        "margins": margins,
        "page_numbers": page_numbers,
        "prepare_js": prepare_js,
        "prepare_arg": prepare_arg,
    }
    return render_htmls_to_pdfs([job], timeout_s)[0]


def merge_pdfs(parts: list[bytes]) -> bytes:
    """PDF 여러 벌 → 한 벌(쪽 차례 그대로 · 같은 글꼴은 한 번만)."""
    from pypdf import PdfWriter

    writer = PdfWriter()
    for part in parts:
        writer.append(io.BytesIO(part))
    writer.compress_identical_objects(remove_duplicates=True, remove_unreferenced=True)
    out = io.BytesIO()
    writer.write(out)
    return out.getvalue()


# ── 엑셀 ──────────────────────────────────────────────────────────────────

BAD_TITLE = re.compile(r"[\[\]:*?/\\]")


def sheet_title(name: str, used: set[str]) -> str:
    """엑셀 시트 이름 — 금지 글자 = 「_」 · 31자 · 같은 이름 = 뒤에 (2)…"""
    base = BAD_TITLE.sub("_", name).strip().strip("'") or "시트"
    title, n = base[:31], 1
    while title.lower() in used:
        n += 1
        tail = f" ({n})"
        title = base[: 31 - len(tail)] + tail
    used.add(title.lower())
    return title


def number_format(decimals: int | None) -> str:
    return "#,##0" + ("." + "0" * decimals if decimals else "")


def print_setup(ws: Any, orientation: str, head_rows: int) -> None:
    """인쇄 설정(302-5) — PDF 와 같은 방향 · A4 · 폭 1쪽 맞춤 · 머리 줄 쪽마다 · 여백 ≈ PDF."""
    from openpyxl.worksheet.page import PageMargins

    ws.page_setup.orientation = orientation
    ws.page_setup.paperSize = ws.PAPERSIZE_A4
    ws.page_setup.fitToWidth = 1
    ws.page_setup.fitToHeight = 0
    ws.sheet_properties.pageSetUpPr.fitToPage = True
    ws.page_margins = PageMargins(left=0.4, right=0.4, top=0.4, bottom=0.55, header=0.2, footer=0.2)
    if head_rows:
        ws.print_title_rows = f"1:{head_rows}"
    ws.oddHeader.left.text = "&A"  # 시트 이름
    ws.oddFooter.center.text = "&P / &N"  # 쪽 번호


if __name__ == "__main__":  # 자식 프로세스 몸통
    job = json.loads(sys.stdin.buffer.read())
    try:
        for out, data in zip(job["outs"], _render(job["jobs"]), strict=True):
            Path(out).write_bytes(data)
        print(json.dumps({"reason": None}))
    except PrintError as error:
        print(json.dumps({"reason": str(error)}, ensure_ascii=False))
    sys.exit(0)