ee2fab3847
- packing_list renders A5 content into top region of an A4 portrait page when paper=a4, so A5 paper loaded horizontally in an A4 tray prints upright - run.generate/print_document/print_api thread the optional paper param through - README: new Print Service section with A4 carrier usage + curl example - Swagger: POST /api/print description documents the A4 carrier mode
262 lines
10 KiB
Python
262 lines
10 KiB
Python
"""
|
|
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"
|
|
"* Packing list supports an A4 carrier mode via `paper=\"a4\"`: the A5 content is "
|
|
"rendered on an A4 portrait page (top region) and printed as A4, so it comes out "
|
|
"correctly on A5 paper loaded horizontally in an A4 tray.\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)"],
|
|
)
|
|
paper: Optional[Literal["a4"]] = Field(
|
|
None,
|
|
description=(
|
|
"Paper carrier mode. Omit (null) for the default A5 landscape packing list. "
|
|
"Set to \"a4\" to render the packing list on an A4 portrait page with the A5 "
|
|
"content placed in the top region, so it prints correctly on A5 paper loaded "
|
|
"horizontally in an A4 tray (saves re-adjusting the paper clamps). Only "
|
|
"applies to `packing_list`; ignored for other report types."
|
|
),
|
|
examples=["a4"],
|
|
)
|
|
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")
|
|
|
|
|
|
@app.post(
|
|
"/api/print",
|
|
tags=["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.\n\n"
|
|
"**A4 Carrier Mode (packing_list only)**\n"
|
|
"To avoid re-adjusting the printer's paper clamps, A5 paper can be loaded "
|
|
"**horizontally** in an A4 tray. But horizontal loading would otherwise print "
|
|
"the A5 content upside-down. This mode fixes it: pass `paper=\"a4\"` and the "
|
|
"A5 content is rendered into the **top 148mm region of an A4 portrait page**, "
|
|
"then printed as A4 — the horizontally loaded A5 sheet receives the top region "
|
|
"and comes out upright and complete. Verified on a real printer.\n\n"
|
|
"Both conditions are required:\n"
|
|
"1. Request carries `paper=\"a4\"` (only affects `packing_list`; ignored for others).\n"
|
|
"2. Physically, the A5 paper is loaded **horizontally** in the A4 tray.\n\n"
|
|
"For a standard A5 tray (vertical loading), omit `paper` to fall back to normal "
|
|
"A5 landscape direct print. The A4-carrier PDF is named with an `_a4` suffix; "
|
|
"always read the returned `pdf` field rather than hard-coding the path."
|
|
),
|
|
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,
|
|
paper=req.paper,
|
|
)
|
|
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)
|