Add HTTP print API for packing list and sign-off receipt

- core/printing.py: SumatraPDF silent print backend (dry_run, timeout, error capture)
- core/print_document.py: orchestrate generate() + print, per-report paper size, printer fallback
- core/print_api.py: FastAPI POST /api/print + GET /api/health, HTTPBearer auth, English Swagger at /api/docs
- serve.py: uvicorn entrypoint
- config: printers/print/api sections in settings.yaml + core/settings.py
- requirements.txt: add fastapi, uvicorn
- .gitignore: ignore logs/

.gitignore

@@ -13,7 +13,7 @@ config/settings.yaml
 
 # generated output
 out/
-
+logs/
 # verification artifacts
 *.preview.png
 _smoke_*

core/print_api.py

@@ -0,0 +1,232 @@
+"""
+FastAPI print service for WareShipManifest.
+
+All endpoints are mounted under the ``/api`` prefix:
+    POST /api/print   generate a shipping document PDF and physically print it
+    GET  /api/health  liveness probe
+
+Interactive Swagger UI is served at ``/api/docs`` and the OpenAPI schema at
+``/api/openapi.json``. All documentation is in English.
+"""
+from __future__ import annotations
+
+import json
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Literal, Optional
+
+from fastapi import APIRouter, Depends, FastAPI, HTTPException
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from pydantic import BaseModel, ConfigDict, Field
+
+from core.settings import PROJECT_ROOT, settings
+from core import print_document as pd
+
+app = FastAPI(
+    title="WareShipManifest Print Service API",
+    description=(
+        "HTTP API for the WareShipManifest print service.\n\n"
+        "## Overview\n"
+        "Generate warehouse shipping documents (packing list / sign-off receipt) "
+        "as PDF and send them to a physical printer through SumatraPDF.\n\n"
+        "## Endpoints\n"
+        "* `POST /api/print` — generate a document and print it "
+        "(or preview with `dry_run`).\n"
+        "* `GET /api/health` — liveness probe.\n\n"
+        "## Authentication\n"
+        "When `api.token` is configured in `settings.yaml`, every request must "
+        "carry a bearer token. Click the **Authorize** button (top right of this "
+        "page), enter the token value, and it will be sent automatically as "
+        "`Authorization: Bearer <token>`. Requests without a valid token are "
+        "rejected with `401` (missing) or `403` (invalid).\n\n"
+        "## Notes\n"
+        "* `dry_run=true` generates the PDF and builds the print command but "
+        "does **not** send paper to the printer — useful for testing.\n"
+        "* The target printer falls back to `printers.default` in `settings.yaml` "
+        "when `printer` is omitted.\n"
+        "* Paper size and orientation are chosen automatically per report type: "
+        "packing_list → A5 landscape, sign_receipt → A4 portrait.\n"
+        "* A JSON line is appended to `logs/print.log` for every request."
+    ),
+    version="1.0.0",
+    docs_url="/api/docs",
+    openapi_url="/api/openapi.json",
+    openapi_tags=[
+        {
+            "name": "print",
+            "description": "Generate shipping documents and send them to a printer.",
+        }
+    ],
+)
+
+api_router = APIRouter(prefix="/api", tags=["print"])
+
+security = HTTPBearer(auto_error=False)
+
+
+class PrintRequest(BaseModel):
+    report_type: Literal["packing_list", "sign_receipt"] = Field(
+        ...,
+        description="Type of document to generate and print.",
+        examples=["packing_list", "sign_receipt"],
+    )
+    params: dict[str, Any] = Field(
+        ...,
+        description=(
+            "Report parameters. packing_list requires `paichan_no` (str) and "
+            "`box_no` (int); sign_receipt requires `receipt_no` (str, 11-digit)."
+        ),
+        examples=[{"paichan_no": "R07425", "box_no": 1}],
+    )
+    printer: Optional[str] = Field(
+        None,
+        description="Target printer name. Falls back to `printers.default` in settings.yaml when omitted.",
+        examples=["Canon G1030 series (网络 USB1)"],
+    )
+    dry_run: bool = Field(
+        False,
+        description="When true, generate the PDF and build the print command but do NOT send it to the printer.",
+        examples=[True, False],
+    )
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "report_type": "packing_list",
+                "params": {"paichan_no": "R07425", "box_no": 1},
+                "dry_run": False,
+            }
+        }
+    )
+
+
+class PrintResponse(BaseModel):
+    ok: bool = Field(..., description="Whether the print job succeeded.")
+    report_type: str = Field(..., description="The report type that was requested.")
+    printer: Optional[str] = Field(None, description="The printer the job was sent to.")
+    pdf: str = Field(..., description="Absolute path of the generated PDF file.")
+    dry_run: bool = Field(
+        ..., description="True when the PDF was generated but not physically printed."
+    )
+    returncode: Optional[int] = Field(
+        None, description="Exit code of the SumatraPDF print process (null in dry_run)."
+    )
+    error: Optional[str] = Field(None, description="Error message, if any.")
+
+
+class PrintError(BaseModel):
+    detail: str = Field(
+        ..., description="Human-readable error message describing what went wrong."
+    )
+
+
+def _verify_token(
+    creds: Optional[HTTPAuthorizationCredentials] = Depends(security),
+) -> None:
+    """Skip auth when no token is configured (local debugging); otherwise verify Bearer."""
+    if not settings.api.token:
+        return
+    if creds is None:
+        raise HTTPException(status_code=401, detail="Missing bearer token")
+    if creds.credentials != settings.api.token:
+        raise HTTPException(status_code=403, detail="Invalid token")
+
+
+def _log(result: dict) -> None:
+    log_path = PROJECT_ROOT / "logs" / "print.log"
+    log_path.parent.mkdir(parents=True, exist_ok=True)
+    entry = {**result, "ts": datetime.now().isoformat()}
+    with open(log_path, "a", encoding="utf-8") as f:
+        f.write(json.dumps(entry, ensure_ascii=False, default=str) + "\n")
+
+
+@api_router.post(
+    "/print",
+    summary="Generate and print a shipping document",
+    description=(
+        "Generate the requested document as a PDF using the existing report "
+        "pipeline, then send the PDF to the target printer via SumatraPDF in "
+        "silent mode.\n\n"
+        "**Flow**\n"
+        "1. Validate the request and the required parameters.\n"
+        "2. Generate the PDF from the database.\n"
+        "3. Build and execute the SumatraPDF print command "
+        "(skipped when `dry_run=true`).\n"
+        "4. Append a JSON line to `logs/print.log`.\n"
+        "5. Return the job result.\n\n"
+        "Set `dry_run=true` to generate the PDF without consuming paper."
+    ),
+    response_description="Print job result, including the generated PDF path.",
+    response_model=PrintResponse,
+    responses={
+        400: {
+            "description": "Invalid or missing parameters (e.g. required field not provided).",
+            "model": PrintError,
+        },
+        401: {"description": "Missing bearer token (auth enabled).", "model": PrintError},
+        403: {"description": "Invalid bearer token.", "model": PrintError},
+        404: {
+            "description": "Referenced record not found (e.g. receipt_no / box does not exist).",
+            "model": PrintError,
+        },
+        422: {
+            "description": "Report generation or printing failed at runtime.",
+            "model": PrintError,
+        },
+        500: {"description": "Unexpected server error.", "model": PrintError},
+    },
+    dependencies=[Depends(_verify_token)],
+)
+def print_endpoint(req: PrintRequest):
+    try:
+        result = pd.print_document(
+            report_type=req.report_type,
+            params=req.params,
+            printer=req.printer,
+            dry_run=req.dry_run,
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    except FileNotFoundError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+    except RuntimeError as e:
+        raise HTTPException(status_code=422, detail=str(e))
+    except Exception as e:  # noqa: BLE001
+        raise HTTPException(status_code=500, detail=f"print failed: {e}")
+
+    _log(result)
+    return {
+        "ok": result.get("ok", False),
+        "report_type": req.report_type,
+        "printer": result.get("printer"),
+        "pdf": str(result.get("pdf")),
+        "dry_run": result.get("dry_run", False),
+        "returncode": result.get("returncode"),
+        "error": result.get("error"),
+    }
+
+
+@api_router.get(
+    "/health",
+    summary="Liveness probe",
+    description=(
+        "Returns `{\"status\": \"ok\"}` when the service is running. "
+        "No authentication is required. Use it for health checks / uptime monitoring."
+    ),
+    response_description="Service status.",
+    responses={
+        200: {
+            "description": "Service is alive.",
+            "content": {
+                "application/json": {
+                    "example": {"status": "ok"},
+                }
+            },
+        }
+    },
+)
+def health():
+    return {"status": "ok"}
+
+
+app.include_router(api_router)

core/print_document.py

@@ -0,0 +1,61 @@
+"""打印编排：生成报表 PDF -> 送打印机。
+
+复用现有 run.generate() 完成「查询→装配→渲染→PDF→页脚盖印」全链路，
+再调用 core.printing 把 PDF 静默打印到默认/指定打印机。
+"""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+PROJECT_ROOT = Path(__file__).resolve().parent.parent
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+# run.py 在项目根，生成逻辑已验证；此处直接复用，不改动 run.py
+import run  # noqa: E402
+
+from core.settings import settings  # noqa: E402
+from core import printing  # noqa: E402
+
+VALID_REPORTS = ("packing_list", "sign_receipt")
+
+
+def print_document(
+    report_type: str,
+    params: dict,
+    printer: str | None = None,
+    dry_run: bool = False,
+) -> dict:
+    """生成报表 PDF 并打印。
+
+    :param report_type: packing_list | sign_receipt
+    :param params: 报表参数（如 paichan_no/box_no 或 receipt_no）
+    :param printer: 指定打印机名；为 None 时用 settings.printers.default
+    :param dry_run: True 时只构造打印命令不真正执行（用于验证链路）
+    :return: 结构化结果（含 pdf 路径、打印命令、ok 状态）
+    """
+    if report_type not in VALID_REPORTS:
+        raise ValueError(f"未知报表类型：{report_type}（支持 {list(VALID_REPORTS)}）")
+
+    # 1. 生成 PDF（run.generate 内部已做 required 参数校验与 DB 查询）
+    output = run._default_output(report_type, params)
+    out_path = run.generate(report_type, params, output)
+
+    # 2. 选打印机与纸张（装箱单 A5 横向 / 签收单 A4 纵向）
+    target_printer = printer or settings.printers.default
+    paper_settings = settings.print.settings_by_report.get(
+        report_type, settings.print.default_settings
+    )
+
+    # 3. 打印
+    result = printing.print_pdf(
+        pdf_path=out_path,
+        printer=target_printer,
+        settings=paper_settings,
+        sumatrapdf_path=settings.print.sumatrapdf_path,
+        dry_run=dry_run,
+    )
+    result["report_type"] = report_type
+    result["params"] = params
+    return result

core/printing.py

@@ -0,0 +1,64 @@
+"""打印后端：把已生成的 PDF 静默送到物理/网络打印机。
+
+MVP 使用 SumatraPDF 命令行打印（轻量、支持 -print-to 指定打印机与纸张，
+配合 -silent 适合服务端无人值守打印）。
+"""
+from __future__ import annotations
+
+import subprocess
+from pathlib import Path
+
+
+def build_command(
+    sumatrapdf_path: str, printer: str, settings: str, pdf_path: str
+) -> list[str]:
+    """构造 SumatraPDF 打印命令。"""
+    return [
+        sumatrapdf_path,
+        "-print-to",
+        printer,
+        "-print-settings",
+        settings,
+        "-silent",
+        pdf_path,
+    ]
+
+
+def print_pdf(
+    pdf_path: str | Path,
+    printer: str,
+    settings: str = "paper=A4,orientation=portrait,color=bw",
+    sumatrapdf_path: str = "C:/Program Files/SumatraPDF/SumatraPDF.exe",
+    timeout: int = 60,
+    dry_run: bool = False,
+) -> dict:
+    """调用 SumatraPDF 把 PDF 静默打印到指定打印机。
+
+    返回结构化结果；dry_run=True 时只构造命令并上报，不真正执行（用于验证链路）。
+    """
+    pdf_path = str(pdf_path)
+    cmd = build_command(sumatrapdf_path, printer, settings, pdf_path)
+    result: dict = {
+        "printer": printer,
+        "pdf": pdf_path,
+        "command": cmd,
+        "dry_run": dry_run,
+    }
+    if dry_run:
+        result["ok"] = True
+        result["note"] = "dry_run: command constructed, not executed"
+        return result
+    try:
+        proc = subprocess.run(
+            cmd, capture_output=True, text=True, timeout=timeout, check=False
+        )
+        result["returncode"] = proc.returncode
+        result["stdout"] = proc.stdout
+        result["stderr"] = proc.stderr
+        result["ok"] = proc.returncode == 0
+        if proc.returncode != 0:
+            result["error"] = (proc.stderr or proc.stdout or "unknown error").strip()
+    except subprocess.TimeoutExpired:
+        result["ok"] = False
+        result["error"] = f"print timeout after {timeout}s"
+    return result

core/settings.py

@@ -48,11 +48,37 @@ class ReportConfig(BaseModel):
     debug_border: bool = False
 
 
+class PrintersConfig(BaseModel):
+    """打印机配置（MVP：单一默认打印机）"""
+
+    default: str
+
+
+class PrintConfig(BaseModel):
+    """打印后端配置"""
+
+    backend: str = "sumatrapdf"
+    sumatrapdf_path: str = "C:/Program Files/SumatraPDF/SumatraPDF.exe"
+    default_settings: str = "paper=A4,orientation=portrait,color=bw"
+    settings_by_report: dict[str, str] = {}
+
+
+class ApiConfig(BaseModel):
+    """HTTP 打印服务配置"""
+
+    host: str = "0.0.0.0"
+    port: int = 8000
+    token: str = ""
+
+
 class Settings(BaseModel):
     """应用配置"""
 
     database: DatabaseConfig
     report: ReportConfig = ReportConfig()
+    printers: PrintersConfig
+    print: PrintConfig = PrintConfig()
+    api: ApiConfig = ApiConfig()
 
     @property
     def database_url(self) -> URL:

requirements.txt

@@ -6,3 +6,7 @@ pydantic-yaml>=0.12
 # dev / verification only (not required at runtime)
 pytest>=8.0
 pymupdf>=1.24
+
+# HTTP 打印服务（新增打印接口）
+fastapi>=0.110
+uvicorn>=0.29

serve.py

@@ -0,0 +1,11 @@
+"""打印服务启动入口：python serve.py"""
+from core.settings import settings
+import uvicorn
+
+if __name__ == "__main__":
+    uvicorn.run(
+        "core.print_api:app",
+        host=settings.api.host,
+        port=settings.api.port,
+        log_level="info",
+    )