# src/certflow/services/certificate_print_service.py
"""合格证打印服务（编排层）
编排完整的合格证打印业务流程，组合编号、打印引擎、日志等子模块。
拆分为：
- config/print_config.py: 配置加载（canonical，原 cert_print_config.py 已迁入）
- cert_numbering.py: 编号与 Certificate 写入
- cert_print_engine.py: GDI/HTML 打印引擎
- certificate_print_service.py (本文件): 编排层（参数补充、批量打印、日志、模板）
"""
from __future__ import annotations
import os
from collections.abc import Callable
from datetime import datetime
from typing import Any
from sqlalchemy.orm import Session
from certflow.config.print_config import get_empty_markers
from certflow.models.certificate import Certificate
from certflow.models.print_log import PrintLog
from certflow.services.cert_numbering import CertNumberingService
from certflow.services.cert_numbering_policy import get_number_model, is_auto_calc_dn_pn
from certflow.services.cert_print_engine import CertPrintEngine
from certflow.services.dn_service import DNService
from certflow.services.pn_service import PNService
from certflow.services.print_history_service import HistoryBackfillService
from certflow.utils.database import DatabaseManager
from certflow.utils.logger import logger
[文档]
class CertificatePrintService:
"""合格证打印服务（编排层）
组合编号、打印引擎、日志等子模块，提供完整的打印业务流程。
编号模式（委托 CertNumberingService）：
- auto_number: 自动编号（正则解析 production_order_no / supply_type，最常用）
- manual_number: 手动指定前缀和起始序号
- copy_number: 复制已有编号（单件直接复制）
打印（委托 CertPrintEngine）：
- print_single: 单台 GDI 精确打印
- batch_print: 批量展开编号 → 逐台打印 → 记录日志
- print_html_fallback: HTML 降级打印
持久化（委托 CertNumberingService）：
- create_certificates: 将已编号的 SalePlan 写入 Certificate 表
- export_to_access: 导出到 Access 数据库
- export_sign_tb_csv: 导出 CSV
日志与历史：
- record_print_log: 记录打印日志到 SQLite print_logs 表
- get_print_history: 查询打印历史
Attributes:
session: SQLAlchemy 数据库会话对象
output_dir: 输出目录路径
printer_name: 默认打印机名称
Examples:
>>> service = CertificatePrintService(session)
>>> # 自动编号
>>> count = service.auto_number([1, 2, 3], prefix_override="2506")
>>> # 写入 Certificate
>>> cert_count = service.create_certificates([1, 2, 3])
>>> # 逐台打印
>>> service.batch_print(certificate_id=1, printer_name="EPSON LQ-635KII")
"""
def __init__(
self,
session: Session,
output_dir: str = "output/certificates",
printer_name: str = "",
) -> None:
"""初始化合格证打印服务
Args:
session: SQLAlchemy 数据库会话对象
output_dir: 输出目录路径，默认为 "output/certificates"
printer_name: 默认打印机名称，为空则使用系统默认打印机
"""
self.session = session
self.output_dir = output_dir
self.printer_name = printer_name
self._numbering = CertNumberingService(session)
self._engine = CertPrintEngine(printer_name)
os.makedirs(output_dir, exist_ok=True)
# ============================================================
# 编号模式（委托 CertNumberingService）
# ============================================================
[文档]
def auto_number(self, ids: list[int], prefix_override: str | None = None) -> int:
"""自动编号（委托 CertNumberingService）
通过正则解析 production_order_no / supply_type 自动生成前缀并编号。
Args:
ids: SalePlan.id 列表
prefix_override: 自定义前缀覆盖（如 "2506"），None 则按规则推导
Returns:
int: 成功编号的记录数
"""
return self._numbering.auto_number(ids, prefix_override)
[文档]
def manual_number(
self, ids: list[int], prefix: str, start: int, quantity: int, suffix: str = ""
) -> int:
"""手动编号（委托 CertNumberingService）
Args:
ids: SalePlan.id 列表
prefix: 编号前缀
start: 起始序号
quantity: 每台数量
suffix: 编号后缀
Returns:
int: 成功编号的记录数
"""
return self._numbering.manual_number(ids, prefix, start, quantity, suffix)
[文档]
def copy_number(self, ids: list[int], source_code: str) -> int:
"""复制编号（委托 CertNumberingService）
将已有编号直接复制到目标记录（单件直接复制）。
Args:
ids: SalePlan.id 列表
source_code: 源编号（作为复制模板）
Returns:
int: 成功编号的记录数
"""
return self._numbering.copy_number(ids, source_code)
# ============================================================
# Certificate 写入与导出（委托 CertNumberingService）
# ============================================================
[文档]
def create_certificates(
self, ids: list[int], printer_name: str = "", print_status: str = "已打印"
) -> int:
"""将已编号的记录写入 Certificate 表（委托 CertNumberingService）
Args:
ids: SalePlan.id 列表
printer_name: 写入的打印机名称
print_status: 写入的打印状态，默认 "已打印"
Returns:
int: 成功创建的 Certificate 记录数
"""
return self._numbering.create_certificates(ids, printer_name, print_status)
[文档]
def export_to_access(self, ids: list[int], mdb_path: str | None = None) -> int:
"""导出到 Access 数据库（委托 CertNumberingService）
Args:
ids: Certificate.id 列表
mdb_path: 目标 .mdb 文件路径，None 则使用默认路径
Returns:
int: 成功导出的记录数
"""
return self._numbering.export_to_access(ids, mdb_path)
[文档]
def export_sign_tb_csv(self, output_path: str | None = None, limit: int = 1000) -> str | None:
"""导出 CSV（委托 CertNumberingService）
Args:
output_path: 输出 CSV 文件路径，None 则使用默认路径
limit: 最大导出条数
Returns:
str | None: 生成的 CSV 文件路径；失败返回 None
"""
return self._numbering.export_sign_tb_csv(output_path, limit)
# ============================================================
# 生成合格证（统一「打印合格证」入口，委托 CertNumberingService 数据访问）
# ============================================================
[文档]
def generate_print_certificates(
self,
ids: list[int],
prefix_override: str | None = None,
print_status: str = "待打印",
) -> tuple[int, int]:
"""为选中的销售计划生成合格证（去重安全）。
跳过已生成 Certificate 的计划（避免证书重复生成与唯一约束冲突），
对剩余计划自动编号并写入 Certificate 表。编号与写库均委托
``CertNumberingService``（数据访问层），本方法仅做编排与去重判定。
Args:
ids: 销售计划 ID 列表
prefix_override: 编号前缀覆盖（None 则按规则推导）
print_status: 写入的打印状态
Returns:
tuple[int, int]: (编号数, 合格证生成数)
"""
from certflow.models import Certificate
if not ids:
return 0, 0
# 去重：跳过已生成合格证的计划
existing = {
c.sale_plan_id
for c in self.session.query(Certificate.sale_plan_id)
.filter(Certificate.sale_plan_id.in_(ids))
.all()
}
new_ids = [i for i in ids if i not in existing]
if not new_ids:
logger.info("选中记录均已生成合格证，无需重复生成")
return 0, 0
numbered = self._numbering.auto_number(new_ids, prefix_override)
cert_count = self._numbering.create_certificates(new_ids, self.printer_name, print_status)
logger.info(f"生成合格证完成 | 编号={numbered} | 证书={cert_count}")
return numbered, cert_count
# ============================================================
# 打印（委托 CertPrintEngine）
# ============================================================
[文档]
def print_single(
self,
data: dict[str, Any],
serial_number: str,
printer_name: str | None = None,
print_fields: list[str] | None = None,
) -> bool:
"""打印单台合格证（委托 CertPrintEngine）
Args:
data: 打印数据字典（含 product_name、dn、pn 等字段）
serial_number: 合格证编号（SN）
printer_name: 打印机名称，None 则使用服务默认打印机
print_fields: 需要打印的字段名列表，None 表示打印全部字段
Returns:
bool: 是否打印成功
"""
return self._engine.print_single(data, serial_number, printer_name, print_fields)
@property
def last_os_job_id(self) -> int | None:
"""最近一次打印提交的 OS 作业 ID（委托 CertPrintEngine 捕获）。
偏差 5 作业 ID 捕获链路末端：队列的 ``_default_print_func`` 经此属性
拿到 spooler 作业 ID，写入 ``PrintJob.os_job_id``，供 OS 级取消使用。
"""
return self._engine.last_os_job_id
[文档]
def print_html_fallback(self, data: dict[str, Any], serial_number: str) -> str | None:
"""HTML 降级打印（委托 CertPrintEngine）
Args:
data: 打印数据字典
serial_number: 合格证编号（SN）
Returns:
str | None: 生成的 HTML 文件路径；失败返回 None
"""
return self._engine.print_html_fallback(data, serial_number)
# ============================================================
# 参数补充（核心编排逻辑）
# ============================================================
[文档]
def check_and_supplement_params(
self,
cert: Certificate,
parent: object | None = None,
supplement_callback: Callable | None = None,
) -> dict[str, Any]:
"""检查 Certificate 参数完整性，必要时弹出参数补充对话框
对标 VBA print.vb 中一连串的 InputBox 调用：
- 阀体材质 InputBox（dd.vb:1066）
- 阀杆材质 InputBox（dd.vb:1074）
- 启闭件材质 InputBox（dd.vb:1082）
- 检验工号 InputBox（dd.vb:894-906）
- 出厂年月修正 InputBox（dd.vb:986,1009）
Args:
cert: Certificate ORM 对象
parent: 对话框父级窗口，服务层不感知其具体类型，仅原样透传给
supplement_callback；由 Controller/View 层传入 QWidget
supplement_callback: 可选回调 (model, prefill, material_grades, parent)
→ (confirmed, skip_all, params)，由 Controller/View 层注入
"""
result: dict[str, Any] = {
"supplemented": False,
"skipped": False,
"params": {},
}
dn_service = DNService(self.session)
pn_service = PNService(self.session)
model = cert.product_model or ""
prefill = self._build_prefill(cert, model, dn_service)
# 判断是否需要弹出对话框
missing_params = self._detect_missing_params(cert, prefill)
if not missing_params:
logger.info(f"Certificate {cert.certificate_no} 参数完整，跳过补充对话框")
return result
logger.info(
f"Certificate {cert.certificate_no} 缺少参数: "
f"{', '.join(missing_params)}，弹出补充对话框"
)
# 弹出参数补充对话框（通过回调委托给上层处理 UI）
try:
material_grades = self._load_material_grades()
if supplement_callback is not None:
confirmed, skip_all, params = supplement_callback(
model=model,
prefill=prefill,
material_grades=material_grades,
parent=parent,
)
else:
logger.warning("无参数补充回调，跳过对话框")
return result
if not confirmed:
return result
result["supplemented"] = True
result["skipped"] = skip_all
result["params"] = params
if not skip_all:
self._apply_supplemented_params(cert, params, dn_service, pn_service)
except Exception as e:
logger.error(f"参数补充对话框异常: {e}")
return result
@staticmethod
def _build_prefill(
cert: Certificate,
model: str,
dn_service: DNService,
) -> dict[str, str]:
"""构建参数预填字典（包含材质、检验工号、出厂年月）
Args:
cert: Certificate ORM 对象
model: 产品型号
dn_service: DN 服务实例
Returns:
dict[str, str]: 预填字典，键含 caliber/pressure/temperature/medium 等
"""
dn_value = ""
if cert.product_spec:
if is_auto_calc_dn_pn():
# 开启自动计算（默认）：从口径串解析标准化口径值
dn_result = dn_service.resolve_caliber(cert.product_spec, model=model)
dn_value = dn_result.get("caliber_value") or cert.product_spec
else:
# 关闭自动计算（§5.7.2-E / 维度8）：仅用已存原始口径，不自动推算
dn_value = cert.product_spec
pn_value = ""
if cert.pn_value:
pn_value = cert.pn_display or cert.pn_value
return {
"caliber": dn_value,
"pressure": pn_value,
"temperature": cert.working_temp or "",
"medium": cert.working_medium or "",
"test_standard": cert.test_standard or "",
"inspector_id": getattr(cert, "inspector_id", "") or "",
"body_material": getattr(cert, "fati_material", "") or "",
"stem_material": getattr(cert, "fagan_material", "") or "",
"disc_material": getattr(cert, "qibijian_material", "") or "",
"manufacture_date": getattr(cert, "issue_date", "") or "",
}
@staticmethod
def _detect_missing_params(cert: Certificate, prefill: dict[str, str]) -> list[str]:
"""检测缺失的参数列表
Args:
cert: Certificate ORM 对象
prefill: 由 _build_prefill 生成的预填字典
Returns:
list[str]: 缺失参数中文名列表（如 ["口径(DN)", "压力(PN)"]）
"""
missing: list[str] = []
# 关闭自动计算口径压力时（§5.7.2-E / 维度8），不就口径/压力缺失弹补充对话框
if is_auto_calc_dn_pn():
if not prefill.get("caliber"):
missing.append("口径(DN)")
if not prefill.get("pressure"):
missing.append("压力(PN)")
if not cert.working_temp:
missing.append("温度")
if not cert.working_medium:
missing.append("介质")
return missing
def _apply_dn_param(
self, cert: Certificate, params: dict[str, str], dn_service: DNService
) -> None:
"""应用 DN/口径 参数
Args:
cert: Certificate ORM 对象
params: 用户填写的参数（含 caliber 键）
dn_service: DN 服务实例
"""
model = cert.product_model or ""
if not params.get("caliber"):
return
dn_result = dn_service.resolve_caliber(params["caliber"], model=model)
cert.product_spec = dn_result.get("caliber_value") or params["caliber"]
if dn_result.get("needs_manual"):
dn_service.auto_learn(
raw_text=params["caliber"],
caliber_value=cert.product_spec,
certificate_no=cert.certificate_no,
)
def _apply_pn_param(
self, cert: Certificate, params: dict[str, str], pn_service: PNService
) -> None:
"""应用 PN/压力 参数
Args:
cert: Certificate ORM 对象
params: 用户填写的参数（含 pressure 键）
pn_service: PN 服务实例
"""
model = cert.product_model or ""
if not params.get("pressure"):
return
pn_info = pn_service.resolve_pn(model=model, template_pn=params["pressure"])
if pn_info.get("pn_value"):
cert.pn_value = pn_info["pn_value"]
cert.pn_unit = pn_info["pn_unit"]
cert.pn_display = pn_info["pn_display"]
pn_service.auto_learn(
model=model,
pn_value=pn_info["pn_value"],
pn_unit=pn_info["pn_unit"],
test_standard=params.get("test_standard", ""),
certificate_no=cert.certificate_no,
)
def _apply_basic_params(self, cert: Certificate, params: dict[str, str]) -> None:
"""应用温度、介质、试压标准、材质、工号、出厂年月等基础参数
Args:
cert: Certificate ORM 对象
params: 用户填写的参数（含 temperature/medium 等键）
"""
if params.get("temperature"):
cert.working_temp = params["temperature"]
if params.get("medium"):
cert.working_medium = params["medium"]
if params.get("test_standard"):
cert.test_standard = params["test_standard"]
# === v9 新增：材质信息（对齐VBA InputBox）===
if params.get("body_material"):
cert.fati_material = params["body_material"]
if params.get("stem_material"):
cert.fagan_material = params["stem_material"]
if params.get("disc_material"):
cert.qibijian_material = params["disc_material"]
# === v9 新增：检验工号（对齐VBA dd.vb:894-906）===
if params.get("inspector_id") and hasattr(cert, "inspector_id"):
cert.inspector_id = params["inspector_id"]
# === v9 新增：出厂年月修正（对齐VBA dd.vb:986,1009）===
if params.get("manufacture_date"):
cert.issue_date = params["manufacture_date"]
def _apply_supplemented_params(
self,
cert: Certificate,
params: dict[str, str],
dn_service: DNService,
pn_service: PNService,
) -> None:
"""将补充的参数应用到 Certificate 对象
Args:
cert: Certificate 对象
params: 用户填写的参数（含材质、检验工号、出厂年月）
dn_service: DN 服务实例
pn_service: PN 服务实例
"""
self._apply_dn_param(cert, params, dn_service)
self._apply_pn_param(cert, params, pn_service)
self._apply_basic_params(cert, params)
# 提交
self.session.commit()
logger.info(
f"参数已补充: cert_id={cert.id}, 口径={params.get('caliber', '')}, "
f"压力={params.get('pressure', '')}, 温度={params.get('temperature', '')}, "
f"介质={params.get('medium', '')}, "
f"阀体材质={params.get('body_material', '')}, "
f"阀杆材质={params.get('stem_material', '')}, "
f"启闭件材质={params.get('disc_material', '')}, "
f"工号={params.get('inspector_id', '')}"
)
# 材质牌号已迁移到 SQLite material_grades 表（见 MaterialGrade 模型与 MaterialGradeService）。
# _load_material_grades 优先按阀门零件类别从数据库分组读取；表为空或异常时回退硬编码，
# 保证打印参数补充对话框不降级。
def _load_material_grades(self) -> dict[str, list[str]]:
"""从 SQLite material_grades 表加载材质牌号列表（按阀门零件类别分组）
优先读取数据库；若表为空或读取失败，回退到内置硬编码字典，保证对话框始终可用。
Returns:
dict: {分类: [牌号列表]}，如 {"body": ["WCB", "CF8"], "stem": ["2Cr13"], ...}
"""
try:
from certflow.services.material_grade_service import MaterialGradeService
grades = MaterialGradeService(self.session).get_grades_by_category()
if grades:
return grades
logger.info("material_grades 表为空，回退硬编码材质牌号")
except Exception as e:
logger.warning(f"从数据库读取材质牌号失败，回退硬编码: {e}")
return self._hardcoded_material_grades()
@staticmethod
def _hardcoded_material_grades() -> dict[str, list[str]]:
"""内置硬编码材质牌号（数据库为空时的兜底）
Returns:
dict[str, list[str]]: {分类: [牌号列表]}，如 {"body": ["WCB"], ...}
"""
# 预设常用阀门材质牌号
return {
"body": [
"WCB",
"LCB",
"LCC",
"WC6",
"WC9",
"CF8",
"CF8M",
"CF3",
"CF3M",
"HT200",
"HT250",
"QT450-10",
],
"stem": [
"2Cr13",
"3Cr13",
"304",
"316",
"17-4PH",
"Monel",
"F51",
"F53",
],
"disc": [
"2Cr13",
"3Cr13",
"304",
"316",
"WCB",
"CF8",
"CF8M",
"F51",
],
}
# ============================================================
# 批量打印 & 日志记录
# ============================================================
[文档]
def batch_print(
self,
certificate_id: int,
printer_name: str | None = None,
progress_callback: Callable[[int, int], None] | None = None,
skip_supplement: bool = False,
parent: object | None = None,
supplement_callback: Callable | None = None,
) -> dict[str, Any]:
"""批量打印：参数补充 → 解析编号 → 展开逐台编号 → 打印 → 记录日志
对标 VBA print.vb 批量打印流程：
1. 检查参数完整性，缺失则弹出补充对话框（对标 InputBox）
2. 解析 Certificate 编号范围
3. 展开为逐台编号后依次打印
4. 每台记录一条 print_log
Args:
certificate_id: Certificate.id
printer_name: 打印机名称
progress_callback: 进度回调 fn(current, total)
skip_supplement: 跳过参数补充对话框（批量自动化时使用）
parent: 父窗口（用于对话框模态），服务层不感知其具体类型，原样透传
supplement_callback: 参数补充对话框回调
(model, prefill, material_grades, parent) → (confirmed, skip_all, params)
Returns:
{"success": int, "failed": int, "serials": [...], "message": str}
"""
cert = self.session.query(Certificate).filter(Certificate.id == certificate_id).first()
if not cert:
return {"success": 0, "failed": 0, "serials": [], "message": "Certificate 不存在"}
# === 参数补充（对标 VBA InputBox）===
if not skip_supplement:
supp_result = self.check_and_supplement_params(
cert, parent=parent, supplement_callback=supplement_callback
)
if not supp_result.get("supplemented", False) and not supp_result.get("skipped", False):
return {"success": 0, "failed": 0, "serials": [], "message": "用户取消打印"}
self.session.refresh(cert)
# 展开编号列表
serials = cert.code_list
if not serials:
product_code = cert.product_code_range or cert.certificate_no or ""
serials = [product_code]
if not serials:
return {"success": 0, "failed": 0, "serials": [], "message": "无法展开编号"}
# 构建打印数据
print_data = self._build_print_data(cert)
printer = printer_name or self.printer_name
total = len(serials)
success_count = 0
failed_count = 0
logger.info(
f"开始批量打印 cert_id={certificate_id}, 编号范围={cert.certificate_no}, 共 {total} 台"
)
for i, sn in enumerate(serials):
# 逐台打印
ok = self.print_single(print_data, sn, printer)
if not ok:
self.print_html_fallback(print_data, sn)
# 记录打印日志到 SQLite（51个字段完全兼容 Access SignTb）
self.record_print_log(
certificate_id=certificate_id,
certificate_no=sn,
cert_data={
# Br(1) ~ Br(9): 基础信息
"plan_date": cert.plan_date,
"customer": cert.customer,
"project_name": cert.project_name,
"product_name": cert.product_name,
"product_model": cert.product_model,
"dn": cert.product_spec,
"pn": cert.pn_display or cert.pn_value or "",
"medium": cert.working_medium,
"temperature": cert.working_temp,
# Br(10) ~ Br(11): 出厂日期年月
"manufacture_year": int(cert.product_code_ym[:4])
if cert.product_code_ym and len(cert.product_code_ym) >= 4
else 0,
"manufacture_month": int(cert.product_code_ym[4:6])
if cert.product_code_ym and len(cert.product_code_ym) >= 6
else 0,
# Br(25) ~ Br(27): 合格证版型/图片/编码模式
"template_style": 1,
"pic_style": 0,
# 步骤 6.2：编码模式由 certificate.numbering.modes 配置判定（默认 0）
"number_model": get_number_model(cert),
# Br(28): 合格证模板名称
"template_name": cert.product_name or "",
# Br(32): 铭牌路径（model_full_name）
"model_full_name": print_data.get("model_full_name", ""),
# SN/KKS 位置标记（落库值，由编号模式规则确定；驱动双编码互换）
"sn_kks_position": getattr(cert, "sn_kks_position", "") or "",
# Br(35) ~ Br(45): 材质/技术要求/销售信息
"body_material": "",
"stem_material": "",
"disc_material": "",
"tech_requirements": cert.tech_requirements,
"equipment_no": "",
"sub_project": "",
"sales_sn": "",
"category": "",
"weight": "",
"seq_no": "",
"requisition_no": "",
# Br(49): 打印模式
"print_mode": "1",
# Br(50) ~ Br(51): 水头
"max_inlet_head": "",
"max_boost_head": "",
# hgz_ShuLiang
"copies": 1,
},
printer_name=printer,
status="success" if ok else "failed",
error_message="" if ok else "GDI 打印失败，已降级为 HTML 打印",
content_summary=self._build_content_summary(print_data, sn),
)
if ok:
success_count += 1
else:
failed_count += 1
if progress_callback:
progress_callback(i + 1, total)
# 更新 Certificate 打印状态
cert.print_status = "已打印"
cert.print_time = datetime.now()
cert.printer_name = printer
# G3: 若本 SalePlan 全部台数已打印，回填 sn_code/kks_code 到 SalePlan
from certflow.services.cert_log_service import backfill_dual_codes_if_fully_printed
backfill_dual_codes_if_fully_printed(self.session, certificate_id)
DatabaseManager.commit_with_retry(self.session)
msg = f"打印完成: 成功{success_count}台, 失败{failed_count}台"
logger.info(msg)
return {
"success": success_count,
"failed": failed_count,
"serials": serials,
"message": msg,
}
def _build_print_data(self, cert: Certificate) -> dict[str, Any]:
"""从 Certificate ORM 对象构建打印数据字典
DN 字段对标 VBA print.vb 第681行：Replace(str_kjA, "DN", "")
PN 字段根据模板类型生成合适的显示文本：
- 全中文模板：纯数值（底图已印 MPa）
- 中英文/俄英文模板：数值+单位
v9 新增：英制处理（LB判断/等级提取/C5C6回退）
Args:
cert: Certificate ORM 对象
Returns:
dict[str, Any]: 打印数据字典（含 product_name、dn、pn、temperature 等）
"""
# === DN 标准化 ===
dn_display = ""
if cert.product_spec:
dn_service = DNService(self.session)
dn_result = dn_service.resolve_caliber(cert.product_spec)
dn_display = dn_result.get("caliber_value") or cert.product_spec
else:
dn_display = ""
# === PN 显示文本 ===
pn_display = ""
pn_service = PNService(self.session)
model = cert.product_model or ""
if cert.pn_value:
template_name = getattr(cert, "template_type", "") or "全中文"
template_type = "全中文"
if "中英文" in template_name or "俄英文" in template_name:
template_type = "中英文"
elif "全英文" in template_name:
template_type = "全英文"
pn_display = pn_service.get_display_text(
pn_value=cert.pn_value,
pn_unit=cert.pn_unit or "",
template_type=template_type,
)
elif cert.pn_display:
pn_display = cert.pn_display
# === 英制处理（对齐VBA dd.vb:723-755, 841-857）===
is_imperial = pn_service.is_imperial(model)
lb_level = pn_service.extract_lb_level(model) if is_imperial else None
imperial_dn = "" # C5: 英制口径备用列
imperial_pn = "" # C6: 英制压力备用列
if is_imperial and lb_level:
imperial_pn = f"{lb_level}Lb"
if cert.product_spec:
imperial_dn = cert.product_spec
return {
"product_name": cert.product_name or "",
"product_model": model,
"dn": dn_display,
"pn": pn_display,
"temperature": cert.working_temp or "",
"medium": cert.working_medium or "",
"check_standard": cert.test_standard or "",
# H3/实物缺工号根因：此前硬编码为 ""，导致队列实际打印永远丢检验工号；
# 改为读取 cert.inspector_id（建证时按语言族烘焙默认检5/No5，订正/下拉亦可覆盖）。
"inspector_id": cert.inspector_id or "",
"manufacture_date": cert.issue_date or datetime.now().strftime("%Y.%m"),
"serial_number": cert.certificate_no or "",
"customer": cert.customer or "",
"project_name": cert.project_name or "",
"template_name": getattr(cert, "template_name", "") or "全中文合格证模板",
"template_type": getattr(cert, "template_type", "") or "全中文",
# 英制信息（用于C5/C6备用列）
"is_imperial": is_imperial,
"lb_level": lb_level,
"imperial_dn": imperial_dn,
"imperial_pn": imperial_pn,
}
@staticmethod
def _build_content_summary(data: dict[str, Any], serial_number: str) -> str:
"""构建打印内容摘要
Args:
data: 打印数据字典
serial_number: 合格证编号（SN）
Returns:
str: 以分号连接的内容摘要字符串
"""
parts = [
f"编号={serial_number}",
f"产品={data.get('product_name', '')}",
f"型号={data.get('product_model', '')}",
f"DN={data.get('dn', '')}",
f"介质={data.get('medium', '')}",
f"温度={data.get('temperature', '')}",
]
return "; ".join(p for p in parts if "=" not in p[-2:])
# ============================================================
# 打印日志记录
# ============================================================
@staticmethod
def _derive_sn_and_kks(
full_code: str, number_model: int = 0, ar28: str = ""
) -> tuple[str, str]:
"""根据 VBA 逻辑从完整编码推导 SN（短编号）和 KKS（长编码）
对应 VBA print.vb 中「将打印内容添加到Access数据库」的赋值逻辑：
- 默认模式: SN (Br(12)) = 去掉中间4位年月，保留 V+流水号+后缀
- 默认模式: KKS (Br(13)) = 完整编码（含年月）
- 编码模式5/6: SN 和 KKS 根据 ar(2,8) 位置互换
- KKS 截断到12位（VBA Br(13) 逻辑）
Args:
full_code: 完整编码，如 "V2604520Y" 或 "V2604123"
number_model: 编码模式 (0=默认, 4=电建三KKS, 5=电建三双编码, 6=合肥水泥双编码)
ar28: ar(2,8) 的值，指示 "B12/B13 哪个是KKS哪个是SN"
Returns:
(SN, KKS) 元组，SN 为短编号，KKS 为完整编码（截断到12位）
"""
if not full_code or not full_code.startswith("V"):
return full_code or "", full_code or ""
# V + 4位年月 + 流水号 + 后缀 = V2604520Y
# SN = V + 流水号 + 后缀（去掉中间4位年月）
sn = full_code[0] + full_code[5:] if len(full_code) > 5 else full_code
kks = full_code
# === 编码模式5/6: SN/KKS 互换（对齐VBA dd.vb:2205-2239）===
if number_model in (5, 6):
if ar28 == "B12单元格为KKS编码，B13单元格为产品编号":
sn, kks = kks, sn
elif ar28 == "B12单元格为产品编号，B13单元格为KKS编码":
pass
# === KKS 截断到12位（对齐VBA dd.vb:2234,2238）===
if len(kks) > 12:
kks = kks[:12]
return sn, kks
@staticmethod
def _printer_name_to_code(printer_name: str) -> int:
"""打印机名称 → 整数代码映射（对齐 VBA dd.vb:2266-2284）
铭牌刻印系统按整数代码解析打印机类型。
Args:
printer_name: 打印机名称字符串
Returns:
int: 打印机代码 (0=未知, 1=Foxit PDF, 2=EPSON针打, 3=网络打印机,
4=HP LaserJet生产, 5=Microsoft XPS, 6=DocuCom PDF, 7=HP LaserJet销售)
"""
if not printer_name:
return 0
mapping = [
("Foxit Reader PDF Printer", 1),
("EPSON LQ-635KII", 2),
("NPI15539C", 3),
("HP LaserJet Pro MFP M225-M226", 4),
("Microsoft XPS Document Writer", 5),
("DocuCom PDF Driver", 6),
]
for keyword, code in mapping:
if keyword in printer_name:
return code
if "HP LaserJet" in printer_name and "M225-M226" in printer_name:
return 4
return 0
@staticmethod
def _clean_temperature(temp: str) -> str:
"""清理温度字符串中的 ≤/℃ 符号（对齐VBA dd.vb:2194-2196）"""
if not temp:
return ""
return temp.replace(" ", "").replace("≤", "").replace("℃", "")
@staticmethod
def _clean_pn_for_access(pn: str, is_imperial: bool = False) -> str:
"""清理 PN 字符串中的单位后缀（对齐VBA dd.vb:2192-2193）"""
if not pn:
return ""
pn = pn.strip()
upper = pn.upper()
if upper.endswith("MPA"):
pn = pn[:-3].strip()
if is_imperial and not upper.endswith("LB"):
pn = f"{pn}Lb"
return pn
[文档]
def record_print_log(
self,
certificate_id: int,
certificate_no: str,
cert_data: dict[str, Any] | None = None,
printer_name: str = "",
status: str = "success",
error_message: str = "",
content_summary: str = "",
) -> PrintLog | None:
"""记录打印日志到 SQLite print_logs 表（字段名与 Access SignTb 兼容）
每次打印操作（单台或批量中的每台）都写入一条记录。
字段命名与 Access SignTb 保持一致，方便铭牌刻印系统直接读取。
SN 和 KKS 赋值遵循 VBA 逻辑：
- SN = 短编号（去掉中间4位年月），如 V520Y
- KKS = 完整编码（含年月），如 V2604520Y
- 编码模式5/6时 SN/KKS 互换
- KKS 截断到12位
字段清理对齐VBA（温度去≤/℃，PN去MPA后缀等）。
Args:
certificate_id: 关联合格证批次ID
certificate_no: 合格证完整编号（如 V2604520Y）
cert_data: 证书完整数据字典
printer_name: 打印机名称
status: 打印状态 (success/failed)
error_message: 错误信息
content_summary: 打印内容摘要
Returns:
PrintLog 对象或 None
"""
try:
cert_data = cert_data or {}
copies = cert_data.get("copies", 1)
# 对齐VBA: 空值填入默认标记（从配置读取）
markers = get_empty_markers()
for key, marker in markers.items():
if not cert_data.get(key):
cert_data[key] = marker
# 按 VBA 逻辑推导 SN（短编号）和 KKS（完整编码）
number_model = cert_data.get("number_model", 0)
sn_kks_position = cert_data.get("sn_kks_position", "") or ""
sn, kks = self._derive_sn_and_kks(certificate_no, number_model, sn_kks_position)
# === 字段清理（对齐VBA dd.vb:2191-2200）===
clean_temp = self._clean_temperature(cert_data.get("temperature", ""))
model = cert_data.get("product_model", "")
is_imperial = PNService.is_imperial(model)
clean_pn = self._clean_pn_for_access(cert_data.get("pn", ""), is_imperial)
dn_raw = cert_data.get("dn", "")
if dn_raw.upper().endswith("MM"):
dn_raw = dn_raw[:-2].strip()
printer_code = self._printer_name_to_code(printer_name)
plan_date = cert_data.get("plan_date", "")
if isinstance(plan_date, str) and plan_date:
try:
plan_date = datetime.strptime(plan_date, "%Y-%m-%d")
except ValueError:
logger.warning(f"无法解析计划日期 [{plan_date}]，使用当前时间")
plan_date = datetime.now()
elif not plan_date:
plan_date = None
sign_model_full_name = cert_data.get("model_full_name", "")
if not sign_model_full_name:
try:
dup_count = self.session.query(PrintLog).filter(sn == PrintLog.SN).count()
sign_model_full_name = f"重复条目数:{dup_count + 1:02d}"
except Exception:
pass
fields = {
# Br(1) ~ Br(13): 基础信息
"JiHuaRiQi": plan_date or datetime.now(),
"DingHuoDanWei": cert_data["customer"],
"XiangMuMingCheng": cert_data["project_name"],
"ChanPinMingCheng": cert_data["product_name"],
"ChanPinXingHao": cert_data["product_model"],
"DN": dn_raw,
"PN": clean_pn,
"ShiYongJieZhi": cert_data["medium"],
"ShiYongWenDu": clean_temp,
"ChuChangRiQiYear": cert_data.get("manufacture_year", 0),
"ChuChangRiQiMonth": cert_data.get("manufacture_month", 0),
"SN": sn,
"KKS": kks,
# Br(14) ~ Br(23): 刻印标记
# Br(24) ~ Br(33): 合格证/铭牌打印信息
"hgz_ShuLiang": copies,
"hgz_YangShi": cert_data.get("template_style", 0),
"hgz_PdfPic": cert_data.get("pic_style", 0),
"hgz_BianMaModel": number_model,
"hgz_xlMuBanName": cert_data.get("template_name") or cert_data.get("template", ""),
"hgz_PrintTime": datetime.now(),
"PrinterName": str(printer_code),
"Sign_PrintCount": 0,
"Sign_ModelFullName": sign_model_full_name,
# Br(35) ~ Br(45): 材质/技术要求/销售信息
"FaTiCaiZhi": cert_data["body_material"],
"FaGanCaiZhi": cert_data["stem_material"],
"QiBiJianCaiZhi": cert_data["disc_material"],
"JiShuYaoQiu": cert_data["tech_requirements"],
"SheBeiWeiHao": cert_data["equipment_no"],
"ZiXiangMu": cert_data["sub_project"],
"SNFromXSB": cert_data["sales_sn"],
"ZhongLei": cert_data["category"],
"ZhongLiang": cert_data["weight"],
"XuHao": cert_data["seq_no"],
"YaoHuoDanHaoFromXSB": cert_data["requisition_no"],
# Br(49) ~ Br(51): 打印模式/水头
"DcOrPlPrint": cert_data.get("print_mode", "1"),
"MaxJinShuiTou": cert_data["max_inlet_head"],
"MaxShengYaShuiTou": cert_data["max_boost_head"],
# 扩展字段
"status": status,
"error_message": error_message,
"content_summary": content_summary,
}
# 单一写点（§11.8 阶段2·B）：保留本方法的 SN/KKS/清洗派生，统一经 record_after_generate 落库
return HistoryBackfillService(self.session).record_after_generate(
certificate_id=certificate_id,
fields=fields,
status=status,
error_message=error_message,
content_summary=content_summary,
)
except Exception as e:
logger.error(f"记录打印日志失败 [{certificate_no}]: {e}")
self.session.rollback()
return None
[文档]
def get_print_history(
self,
limit: int = 100,
status_filter: str | None = None,
) -> list[dict[str, Any]]:
"""查询打印历史（返回字段名与 Access SignTb 兼容）
Args:
limit: 最大返回条数
status_filter: 状态筛选 (success/failed)，None 表示全部
Returns:
打印历史列表，每条记录包含 SignTb 兼容字段
"""
query = self.session.query(PrintLog).order_by(PrintLog.hgz_PrintTime.desc())
if status_filter:
query = query.filter(PrintLog.status == status_filter)
logs = query.limit(limit).all()
return [
{
# 主键和关联
"id": log.id,
"certificate_id": log.certificate_id,
# Br(1) ~ Br(13): 基础信息
"JiHuaRiQi": log.JiHuaRiQi.strftime("%Y-%m-%d") if log.JiHuaRiQi else "",
"DingHuoDanWei": log.DingHuoDanWei,
"XiangMuMingCheng": log.XiangMuMingCheng,
"ChanPinMingCheng": log.ChanPinMingCheng,
"ChanPinXingHao": log.ChanPinXingHao,
"DN": log.DN,
"PN": log.PN,
"ShiYongJieZhi": log.ShiYongJieZhi,
"ShiYongWenDu": log.ShiYongWenDu,
"ChuChangRiQiYear": log.ChuChangRiQiYear,
"ChuChangRiQiMonth": log.ChuChangRiQiMonth,
"SN": log.SN,
"KKS": log.KKS,
# Br(14) ~ Br(23): 刻印标记
"Sign_XingHao": log.Sign_XingHao,
"Sign_DN": log.Sign_DN,
"Sign_PN": log.Sign_PN,
"Sign_JieZhi": log.Sign_JieZhi,
"Sign_WenDu": log.Sign_WenDu,
"Sign_Year": log.Sign_Year,
"Sign_Month": log.Sign_Month,
"Sign_SN": log.Sign_SN,
"Sign_KKS": log.Sign_KKS,
"Sign_ALL": log.Sign_ALL,
# Br(24) ~ Br(33): 合格证/铭牌打印信息
"hgz_ShuLiang": log.hgz_ShuLiang,
"hgz_YangShi": log.hgz_YangShi,
"hgz_PdfPic": log.hgz_PdfPic,
"hgz_BianMaModel": log.hgz_BianMaModel,
"hgz_xlMuBanName": log.hgz_xlMuBanName,
"hgz_PrintTime": log.hgz_PrintTime.strftime("%Y-%m-%d %H:%M:%S")
if log.hgz_PrintTime
else "",
"PrinterName": log.PrinterName,
"Sign_PrintTime": log.Sign_PrintTime.strftime("%Y-%m-%d %H:%M:%S")
if log.Sign_PrintTime
else "",
"Sign_ModelFullName": log.Sign_ModelFullName,
"Sign_PrintCount": log.Sign_PrintCount,
# Br(35) ~ Br(45): 材质/技术要求/销售信息
"FaTiCaiZhi": log.FaTiCaiZhi,
"FaGanCaiZhi": log.FaGanCaiZhi,
"QiBiJianCaiZhi": log.QiBiJianCaiZhi,
"JiShuYaoQiu": log.JiShuYaoQiu,
"SheBeiWeiHao": log.SheBeiWeiHao,
"ZiXiangMu": log.ZiXiangMu,
"SNFromXSB": log.SNFromXSB,
"ZhongLei": log.ZhongLei,
"ZhongLiang": log.ZhongLiang,
"XuHao": log.XuHao,
"YaoHuoDanHaoFromXSB": log.YaoHuoDanHaoFromXSB,
# Br(46) ~ Br(51): 刻印标记/打印模式/水头
"Sign_FTCZ": log.Sign_FTCZ,
"Sign_FGCZ": log.Sign_FGCZ,
"Sign_QBJCZ": log.Sign_QBJCZ,
"DcOrPlPrint": log.DcOrPlPrint,
"MaxJinShuiTou": log.MaxJinShuiTou,
"MaxShengYaShuiTou": log.MaxShengYaShuiTou,
# 扩展字段
"status": log.status,
"error_message": log.error_message,
"content_summary": log.content_summary,
}
for log in logs
]
# ============================================================
# 模板匹配 & 回填
# ============================================================
[文档]
def apply_template_to_certificate(self, cert_id: int, template: dict[str, Any]) -> bool:
"""将 templates.xlsx 中的模板参数应用到 Certificate 记录
Args:
cert_id: Certificate.id
template: 从 TemplateManager.query_by_model 返回的模板 dict
Returns:
bool: 是否成功
"""
cert = self.session.query(Certificate).filter(Certificate.id == cert_id).first()
if not cert:
logger.warning(f"Certificate 不存在: id={cert_id}")
return False
field_map = {
"product_name": "product_name",
"product_model": "product_model",
"dn": "product_spec",
"temperature": "working_temp",
"medium": "working_medium",
"check_standard": "test_standard",
"inspector_id": "inspector_id",
}
updated = 0
for tmpl_key, cert_key in field_map.items():
val = template.get(tmpl_key, "").strip()
if val and hasattr(cert, cert_key):
setattr(cert, cert_key, val)
updated += 1
# === PN 处理 ===
tmpl_pn = template.get("pn", "").strip()
if tmpl_pn:
pn_service = PNService(self.session)
model = cert.product_model or ""
pn_info = pn_service.resolve_pn(model=model, template_pn=tmpl_pn)
cert.pn_value = pn_info["pn_value"]
cert.pn_unit = pn_info["pn_unit"]
cert.pn_display = pn_info["pn_display"]
updated += 1
logger.info(
f"PN 解析: 型号={model}, 模板PN={tmpl_pn} → "
f"value={cert.pn_value}, unit={cert.pn_unit}, display={cert.pn_display}"
)
if template.get("manufacture_date", "").strip():
cert.issue_date = template["manufacture_date"].strip()
self.session.commit()
logger.info(f"应用模板到 cert_id={cert_id}: {updated} 个字段")
return True
[文档]
def writeback_template_to_xlsx(self, cert_id: int, xlsx_path: str) -> bool:
"""打印完成后回填到 templates.xlsx「数据库」表
Args:
cert_id: Certificate.id
xlsx_path: templates.xlsx 文件路径
Returns:
bool: 是否成功
"""
from certflow.handlers.template_handler import TemplateManager
cert = self.session.query(Certificate).filter(Certificate.id == cert_id).first()
if not cert:
return False
tm = TemplateManager(templates_xlsx_path=xlsx_path)
return tm.writeback_to_xlsx(
{
"product_name": cert.product_name,
"product_model": cert.product_model,
"product_spec": cert.product_spec,
"dn": cert.product_spec,
"pn": cert.pn_display or cert.pn_value or "",
"working_temp": cert.working_temp,
"working_medium": cert.working_medium,
"test_standard": cert.test_standard,
"inspector_id": getattr(cert, "inspector_id", ""),
"manufacture_date": cert.issue_date,
"certificate_no": cert.certificate_no,
},
xlsx_path,
)