"""合格证数据导出服务
从 CertificatePrintService 迁移出来，负责：
- Access 数据库导出 (SignDB.mdb)
- SignTb CSV 导出（Access 桥接方案）
- 打印历史查询
"""
from __future__ import annotations
import csv
import os
import re
from datetime import datetime
from typing import Any
from sqlalchemy.orm import Session
from certflow.utils.logger import logger
# SN 短码派生（与 cert_log_service.CertLogService.derive_sn_kks 同源）：
# 完整编号 "V2512123Y" → 剥掉中间 4 位年月 → 短码 "V123Y"。
# 区间串（含 "---"）无法拆单台，回退取首段完整编号的短码。
_SN_RE = re.compile(r"^([A-Z])(\d{4})(\d+)([A-Z]?)$")
def _to_short_sn(code: str) -> str:
"""从完整产品编号（或区间串首段）派生 SN 短码。
- 单号 ``V2512123Y`` → ``V123Y``（剥掉 YYMM）。
- 区间串 ``V2604520Y---524Y`` → 取首段 ``V2604520Y`` 派生 ``V520Y``。
- 非 V 体系（如电厂件真 KKS）→ 原样返回。
"""
if not code:
return code
first = code.split("---")[0].split("-")[0].strip()
m = _SN_RE.match(first)
if m:
prefix, _ym, seq, suffix = m.group(1), m.group(2), int(m.group(3)), m.group(4) or ""
return f"{prefix}{seq:03d}{suffix}"
return code
[文档]
class CertExportService:
"""合格证数据导出服务"""
def __init__(self, session: Session) -> None:
"""初始化合格证导出服务。
Args:
session: SQLAlchemy 数据库会话。
"""
self.session = session
# ============================================================
# Access 导出
# ============================================================
[文档]
def export_to_access(self, ids: list[int], mdb_path: str | None = None) -> int:
"""导出到 Access 数据库（默认禁用）
控制方式（优先级从高到低）：
1. 环境变量 CERTFLOW_ENABLE_ACCESS_EXPORT=true 强制启用
2. 配置文件 config.yaml → export.access.enabled: true
3. 以上均未设置时默认禁用
注意: 64-bit Python 环境可能无法连接 32-bit Access ODBC 驱动。
推荐使用 record_print_log 记录到 SQLite print_logs 表，或使用 export_sign_tb_csv 导出 CSV。
Args:
ids: 合格证记录 ID 列表。
mdb_path: Access 数据库路径，为 None 时按配置或默认值解析。
Returns:
int: 成功写入的记录数；导出被禁用或失败时返回 0。
"""
from certflow.utils.access_db import is_access_enabled
if not is_access_enabled():
logger.debug(
"Access 导出已禁用"
"（设置 export.access.enabled=true 或 CERTFLOW_ENABLE_ACCESS_EXPORT=true 启用）"
)
return 0
from certflow.models.certificate import Certificate
if mdb_path is None:
try:
from certflow.config.settings import get_config_value
mdb_path = get_config_value(
"paths.access.primary", default="D:\\DataSource\\SignDB.mdb"
)
except Exception as e:
logger.warning(f"读取 Access 路径配置失败，使用默认路径: {e}")
mdb_path = "D:\\DataSource\\SignDB.mdb"
certs = self.session.query(Certificate).filter(Certificate.id.in_(ids)).all()
try:
from certflow.utils.access_db import _import_pyodbc
pyodbc = _import_pyodbc()
conn_str = f"Driver={{Microsoft Access Driver (*.mdb, *.accdb)}};DBQ={mdb_path};"
conn = pyodbc.connect(conn_str)
cursor = conn.cursor()
for cert in certs:
# B2：SN/KKS 同源 certificate_no 是 bug——KKS 取完整产品编号（长码），
# SN 取剥掉年月(YYMM)的短码；优先用实际打印编号 product_code_range。
kks = (cert.product_code_range or cert.certificate_no or "").strip()
sn = _to_short_sn(kks) if kks else (cert.certificate_no or "")
cursor.execute(
"""
INSERT INTO SignTb (
JiHuaRiQi, DingHuoDanWei, XiangMuMingCheng,
ChanPinMingCheng, ChanPinXingHao, DN, PN,
ShiYongJieZhi, ShiYongWenDu, SN, KKS,
hgz_ShuLiang, hgz_PrintTime, PrinterName
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
cert.plan_date,
cert.customer,
cert.project_name,
cert.product_name,
cert.product_model,
cert.product_spec,
cert.pn_display or cert.pn_value or "",
cert.working_medium or "",
cert.working_temp or "",
sn,
kks,
cert.quantity,
cert.print_time,
cert.printer_name or "",
),
)
conn.commit()
conn.close()
logger.info(f"写入 Access 完成 | 数量={len(certs)}")
return len(certs)
except ImportError:
logger.warning("pyodbc 未安装，跳过 Access 导出")
return 0
except Exception as e:
logger.error(f"写入 Access 失败: {e}")
return 0
# ============================================================
# CSV 导出 (SignTb 兼容)
# ============================================================
[文档]
def export_sign_tb_csv(self, output_path: str | None = None, limit: int = 1000) -> str | None:
"""导出 print_logs 为 SignTb 兼容 CSV（Access 桥接方案）
CSV 列序完全对齐 Access SignTb 51字段，铭牌刻印系统可导入。
Args:
output_path: 输出 CSV 路径，为 None 时自动生成到 output/export 目录。
limit: 最多导出的打印日志条数。
Returns:
str | None: 成功时返回输出文件路径，无数据或失败时返回 None。
"""
from certflow.models.print_log import PrintLog
if output_path is None:
os.makedirs("output/export", exist_ok=True)
output_path = (
f"output/export/SignTb_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
)
logs = (
self.session.query(PrintLog).order_by(PrintLog.hgz_PrintTime.desc()).limit(limit).all()
)
if not logs:
logger.warning("无打印日志可导出")
return None
fieldnames = [
"JiHuaRiQi",
"DingHuoDanWei",
"XiangMuMingCheng",
"ChanPinMingCheng",
"ChanPinXingHao",
"DN",
"PN",
"ShiYongJieZhi",
"ShiYongWenDu",
"ChuChangRiQiYear",
"ChuChangRiQiMonth",
"SN",
"KKS",
"Sign_XingHao",
"Sign_DN",
"Sign_PN",
"Sign_JieZhi",
"Sign_WenDu",
"Sign_Year",
"Sign_Month",
"Sign_SN",
"Sign_KKS",
"Sign_ALL",
"hgz_ShuLiang",
"hgz_YangShi",
"hgz_PdfPic",
"hgz_BianMaModel",
"hgz_xlMuBanName",
"hgz_PrintTime",
"PrinterName",
"Sign_PrintTime",
"Sign_ModelFullName",
"Sign_PrintCount",
"FaTiCaiZhi",
"FaGanCaiZhi",
"QiBiJianCaiZhi",
"JiShuYaoQiu",
"SheBeiWeiHao",
"ZiXiangMu",
"SNFromXSB",
"ZhongLei",
"ZhongLiang",
"XuHao",
"YaoHuoDanHaoFromXSB",
"Sign_FTCZ",
"Sign_FGCZ",
"Sign_QBJCZ",
"DcOrPlPrint",
"MaxJinShuiTou",
"MaxShengYaShuiTou",
]
try:
with open(output_path, "w", newline="", encoding="utf-8-sig") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
for log in logs:
writer.writerow(self._log_to_dict(log))
logger.info(f"SignTb CSV 导出完成: {output_path} ({len(logs)} 条)")
return output_path
except Exception as e:
logger.error(f"SignTb CSV 导出失败: {e}")
return None
@staticmethod
def _log_to_dict(log: Any) -> dict[str, Any]:
"""PrintLog → SignTb CSV 行字典。
Args:
log: PrintLog 记录实例。
Returns:
dict[str, Any]: 与 Access SignTb 字段对齐的 CSV 行字典。
"""
return {
"JiHuaRiQi": log.JiHuaRiQi.strftime("%Y-%m-%d") if log.JiHuaRiQi else "",
"DingHuoDanWei": log.DingHuoDanWei or "",
"XiangMuMingCheng": log.XiangMuMingCheng or "",
"ChanPinMingCheng": log.ChanPinMingCheng or "",
"ChanPinXingHao": log.ChanPinXingHao or "",
"DN": log.DN or "",
"PN": log.PN or "",
"ShiYongJieZhi": log.ShiYongJieZhi or "",
"ShiYongWenDu": log.ShiYongWenDu or "",
"ChuChangRiQiYear": log.ChuChangRiQiYear or 0,
"ChuChangRiQiMonth": log.ChuChangRiQiMonth or 0,
"SN": log.SN or "",
"KKS": log.KKS or "",
"Sign_XingHao": 1 if log.Sign_XingHao else 0,
"Sign_DN": 1 if log.Sign_DN else 0,
"Sign_PN": 1 if log.Sign_PN else 0,
"Sign_JieZhi": 1 if log.Sign_JieZhi else 0,
"Sign_WenDu": 1 if log.Sign_WenDu else 0,
"Sign_Year": 1 if log.Sign_Year else 0,
"Sign_Month": 1 if log.Sign_Month else 0,
"Sign_SN": 1 if log.Sign_SN else 0,
"Sign_KKS": 1 if log.Sign_KKS else 0,
"Sign_ALL": 1 if log.Sign_ALL else 0,
"hgz_ShuLiang": log.hgz_ShuLiang or 1,
"hgz_YangShi": log.hgz_YangShi or 0,
"hgz_PdfPic": log.hgz_PdfPic or 0,
"hgz_BianMaModel": log.hgz_BianMaModel or 0,
"hgz_xlMuBanName": log.hgz_xlMuBanName or "",
"hgz_PrintTime": log.hgz_PrintTime.strftime("%Y-%m-%d %H:%M:%S")
if log.hgz_PrintTime
else "",
"PrinterName": log.PrinterName or "",
"Sign_PrintTime": log.Sign_PrintTime.strftime("%Y-%m-%d %H:%M:%S")
if log.Sign_PrintTime
else "",
"Sign_ModelFullName": log.Sign_ModelFullName or "",
"Sign_PrintCount": log.Sign_PrintCount or 0,
"FaTiCaiZhi": log.FaTiCaiZhi or "",
"FaGanCaiZhi": log.FaGanCaiZhi or "",
"QiBiJianCaiZhi": log.QiBiJianCaiZhi or "",
"JiShuYaoQiu": log.JiShuYaoQiu or "",
"SheBeiWeiHao": log.SheBeiWeiHao or "",
"ZiXiangMu": log.ZiXiangMu or "",
"SNFromXSB": log.SNFromXSB or "",
"ZhongLei": log.ZhongLei or "",
"ZhongLiang": log.ZhongLiang or "",
"XuHao": log.XuHao or "",
"YaoHuoDanHaoFromXSB": log.YaoHuoDanHaoFromXSB or "",
"Sign_FTCZ": 1 if log.Sign_FTCZ else 0,
"Sign_FGCZ": 1 if log.Sign_FGCZ else 0,
"Sign_QBJCZ": 1 if log.Sign_QBJCZ else 0,
"DcOrPlPrint": log.DcOrPlPrint or "",
"MaxJinShuiTou": log.MaxJinShuiTou or "",
"MaxShengYaShuiTou": log.MaxShengYaShuiTou or "",
}
# ============================================================
# 打印历史查询
# ============================================================
[文档]
def get_print_history(self, limit: int = 100, status_filter: str | None = None) -> list[dict]:
"""查询打印历史（返回字段名与 Access SignTb 兼容）。
Args:
limit: 最多返回的记录条数。
status_filter: 按打印状态过滤，为 None 时不筛选。
Returns:
list[dict]: 打印历史记录列表（字段名与 SignTb 对齐）。
"""
from certflow.models.print_log import PrintLog
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 [self._log_to_dict(log) for log in logs]