"""报告生成处理器模块
基于ReportLab库生成PDF格式的试压报告、材质报告和质保书，
支持中英文双语、自定义模板样式、合并单元格和分页等功能。
"""
from __future__ import annotations
from datetime import datetime
from pathlib import Path
from typing import Any
from loguru import logger
from reportlab.lib import colors
from reportlab.lib.enums import TA_CENTER, TA_LEFT
from reportlab.lib.pagesizes import A4
from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet
from reportlab.lib.units import mm
from reportlab.pdfbase import pdfmetrics
from reportlab.pdfbase.ttfonts import TTFont
from reportlab.platypus import PageBreak, Paragraph, SimpleDocTemplate, Spacer, Table, TableStyle
[文档]
class ReportGenerator:
"""报告生成处理器 - 支持复杂模板和合并单元格
基于ReportLab生成PDF报告，支持试压报告、材质报告和质保书三种类型，
提供中文字体注册、自定义样式、分页和签名栏等功能。
Attributes:
output_dir: 输出目录路径，用于存放生成的PDF文件
styles: 报告样式集合，包含自定义的中文段落样式
"""
def __init__(self, output_dir: str = "reports") -> None:
"""初始化报告生成器
创建输出目录，注册中文字体并设置自定义样式。
Args:
output_dir: PDF文件输出目录，默认为"reports"
"""
self.output_dir: Path = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
# 注册中文字体
self._register_chinese_fonts()
# 样式
self.styles = getSampleStyleSheet()
self._setup_custom_styles()
def _register_chinese_fonts(self) -> None:
"""注册中文字体
尝试从系统字体目录注册中文字体(黑体或宋体)，
支持Windows和Linux平台。注册失败时使用默认字体。
"""
try:
# Windows 字体路径
font_paths = [
"C:/Windows/Fonts/simhei.ttf", # 黑体
"C:/Windows/Fonts/simsun.ttc", # 宋体
"/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", # Linux
]
for font_path in font_paths:
if Path(font_path).exists():
pdfmetrics.registerFont(TTFont("ChineseFont", font_path))
logger.info(f"中文字体注册成功: {font_path}")
return
logger.warning("未找到中文字体，使用默认字体")
except Exception as e:
logger.warning(f"字体注册失败: {e}")
def _setup_custom_styles(self) -> None:
"""设置自定义样式
创建中文标题、中文正文和中文居中三种自定义段落样式，
并将它们添加到样式集合中供报告生成使用。
"""
self.styles.add(
ParagraphStyle(
name="ChineseTitle",
parent=self.styles["Heading1"],
fontName="ChineseFont",
fontSize=16,
alignment=TA_CENTER,
spaceAfter=20,
)
)
self.styles.add(
ParagraphStyle(
name="ChineseNormal",
parent=self.styles["Normal"],
fontName="ChineseFont",
fontSize=10,
alignment=TA_LEFT,
)
)
self.styles.add(
ParagraphStyle(
name="ChineseCenter",
parent=self.styles["Normal"],
fontName="ChineseFont",
fontSize=10,
alignment=TA_CENTER,
)
)
[文档]
def generate_pressure_test_report(
self,
data: list[dict[str, Any]],
template_name: str = "pressure_test",
language: str = "zh",
records_per_page: int = 5,
) -> str:
"""生成试压报告
根据测试数据生成PDF格式的试压报告，支持分页和签名栏。
Args:
data: 测试数据字典列表，每条记录包含试验压力、保压时间等字段
template_name: 模板名称，默认为"pressure_test"
language: 语言，可选"zh"(中文)或"en"(英文)，默认为"zh"
records_per_page: 每页记录数，可选5/6/9，默认为5
Returns:
str: 生成的PDF文件完整路径
Examples:
>>> generator = ReportGenerator("output/reports")
>>> test_data = [
... {"test_pressure": "1.5", "hold_time": "60", "test_result": "合格"},
... {"test_pressure": "2.0", "hold_time": "90", "test_result": "合格"}
... ]
>>> filepath = generator.generate_pressure_test_report(test_data)
>>> print(f"报告已生成: {filepath}")
"""
filename = f"pressure_test_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf"
filepath = self.output_dir / filename
# 创建PDF文档
doc = SimpleDocTemplate(
str(filepath),
pagesize=A4,
topMargin=20 * mm,
bottomMargin=20 * mm,
leftMargin=15 * mm,
rightMargin=15 * mm,
)
# 构建内容
story = []
# 添加标题
title_text = "试压报告" if language == "zh" else "Pressure Test Report"
story.append(Paragraph(title_text, self.styles["ChineseTitle"]))
story.append(Spacer(1, 10))
# 添加基本信息
info_data = self._get_report_info(data, language)
info_table = self._create_info_table(info_data, language)
story.append(info_table)
story.append(Spacer(1, 20))
# 分页处理数据
pages_data = [data[i : i + records_per_page] for i in range(0, len(data), records_per_page)]
for page_idx, page_data in enumerate(pages_data):
# 创建测试数据表格
table_data = self._build_pressure_test_table(page_data, language)
table = self._create_table_with_style(table_data, template_name)
story.append(table)
# 添加签名栏
if page_idx < len(pages_data) - 1:
story.append(PageBreak())
else:
story.append(Spacer(1, 30))
signature_table = self._create_signature_table(language)
story.append(signature_table)
# 生成PDF
doc.build(story)
logger.info(f"试压报告生成成功: {filepath}")
return str(filepath)
[文档]
def generate_material_certificate(
self,
data: list[dict[str, Any]],
material_type: str = "steel",
language: str = "zh",
records_per_page: int = 6,
) -> str:
"""生成材质报告/质保书
根据材质数据生成PDF格式的材质证明书。
Args:
data: 材质数据字典列表，包含材质牌号、规格、炉号等字段
material_type: 材质类型，可选"steel"(钢材)/"valve"(阀门)/"pipe"(管材)，默认为"steel"
language: 语言，可选"zh"(中文)或"en"(英文)，默认为"zh"
records_per_page: 每页记录数，默认为6
Returns:
str: 生成的PDF文件完整路径
Examples:
>>> generator = ReportGenerator()
>>> material_data = [
... {"material_grade": "Q235B", "specification": "10mm", "heat_no": "H12345"},
... {"material_grade": "304", "specification": "8mm", "heat_no": "H67890"}
... ]
>>> filepath = generator.generate_material_certificate(material_data)
>>> print(f"材质报告已生成: {filepath}")
"""
filename = f"material_cert_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf"
filepath = self.output_dir / filename
doc = SimpleDocTemplate(str(filepath), pagesize=A4)
story = []
# 标题
title_text = "材质证明书" if language == "zh" else "Material Certificate"
story.append(Paragraph(title_text, self.styles["ChineseTitle"]))
story.append(Spacer(1, 10))
# 分页
pages_data = [data[i : i + records_per_page] for i in range(0, len(data), records_per_page)]
for page_data in pages_data:
table_data = self._build_material_table(page_data, material_type, language)
table = self._create_table_with_style(table_data, "material")
story.append(table)
if page_data != pages_data[-1]:
story.append(PageBreak())
doc.build(story)
logger.info(f"材质报告生成成功: {filepath}")
return str(filepath)
[文档]
def generate_quality_certificate(self, data: dict[str, Any], language: str = "zh") -> str:
"""生成质保书(单页)
根据质保数据生成单页PDF格式的产品质量保证书。
Args:
data: 质保书数据字典，包含产品名称、规格型号、数量等字段
language: 语言，可选"zh"(中文)或"en"(英文)，默认为"zh"
Returns:
str: 生成的PDF文件完整路径
Examples:
>>> generator = ReportGenerator()
>>> quality_data = {
... "product_name": "阀门",
... "specification": "DN100",
... "quantity": "100",
... "batch_no": "B2024001"
... }
>>> filepath = generator.generate_quality_certificate(quality_data)
>>> print(f"质保书已生成: {filepath}")
"""
filename = f"quality_cert_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pdf"
filepath = self.output_dir / filename
doc = SimpleDocTemplate(str(filepath), pagesize=A4)
story = []
# 标题
title_text = "产品质量保证书" if language == "zh" else "Quality Certificate"
story.append(Paragraph(title_text, self.styles["ChineseTitle"]))
story.append(Spacer(1, 20))
# 质保书内容
content_data = self._build_quality_table(data, language)
table = self._create_table_with_style(content_data, "quality")
story.append(table)
doc.build(story)
logger.info(f"质保书生成成功: {filepath}")
return str(filepath)
def _build_pressure_test_table(self, data: list[dict[str, Any]], language: str) -> list[list]:
"""构建试压报告表格数据(支持合并单元格)
将测试数据列表转换为表格行数据，包含序号、试验压力、保压时间、
试验结果、测试人和测试日期列，并应用合并单元格逻辑。
Args:
data: 测试数据字典列表
language: 语言，可选"zh"或"en"
Returns:
List[List]: 表格行数据列表，第一行为表头，后续为数据行
"""
headers = self._get_pressure_test_headers(language)
table_data = [headers]
for idx, record in enumerate(data, 1):
row = [
str(idx), # 序号
record.get("test_pressure", "0"), # 试验压力
record.get("hold_time", "0"), # 保压时间
record.get("test_result", "合格"), # 结果
record.get("tester", ""), # 测试人
record.get("test_date", datetime.now().strftime("%Y-%m-%d")), # 测试日期
]
table_data.append(row)
# 如果需要合并单元格(例如: 相同批次合并)
return self._apply_merge_cells(table_data, "pressure_test")
def _build_material_table(
self, data: list[dict[str, Any]], material_type: str, language: str
) -> list[list]:
"""构建材质报告表格
将材质数据列表转换为表格行数据，包含序号、材质牌号、规格、
数量、炉号、化学成分、机械性能和证书号列。
Args:
data: 材质数据字典列表
material_type: 材质类型，用于决定表格列配置
language: 语言，可选"zh"或"en"
Returns:
List[List]: 表格行数据列表，第一行为表头，后续为数据行
"""
headers = self._get_material_headers(language)
table_data = [headers]
for idx, record in enumerate(data, 1):
row = [
str(idx),
record.get("material_grade", ""), # 材质牌号
record.get("specification", ""), # 规格
record.get("quantity", "1"), # 数量
record.get("heat_no", ""), # 炉号
record.get("chemical_composition", ""), # 化学成分
record.get("mechanical_property", ""), # 机械性能
record.get("cert_no", ""), # 证书号
]
table_data.append(row)
return table_data
def _build_quality_table(self, data: dict[str, Any], language: str) -> list[list]:
"""构建质保书表格
将质保数据转换为两列表格行数据。
Args:
data: 质保书数据字典
language: 语言，可选"zh"或"en"
Returns:
List[List]: 表格行数据列表，格式为[[字段名, 值], ...]
"""
return [
["产品名称", data.get("product_name", "")],
["规格型号", data.get("specification", "")],
["数量", data.get("quantity", "")],
["生产批号", data.get("batch_no", "")],
["检验标准", data.get("standard", "GB/T")],
["化学成分", data.get("chemical_composition", "")],
["机械性能", data.get("mechanical_property", "")],
["检验结果", data.get("result", "合格")],
["检验日期", data.get("test_date", datetime.now().strftime("%Y-%m-%d"))],
["有效期至", data.get("expiry_date", "")],
]
def _create_table_with_style(self, table_data: list[list], template_type: str) -> Table:
"""创建带样式的表格
根据模板类型计算列宽，创建表格并应用基础样式、表头样式和交替行颜色。
Args:
table_data: 表格行数据列表
template_type: 模板类型，如"pressure_test"、"material"、"quality"
Returns:
Table: 带样式的ReportLab表格对象
"""
# 计算列宽
col_widths = self._calculate_col_widths(table_data, template_type)
table = Table(table_data, colWidths=col_widths, repeatRows=1)
# 基础样式
style = TableStyle(
[
("GRID", (0, 0), (-1, -1), 1, colors.black),
("FONTNAME", (0, 0), (-1, -1), "ChineseFont"),
("FONTSIZE", (0, 0), (-1, -1), 9),
("ALIGN", (0, 0), (-1, -1), "CENTER"),
("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
]
)
# 表头样式
style.add("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#4472C4"))
style.add("TEXTCOLOR", (0, 0), (-1, 0), colors.white)
style.add("FONTNAME", (0, 0), (-1, 0), "ChineseFont")
style.add("FONTSIZE", (0, 0), (-1, 0), 10)
style.add("BOLD", (0, 0), (-1, 0), 1)
# 交替行颜色
for i in range(1, len(table_data)):
if i % 2 == 0:
style.add("BACKGROUND", (0, i), (-1, i), colors.HexColor("#E9ECF4"))
table.setStyle(style)
return table
def _calculate_col_widths(self, table_data: list[list], template_type: str) -> list[float]:
"""计算列宽
根据模板类型返回预定义的列宽列表，未知模板则平均分配A4页面宽度。
Args:
table_data: 表格行数据列表(用于确定列数)
template_type: 模板类型
Returns:
List[float]: 列宽列表，单位为毫米(mm)
"""
num_cols = len(table_data[0]) if table_data else 1
if template_type == "pressure_test":
widths = [30, 50, 50, 50, 80, 80] # mm
elif template_type == "material":
widths = [30, 60, 60, 40, 60, 80, 80, 80]
elif template_type == "quality":
widths = [60, 120]
else:
# 平均分配
page_width = A4[0] - 30 * mm
widths = [page_width / num_cols] * num_cols
return widths
def _apply_merge_cells(self, table_data: list[list], template_type: str) -> list[list]:
"""应用合并单元格逻辑
对试压报告模板中相同测试压力的行进行合并(预留接口)。
Args:
table_data: 表格行数据列表
template_type: 模板类型
Returns:
List[List]: 处理后的表格行数据列表
"""
if template_type != "pressure_test":
return table_data
# 示例: 合并相同测试压力的行
if len(table_data) > 2:
# 这里可以根据业务需求实现合并逻辑
pass
return table_data
def _get_pressure_test_headers(self, language: str) -> list[str]:
"""获取试压报告表头
根据语言返回中文或英文的试压报告列标题。
Args:
language: 语言，可选"zh"或"en"
Returns:
List[str]: 表头字符串列表
"""
if language == "zh":
return ["序号", "试验压力(MPa)", "保压时间(s)", "试验结果", "测试人", "测试日期"]
return ["No.", "Test Pressure(MPa)", "Hold Time(s)", "Result", "Tester", "Test Date"]
def _get_material_headers(self, language: str) -> list[str]:
"""获取材质报告表头
根据语言返回中文或英文的材质报告列标题。
Args:
language: 语言，可选"zh"或"en"
Returns:
List[str]: 表头字符串列表
"""
if language == "zh":
return ["序号", "材质牌号", "规格", "数量", "炉号", "化学成分", "机械性能", "证书号"]
return ["No.", "Grade", "Spec", "Qty", "Heat No.", "Chem Comp", "Mech Prop", "Cert No."]
def _get_report_info(self, data: list[dict[str, Any]], language: str) -> dict[str, Any]:
"""获取报告基本信息
从数据列表中提取报告编号、客户名称、订单号、测试日期和总数量。
Args:
data: 数据字典列表
language: 语言，可选"zh"或"en"
Returns:
Dict[str, Any]: 报告基本信息字典，包含以下键:
- report_no: 报告编号
- customer: 客户名称
- order_no: 订单号
- test_date: 测试日期
- total_count: 总数量
"""
return {
"report_no": f"RPT-{datetime.now().strftime('%Y%m%d%H%M%S')}",
"customer": data[0].get("customer", "") if data else "",
"order_no": data[0].get("order_no", "") if data else "",
"test_date": datetime.now().strftime("%Y-%m-%d"),
"total_count": len(data),
}
def _create_info_table(self, info_data: dict[str, Any], language: str) -> Table:
"""创建信息表格
根据语言创建报告基本信息表格，包含报告编号、客户、订单号等。
Args:
info_data: 报告基本信息字典
language: 语言，可选"zh"或"en"
Returns:
Table: ReportLab表格对象
"""
headers = (
["报告编号", "客户名称", "订单号", "测试日期", "总数量"]
if language == "zh"
else ["Report No.", "Customer", "Order No.", "Test Date", "Total"]
)
values = [
info_data["report_no"],
info_data["customer"],
info_data["order_no"],
info_data["test_date"],
str(info_data["total_count"]),
]
table_data = [headers, values]
table = Table(table_data, colWidths=[60, 80, 80, 60, 40])
table.setStyle(
TableStyle(
[
("GRID", (0, 0), (-1, -1), 1, colors.black),
("FONTNAME", (0, 0), (-1, -1), "ChineseFont"),
("FONTSIZE", (0, 0), (-1, -1), 9),
("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#D9E1F2")),
("ALIGN", (0, 0), (-1, -1), "CENTER"),
]
)
)
return table
def _create_signature_table(self, language: str) -> Table:
"""创建签名表格
根据语言创建包含检验员、审核员、批准人和日期的签名栏表格。
Args:
language: 语言，可选"zh"或"en"
Returns:
Table: ReportLab表格对象
"""
if language == "zh":
data = [
["检验员", "审核员", "批准人", "日期"],
["", "", "", datetime.now().strftime("%Y-%m-%d")],
]
else:
data = [
["Inspector", "Reviewer", "Approver", "Date"],
["", "", "", datetime.now().strftime("%Y-%m-%d")],
]
table = Table(data, colWidths=[80, 80, 80, 80])
table.setStyle(
TableStyle(
[
("GRID", (0, 0), (-1, -1), 1, colors.black),
("FONTNAME", (0, 0), (-1, -1), "ChineseFont"),
("FONTSIZE", (0, 0), (-1, -1), 9),
("ALIGN", (0, 0), (-1, -1), "CENTER"),
("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
]
)
)
return table