"""合格证打印日志服务
从 PrintView 迁移出来，负责打印后数据持久化：
- PrintLog 51 字段构建与写入
- Certificate 打印状态更新与回填
- SN/KKS 推导、年月提取等数据清理
对标 VBA: 将打印内容添加到数据库() + 取消自动筛选后静默保存当前工作簿
"""
from __future__ import annotations
import re
from contextlib import suppress
from datetime import datetime
from typing import TYPE_CHECKING, Any
from sqlalchemy.orm import Session
from certflow.services.print_history_service import HistoryBackfillService
from certflow.utils.database import DatabaseManager
from certflow.utils.logger import logger
if TYPE_CHECKING:
from certflow.models.print_log import PrintLog
def _cfg_get(key: str, default: Any) -> Any:
"""读取配置（隔离导入失败场景，配置不可用时回退默认）。"""
try:
from certflow.config.settings import cfg
except ImportError:
return default
return cfg(key, default)
def _resolve_duplicate_config() -> tuple[bool, list[str], list[str], int]:
"""解析 ``certificate.print_duplicate_check`` 配置。
兼容 dict（测试桩）与 pydantic 模型（运行时配置）两种形态。
返回 ``(enabled, customers, compare_fields, threshold)``；配置缺失时返回全关。
"""
cfg_root = _cfg_get("certificate.print_duplicate_check", None)
if cfg_root is None:
return False, [], [], 0
if hasattr(cfg_root, "enabled"):
enabled = bool(cfg_root.enabled)
customers = list(cfg_root.customers or [])
compare_fields = list(cfg_root.compare_fields or [])
threshold = int(cfg_root.threshold or 0)
else:
enabled = bool(cfg_root.get("enabled", False))
customers = cfg_root.get("customers") or []
compare_fields = cfg_root.get("compare_fields") or []
threshold = int(cfg_root.get("threshold", 0) or 0)
return enabled, customers, compare_fields, threshold
def _count_matched_fields(log: PrintLog, rec: PrintLog, fields: list[str]) -> int:
"""统计两条 PrintLog 在给定字段上相等的数量（含双 None 视为相等）。"""
matched = 0
for field in fields:
cur = getattr(log, field, None)
old = getattr(rec, field, None)
if (
cur is None
and old is None
or cur is not None
and old is not None
and str(cur) == str(old)
):
matched += 1
return matched
[文档]
def is_duplicate_print_record(session: Session, log: PrintLog) -> bool:
"""打印存库重复检测（§5.7.2-C / 维度11，对齐 VBA「大连大高重复跳过存库」）。
仅当配置 ``certificate.print_duplicate_check.enabled=true`` 且当前打印记录的
订货单位（``DingHuoDanWei``）命中 ``customers`` 关键词时启用：在 PrintLog 表中
查找与当前记录在 ``compare_fields`` 上字段值相等数 ≥ ``threshold`` 的既有记录，
命中则返回 True（调用方跳过新增 PrintLog 行，仅更新合格证清单/ Certificate 状态）。
默认配置 ``enabled=false`` → 永远返回 False，零差异（所有客户照常落库）。
Args:
session: SQLAlchemy 会话。
log: 已构建但未落库的 PrintLog 对象（取 ``DingHuoDanWei`` / ``compare_fields`` 值）。
Returns:
bool: 是否应跳过落库（True=重复，跳过）。
"""
enabled, customers, compare_fields, threshold = _resolve_duplicate_config()
if not enabled or not compare_fields or threshold <= 0:
return False
if customers:
customer = str(getattr(log, "DingHuoDanWei", "") or "")
if not any(str(c) in customer for c in customers):
return False
from certflow.models.print_log import PrintLog
# 逐条比对既有 PrintLog（阈值通常较小，全表扫可接受；大连大高场景量可控）
existing = session.query(PrintLog).all()
return any(_count_matched_fields(log, rec, compare_fields) >= threshold for rec in existing)
# 检验工号默认（按模板语言族，BUG-006 I / H3）：
# 仅全中文默认「检5」；其余语言族（中英文 / 全英文 / 俄英文）默认「No 5」。
# 真实打印流程中操作者可在参数补充对话框录入具体工号；未录入时回退到此默认值。
# 主路径改为配置驱动（config/certificate.yaml 的 certificate.inspector_id），
# 下方硬编码仅作为「配置不可用」时的最终回退，保持既有行为。
_INSPECTOR_ID_DEFAULTS = {
"全中文": "检5",
"中英文": "No 5",
"全英文": "No 5",
"俄英文": "No 5",
}
def _default_inspector_id(template_type: str) -> str:
"""按模板语言族返回检验工号默认值（配置驱动，回退硬编码）。
读取 ``certificate.inspector_id``：
- ``enabled=False`` → 返回空串（不自动填默认工号，由操作者手动录入）。
- 命中 ``defaults[语言族]`` → 该值；否则用 ``fallback``。
配置不可用时回退 ``_INSPECTOR_ID_DEFAULTS``（兼容既有行为）。
"""
try:
from certflow.config.settings import cfg
inspector_id = cfg("certificate.inspector_id", None)
except ImportError:
inspector_id = None
if inspector_id:
if not bool(inspector_id.get("enabled", True)):
return ""
defaults = inspector_id.get("defaults", {}) or {}
if template_type in defaults:
return str(defaults[template_type])
return str(inspector_id.get("fallback", "检5"))
# 配置不可用：回退硬编码默认
return _INSPECTOR_ID_DEFAULTS.get(template_type, "检5")
[文档]
class CertLogService:
"""打印日志保存服务"""
def __init__(self, session: Session, certificate_id: int | None = None) -> None:
"""初始化打印日志服务。
Args:
session: SQLAlchemy 数据库会话。
certificate_id: 关联的合格证 ID，可为 None（仅构建日志时不回填）。
"""
self.session = session
self.certificate_id = certificate_id
# ============================================================
# SN/KKS 推导
# ============================================================
[文档]
@staticmethod
def derive_sn_kks(serial: str) -> tuple[str, str]:
"""从完整编码推导 SN（短编号）和 KKS（完整编码）
- SN = 去掉中间4位年月，如 "V2512123Y" → "V123Y"
- KKS = 完整编码原样返回
"""
m = re.match(r"^([A-Z])(\d{4})(\d+)([A-Z]?)$", serial)
if m:
prefix, seq, suffix = m.group(1), int(m.group(3)), m.group(4) or ""
return f"{prefix}{seq:03d}{suffix}", serial
return serial, serial
[文档]
@staticmethod
def derive_manufacture_ym(serial: str) -> tuple[int | None, int | None]:
"""从完整编码提取出厂年月，例如 "V2512123Y" → (25, 12)"""
m = re.match(r"^[A-Z](\d{2})(\d{2})\d+[A-Z]?$", serial)
if m:
return int(m.group(1)), int(m.group(2))
return None, None
# ============================================================
# 数据获取
# ============================================================
[文档]
def get_cert_data(self) -> dict[str, Any]:
"""从 Certificate 表获取当前批次的业务数据，关联 SalePlan 获取销售信息。
Returns:
dict[str, Any]: 合格证及关联销售计划的业务数据字典。
"""
from certflow.models.certificate import Certificate
from certflow.models.sale_plan import SalePlan
if not self.certificate_id:
return {}
try:
cert = self.session.get(Certificate, self.certificate_id)
if not cert:
return {}
sp_data = {}
if cert.sale_plan_id:
sp = self.session.get(SalePlan, cert.sale_plan_id)
if sp:
sp_data = {
"product_code": sp.product_code or "",
"category": sp.category or "",
"weight": str(sp.weight) if sp.weight else "",
"sort_order": str(sp.sort_order) if sp.sort_order else "",
"plan_no": sp.plan_no or "",
}
return {
"plan_date": cert.plan_date or "",
"customer": cert.customer or "",
"project_name": cert.project_name or "",
"product_name": cert.product_name or "",
"product_model": cert.product_model or "",
"dn": cert.product_spec or "",
"pn": cert.pn_display or cert.pn_value or "",
"temperature": cert.working_temp or "",
"medium": cert.working_medium or "",
"check_standard": cert.test_standard or "",
"fati_material": cert.fati_material or "",
"fagan_material": cert.fagan_material or "",
"qibijian_material": cert.qibijian_material or "",
"tech_requirements": cert.tech_requirements or "",
"sales_order_no": cert.sales_order_no or "",
"production_order_no": cert.production_order_no or "",
"quantity": cert.quantity or 1,
**sp_data,
}
except Exception as e:
logger.warning(f"[CertLog] 获取 Certificate 数据失败: {e}")
return {}
[文档]
def get_sn_dup_count(self, sn: str) -> str:
"""查询 SN 重复条目数，格式 "重复条目数:02" """
from certflow.models.print_log import PrintLog
if not sn or not self.session:
return "重复条目数:01"
try:
count = self.session.query(PrintLog).filter(sn == PrintLog.SN).count()
return f"重复条目数:{count + 1:02d}"
except Exception:
return "重复条目数:01"
# ============================================================
# PrintLog 构建
# ============================================================
[文档]
def build_print_log(
self, serial: str, data: dict[str, Any], printer_name: str, now: datetime | None = None
) -> PrintLog:
"""构建单条 PrintLog 记录（51字段对齐 Access SignTb）。
Args:
serial: 产品完整编码。
data: 表单/打印数据字典。
printer_name: 打印机名称。
now: 打印时间戳，为 None 时使用当前时间。
Returns:
PrintLog: 已构建但尚未提交的打印日志对象。
"""
from certflow.models.vba_mapping import VBAMapping
now = now or datetime.now()
sn, kks = self.derive_sn_kks(serial)
# §5.7.2-B：清单保存用 list 分隔符（默认 "-"，如 V2607099Y-1000Y），
# 与编号阶段 stage 分隔符（"---"）区分；PrintLog/SN/KKS 落库即清单形态。
from certflow.services.certificate_number_service import CertificateNumberService
sn = CertificateNumberService.to_list_format(sn)
kks = CertificateNumberService.to_list_format(kks)
year, month = self.derive_manufacture_ym(serial)
cert_data = self.get_cert_data()
def _val(key: str, default: str = "") -> str:
v = data.get(key) or cert_data.get(key)
return str(v) if v else default
# 温度/DN/PN 清理
temp = _val("temperature", "").replace(" ", "").replace("≤", "").replace("℃", "")
dn = _val("dn", "")
if dn.upper().startswith("DN"):
dn = dn[2:]
if dn.upper().endswith("MM"):
dn = dn[:-2]
pn = _val("pn", "")
if pn.upper().endswith("MPA"):
pn = pn[:-3]
if data.get("product_model", "").upper().find("LB") > 0 and not pn.upper().endswith("LB"):
pn = f"{pn}Lb"
jhrq = None
plan_date_str = cert_data.get("plan_date", "")
if plan_date_str:
with suppress(ValueError):
jhrq = datetime.strptime(plan_date_str, "%Y-%m-%d")
template_type = data.get("template_type", "全中文")
number_mode = data.get("number_mode", "直接复制")
style_code = VBAMapping.get_style_code(self.session, template_type)
number_code = VBAMapping.get_number_mode_code(self.session, number_mode)
printer_code = VBAMapping.get_printer_code(self.session, printer_name)
fields = {
"JiHuaRiQi": jhrq or now,
"DingHuoDanWei": _val("customer", "【空白订货单位】"),
"XiangMuMingCheng": _val("project_name", "【空白项目名称】"),
"ChanPinMingCheng": _val("product_name", "【空白产品名称】"),
"ChanPinXingHao": _val("product_model", "【空白产品型号】"),
"DN": dn,
"PN": pn,
"ShiYongJieZhi": _val("medium", "【空白适用介质】"),
"ShiYongWenDu": temp,
"ChuChangRiQiYear": year or 1900,
"ChuChangRiQiMonth": month or 1,
"SN": sn,
"KKS": kks,
"hgz_ShuLiang": data.get("quantity", cert_data.get("quantity", 1)),
"hgz_YangShi": style_code,
"hgz_BianMaModel": number_code,
"hgz_xlMuBanName": data.get("template_name") or data.get("template_type", ""),
"hgz_PrintTime": now,
# B3：铭牌打印时间 = 合格证打印时间（打印即铭牌刻印产出，合理回填）；
# 其余 Sign_* 刻印标记（Br14-23/46-48）与 水头字段（Br50/51）由铭牌刻印
# 环节置位 / 暂无业务数据源，保持模型默认（False / 空），不臆造。
"Sign_PrintTime": now,
"PrinterName": str(printer_code),
"Sign_PrintCount": 0,
"Sign_ModelFullName": self.get_sn_dup_count(sn),
"FaTiCaiZhi": _val("fati_material", "【未录入阀体材质】"),
"FaGanCaiZhi": _val("fagan_material", "【未录入阀杆材质】"),
"QiBiJianCaiZhi": _val("qibijian_material", "【未录入启闭件材质】"),
"JiShuYaoQiu": cert_data.get("tech_requirements", "【空白技术要求】"),
"SheBeiWeiHao": cert_data.get("sales_order_no", ""),
"ZiXiangMu": cert_data.get("production_order_no", ""),
"SNFromXSB": cert_data.get("product_code", ""),
"ZhongLei": cert_data.get("category", ""),
"ZhongLiang": cert_data.get("weight", ""),
"XuHao": cert_data.get("sort_order", ""),
"YaoHuoDanHaoFromXSB": cert_data.get("plan_no", ""),
"DcOrPlPrint": "1",
"status": "success",
"content_summary": f"{_val('product_name')} {_val('product_model')}",
}
# 单一 PrintLog 构造器（§11.8 阶段2·B）：消除 cert_log_service / certificate_print_service 重复构造
return HistoryBackfillService(self.session).assemble_print_log(
certificate_id=self.certificate_id, fields=fields
)
# ============================================================
# Certificate 回填
# ============================================================
[文档]
def update_certificate_status(
self, data: dict[str, Any], printer_name: str, now: datetime | None = None
) -> None:
"""更新 Certificate 打印状态，用表单数据覆盖（以实际打印内容为准）。
Args:
data: 表单/打印数据字典。
printer_name: 打印机名称。
now: 打印时间戳，为 None 时使用当前时间。
"""
from certflow.models.certificate import Certificate
if not self.certificate_id:
return
cert = self.session.get(Certificate, self.certificate_id)
if not cert:
return
cert.print_status = "已打印"
cert.print_time = now or datetime.now()
cert.printer_name = printer_name
# A3：落库写证/打印时选择的模板语言族，供重打恢复
cert.template_type = data.get("template_type", "全中文")
cert.working_temp = data.get("temperature", "")
cert.working_medium = data.get("medium", "")
cert.test_standard = data.get("check_standard", "")
# H3：真实回填检验工号（此前漏写）；未提供时按模板语言族取默认值
cert.inspector_id = data.get("inspector_id") or _default_inspector_id(
data.get("template_type", "全中文")
)
if data.get("pn"):
cert.pn_display = data["pn"]
# ============================================================
# 批量保存
# ============================================================
[文档]
def get_recent_logs(self, limit: int = 200) -> list[PrintLog]:
"""获取最近的打印日志
Args:
limit: 返回条数上限，默认 200
Returns:
PrintLog 列表，按 id 降序排列
"""
from certflow.models.print_log import PrintLog
try:
return self.session.query(PrintLog).order_by(PrintLog.id.desc()).limit(limit).all()
except Exception as e:
logger.error(f"获取打印历史失败: {e}")
return []
[文档]
def query_logs(
self,
filters: dict | None = None,
limit: int = 2000,
) -> list[PrintLog]:
"""按条件查询打印日志（参数化，防注入）
支持按日期区间（hgz_PrintTime 或 JiHuaRiQi）与产品型号模糊搜索。
Args:
filters: 过滤条件字典，可选键：
date_field: "hgz_PrintTime"(默认) 或 "JiHuaRiQi"
date_from: datetime 起始（含）
date_to: datetime 结束（含）
model_keyword: 产品型号模糊关键字（LIKE，不区分大小写）
limit: 返回条数上限，默认 2000
Returns:
PrintLog 列表，按 id 降序排列
"""
from certflow.models.print_log import PrintLog
filters = filters or {}
try:
query = self.session.query(PrintLog)
date_field = filters.get("date_field", "hgz_PrintTime")
date_col = (
PrintLog.hgz_PrintTime if date_field == "hgz_PrintTime" else PrintLog.JiHuaRiQi
)
date_from = filters.get("date_from")
if date_from is not None:
query = query.filter(date_col >= date_from)
date_to = filters.get("date_to")
if date_to is not None:
query = query.filter(date_col <= date_to)
keyword = filters.get("model_keyword")
if keyword:
query = query.filter(PrintLog.ChanPinXingHao.ilike(f"%{keyword}%"))
return query.order_by(PrintLog.id.desc()).limit(limit).all()
except Exception as e:
logger.error(f"查询打印历史失败: {e}")
return []
[文档]
def auto_save(self, serials: list[str], data: dict[str, Any], printer_name: str) -> None:
"""逐台保存 PrintLog + 更新 Certificate，统一提交。
Args:
serials: 产品编码列表（逐台打印）。
data: 表单/打印数据字典。
printer_name: 打印机名称。
"""
if not self.session:
logger.warning("[CertLog] 无数据库会话，跳过保存")
return
now = datetime.now()
failures: list[str] = []
skipped_duplicates: list[str] = []
for serial in serials:
try:
log = self.build_print_log(serial, data, printer_name, now)
# §5.7.2-C：指定客户打印时检测 PrintLog 重复，命中则跳过落库行
# （仅更新合格证清单/ Certificate 状态，与 VBA「重复跳过存库」一致）。
if is_duplicate_print_record(self.session, log):
skipped_duplicates.append(serial)
logger.info(f"[CertLog] 检测到重复打印记录，跳过落库: {serial}")
continue
self.session.add(log)
except Exception as e:
failures.append(f"{serial}: {e}")
logger.error(f"[CertLog] 添加打印日志失败 [{serial}]: {e}")
try:
self.update_certificate_status(data, printer_name, now)
except Exception as e:
logger.error(f"[CertLog] 更新 Certificate 状态失败: {e}")
# G3: 打印后回填 SalePlan 的 sn_code/kks_code（仅在该计划全部台数打印完毕）
if self.certificate_id:
backfill_dual_codes_if_fully_printed(self.session, self.certificate_id)
if failures:
# F3(a): 不再静默吞异常谎报成功——汇总失败并抛出，交由调用方处理
summary = "; ".join(failures[:3])
more = "" if len(failures) <= 3 else f" 等共 {len(failures)} 条"
raise RuntimeError(
f"[CertLog] {len(failures)}/{len(serials)} 条打印日志保存失败: {summary}{more}"
)
try:
DatabaseManager.commit_with_retry(self.session)
skip_msg = f"，跳过重复 {len(skipped_duplicates)} 条" if skipped_duplicates else ""
logger.info(f"[CertLog] 已保存 {len(serials)} 条打印记录{skip_msg}")
except Exception as e:
logger.error(f"[CertLog] 提交打印记录失败: {e}")
self.session.rollback()
raise
# G3 回填已在 commit 前（update_certificate_status 之后）执行，此处无需重复。
[文档]
def backfill_dual_codes_if_fully_printed(session: Session, certificate_id: int) -> bool:
"""G3: 打印后回填 SalePlan 的 sn_code/kks_code。
设计意图（§11.4.1）：SN/KKS 为「打印后记录」字段——编号阶段不再派生，
打印时逐台由 product_code_range 正则拆分得到短码 SN(如 ``V520Y``)/长码
KKS(如 ``V2604520Y``) 写入 PrintLog；SalePlan 的 sn_code/kks_code 仅在
该计划全部台数（quantity）对应的 PrintLog 均已写入时才回填。
G5 补充触发点：``SalePlan.fully_shipped``（全部台数发完，等价
``shipping_status='已发货'``）为真时也执行回填——满足「全部发完才回填」
的语义，且不必等到逐台 PrintLog 全部落库。
回填值由**实际打印的编号**推导（补丁36 修正）：优先取
``Certificate.product_code_range``（用户可能手动修订过编号，打印出的
serial 即来自此）；仅在空时回退 ``SalePlan.product_code``。
一般件 kks_code=长 V 码区间串、sn_code=剥 YYMM 短码；电厂件已录入的真
KKS（不以 V 开头）受保护不被覆盖。
Args:
session: SQLAlchemy 会话
certificate_id: 本次打印关联的 Certificate.id
Returns:
bool: 是否执行了回填（False 表示尚未全部打印/发完或无需回填）
"""
from certflow.models.certificate import Certificate
from certflow.models.print_log import PrintLog
from certflow.models.sale_plan import SalePlan
cert = session.get(Certificate, certificate_id)
if not cert or not cert.sale_plan_id:
return False
sp = session.get(SalePlan, cert.sale_plan_id)
if not sp or not (sp.quantity or 0) > 0:
return False
# G5 触发点①：全部台数已发完（fully_shipped）即回填。
# 触发点②（G3 原逻辑）：统计该 SalePlan 下所有 Certificate 关联的
# PrintLog 台数，仅当全部台数打印完毕才回填。两者满足其一即可。
if not sp.fully_shipped:
cert_ids = [
c.id
for c in session.query(Certificate.id).filter(Certificate.sale_plan_id == sp.id).all()
]
if not cert_ids:
return False
printed = session.query(PrintLog).filter(PrintLog.certificate_id.in_(cert_ids)).count()
if printed < sp.quantity:
return False
# 补丁36 修正：回填必须使用「实际打印的编号」，而非 SalePlan 原始 product_code。
# 用户可能手动修订过编号，打印出的 serial 来自 Certificate.product_code_range；
# 若取 sp.product_code 会得到与原编号一致、但非实际打印值的 sn/kks
# （日志实证：实际打印 V2607460/461，旧逻辑回填 V001---002）。
from certflow.services.dual_code import _is_real_kks, to_sn_code
printed = (cert.product_code_range or "").strip() or (sp.product_code or "").strip()
# 电厂件真 KKS（不以 V 开头）受保护，不被实际打印的 V 码覆盖（B1-sn-elec）
if not _is_real_kks(sp.kks_code or ""):
sp.kks_code = printed
sp.sn_code = to_sn_code(printed) if printed else to_sn_code(sp.product_code or "")
logger.info(
f"[CertLog] G3 回填 sn/kks | sale_plan_id={sp.id} | sn={sp.sn_code!r} kks={sp.kks_code!r}"
f" | 来源={'实际打印=' + cert.product_code_range if cert.product_code_range else 'sp.product_code'}"
)
return True