# src/certflow/services/scan_service.py
"""合格证扫描件生成服务
提供扫描件生成的业务逻辑：
- 批量生成扫描件
- 模板坐标管理
- 调用 Handler 进行底层绘制
"""
from __future__ import annotations
import contextlib
import os
import re
from collections.abc import Callable
from typing import TYPE_CHECKING, Any
from PIL import Image
from certflow.handlers.scan_handler import (
ScanExportHandler,
ScanImageHandler,
ScanLayoutHandler,
)
from certflow.services.cert_numbering_policy import parse_product_code_meta
from certflow.utils.logger import logger
if TYPE_CHECKING:
from sqlalchemy.orm import Session
from certflow.models.certificate import Certificate
[文档]
class ScanService:
"""扫描件生成服务
封装扫描件生成的业务逻辑，支持单张与多张拼接两种模式，
并委托底层 Handler 完成画布创建、字段绘制与文件导出。
"""
def __init__(self, session: Session | None = None) -> None:
"""初始化扫描件服务
Args:
session: SQLAlchemy 会话对象，用于查询 Certificate，可为 None。
"""
self.session = session
self.image_handler = ScanImageHandler()
self._coordinates_cache = {}
[文档]
def load_certificates(self, certificate_ids: list[int]) -> list[Certificate]:
"""加载合格证记录
Args:
certificate_ids: 合格证 ID 列表
Returns:
合格证列表
"""
from certflow.models.certificate import Certificate
return self.session.query(Certificate).filter(Certificate.id.in_(certificate_ids)).all()
[文档]
def generate_scans(
self,
certificate_ids: list[int],
output_dir: str,
template_type: str = "全中文",
output_format: str = "JPG",
rows: int = 1,
cols: int = 1,
paper_width_mm: float = 60,
paper_height_mm: float = 100,
data_overrides: dict[int, dict] | None = None,
dual_code: bool = False,
progress_callback: Callable[[int, int, str], None] | None = None,
) -> dict[str, Any]:
"""生成扫描件
Args:
certificate_ids: 合格证 ID 列表
output_dir: 输出目录
template_type: 模板类型 (全中文/中英文/俄英文)
output_format: 输出格式 (JPG/PDF)
rows: 每页行数
cols: 每页列数
paper_width_mm: 纸张宽度 (mm)
paper_height_mm: 纸张高度 (mm)
progress_callback: 进度回调函数 fn(current, total, message)
Returns:
生成结果字典: {
"success": int, # 成功生成的文件数
"failed": int, # 失败的数量
"files": list[str], # 生成的文件路径列表
"message": str # 结果消息
}
"""
from certflow.models.certificate import Certificate
# 查询合格证
certificates = (
self.session.query(Certificate).filter(Certificate.id.in_(certificate_ids)).all()
)
if not certificates:
return {
"success": 0,
"failed": len(certificate_ids),
"files": [],
"message": "未找到合格证记录",
}
# 加载模板坐标
coordinates = self._load_coordinates(template_type)
# 逐台展开：每个 Certificate 为批次级（product_code_range 是区间串，如
# "V2607050-057"），需拆分为单台编号 V2607050..V2607057，逐台生成「一证一号」
# 的扫描件（而非整批共用区间串）。多张拼接（拼版）时同样按单台单元排列。
units = self._expand_units(certificates, data_overrides)
total = len(units)
layout = ScanLayoutHandler(rows, cols)
export_handler = ScanExportHandler(output_dir, output_format)
export_handler.ensure_output_dir()
output_files = []
if not layout.is_multi_page():
# 单张模式：每份合格证一个文件（每单台编号一个文件）
for i, (cert, serial) in enumerate(units):
if progress_callback:
progress_callback(i + 1, total, f"正在处理: {serial}")
file_path = self._generate_single_certificate(
cert,
export_handler,
coordinates,
paper_width_mm,
paper_height_mm,
template_type,
serial_number=serial,
data_overrides=data_overrides,
dual_code=dual_code,
)
if file_path:
output_files.append(file_path)
else:
# 多张拼接模式：每页排 rows×cols 份合格证（拼版到一张 A4）
total_pages = layout.get_total_pages(total)
processed = 0
for page_num, batch_start in enumerate(range(0, total, layout.per_page), 1):
batch = units[batch_start : batch_start + layout.per_page]
if progress_callback:
progress_callback(
processed + 1,
total,
f"正在生成第 {page_num} 页 ({len(batch)} 张)",
)
file_path = self._generate_multi_page(
batch,
page_num,
total_pages,
export_handler,
coordinates,
layout,
paper_width_mm,
paper_height_mm,
template_type,
data_overrides=data_overrides,
dual_code=dual_code,
)
if file_path:
output_files.append(file_path)
processed += len(batch)
return {
"success": len(output_files),
"failed": total - len(output_files),
"files": output_files,
"message": f"生成完成: {len(output_files)} 个文件",
}
def _load_coordinates(self, template_type: str) -> dict[str, Any]:
"""加载模板坐标（带缓存）。
Args:
template_type: 模板类型（全中文/中英文/俄英文）。
Returns:
dict[str, Any]: 包含 fields、background_path、template_key 的坐标字典。
"""
if template_type in self._coordinates_cache:
return self._coordinates_cache[template_type]
try:
from certflow.config.settings import cfg
from certflow.services.printer.template_field_formatter import resolve_template_key
templates = cfg("print_templates", {})
# 经统一解析器收口（取代本地 TEMPLATE_TYPE_MAP 双份硬编码，
# 见 BLUEPRINT §2.3 偏差 8）：入参已是合法 key 则直接用，
# 否则按语言族解析 enabled 变体 / 首个变体 / 兜底 full_chinese_3。
template_key = resolve_template_key(template_type)
tmpl = templates.get(template_key, {})
fields = tmpl.get("fields", {})
background_path = tmpl.get("background", "")
# 支持 print.yaml background.image_path 自定义底板覆盖（补丁37）
custom = cfg("background.image_path", "") or ""
if custom:
background_path = custom
# 转换相对路径为绝对路径
if background_path:
from certflow.utils.path_utils import get_project_root
background_path = str(get_project_root() / background_path)
if not os.path.exists(background_path):
logger.warning(
f"[Scan] 底板文件缺失: {background_path} (template={template_key})"
)
result = {
"fields": fields,
"background_path": background_path,
"template_key": template_key,
}
self._coordinates_cache[template_type] = result
logger.info(f"加载模板坐标: {template_key}, 字段数={len(fields)}")
return result
except Exception as e:
logger.warning(f"加载模板坐标失败: {e}")
return {"fields": {}, "background_path": ""}
def _generate_single_certificate(
self,
cert: Certificate,
export_handler: ScanExportHandler,
coordinates: dict[str, Any],
width_mm: float,
height_mm: float,
template_type: str = "全中文",
serial_number: str | None = None,
data_overrides: dict[int, dict] | None = None,
dual_code: bool = False,
) -> str | None:
"""生成单张合格证扫描件（一证一号）
Args:
cert: Certificate 对象
export_handler: 导出处理器
coordinates: 模板坐标
width_mm: 纸张宽度
height_mm: 纸张高度
serial_number: 单台产品编号（逐台展开后传入），缺省回退证书区间串
data_overrides: 合格证 ID → 覆盖字段字典（UI 编辑态，落库前预览/生成用）
Returns:
生成的文件路径，失败返回 None
"""
try:
# 创建画布
img = self.image_handler.create_canvas(
width_mm, height_mm, coordinates.get("background_path", "")
)
# 绘制字段（serial_number 覆盖为单台编号；data_overrides 覆盖字段值）
data = self._build_cert_data(
cert,
template_type,
serial_number=serial_number,
overrides=data_overrides.get(cert.id) if data_overrides else None,
dual_code=dual_code,
)
for field_key, pos in coordinates.get("fields", {}).items():
value = data.get(field_key, "")
if value:
self.image_handler.draw_field(
img,
value,
pos.get("x", 0),
pos.get("y", 0),
pos.get("font_size", 9),
width_mm=pos.get("width"),
height_mm=pos.get("height"),
align=pos.get("align", "left"),
)
# 导出文件（文件名用单台编号，便于 8 台生成 8 个独立文件）
file_path = export_handler.generate_filename_single(
serial_number or cert.certificate_no or "cert"
)
if export_handler.output_format == "PDF":
return self.image_handler.save_pdf(img, file_path)
return self.image_handler.save_jpg(img, file_path)
except Exception as e:
logger.error(f"生成单张扫描件失败 [{cert.certificate_no}]: {e}")
return None
def _generate_multi_page(
self,
units: list[tuple[Certificate, str]],
page_num: int,
total_pages: int,
export_handler: ScanExportHandler,
coordinates: dict[str, Any],
layout: ScanLayoutHandler,
width_mm: float,
height_mm: float,
template_type: str = "全中文",
data_overrides: dict[int, dict] | None = None,
dual_code: bool = False,
) -> str | None:
"""生成多张拼接扫描件（拼版到一张 A4，每格一证一号）
Args:
units: 本页的 (Certificate, 单台编号) 列表
page_num: 页码
total_pages: 总页数
export_handler: 导出处理器
coordinates: 模板坐标
layout: 布局处理器
width_mm: 单张宽度
height_mm: 单张高度
data_overrides: 合格证 ID → 覆盖字段字典（UI 编辑态）
Returns:
生成的文件路径，失败返回 None
"""
try:
# 计算画布尺寸：按本页实际占用的单元格裁剪（最后一页非满时去掉空白行/列，
# 避免生成整块 180×200 的空白合格证；满页与 calculate_canvas_size_mm 等价）。
used_rows, used_cols = layout.effective_grid(len(units))
canvas_width_mm = width_mm * used_cols
canvas_height_mm = height_mm * used_rows
canvas_width_px = self.image_handler.mm_to_px(canvas_width_mm)
canvas_height_px = self.image_handler.mm_to_px(canvas_height_mm)
# 创建画布
img = Image.new("RGB", (canvas_width_px, canvas_height_px), "white")
# 绘制每张合格证（每格用各自单台编号）
for idx, (cert, serial) in enumerate(units):
x_offset_mm, y_offset_mm, _, _ = layout.get_position(idx, width_mm, height_mm)
# 绘制背景
self._draw_background_at_position(
img,
coordinates.get("background_path", ""),
width_mm,
height_mm,
x_offset_mm,
y_offset_mm,
)
# 绘制字段（serial 覆盖为单台编号；data_overrides 覆盖字段值）
data = self._build_cert_data(
cert,
template_type,
serial_number=serial,
overrides=data_overrides.get(cert.id) if data_overrides else None,
dual_code=dual_code,
)
for field_key, pos in coordinates.get("fields", {}).items():
value = data.get(field_key, "")
if value:
self.image_handler.draw_field(
img,
value,
x_offset_mm + pos.get("x", 0),
y_offset_mm + pos.get("y", 0),
pos.get("font_size", 9),
width_mm=pos.get("width"),
height_mm=pos.get("height"),
align=pos.get("align", "left"),
)
# 导出文件（文件名带本页起止单台编号，便于识别拼版内容）
first_serial = units[0][1] if units else f"p{page_num}"
last_serial = units[-1][1] if units else f"p{page_num}"
serial_range = (
f"{first_serial}-{last_serial}" if first_serial != last_serial else first_serial
)
file_path = export_handler.generate_filename(serial_range, page_num, total_pages)
if export_handler.output_format == "PDF":
return self.image_handler.save_pdf(img, file_path)
return self.image_handler.save_jpg(img, file_path)
except Exception as e:
logger.error(f"生成多页扫描件失败 [第 {page_num} 页]: {e}")
return None
def _draw_background_at_position(
self,
img: Image.Image,
background_path: str,
width_mm: float,
height_mm: float,
x_offset_mm: float,
y_offset_mm: float,
) -> None:
"""在指定位置绘制背景图片"""
if not background_path or not os.path.exists(background_path):
return
try:
from PIL import Image
width_px = self.image_handler.mm_to_px(width_mm)
height_px = self.image_handler.mm_to_px(height_mm)
x_offset_px = self.image_handler.mm_to_px(x_offset_mm)
y_offset_px = self.image_handler.mm_to_px(y_offset_mm)
bg = self.image_handler.load_background_rgb(background_path)
bg = bg.resize((width_px, height_px), Image.Resampling.LANCZOS)
img.paste(bg, (x_offset_px, y_offset_px))
except Exception as e:
logger.warning(f"绘制背景图片失败: {e}")
@staticmethod
def _format_issue_ym(issue_date: str | None) -> str:
"""出厂日期格式化为 YYYY.MM（用户只需年月，BUG-006 J）。"""
if not issue_date:
return ""
s = str(issue_date).replace("/", "-")
parts = s.split("-")
if len(parts) >= 2:
return f"{parts[0]}.{parts[1]}"
return str(issue_date)
def _build_cert_data(
self,
cert: Certificate,
template_type: str = "全中文",
serial_number: str | None = None,
overrides: dict[str, Any] | None = None,
dual_code: bool = False,
) -> dict[str, Any]:
"""从 Certificate 构建打印数据（BUG-006 D/E/J/K 修复）。
Args:
cert: Certificate 对象
template_type: 模板类型（全中文/中英文/全英文/俄英文），
用于决定口径/压力的单位策略。
serial_number: 单台产品编号覆盖值（逐台展开后传入），缺省用区间串。
overrides: UI 编辑态覆盖字段（底层字段名 → 值），生成前合并，
打印预览与正式生成共用此路径以保证所见即所得。
dual_code: 是否渲染双编码（SN 短码 / KKS 长码）。默认关闭，
由配置 output.targets.scan.dual_code.enabled 驱动，
仅特殊用户场景启用。
Returns:
打印数据字典
"""
from certflow.services.printer.template_field_formatter import (
family_of,
get_field_formatter,
)
ov = overrides or {}
family = family_of(template_type)
formatter = get_field_formatter()
# 口径（公称通径，DN）独立于 pn 系列（pn_value/pn_display/pn_unit），其单位
# 前缀是 DN，绝非 MPa/mm 等压力/长度单位。旧逻辑误补 mm 后缀（且与预览二次
# format 产生 DN 前缀不一致），此处改为与打印侧同口径：直接按模板语言族用
# TemplateFieldFormatter 渲染——全中文底板预印单位→只打数值；中英文/全英文/
# 俄英文→补 DN 前缀。formatter.format 幂等，下游绘制/预览再次 format 也安全。
dn = ov.get("product_spec", cert.product_spec) or ""
dn = formatter.format("dn", dn, family)[0]
# 公称压力（pn 系列）按族格式化：中文仅数值，其余带单位（MPa/Lb）。
pn = formatter.format_pn(
ov.get("pn_value", cert.pn_value),
ov.get("pn_unit", cert.pn_unit),
family,
)
data: dict[str, Any] = {
"product_name": ov.get("product_name", cert.product_name) or "",
"product_model": ov.get("product_model", cert.product_model) or "",
"dn": dn,
"pn": pn,
"temperature": ov.get("working_temp", cert.working_temp) or "",
"medium": ov.get("working_medium", cert.working_medium) or "",
"check_standard": ov.get("test_standard", cert.test_standard) or "",
"inspector_id": ov.get("inspector_id", getattr(cert, "inspector_id", "")) or "",
# 出厂日期仅取年月 YYYY.MM（BUG-006 J）
"manufacture_date": self._format_issue_ym(ov.get("issue_date", cert.issue_date)),
# 产品编号优先用逐台单号（serial_number），缺省回退区间串（V...）；
# 不再误用 CERT- 批次号（BUG-006 K）。
"serial_number": serial_number
if serial_number
else (ov.get("product_code_range", cert.product_code_range) or ""),
}
# 双编码（SN 短码 / KKS 长码）：与打印侧 _derive_sn_and_kks 同口径。
# 电厂件真 KKS（不以 V 开头，落库于 SalePlan.kks_code）受保护不被 V 码覆盖。
if dual_code:
sn, kks = self._derive_dual_codes(cert, serial_number, ov)
data["sn_code"] = ov.get("sn_code", sn)
data["kks_code"] = ov.get("kks_code", kks)
return data
@staticmethod
def _derive_dual_codes(
cert: Certificate, serial_number: str | None, ov: dict[str, Any] | None
) -> tuple[str, str]:
"""由逐台单号派生 SN 短码 / KKS 长码（与打印侧同口径）。
- 默认模式: SN = 剥 4 位 YYMM 短码；KKS = 完整长 V 码。
- 编码模式 5/6: SN/KKS 按 sn_kks_position 互换。
- 电厂件真 KKS（SalePlan.kks_code 不以 V 开头）受保护，优先采用。
Args:
cert: Certificate 对象（用于取编码模式/关联 SalePlan 真 KKS）。
serial_number: 当前逐台产品编号（V 码），缺省回退区间串。
ov: UI 覆盖字典（可能含 sn_kks_position）。
Returns:
(sn, kks) 元组。
"""
from certflow.services.cert_numbering_policy import get_number_model
from certflow.services.certificate_print_service import CertificatePrintService
ov = ov or {}
unit = serial_number or (
ov.get("product_code_range", getattr(cert, "product_code_range", "")) or ""
)
number_model = 0
with contextlib.suppress(Exception): # 编码模式缺失时按默认模式处理
number_model = get_number_model(cert)
ar28 = ov.get("sn_kks_position", getattr(cert, "sn_kks_position", "") or "")
sn, kks = CertificatePrintService._derive_sn_and_kks(unit, number_model, ar28)
# 电厂件真 KKS 保护：SalePlan.kks_code 不以 V 开头时，覆盖推导出的 V 码长码
real_kks = ScanService._real_kks_for_cert(cert)
if real_kks:
kks = real_kks
return sn, kks
@staticmethod
def _real_kks_for_cert(cert: Certificate) -> str:
"""取证书关联 SalePlan 的真 KKS（电厂件，不以 V 开头）。无则返回空串。"""
sale_plan_id = getattr(cert, "sale_plan_id", None)
session = getattr(cert, "session", None) or ScanService._session_for(cert)
if not sale_plan_id or session is None:
return ""
try:
from certflow.models.sale_plan import SalePlan
plan = session.get(SalePlan, sale_plan_id)
kks = getattr(plan, "kks_code", "") or "" if plan else ""
return kks if kks and not kks.startswith("V") else ""
except Exception: # noqa: BLE001 - 查询失败时不阻断扫描件生成
return ""
@staticmethod
def _session_for(cert: Certificate) -> Any: # noqa: ANN401 - 仅内部兜底取 session
"""从 Certificate 实例尽力取出其所属 session（无则返回 None）。"""
try:
return cert._sa_instance_state.session if cert._sa_instance_state else None
except Exception: # noqa: BLE001
return None
# ------------------------------------------------------------------
# 逐台展开：批次级 Certificate → 单台编号列表
# ------------------------------------------------------------------
@staticmethod
def _format_seq(seq: int) -> str:
"""流水号补零（与打印视图一致：<1000 补 3 位，否则原值）。"""
return f"{seq:03d}" if seq < 1000 else str(seq)
@staticmethod
def _codes_from_structured(cert: Certificate) -> list[str]:
"""用产品编号拆填的结构化列展开（与打印视图一致，区间串不可用时兜底）。
数据模型里 ``product_code_prefix`` / ``product_code_ym`` /
``product_code_seq_start`` / ``product_code_seq_end`` / ``quantity`` 才是
逐台编号的权威来源；``product_code_range`` 只是派生显示串。当区间串为空或
不可解析时（如用户库仅填了结构化列），仍须据此展开，否则扫描件会退化为
整批 1 页（BUG：打印可翻页而扫描只有一页的根因）。
Returns:
list[str]: 逐台编码（如 ``['V2607018', ...]``）；列不全/无效返回 ``[]``。
"""
prefix = getattr(cert, "product_code_prefix", None)
ym = getattr(cert, "product_code_ym", None)
try:
seq_start = int(getattr(cert, "product_code_seq_start", None) or 0)
seq_end = int(getattr(cert, "product_code_seq_end", None) or 0)
except (TypeError, ValueError):
return []
if not (prefix and ym) or seq_start <= 0:
return []
if seq_end < seq_start:
# 仅末流水缺失时，用 quantity 推算结束流水（与打印视图一致）
qty = int(getattr(cert, "quantity", None) or 1)
seq_end = seq_start + max(0, qty - 1)
if seq_end < seq_start:
return []
return [
f"{prefix}{ym}{ScanService._format_seq(seq)}" for seq in range(seq_start, seq_end + 1)
]
def _individual_codes(
self, cert: Certificate, override: dict[str, Any] | None = None
) -> list[str]:
"""从 Certificate 解析单台产品编号列表（展开为逐台编码，而非区间串）。
展开策略（依次尝试，命中即止）：
1. 已拆填的逐台字段 ``cert.code_list``（兼容预留）；
2. 解析 ``product_code_range`` 区间串（``parse_product_code_meta``，
形如 ``V2607050Y---057Y``，可保留后缀）；
3. 结构化列兜底（``product_code_prefix/ym/seq_start/seq_end``，
与打印视图同口径，区间串为空/不可解析时仍能展开）；
4. 纯数字区间串兜底（如 ``001-012``）→ 按结束位宽补零展开为 001..012。
覆写（override）优先，便于编辑态即时预览/生成。
"""
ov = override or {}
# 覆写里的区间串优先
range_str = ov.get("product_code_range") or getattr(cert, "product_code_range", "") or ""
try:
codes = getattr(cert, "code_list", None) or []
except Exception: # noqa: BLE001
codes = []
if codes:
return codes
meta = parse_product_code_meta(range_str)
if meta:
prefix = meta["prefix"]
ym = meta["ym"]
suffix = meta["suffix"] or ""
seq_start = meta["seq_start"]
seq_end = meta["seq_end"]
return [
f"{prefix}{ym}{self._format_seq(seq)}{suffix}"
for seq in range(seq_start, seq_end + 1)
]
# 兜底②：结构化列（与打印视图对齐；区间串缺失/不可解析时仍能正确展开）
structured = self._codes_from_structured(cert)
if structured:
return structured
# 兜底③：纯数字区间（如 "001-012"）
plain = re.match(r"^\s*(\d+)\s*-\s*(\d+)\s*$", range_str.strip())
if plain:
start = int(plain.group(1))
end = int(plain.group(2))
width = len(plain.group(2))
return [f"{i:0{width}d}" for i in range(start, end + 1)]
return []
def _expand_units(
self,
certificates: list[Certificate],
data_overrides: dict[int, dict] | None = None,
) -> list[tuple[Certificate, str]]:
"""将批次级 Certificate 展开为单台扫描单元。
每个 Certificate 的 ``product_code_range`` 是区间串（如 ``V2607050-057``），
需拆分为单台编号 ``V2607050``..``V2607057``；逐台生成「一证一号」扫描件。
解析失败或无区间时回退为单台（用区间串/证书号本身）。
Args:
certificates: 批次级 Certificate 列表。
data_overrides: 合格证 ID → 覆盖字段字典（编辑态，展开时用其区间串）。
Returns:
list[tuple[Certificate, str]]: (证书, 单台编号) 列表。
"""
units: list[tuple[Certificate, str]] = []
for cert in certificates:
ov = (data_overrides or {}).get(cert.id)
codes = self._individual_codes(cert, ov)
if not codes:
fallback = (
(ov or {}).get("product_code_range")
or cert.product_code_range
or cert.certificate_no
or ""
)
units.append((cert, fallback))
else:
for code in codes:
units.append((cert, code))
return units