"""模板配置管理器
管理合格证打印模板的配置，包括内置模板（全中文/中英文/全英文/俄英文）
和用户自定义模板。提供 templates.xlsx 历史模板数据的读写功能。
从 certificate_printer v2.0 移植，整合到 CertFlow 架构中。
分层说明:
本模块为纯数据处理层（handlers），不依赖 PySide6，可在无 Qt
（headless / 无 X11）环境导入。模板切换通知采用普通回调机制
（``on_template_changed``），由上层 View 自行桥接到 Qt 信号。
"""
from __future__ import annotations
import json
import os
from collections.abc import Callable
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any
import pandas as pd
from certflow.handlers.excel_handler import ExcelHandler
from certflow.utils.logger import logger
[文档]
@dataclass
class FieldMapping:
"""字段映射 — 数据源字段到打印目标位置的映射"""
source_field: str # 数据源字段名
target_cell: str = "" # 目标单元格位置 (如 "B3", "C5") — Excel 回填用
display_name: str = "" # 显示名称
show_label: bool = False # 是否在预览中显示标签前缀
x: float = 0 # X 坐标 (mm)，绝对定位用
y: float = 0 # Y 坐标 (mm)，绝对定位用
align: str = "left" # 对齐方式: left / center / right
font_size: int = 8 # 字体大小 (pt)
is_required: bool = False
default_value: str = ""
format_string: str = ""
[文档]
@dataclass
class TemplateConfig:
"""合格证打印模板配置"""
name: str # 内部标识名
display_name: str # 用户可见名称
template_type: str = "全中文" # 全中文 / 中英文 / 全英文 / 俄英文
page_width: float = 60 # mm
page_height: float = 100 # mm
background_image: str = ""
fields: list[FieldMapping] = field(default_factory=list)
[文档]
def to_dict(self) -> dict[str, Any]:
"""将模板配置序列化为字典
Args:
无
Returns:
dict[str, Any]: 模板配置字典，包含 name/display_name/fields 等键
"""
return {
"name": self.name,
"display_name": self.display_name,
"template_type": self.template_type,
"page_width": self.page_width,
"background_image": self.background_image,
"page_height": self.page_height,
"fields": [
{
"source_field": f.source_field,
"target_cell": f.target_cell,
"display_name": f.display_name,
"show_label": f.show_label,
"x": f.x,
"y": f.y,
"align": f.align,
"font_size": f.font_size,
"is_required": f.is_required,
"default_value": f.default_value,
"format_string": f.format_string,
}
for f in self.fields
],
}
[文档]
@classmethod
def from_dict(cls, data: dict[str, Any]) -> TemplateConfig:
"""从字典反序列化为 TemplateConfig 实例
Args:
data: 模板配置字典（含 name/display_name/fields 等键）
Returns:
TemplateConfig: 重建的模板配置对象
"""
fields = [
FieldMapping(
source_field=f["source_field"],
target_cell=f.get("target_cell", ""),
display_name=f.get("display_name", ""),
show_label=f.get("show_label", True),
x=f.get("x", 0),
y=f.get("y", 0),
align=f.get("align", "left"),
font_size=f.get("font_size", 8),
is_required=f.get("is_required", False),
default_value=f.get("default_value", ""),
format_string=f.get("format_string", ""),
)
for f in data.get("fields", [])
]
return cls(
name=data["name"],
display_name=data["display_name"],
template_type=data.get("template_type", "全中文"),
page_width=data.get("page_width", 60),
page_height=data.get("page_height", 100),
background_image=data.get("background_image", ""),
fields=fields,
)
# ============================================================
# 最小回退模板（仅 config.yaml 加载失败时使用）
# ============================================================
_FALLBACK_FULL_CHINESE_FIELDS: list[FieldMapping] = [
FieldMapping(
"product_name", "B3", "产品名称", x=30, y=20, align="center", font_size=10, is_required=True
),
FieldMapping(
"product_model",
"B4",
"产品型号",
x=30,
y=28,
align="center",
font_size=10,
is_required=True,
),
FieldMapping("dn", "B5", "公称通径", x=12, y=38, align="left", font_size=8, is_required=True),
FieldMapping("pn", "B6", "公称压力", x=12, y=45, align="left", font_size=8, is_required=True),
FieldMapping(
"temperature", "B7", "工作温度", x=12, y=52, align="left", font_size=8, is_required=True
),
FieldMapping(
"medium", "B8", "适用介质", x=12, y=59, align="left", font_size=8, is_required=True
),
FieldMapping(
"check_standard", "B9", "检验标准", x=12, y=66, align="left", font_size=7, is_required=False
),
FieldMapping(
"inspector_id",
"B10",
"检验工号",
x=30,
y=80,
align="center",
font_size=8,
is_required=False,
),
FieldMapping(
"manufacture_date",
"B11",
"出厂日期",
x=30,
y=88,
align="center",
font_size=8,
is_required=True,
),
FieldMapping(
"serial_number", "B12", "产品编号", x=50, y=88, align="right", font_size=8, is_required=True
),
]
[文档]
class TemplateManager:
"""模板管理器
管理合格证打印模板的配置，支持内置模板和用户自定义模板。
提供 templates.xlsx 历史模板数据的查询和回填功能。
通知机制:
模板切换时回调 ``on_template_changed`` 注册的监听器（携带模板名称）。
本类不依赖 Qt；如需 Qt 信号，由上层 View 注册回调后自行转发。
"""
def __init__(self, templates_xlsx_path: str = "") -> None:
"""初始化模板管理器
Args:
templates_xlsx_path: templates.xlsx 历史模板文件路径，默认为空
"""
self.templates: dict[str, TemplateConfig] = {}
self.current_template: TemplateConfig | None = None
self.templates_xlsx_path = templates_xlsx_path
self._template_changed_callbacks: list[Callable[[str], None]] = []
self._load_default_templates()
# ================================================================
# 变更通知（替代原 Qt Signal，保持 handlers 层无 Qt）
# ================================================================
[文档]
def on_template_changed(self, callback: Callable[[str], None]) -> None:
"""注册模板切换监听回调
Args:
callback: 模板切换时被调用，入参为模板内部名称
Returns:
None
"""
self._template_changed_callbacks.append(callback)
def _emit_template_changed(self, name: str) -> None:
"""通知所有已注册的模板切换监听器
单个回调异常不影响其余回调执行。
Args:
name: 模板内部名称
Returns:
None
"""
for callback in self._template_changed_callbacks:
try:
callback(name)
except Exception as e:
logger.warning(f"模板切换回调执行失败: {e}")
# ================================================================
# 模板 CRUD
# ================================================================
def _load_default_templates(self) -> None:
"""加载内置模板：优先从 config.yaml，回退到最小硬编码"""
yaml_templates = self._load_templates_from_config()
if yaml_templates:
self.templates.update(yaml_templates)
logger.info(
f"从 config.yaml 加载 {len(yaml_templates)} 个模板: {list(yaml_templates.keys())}"
)
else:
# 最小回退（只保留一个全中文模板，避免两份数据不同步）
logger.warning("config.yaml 模板加载失败，使用最小回退模板")
self.templates["full_chinese"] = TemplateConfig(
name="full_chinese",
display_name="全中文合格证模板",
template_type="全中文",
page_width=60,
page_height=100,
fields=_FALLBACK_FULL_CHINESE_FIELDS,
)
self._load_user_templates()
logger.debug(f"模板管理器初始化: {len(self.templates)} 个模板")
@staticmethod
def _load_templates_from_config() -> dict[str, TemplateConfig]:
"""从 ui.yaml 加载模板配置。
Returns:
模板字典，如果配置不可用则返回空字典。
"""
try:
from certflow.config.settings import cfg
except ImportError:
return {}
templates_cfg = cfg("field_templates", {})
if not templates_cfg:
return {}
result: dict[str, TemplateConfig] = {}
for template_name, cfg in templates_cfg.items():
try:
fields = []
for f in cfg.get("fields", []):
fields.append(
FieldMapping(
source_field=f["source_field"],
target_cell=f.get("target_cell", ""),
display_name=f.get("display_name", ""),
x=f.get("x", 0),
y=f.get("y", 0),
align=f.get("align", "left"),
font_size=f.get("font_size", 8),
is_required=f.get("is_required", False),
)
)
template = TemplateConfig(
name=cfg.get("name", template_name),
display_name=cfg.get("display_name", template_name),
template_type=cfg.get("template_type", "全中文"),
page_width=cfg.get("page_width", 60),
page_height=cfg.get("page_height", 100),
background_image=cfg.get("background_image", ""),
fields=fields,
)
result[template.name] = template
except Exception as e:
logger.warning(f"加载模板配置失败 {template_name}: {e}")
return result
def _load_user_templates(self) -> None:
"""加载用户自定义模板"""
template_dir = "templates"
if os.path.isdir(template_dir):
for file_name in os.listdir(template_dir):
if file_name.endswith(".json"):
try:
file_path = os.path.join(template_dir, file_name)
with open(file_path, encoding="utf-8") as f:
data = json.load(f)
template = TemplateConfig.from_dict(data)
self.templates[template.name] = template
except Exception as e:
logger.warning(f"加载模板失败 {file_name}: {e}")
[文档]
def add_template(self, template: TemplateConfig) -> None:
"""添加模板
Args:
template: 要添加的模板配置对象
Returns:
None
"""
self.templates[template.name] = template
logger.debug(f"添加模板: {template.display_name}")
[文档]
def remove_template(self, template_name: str) -> None:
"""删除模板
Args:
template_name: 要删除的模板内部名称
Returns:
None
"""
if template_name in self.templates:
del self.templates[template_name]
[文档]
def get_template(self, name: str) -> TemplateConfig | None:
"""获取模板
Args:
name: 模板内部名称
Returns:
TemplateConfig | None: 命中的模板配置，未找到时返回 None
"""
return self.templates.get(name)
[文档]
def set_current_template(self, name: str) -> None:
"""设置当前模板
Args:
name: 模板内部名称
Returns:
None: 设置成功后回调已注册的模板切换监听器
"""
if name in self.templates:
self.current_template = self.templates[name]
self._emit_template_changed(name)
[文档]
def get_template_list(self) -> list[str]:
"""获取模板名称列表（用户可见）
Args:
无
Returns:
list[str]: 各模板的 display_name 列表
"""
return [t.display_name for t in self.templates.values()]
[文档]
def get_template_names(self) -> list[str]:
"""获取模板内部名称列表
Args:
无
Returns:
list[str]: 各模板的内部 name 列表
"""
return list(self.templates.keys())
[文档]
def get_template_by_display_name(self, display_name: str) -> TemplateConfig | None:
"""根据显示名称获取模板
Args:
display_name: 模板用户可见名称
Returns:
TemplateConfig | None: 命中的模板配置，未找到时返回 None
"""
for t in self.templates.values():
if t.display_name == display_name:
return t
return None
[文档]
def export_template(self, template_name: str, file_path: str) -> bool:
"""导出模板配置为 JSON
Args:
template_name: 要导出的模板内部名称
file_path: 目标 JSON 文件路径
Returns:
bool: 导出成功返回 True，模板不存在时返回 False
"""
template = self.templates.get(template_name)
if template:
with open(file_path, "w", encoding="utf-8") as f:
json.dump(template.to_dict(), f, ensure_ascii=False, indent=2)
return True
return False
[文档]
def import_template(self, file_path: str) -> TemplateConfig | None:
"""从 JSON 导入模板配置
Args:
file_path: 源 JSON 文件路径
Returns:
TemplateConfig | None: 导入成功的模板配置，异常时为 None
"""
with open(file_path, encoding="utf-8") as f:
data = json.load(f)
template = TemplateConfig.from_dict(data)
self.add_template(template)
return template
# ================================================================
# templates.xlsx 历史模板读写
# ================================================================
[文档]
def load_from_xlsx(self, xlsx_path: str | None = None) -> pd.DataFrame | None:
"""从 templates.xlsx 的「数据库」工作表加载历史模板数据
Args:
xlsx_path: xlsx 文件路径，默认使用初始化时指定的路径
Returns:
DataFrame 或 None
"""
path = xlsx_path or self.templates_xlsx_path
if not path or not os.path.exists(path):
logger.debug(f"templates.xlsx 不存在: {path}")
return None
try:
wb = ExcelHandler.load_workbook(Path(path), data_only=True)
if "数据库" not in wb.sheetnames:
logger.warning("templates.xlsx 中没有「数据库」工作表")
wb.close()
return None
ws = wb["数据库"]
total_rows = ws.max_row - 1
logger.debug(f"templates.xlsx「数据库」: {total_rows} 行")
# 列映射: (列索引1-based, 字段名)
column_mapping = [
(1, "row_index"),
(2, "template_name"),
(3, "product_name"),
(4, "product_model"),
(5, "dn"),
(6, "pn"),
(7, "temperature"),
(8, "medium"),
(9, "check_standard"),
(10, "inspector_id"),
(11, "manufacture_date"),
(12, "serial_number"),
(13, "dn_inch"),
(14, "pn_lb"),
(16, "print_time"),
(17, "workbook_name"),
(19, "sheet_name"),
]
data_rows = []
for row_idx in range(2, ws.max_row + 1):
record: dict[str, Any] = {}
for col_idx, field_name in column_mapping:
cell_value = ws.cell(row=row_idx, column=col_idx).value
record[field_name] = str(cell_value).strip() if cell_value is not None else ""
data_rows.append(record)
wb.close()
df = pd.DataFrame(data_rows)
logger.debug(f"templates.xlsx 加载完成: {len(df)} 行")
return df
except Exception as e:
logger.error(f"templates.xlsx 加载失败: {e}")
return None
[文档]
def query_by_model(
self, product_model: str, xlsx_path: str | None = None
) -> list[dict[str, Any]]:
"""按产品型号从 templates.xlsx 查询历史模板
在「数据库」工作表的第4列（产品型号）进行部分匹配。
Args:
product_model: 产品型号（支持部分匹配）
xlsx_path: xlsx 文件路径
Returns:
匹配的模板列表，每条包含完整打印参数
"""
path = xlsx_path or self.templates_xlsx_path
if not path or not os.path.exists(path):
logger.debug("templates.xlsx 不存在，无法查询模板")
return []
try:
wb = ExcelHandler.load_workbook(Path(path), data_only=True)
ws = wb["数据库"]
results = []
for row_idx in range(2, ws.max_row + 1):
cell_val = ws.cell(row=row_idx, column=4).value
if cell_val is None:
continue
if product_model.upper() in str(cell_val).strip().upper():
row_index = ws.cell(row=row_idx, column=1).value
record = {
"row_index": int(row_index) if row_index else row_idx - 1,
"template_name": str(ws.cell(row=row_idx, column=2).value or "").strip(),
"product_name": str(ws.cell(row=row_idx, column=3).value or "").strip(),
"product_model": str(ws.cell(row=row_idx, column=4).value or "").strip(),
"dn": str(ws.cell(row=row_idx, column=5).value or "").strip(),
"pn": str(ws.cell(row=row_idx, column=6).value or "").strip(),
"temperature": str(ws.cell(row=row_idx, column=7).value or "").strip(),
"medium": str(ws.cell(row=row_idx, column=8).value or "").strip(),
"check_standard": str(ws.cell(row=row_idx, column=9).value or "").strip(),
"inspector_id": str(ws.cell(row=row_idx, column=10).value or "").strip(),
"manufacture_date": str(
ws.cell(row=row_idx, column=11).value or ""
).strip(),
"serial_number": str(ws.cell(row=row_idx, column=12).value or "").strip(),
}
results.append(record)
wb.close()
logger.info(f"按型号 '{product_model}' 查询到 {len(results)} 条模板")
return results
except Exception as e:
logger.error(f"查询模板失败 [model={product_model}]: {e}")
return []
[文档]
def writeback_to_xlsx(
self,
certificate_data: dict[str, Any],
xlsx_path: str | None = None,
) -> bool:
"""打印完成后回填到 templates.xlsx「数据库」表
如果同型号+同编号已有记录则更新，否则新增行。
Args:
certificate_data: 合格证数据字典
xlsx_path: xlsx 文件路径
Returns:
bool: 是否回填成功
"""
path = xlsx_path or self.templates_xlsx_path
if not path or not os.path.exists(path):
logger.warning("templates.xlsx 不存在，无法回填")
return False
try:
wb = ExcelHandler.load_workbook(Path(path), data_only=False)
ws = wb["数据库"]
product_model = certificate_data.get("product_model", "").strip()
cert_no = certificate_data.get("certificate_no", "").strip()
# 查找同型号+同编号范围的记录
target_row = None
for row_idx in range(2, ws.max_row + 1):
cell_model = str(ws.cell(row=row_idx, column=4).value or "").strip()
cell_sn = str(ws.cell(row=row_idx, column=12).value or "").strip()
if cell_model.upper() == product_model.upper() and cell_sn == cert_no:
target_row = row_idx
break
# 数据映射 (列索引, 值)
col_values = {
3: certificate_data.get("product_name", ""),
4: product_model,
5: certificate_data.get("product_spec", "") or certificate_data.get("dn", ""),
6: certificate_data.get("pn", ""),
7: certificate_data.get("working_temp", ""),
8: certificate_data.get("working_medium", ""),
9: certificate_data.get("test_standard", ""),
10: certificate_data.get("inspector_id", ""),
11: certificate_data.get("manufacture_date", ""),
12: cert_no,
16: datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
}
if target_row:
# 更新已有记录
for col, val in col_values.items():
if val:
ws.cell(row=target_row, column=col).value = val
logger.info(f"回填 templates.xlsx: 更新行 {target_row}, 型号={product_model}")
else:
# 新增行
new_row = ws.max_row + 1
ws.cell(row=new_row, column=1).value = ws.max_row # 序号
for col, val in col_values.items():
if val:
ws.cell(row=new_row, column=col).value = val
logger.info(f"回填 templates.xlsx: 新增行 {new_row}, 型号={product_model}")
wb.save(path)
wb.close()
return True
except Exception as e:
logger.error(f"回填 templates.xlsx 失败: {e}")
return False