# src/certflow/services/cert_print_engine.py
"""合格证打印引擎
提供 GDI 精确打印、HTML 降级打印和打印策略选择。
"""
from __future__ import annotations
import contextlib
import tempfile
import traceback
from collections.abc import Callable
from typing import Any
from certflow.config.print_config import (
fields_to_tuples,
get_font,
get_layout,
get_paper,
get_print_fields,
load_print_config,
)
from certflow.utils.logger import logger
[文档]
class CertPrintEngine:
"""合格证打印引擎
提供打印的核心能力：
- GDI 精确打印到预印卡纸
- HTML 降级打印方案
- 打印策略选择（GDI / ESC/P-K）
Attributes:
printer_name: 默认打印机名称
"""
def __init__(self, printer_name: str = "") -> None:
self.printer_name = printer_name
self._strategy = None
self._last_os_job_id: int | None = None # 最近一次打印提交的 OS 作业 ID
@property
def last_os_job_id(self) -> int | None:
"""最近一次打印提交的 OS 作业 ID（GDI 路径经 ``hdc.StartDoc``、ESC 路径
经策略委托 ``PrinterService`` 捕获）。headless 下为 ``None``。
"""
return self._last_os_job_id
# ============================================================
# GDI 精确打印（核心方法）
# ============================================================
@staticmethod
def _resolve_printer(printer_name: str | None) -> str:
"""解析打印机名称，返回可用的打印机名
Args:
printer_name: 用户指定的打印机名称，None 或空串则取系统默认打印机
Returns:
str: 实际使用的打印机名称，无可用打印机时返回空串
"""
import win32print
if printer_name:
return printer_name
printer = win32print.GetDefaultPrinter()
if not printer:
logger.error("没有可用打印机")
return printer or ""
# ============================================================
# 打印策略
# ============================================================
def _get_strategy(self) -> Any:
"""获取打印策略（懒加载，从配置文件读取引擎类型）
Returns:
Any: 打印策略实例（GDI 或 ESC/P-K）
"""
if self._strategy is None:
from .printer.print_strategy import PrintStrategyFactory
self._strategy = PrintStrategyFactory.from_config(self.printer_name)
return self._strategy
[文档]
def print_single(
self,
data: dict[str, Any],
serial_number: str,
printer_name: str | None = None,
print_fields: list[str] | None = None,
) -> bool:
"""打印单台合格证（策略模式入口）
根据 config.yaml → print.engine 自动选择 GDI 或 ESC/P-K 引擎。
Args:
data: 打印数据字典（含 product_name、dn、pn 等字段）
serial_number: 合格证编号（SN）
printer_name: 打印机名称，None 则使用默认打印机
print_fields: 需要打印的字段名列表，None 表示打印全部字段
Returns:
bool: 是否打印成功
"""
strategy = self._get_strategy()
try:
result = strategy.print_certificate(data, serial_number, copies=1)
# 捕获 OS 作业 ID（偏差 5 作业 ID 捕获链路）：strategy 透出最近一次
# 提交的 spooler 作业 ID，供队列做 OS 级取消。
self._last_os_job_id = getattr(strategy, "last_os_job_id", None)
if result:
logger.info(f"打印成功 [{strategy.__class__.__name__}]: {serial_number}")
else:
logger.warning(f"打印失败 [{strategy.__class__.__name__}]: {serial_number}")
return result
except Exception as e:
logger.error(
f"打印异常 [{strategy.__class__.__name__}]: "
f"{serial_number}: {e}\n{traceback.format_exc()}"
)
return False
@staticmethod
def _resolve_user_offsets(layout: dict) -> tuple[float, float]:
"""从 userconfig.yaml 读取用户覆盖的偏移量
Args:
layout: 布局配置字典
Returns:
tuple[float, float]: (x_offset, y_offset)
"""
try:
from certflow.config.settings import _get_user_config
user_cfg = _get_user_config()
printer_offsets = user_cfg.get("printer_offsets", {})
x = printer_offsets.get("x_offset")
if x is None:
x = layout.get("x_offset", 0)
if x == 0:
x = 19 # 默认值兜底
y = printer_offsets.get("y_offset", layout.get("y_offset", 0))
return x, y
except ImportError:
return layout.get("x_offset", 0), layout.get("y_offset", 0)
@staticmethod
def _render_print_field(
hdc,
mm_to_px: Callable[[float], int],
dpi_x: int,
field_info: tuple,
data: dict[str, Any],
serial_number: str,
layout: dict,
font_cfg: dict,
paper: dict,
x_offset: float,
y_offset: float,
value_label_gap: float,
) -> bool:
"""渲染单个打印字段，返回 True 表示实际打印了内容
Args:
hdc: GDI 设备上下文句柄
mm_to_px: 毫米转像素的Callable
dpi_x: 横向 DPI
field_info: 字段元组 (label, key, x/y/offset...)
data: 打印数据字典
serial_number: 合格证编号（SN）
layout: 布局配置字典
font_cfg: 字体配置字典
paper: 纸张配置字典
x_offset: 用户 X 偏移
y_offset: 用户 Y 偏移
value_label_gap: 标签与值的水平间距
Returns:
bool: 是否实际打印了该字段（值为空则不打印）
"""
# debug: 打印字段信息
logger.debug(
f"渲染字段: field_info[0]={field_info[0] if len(field_info) > 0 else 'N/A'}, "
f"len={len(field_info)}, "
f"x={field_info[2] if len(field_info) >= 6 else layout['label_x']}, "
f"font_size={field_info[4] if len(field_info) >= 6 else layout['font_size']}, "
f"dpi_x={dpi_x}"
)
# 兼容两种格式
if len(field_info) >= 6:
# 模板坐标格式: (label, key, x, y, font_size, align)
label, key, field_x, field_y, field_font_size, field_align = field_info[:6]
use_label_x = (field_x if field_x is not None else layout["label_x"]) + x_offset
use_y = (field_y if field_y is not None else layout["start_y"]) + y_offset
use_font_size = field_font_size if field_font_size else layout["font_size"]
use_align = field_align if field_align else "left"
else:
# 回退格式: (label, key, offset_idx)
label, key, offset_idx = field_info[:3]
use_label_x = layout["label_x"] + x_offset
use_y = layout["start_y"] + offset_idx * layout["line_spacing"] + y_offset
use_font_size = layout["font_size"]
use_align = "left"
# debug: 打印实际使用的字号
logger.debug(
f"字段={key}, use_font_size={use_font_size}, "
f"layout.font_size={layout['font_size']}, "
f"是否切换={use_font_size != layout['font_size']}, "
f"height={int(use_font_size * dpi_x / 72)}"
)
value = data.get(key, "") or ""
if key == "serial_number":
value = serial_number
if not str(value).strip():
return False
# 针打统一使用默认字号，不做逐字段切换
# （GDI CreateFont 在针式打印机上 height 处理不一致，切换反而导致大小异常）
# 对齐方式
if use_align == "center":
hdc.SetTextAlign(1) # TA_CENTER
center_x = mm_to_px(paper["width"] / 2)
hdc.TextOut(center_x, mm_to_px(use_y), str(value))
hdc.SetTextAlign(0)
elif use_align == "right":
hdc.SetTextAlign(2) # TA_RIGHT
right_x = mm_to_px(paper["width"] - (use_label_x - x_offset))
hdc.TextOut(right_x, mm_to_px(use_y), str(value))
hdc.SetTextAlign(0)
else:
# 只打印值，不打印标签（标签已在预印底板上）
hdc.TextOut(mm_to_px(use_label_x), mm_to_px(use_y), str(value))
return True
def _print_single_gdi(
self,
data: dict[str, Any],
serial_number: str,
printer_name: str | None = None,
print_fields: list[str] | None = None,
) -> bool:
"""使用 win32 GDI 精确打印单台合格证到预印卡纸
Args:
data: 打印数据字典
serial_number: 合格证编号（SN）
printer_name: 打印机名称，None 则使用默认打印机
print_fields: 需要打印的字段名列表，None 表示打印全部字段
Returns:
bool: 是否打印成功
"""
import win32print
import win32ui
# 清除缓存，确保每次打印都读取最新配置
load_print_config.cache_clear()
printer = self._resolve_printer(printer_name or self.printer_name)
if not printer:
return False
hprinter = None
fields = print_fields if print_fields else fields_to_tuples()
paper = get_paper()
layout = get_layout()
font_cfg = get_font()
try:
logger.debug(f"GDI 打印: 编号={serial_number}, 打印机={printer}")
hprinter = win32print.OpenPrinter(printer)
devmode = win32print.GetPrinter(hprinter, 2)["pDevMode"]
devmode.PaperSize = 256
devmode.PaperWidth = int(paper["width"] * 10)
devmode.PaperLength = int(paper["height"] * 10)
hdc = win32ui.CreateDC()
hdc.CreatePrinterDC(printer)
job_id = hdc.StartDoc(f"合格证_{serial_number}")
self._last_os_job_id = job_id
hdc.StartPage()
dpi_x = hdc.GetDeviceCaps(88)
def mm_to_px(mm_val: float) -> int:
return int(mm_val * dpi_x / 25.4)
# 默认字号: 正值 = 单元格高度（点阵高度），适配针打
font = win32ui.CreateFont(
{
"name": font_cfg["name"],
"height": int(layout["font_size"] * dpi_x / 72),
"weight": font_cfg["weight"],
}
)
hdc.SelectObject(font)
hdc.SetTextAlign(0)
hdc.SetBkMode(1)
x_offset, y_offset = self._resolve_user_offsets(layout)
value_label_gap = layout["value_x"] - layout["label_x"]
logger.debug(
f"打印坐标: label_x={layout['label_x']}, value_x={layout['value_x']}, "
f"x_offset={x_offset}, y_offset={y_offset}"
)
field_count = 0
for field_info in fields:
if self._render_print_field(
hdc,
mm_to_px,
dpi_x,
field_info,
data,
serial_number,
layout,
font_cfg,
paper,
x_offset,
y_offset,
value_label_gap,
):
field_count += 1
hdc.EndPage()
hdc.EndDoc()
hdc.DeleteDC()
logger.info(f"GDI 打印成功: {serial_number} (共 {field_count} 个字段)")
return True
except Exception as e:
logger.error(f"GDI 打印失败 [{serial_number}]: {e}\n{traceback.format_exc()}")
return False
finally:
if hprinter is not None:
with contextlib.suppress(Exception):
win32print.ClosePrinter(hprinter)
[文档]
def print_html_fallback(self, data: dict[str, Any], serial_number: str) -> str | None:
"""HTML 降级打印方案
GDI 打印失败时生成 HTML 文件用浏览器打印。
Args:
data: 打印数据字典
serial_number: 合格证编号（SN）
Returns:
str | None: 临时 HTML 文件路径，失败返回 None
"""
try:
html = self._generate_certificate_html(data, serial_number)
with tempfile.NamedTemporaryFile(
mode="w", suffix=".html", delete=False, encoding="utf-8"
) as f:
f.write(html)
temp_file = f.name
import webbrowser
webbrowser.open(temp_file)
logger.info(f"HTML 降级打印: {serial_number} → {temp_file}")
return temp_file
except Exception as e:
logger.error(f"HTML 降级打印失败 [{serial_number}]: {e}")
return None
@staticmethod
def _generate_certificate_html(data: dict[str, Any], serial_number: str) -> str:
"""生成合格证预览 HTML（配置驱动）
Args:
data: 打印数据字典
serial_number: 合格证编号（SN）
Returns:
str: 完整 HTML 文档字符串
"""
paper = get_paper()
layout = get_layout()
fields = get_print_fields()
field_rows = ""
for f in fields:
value = data.get(f["key"], "") or ""
if f["key"] == "serial_number":
value = serial_number
if not str(value).strip():
continue
top = layout["start_y"] + f["offset"] * layout["line_spacing"]
field_rows += f"""
<div class="field-row" style="top: {top}mm;">
<span class="field-label">{f["label"]}</span>
<span class="field-value" style="left: {layout["value_x"]}mm;">{value}</span>
</div>"""
return f"""<!DOCTYPE html>
<html><head><meta charset="UTF-8">
<style>
@page {{ size: {paper["width"]}mm {paper["height"]}mm; margin: 0; }}
body {{
width: {paper["width"]}mm; height: {paper["height"]}mm;
margin: 0; padding: 0;
font-family: "SimSun", "宋体", sans-serif;
position: relative; overflow: hidden;
}}
.field-row {{
position: absolute; left: {layout["label_x"]}mm;
font-size: 8pt; line-height: 1.4; white-space: nowrap;
}}
.field-label {{ color: #333; }}
.field-value {{ position: absolute; font-weight: bold; color: #000; }}
@media print {{ body {{ background-image: none !important; }} }}
</style>
<script>window.onload=function(){{window.print();}}</script>
</head><body>{field_rows}</body></html>"""