# src/certflow/controllers/printer_controller.py
"""打印机控制器
提供合格证打印的异步控制，封装 PrintWorker 线程和 PrinterController 信号机制。
通过 Qt 信号槽模式实现非阻塞打印，支持打印机切换和状态查询。
支持 PrintConfig 聚合配置透传。
Signals:
printer_changed: 打印机切换时发出，携带新打印机名称
status_changed: 打印状态变化时发出，携带状态描述文本
print_completed: 打印完成时发出，携带 (成功标志, 消息)
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from PySide6.QtCore import QObject, QThread, Signal
from certflow.services.cert_log_service import _default_inspector_id
from certflow.services.cert_numbering_policy import (
get_issue_date_source,
parse_product_code_meta,
)
from certflow.services.output_guard import run_guard, validate_pn_text
from certflow.services.printer import PrinterService
from certflow.services.printer.template_field_formatter import (
family_of,
get_field_formatter,
)
from certflow.services.printer.template_manager import get_template_manager
if TYPE_CHECKING:
from certflow.config.print_config import RuntimePrintConfig
[文档]
class PrintWorker(QThread):
"""打印工作线程
将打印操作放在独立线程中执行，避免阻塞 UI 主线程。
支持两种模式：
- 传入 PrintConfig：使用策略模式（ESC/P-K 或 GDI）
- 不传 PrintConfig：回退到 PrinterService.print_certificate() 文本模式
Signals:
progress: 打印进度信号 (current, total)
finished: 打印完成信号 (success: bool, message: str)
"""
progress = Signal(int, int)
"""打印进度信号 (current, total)"""
finished = Signal(bool, str)
"""打印完成信号 (success, message)"""
def __init__(
self,
service: PrinterService,
data: dict[str, Any],
copies: int = 1,
config: RuntimePrintConfig | None = None,
):
"""初始化打印工作线程
Args:
service: PrinterService 打印机服务实例
data: 打印数据字典，包含 product_name/product_model 等字段
copies: 打印份数
config: RuntimePrintConfig 聚合配置，传入时使用策略模式打印
"""
super().__init__()
self.service = service
self.data = data
self.copies = copies
self.config = config
[文档]
def run(self) -> None:
"""线程主执行方法（由 QThread.start() 触发）"""
try:
if self.config:
# 策略模式：通过 PrintStrategy 打印
success = self._print_with_strategy()
else:
# 回退：文本模式打印
success = self.service.print_certificate(self.data, self.copies)
self.finished.emit(success, "打印完成" if success else "打印失败")
except Exception as e:
self.finished.emit(False, str(e))
def _print_with_strategy(self) -> bool:
"""使用 RuntimePrintConfig 驱动的策略打印"""
from certflow.services.printer.print_strategy import PrintStrategyFactory
strategy = PrintStrategyFactory.from_config(
printer_name=self.service.printer_name,
config=self.config,
)
# 从数据中提取 serial_number
serial_number = self.data.get("cert_number") or self.data.get("serial_number", "")
return strategy.print_certificate(
data=self.data,
serial_number=serial_number,
copies=self.copies,
config=self.config,
)
[文档]
class PrinterController(QObject):
"""打印机控制器
管理打印机选择、打印任务调度和状态监控。
通过 Qt 信号槽实现 UI 与打印逻辑的解耦。
支持 PrintConfig 聚合配置透传。
Attributes:
_printer_name: 当前选择的打印机名称，None 表示未初始化
_service: 打印机服务实例（惰性创建，按名称缓存）
_print_worker: 当前活动的打印工作线程
Signals:
printer_changed: 打印机名称变更信号
status_changed: 打印状态文本信号
print_completed: 打印完成信号 (success, message)
"""
printer_changed = Signal(str)
"""打印机切换信号，携带新打印机名称"""
status_changed = Signal(str)
"""状态变更信号，携带状态描述文本"""
print_completed = Signal(bool, str)
"""打印完成信号 (success: bool, message: str)"""
def __init__(self) -> None:
"""初始化打印机控制器"""
super().__init__()
self._printer_name: str | None = None
self._service: PrinterService | None = None
# 持有所有活动打印线程的引用，避免被 GC 回收导致线程在运行时被销毁
# （批量打印会连续 start 多个线程，若仅保留最后一个引用，前序线程会被
# 提前销毁，表现为「指令未发送」/ QThread Destroyed while running）
self._print_workers: list[PrintWorker] = []
def _get_service(self) -> PrinterService:
"""获取或创建打印机服务实例（按打印机名称缓存）
Returns:
PrinterService: 打印机服务实例
"""
if self._service is None or self._service.printer_name != self._printer_name:
self._service = PrinterService(self._printer_name)
return self._service
@property
def current_printer(self) -> str:
"""当前打印机名称（惰性获取默认值）
Returns:
str: 打印机名称字符串，未获取到默认打印机时返回空字符串
"""
if self._printer_name is None:
self._printer_name = self._get_service().get_default_printer()
return self._printer_name or ""
[文档]
def set_printer(self, printer_name: str) -> None:
"""设置当前打印机并发出变更信号
Args:
printer_name: 目标打印机名称
Returns:
None
"""
self._printer_name = printer_name
self._service = PrinterService(printer_name)
self.printer_changed.emit(printer_name)
[文档]
def get_printers(self) -> list[str]:
"""获取系统可用打印机列表
Returns:
list[str]: 打印机名称列表
"""
return PrinterService().get_available_printers()
[文档]
def print_certificate(
self,
data: dict[str, Any],
copies: int = 1,
config: RuntimePrintConfig | None = None,
) -> None:
"""异步打印合格证（在独立线程中执行）
Args:
data: 打印数据字典，包含 product_name/product_model/dn/pn 等字段
copies: 打印份数，默认为 1
config: RuntimePrintConfig 聚合配置，传入时使用策略模式打印；
为 None 时回退到文本模式
Returns:
None
Examples:
>>> controller = PrinterController()
>>> controller.set_printer("HP LaserJet")
>>> controller.print_certificate(
... {"product_name": "闸阀", "product_model": "Z41H-16C"},
... copies=2,
... )
"""
service = self._get_service()
worker = PrintWorker(service, data, copies, config)
worker.finished.connect(self._on_print_finished)
# 完成后从引用列表移除，避免内存无限增长（同时保证运行期不被 GC）
worker.finished.connect(lambda *_: self._release_worker(worker))
self._print_workers.append(worker)
worker.start()
self.status_changed.emit(f"正在打印 {copies} 份...")
def _release_worker(self, worker: PrintWorker) -> None:
"""从活动线程列表移除已结束的 worker（保留运行期强引用）。"""
if worker in self._print_workers:
self._print_workers.remove(worker)
def _on_print_finished(self, success: bool, message: str) -> None:
"""打印完成回调（内部槽函数）
Args:
success: 是否打印成功
message: 结果描述信息
"""
self.print_completed.emit(success, message)
self.status_changed.emit(message)
[文档]
def get_printer_status(self) -> bool:
"""检查当前打印机是否就绪
Returns:
bool: True 表示就绪可打印，False 表示离线/缺纸/错误/暂停
"""
return self._get_service().is_printer_ready()
# ------------------------------------------------------------------
# 视图层穿透收口（阶段 C）：PrintView 原先直连多个 services 子模块
# （cert_numbering_policy / output_guard / template_field_formatter /
# template_manager / cert_log_service）的纯函数与校验器。现统一经本
# 控制器的静态门面委托，消除视图对 L3 服务层的散落依赖。
# 设计为类级静态方法：即使 printer_controller 实例未注入，视图仍可
# 经 ``PrinterController.xxx(...)`` 调用，保证旧调用方 / 测试零改动。
# ------------------------------------------------------------------
[文档]
@staticmethod
def family_of(template_key: str) -> str:
"""模板语言族判定（委托 ``TemplateFieldFormatter.family_of``）。"""
return family_of(template_key)
[文档]
@staticmethod
def get_issue_date_source() -> str:
"""获取出厂日期源配置（委托 ``cert_numbering_policy.get_issue_date_source``）。"""
return get_issue_date_source()
[文档]
@staticmethod
def validate_pn_text(family: str, pn_text: str) -> str:
"""校验压力文本（委托 ``output_guard.validate_pn_text``），返回错误信息或空串。"""
return validate_pn_text(family, pn_text)
[文档]
@staticmethod
def run_guard(*args: Any, **kwargs: Any) -> tuple[bool, dict[str, Any]]:
"""运行输出护栏（委托 ``output_guard.run_guard``），返回 (ok, updated)。"""
return run_guard(*args, **kwargs)
[文档]
@staticmethod
def get_template_manager() -> Any:
"""获取模板管理器（委托 ``template_manager.get_template_manager``）。"""
return get_template_manager()
[文档]
@staticmethod
def default_inspector_id(family: str) -> str:
"""获取指定语言族的默认检验工号（委托 ``cert_log_service._default_inspector_id``）。"""
return _default_inspector_id(family)