统一风险报告卡片与导出数据逻辑

app/schemas/risk_prediction.py

@@ -1,5 +1,7 @@
 """Risk prediction request and response contracts."""
 
+from typing import Literal
+
 from pydantic import Field
 
 from app.domain.risk_engine import RiskLevel
@@ -103,6 +105,44 @@ class AiSolutionReport(ApiSchema):
     warning: str | None = None
 
 
+class ReportCard(ApiSchema):
+    """A response card explicitly appended to the risk report.
+
+    The card is intentionally a superset of the five assistant card types so
+    that the report page and both export formats can consume the exact same
+    payload without type-specific ad-hoc transformations.
+    """
+
+    type: Literal["supplier", "material", "standard", "compliance", "solution"]
+    title: str = "风险分析卡片"
+    name: str | None = None
+    party_type: str | None = None
+    material: str | None = None
+    category: str | None = None
+    question: str | None = None
+    anomaly_count: int | None = None
+    cooperation_count: int | None = None
+    frequency_coefficient: float | None = None
+    supplier_base_score: float | None = None
+    supplier_score: float | None = None
+    tier: str | None = None
+    matched_count: int | None = None
+    detected_values: list[dict[str, str]] = Field(default_factory=list)
+    top_problems: list[dict[str, int | str]] = Field(default_factory=list)
+    frequent_problems: list[dict[str, int | str]] = Field(default_factory=list)
+    standards: list[StandardJudgment] = Field(default_factory=list)
+    similar_cases: list[SimilarCase] = Field(default_factory=list)
+    confidence: str | None = None
+    analysis_base: str | None = None
+    standard_fallback_enabled: bool = False
+    risks: list[AiSolutionRisk] = Field(default_factory=list)
+    suggestions: list[str] = Field(default_factory=list)
+    note: str | None = None
+    warning: str | None = None
+    frequency_basis: str | None = None
+    data_available: bool | None = None
+
+
 class RiskAssessmentResult(ApiSchema):
     subject_id: str
     score: float
@@ -118,6 +158,7 @@ class RiskAssessmentResult(ApiSchema):
     similar_cases: list[SimilarCase] = Field(default_factory=list)
     ai_suggestions: list[str] = Field(default_factory=list)
     ai_solution_cards: list[AiSolutionReport] = Field(default_factory=list)
+    report_cards: list[ReportCard] = Field(default_factory=list)
 
 
 class ModelInfo(ApiSchema):

app/services/conversations.py

@@ -345,6 +345,14 @@ class ConversationService:
 
         if cls._is_placeholder_supplier_name(text):
             return "general"
