# src/certflow/services/printer/printer_service.py
"""打印机业务服务 - 专为 LQ-635KII 设计"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
try:
import win32print
_HAS_WIN32 = True
except ImportError:
_HAS_WIN32 = False
win32print = None # type: ignore[assignment]
from .lq635kii_printer import LQ635KIIPrinter, PaperSize, PrintQuality
from .template_manager import get_template_manager
if TYPE_CHECKING:
from certflow.config.print_config import RuntimePrintConfig
[文档]
class PrinterService:
"""LQ-635KII 打印机服务
提供合格证打印等高级功能
"""
# 合格证字段定义 (Y坐标基于 1/6 英寸行距)
CERT_FIELDS = [
("product_name", "产品名称：", 1),
("product_model", "产品型号：", 2),
("dn", "公称通径：", 3),
("pn", "公称压力：", 4),
("temperature", "工作温度：", 5),
("medium", "适用介质：", 6),
("check_standard", "检验标准：", 7),
("inspector_id", "检验工号：", 8),
("manufacture_date", "出厂日期：", 9),
("serial_number", "产品编号：", 10),
]
def __init__(self, printer_name: str | None = None) -> None:
"""初始化打印机服务
Args:
printer_name: 打印机名称，为 None 时使用系统默认打印机
"""
self.printer_name = printer_name
if printer_name is None:
self.printer_name = win32print.GetDefaultPrinter() if _HAS_WIN32 else ""
self._printer_cal_x_offset = self._load_x_offset()
self._printer_cal_y_offset = self._load_y_offset()
# 复用全局单例，避免每次打印作业都重新加载并打日志
self._template_manager = get_template_manager()
self._last_os_job_id: int | None = None # 最近一次打印提交的 OS 作业 ID
def _load_x_offset(self) -> float:
"""从 config.yaml 加载 X 偏移量
Returns:
float: X 方向偏移量 (mm)，默认 14.0
"""
try:
from certflow.config.settings import cfg
return cfg("print.layout.x_offset", 14.0)
except Exception:
return 14.0
def _load_y_offset(self) -> float:
"""从 config.yaml 加载 Y 校准偏移量（原 LQ635KIIPrinter 硬编码 -5，已配置化）
物理装纸/打印头起点偏差对**所有**文字统一叠加，正值向下、负值向上。
实际偏移需按 EPSON LQ-635KII 实机微调，故配置化暴露。
Returns:
float: Y 方向偏移量 (mm)，默认 0.0（与预览 RuntimePrintConfig 回退一致）
"""
try:
from certflow.config.settings import cfg
return cfg("print.layout.y_offset", 0.0)
except Exception:
return 0.0
[文档]
def get_printer(self) -> LQ635KIIPrinter:
"""获取打印机实例
Returns:
LQ635KIIPrinter: 打印机控制实例
"""
return LQ635KIIPrinter(
self.printer_name,
calibration_x_offset_mm=self._printer_cal_x_offset,
y_offset_mm=self._printer_cal_y_offset,
)
@property
def last_os_job_id(self) -> int | None:
"""最近一次打印提交的 OS 作业 ID（由 ``LQ635KIIPrinter._flush`` 捕获）。
用于打印队列的 OS 级取消 / 状态查询（BLUEPRINT §2.3 偏差 5 作业 ID
捕获链路）。每次新打印前由 ``_flush`` 覆盖，无 win32print（headless）
时为 ``None``。
"""
return self._last_os_job_id
[文档]
def print_certificate(self, data: dict[str, Any], copies: int = 1) -> bool:
"""简单文本模式打印（备用方案）
精确定位请使用 print_by_template_config（读 coordinates.yaml，单一坐标源）
使用文本模式打印，LQ-635KII 自动处理中文
Args:
data: 打印数据
copies: 打印份数
Returns:
是否成功
"""
try:
with self.get_printer() as printer:
for _ in range(copies):
printer.init()
printer.set_paper_size(PaperSize.certificate())
printer.set_quality(PrintQuality.LETTER)
printer.set_line_spacing_1_6() # 1/6 英寸行距
# 打印内容 (使用制表位对齐)
for field_key, label, _line_num in self.CERT_FIELDS:
value = data.get(field_key, "")
if field_key == "serial_number":
value = data.get("cert_number", value)
if value:
# 使用制表位实现标签和值的对齐
printer.println(f"{label}{value}")
else:
# 空值只打印标签
printer.println(label)
printer.form_feed() # 换页
printer.flush()
self._last_os_job_id = getattr(printer, "_last_job_id", None)
return True
except Exception as e:
print(f"打印失败: {e}")
return False
[文档]
def print_by_template(
self,
template_key: str,
data: dict[str, Any],
copies: int = 1,
x_offsets: dict[str, float] | None = None,
) -> bool:
"""根据模板打印合格证
使用 coordinates.yaml 中定义的模板坐标进行精确定位打印。
Args:
template_key: 模板键名（如 "russian", "full_chinese_1"）
data: 打印数据字典
copies: 打印份数
x_offsets: 英制字段 X 偏移（mm），键为字段名（如 ``pn``/``dn``），
非零时将该字段坐标右移，使数值落在预印单位标签（MPa/Lb）下。
Returns:
是否成功
"""
template = self._template_manager.get_template(template_key)
if not template:
print(f"模板不存在: {template_key}")
return False
try:
with self.get_printer() as printer:
printer.init()
printer.set_paper_size(PaperSize.certificate())
for _ in range(copies):
for field_name, pos in template["fields"].items():
value = data.get(field_name, "")
if field_name == "serial_number":
value = data.get("cert_number", value)
if not value:
continue
off = x_offsets.get(field_name, 0.0) if x_offsets else 0.0
printer.set_absolute_x_mm(pos["x_mm"] + off)
printer.set_absolute_y_mm(pos["y_mm"])
printer.print_text(str(value))
printer.form_feed()
printer.flush()
self._last_os_job_id = getattr(printer, "_last_job_id", None)
return True
except Exception as e:
print(f"模板打印失败 [{template_key}]: {e}")
return False
[文档]
def print_by_template_config(
self,
config: RuntimePrintConfig,
data: dict[str, Any],
copies: int = 1,
x_offsets: dict[str, float] | None = None,
) -> bool:
"""根据 RuntimePrintConfig 打印合格证
使用 RuntimePrintConfig.template.fields 中定义的坐标进行精确定位打印。
相比 print_by_template()，不再依赖 TemplateManager，直接使用 RuntimePrintConfig。
Args:
config: 聚合打印配置（包含 engine/template/layout 等）
data: 打印数据字典
copies: 打印份数
x_offsets: 英制字段 X 偏移（mm），键为字段名（如 ``pn``/``dn``），
非零时将该字段坐标右移，使数值落在预印单位标签（MPa/Lb）下。
Returns:
是否成功
"""
if not config.template or not config.template.fields:
print("PrintConfig 中无模板字段，无法执行精确定位打印")
return False
try:
with self.get_printer() as printer:
printer.init()
printer.set_paper_size(PaperSize.certificate())
printer.set_quality(PrintQuality.LETTER)
for _ in range(copies):
for field_def in config.template.fields:
value = data.get(field_def.key, "")
if field_def.key == "serial_number":
value = data.get("cert_number", value)
if not value:
continue
off = x_offsets.get(field_def.key, 0.0) if x_offsets else 0.0
printer.set_absolute_x_mm(field_def.x_mm + off)
printer.set_absolute_y_mm(field_def.y_mm)
printer.print_text(str(value))
printer.form_feed()
printer.flush()
self._last_os_job_id = getattr(printer, "_last_job_id", None)
return True
except Exception as e:
print(f"模板配置打印失败 [{config.template.key}]: {e}")
return False
[文档]
def render_certificate_bytes(
self,
config: RuntimePrintConfig,
data: dict[str, Any],
copies: int = 1,
x_offsets: dict[str, float] | None = None,
) -> bytes:
"""渲染合格证为 ESC/P-K 原始字节流（**不发送到打印机**）。
与 :meth:`print_by_template_config` 共用同一套命令构建逻辑，但刻意
不调用 ``open()`` / ``flush()``，因此无需真实打印机、不会真正出纸，
可直接拿到「即将发送给打印机的完整字节」，供调试落盘 / 十六进制分析。
Args:
config: RuntimePrintConfig 聚合配置（含 ``template.fields`` 坐标）。
data: 打印数据字典（含 product_name/dn/pn/... 及 ``cert_number``）。
copies: 份数，默认 1。
x_offsets: 英制字段 X 偏移（mm），键为字段名。
Returns:
bytes: 完整 ESC/P-K 命令字节流；配置无模板字段时返回空字节。
"""
if not config.template or not config.template.fields:
return b""
# 仅构造，不打开真实打印机（open() 会调用 win32print.OpenPrinter，
# 在无打印机/CI 环境下会失败；渲染过程本身只向 _buffer 追加字节）。
printer = self.get_printer()
printer.init()
printer.set_paper_size(PaperSize.certificate())
printer.set_quality(PrintQuality.LETTER)
for _ in range(max(1, copies)):
for field_def in config.template.fields:
value = data.get(field_def.key, "")
if field_def.key == "serial_number":
value = data.get("cert_number", value)
if not value:
continue
off = x_offsets.get(field_def.key, 0.0) if x_offsets else 0.0
printer.set_absolute_x_mm(field_def.x_mm + off)
printer.set_absolute_y_mm(field_def.y_mm)
printer.print_text(str(value))
printer.form_feed()
return bytes(printer._buffer)
[文档]
def get_available_printers(self) -> list[str]:
"""获取可用打印机列表
Returns:
list[str]: 本地和网络连接的所有打印机名称列表
"""
try:
printers = win32print.EnumPrinters(
win32print.PRINTER_ENUM_LOCAL | win32print.PRINTER_ENUM_CONNECTIONS
)
return [p[2] for p in printers]
except Exception:
return []
[文档]
def get_default_printer(self) -> str:
"""获取系统默认打印机
Returns:
str: 默认打印机名称，失败时返回空字符串
"""
try:
return win32print.GetDefaultPrinter()
except Exception:
return ""
[文档]
def is_printer_ready(self) -> bool:
"""检查打印机是否就绪
通过查询打印机状态位判断是否处于离线、缺纸、错误或暂停状态。
Returns:
bool: True 表示打印机就绪可用
"""
try:
handle = win32print.OpenPrinter(self.printer_name)
info = win32print.GetPrinter(handle, 2)
win32print.ClosePrinter(handle)
status = info.get("Status", 0)
offline = win32print.PRINTER_STATUS_OFFLINE
paper_out = win32print.PRINTER_STATUS_PAPER_OUT
error = win32print.PRINTER_STATUS_ERROR
paused = win32print.PRINTER_STATUS_PAUSED
return not (status & (offline | paper_out | error | paused))
except Exception:
return False