378 lines
16 KiB
Python
378 lines
16 KiB
Python
"""PDF rendering for the prototype-aligned risk report."""
|
||
|
||
from io import BytesIO
|
||
from pathlib import Path
|
||
|
||
from docx import Document
|
||
from docx.enum.table import WD_CELL_VERTICAL_ALIGNMENT
|
||
from docx.enum.text import WD_ALIGN_PARAGRAPH
|
||
from docx.oxml import OxmlElement
|
||
from docx.shared import Pt, RGBColor
|
||
from reportlab.lib import colors
|
||
from reportlab.lib.enums import TA_CENTER
|
||
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 Paragraph, SimpleDocTemplate, Table, TableStyle
|
||
|
||
from app.schemas.risk_prediction import RiskAssessmentResult
|
||
|
||
LEVEL_LABELS = {
|
||
"low": "低风险",
|
||
"moderately_low": "较低风险",
|
||
"medium": "中等风险",
|
||
"high": "高风险",
|
||
"very_high": "极高风险",
|
||
}
|
||
|
||
|
||
def _font_name() -> str:
|
||
candidates = [
|
||
Path("C:/Windows/Fonts/msyh.ttc"),
|
||
Path("C:/Windows/Fonts/msyh.ttf"),
|
||
Path("C:/Windows/Fonts/simhei.ttf"),
|
||
]
|
||
for path in candidates:
|
||
if path.exists():
|
||
name = "DieyaCjk"
|
||
pdfmetrics.registerFont(TTFont(name, str(path), subfontIndex=0))
|
||
return name
|
||
return "Helvetica"
|
||
|
||
|
||
def build_risk_report_pdf(result: RiskAssessmentResult) -> bytes:
|
||
font = _font_name()
|
||
styles = getSampleStyleSheet()
|
||
title = ParagraphStyle(
|
||
"title", parent=styles["Title"], fontName=font, fontSize=20, alignment=TA_CENTER, spaceAfter=4
|
||
)
|
||
subtitle = ParagraphStyle(
|
||
"subtitle",
|
||
parent=styles["Normal"],
|
||
fontName=font,
|
||
fontSize=9,
|
||
textColor=colors.HexColor("#64748b"),
|
||
alignment=TA_CENTER,
|
||
spaceAfter=14,
|
||
)
|
||
section = ParagraphStyle(
|
||
"section",
|
||
parent=styles["Heading2"],
|
||
fontName=font,
|
||
fontSize=11,
|
||
textColor=colors.HexColor("#111827"),
|
||
leftIndent=5,
|
||
borderPadding=3,
|
||
spaceBefore=12,
|
||
spaceAfter=7,
|
||
)
|
||
body = ParagraphStyle("body", parent=styles["BodyText"], fontName=font, fontSize=9, leading=15)
|
||
small = ParagraphStyle("small", parent=body, fontSize=8, textColor=colors.HexColor("#64748b"))
|
||
supplier_name = result.supplier_freq.supplier_name if result.supplier_freq else None
|
||
report_number = f"RP-{result.subject_id[:8]}-{round(result.score)}"
|
||
generated_at = __import__("datetime").datetime.now().strftime("%Y-%m-%d %H:%M")
|
||
story = [
|
||
Paragraph("新品风险预估报告", title),
|
||
Paragraph(f"蝶雅产品部 · 风险预估系统 · 自动生成 · {generated_at}", subtitle),
|
||
Paragraph(
|
||
f"产品/评估对象：{result.subject_id} 供应商：{supplier_name or '未指定'} 报告编号：{report_number}",
|
||
body,
|
||
),
|
||
]
|
||
|
||
def table(rows: list[list[str]], widths: list[float] | None = None) -> Table:
|
||
converted = [[Paragraph(str(cell), body) for cell in row] for row in rows]
|
||
item = Table(converted, colWidths=widths, repeatRows=1)
|
||
item.setStyle(
|
||
TableStyle(
|
||
[
|
||
("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#111827")),
|
||
("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
|
||
("FONTNAME", (0, 0), (-1, -1), font),
|
||
("GRID", (0, 0), (-1, -1), 0.45, colors.HexColor("#dbe3ef")),
|
||
("VALIGN", (0, 0), (-1, -1), "TOP"),
|
||
("LEFTPADDING", (0, 0), (-1, -1), 7),
|
||
("RIGHTPADDING", (0, 0), (-1, -1), 7),
|
||
("TOPPADDING", (0, 0), (-1, -1), 6),
|
||
("BOTTOMPADDING", (0, 0), (-1, -1), 6),
|
||
]
|
||
)
|
||
)
|
||
return item
|
||
|
||
story += [Paragraph("一、风险结论", section)]
|
||
rows = [
|
||
["评估项", "结论"],
|
||
[
|
||
"综合评分",
|
||
f"{round(result.score)} / 100 · "
|
||
f"{LEVEL_LABELS.get(result.level.value, result.level.value)} · "
|
||
f"{result.risk_badge or '需关注'}",
|
||
],
|
||
["风险警示", result.alert_text or "当前暂无足够历史数据，建议按标准库完成基础验证。"],
|
||
]
|
||
dimensions = result.dimension_scores or [
|
||
{"dimension": factor.name, "score": 100 - factor.value, "weight": factor.weight, "reason": None}
|
||
for factor in result.factors
|
||
]
|
||
for item in dimensions:
|
||
name = item.dimension if hasattr(item, "dimension") else item["dimension"]
|
||
if name in {"节点阶段", "缺陷等级"}:
|
||
continue
|
||
score = item.score if hasattr(item, "score") else item["score"]
|
||
weight = item.weight if hasattr(item, "weight") else item["weight"]
|
||
reason = item.reason if hasattr(item, "reason") else item.get("reason")
|
||
supplier = result.supplier_freq
|
||
if name == "供应商表现" and supplier:
|
||
order_text = f"合作 {supplier.cooperation_count} 次、" if supplier.cooperation_count is not None else ""
|
||
conclusion = f"{round(score)} 分 · {order_text}异常 {supplier.anomaly_count or 0} 次"
|
||
elif name == "标准符合性" and result.standard_judgment:
|
||
conclusion = f"{round(score)} 分 · 命中 {len(result.standard_judgment)} 条判定标准，待检测项需人工确认"
|
||
else:
|
||
conclusion = f"{round(score)} 分" + (f" · {reason}" if reason else "")
|
||
rows.append([f"{name}（{weight:g}%）", conclusion])
|
||
story += [table(rows, [55 * mm, 105 * mm])]
|
||
|
||
if result.category_warning:
|
||
story += [
|
||
Paragraph("二、品类风险预警", section),
|
||
table(
|
||
[
|
||
["品类", "历史异常", "风险等级", "建议"],
|
||
[
|
||
result.category_warning.category,
|
||
f"{result.category_warning.anomaly_count} 条",
|
||
result.category_warning.risk_level,
|
||
result.category_warning.suggestion or "结合相似品类和标准库预判",
|
||
],
|
||
],
|
||
[35 * mm, 30 * mm, 30 * mm, 65 * mm],
|
||
),
|
||
]
|
||
|
||
supplier = result.supplier_freq
|
||
if supplier and supplier.supplier_name:
|
||
supplier_rows = [["维度", "数值", "说明"]]
|
||
if supplier.cooperation_count is not None:
|
||
supplier_rows.append(["合作次数", f"{supplier.cooperation_count} 次", "近三年采购订单去重单据数"])
|
||
supplier_rows.append(["异常次数", f"{supplier.anomaly_count or 0} 次", "异常案例库统计"])
|
||
if supplier.alpha is not None:
|
||
supplier_rows.append(
|
||
["频次风险系数", f"×{supplier.alpha:.2f}", "α = 1.0 +（采购订单数 / 100）× 0.3，最高1.5"]
|
||
)
|
||
if supplier.tier:
|
||
supplier_rows.append(["供应商层级", supplier.tier, supplier.suggestion or ""])
|
||
story += [Paragraph("三、供应商合作频次分析", section), table(supplier_rows, [38 * mm, 35 * mm, 87 * mm])]
|
||
|
||
story += [Paragraph(f"四、相似案例匹配（{len(result.similar_cases)} 条）", section)]
|
||
if result.similar_cases:
|
||
story.append(
|
||
table(
|
||
[["日期", "产品", "问题", "处理方案"]]
|
||
+ [
|
||
[case.date or "未知", case.product or "未命名", case.problem or "未填写", case.solution or "未填写"]
|
||
for case in result.similar_cases
|
||
],
|
||
[28 * mm, 42 * mm, 48 * mm, 42 * mm],
|
||
)
|
||
)
|
||
else:
|
||
story.append(Paragraph("暂无相关案例，RAG 相似案例检索未返回匹配结果。", small))
|
||
|
||
story += [Paragraph("五、预防建议", section)]
|
||
for index, suggestion in enumerate(
|
||
result.ai_suggestions or ["优先针对低分维度安排专项验证，并明确责任人与完成节点。"], 1
|
||
):
|
||
story.append(Paragraph(f"{index}. {suggestion}", body))
|
||
story += [
|
||
Paragraph("六、AI 推理方案（无历史案例场景）", section),
|
||
Paragraph(
|
||
"本次评估已采用历史案例与标准库双引擎结论；若后续未匹配到足够历史案例，将基于相似品类、相似物料生成类比方案，并由标准库进行规则兜底。",
|
||
body,
|
||
),
|
||
]
|
||
|
||
story += [Paragraph("七、标准判定结果", section)]
|
||
if result.standard_judgment:
|
||
story.append(
|
||
table(
|
||
[["判定维度", "判定标准", "标准来源", "当前状态"]]
|
||
+ [
|
||
[f"{item.category} · {item.risk_point}", item.rule, item.source or "标准文件.xlsx", item.status]
|
||
for item in result.standard_judgment
|
||
],
|
||
[38 * mm, 55 * mm, 40 * mm, 27 * mm],
|
||
)
|
||
)
|
||
else:
|
||
story.append(Paragraph("暂无相关标准。", small))
|
||
story.append(Paragraph("双引擎融合：历史异常经验引擎提供风险优先级，标准库提供可执行的检测阈值与验证任务。", small))
|
||
|
||
output = BytesIO()
|
||
doc = SimpleDocTemplate(
|
||
output,
|
||
pagesize=A4,
|
||
rightMargin=18 * mm,
|
||
leftMargin=18 * mm,
|
||
topMargin=16 * mm,
|
||
bottomMargin=16 * mm,
|
||
title="蝶雅新品风险预估报告",
|
||
)
|
||
doc.build(story)
|
||
return output.getvalue()
|
||
|
||
|
||
def build_risk_report_docx(result: RiskAssessmentResult) -> bytes:
|
||
document = Document()
|
||
section = document.sections[0]
|
||
section.page_width = 210 * 36000
|
||
section.page_height = 297 * 36000
|
||
section.top_margin = 16 * 36000
|
||
section.bottom_margin = 16 * 36000
|
||
section.left_margin = 18 * 36000
|
||
section.right_margin = 18 * 36000
|
||
normal = document.styles["Normal"]
|
||
normal.font.name = "微软雅黑"
|
||
normal.font.size = Pt(9)
|
||
title = document.add_paragraph()
|
||
title.alignment = WD_ALIGN_PARAGRAPH.CENTER
|
||
run = title.add_run("新品风险预估报告")
|
||
run.bold = True
|
||
run.font.size = Pt(20)
|
||
subtitle = document.add_paragraph()
|
||
subtitle.alignment = WD_ALIGN_PARAGRAPH.CENTER
|
||
subtitle.add_run(f"蝶雅产品部 · 风险预估系统 · 自动生成 · {result.subject_id}").font.size = Pt(9)
|
||
|
||
def heading(text: str) -> None:
|
||
p = document.add_paragraph()
|
||
p.paragraph_format.space_before = Pt(12)
|
||
p.paragraph_format.space_after = Pt(6)
|
||
p.paragraph_format.left_indent = Pt(5)
|
||
r = p.add_run(text)
|
||
r.bold = True
|
||
r.font.size = Pt(11)
|
||
|
||
def add_table(rows: list[list[str]]) -> None:
|
||
table = document.add_table(rows=len(rows), cols=len(rows[0]))
|
||
table.style = "Table Grid"
|
||
table.autofit = True
|
||
for row_index, row in enumerate(rows):
|
||
for col_index, value in enumerate(row):
|
||
cell = table.cell(row_index, col_index)
|
||
cell.vertical_alignment = WD_CELL_VERTICAL_ALIGNMENT.TOP
|
||
cell.text = str(value)
|
||
for paragraph in cell.paragraphs:
|
||
paragraph.paragraph_format.space_after = Pt(0)
|
||
for run in paragraph.runs:
|
||
run.font.name = "微软雅黑"
|
||
run.font.size = Pt(9)
|
||
if row_index == 0:
|
||
run.bold = True
|
||
run.font.color.rgb = RGBColor(255, 255, 255)
|
||
if row_index == 0:
|
||
shading = OxmlElement("w:shd")
|
||
shading.set("{http://schemas.openxmlformats.org/wordprocessingml/2006/main}fill", "111827")
|
||
cell._tc.get_or_add_tcPr().append(shading)
|
||
document.add_paragraph()
|
||
|
||
heading("一、风险结论")
|
||
rows = [
|
||
["评估项", "结论"],
|
||
[
|
||
"综合评分",
|
||
f"{round(result.score)} / 100 · "
|
||
f"{LEVEL_LABELS.get(result.level.value, result.level.value)} · "
|
||
f"{result.risk_badge or '需关注'}",
|
||
],
|
||
["风险警示", result.alert_text or "当前暂无足够历史数据，建议按标准库完成基础验证。"],
|
||
]
|
||
dimensions = result.dimension_scores or [
|
||
{
|
||
"dimension": factor.name,
|
||
"score": 100 - factor.value,
|
||
"weight": factor.weight,
|
||
"reason": None,
|
||
}
|
||
for factor in result.factors
|
||
]
|
||
for item in dimensions:
|
||
name = item.dimension if hasattr(item, "dimension") else item["dimension"]
|
||
if name in {"节点阶段", "缺陷等级"}:
|
||
continue
|
||
score = item.score if hasattr(item, "score") else item["score"]
|
||
weight = item.weight if hasattr(item, "weight") else item["weight"]
|
||
reason = item.reason if hasattr(item, "reason") else item.get("reason")
|
||
supplier = result.supplier_freq
|
||
if name == "供应商表现" and supplier:
|
||
order_text = f"合作 {supplier.cooperation_count} 次、" if supplier.cooperation_count is not None else ""
|
||
conclusion = f"{round(score)} 分 · {order_text}异常 {supplier.anomaly_count or 0} 次"
|
||
elif name == "标准符合性" and result.standard_judgment:
|
||
conclusion = f"{round(score)} 分 · 命中 {len(result.standard_judgment)} 条判定标准，待检测项需人工确认"
|
||
else:
|
||
conclusion = f"{round(score)} 分" + (f" · {reason}" if reason else "")
|
||
rows.append([f"{name}（{weight:g}%）", conclusion])
|
||
add_table(rows)
|
||
if result.category_warning:
|
||
heading("二、品类风险预警")
|
||
add_table(
|
||
[
|
||
["品类", "历史异常", "风险等级", "建议"],
|
||
[
|
||
result.category_warning.category,
|
||
f"{result.category_warning.anomaly_count} 条",
|
||
result.category_warning.risk_level,
|
||
result.category_warning.suggestion or "结合相似品类和标准库预判",
|
||
],
|
||
]
|
||
)
|
||
if result.supplier_freq and result.supplier_freq.supplier_name:
|
||
heading("三、供应商合作频次分析")
|
||
rows = [["维度", "数值", "说明"]]
|
||
if result.supplier_freq.cooperation_count is not None:
|
||
rows.append(["合作次数", f"{result.supplier_freq.cooperation_count} 次", "近三年采购订单去重单据数"])
|
||
rows.append(["异常次数", f"{result.supplier_freq.anomaly_count or 0} 次", "异常案例库统计"])
|
||
if result.supplier_freq.alpha is not None:
|
||
rows.append(
|
||
["频次风险系数", f"×{result.supplier_freq.alpha:.2f}", "α = 1.0 +（采购订单数 / 100）× 0.3，最高1.5"]
|
||
)
|
||
if result.supplier_freq.tier:
|
||
rows.append(["供应商层级", result.supplier_freq.tier, result.supplier_freq.suggestion or ""])
|
||
add_table(rows)
|
||
heading(f"四、相似案例匹配（{len(result.similar_cases)} 条）")
|
||
if result.similar_cases:
|
||
add_table(
|
||
[["日期", "产品", "问题", "处理方案"]]
|
||
+ [
|
||
[case.date or "未知", case.product or "未命名", case.problem or "未填写", case.solution or "未填写"]
|
||
for case in result.similar_cases
|
||
]
|
||
)
|
||
else:
|
||
document.add_paragraph("暂无相关案例，RAG 相似案例检索未返回匹配结果。")
|
||
heading("五、预防建议")
|
||
for index, suggestion in enumerate(
|
||
result.ai_suggestions or ["优先针对低分维度安排专项验证，并明确责任人与完成节点。"], 1
|
||
):
|
||
document.add_paragraph(f"{index}. {suggestion}")
|
||
heading("六、AI 推理方案（无历史案例场景）")
|
||
document.add_paragraph(
|
||
"本次评估已采用历史案例与标准库双引擎结论；若后续未匹配到足够历史案例，将基于相似品类、相似物料生成类比方案，并由标准库进行规则兜底。"
|
||
)
|
||
heading("七、标准判定结果")
|
||
if result.standard_judgment:
|
||
add_table(
|
||
[["判定维度", "判定标准", "标准来源", "当前状态"]]
|
||
+ [
|
||
[f"{item.category} · {item.risk_point}", item.rule, item.source or "标准文件.xlsx", item.status]
|
||
for item in result.standard_judgment
|
||
]
|
||
)
|
||
else:
|
||
document.add_paragraph("暂无相关标准。")
|
||
document.add_paragraph("双引擎融合：历史异常经验引擎提供风险优先级，标准库提供可执行的检测阈值与验证任务。")
|
||
output = BytesIO()
|
||
document.save(output)
|
||
return output.getvalue()
|