"""质保书系列报告的 xlsx 填充引擎（openpyxl，保格式 + 分页 + 以下空白）。
设计依据（提取自 VBA 模块 Mod_制作质保书在用.txt + 真实模板几何）：
- 单工作表内按「页」堆叠：每页 = 标题/表头(静态) + K 个产品行 + 材质段(每产品 p 行) + 落款(仅末页)。
- 行高/列宽/边框来自模板母行，原样保留 -> 打印按行高精准分页不偏。
- 末页数据不足 K 条时：首个空产品槽合并 A:C 写「以下空白」，其余空行纯留空(仅边框)。
材质段同理：首个空零件组写「以下空白」，后续空行清内容、复制格式。
- 分页：每页末行加 row_break；表头行(1:4)设 print_title_rows 重复。
模块泛型化：模板差异由 ZhiBaoShuSpec 描述，引擎本身不硬编码列号。
（独立于 certflow PySide GUI 包，便于无头环境测试与迁移工具复用。）
"""
from __future__ import annotations
from copy import copy
from dataclasses import dataclass, field
from datetime import date, datetime
import openpyxl
from certflow.services.report_migration.material_provider import MaterialPart, MaterialProvider
[文档]
@dataclass
class ProductRecord:
"""一条产品记录（写入质保书产品行 + 驱动材质解析）。
字段与 VBA 产品行列映射一一对应；parts 为材质块主要零件列表。
"""
序号: int = 0
产品名称: str = ""
产品型号: str = ""
规格: str = ""
数量: object = ""
产品编号: str = ""
试验标准: str = ""
壳体压力: str = ""
壳体时间: str = ""
上密封压力: str = ""
上密封时间: str = ""
密封压力: str = ""
密封时间: str = ""
结果: str = "合格"
适用温度: str = "" # 9行「及温度」变体：适用温度（≤℃）
生产令号: str = ""
parts: list[MaterialPart] | None = None
# 阶段 C：BOM 待核对明细（未命中/多对一/缺数据零件），随记录传递供报告页展示，不落模板
待核对明细: list[dict[str, str]] | None = None
# ----------------------------- 样式/合并复制工具 -----------------------------
[文档]
def copy_cell_style(src, dst) -> None:
dst.font = copy(src.font)
dst.border = copy(src.border)
dst.fill = copy(src.fill)
dst.alignment = copy(src.alignment)
dst.number_format = src.number_format
dst.protection = copy(src.protection)
def _unmerge_intersecting(ws, r1: int, c1: int, r2: int, c2: int) -> None:
"""移除与给定矩形相交的所有合并区域，避免重新合并时报错。"""
for mr in list(ws.merged_cells.ranges):
if not (mr.max_row < r1 or mr.min_row > r2 or mr.max_col < c1 or mr.min_col > c2):
ws.unmerge_cells(
start_row=mr.min_row,
start_column=mr.min_col,
end_row=mr.max_row,
end_column=mr.max_col,
)
[文档]
def merge_cells_safe(ws, r1: int, c1: int, r2: int, c2: int) -> None:
_unmerge_intersecting(ws, r1, c1, r2, c2)
ws.merge_cells(start_row=r1, start_column=c1, end_row=r2, end_column=c2)
[文档]
def copy_block_styles(
src_ws, dst_ws, src_top: int, src_bottom: int, dst_top: int, ncols: int, src_left: int = 1
) -> None:
"""把 [src_top..src_bottom] 行的样式(单元格样式+行高+合并) 整体偏移复制到 dst_top。
src_ws 为模板源（始终读模板母行），dst_ws 为输出表。
"""
offset = dst_top - src_top
for r in range(src_top, src_bottom + 1):
tr = r + offset
dst_ws.row_dimensions[tr].height = src_ws.row_dimensions[r].height
for c in range(src_left, src_left + ncols):
src_cell = src_ws.cell(row=r, column=c)
dst_cell = dst_ws.cell(row=tr, column=c)
copy_cell_style(src_cell, dst_cell)
if src_cell.value is not None:
dst_cell.value = src_cell.value
# 合并区域整体偏移（基于模板源）
for mr in list(src_ws.merged_cells.ranges):
if src_top <= mr.min_row <= src_bottom:
_unmerge_intersecting(
dst_ws, mr.min_row + offset, mr.min_col, mr.max_row + offset, mr.max_col
)
dst_ws.merge_cells(
start_row=mr.min_row + offset,
start_column=mr.min_col,
end_row=mr.max_row + offset,
end_column=mr.max_col,
)
# ------------------------------- 模板规格 -------------------------------
[文档]
@dataclass
class ZhiBaoShuSpec:
name: str # 模板 sheet 名
k: int # 每页产品数
单条行数: int # 每产品零件行数
page_row_count: int # 每页总行数 (PageRowCount)
page_col_count: int # 每页总列数 (PageColCount)
# 区块相对行号（相对每页顶部）
title_rows: tuple[int, int] = (1, 4)
product_start: int = 5
material_header_rows: tuple[int, int] = (11, 13)
material_start: int = 14
footer_rows: tuple[int, int] = (27, 29)
has_material: bool = True # 试压报告等无材质块模板置 False
# 每页报告头（R2 相对页顶）的日期 / 报告单号 单元格（对标 VBA Mod_制作质保书在用）
date_row: int = 2
date_col: int = 8 # H 列: "日期：yyyy.mm.dd"
number_row: int = 2
number_col: int = 15 # O 列: "C/1 编号：9.1-13-yymmdd{serial}"
number_prefix: str = "C/1 编号：" # 报告单号前缀；只有标准号=C/0，试压=C/1 N0：
report_code: str = "9.1-13" # 报告单号中的标准代号；试压(中英)=8.2-20
# 日期单元格写入格式；试压(中英) 模板日期格只存日期本身（无"日期："前缀），故置空
date_label: str = "日期："
date_fmt: str = "%Y.%m.%d"
# 产品块：每条产品占的行数（质保书/多数试压=1；试压(中英) 双语每产品占 2 行合并）
product_row_span: int = 1
# 是否有独立产品信息表（质保书/试压=True；材质证明书为 False，产品身份并入材质组）
product_block: bool = True
# 材质证明书模式：产品身份(序号/产品名称/型号规格)写进材质组首行并向下合并
caizhi: bool = False
# 产品行写格列映射：字段名 -> 列号(1-based)
product_cols: dict[str, int] = field(default_factory=dict)
# 产品行合并跨列：字段名 -> (首列, 末列)
product_merges: dict[str, tuple[int, int]] = field(default_factory=dict)
# 材质行写格列映射
material_cols: dict[str, int] = field(default_factory=dict)
# ------------------------------- 填充引擎 -------------------------------
[文档]
class ReportFillEngine:
def __init__(self, provider: MaterialProvider, material_annotation: bool = False):
self.provider = provider
# 材质单元格溯源批注开关（REPORT_MATERIAL_BLUEPRINT §6，默认关以保向后兼容）
self.material_annotation = material_annotation
# ---- 批注 / 溢出列（§3/§4）----
def _fill_overflow_chem(self, ws, row: int, part: MaterialPart, spec: ZhiBaoShuSpec) -> None:
"""B 类：不在表头映射集的元素，用换行符 join 填入第 11 列溢出格。
仅当 spec 声明了溢出列（如质保书 col 14）且该格当前为空才写；
若该列已被表头元素占用（如材质证明书 Ti 占 col 14）则跳过（改由批注承载）。
"""
overflow_col = spec.material_cols.get("化学成分溢出")
if not overflow_col or not part.表头外元素:
return
if ws.cell(row=row, column=overflow_col).value not in (None, ""):
return # 溢出列已被占用，不覆盖，交批注
text = "\n".join(f"{k}={v}" for k, v in part.表头外元素.items())
ws.cell(row=row, column=overflow_col).value = text
def _attach_material_comment(
self, ws, row: int, col: int, part: MaterialPart, spec: ZhiBaoShuSpec
) -> None:
"""按 §3 规范给 材料 单元格挂溯源批注；无溯源信息则不挂。"""
if not (part.来源 != "cert" or part.is_virtual or part.表头外元素):
return # 纯真实数据不挂批注
lines: list[str] = []
if part.来源 == "bom":
lines.append(f"[溯源] 来源：BOM派生(生产令号 {part.来源令号})")
elif part.来源 == "unknown":
lines.append("[溯源] 来源：未知（无牌号）")
if part.is_virtual:
lines.append("[成分] 为按标准区间虚拟示意值，非实测；区间取自 material_grades")
if part.表头外元素:
extra = "\n ".join(f"{k}={v}" for k, v in part.表头外元素.items())
lines.append(f"[缺列] 以下元素不在表头、未写入单元格（模拟值）：\n {extra}")
from openpyxl.comments import Comment
ws.cell(row=row, column=col).comment = Comment("\n".join(lines), "CertFlow")
# ---- 产品行 / 材质行 单条写入 ----
def _write_product_row(self, ws, abs_row: int, rec, spec: ZhiBaoShuSpec) -> None:
# 泛型按 product_cols 写入：字段名与 ProductRecord 属性名一致，
# 缺字段（如「只有标准号」无 产品编号、「及温度」才有 适用温度）自动跳过。
for field_name, col in spec.product_cols.items():
if not col:
continue
val = getattr(rec, field_name, None)
# 型号规格：试压(中英) 单列合并 产品型号+规格
if val is None and field_name == "型号规格":
val = " ".join(str(x) for x in [rec.产品型号, rec.规格] if x)
cell = ws.cell(row=abs_row, column=col)
cell.value = val
# 序号列：个别 9 行/试压(中英)/材质 模板把 A 列设成日期格式，写入整数会显示为日期，
# 强制 General 保证序号正确显示（对标 VBA 实际输出为 1/2/3...）。
if field_name == "序号":
cell.number_format = "General"
def _write_material_group(
self, ws, abs_first_row: int, parts: list[MaterialPart], spec: ZhiBaoShuSpec, seq: int
) -> bool:
"""写入一个产品的质保书材质组（C4：阀体/阀杆/启闭件并行展开）。
逐零件写入连续行（封顶 ``spec.单条行数``），使质保书材质块真正展开多零件，
与材质证明书 ``_write_caizhi_group`` 行为对齐；序号仅首行。
无材质时整组留空（不写序号），交由调用方在该页首个空组处写「以下空白」。
数据来源：``records[idx].parts``（由 ``_resolve_materials`` 按合格证三字段
fati/fagan/qibijian 解析，C1 已产出多零件）或 provider 回退；本方法只负责把
已解析的多零件逐行落盘。
"""
mc = spec.material_cols
if not parts:
return False
parts = parts[: spec.单条行数]
for j, part in enumerate(parts):
r = abs_first_row + j
if j == 0:
ws.cell(row=r, column=mc["序号"]).value = seq
ws.cell(row=r, column=mc["零件名称"]).value = part.零件名称
ws.cell(row=r, column=mc["材料"]).value = part.材料
chem = mc.get("化学成分") or {}
for key, col in chem.items():
if key in part.化学成分:
ws.cell(row=r, column=col).value = part.化学成分[key]
mech = mc.get("力学性能") or {}
for key, col in mech.items():
if key in part.力学性能:
ws.cell(row=r, column=col).value = part.力学性能[key]
ws.cell(row=r, column=mc["结果"]).value = part.结果
if self.material_annotation:
self._fill_overflow_chem(ws, r, part, spec)
self._attach_material_comment(ws, r, mc["材料"], part, spec)
return True
def _write_caizhi_group( # noqa: C901
self, ws, abs_first_row: int, rec, spec: ZhiBaoShuSpec, seq: int
) -> bool:
"""材质证明书模式：产品身份写进材质组首行并向下合并，逐行写零件材质。
与质保书材质块的区别：这里没有独立产品信息表，序号/产品名称/型号规格
直接落在材质组首行，并按 单条行数 向下合并（对标模板 A7:A9 等合并）。
"""
mc = spec.material_cols
ws.cell(row=abs_first_row, column=mc["序号"]).value = seq
ws.cell(row=abs_first_row, column=mc["序号"]).number_format = "General"
ws.cell(row=abs_first_row, column=mc["产品名称"]).value = rec.产品名称
mc_val = " ".join(str(x) for x in [rec.产品型号, rec.规格] if x)
ws.cell(row=abs_first_row, column=mc["型号规格"]).value = mc_val
if spec.单条行数 > 1:
for f in ("序号", "产品名称", "型号规格"):
c = mc[f]
merge_cells_safe(ws, abs_first_row, c, abs_first_row + spec.单条行数 - 1, c)
# 零件材质：优先用记录自带 parts，否则经 provider 解析（绝不伪造）
parts = rec.parts or []
if not parts and self.provider is not None:
parts = (
self.provider.resolve(
getattr(rec, "生产令号", ""),
型号=rec.产品型号,
规格=rec.规格,
零件数=spec.单条行数,
)
or []
)
# 来自外部数据源（非合格证手工录入）→ 标记 BOM 派生，供批注溯源
for _pp in parts:
if _pp.来源 == "cert":
_pp.来源 = "bom"
parts = parts[: spec.单条行数]
for j, part in enumerate(parts):
r = abs_first_row + j
ws.cell(row=r, column=mc["零件名称"]).value = part.零件名称
ws.cell(row=r, column=mc["材料"]).value = part.材料
chem = mc.get("化学成分") or {}
for key, col in chem.items():
if key in part.化学成分:
ws.cell(row=r, column=col).value = part.化学成分[key]
mech = mc.get("力学性能") or {}
for key, col in mech.items():
if key in part.力学性能:
ws.cell(row=r, column=col).value = part.力学性能[key]
ws.cell(row=r, column=mc["结果"]).value = part.结果
if self.material_annotation:
self._fill_overflow_chem(ws, r, part, spec)
self._attach_material_comment(ws, r, mc["材料"], part, spec)
return True
# ---- 区块清空（保留样式，清值）----
def _clear_block(self, ws, r1: int, r2: int, c1: int, c2: int) -> None:
from openpyxl.cell.cell import MergedCell
for r in range(r1, r2 + 1):
for c in range(c1, c2 + 1):
cell = ws.cell(row=r, column=c)
if isinstance(cell, MergedCell):
continue # 合并区非左上角只读，左上角已清即可
cell.value = None
# ---- 主流程 ----
[文档]
def fill( # noqa: C901
self,
wb_out: openpyxl.Workbook,
template_ws,
spec: ZhiBaoShuSpec,
records: list,
sheet_name: str = "报告",
report_date: date | None = None,
page_numbers: list[str] | None = None,
) -> openpyxl.worksheet.worksheet.Worksheet:
ws = wb_out.create_sheet(sheet_name)
# 列宽：只用模板中属于本页（<=page_col_count）的列，跳过钉钉/复制遗留的幽灵列宽
from openpyxl.utils import column_index_from_string
for col, dim in template_ws.column_dimensions.items():
ci = column_index_from_string(col)
if 1 <= ci <= spec.page_col_count and dim.width:
ws.column_dimensions[col].width = dim.width
copy_block_styles(
template_ws, ws, 1, spec.page_row_count, 1, spec.page_col_count, src_left=1
)
N = len(records)
k = spec.k
prs = spec.product_row_span
pages = (N + k - 1) // k
i_end = N % k
if i_end == 0:
i_end = k
date_str = None
if report_date is not None:
date_str = (
(spec.date_label + report_date.strftime(spec.date_fmt))
if spec.date_label
else report_date.strftime(spec.date_fmt)
)
for p in range(1, pages + 1):
base = (p - 1) * spec.page_row_count
if p > 1:
copy_block_styles(
template_ws,
ws,
1,
spec.page_row_count,
base + 1,
spec.page_col_count,
src_left=1,
)
# 清空产品块 / 材质块（按各自占用行数）
if spec.product_block:
self._clear_block(
ws,
base + spec.product_start,
base + spec.product_start + k * prs - 1,
1,
spec.page_col_count,
)
if spec.has_material:
self._clear_block(
ws,
base + spec.material_start,
base + spec.material_start + k * spec.单条行数 - 1,
1,
spec.page_col_count,
)
start = (p - 1) * k
# 产品行（每产品占 prs 行；质保书 prs=1，试压(中英) prs=2）
if spec.product_block:
for i in range(k):
idx = start + i
abs_row = base + spec.product_start + i * prs
if idx < N:
self._write_product_row(ws, abs_row, records[idx], spec)
# 材质组（§12 空洞修复）：区分中间洞与尾部空槽
if spec.has_material:
for i in range(k):
idx = start + i
abs_first = base + spec.material_start + i * spec.单条行数
wrote = False
if idx < N:
if spec.caizhi:
self._write_caizhi_group(ws, abs_first, records[idx], spec, i + 1)
wrote = True # 材质证明书每个产品都写一组
else:
parts = records[idx].parts or (
self.provider.resolve(
getattr(records[idx], "生产令号", ""),
型号=records[idx].产品型号,
规格=records[idx].规格,
零件数=spec.单条行数,
)
if self.provider
else []
)
wrote = self._write_material_group(ws, abs_first, parts, spec, i + 1)
if not wrote:
# 中间洞（idx<N 且查无材质）→「数据缺失」；尾部（idx>=N 订单不足）→「以下空白」
text = "数据缺失" if idx < N else "以下空白"
ws.cell(row=abs_first, column=1).value = text
merge_cells_safe(ws, abs_first, 1, abs_first + spec.单条行数 - 1, 3)
# 每页报告头：日期 + 报告单号（R2 相对页顶，对标 VBA）
if date_str:
ws.cell(row=base + spec.date_row, column=spec.date_col).value = date_str
# 一页一号：每页写各自编号；无逐页编号（兼容旧调用）则留空
if page_numbers:
page_no = page_numbers[p - 1] if p - 1 < len(page_numbers) else None
if page_no:
ws.cell(row=base + spec.number_row, column=spec.number_col).value = page_no
if p == pages and spec.product_block and i_end < k:
# 产品块：末页首空槽写「以下空白」（跨 prs 行合并 A:C）
prod_blank = base + spec.product_start + i_end * prs
ws.cell(row=prod_blank, column=1).value = "以下空白"
merge_cells_safe(ws, prod_blank, 1, prod_blank + prs - 1, 3)
from openpyxl.worksheet.pagebreak import Break, RowBreak
rb = RowBreak()
for p in range(1, pages):
rb.append(Break(id=p * spec.page_row_count))
ws.row_breaks = rb
ws.print_title_rows = f"{spec.title_rows[0]}:{spec.title_rows[1]}"
ws.page_setup.fitToWidth = 1
ws.page_setup.fitToHeight = 0
# 不打印批注（§7）：显式兜底，确保批注仅悬停可见、不进打印/导出 PDF
ws.print_options.comments = None
# 格式归一：凡「日期格式单元格 + 写入值非日期」一律改 General，
# 避免数量/力学性能/材质块序号等数值被日期格式渲染成 1900 序列日期
# （补充引擎此前仅对主序号列做格式纠正的遗漏；模板母行误设日期格式所致）。
from openpyxl.cell.cell import MergedCell
for row in ws.iter_rows():
for cell in row:
if isinstance(cell, MergedCell):
continue
if cell.value is None:
continue
if isinstance(cell.value, (datetime, date)):
continue
if "y" in cell.number_format.lower():
cell.number_format = "General"
return ws