# src/certflow/views/print_view.py
"""合格证打印视图 - 只负责打印，不包含数据源/历史/设置
从 CertificatePrintView 拆出，保留核心打印功能：
- 模板选择
- 产品信息编辑
- 编号设置
- 打印预览
- 单次/批量打印
"""
from __future__ import annotations
import json
import os
from datetime import datetime
from typing import TYPE_CHECKING, Any
from PySide6.QtCore import Qt, QTimer, Signal
from PySide6.QtWidgets import (
QCheckBox,
QComboBox,
QFileDialog,
QFrame,
QGridLayout,
QGroupBox,
QHBoxLayout,
QInputDialog,
QLabel,
QLineEdit,
QMessageBox,
QProgressBar,
QPushButton,
QScrollArea,
QSpinBox,
QSplitter,
QVBoxLayout,
QWidget,
)
from certflow.config.paths_override import cfg
from certflow.controllers.model_param_mapping_controller import ModelParamMappingController
from certflow.controllers.print_queue_controller import PrintQueueController
from certflow.controllers.printer_controller import PrinterController
from certflow.utils.logger import logger
from certflow.views.certificate_preview_painter import render_certificate_pixmap
from certflow.views.queue_panel import QueuePanel
if TYPE_CHECKING:
pass
[文档]
class PrintView(QWidget):
"""合格证打印视图
Signals:
print_completed: 打印完成信号，携带 certificate_ids 列表
"""
print_completed = Signal(list)
# 跳转扫描件生成：携带已加载的合格证 ID 列表（供主窗口切换到扫描件视图）
navigate_to_scan = Signal(list)
def __init__(
self,
session: Any,
config: dict = None,
db_manager: Any = None,
parent: Any = None,
printer_controller: Any = None,
) -> None:
"""初始化PrintView。
Args:
printer_controller: 注入共享打印机控制器（OutputController 统一调度，
§11.8 阶段4）；为 None 时沿用原有懒加载逻辑（测试/旧调用方零改动）。
Args:
session: SQLAlchemy 会话对象
config: 配置字典
db_manager: 数据库管理器实例
parent: 父控件（可选）
Returns:
None
"""
super().__init__(parent)
self.db_manager = db_manager
self.session = session
self.config = config or {}
# 会话生命周期统一由 db_manager 管理；优先 db_manager，回退裸 session（测试/旧调用方）。
# 视图不再直接把裸 session 注入 Controller，统一经 db_manager 派生（阶段 C 补丁 59）。
self._printer_controller = None
if printer_controller is not None:
self._printer_controller = printer_controller
# 连接打印完成信号，感知异步打印结果（与懒加载路径保持一致）
self._printer_controller.print_completed.connect(self._on_print_done)
# 型号字典控制器（懒加载）
self._mpm_controller = None
# 状态
self._current_certificate_id: int | None = None
self._current_cert: Any = None # 当前绑定的 Certificate（供模式切换重加载）
self._current_sale_plan: Any = None # 当前绑定的 SalePlan（供编号回填取整批区间）
self._loaded_plans: list = [] # 本次加载的多个销售计划（支持一次打印多条记录）
self._loaded_certs: list = [] # 对应的 Certificate（按 unique_key 取最新一条）
self._serial_numbers: list[str] = []
self._current_serial_index: int = 0
self._three_digit_mode: bool = False # 编号模式「三位流水」时强制 3 位流水
# 编号续号：配置驱动自动续号 + 控制器懒加载 + 加载记录抑制标志
self._auto_continue = bool(cfg("certificate.numbering.auto_continue", True))
self._cert_controller = None
self._loading_record = False # 加载记录时抑制「目标年月」自动覆盖起始
# 防抖定时器
self._refresh_timer = QTimer()
self._refresh_timer.setSingleShot(True)
self._refresh_timer.timeout.connect(self._do_refresh_preview)
# 打印防重复标志
self._is_printing = False
# 扫描件生成控制器（懒加载，仅当勾选「生成扫描件」时触发）
self._scan_controller = None
# 打印机检查缓存（同一批次只检查一次）
self._checked_printer: str | None = None
self._init_config()
self._init_queue()
self.setup_ui()
self._connect_signals()
self._refresh_preview()
# ============================================================
# 配置加载
# ============================================================
def _init_config(self) -> None:
"""加载配置"""
print_cfg = self.config.get("print", {})
self._default_printer = print_cfg.get("default_printer", "默认打印机")
self._default_copies = print_cfg.get("default_copies", 1)
layout_cfg = print_cfg.get("layout", {})
paper_cfg = print_cfg.get("paper", {})
self._label_x = layout_cfg.get("label_x", 8)
self._value_x = layout_cfg.get("value_x", 28)
self._start_y = layout_cfg.get("start_y", 28)
self._line_spacing = layout_cfg.get("line_spacing", 7.5)
self._paper_width = paper_cfg.get("width", 60)
self._paper_height = paper_cfg.get("height", 100)
# 模板列表（A1：动态从 TemplateManager 读取全部变体，含印章差异）
self._template_options = self._load_template_options() # list[(显示标签, 配置键)]
self._current_template_key = (
self._template_options[0][1] if self._template_options else "full_chinese_1"
)
# 配置驱动默认模板（print.default_template，设置页可改；匹配 key/标签/去变体后缀，找不到回退首个）
cfg_default = cfg("print.default_template", "")
if cfg_default:
for _label, _key in self._template_options:
if (
_key == cfg_default
or _label == cfg_default
or _label.split(" · ")[0] == cfg_default
):
self._current_template_key = _key
break
# 打印选项默认值（配置驱动：output.targets.print.options，回退 True）
self._opt_auto_fill = bool(cfg("output.targets.print.options.auto_fill", True))
self._opt_save_log = bool(cfg("output.targets.print.options.save_log", True))
# 预览是否显示字段标题（配置驱动：优先 output.targets.print.options.preview_labels，
# 回退 print.preview.show_labels；预印合格证可关，仅看值）
self._show_labels = bool(
cfg(
"output.targets.print.options.preview_labels",
cfg("print.preview.show_labels", True),
)
)
# 扫描件生成选项默认值（output.targets.scan.defaults，供「生成扫描件」复选框使用）
self._scan_defaults = dict(cfg("output.targets.scan.defaults", {}) or {})
# ============================================================
# 模板选项（A1/A2/A3：动态填充 + 语言族映射 + 重打恢复）
# ============================================================
# 语言族判定统一收口到 TemplateFieldFormatter.family_of（蓝图 §9-5 清理：删除本类副本）
def _load_template_options(self) -> list[tuple[str, str]]:
"""从 TemplateManager 读取全部模板变体，返回 [(显示标签, 配置键), ...]
显示标签为 ``名称 · 变体``（如「全中文合格证 · 检验5灰度」），
暴露 coordinates.yaml 中全部 9 个变体（含印章差异）。加载失败时回退到
基础 4 语言族，保证 UI 始终可用。
"""
try:
manager = PrinterController.get_template_manager()
options = [
(f"{name} · {variant}" if variant else name, key)
for key, name, variant in manager.list_enabled_templates()
]
if options:
return options
except Exception as exc: # pragma: no cover - 配置缺失时回退
logger.warning(f"[PrintView] 加载模板列表失败，回退默认: {exc}")
return [
("全中文合格证 · 检验5灰度", "full_chinese_1"),
("中英文合格证 · 检5灰度", "bilingual_1"),
("俄英文合格证", "russian"),
("全英文合格证", "full_english"),
]
def _restore_template_by_family(self, family: str) -> None:
"""按语言族恢复下拉选中（A3：重打时选中上次版式所属语言族的首个变体）"""
if not family:
return
for index, (_label, key) in enumerate(self._template_options):
if PrinterController.family_of(key) == family:
self.template_combo.setCurrentIndex(index)
self._current_template_key = key
return
def _init_queue(self) -> None:
"""初始化打印队列控制器与面板（启动时报废未完成任务）"""
self._queue_controller = PrintQueueController(db_manager=self.db_manager)
if self._default_printer and self._default_printer != "默认打印机":
self._queue_controller.set_printer(self._default_printer)
self._queue_panel = QueuePanel(self._queue_controller)
try:
recovered = self._queue_controller.recover()
if recovered:
logger.info(f"打印队列恢复 {recovered} 个未完成任务")
except Exception as e: # noqa: BLE001
logger.warning(f"打印队列恢复失败（可忽略）: {e}")
# H2 事件驱动重载：队列 worker 完成（含参数补充确认写库 + 打印回填）
# 后，按 job 关联的 certificate_id 重载当前表单，保证「选中快照」与库一致。
self._queue_controller.job_finished.connect(self._on_queue_job_finished)
# ============================================================
# 信号连接（集中管理）
# ============================================================
def _connect_signals(self) -> None:
"""集中管理信号连接"""
# 模板切换
self.template_combo.currentIndexChanged.connect(self._on_template_changed)
# 产品信息 - 使用防抖
self.product_name_edit.textChanged.connect(self._on_data_changed)
self.product_model_edit.textChanged.connect(self._on_data_changed)
self.dn_edit.textChanged.connect(self._on_data_changed)
self.pn_edit.textChanged.connect(self._on_data_changed)
self.temperature_edit.textChanged.connect(self._on_data_changed)
self.medium_edit.textChanged.connect(self._on_data_changed)
self.manufacture_date_edit.textChanged.connect(self._on_data_changed)
self.check_standard_edit.textChanged.connect(self._on_data_changed)
self.inspector_id_edit.textChanged.connect(self._on_data_changed)
self.inspector_id_edit.textChanged.connect(self._sync_inspector_combo)
# 编号设置 - 变化需同步刷新编号导航（重置/重算），并防抖刷新预览
self.prefix_edit.textChanged.connect(self._on_number_field_changed)
self.year_month_edit.textChanged.connect(self._on_number_field_changed)
# 目标年月变化：自动续号 + 下一号预览（独立于导航刷新，避免相互干扰）
self.year_month_edit.textChanged.connect(self._on_target_ym_changed)
self.start_number_spin.valueChanged.connect(self._on_number_field_changed)
self.quantity_spin.valueChanged.connect(self._on_number_field_changed)
self.suffix_edit.textChanged.connect(self._on_number_field_changed)
# 编号模式切换
self.number_mode_combo.currentTextChanged.connect(self._on_number_mode_changed)
# H2：活化「自动填写」复选框——作为事件驱动重载的覆盖策略开关。
# 勾选 ⇒ 重载时 DB 值覆盖手改；取消 ⇒ 重载时保留手改（仅填空白）。
# 切换即时按新策略重载当前证书（对齐文档 §10.4-1 H2 接线要求）。
self.auto_fill_check.toggled.connect(self._on_auto_fill_toggled)
# ============================================================
# UI 构建
# ============================================================
[文档]
def setup_ui(self) -> None:
"""构建打印界面
整体：顶部【主操作面：队列传输控制条】+ QSplitter(左控制面板 / 右预览面板)
+ 底部【打印队列表格】。传输控制条（启动/暂停/继续/置顶/上移/取消/清空已完成）
提升到顶部，用户可显式控制队列；左右均包 QScrollArea 兜底防裁切。
"""
layout = QVBoxLayout(self)
layout.setContentsMargins(4, 4, 4, 4)
layout.setSpacing(4)
# —— 主操作面：同级一排 —— 左：配置动作（保存订正/入队/入队启动/生成扫描件，靠前高频）；
# 右：队列传输控制（启动/暂停/继续/置顶/上移/取消/清空已完成）——
layout.addWidget(self._build_top_action_bar())
left_panel = self._build_left_panel()
right_panel = self._build_right_panel()
splitter = QSplitter()
splitter.addWidget(self._wrap_in_scroll(left_panel, "控制面板"))
splitter.addWidget(self._wrap_in_scroll(right_panel, "预览面板"))
splitter.setSizes([520, 420])
layout.addWidget(splitter, stretch=1)
# 队列面板（仅任务表格 + 状态）限高：避免抢走 splitter 空间；
# 内容超出时面板内表格自带滚动条，用户可下拉查看历史任务。
self._queue_panel.setMaximumHeight(180)
layout.addWidget(self._queue_panel)
@staticmethod
def _wrap_in_scroll(widget: QWidget, name: str) -> QWidget:
"""将任意 widget 包裹到 QScrollArea，保证内容溢出时可滚动。"""
scroll = QScrollArea()
scroll.setWidgetResizable(True)
scroll.setFrameShape(QFrame.Shape.NoFrame)
scroll.setObjectName(name)
scroll.setWidget(widget)
return scroll
def _build_left_panel(self) -> QWidget:
"""左侧控制面板（被 QScrollArea 包裹：内容溢出可滚动）"""
panel = QWidget()
layout = QVBoxLayout(panel)
layout.setContentsMargins(4, 4, 4, 4)
layout.setSpacing(6)
# 模板选择
layout.addWidget(self._build_template_group())
# 产品信息
layout.addWidget(self._build_product_group())
# 编号设置
layout.addWidget(self._build_number_group())
# 打印选项（紧凑两列，回到左侧以恢复左右面板均衡）
layout.addWidget(self._build_print_options_group())
layout.addStretch()
return panel
def _build_template_group(self) -> QGroupBox:
"""模板选择组"""
group = QGroupBox("模板")
layout = QVBoxLayout(group)
layout.setContentsMargins(4, 6, 4, 4)
self.template_combo = QComboBox()
for _label, _key in self._template_options:
self.template_combo.addItem(_label, _key)
# 选中配置驱动默认模板（print.default_template，回退首个）
for _i, (_label, _key) in enumerate(self._template_options):
if _key == self._current_template_key:
self.template_combo.setCurrentIndex(_i)
break
layout.addWidget(self.template_combo)
return group
def _build_product_group(self) -> QGroupBox:
"""产品信息组"""
group = QGroupBox("产品信息")
layout = QGridLayout(group)
layout.setContentsMargins(6, 10, 6, 6)
layout.setHorizontalSpacing(8)
fields = [
("产品名称", "product_name_edit", 0, 0),
("产品型号", "product_model_edit", 0, 1),
("公称通径", "dn_edit", 1, 0),
("公称压力", "pn_edit", 1, 1),
("适用温度", "temperature_edit", 2, 0),
("适用介质", "medium_edit", 2, 1),
("出厂年月", "manufacture_date_edit", 3, 0),
("检验标准", "check_standard_edit", 3, 1),
("检验工号", "inspector_id_edit", 4, 0),
]
for label_text, attr_name, row, col in fields:
label = QLabel(label_text)
label.setMinimumWidth(60)
layout.addWidget(label, row, col * 2)
edit = QLineEdit()
setattr(self, attr_name, edit)
layout.addWidget(edit, row, col * 2 + 1)
# 检验工号快速切换下拉（检5 / No 5）：与模板语言族默认同源，
# 提供一个显式控件在两者间一键切换，覆盖自动逻辑（手动输入其他值则下拉置空）。
self.inspector_combo = QComboBox()
self.inspector_combo.addItems(["检5", "No 5"])
self.inspector_combo.setToolTip(
"按模板语言族快速切换检验工号默认（中文→检5 / 其他→No 5）；手动输入其他值后下拉显示为空"
)
self.inspector_combo.setMinimumWidth(60)
self.inspector_combo.currentTextChanged.connect(self._on_inspector_combo_changed)
# inspector_id_edit 在 row 4 col 1；右侧 col 2/3 空闲，放下拉
layout.addWidget(self.inspector_combo, 4, 2)
self.manufacture_date_edit.setText(datetime.now().strftime("%Y.%m"))
# 出厂日期来源选择（配置驱动 + UI 覆盖）：
# 默认勾选态来自 config certificate.issue_date_source（current_ym→勾选）；
# 勾选=强制当前年月；不勾选=配置默认（解析编号年月，加载记录后显示编号年月）。
self.use_current_ym_check = QCheckBox("出厂日期用当前年月")
self.use_current_ym_check.setChecked(
PrinterController.get_issue_date_source() == "current_ym"
)
self.use_current_ym_check.setToolTip(
"勾选则出厂年月强制取当前年月；不勾选则按配置默认（解析编号年月）"
)
self.use_current_ym_check.toggled.connect(self._on_use_current_ym_toggled)
layout.addWidget(self.use_current_ym_check, 5, 0, 1, 4)
# 缓存最近加载记录解析出的编号年月（供取消勾选时回显）
self._last_issue_ym = None
self._apply_issue_date_display()
return group
def _on_use_current_ym_toggled(self, checked: bool) -> None:
"""勾选/取消「出厂日期用当前年月」时刷新显示值。"""
self._apply_issue_date_display()
def _apply_issue_date_display(self) -> None:
"""根据「出厂日期用当前年月」复选框决定出厂年月显示值。"""
if self.use_current_ym_check.isChecked():
self.manufacture_date_edit.setText(datetime.now().strftime("%Y.%m"))
else:
# 配置默认：编号年月解析值；无加载记录时回退今天（加载后会被覆盖）
self.manufacture_date_edit.setText(
self._last_issue_ym or datetime.now().strftime("%Y.%m")
)
def _build_number_group(self) -> QGroupBox:
"""编号设置组（目标年月 / 自动续号 / 起始流水手填覆盖）
目标年月默认当前月；补历史编号填如 ``2604``。自动续号（配置
``certificate.numbering.auto_continue``，默认开）下，「起始」由系统按该月
已用最大流水 + 1 自动填充（框置灰），勾选「起始流水手填覆盖」可手动插入。
"""
group = QGroupBox("编号设置")
layout = QGridLayout(group)
layout.setContentsMargins(6, 10, 6, 6)
# 第 0 行：模式 + 起始流水手填覆盖 + 下一号预览（同排，避免模式独占一行）
layout.addWidget(QLabel("模式"), 0, 0)
self.number_mode_combo = QComboBox()
self.number_mode_combo.addItems(["读取编号", "正则刷新", "直接复制", "三位流水"])
layout.addWidget(self.number_mode_combo, 0, 1, 1, 2)
self._start_override_check = QCheckBox("起始流水手填覆盖")
self._start_override_check.setChecked(not self._auto_continue)
self._start_override_check.setToolTip(
"勾选后可手动填写「起始」流水（特殊插入）；不勾选则按目标年月自动续该月最大流水"
)
self._start_override_check.toggled.connect(self._on_start_override_toggled)
layout.addWidget(self._start_override_check, 0, 3, 1, 2)
self._next_serial_label = QLabel("下一号: —")
self._next_serial_label.setStyleSheet("color: #555;")
layout.addWidget(self._next_serial_label, 0, 5)
# 第 1 行：前缀 + 目标年月 + 起始（标签在偶数列、控件在奇数列）
layout.addWidget(QLabel("前缀"), 1, 0)
self.prefix_edit = QLineEdit("V")
layout.addWidget(self.prefix_edit, 1, 1)
layout.addWidget(QLabel("目标年月"), 1, 2)
self.year_month_edit = QLineEdit(datetime.now().strftime("%y%m"))
self.year_month_edit.setToolTip("编号年月：补历史编号填如 2604；留空/默认当前月")
layout.addWidget(self.year_month_edit, 1, 3)
layout.addWidget(QLabel("起始"), 1, 4)
self.start_number_spin = QSpinBox()
self.start_number_spin.setRange(1, 99999)
self.start_number_spin.setValue(1)
layout.addWidget(self.start_number_spin, 1, 5)
# 第 2 行：数量 + 后缀 + 结束
layout.addWidget(QLabel("数量"), 2, 0)
self.quantity_spin = QSpinBox()
self.quantity_spin.setRange(1, 1000)
self.quantity_spin.setValue(1)
layout.addWidget(self.quantity_spin, 2, 1)
layout.addWidget(QLabel("后缀"), 2, 2)
self.suffix_edit = QLineEdit()
layout.addWidget(self.suffix_edit, 2, 3)
layout.addWidget(QLabel("结束"), 2, 4)
self.end_preview_label = QLabel("—")
self.end_preview_label.setStyleSheet("font-weight: bold; color: #0078d4;")
layout.addWidget(self.end_preview_label, 2, 5)
# 自动续号时「起始」由系统控制（禁用手填），避免与自动值冲突
self.start_number_spin.setEnabled(self._start_override_check.isChecked())
# 初始化预览（当前月下一号）
self._update_next_serial_label()
return group
# ============================================================
# 编号续号（目标年月 / 自动续号 / 手填覆盖）
# ============================================================
def _compose_prefix(self) -> str:
"""由前缀与年月框组合出编号前缀（如 V + 2604 → V2604）。"""
return (self.prefix_edit.text().strip() or "V") + (
self.year_month_edit.text().strip() or ""
)
def _get_cert_controller(self) -> Any:
"""懒加载合格证控制器（编号续号后台支撑）。
优先使用 ``db_manager`` 派生会话（会话生命周期由 Manager 统一管理），
回退裸 ``session``（测试/旧调用方零改动）。
"""
if self._cert_controller is None:
from certflow.controllers.certificate_controller import CertificateController
if self.db_manager is not None:
self._cert_controller = CertificateController(db_manager=self.db_manager)
else:
self._cert_controller = CertificateController(db_session=self.session)
return self._cert_controller
def _update_next_serial_label(self) -> None:
"""只读刷新「下一号」预览（不改动起始）。"""
ym = self.year_month_edit.text().strip()
if len(ym) != 4 or not ym.isdigit():
self._next_serial_label.setText("下一号: —")
return
try:
nxt = self._get_cert_controller().get_next_serial(self._compose_prefix())
except Exception as e: # noqa: BLE001
logger.debug(f"[PrintView] 预瞄下一号失败: {e}")
self._next_serial_label.setText("下一号: —")
return
self._next_serial_label.setText(f"下一号: {nxt:03d}")
def _on_target_ym_changed(self, text: str) -> None:
"""目标年月变化：刷新下一号预览；自动续号模式下把「起始」填为该月下一号。"""
self._update_next_serial_label()
if self._loading_record:
return
if self._start_override_check.isChecked():
return # 手填覆盖：尊重用户手动起始
ym = text.strip()
if len(ym) != 4 or not ym.isdigit():
return
try:
nxt = self._get_cert_controller().get_next_serial(self._compose_prefix())
except Exception as e: # noqa: BLE001
logger.debug(f"[PrintView] 自动续号失败: {e}")
return
self.start_number_spin.setValue(nxt)
# 起始变化需同步刷新编号导航与预览
self._current_serial_index = 0
self._serial_numbers = []
self._update_serial_nav()
self._refresh_timer.start(300)
def _on_start_override_toggled(self, checked: bool) -> None:
"""手填覆盖开关：启用/禁用「起始」自旋框；切回自动时重算下一号并填起始。"""
self.start_number_spin.setEnabled(bool(checked))
if not checked:
self._on_target_ym_changed(self.year_month_edit.text())
def _build_print_options_group(self) -> QGroupBox:
"""打印选项组（紧凑两列 + 扫描件子组按需插入：缓解左侧行高，默认窗口可见按钮区）
配置驱动：复选框默认态从 output.targets.print.options 读取。
扫描件子组默认**不插入**布局（隐藏），仅勾选「生成扫描件」时才插入并启用，
节省默认状态下的垂直空间，让按钮区在默认窗口内直接可见。
"""
group = QGroupBox("选项")
self._print_options_grid = QGridLayout(group)
self._print_options_grid.setContentsMargins(6, 8, 6, 6)
self._print_options_grid.setHorizontalSpacing(10)
self._print_options_grid.setVerticalSpacing(4)
# 自动填写：默认态来自配置；勾选为用户意图开关（_autofill_from_dict 内再判配置总闸）
self.auto_fill_check = QCheckBox("自动填写")
self.auto_fill_check.setChecked(self._opt_auto_fill)
self._print_options_grid.addWidget(self.auto_fill_check, 0, 0)
# 保存日志：默认态来自配置；_get_form_data 据此决定是否写回 PrintLog
self.save_to_db_check = QCheckBox("保存日志")
self.save_to_db_check.setChecked(self._opt_save_log)
self._print_options_grid.addWidget(self.save_to_db_check, 0, 1)
self.show_labels_check = QCheckBox("预览显示标题")
self.show_labels_check.setChecked(self._show_labels)
self.show_labels_check.setToolTip(
"取消勾选则预览仅显示字段值（适配预印合格证：标题已由印刷底板提供，只填值）"
)
self.show_labels_check.toggled.connect(self._on_show_labels_toggled)
self._print_options_grid.addWidget(self.show_labels_check, 1, 0)
# —— 生成扫描件：勾选后「入队启动」会在打印同时，按下方选项生成图片/PDF 扫描件（配置驱动）——
self.generate_scan_check = QCheckBox("生成扫描件")
self.generate_scan_check.setChecked(False)
self.generate_scan_check.setToolTip(
"勾选后，点击「入队启动」会在打印的同时，按下方扫描件选项生成扫描件（后台并行，不阻塞打印）"
)
self.generate_scan_check.toggled.connect(self._on_generate_scan_toggled)
self._print_options_grid.addWidget(self.generate_scan_check, 1, 1)
# 扫描件选项子组（紧凑一行：模板·格式·每行·每列；默认隐藏，按需插入布局）
scan_box = QGroupBox("扫描件选项")
sgrid = QGridLayout(scan_box)
sgrid.setContentsMargins(6, 8, 6, 6)
sgrid.setHorizontalSpacing(8)
sgrid.setVerticalSpacing(4)
defaults = self._scan_defaults
self._scan_template_combo = QComboBox()
self._scan_template_combo.addItems(["全中文", "中英文", "俄英文", "全英文"])
self._scan_template_combo.setCurrentText(defaults.get("template_type", "全中文"))
sgrid.addWidget(QLabel("模板"), 0, 0)
sgrid.addWidget(self._scan_template_combo, 0, 1)
self._scan_format_combo = QComboBox()
self._scan_format_combo.addItems(["JPG", "PDF"])
self._scan_format_combo.setCurrentText(defaults.get("format", "JPG"))
sgrid.addWidget(QLabel("格式"), 0, 2)
sgrid.addWidget(self._scan_format_combo, 0, 3)
self._scan_rows_spin = QSpinBox()
self._scan_rows_spin.setRange(1, 10)
self._scan_rows_spin.setValue(int(defaults.get("rows", 1)))
sgrid.addWidget(QLabel("每行"), 0, 4)
sgrid.addWidget(self._scan_rows_spin, 0, 5)
self._scan_cols_spin = QSpinBox()
self._scan_cols_spin.setRange(1, 10)
self._scan_cols_spin.setValue(int(defaults.get("cols", 1)))
sgrid.addWidget(QLabel("每列"), 0, 6)
sgrid.addWidget(self._scan_cols_spin, 0, 7)
self._scan_options_box = scan_box
scan_box.setEnabled(False)
scan_box.hide() # 默认隐藏；勾选「生成扫描件」时由 _on_generate_scan_toggled 插入布局
return group
def _on_generate_scan_toggled(self, checked: bool) -> None:
"""勾选「生成扫描件」时插入扫描件选项子组；取消时移除（省垂直空间）"""
if checked:
self._scan_options_box.show()
self._scan_options_box.setEnabled(True)
# 已插入则跳过
if self._scan_options_box.parentWidget() is None:
self._print_options_grid.addWidget(self._scan_options_box, 2, 0, 1, 2)
else:
self._print_options_grid.removeWidget(self._scan_options_box)
self._scan_options_box.setEnabled(False)
self._scan_options_box.hide()
def _build_top_action_bar(self) -> QWidget:
"""顶部主操作栏：同级一排，左配置动作（靠前）+ 右队列传输控制。
- 左：保存订正 / 入队 / 入队启动 / 生成扫描件（高频配置动作，置前）
- 右：队列传输控制条（启动/暂停/继续/置顶/上移/取消/清空已完成，
由 QueuePanel 提供，提升到此与主操作并列）
两组用 stretch 隔开，配置动作靠左、队列控制靠右，层级清晰。
"""
bar = QWidget()
layout = QHBoxLayout(bar)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(6)
# 左：配置动作（靠前）
self._build_config_action_buttons(layout)
# 隔开两组
layout.addStretch(1)
# 右：队列传输控制
layout.addWidget(self._queue_panel.transport_bar)
return bar
def _build_config_action_buttons(self, layout: QHBoxLayout) -> None:
"""构建高频配置动作按钮（保存订正 / 入队 / 入队启动 / 生成扫描件）并入指定布局。
流水线：核对补录 → 保存订正 → 入队（/入队启动）。按 momo 在 ISSUE #2 的要求：
- 「保存订正」把补录内容写回数据库（Certificate）；
- 「入队」先校验无缺失 → 保存订正 → 从数据库重新读取最新内容 → 入队到
print_jobs，不立即启动，交由队列面板显式控制打印节奏（批量入队后统一排序打印）；
- 「入队启动」是「入队」的快捷方式：入队后立即启动队列 worker（立刻打这一张）；
- 「生成扫描件」跳转扫描件视图，传入当前已加载合格证 ID。
移除旧版「单次/批量」直发路径——它绕过队列、用内存表单直接发打印，且与队列
worker 抢锁引发 database is locked。
"""
buttons = [
("保存订正", "#0078d4", self._on_save_correction),
("入队", "#17a2b8", self._on_enqueue),
("入队启动", "#28a745", self._on_enqueue_and_start),
("生成扫描件", "#ff9800", self._on_navigate_to_scan),
("保存图片", "#6c757d", self._save_preview_image),
]
for text, color, slot in buttons:
button = QPushButton(text)
button.setMinimumHeight(30)
button.setStyleSheet(f"""
QPushButton {{
background-color: {color};
color: white;
font-weight: bold;
border-radius: 4px;
}}
QPushButton:hover {{ background-color: {color}cc; }}
QPushButton:disabled {{ background-color: #aaa; color: #eee; }}
""")
button.clicked.connect(slot)
layout.addWidget(button)
def _get_loaded_certificate_ids(self) -> list[int]:
"""返回本次已加载的合格证 ID 列表（过滤 None）。
供「生成扫描件」按钮跳转时把当前视图里的合格证传入扫描件视图，
实现「打印视图 → 扫描件生成」一键衔接（scan_view 提示「请从打印页面跳转」）。
"""
return [c.id for c in self._loaded_certs if c]
def _on_navigate_to_scan(self) -> None:
"""跳转扫描件生成：把当前已加载的合格证 ID 传入扫描件视图。
服务端 ScanService / ScanController / ScanView 已完整实现，这里仅做控制器
接线——通过 navigate_to_scan 信号通知主窗口切换到扫描件视图并自动加载证书。
"""
cert_ids = self._get_loaded_certificate_ids()
if not cert_ids:
QMessageBox.warning(self, "提示", "尚未加载合格证，请先从查询清单选择编号")
return
self.navigate_to_scan.emit(cert_ids)
def _build_right_panel(self) -> QWidget:
"""右侧预览面板"""
panel = QWidget()
panel.setMaximumWidth(420)
lyt = QVBoxLayout(panel)
lyt.setContentsMargins(4, 4, 4, 4)
# 预览头部
header = QHBoxLayout()
self.prev_btn = QPushButton("◀")
self.prev_btn.setFixedSize(28, 26)
self.prev_btn.clicked.connect(self._prev_serial)
header.addWidget(self.prev_btn)
self.serial_label = QLabel("0 / 0")
self.serial_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
self.serial_label.setFixedWidth(50)
header.addWidget(self.serial_label)
self.next_btn = QPushButton("▶")
self.next_btn.setFixedSize(28, 26)
self.next_btn.clicked.connect(self._next_serial)
header.addWidget(self.next_btn)
header.addStretch()
lyt.addLayout(header)
# 预览（单一 QLabel，由 QPainter 绘制合格证位图，C2 去 Chromium）
self.preview_label = QLabel()
self.preview_label.setFixedWidth(280)
self.preview_label.setFixedHeight(467) # 280 / 60 * 100
self.preview_label.setStyleSheet("border: 1px solid #ccc; background: white;")
self.preview_label.setAlignment(Qt.AlignmentFlag.AlignTop | Qt.AlignmentFlag.AlignLeft)
lyt.addWidget(self.preview_label, stretch=1)
# 选项组已移回左侧（保持左右面板视觉均衡；此处不再添加）
# 状态标签
self.status_label = QLabel("就绪")
self.status_label.setStyleSheet("color: #666; font-size: 11px;")
self.status_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
lyt.addWidget(self.status_label)
# 进度条
self.progress_bar = QProgressBar()
self.progress_bar.setVisible(False)
lyt.addWidget(self.progress_bar)
return panel
# ============================================================
# 打印操作
# ============================================================
# 流水线：核对补录 → 保存订正 → 从 DB 重读 → 入队启动
# ============================================================
# 打印前必须补齐的表单控件（key = 控件属性名，label = 中文名）
# 注：出厂年月控件在初始化时已预填当天日期（_%Y.%m），属"非空白"，
# 不计入"遗漏的空白控件"；如需强制改填可在界面手动覆盖。
_REQUIRED_FIELDS: tuple[tuple[str, str], ...] = (
("product_name_edit", "产品名称"),
("product_model_edit", "产品型号"),
("dn_edit", "公称通径"),
("pn_edit", "公称压力"),
("temperature_edit", "适用温度"),
("medium_edit", "适用介质"),
("check_standard_edit", "检验标准"),
("inspector_id_edit", "检验工号"),
)
def _check_missing_fields(self) -> list[str]:
"""校验打印前必填控件是否有空白。
Returns:
list[str]: 缺失字段的中文名列表；为空表示全部已填。
"""
missing: list[str] = []
for attr, label in self._REQUIRED_FIELDS:
edit = getattr(self, attr, None)
if edit is None or not edit.text().strip():
missing.append(label)
return missing
def _on_save_correction(self) -> None:
"""保存订正：把当前表单控件的内容写回数据库（Certificate）。
这是打印前的强制前置步骤——只有把补录内容落库，队列打印时
才能「从数据库读取最新的、无缺失的打印内容」。
"""
if self._current_certificate_id is None:
QMessageBox.warning(self, "提示", "尚未加载合格证，请先从查询清单选择编号")
return
# 先校验，避免把残缺数据写回库
missing = self._check_missing_fields()
if missing:
QMessageBox.warning(
self,
"信息缺失",
"以下必填项仍为空白，请补充录入后再保存订正：\n\n" + "、".join(missing),
)
return
# 模板族压力单位门限（中文放行不门限；中英文/全英文/俄英文必须带单位，缺则拦截）
pn_err = PrinterController.validate_pn_text(
PrinterController.family_of(self._current_template_key),
self._get_form_data().get("pn", ""),
)
if pn_err:
QMessageBox.warning(self, "压力单位不符模板", pn_err)
return
try:
self._write_correction_to_db()
self.status_label.setText("已保存订正（已写回数据库）")
QMessageBox.information(self, "已保存", "订正内容已写回数据库，可入队启动打印")
except Exception as e: # noqa: BLE001
logger.error(f"[PrintView] 保存订正失败: {e}")
QMessageBox.critical(self, "保存失败", f"写回数据库失败：\n{e}")
def _write_correction_to_db(self) -> None:
"""把表单控件写回 Certificate（含编号组），经 CertificateController 收口。
编号范围串按当前 prefix/year_month/start/quantity/suffix 重组，
同时回填 product_code_* 拆分字段，保证队列打印展开编号时数据完整。
视图不再直连 session / ORM，统一委托 CertificateController.save_correction。
"""
data = self._get_form_data()
self._get_cert_controller().save_correction(
self._current_certificate_id, data, self._current_template_key
)
def _prepare_and_enqueue(self) -> str | None:
"""入队前置流水线（「入队」与「入队启动」共用）：
1. 检查打印机 (_check_printer，main 护栏能力，入队前必检)；
2. 字典自动回填 (_autofill_from_dict，print.guard 链)；
3. 缺失字段交互补录 (_validate_and_prompt_fields，print.guard 链)；
4. 学习回写字典 (_learn_back，print.guard 链)；
5. 保存订正（写回数据库）；
6. **从数据库重新读取最新 Certificate**（确保信息无缺失、与表单一致）；
若 print.guard 未启用或补录后仍缺失，由 _check_missing_fields 作为最终安全网拦截；
7. 入队到 print_jobs（**不启动**，启动与否由调用方决定）。
队列 worker 内部再按 certificate_id 从数据库读取 Certificate 构建打印
数据——最终送上打印机的一定是「数据库里最新、无缺失」的内容，
而不是内存表单里可能残缺的值。
Returns:
str | None: 成功返回 SN 串；任意一步失败（已弹框提示）返回 None。
"""
if self._current_certificate_id is None:
QMessageBox.warning(self, "提示", "尚未加载合格证，请先从查询清单选择编号")
return None
cert = self._get_cert_controller().get_certificate_by_id(self._current_certificate_id)
if cert is None:
QMessageBox.critical(self, "错误", "合格证记录不存在")
return None
# ============================================================
# 步骤1: 检查打印机 (main 护栏能力，入队前必检，避免无效入队)
# 仅当显式配置了指定打印机时才做严格匹配；未配置（默认打印机占位）则
# 跳过严格检查，避免无配置环境下误拦截入队（与旧流水线行为一致）。
# ============================================================
if (
self._default_printer
and self._default_printer != "默认打印机"
and self._check_printer() is None
):
# _check_printer 内部已弹框提示（CI 下被桩掉）
return None
# ============================================================
# 步骤2: 字典自动回填 (_autofill_from_dict，print.guard 链)
# 从 model_param_mappings 回填空白控件（已填的人工值保留）
# ============================================================
self._autofill_from_dict(cert)
# ============================================================
# 步骤3: 缺失字段交互补录 (_validate_and_prompt_fields，print.guard 链)
# 该方法依赖当前表单数据；先构建 data，再交给护栏弹框补录
# （print.guard 未启用时为 no-op，由下方 _check_missing_fields 兜底）
# ============================================================
data = self._get_form_data()
if not self._validate_and_prompt_fields(data):
# 用户取消或补录失败 → 终止流水线（绝不发空白指令）
return None
# 补录可能已写回控件，重读表单数据（并同步控件 → data）
self._sync_form_fields_from_data(data)
# 检验工号空值兜底：入队前若仍空白，按模板语言族预填默认工号（配置驱动）
self._apply_inspector_default()
# ============================================================
# 步骤4: 学习回写字典 (_learn_back，print.guard 链)
# 将用户补录的值写回 model_param_mappings，下次同一型号自动填充
# ============================================================
self._learn_back(cert, data)
# 模板族压力单位门限（中文放行不门限；中英文/全英文/俄英文必须带单位，缺则拦截）
pn_err = PrinterController.validate_pn_text(
PrinterController.family_of(self._current_template_key), (data or {}).get("pn", "")
)
if pn_err:
QMessageBox.warning(self, "压力单位不符模板", pn_err)
return None
# ============================================================
# 步骤5: 保存订正（把补录写回 DB）
# ============================================================
try:
self._write_correction_to_db()
except Exception as e: # noqa: BLE001
logger.error(f"[PrintView] 入队前保存订正失败: {e}")
QMessageBox.critical(self, "保存失败", f"写回数据库失败：\n{e}")
return None
# ============================================================
# 步骤6: 从数据库重新读取最新 Certificate（信息无缺失）
# ============================================================
fresh = self._get_cert_controller().get_certificate_by_id(self._current_certificate_id)
if fresh is None:
QMessageBox.critical(self, "读取失败", "无法从数据库读取最新合格证内容")
return None
# 入队前最后一道闸门：DB 里的关键字段必须无缺失，否则不让启动打印
# （保留 _check_missing_fields 作为最终安全网——覆盖 print.guard 未启用
# 或补录后仍缺失的情况，避免把残缺信息直接送上打印机，这正是旧流程的问题）
db_missing = self._check_missing_fields()
if db_missing:
QMessageBox.critical(
self,
"数据库内容仍缺失",
"已写回数据库，但以下字段在库中仍为空，请重新核对补录：\n\n"
+ "、".join(db_missing),
)
return None
serial = fresh.product_code_range or fresh.certificate_no or ""
self.status_label.setText(f"已从数据库读取最新内容，准备入队：{serial}")
# ============================================================
# 🎯 入队前抓取 raw 打印字节（可选，配置 print.raw_capture.enabled 控制）
# 此时数据已通过护栏补录、保存订正、DB 重读，是「最终要打印的完整数据」。
# 复用真实渲染链路生成 ESC/P-K 字节但不真正出纸，便于落盘 / 十六进制分析。
# 抓取失败不影响入队（内部 try/except 已包裹）。
# ============================================================
self._capture_raw_if_enabled()
# ============================================================
# 步骤7: 入队（不启动）
# ============================================================
try:
# replace_pending=True：入队前取消该证已有 pending，避免重复点击
# "入队启动"或"订正后再入队"产生同一证多份 pending 导致重复打印。
self._queue_controller.enqueue(
certificate_id=self._current_certificate_id,
sn=serial,
priority=0,
replace_pending=True,
)
except Exception as e: # noqa: BLE001
logger.error(f"[PrintView] 入队失败: {e}")
QMessageBox.critical(self, "入队失败", f"入队失败：\n{e}")
return None
return serial
# ============================================================
# 入队前 raw 打印字节抓取（调试/分析用，配置驱动，不影响入队）
# ============================================================
def _capture_raw_if_enabled(self) -> None:
"""入队前可选抓取 raw 字节（配置 print.raw_capture.enabled 控制）。"""
if not self._should_capture_raw():
return
raw_bytes, raw_data = self._capture_raw_print_data()
if raw_bytes:
self._save_raw_binary(raw_bytes, raw_data)
def _should_capture_raw(self) -> bool:
"""是否启用入队前 raw 字节抓取。
优先级：userconfig 覆盖层 print_raw_capture.enabled > 主配置 print.raw_capture.enabled；
两者皆缺省时回退 True（初期调试默认开启，后期可在 UI 或配置关闭）。
"""
from certflow.config.settings import get_override
ov = get_override("print_raw_capture", "enabled", None)
if ov is not None:
return bool(ov)
return bool(cfg("print.raw_capture.enabled", True))
def _capture_raw_print_data(self) -> tuple[bytes | None, dict | None]:
"""生成即将发送给打印机的 ESC/P-K 原始字节流（不真正出纸）。
复用真实打印渲染链路（PrinterService.render_certificate_bytes），
在当前「最终完整数据」上渲染，得到与打印启动后完全一致的字节；
同时返回结构化 print_data 供落盘 JSON 便于人工分析。
Returns:
(bytes | None, dict | None): 字节流与结构化数据；失败返回 (None, None)。
"""
try:
data = self._get_form_data()
serial = self._get_current_serial()
print_data = self._format_print_data(data)
print_data["serial_number"] = serial
print_data["cert_number"] = serial
config = self._build_print_config()
if config is None:
return None, None
printer_name = self._default_printer
if not printer_name or printer_name == "默认打印机":
printer_name = None
raw = self._get_cert_controller().render_certificate_bytes(
config,
print_data,
copies=1,
x_offsets=print_data.get("_x_offsets"),
printer_name=printer_name,
)
return raw, print_data
except Exception as e: # noqa: BLE001
logger.error(f"[PrintView] 抓取 raw 打印数据失败: {e}")
return None, None
def _save_raw_binary(self, data: bytes, print_data: dict | None = None) -> str | None:
"""保存 raw 字节（.prn）及结构化 JSON 到目录（配置 print.raw_capture.output_dir）。
Returns:
str | None: 保存的文件路径；失败返回 None。
"""
try:
serial = self._get_current_serial() or "unknown"
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
output_dir = cfg("print.raw_capture.output_dir", "raw_print")
os.makedirs(output_dir, exist_ok=True)
filename = f"raw_print_{serial}_{timestamp}.prn"
filepath = os.path.join(output_dir, filename)
with open(filepath, "wb") as f:
f.write(data)
logger.info(f"[PrintView] Raw 二进制已保存: {filepath}")
if cfg("print.raw_capture.save_json", True) and print_data is not None:
jpath = os.path.join(output_dir, f"raw_print_{serial}_{timestamp}.json")
with open(jpath, "w", encoding="utf-8") as jf:
json.dump(print_data, jf, ensure_ascii=False, indent=2)
self.status_label.setText(f"Raw 已保存: {filename} ({len(data)} bytes)")
return filepath
except Exception as e: # noqa: BLE001
logger.error(f"[PrintView] 保存 raw 二进制失败: {e}")
return None
def _on_enqueue(self) -> None:
"""独立「入队」：仅入队到 print_jobs，不立即启动。
适合批量入队后统一在「打印队列」面板点「启动」控制打印节奏；
入队后队列面板的 启动/置顶/上移 等按钮恢复为有效主操作。
"""
serial = self._prepare_and_enqueue()
if serial is None:
return
if self._queue_panel is not None:
self._queue_panel.status_label.setText("已入队，待打印——请在队列面板点「启动」")
QMessageBox.information(
self,
"已入队",
f"合格证已入队（待打印）：\n{serial}\n\n"
"请在下方「打印队列」面板点击「启动」开始打印；\n"
"可多张入队后统一排序（置顶/上移）再启动。",
)
def _on_enqueue_and_start(self) -> None:
"""入队启动（快捷方式）：入队后立即启动队列 worker。
校验/补录/订正/从 DB 重读 与「入队」共用 _prepare_and_enqueue，
区别仅在入队后立即 start()（适合「立刻打这一张」）。
"""
serial = self._prepare_and_enqueue()
if serial is None:
return
try:
self._queue_controller.start()
# 若勾选「生成扫描件」：同一批合格证在后台一并生成扫描件（不阻塞打印）
if self.generate_scan_check.isChecked():
self._trigger_scan_generation()
QMessageBox.information(
self,
"已入队",
f"合格证已入队并启动打印：\n{serial}\n\n"
"队列将从数据库读取最新内容（无缺失）后逐台打印。",
)
except Exception as e: # noqa: BLE001
logger.error(f"[PrintView] 入队启动失败: {e}")
QMessageBox.critical(self, "入队失败", f"入队/启动失败：\n{e}")
# ============================================================
# 勾选「生成扫描件」时：入队启动打印的同时，在后台一并生成扫描件
# 选项来自 output.targets.scan.defaults（配置驱动），UI 内可改。
# ============================================================
def _trigger_scan_generation(self) -> None:
"""后台触发扫描件生成（与打印并行，不阻塞 UI）。
复用 ScanController / ScanService；选项取 UI 当前值，回退到
``output.targets.scan.defaults`` 配置默认。双编码开关仍走配置
``output.targets.scan.dual_code.enabled``（默认 false，长期不用）。
"""
cert_ids = self._get_loaded_certificate_ids()
if not cert_ids:
return
try:
defaults = self._scan_defaults
template_type = self._scan_template_combo.currentText() or defaults.get(
"template_type", "全中文"
)
output_format = self._scan_format_combo.currentText() or defaults.get("format", "JPG")
rows = self._scan_rows_spin.value()
cols = self._scan_cols_spin.value()
output_dir = defaults.get("output_dir") or cfg("paths.output.scans", "output/scans")
from certflow.controllers.scan_controller import ScanController
if self._scan_controller is None:
if self.db_manager is not None:
self._scan_controller = ScanController(db_manager=self.db_manager)
else:
self._scan_controller = ScanController(self.session)
self._scan_controller.completed.connect(self._on_scan_completed)
self._scan_controller.error.connect(self._on_scan_error)
self._scan_controller.generate_scans(
certificate_ids=cert_ids,
output_dir=output_dir,
template_type=template_type,
output_format=output_format,
rows=rows,
cols=cols,
dual_code=bool(cfg("output.targets.scan.dual_code.enabled", False)),
)
self.status_label.setText(f"已触发扫描件生成（后台）：{len(cert_ids)} 份合格证")
except Exception as e: # noqa: BLE001
logger.error(f"[PrintView] 触发扫描件生成失败: {e}")
QMessageBox.warning(self, "扫描件生成失败", f"触发扫描件生成失败：\n{e}")
def _on_scan_completed(self, result: dict) -> None:
"""扫描件生成完成回调。"""
ok = result.get("success", 0)
failed = result.get("failed", 0)
msg = result.get("message", "")
self.status_label.setText(f"扫描件生成完成：成功 {ok} / 失败 {failed}；{msg}")
def _on_scan_error(self, err: str) -> None:
"""扫描件生成错误回调。"""
self.status_label.setText(f"扫描件生成失败：{err}")
logger.error(f"[PrintView] 扫描件生成错误: {err}")
def _sync_form_fields_from_data(self, data: dict) -> None:
"""把护栏补录后的 data 回写到表单控件，保证控件与 data 一致。
_validate_and_prompt_fields 已通过 _set_form_field 写回控件，这里做一次
幂等同步，避免控件与内存 data 不一致（尤其 CI 下 QInputDialog 被桩为
返回空 ok 的场景）。
"""
for key in (
"product_name",
"product_model",
"dn",
"pn",
"temperature",
"medium",
"check_standard",
"inspector_id",
):
value = (data.get(key) or "").strip()
widget = {
"product_name": self.product_name_edit,
"product_model": self.product_model_edit,
"dn": self.dn_edit,
"pn": self.pn_edit,
"temperature": self.temperature_edit,
"medium": self.medium_edit,
"check_standard": self.check_standard_edit,
"inspector_id": self.inspector_id_edit,
}.get(key)
if widget is not None and value and not widget.text().strip():
widget.setText(value)
def _do_single_print(self) -> None:
"""单次打印（防重复）。
说明：本 PR 默认 GUI 入口为「入队启动」队列流水线（见 _on_enqueue_and_start，
对应 ISSUE #2：从 DB 重读无缺失再打印、规避写锁）。本直发方法保留为底层
能力，供 main 的 print-guard 端到端测试与内部调用使用，二者互不冲突。
"""
if self._is_printing:
logger.warning("打印进行中，忽略重复调用")
return
self._is_printing = True
try:
data = self._get_form_data()
# ② ③ 发送前护栏（单张：multi=False）
ok, data = self._guard_before_print(data, multi=False)
if not ok:
QMessageBox.warning(self, "已中止", "打印已取消：用户中止了护栏检查。")
return
serial = self._get_current_serial()
print_data = self._format_print_data(data)
print_data["serial_number"] = serial
# 使用 RuntimePrintConfig
config = self._build_print_config()
if config is None:
return
config.copies = 1
controller = self._get_printer_controller()
controller.print_certificate(print_data, copies=1, config=config)
self._advance_start_after_print(1)
if data.get("save_to_db", True):
self._auto_save([serial], data)
# ④ 成功打印后学回字典（保守合并）
self._learn_back(self._current_cert, data)
finally:
self._is_printing = False
def _do_batch_print(self) -> None:
"""批量打印（防重复）
加载了多条销售计划时，依次打印全部计划（每条按各自整批区间）；
仅加载一条时保持原有单计划交互（先发第一张再确认）。
"""
if self._is_printing:
logger.warning("打印进行中，忽略重复调用")
return
if len(self._loaded_plans) > 1:
self._do_batch_print_all()
return
self._update_serial_nav()
serials = self._serial_numbers
if not serials:
QMessageBox.warning(self, "提示", "无可打印的编号")
return
if len(serials) == 1:
self._do_single_print()
return
self._is_printing = True
try:
# 先打印第一张预览
data = self._get_form_data()
# ② ③ 发送前护栏（单计划批量：multi=False）
ok, data = self._guard_before_print(data, multi=False)
if not ok:
QMessageBox.warning(self, "已中止", "打印已取消：用户中止了护栏检查。")
return
print_data = self._format_print_data(data)
print_data["serial_number"] = serials[0]
config = self._build_print_config()
if config is None:
return
config.copies = 1
controller = self._get_printer_controller()
controller.print_certificate(print_data, copies=1, config=config)
reply = QMessageBox.question(
self,
"确认批量打印",
f"已发送第一张打印: {serials[0]}\n\n剩余 {len(serials) - 1} 张，是否继续？",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
)
if reply != QMessageBox.StandardButton.Yes:
self.status_label.setText("批量打印已取消")
return
self._emit_remaining(serials, data, config, controller)
QMessageBox.information(self, "完成", f"批量打印完成，共 {len(serials)} 张")
self._advance_start_after_print(len(serials))
if data.get("save_to_db", True):
self._auto_save(serials, data)
# ④ 成功打印后学回字典（保守合并）
self._learn_back(self._current_cert, data)
finally:
self._is_printing = False
def _emit_remaining(self, serials: list[str], data: dict, config: Any, controller: Any) -> None:
"""发送 serials[1:] 的剩余编号到打印机（共享首张已建好的 config）。"""
self.progress_bar.setVisible(True)
self.progress_bar.setMaximum(max(1, len(serials) - 1))
self.progress_bar.setValue(0)
for i, serial in enumerate(serials[1:], 1):
batch_data = self._format_print_data(data)
batch_data["serial_number"] = serial
controller.print_certificate(batch_data, copies=1, config=config)
self.progress_bar.setValue(i)
self.progress_bar.setVisible(False)
def _do_batch_print_all(self) -> None:
"""多销售计划批量打印：依次逐条打印全部计划（每条按各自整批区间）。"""
if self._is_printing:
return
plans = list(zip(self._loaded_plans, self._loaded_certs, strict=True))
if not plans:
return
total_units = sum(self._plan_quantity(sp, cert) for sp, cert in plans)
reply = QMessageBox.question(
self,
"确认批量打印",
f"将对 {len(plans)} 条记录共 {total_units} 台合格证依次打印，是否继续？",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
)
if reply != QMessageBox.StandardButton.Yes:
self.status_label.setText("批量打印已取消")
return
config = self._build_print_config()
if config is None:
return
config.copies = 1
controller = self._get_printer_controller()
self._is_printing = True
try:
printed = 0
for plan, cert in plans:
# 绑定该计划到表单（产品信息 + 编号设置），使打印数据正确
self._bind_active_record(plan, cert)
self._current_serial_index = 0
self._serial_numbers = []
self._update_serial_nav()
serials = self._serial_numbers
if not serials:
continue
data = self._get_form_data()
# ② ③ 发送前护栏（多计划批量每批：multi=True）
ok, data = self._guard_before_print(data, multi=True)
if not ok:
self.status_label.setText("批量打印已中止")
break
print_data = self._format_print_data(data)
print_data["serial_number"] = serials[0]
controller.print_certificate(print_data, copies=1, config=config)
self._emit_remaining(serials, data, config, controller)
printed += len(serials)
if data.get("save_to_db", True):
self._auto_save(serials, data)
# ④ 成功打印后学回字典（保守合并）
self._learn_back(cert, data)
QMessageBox.information(self, "完成", f"批量打印完成，共 {printed} 张")
self.status_label.setText(f"批量打印完成，共 {printed} 张")
finally:
self._is_printing = False
# ============================================================
# 打印防错护栏：① 自动带出 / ② 必填校验+InputBox / ③ 确认闸口 / ④ 字典学回
# 设计见 docs/PRINT_BLUEPRINT.md
# ============================================================
# ---- 配置与字典查询 ----
def _guard_enabled(self) -> bool:
"""print.guard 段已配置（存在即启用护栏）。"""
return cfg("print.guard", None) is not None
def _get_mpm_controller(self) -> ModelParamMappingController:
if self._mpm_controller is None:
if self.db_manager is not None:
self._mpm_controller = ModelParamMappingController(db_manager=self.db_manager)
else:
self._mpm_controller = ModelParamMappingController(db_session=self.session)
return self._mpm_controller
def _lookup_model_param(self, product_model: str) -> Any:
"""按型号查 model_param_mappings，返回记录或 None。"""
if not product_model:
return None
try:
ctrl = self._get_mpm_controller()
return ctrl.get_by_model(product_model)
except Exception as exc: # noqa: BLE001
logger.warning(f"[PrintView] 查询型号字典失败({product_model}): {exc}")
return None
# ---- ① 加载自动带出 ----
def _apply_autofill_field(self, widget: Any, db_col: str, rec: Any) -> None:
"""按型号字典预填单个字段（仅在控件为空时写入，人工优先）。"""
if widget.text().strip():
return # 已有值不覆盖（人工优先）
val = getattr(rec, db_col, None) or ""
if val:
widget.setText(str(val))
def _validate_and_prompt_fields(self, data: dict) -> bool:
"""③ 缺失字段交互补录（print.guard 链 — 仅必填项补录）。
入队流水线专用：只做必填字段补录，不含差异确认与确认闸口
（差异确认和闸口在发送前护栏 _guard_before_print 中处理）。
Returns:
True 表示补录完成（或无缺失）；False 表示用户取消。
"""
if not self._guard_enabled():
return True
guard = cfg("print.guard", {}) or {}
required_fields = guard.get("required_fields", []) or []
if not required_fields:
return True
for f in required_fields:
key = f.get("key")
if not key:
continue
required = f.get("required", True)
cur = (data.get(key) or "").strip()
if cur or not required:
continue
# 弹框补录
label = self._field_label(key)
text, ok = QInputDialog.getText(
self, "补全必填字段", f"请填写{label}：", QLineEdit.Normal, ""
)
if not ok:
return False
text = (text or "").strip()
if not text:
return False
data[key] = text
self._set_form_field(key, text)
return True
# 字段 key → 中文标签映射（供必填补录弹框展示，§print.guard）
_FIELD_LABELS: dict[str, str] = {
"product_name": "产品名称",
"product_model": "产品型号",
"dn": "公称通径",
"pn": "公称压力",
"temperature": "工作温度",
"medium": "适用介质",
"check_standard": "检验标准",
"inspector_id": "检验员工号",
}
def _autofill_from_dict(self, cert: Any) -> None:
"""_fill_product_fields 之后调用：表单字段为空时按型号预填字典建议值。"""
if not self._guard_enabled():
return
if not cfg("print.guard.autofill.enabled", False):
return
# 「自动填写」复选框为用户意图开关：配置总闸开启且用户勾选才真正预填
if not self.auto_fill_check.isChecked():
return
fields_map = cfg("print.guard.autofill.fields", {}) or {}
if not fields_map:
return
model = (getattr(cert, "product_model", None) or "").strip()
if not model:
return
rec = self._lookup_model_param(model)
if rec is None:
return
widget_for = {
"temperature": self.temperature_edit,
"medium": self.medium_edit,
"check_standard": self.check_standard_edit,
"pn": self.pn_edit,
"dn": self.dn_edit,
}
for form_field, db_col in fields_map.items():
widget = widget_for.get(form_field)
if widget is None:
continue
self._apply_autofill_field(widget, db_col, rec)
def _apply_inspector_default(self) -> None:
"""检验工号空值兜底（配置驱动 print.guard.autofill.inspector_default）。
表单补充/绑定后若检验工号仍空白，按模板语言族填入默认工号：
- 中文合格证（全中文模板）→ chinese 值（默认「检5」）
- 其他合格证（中/俄/全英文模板）→ other 值（默认「No 5」）
与字典自动带出同口径：受「自动填写」复选框 + print.guard.autofill.enabled 总闸约束；
人工已填则保留（不覆盖）。配置缺失或总闸关闭时不做任何动作。
"""
if not self.auto_fill_check.isChecked():
return
if not cfg("print.guard.autofill.enabled", False):
return
inspector_default = cfg("print.guard.autofill.inspector_default", None)
if not inspector_default:
return
if self.inspector_id_edit.text().strip():
return # 人工已填，优先
family = PrinterController.family_of(self._current_template_key)
default = cfg.get("chinese" if family == "全中文" else "other")
if not default:
return
self.inspector_id_edit.setText(str(default))
self._sync_inspector_combo()
def _sync_inspector_for_template_change(self, old_family: str) -> None:
"""模板语言切换时同步检验工号默认值（与生成期烘焙同源：``_default_inspector_id``）。
需求：切换模板下拉（不同语言）时，检验工号要跟随语言：
- 中文合格证（全中文）→「检5」，其他（中/俄/全英文）→「No 5」。
仅当当前检验工号等于「旧语言的默认工号」（即此前是自动带出、非人工手填）
才换成新语言默认值；人工手填了其他值则保留不动，避免误覆盖。
当前为空也按新语言默认补填。
注意：此联动是「模板语言族属性」，与「字典自动带出（自动填写复选框 /
print.guard.autofill.enabled）」无关，故不受其约束——否则关掉自动填写
后切换模板，检5/No5 不会刷新（原实现的坑）。统一复用生成期同款
``_default_inspector_id``，保证加载/生成/切换三处默认值一致。
"""
current = self.inspector_id_edit.text().strip()
old_default = PrinterController.default_inspector_id(old_family)
# 人工手填（不等于旧默认）则保留，不误覆盖
if current and current != old_default:
return
new_family = PrinterController.family_of(self._current_template_key)
new_default = PrinterController.default_inspector_id(new_family)
if new_default:
self.inspector_id_edit.setText(str(new_default))
self._sync_inspector_combo()
def _on_inspector_combo_changed(self, text: str) -> None:
"""下拉选择检验工号默认（检5 / No 5）时，写回文本框并触发预览刷新。
下拉是「显式覆盖」控件：选中即覆盖当前值，不再走模板自动切换的
「非人工手填才刷新」约束。
"""
if not text:
return
self.inspector_id_edit.setText(text)
self._on_data_changed()
def _sync_inspector_combo(self) -> None:
"""把下拉选中态同步到文本框当前值：
- 文本框值为 检5 / No 5 → 下拉选中对应项；
- 文本框为其他值（人工手填）→ 下拉置空（index -1），提示非默认值。
"""
val = self.inspector_id_edit.text().strip()
idx = self.inspector_combo.findText(val, Qt.MatchFixedString)
self.inspector_combo.setCurrentIndex(idx if idx >= 0 else -1)
# ---- ② 必填校验 + VBA 式 InputBox 补录 ----
def _field_label(self, key: str) -> str:
return self._FIELD_LABELS.get(key, key)
def _set_form_field(self, key: str, text: str) -> None:
widget = {
"product_name": self.product_name_edit,
"product_model": self.product_model_edit,
"dn": self.dn_edit,
"pn": self.pn_edit,
"temperature": self.temperature_edit,
"medium": self.medium_edit,
"check_standard": self.check_standard_edit,
"inspector_id": self.inspector_id_edit,
"manufacture_date": self.manufacture_date_edit,
}.get(key)
# 容错：部分字段（如 serial_number）为系统自动生成、无对应录入控件，
# 缺失控件时仅更新 data 字典（由调用方处理），不回写 UI。
if widget is not None:
widget.setText(text)
# ---- 统一护栏管道（抽公共到 services.output_guard，print/scan 共用，§11.5/§11.8 阶段1）----
def _guard_before_print(self, data: dict, *, multi: bool) -> tuple[bool, dict]:
"""发送前统一护栏（② 必填补录 / ③ reconfirm / ④ 差异确认 / ⑤ 确认闸口）。
返回 ``(ok, updated_data)``；``ok=False`` 表示用户取消，本批应中止。
内部将更新后的值写回表单控件（UX），并返回更新后的 data 供后续打印使用。
"""
if not self._guard_enabled():
return True, data
guard = cfg("print.guard", {}) or {}
history_last = self._get_history_last(data)
prompt = self._QtGuardPrompt(self)
ok, updated = PrinterController.run_guard(
data, guard=guard, prompt=prompt, history_last=history_last, multi=multi
)
if ok:
self._sync_form_from_data(updated)
return ok, updated
def _get_history_last(self, data: dict) -> dict | None:
"""取同型号上次实际打印值（供 ④ diff 闸）。无历史/异常返回 None。"""
model = (data.get("product_model") or "").strip()
if not model:
return None
try:
ctrl = self._get_cert_controller()
return ctrl.get_history_last_values(model)
except Exception as exc: # noqa: BLE001
logger.debug(f"[PrintView] 历史候选获取失败: {exc}")
return None
def _sync_form_from_data(self, data: dict) -> None:
"""将护栏更新后的字段值写回对应表单控件（UX 同步）。"""
for key, val in data.items():
self._set_form_field(key, val)
class _QtGuardPrompt:
"""GuardPrompt 的 PyQt 实现：弹 InputBox 补录/确认 + QMessageBox 确认闸口。"""
def __init__(self, view: PrintView) -> None:
self._view = view
def prompt_field(
self, *, key: str, label: str, current: str, suggestion: str
) -> str | None:
suggest = suggestion or self._dict_suggestion(key)
title = "确认字段" if current else "补全必填字段"
verb = "确认" if current else "填写"
text, ok = QInputDialog.getText(
self._view,
title,
f"请{verb}{label}：",
QLineEdit.Normal,
current or suggest,
)
if not ok:
return None
return text
def confirm(self, *, message: str) -> bool:
return (
QMessageBox.question(
self._view,
"确认打印",
message,
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
)
== QMessageBox.StandardButton.Yes
)
def _dict_suggestion(self, key: str) -> str:
"""从 model_param_mappings 取该字段建议值（仅 temperature/medium/check_standard/pn 有意义）。"""
mapping = {
"temperature": "working_temp",
"medium": "working_medium",
"check_standard": "test_standard",
"pn": "pressure_value",
}
col = mapping.get(key)
if not col:
return ""
model = (self._view.product_model_edit.text() or "").strip()
rec = self._view._lookup_model_param(model) if model else None
if rec is None:
return ""
return getattr(rec, col, None) or ""
# ---- ④ 学回字典（保守合并，manual 永不覆盖） ----
def _learn_back(self, cert: Any, data: dict) -> None:
if not self._guard_enabled():
return
if not cfg("print.guard.learn_back.enabled", False):
return
fields = cfg("print.guard.learn_back.fields", []) or []
if not fields:
return
model = (getattr(cert, "product_model", None) or "") or (data.get("product_model") or "")
model = (model or "").strip()
if not model:
return
form_to_db = {
"working_temp": data.get("temperature", ""),
"working_medium": data.get("medium", ""),
"test_standard": data.get("check_standard", ""),
"pressure_value": data.get("pn", ""),
}
updates = {col: (val or "").strip() for col, val in form_to_db.items() if col in fields}
if not any(updates.values()):
return
strategy = cfg("print.guard.learn_back.strategy", "conservative")
overwrite_manual = cfg("print.guard.learn_back.overwrite_manual", False)
try:
self._get_mpm_controller().upsert_for_learnback(
model,
updates,
strategy=strategy,
overwrite_manual=overwrite_manual,
)
except Exception as exc: # noqa: BLE001
logger.warning(f"[PrintView] 学回型号字典失败({model}): {exc}")
def _save_preview_image(self) -> None:
"""保存预览为图片（PNG）。
对齐补丁 19：对话框「导出 PDF」已改为「导出图片」，主预览页此按钮同步落地，
复用 ``render_certificate_pixmap`` 渲染位图并 ``QPixmap.save`` 落盘，消除
「PDF 生成功能开发中」占位（蓝图 D-13）。PDF 导出暂不在范围内。
"""
serial = self._get_current_serial()
file_path, _ = QFileDialog.getSaveFileName(
self, "保存图片", f"合格证_{serial}.png", "PNG 图片 (*.png)"
)
if not file_path:
return
data = self._get_form_data()
pixmap = render_certificate_pixmap(
data,
serial,
self._current_template_key,
show_labels=self._show_labels,
paper_w_mm=self._paper_width,
paper_h_mm=self._paper_height,
preview_w_px=280,
label_x_mm=self._label_x,
)
if pixmap.isNull():
QMessageBox.warning(self, "保存失败", "预览渲染为空，无法保存图片")
return
if pixmap.save(file_path, "PNG"):
QMessageBox.information(self, "已保存", f"预览图片已保存到：\n{file_path}")
else:
QMessageBox.warning(self, "保存失败", "图片保存失败，请检查路径与权限")
# ============================================================
# 预览相关
# ============================================================
# 预览渲染已迁移至 certificate_preview_painter.render_certificate_pixmap（C2 去 Chromium）。
# 字段标题映射 (_FIELD_LABELS) 与模板感知格式化 (TemplateFieldFormatter) 均在该模块内；
# 移除原 QWebEngineView/HTML 渲染（含 _is_headless 段错误绕过、底板 data URI 内嵌等）。
def _refresh_preview(self) -> None:
"""刷新预览（QPainter 绘制，替代 QWebEngineView / HTML）"""
if not hasattr(self, "preview_label"):
return
data = self._get_form_data()
serial = self._get_current_serial()
pixmap = render_certificate_pixmap(
data,
serial,
self._current_template_key,
show_labels=self._show_labels,
paper_w_mm=self._paper_width,
paper_h_mm=self._paper_height,
preview_w_px=280,
label_x_mm=self._label_x,
)
self.preview_label.setPixmap(pixmap)
# ============================================================
# 辅助方法
# ============================================================
def _get_form_data(self) -> dict:
"""获取表单数据"""
return {
"product_name": self.product_name_edit.text().strip(),
"product_model": self.product_model_edit.text().strip(),
"dn": self.dn_edit.text().strip(),
"pn": self.pn_edit.text().strip(),
"temperature": self.temperature_edit.text().strip(),
"medium": self.medium_edit.text().strip(),
"manufacture_date": self.manufacture_date_edit.text().strip(),
"check_standard": self.check_standard_edit.text().strip(),
"inspector_id": self.inspector_id_edit.text().strip(),
"prefix": self.prefix_edit.text().strip(),
"year_month": self.year_month_edit.text().strip(),
"start_number": self.start_number_spin.value(),
"quantity": self.quantity_spin.value(),
"suffix": self.suffix_edit.text().strip(),
"number_mode": self.number_mode_combo.currentText(),
"template_type": PrinterController.family_of(self._current_template_key),
"template_name": self.template_combo.currentText(),
"save_to_db": self.save_to_db_check.isChecked(),
}
# ============================================================
# 模板感知字段格式化（预览与打印共用，配置驱动）
# 实现见 certflow.services.printer.template_field_formatter（docs/PRINT_BLUEPRINT.md §4）
# ============================================================
def _format_print_data(self, data: dict) -> dict:
"""返回一份用于「实际打印」的格式化副本（不污染原始 data，避免影响落库）。
除按模板语言族格式化文本外，还会计算英制字段的 X 偏移（mm），
以 ``_x_offsets`` 随副本透传，供 ESC/P 引擎将数值右移到预印单位标签下方
（详见 docs/PRINT_BLUEPRINT.md §4.2）。
"""
out = dict(data)
formatter = PrinterController.get_field_formatter()
family = PrinterController.family_of(self._current_template_key)
x_offsets: dict[str, float] = {}
for f in ("pn", "temperature", "medium", "dn"):
raw = out.get(f)
if not raw:
continue
text, x_off = formatter.format(f, raw, family)
out[f] = text
if x_off:
x_offsets[f] = x_off
if x_offsets:
out["_x_offsets"] = x_offsets
return out
def _compose_serial(
self,
prefix: str,
year_month: str,
start: int,
quantity: int,
suffix: str,
three_digit: bool = False,
) -> str:
"""单一编号组合函数（编号设置组与导航共用，避免两套拼接逻辑漂移）。
- 单台(quantity==1)：``{prefix}{year_month}{起始:03d}{suffix}``
- 多台：``{prefix}{year_month}{起始:03d}{suffix}---{结束:03d}{suffix}``
``three_digit=True``（模式「三位流水」）强制 3 位流水；否则按原规则
（<1000 用 3 位，>=1000 用原始位数）保持兼容。
"""
start_str = (
f"{start:03d}" if three_digit else f"{start:03d}" if start < 1000 else str(start)
)
if quantity == 1:
return f"{prefix}{year_month}{start_str}{suffix}"
end = start + quantity - 1
end_str = f"{end:03d}" if three_digit else f"{end:03d}" if end < 1000 else str(end)
return f"{prefix}{year_month}{start_str}{suffix}---{end_str}{suffix}"
def _get_current_serial(self) -> str:
"""获取当前编号（优先取导航列表，否则按表单实时组合）"""
if self._serial_numbers and 0 <= self._current_serial_index < len(self._serial_numbers):
return self._serial_numbers[self._current_serial_index]
prefix = self.prefix_edit.text().strip()
year_month = self.year_month_edit.text().strip()
start = self.start_number_spin.value()
suffix = self.suffix_edit.text().strip()
quantity = self.quantity_spin.value()
return self._compose_serial(
prefix, year_month, start, quantity, suffix, self._three_digit_mode
)
def _update_serial_nav(self) -> None:
"""更新编号导航（仅在列表为空时按表单重算，避免覆盖已生成的导航）"""
if not self._serial_numbers:
prefix = self.prefix_edit.text().strip()
year_month = self.year_month_edit.text().strip()
start = self.start_number_spin.value()
suffix = self.suffix_edit.text().strip()
quantity = self.quantity_spin.value()
self._serial_numbers = [
self._compose_serial(prefix, year_month, s, 1, suffix, self._three_digit_mode)
for s in range(start, start + quantity)
]
total = len(self._serial_numbers)
current = self._current_serial_index + 1 if total > 0 else 0
self.serial_label.setText(f"{current} / {total}")
self.prev_btn.setEnabled(self._current_serial_index > 0)
self.next_btn.setEnabled(self._current_serial_index < total - 1)
if self._serial_numbers:
self.end_preview_label.setText(self._serial_numbers[-1])
def _auto_save(self, serials: list[str], data: dict) -> None:
"""自动保存打印记录到数据库（经 CertificateController 收口）。"""
printer_name = self._checked_printer or self._default_printer
self._get_cert_controller().auto_save_log(
serials, data, printer_name, self._current_certificate_id
)
def _prev_serial(self) -> None:
if self._current_serial_index > 0:
self._current_serial_index -= 1
self._update_serial_nav()
self._refresh_preview()
def _next_serial(self) -> None:
if self._current_serial_index < len(self._serial_numbers) - 1:
self._current_serial_index += 1
self._update_serial_nav()
self._refresh_preview()
def _on_data_changed(self) -> None:
"""数据变化（使用防抖，300ms 后统一刷新）"""
self._refresh_timer.start(300)
def _on_number_field_changed(self) -> None:
"""编号设置字段变化：重置导航索引、清空已生成列表并重算，再防抖刷新预览。
修复原逻辑中改起始/数量后右侧编号导航与结束号不刷新的不一致。
"""
self._current_serial_index = 0
self._serial_numbers = []
self._update_serial_nav()
self._refresh_timer.start(300)
def _do_refresh_preview(self) -> None:
"""实际刷新预览（由防抖定时器触发）"""
self._refresh_preview()
def _on_template_changed(self, _index: int = 0) -> None:
"""模板变化（下拉改选项即刷新预览 + 同步检验工号默认）"""
key = self.template_combo.currentData()
if key:
old_family = PrinterController.family_of(self._current_template_key)
self._current_template_key = key
# 切换语言族时同步检验工号默认值（中文→检5 / 其他→No 5）
self._sync_inspector_for_template_change(old_family)
self._refresh_preview()
def _on_show_labels_toggled(self, checked: bool) -> None:
"""预览「显示标题」开关：仅影响预览渲染（预印合格证可关，只显示值）"""
self._show_labels = bool(checked)
self._refresh_preview()
def _on_number_mode_changed(self, mode: str) -> None:
"""编号模式变化：切换流水位数 / 重新从证书加载编号。
- 「三位流水」：强制 3 位流水（覆盖默认的 <1000 用 3 位、>=1000 用原位的规则）。
- 「读取编号」：从当前绑定的 Certificate 重新回填编号设置组。
- 其余模式：仅刷新编号导航与预览。
"""
logger.info(f"编号模式切换为: {mode}")
self._three_digit_mode = mode == "三位流水"
if mode == "读取编号" and self._current_cert is not None:
self._bind_number_fields(self._current_cert)
self._current_serial_index = 0
self._serial_numbers = []
self._update_serial_nav()
self._refresh_preview()
def _on_auto_fill_toggled(self, checked: bool) -> None:
"""H2：「自动填写」开关切换 → 按新策略即时重载当前证书。
- 勾选（checked=True）：重载时 DB 值**覆盖**手改（force=True）；
- 取消（checked=False）：重载时**保留**手改、仅填空白（force=False）。
"""
if self._current_certificate_id is None:
return
self._reload_current_certificate(force=bool(checked))
# 表单控件属性名 → Certificate 列名（用于入队前校验 DB 中是否仍缺失）
_FIELD_TO_CERT_ATTR: dict[str, str] = {
"product_name_edit": "product_name",
"product_model_edit": "product_model",
"dn_edit": "product_spec",
"pn_edit": "pn_display",
"temperature_edit": "working_temp",
"medium_edit": "working_medium",
"check_standard_edit": "test_standard",
"inspector_id_edit": "inspector_id",
}
[文档]
def load_selected_sale_plan_ids(self, sale_plan_ids: list[int]) -> None:
"""加载选中的销售计划 ID 列表，支持一次打印多条记录。
多条记录均加入打印视图：各自绑定最新 Certificate 并按整批编号区间
回填；批量打印时依次逐条打印（如 12 台 + 32 台 = 44 台）。表单编辑 /
预览默认展示第一条，可通过「批量」按钮一次打印全部。
"""
if not sale_plan_ids:
return
# 收集所有销售计划及其关联的最新 Certificate（按 unique_key）
self._loaded_plans = []
self._loaded_certs = []
for pid in sale_plan_ids:
sp = self._get_cert_controller().get_sale_plan_by_id(pid)
if sp is None:
continue
self._loaded_plans.append(sp)
cert = (
self._get_cert_controller().get_certificate_by_unique_key(sp.unique_key)
if sp.unique_key
else None
)
self._loaded_certs.append(cert)
if not self._loaded_plans:
return
# 绑定第一条到表单（产品信息 / 编号设置，供编辑与预览）
sp0 = self._loaded_plans[0]
cert0 = self._loaded_certs[0]
self._bind_active_record(sp0, cert0)
total_units = sum(
self._plan_quantity(sp, cert)
for sp, cert in zip(self._loaded_plans, self._loaded_certs, strict=True)
)
logger.info(
"[PrintView] 加载销售计划: {} 条, 合计 {} 台, cert_id={}",
len(self._loaded_plans),
total_units,
self._current_certificate_id,
)
self.status_label.setText(f"已加载 {len(self._loaded_plans)} 条记录 / 共 {total_units} 台")
def _bind_active_record(self, plan: Any, cert: Any) -> None:
"""将一条 (销售计划, Certificate) 绑定到表单（产品信息 + 编号设置）。"""
if cert:
self._current_certificate_id = cert.id
self._current_cert = cert
self._fill_product_fields(cert)
# A3：恢复写证/打印时记录的模板（按语言族选中对应变体）
self._restore_template_by_family(cert.template_type or "")
# H1：补全出厂年月 / 检验工号
self._last_issue_ym = self._format_issue_ym(cert.issue_date)
self._apply_issue_date_display()
self.inspector_id_edit.setText(cert.inspector_id or "")
# 检验工号空值兜底：绑定即按模板语言族预填默认工号（配置驱动）
self._apply_inspector_default()
self._sync_inspector_combo()
self._current_sale_plan = plan
# 编号设置：按销售计划整批区间回填（前缀/年月/起始/数量/后缀）
self._bind_number_fields(cert, plan)
def _fill_product_fields(self, cert: Any) -> None:
"""从 Certificate 回填产品信息字段（名称/型号/口径/压力/温度/介质/标准）。"""
self.product_name_edit.setText(cert.product_name or "")
self.product_model_edit.setText(cert.product_model or "")
self.dn_edit.setText(cert.product_spec or "")
self.pn_edit.setText(cert.pn_display or "")
self.temperature_edit.setText(cert.working_temp or "")
self.medium_edit.setText(cert.working_medium or "")
self.check_standard_edit.setText(cert.test_standard or "")
# ① 加载自动带出：表单字段为空时按型号预填字典建议值（docs/PRINT_BLUEPRINT.md）
self._autofill_from_dict(cert)
def _reload_current_certificate(self, force: bool | None = None) -> None:
"""H2 事件驱动重载：从数据库重读当前 Certificate 并完整重绑表单。
解决簇 H 核心缺口——表单是「选中时拷贝一次的快照」，Certificate 被改
（参数补充确认 / 打印回填 / 订正写库）后表单不重载，导致用户看到的表单
与库里真实值不一致。本方法是「及时刷新」的统一入口，被以下事件驱动：
- 打印队列 worker 完成（``job_finished`` 信号，覆盖参数补充确认 + 打印回填）；
- 参数补充确认后即时重载（``supplement_callback`` 包装层）；
- 保存订正写库成功后重载（可选显式调用）。
重载策略（覆盖 vs 保留手改值）由 ``auto_fill_check`` 决定，对齐文档
§10.4-1 H2「接线 auto_fill_check 决定是否自动覆盖手改值」：
- ``force=True``：DB 值强制覆盖所有控件（无视手改）；
- ``force=False``：仅填空白控件，保留用户手改值；
- ``force=None``（默认）：由 ``auto_fill_check.isChecked()`` 决定——
勾选 ⇒ 覆盖（True），未勾选 ⇒ 保留手改（False）。
重载后刷新预览（防抖）。
"""
if self._current_certificate_id is None:
return
# 重读数据库最新值：worker 在独立线程/会话写回并已 commit，主线程会话
# 的 identity map 可能缓存旧对象——先 expire 再经 controller 门面重查，
# 保证拿到库里最新内容（controller 在 db_manager 模式下用独立会话，天然最新）。
self.session.expire_all()
fresh = self._get_cert_controller().get_certificate_by_id(self._current_certificate_id)
if fresh is None:
logger.warning(
f"[PrintView] H2 重载失败：cert_id={self._current_certificate_id} 不存在"
)
return
overwrite = force if force is not None else self.auto_fill_check.isChecked()
# 产品信息字段：按策略填/覆盖
self._apply_reload_field(self.product_name_edit, fresh.product_name, overwrite)
self._apply_reload_field(self.product_model_edit, fresh.product_model, overwrite)
self._apply_reload_field(self.dn_edit, fresh.product_spec, overwrite)
self._apply_reload_field(self.pn_edit, fresh.pn_display, overwrite)
self._apply_reload_field(self.temperature_edit, fresh.working_temp, overwrite)
self._apply_reload_field(self.medium_edit, fresh.working_medium, overwrite)
self._apply_reload_field(self.check_standard_edit, fresh.test_standard, overwrite)
self._apply_reload_field(self.inspector_id_edit, fresh.inspector_id, overwrite)
# 出厂年月：由 cert.issue_date 格式化（H1 已补，重载时同步最新值）
self._last_issue_ym = self._format_issue_ym(fresh.issue_date)
if overwrite or not self.manufacture_date_edit.text().strip():
self._apply_issue_date_display()
self._sync_inspector_combo()
# 编号设置组：始终按 DB 最新区间回填（编号是派生数据，非用户手改焦点）
self._bind_number_fields(fresh, self._current_sale_plan)
# 模板：按 DB 记录的语言族恢复（A3）
self._restore_template_by_family(fresh.template_type or "")
# 检验工号空值兜底（配置驱动，受 auto_fill_check 约束）
self._apply_inspector_default()
# 刷新预览（防抖，复用既有机制）
self._refresh_timer.start(300)
logger.info(f"[PrintView] H2 表单已事件驱动重载 cert_id={fresh.id} | overwrite={overwrite}")
def _apply_reload_field(self, widget: Any, value: Any, overwrite: bool) -> None:
"""按重载策略写回单个控件。
- ``overwrite=True``：DB 值覆盖控件当前内容；
- ``overwrite=False``：仅当控件为空时填 DB 值，保留用户手改。
"""
new_val = (value or "").strip()
if overwrite or not widget.text().strip():
widget.setText(new_val)
def _plan_quantity(self, plan: Any, cert: Any) -> int:
"""计算一条销售计划对应的合格证台数（整批区间优先，回退单张证书）。"""
meta = None
if plan and getattr(plan, "product_code", None):
meta = PrinterController.parse_product_code_meta(plan.product_code)
if meta is None and cert and cert.product_code_range:
meta = PrinterController.parse_product_code_meta(cert.product_code_range)
if meta:
return int(meta["quantity"])
return int(cert.quantity) if cert and cert.quantity else 1
# ============================================================
# H1：表单信源补全（出厂年月 / 检验工号 / 产品编号）
# ============================================================
@staticmethod
def _format_issue_ym(issue_date: str | None) -> str:
"""出厂日期格式化为 YYYY.MM（与 BUG-006 J 展示口径一致）。"""
if not issue_date:
return datetime.now().strftime("%Y.%m")
s = str(issue_date).replace("/", "-")
parts = s.split("-")
if len(parts) >= 2:
return f"{parts[0]}.{parts[1]}"
return str(issue_date)
def _bind_number_fields(self, cert: Any = None, sale_plan: Any = None) -> None:
"""从 SalePlan / Certificate 回填编号设置组（前缀/年月/起始/数量/后缀）。
数据源绑定的一部分：``load_selected_sale_plan_ids`` 负责产品信息字段，
本方法负责编号设置组，二者解耦清晰。
编号设置以「销售计划整批编号区间」为准（BUG-006 按销售计划关联 + 打印
按数量分批）：优先正则拆分 ``sale_plan.product_code`` 区间串（如
``V2607012-023`` → 前缀 V / 年月 2607 / 起始 12 / 数量 12 / 后缀），
使打印设置直接反映整批台数，便于按数量批量打印（数量 12 = 第一批，
后续改数量 32 = 第二批）。拆分失败再回退单张 Certificate 的
``product_code_range`` 或已拆填的 ``product_code_*`` 字段。
"""
# 优先取销售计划整批编号区间（含整批台数），其次单张证书区间
sp = sale_plan or getattr(self, "_current_sale_plan", None)
meta = None
if sp and getattr(sp, "product_code", None):
meta = PrinterController.parse_product_code_meta(sp.product_code)
if meta is None and cert and cert.product_code_range:
meta = PrinterController.parse_product_code_meta(cert.product_code_range)
if meta:
prefix = meta["prefix"]
ym = meta["ym"]
seq_start = meta["seq_start"]
quantity = meta["quantity"]
suffix = meta["suffix"] or ""
else:
prefix = (cert.product_code_prefix if cert else None) or "V"
ym = (cert.product_code_ym if cert else None) or datetime.now().strftime("%y%m")
seq_start = (cert.product_code_seq_start if cert else None) or 1
quantity = (cert.quantity if cert else None) or 1
suffix = ""
self.prefix_edit.setText(prefix or "V")
self.year_month_edit.setText(ym or datetime.now().strftime("%y%m"))
self.start_number_spin.setValue(int(seq_start) if seq_start else 1)
self.quantity_spin.setValue(int(quantity) if quantity else 1)
self.suffix_edit.setText(suffix or "")
# 加载期间抑制「目标年月」自动覆盖起始；加载后刷新下一号预览（不改起始）
self._loading_record = True
# （上面 setText 已触发 _on_target_ym_changed，因 _loading_record 被跳过自动填充）
self._loading_record = False
self._update_next_serial_label()
# ============================================================
# 以下为 main 分支的「直发/护栏」底层基础设施方法（合并自 origin/main）。
# 本 PR 的 GUI 默认入口为「保存订正 / 入队启动」队列流水线（见 _on_save_correction /
# _on_enqueue_and_start），但保留这些底层方法，以便 main 的 print-guard 端到端
# 测试（tests/test_print_guard_e2e_smoke.py）及内部直发调用（_do_batch_print_all 等）
# 仍可正常工作；它们与队列流水线互不冲突。
# ============================================================
def _get_printer_controller(self) -> Any:
"""获取打印机控制器"""
if self._printer_controller is None:
from certflow.controllers.printer_controller import PrinterController
self._printer_controller = PrinterController()
# 使用当前活动打印机名（由 PrinterController.current_printer 惰性获取系统默认）
real_printer = self._printer_controller.current_printer
logger.info(f"[PrintConfig] 系统默认打印机: {real_printer}")
if real_printer:
self._printer_controller.set_printer(real_printer)
# 连接打印完成信号，感知异步打印结果
self._printer_controller.print_completed.connect(self._on_print_done)
return self._printer_controller
def _on_print_done(self, success: bool, message: str) -> None:
"""打印完成回调（异步）"""
if success:
self.status_label.setText(f"打印成功: {message}")
else:
self.status_label.setText(f"打印失败: {message}")
logger.error(f"[PrintConfig] 打印失败: {message}")
def _on_queue_job_finished(self, job_id: int, success: bool, message: str) -> None:
"""H2 事件驱动重载：打印队列任务完成后按 job 关联证书重载表单。
``PrintQueueController.job_finished`` 信号（主线程 emit，worker 写库已
commit）携带 ``(job_id, success, message)``。任务成功且关联证书正是当前
加载的证时，重读数据库并重绑表单——覆盖「参数补充确认写库」与「打印回填」
两类 Certificate 变更，使表单不再停留在选中时的旧快照。
重载策略遵循 ``auto_fill_check``（见 ``_reload_current_certificate``）。
"""
if not success:
return
# 经 CertificateController 门面查 job 关联的 certificate_id（视图不直连 models）
try:
job_cert_id = self._get_cert_controller().get_job_certificate_id(job_id)
except Exception as e: # noqa: BLE001
logger.warning(f"[PrintView] H2 查 job 失败，跳过重载: {e}")
return
if job_cert_id is None:
return
if job_cert_id != self._current_certificate_id:
return
logger.info(f"[PrintView] H2 队列任务完成触发重载 job_id={job_id} cert_id={job_cert_id}")
self._reload_current_certificate()
def _check_printer(self) -> str | None:
"""严格检查打印机：配置指定名称必须在系统列表中，并设为活动打印机
同一批次多次调用时，使用缓存结果避免重复枚举打印机列表。
Returns:
匹配的打印机名称，不匹配时弹框提示并返回 None
"""
import win32print
# 缓存：同一批次已检查过则直接返回
if self._checked_printer:
return self._checked_printer
# 1. 从配置文件读取期望的打印机名
expected = self._default_printer
if not expected or expected == "默认打印机":
logger.warning("[PrintConfig] 配置中未指定打印机，跳过检查")
return None
# 2. 获取系统打印机列表
try:
printers = [
p[2]
for p in win32print.EnumPrinters(
win32print.PRINTER_ENUM_LOCAL | win32print.PRINTER_ENUM_CONNECTIONS
)
]
except Exception:
printers = []
# 3. 精确匹配：只认定配置中的单一打印机，名称不一致即视为不可用
# （不接受模糊匹配，避免错打到其它设备）
matched = expected if expected in printers else None
if matched is None:
logger.error(
f"[PrintConfig] 打印机未找到: 配置={expected}, 系统可用={', '.join(printers[:5])}"
)
QMessageBox.warning(
self,
"打印机未就绪",
f"配置指定的打印机:\n {expected}\n\n"
"在系统中未找到。\n\n"
"请确认:\n"
"1. 打印机已开机并连接\n"
"2. 驱动程序已正确安装\n"
"3. 打印机名称与 config.yaml 中一致",
)
return None
# 5. 设为系统默认打印机，确保打印作业发送到正确设备
current_default = win32print.GetDefaultPrinter()
if current_default != matched:
logger.info(f"[PrintConfig] 切换默认打印机: {current_default} → {matched}")
try:
win32print.SetDefaultPrinter(matched)
except Exception as e:
logger.error(f"[PrintConfig] 设置默认打印机失败: {e}")
QMessageBox.warning(
self,
"打印机设置失败",
f"无法将默认打印机切换为:\n {matched}\n\n错误: {e}",
)
return None
logger.info(f"[PrintConfig] 打印机就绪: {matched}")
self._checked_printer = matched
return matched
def _build_print_config(self) -> Any:
"""构建打印配置，打印机不可用时返回 None"""
from certflow.config.print_config import RuntimePrintConfig
# 严格检查打印机
printer_name = self._check_printer()
if printer_name is None:
return None
# 根据模板配置键映射（A2：不再将「全英文」静默降级到 bilingual_1）
key = self._current_template_key or "full_chinese_1"
return RuntimePrintConfig.from_template(
template_key=key,
printer_name=printer_name,
copies=1,
)
def _advance_start_after_print(self, count: int) -> None:
"""打印后推进起始编号"""
current = self.start_number_spin.value()
self.start_number_spin.setValue(current + count)
self._serial_numbers = []
self._update_serial_nav()