diff --git a/app/services/reports.py b/app/services/reports.py index 666ea6b..8f46c6b 100644 --- a/app/services/reports.py +++ b/app/services/reports.py @@ -2,6 +2,7 @@ from io import BytesIO from pathlib import Path +from typing import Any from docx import Document from docx.enum.table import WD_CELL_VERTICAL_ALIGNMENT @@ -143,6 +144,65 @@ def report_card_lines(card: ReportCard) -> list[tuple[str, str]]: return lines +def _unique_by(items: list[Any], key) -> list[Any]: # noqa: ANN001 + seen: set[Any] = set() + result: list[Any] = [] + for item in items: + value = key(item) + if value in seen: + continue + seen.add(value) + result.append(item) + return result + + +def aggregate_report_data(result: RiskAssessmentResult) -> dict[str, Any]: + """Merge appended cards into the report's native sections. + + A card is an input action, not a report section. Its fields are folded + into the matching section so the browser, PDF and Word output have the + same information architecture. + """ + + cards = effective_report_cards(result) + supplier_cards = [card for card in cards if card.type == "supplier"] + material_cards = [card for card in cards if card.type == "material"] + solution_cards = [card for card in cards if card.type == "solution"] + standards = _unique_by( + [*result.standard_judgment, *(item for card in cards for item in card.standards)], + lambda item: (item.category, item.risk_point, item.rule, item.status), + ) + similar_cases = _unique_by( + [*result.similar_cases, *(item for card in cards for item in card.similar_cases)], + lambda item: ( + item.case_id, + item.date, + item.product, + item.problem or item.description, + item.solution, + ), + ) + suggestions = _unique_by( + [*result.ai_suggestions, *(item for card in cards for item in card.suggestions)], + lambda item: item, + ) + detected_values = [item for card in cards for item in card.detected_values] + risks = [(card.title, risk) for card in solution_cards for risk in card.risks] + analyses = [card for card in solution_cards if card.analysis_base or card.confidence or card.warning] + return { + "cards": cards, + "supplier_cards": supplier_cards, + "material_cards": material_cards, + "solution_cards": solution_cards, + "standards": standards, + "similar_cases": similar_cases, + "suggestions": suggestions, + "detected_values": detected_values, + "risks": risks, + "analyses": analyses, + } + + def _font_name() -> str: candidates = [ Path("C:/Windows/Fonts/msyh.ttc"), @@ -188,6 +248,10 @@ def build_risk_report_pdf(result: RiskAssessmentResult) -> bytes: 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") + report_data = aggregate_report_data(result) + standards = report_data["standards"] + similar_cases = report_data["similar_cases"] + suggestions = report_data["suggestions"] story = [ Paragraph("新品风险预估报告", title), Paragraph(f"蝶雅产品部 · 风险预估系统 · 自动生成 · {generated_at}", subtitle), @@ -268,27 +332,52 @@ def build_risk_report_pdf(result: RiskAssessmentResult) -> bytes: ] supplier = result.supplier_freq - if supplier and supplier.supplier_name: + supplier_cards = report_data["supplier_cards"] + if (supplier and supplier.supplier_name) or supplier_cards: supplier_rows = [["维度", "数值", "说明"]] - if supplier.cooperation_count is not None: + if supplier and 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: + if supplier: + supplier_rows.append(["异常次数", f"{supplier.anomaly_count or 0} 次", "异常案例库统计"]) + if supplier and supplier.alpha is not None: supplier_rows.append( ["频次风险系数", f"×{supplier.alpha:.2f}", "α = 1.0 +（采购订单数 / 100）× 0.3，最高1.5"] ) - if supplier.tier: + if supplier and supplier.tier: supplier_rows.append(["供应商层级", supplier.tier, supplier.suggestion or ""]) + for card in supplier_cards: + if card.name: + supplier_rows.append(["供应商名称", card.name, "供应商画像统计"]) + for label, value in report_card_lines(card): + if label not in {"卡片类型", "分析对象", "对象类型"}: + supplier_rows.append([label, value, "追加供应商画像数据"]) story += [Paragraph("三、供应商合作频次分析", section), table(supplier_rows, [38 * mm, 35 * mm, 87 * mm])] - story += [Paragraph(f"四、相似案例匹配（{len(result.similar_cases)} 条）", section)] - if result.similar_cases: + material_cards = report_data["material_cards"] + if material_cards: + story += [Paragraph("四、物料风险分析", section)] + material_rows = [["物料/字段", "分析结果"]] + for card in material_cards: + material_rows.append( + [ + card.material or card.title, + "；".join( + f"{label}：{value}" + for label, value in report_card_lines(card) + if label not in {"卡片类型", "分析对象"} + ), + ] + ) + story.append(table(material_rows, [45 * mm, 115 * mm])) + + story += [Paragraph(f"五、相似案例匹配（{len(similar_cases)} 条）", section)] + if similar_cases: story.append( table( [["日期", "产品", "问题", "处理方案"]] + [ [case.date or "未知", case.product or "未命名", case.problem or "未填写", case.solution or "未填写"] - for case in result.similar_cases + for case in similar_cases ], [28 * mm, 42 * mm, 48 * mm, 42 * mm], ) @@ -296,27 +385,35 @@ def build_risk_report_pdf(result: RiskAssessmentResult) -> bytes: else: story.append(Paragraph("暂无相关案例，RAG 相似案例检索未返回匹配结果。", small)) - story += [Paragraph("五、预防建议", section)] - for index, suggestion in enumerate( - result.ai_suggestions or ["优先针对低分维度安排专项验证，并明确责任人与完成节点。"], 1 - ): + story += [Paragraph("六、预防建议", section)] + for index, suggestion in enumerate(suggestions or ["优先针对低分维度安排专项验证，并明确责任人与完成节点。"], 1): story.append(Paragraph(f"{index}. {suggestion}", body)) story += [ - Paragraph("六、AI 推理方案（无历史案例场景）", section), + Paragraph("七、AI 推理方案（无历史案例场景）", section), Paragraph( "本次评估已采用历史案例与标准库双引擎结论；若后续未匹配到足够历史案例，将基于相似品类、相似物料生成类比方案，并由标准库进行规则兜底。", body, ), ] - story += [Paragraph("七、标准判定结果", section)] - if result.standard_judgment: + for card in report_data["solution_cards"]: + if card.analysis_base: + story.append(Paragraph(f"分析基础：{card.analysis_base}", body)) + if card.confidence: + story.append(Paragraph(f"置信度：{card.confidence}", body)) + for risk in card.risks: + story.append(Paragraph(f"{risk.level} · {risk.title}：{risk.detail}", body)) + if card.warning: + story.append(Paragraph(f"风险提示：{card.warning}", body)) + + story += [Paragraph("八、标准判定结果", section)] + if standards: story.append( table( [["判定维度", "判定标准", "标准来源", "当前状态"]] + [ [f"{item.category} · {item.risk_point}", item.rule, item.source or "标准文件.xlsx", item.status] - for item in result.standard_judgment + for item in standards ], [38 * mm, 55 * mm, 40 * mm, 27 * mm], ) @@ -325,18 +422,6 @@ def build_risk_report_pdf(result: RiskAssessmentResult) -> bytes: story.append(Paragraph("暂无相关标准。", small)) story.append(Paragraph("双引擎融合：历史异常经验引擎提供风险优先级，标准库提供可执行的检测阈值与验证任务。", small)) - cards = effective_report_cards(result) - if cards: - story += [Paragraph("八、追加卡片分析", section)] - for index, card in enumerate(cards, 1): - story.append(Paragraph(f"{index}. {card.title}", body)) - story.append( - table( - [["字段", "内容"]] + [[label, value] for label, value in report_card_lines(card)], - [38 * mm, 122 * mm], - ) - ) - output = BytesIO() doc = SimpleDocTemplate( output, @@ -371,6 +456,10 @@ def build_risk_report_docx(result: RiskAssessmentResult) -> bytes: subtitle = document.add_paragraph() subtitle.alignment = WD_ALIGN_PARAGRAPH.CENTER subtitle.add_run(f"蝶雅产品部 · 风险预估系统 · 自动生成 · {result.subject_id}").font.size = Pt(9) + report_data = aggregate_report_data(result) + standards = report_data["standards"] + similar_cases = report_data["similar_cases"] + suggestions = report_data["suggestions"] def heading(text: str) -> None: p = document.add_paragraph() @@ -454,58 +543,83 @@ def build_risk_report_docx(result: RiskAssessmentResult) -> bytes: ], ] ) - if result.supplier_freq and result.supplier_freq.supplier_name: + supplier_cards = report_data["supplier_cards"] + if (result.supplier_freq and result.supplier_freq.supplier_name) or supplier_cards: heading("三、供应商合作频次分析") rows = [["维度", "数值", "说明"]] - if result.supplier_freq.cooperation_count is not None: + if result.supplier_freq and 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: + if result.supplier_freq: + rows.append(["异常次数", f"{result.supplier_freq.anomaly_count or 0} 次", "异常案例库统计"]) + if result.supplier_freq and 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: + if result.supplier_freq and result.supplier_freq.tier: rows.append(["供应商层级", result.supplier_freq.tier, result.supplier_freq.suggestion or ""]) + for card in supplier_cards: + if card.name: + rows.append(["供应商名称", card.name, "供应商画像统计"]) + for label, value in report_card_lines(card): + if label not in {"卡片类型", "分析对象", "对象类型"}: + rows.append([label, value, "追加供应商画像数据"]) add_table(rows) - heading(f"四、相似案例匹配（{len(result.similar_cases)} 条）") - if result.similar_cases: + material_cards = report_data["material_cards"] + if material_cards: + heading("四、物料风险分析") + rows = [["物料/字段", "分析结果"]] + for card in material_cards: + rows.append( + [ + card.material or card.title, + "；".join( + f"{label}：{value}" + for label, value in report_card_lines(card) + if label not in {"卡片类型", "分析对象"} + ), + ] + ) + add_table(rows) + heading(f"五、相似案例匹配（{len(similar_cases)} 条）") + if similar_cases: add_table( [["日期", "产品", "问题", "处理方案"]] + [ [case.date or "未知", case.product or "未命名", case.problem or "未填写", case.solution or "未填写"] - for case in result.similar_cases + for case in similar_cases ] ) else: document.add_paragraph("暂无相关案例，RAG 相似案例检索未返回匹配结果。") - heading("五、预防建议") - for index, suggestion in enumerate( - result.ai_suggestions or ["优先针对低分维度安排专项验证，并明确责任人与完成节点。"], 1 - ): + heading("六、预防建议") + for index, suggestion in enumerate(suggestions or ["优先针对低分维度安排专项验证，并明确责任人与完成节点。"], 1): document.add_paragraph(f"{index}. {suggestion}") - heading("六、AI 推理方案（无历史案例场景）") + heading("七、AI 推理方案（无历史案例场景）") document.add_paragraph( "本次评估已采用历史案例与标准库双引擎结论；若后续未匹配到足够历史案例，将基于相似品类、相似物料生成类比方案，并由标准库进行规则兜底。" ) - heading("七、标准判定结果") - if result.standard_judgment: + for card in report_data["solution_cards"]: + if card.analysis_base: + document.add_paragraph(f"分析基础：{card.analysis_base}") + if card.confidence: + document.add_paragraph(f"置信度：{card.confidence}") + for risk in card.risks: + document.add_paragraph(f"{risk.level} · {risk.title}：{risk.detail}") + if card.warning: + document.add_paragraph(f"风险提示：{card.warning}") + heading("八、标准判定结果") + if standards: add_table( [["判定维度", "判定标准", "标准来源", "当前状态"]] + [ [f"{item.category} · {item.risk_point}", item.rule, item.source or "标准文件.xlsx", item.status] - for item in result.standard_judgment + for item in standards ] ) else: document.add_paragraph("暂无相关标准。") document.add_paragraph("双引擎融合：历史异常经验引擎提供风险优先级，标准库提供可执行的检测阈值与验证任务。") - cards = effective_report_cards(result) - if cards: - heading("八、追加卡片分析") - for index, card in enumerate(cards, 1): - document.add_paragraph(f"{index}. {card.title}") - add_table([["字段", "内容"]] + [[label, value] for label, value in report_card_lines(card)]) output = BytesIO() document.save(output) return output.getvalue()