+        if text.startswith("查物料风险"):
+            return "material"
+        if text.startswith("搜索相似案例"):
+            return "risk_assessment"
+        if text.startswith("生成 AI 案例推理方案"):
+            return "solution"
+        if text.startswith("评估新品"):
+            return "risk_assessment"
         intent = str(semantics.get("intent") or "").strip().lower()
         if intent == "supplier" or (
             any(word in text for word in ("供应商", "加工厂", "工厂"))
@@ -354,6 +362,8 @@ class ConversationService:
             return "supplier"
         if intent == "material" or "物料" in text or "类物料" in text:
             return "material"
+        if intent == "cases":
+            return "risk_assessment"
         if intent == "standard" or any(word in text for word in ("检测报告", "检测值", "判定", "pH", "PH")):
             return "standard"
         if intent == "compliance" or any(word in text for word in ("合规", "宣称", "备案", "法规")):
@@ -667,6 +677,8 @@ class ConversationService:
 
     async def _material_card(self, text: str, semantics: dict[str, Any]) -> dict[str, Any]:
         material = self._first_match(text, ("软管", "泵头", "瓶体", "瓶子", "彩盒", "铝瓶", "喷头", "PET"))
+        if not material and "查物料风险：" in text:
+            material = text.split("查物料风险：", 1)[1].strip().split("，", 1)[0]
         material = material or self._semantic_text(semantics, "packaging") or "未指定物料"
         query_term = "软管" if "软管" in material else material
         count = await self._count_anomalies(AnomalyCase.material_type, query_term)

app/services/reports.py

@@ -17,7 +17,7 @@ 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
+from app.schemas.risk_prediction import ReportCard, RiskAssessmentResult
 
 LEVEL_LABELS = {
     "low": "低风险",
@@ -28,6 +28,121 @@ LEVEL_LABELS = {
 }
 
 
+def effective_report_cards(result: RiskAssessmentResult) -> list[ReportCard]:
+    """Return the single canonical collection used by every report renderer.
+
+    ``ai_solution_cards`` predates the generic card report feature.  It is
+    accepted as a read-only compatibility input when a client sends an old
+    report payload, while new payloads always use ``report_cards``.
+    """
+
+    if result.report_cards:
+        return list(result.report_cards)
+    return [
+        ReportCard(
+            type="solution",
+            title=item.title,
+            confidence=item.confidence,
+            analysis_base=item.analysis_base,
+            standard_fallback_enabled=item.standard_fallback_enabled,
+            standards=item.standards,
+            similar_cases=item.similar_cases,
+            risks=item.risks,
+            suggestions=item.suggestions,
+            warning=item.warning,
+        )
+        for item in result.ai_solution_cards
+    ]
+
+
+def report_card_lines(card: ReportCard) -> list[tuple[str, str]]:
+    """Flatten a card into ordered, renderer-neutral report lines.
+
+    PDF, Word and the web page all consume this same field contract.  Keeping
+    formatting here prevents one export from silently dropping appended card
+    details that are visible in another format.
+    """
+
+    lines: list[tuple[str, str]] = [("卡片类型", card.type)]
+    subject = card.name or card.material or card.category
+    if subject:
+        lines.append(("分析对象", subject))
+    if card.party_type:
+        lines.append(("对象类型", card.party_type))
+    if card.question:
+        lines.append(("原始问题", card.question))
+    if card.anomaly_count is not None:
+        lines.append(("历史异常", f"{card.anomaly_count} 条"))
+    if card.cooperation_count is not None:
+        lines.append(("合作次数", f"{card.cooperation_count} 次"))
+    if card.frequency_coefficient is not None:
+        lines.append(("频次系数", f"×{card.frequency_coefficient:.2f}"))
+    if card.supplier_base_score is not None:
+        lines.append(("供应商基础分", f"{card.supplier_base_score:.2f} 分"))
+    if card.supplier_score is not None:
+        lines.append(("供应商得分", f"{card.supplier_score:.2f} 分"))
+    if card.tier:
+        lines.append(("供应商层级", card.tier))
+    if card.matched_count is not None:
+        lines.append(("匹配数量", f"{card.matched_count} 条"))
+    if card.frequency_basis:
+        lines.append(("数据口径", card.frequency_basis))
+    if card.data_available is not None:
+        lines.append(("数据状态", "有匹配数据" if card.data_available else "暂无匹配数据"))
+    if card.note:
+        lines.append(("备注", card.note))
+    if card.detected_values:
+        values = "、".join(f"{item.get('name', '')}：{item.get('value', '')}" for item in card.detected_values)
+        lines.append(("检测值", values))
+    if card.confidence:
+        lines.append(("置信度", card.confidence))
+    if card.analysis_base:
+        lines.append(("分析基础", card.analysis_base))
+    if card.standard_fallback_enabled:
+        lines.append(("标准兜底校验", "已启用（无直接历史案例）"))
+    if card.top_problems:
+        lines.append(
+            ("高频问题", "、".join(f"{item.get('name', '')}（{item.get('count', 0)}次）" for item in card.top_problems))
+        )
+    if card.frequent_problems:
+        lines.append(
+            (
+                "常见问题",
+                "、".join(f"{item.get('name', '')}（{item.get('count', 0)}次）" for item in card.frequent_problems),
+            )
+        )
+    if card.standards:
+        lines.append(
+            (
+                "关联标准",
+                "；".join(
+                    f"{item.category} · {item.risk_point}：{item.rule}"
+                    f"（{item.source or '标准文件.xlsx'}）· {item.status}"
+                    for item in card.standards
+                ),
+            )
+        )
+    if card.similar_cases:
+        lines.append(
+            (
+                "相似案例",
+                "；".join(
+                    f"{item.date or '日期未知'} · {item.product or '未命名产品'} · "
+                    f"{item.problem or item.description or '异常问题未填写'} → {item.solution or '解决方案未填写'}"
+                    + (f"（相似度 {round(item.similarity * 100)}%）" if item.similarity is not None else "")
+                    for item in card.similar_cases
+                ),
+            )
+        )
+    if card.risks:
+        lines.append(("风险项", "；".join(f"{item.level} · {item.title}：{item.detail}" for item in card.risks)))
+    if card.suggestions:
+        lines.append(("建议方案", "；".join(card.suggestions)))
+    if card.warning:
+        lines.append(("风险提示", card.warning))
+    return lines
+
+
 def _font_name() -> str:
     candidates = [
         Path("C:/Windows/Fonts/msyh.ttc"),
@@ -210,6 +325,18 @@ 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,
@@ -372,6 +499,13 @@ def build_risk_report_docx(result: RiskAssessmentResult) -> bytes:
     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()

tests/test_conversations.py

@@ -104,6 +104,19 @@ def test_only_explicit_risk_requests_generate_risk_cards() -> None:
     ) is True
 
 
+@pytest.mark.parametrize(
+    ("text", "expected"),
+    [
+        ("评估新品：洗发水，包装材质：泵头", "risk_assessment"),
+        ("查物料风险：软管", "material"),
+        ("搜索相似案例：软管印刷遇水脱落", "risk_assessment"),
+        ("生成 AI 案例推理方案：铝瓶+喷头密封验证", "solution"),
+    ],
+)
+def test_tool_requests_keep_their_intent_for_card_routing(text: str, expected: str) -> None:
+    assert ConversationService._response_type(text, {"intent": "general"}) == expected
+
+
 def test_legacy_general_semantics_hide_old_risk_card() -> None:
     messages = [
         ConversationMessage(

tests/test_reports.py

@@ -0,0 +1,62 @@
+from io import BytesIO
+
+from app.schemas.risk_prediction import AiSolutionReport, ReportCard, RiskAssessmentResult
+from app.services.reports import (
+    build_risk_report_docx,
+    build_risk_report_pdf,
+    effective_report_cards,
+    report_card_lines,
+)
+from docx import Document
+
+
+def _result() -> RiskAssessmentResult:
+    return RiskAssessmentResult(
+        subject_id="洗发水新品",
+        score=72,
+        level="medium",
+        model_version="test",
+        factors=[],
+        report_cards=[
+            ReportCard(
+                type="supplier",
+                title="供应商深度画像",
+                name="佳供应商",
+                party_type="供应商",
+                supplier_score=88.5,
+                frequency_coefficient=1.2,
+                top_problems=[{"name": "泄漏", "count": 3}],
+                suggestions=["安排首单全检"],
+            )
+        ],
+    )
+
+
+def test_report_card_lines_are_canonical() -> None:
+    lines = dict(report_card_lines(_result().report_cards[0]))
+    assert lines["分析对象"] == "佳供应商"
+    assert lines["供应商得分"] == "88.50 分"
+    assert lines["高频问题"] == "泄漏（3次）"
+    assert lines["建议方案"] == "安排首单全检"
+
+
+def test_old_ai_solution_payload_is_supported() -> None:
+    result = _result().model_copy(
+        update={"report_cards": [], "ai_solution_cards": [AiSolutionReport(title="旧方案")]}
+    )
+    cards = effective_report_cards(result)
+    assert len(cards) == 1
+    assert cards[0].title == "旧方案"
+
+
+def test_pdf_and_word_include_appended_card_content() -> None:
+    result = _result()
+    pdf = build_risk_report_pdf(result)
+    word = build_risk_report_docx(result)
+    assert pdf.startswith(b"%PDF")
+    document = Document(BytesIO(word))
+    text = "\n".join(paragraph.text for paragraph in document.paragraphs)
+    text += "\n" + "\n".join(cell.text for table in document.tables for row in table.rows for cell in row.cells)
+    assert "佳供应商" in text
+    assert "88.50 分" in text
+    assert "安排首单全检" in text