app/api/router.py

@@ -9,6 +9,7 @@ from app.api.v1.health import router as health_router
 from app.api.v1.risk_items import router as risk_items_router
 from app.api.v1.risk_predictions import router as risk_predictions_router
 from app.api.v1.standards import router as standards_router
+from app.api.v1.suppliers import router as suppliers_router
 
 api_router = APIRouter(prefix="/api/v1")
 api_router.include_router(health_router)
@@ -18,3 +19,4 @@ api_router.include_router(dashboard_router)
 api_router.include_router(risk_items_router)
 api_router.include_router(risk_predictions_router)
 api_router.include_router(standards_router)
+api_router.include_router(suppliers_router)

app/api/v1/suppliers.py

@@ -0,0 +1,32 @@
+"""Supplier profile endpoints."""
+
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Query
+
+from app.api.dependencies import DatabaseSession, require_permission
+from app.models.entities import UserAccount
+from app.schemas.common import ApiResponse
+from app.schemas.suppliers import SupplierListData, SupplierProfileData
+from app.services.suppliers import SupplierProfileService
+
+router = APIRouter(prefix="/suppliers", tags=["suppliers"])
+
+
+@router.get("", response_model=ApiResponse[SupplierListData])
+async def list_suppliers(
+    session: DatabaseSession,
+    _current_user: Annotated[UserAccount, Depends(require_permission("supplier.read"))],
+    keyword: str | None = Query(default=None, max_length=100),
+    limit: int = Query(default=100, ge=1, le=100),
+) -> ApiResponse[SupplierListData]:
+    return ApiResponse(data=await SupplierProfileService(session).list(keyword=keyword, limit=limit))
+
+
+@router.get("/{supplier_name}", response_model=ApiResponse[SupplierProfileData])
+async def get_supplier_profile(
+    supplier_name: str,
+    session: DatabaseSession,
+    _current_user: Annotated[UserAccount, Depends(require_permission("supplier.read"))],
+) -> ApiResponse[SupplierProfileData]:
+    return ApiResponse(data=await SupplierProfileService(session).detail(supplier_name))

app/schemas/dashboard.py

@@ -32,9 +32,9 @@ class DashboardMatrixRow(ApiSchema):
     name: str
     party_type: str
     categories: dict[str, int]
-    cooperation_total: int
+    cooperation_total: int | None
     anomaly_count: int
-    anomaly_rate: float
+    anomaly_rate: float | None
     tier: str
 
 

app/schemas/risk_prediction.py

@@ -43,9 +43,12 @@ class SupplierFrequency(ApiSchema):
     supplier_name: str | None = None
     cooperation_count: int | None = None
     anomaly_count: int | None = None
-    alpha: float = 1.0
+    alpha: float | None = None
     tier: str | None = None
     suggestion: str | None = None
+    frequency_basis: str = "近3年异常记录数"
+    note: str | None = None
+    data_available: bool = False
 
 
 class StandardJudgment(ApiSchema):

app/schemas/suppliers.py

@@ -0,0 +1,64 @@
+"""Supplier profile API contracts."""
+
+from pydantic import Field
+
+from app.schemas.common import ApiSchema
+
+
+class SupplierOption(ApiSchema):
+    name: str
+    anomaly_count: int
+    tier: str
+
+
+class SupplierListData(ApiSchema):
+    items: list[SupplierOption] = Field(default_factory=list)
+    default_supplier: str | None = None
+
+
+class SupplierMetric(ApiSchema):
+    cooperation_total: int | None = None
+    anomaly_total: int
+    anomaly_rate: float | None = None
+    frequency_alpha: float
+    risk_score: float | None = None
+    main_category_count: int
+    cooperation_count_basis: str = "暂无采购订单台账；合作次数不可由异常记录推算"
+    anomaly_rate_basis: str = "暂无采购订单/生产批次总量，无法计算真实异常率"
+
+
+class SupplierCategoryBreakdown(ApiSchema):
+    category: str
+    anomaly_count: int
+    anomaly_rate: float | None = None
+    top_material: str | None = None
+    top_problem: str | None = None
+    anomaly_rate_basis: str = "暂无供应商品类合作总量，无法计算真实异常率"
+
+
+class SupplierRecentRecord(ApiSchema):
+    case_id: int
+    month: str | None = None
+    product: str | None = None
+    anomaly: str | None = None
+    defect_level: str | None = None
+
+
+class FactoryRecommendation(ApiSchema):
+    category: str
+    preferred_factory: str | None = None
+    alternative_factory: str | None = None
+    reason: str
+    recommendation_basis: str = "暂无加工厂生产总量，无法计算真实异常率"
+
+
+class SupplierProfileData(ApiSchema):
+    supplier_name: str
+    tier: str
+    tier_condition: str
+    metric: SupplierMetric
+    category_breakdown: list[SupplierCategoryBreakdown] = Field(default_factory=list)
+    monthly_trend: list[dict[str, int | str]] = Field(default_factory=list)
+    recent_records: list[SupplierRecentRecord] = Field(default_factory=list)
+    factory_recommendations: list[FactoryRecommendation] = Field(default_factory=list)
+    recommendation_note: str | None = None

app/services/conversations.py

@@ -4,7 +4,7 @@ import json
 import logging
 import re
 from collections.abc import AsyncGenerator, Sequence
-from datetime import UTC, datetime
+from datetime import UTC, date, datetime, timedelta
 from typing import Any, Literal
 
 from sqlalchemy import func, or_, select
@@ -23,6 +23,7 @@ from app.schemas.risk_prediction import (
     RiskAssessmentResult,
     RiskDimensionScore,
     RiskFactorResult,
+    SimilarCase,
     StandardJudgment,
     SupplierFrequency,
 )
@@ -34,6 +35,7 @@ logger = logging.getLogger("dieya_risk_prediction.conversations")
 
 
 class ConversationService:
+    _supplier_history_days = 365 * 3
     _engine = BaselineRiskEngine()
     _risk_intents = {
         "assessment",
@@ -232,37 +234,53 @@ class ConversationService:
             yield {"type": "error", "message": "LLM 未返回有效内容"}
             return
 
-        show_risk = self._is_risk_request(text, semantics)
-        response_type = self._response_type(text, semantics)
-        response_type = await self._use_solution_card_without_history(response_type, text, semantics)
-        risk_result = (
-            await self._enrich_risk_result(text, semantics, self._predict(text))
-            if show_risk and response_type == "risk_assessment"
-            else None
-        )
-        response_card = (
-            await self._build_response_card(response_type, text, semantics)
-            if response_type != "general"
-            else None
-        )
+        # Persist the complete assistant text before optional card enrichment.
+        # A slow or failed RAG/score lookup must not lose the conversation turn.
         assistant_message = ConversationMessage(
             conversation_id=conversation.id,
             role="assistant",
             message_type="text",
             content=reply,
-            structured_data={
-                "semantic": semantics,
-                **(
-                    {"riskResult": risk_result.model_dump(mode="json", by_alias=False)}
-                    if risk_result
-                    else {}
-                ),
-                **({"responseCard": response_card} if response_card else {}),
-            },
+            structured_data={"semantic": semantics},
         )
         self.session.add(assistant_message)
         await self.session.flush()
         await self.session.commit()
+
+        show_risk = self._is_risk_request(text, semantics)
+        response_type = self._response_type(text, semantics)
+        risk_result = None
+        response_card = None
+        try:
+            response_type = await self._use_solution_card_without_history(response_type, text, semantics)
+            risk_result = (
+                await self._enrich_risk_result(text, semantics, self._predict(text))
+                if show_risk and response_type == "risk_assessment"
+                else None
+            )
+            response_card = (
+                await self._build_response_card(response_type, text, semantics)
+                if response_type != "general"
+                else None
+            )
+        except Exception as exc:  # noqa: BLE001 - text reply must remain persisted
+            logger.exception(
+                "Conversation enrichment failed after assistant text was persisted error=%s",
+                exc.__class__.__name__,
+            )
+            await self.session.rollback()
+
+        assistant_message.structured_data = {
+            "semantic": semantics,
+            **(
+                {"riskResult": risk_result.model_dump(mode="json", by_alias=False)}
+                if risk_result
+                else {}
+            ),
+            **({"responseCard": response_card} if response_card else {}),
+        }
+        await self.session.flush()
+        await self.session.commit()
         persisted = await self.repository.get_for_user(conversation.id, user.id)
         if persisted is None:  # pragma: no cover - the row already existed
             yield {"type": "error", "message": "对话读取失败"}
@@ -537,25 +555,28 @@ class ConversationService:
             name = match.group(1).strip() if match else "未指定"
         is_factory = any(word in text for word in ("加工厂", "工厂", "生产厂")) and "供应商" not in text
         column = AnomalyCase.factory if is_factory else AnomalyCase.supplier
-        anomaly_count = await self._count_anomalies(column, name)
-        problems = await self._top_anomaly_types(column, name)
+        anomaly_count, a_count, b_count = await self._supplier_history_stats(column, name)
+        problems = await self._top_anomaly_types(
+            column,
+            name,
+            since=self._supplier_history_start(),
+        )
         party_type = "加工厂" if is_factory else "供应商"
+        tier = self._supplier_tier(anomaly_count, a_count=a_count, b_count=b_count)
         return {
             "type": "supplier",
             "title": "供应商深度画像",
             "name": name,
             "partyType": party_type,
             "anomalyCount": anomaly_count,
-            "cooperationCount": None,
-            "tier": self._supplier_tier(anomaly_count),
+            "cooperationCount": anomaly_count,
+            "tier": tier,
             "frequencyCoefficient": round(min(1.5, 1 + anomaly_count / 100 * 0.3), 2) if anomaly_count else 1.0,
             "topProblems": problems,
-            "suggestions": (
-                ["加强巡检并准备备选供应商", "重点复核高频异常对应的来料和过程检验项目"]
-                if anomaly_count
-                else ["暂无历史合作数据，建议首单全检并安排驻厂巡检"]
-            ),
-            "note": "合作次数需要采购合作台账接入后统计，当前仅展示异常案例库中的真实记录。",
+            "suggestions": self._supplier_management_suggestions(tier),
+            "frequencyBasis": "近3年异常记录数",
+            "note": "合作次数为近3年异常记录数代理值，不代表真实采购订单合作次数。",
+            "dataAvailable": True,
         }
 
     async def _material_card(self, text: str, semantics: dict[str, Any]) -> dict[str, Any]:
@@ -801,12 +822,27 @@ class ConversationService:
             for row in rows
         ]
 
-    async def _top_anomaly_types(self, column, value: str | None) -> Sequence[dict[str, Any]]:  # noqa: ANN001
+    async def _top_anomaly_types(
+        self,
+        column,  # noqa: ANN001
+        value: str | None,
+        *,
+        since: date | None = None,
+    ) -> Sequence[dict[str, Any]]:
         if not value:
             return []
+        filters = [column.ilike(f"%{value}%"), AnomalyCase.anomaly_type.is_not(None)]
+        if since:
+            filters.extend(
+                [
+                    AnomalyCase.date.is_not(None),
+                    AnomalyCase.date >= since,
+                    AnomalyCase.date <= datetime.now(UTC).date(),
+                ]
+            )
         rows = await self.session.execute(
             select(AnomalyCase.anomaly_type, func.count())
-            .where(column.ilike(f"%{value}%"), AnomalyCase.anomaly_type.is_not(None))
+            .where(*filters)
             .group_by(AnomalyCase.anomaly_type)
             .order_by(func.count().desc())
             .limit(5)
@@ -917,7 +953,10 @@ class ConversationService:
             defect_level=self._first_match(text, ("A级", "B级", "C级")),
         )
         category_count = await self._count_category_anomalies(category)
-        supplier_count = await self._count_anomalies(AnomalyCase.supplier, supplier)
+        supplier_count, supplier_a_count, supplier_b_count = await self._supplier_history_stats(
+            AnomalyCase.supplier,
+            supplier,
+        )
         factory_count = await self._count_anomalies(AnomalyCase.factory, factory)
         anomaly_count = supplier_count if supplier else factory_count
         standard_rows = await self._standards_for(category, text)
@@ -936,7 +975,15 @@ class ConversationService:
 
         alpha_count = supplier_count if supplier else 0
         alpha = round(min(1.5, 1 + (alpha_count / 100) * 0.3), 2) if alpha_count else 1.0
-        tier = self._supplier_tier(alpha_count) if supplier else None
+        tier = (
+            self._supplier_tier(
+                alpha_count,
+                a_count=supplier_a_count,
+                b_count=supplier_b_count,
+            )
+            if supplier
+            else None
+        )
         level_text = self._risk_level_text(database_score.level)
         alert_text = self._alert_text(
             category,
@@ -998,21 +1045,30 @@ class ConversationService:
                         else "建议结合当前品类高频异常制定验证计划"
                     ),
                 ),
-                "supplier_freq": SupplierFrequency(
-                    supplier_name=supplier_or_factory,
-                    cooperation_count=None,
-                    anomaly_count=anomaly_count or None,
-                    alpha=alpha,
-                    tier=tier,
-                    suggestion=(
-                        "无历史数据，建议首单全检并安排驻厂巡检"
-                        if not anomaly_count
-                        else "建议加强巡检并准备备选供应商"
-                    ),
+                "supplier_freq": (
+                    SupplierFrequency(
+                        supplier_name=supplier,
+                        cooperation_count=supplier_count,
+                        anomaly_count=supplier_count,
+                        alpha=alpha,
+                        tier=tier,
+                        suggestion="；".join(self._supplier_management_suggestions(tier)),
+                        frequency_basis="近3年异常记录数",
+                        note="合作次数为近3年异常记录数代理值，不代表真实采购订单合作次数。",
+                        data_available=True,
+                    )
+                    if supplier
+                    else SupplierFrequency(
+                        note="未指定供应商，暂无供应商频次分析。",
+                        frequency_basis="未提供供应商",
+                    )
                 ),
                 "standard_judgment": standard_rows,
                 "dimension_scores": dimensions,
-                "similar_cases": [item.case.model_dump(mode="json", by_alias=False) for item in similar_cases],
+                "similar_cases": [
+                    SimilarCase.model_validate(item.case.model_dump(mode="json", by_alias=False))
+                    for item in similar_cases
+                ],
                 "ai_suggestions": suggestions,
             }
         )
@@ -1021,10 +1077,47 @@ class ConversationService:
         if not value or value == "未归类":
             return 0
         count = await self.session.scalar(
-            select(func.count()).select_from(AnomalyCase).where(column.ilike(f"%{value}%"))
+            select(func.count())
+            .select_from(AnomalyCase)
+            .where(
+                column.ilike(f"%{value}%"),
+                AnomalyCase.date.is_not(None),
+                AnomalyCase.date >= self._supplier_history_start(),
+                AnomalyCase.date <= datetime.now(UTC).date(),
+            )
         )
         return int(count or 0)
 
+    @classmethod
+    def _supplier_history_start(cls) -> date:
+        return datetime.now(UTC).date() - timedelta(days=cls._supplier_history_days)
+
+    async def _supplier_history_stats(
+        self,
+        column,  # noqa: ANN001
+        value: str | None,
+    ) -> tuple[int, int, int]:
+        """Return the PRD's three-year N, A-count, and B-count inputs."""
+
+        if not value or value == "未归类":
+            return 0, 0, 0
+        rows = (
+            await self.session.execute(
+                select(AnomalyCase.defect_level, func.count())
+                .where(
+                    column.ilike(f"%{value}%"),
+                    AnomalyCase.date.is_not(None),
+                    AnomalyCase.date >= self._supplier_history_start(),
+                    AnomalyCase.date <= datetime.now(UTC).date(),
+                )
+                .group_by(AnomalyCase.defect_level)
+            )
+        ).all()
+        total = sum(int(count or 0) for _, count in rows)
+        a_count = sum(int(count or 0) for level, count in rows if self._defect_code(level) == "A")
+        b_count = sum(int(count or 0) for level, count in rows if self._defect_code(level) == "B")
+        return total, a_count, b_count
+
     async def _generate_ai_suggestions(
         self,
         *,
@@ -1132,7 +1225,10 @@ class ConversationService:
                 or_(
                     AnomalyCase.product_category.ilike(f"%{category}%"),
                     AnomalyCase.product_name.ilike(f"%{category}%"),
-                )
+                ),
+                AnomalyCase.date.is_not(None),
+                AnomalyCase.date >= self._supplier_history_start(),
+                AnomalyCase.date <= datetime.now(UTC).date(),
             )
         )
         return int(count or 0)
@@ -1213,17 +1309,40 @@ class ConversationService:
         }[level]
 
     @staticmethod
-    def _supplier_tier(anomaly_count: int) -> str | None:
-        if not anomaly_count:
+    def _defect_code(value: str | None) -> str | None:
+        match = re.search(r"[ABCD]", (value or "").upper())
+        return match.group(0) if match else None
+
+    @staticmethod
+    def _supplier_tier(
+        anomaly_count: int,
+        *,
+        a_count: int = 0,
+        b_count: int = 0,
+    ) -> str:
+        if anomaly_count == 0:
             return "新供应商"
-        if anomaly_count < 20:
-            return "S级 · 战略"
-        if anomaly_count <= 60:
-            return "A级 · 优质"
-        if anomaly_count <= 120:
+        ab_ratio = (a_count + b_count) / anomaly_count
+        if anomaly_count > 120 or a_count > 0 or ab_ratio > 0.5:
+            return "C级 · 高风险"
+        if anomaly_count > 60 or ab_ratio >= 0.3:
             return "B级 · 关注"
+        if anomaly_count >= 20:
+            return "A级 · 优质"
+        if a_count == 0:
+            return "S级 · 战略"
         return "C级 · 高风险"
 
+    @staticmethod
+    def _supplier_management_suggestions(tier: str | None) -> Sequence[str]:
+        return {
+            "S级 · 战略": ["正常合作", "季度回顾"],
+            "A级 · 优质": ["常规管理", "年度评审"],
+            "B级 · 关注": ["加强巡检", "新品打样加严"],
+            "C级 · 高风险": ["安排驻厂巡检", "储备备选供应商", "限制新品使用"],
+            "新供应商": ["无历史数据", "首单全检", "安排驻厂巡检"],
+        }.get(tier, [])
+
     @staticmethod
     def _alert_text(
         category: str,

app/services/dashboard.py

@@ -286,8 +286,8 @@ class DashboardService:
         result = []
         for (name, party_type), item in sorted(actors.items(), key=lambda entry: -entry[1]["count"]):
             anomaly_count = item["count"]
-            # Keep the existing cooperation-total calculation unchanged.
-            cooperation_total = item["count"]
+            # No order or production denominator exists in anomaly_cases.
+            cooperation_total = None
             values = {category: 0 for category in categories}
             for row in item["items"]:
                 category = cls._category(row["product_name"], row["material_type"])
@@ -300,11 +300,7 @@ class DashboardService:
                     categories=values,
                     cooperation_total=cooperation_total,
                     anomaly_count=anomaly_count,
-                    anomaly_rate=(
-                        round(anomaly_count / cooperation_total * 100, 1)
-                        if cooperation_total
-                        else 0
-                    ),
+                    anomaly_rate=None,
                     tier=item["tier"],
                 )
             )

app/services/suppliers.py

@@ -0,0 +1,208 @@
+"""Live supplier profile aggregation from the approved anomaly case data."""
+
+from __future__ import annotations
+
+from collections import Counter, defaultdict
+from datetime import UTC, date, datetime, timedelta
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.exceptions import AppException
+from app.models.entities import AnomalyCase
+from app.schemas.suppliers import (
+    SupplierCategoryBreakdown,
+    SupplierListData,
+    SupplierMetric,
+    SupplierOption,
+    SupplierProfileData,
+    SupplierRecentRecord,
+)
+from app.services.database_risk_scoring import DatabaseRiskScoringService, normalize_category
+
+
+class SupplierProfileService:
+    history_days = 365 * 3
+    recent_months = 6
+    max_categories = 5
+    max_recent_records = 10
+
+    def __init__(self, session: AsyncSession, *, as_of: date | None = None) -> None:
+        self.session = session
+        self.as_of = as_of or datetime.now(UTC).date()
+
+    async def list(self, *, keyword: str | None = None, limit: int = 100) -> SupplierListData:
+        rows = await self._load_rows()
+        grouped = self._group_suppliers(rows, keyword)
+        items = [
+            SupplierOption(
+                name=name,
+                anomaly_count=len(items),
+                tier=self.tier_for(
+                    len(items),
+                    sum(self.defect_code(row.get("defect_level")) == "A" for row in items),
+                    sum(self.defect_code(row.get("defect_level")) == "B" for row in items),
+                ),
+            )
+            for name, items in sorted(grouped.items(), key=lambda item: (-len(item[1]), item[0]))[:limit]
+        ]
+        return SupplierListData(
+            items=items,
+            default_supplier=items[0].name if items else None,
+        )
+
+    async def detail(self, supplier_name: str) -> SupplierProfileData:
+        rows = [row for row in await self._load_rows() if self.matches(row.get("supplier"), supplier_name)]
+        if not rows:
+            raise AppException("供应商不存在或暂无近3年异常记录", code=404, status_code=404)
+
+        anomaly_total = len(rows)
+        a_count = sum(self.defect_code(row.get("defect_level")) == "A" for row in rows)
+        b_count = sum(self.defect_code(row.get("defect_level")) == "B" for row in rows)
+        tier = self.tier_for(anomaly_total, a_count, b_count)
+        categories = self._categories(rows)
+        score = await DatabaseRiskScoringService(self.session, as_of=self.as_of).score(supplier=supplier_name)
+        return SupplierProfileData(
+            supplier_name=supplier_name,
+            tier=tier,
+            tier_condition=self.tier_condition(tier),
+            metric=SupplierMetric(
+                cooperation_total=None,
+                anomaly_total=anomaly_total,
+                anomaly_rate=None,
+                frequency_alpha=self.alpha(anomaly_total),
+                risk_score=score.score,
+                main_category_count=len(categories),
+            ),
+            category_breakdown=categories[: self.max_categories],
+            monthly_trend=self._monthly_trend(rows),
+            recent_records=self._recent_records(rows),
+            factory_recommendations=[],
+            recommendation_note=(
+                "合作次数、供应商真实异常率和代工厂真实异常率需要采购订单或生产批次总量；"
+                "当前已接入近3年异常库聚合，但不会用异常记录数伪造上述指标。"
+            ),
+        )
+
+    async def _load_rows(self) -> list[dict]:
+        start = self.as_of - timedelta(days=self.history_days)
+        statement = select(
+            AnomalyCase.id,
+            AnomalyCase.date,
+            AnomalyCase.product_name,
+            AnomalyCase.product_category,
+            AnomalyCase.material_type,
+            AnomalyCase.supplier,
+            AnomalyCase.factory,
+            AnomalyCase.anomaly_type,
+            AnomalyCase.defect_level,
+        ).where(
+            AnomalyCase.date.is_not(None),
+            AnomalyCase.date >= start,
+            AnomalyCase.date <= self.as_of,
+        )
+        return [dict(row) for row in (await self.session.execute(statement)).mappings().all()]
+
+    @staticmethod
+    def matches(value: str | None, target: str) -> bool:
+        return bool(value and target.strip().lower() in value.lower())
+
+    @classmethod
+    def _group_suppliers(cls, rows: list[dict], keyword: str | None) -> dict[str, list[dict]]:
+        grouped: dict[str, list[dict]] = defaultdict(list)
+        for row in rows:
+            name = (row.get("supplier") or "").strip()
+            if name and (not keyword or cls.matches(name, keyword)):
+                grouped[name].append(row)
+        return grouped
+
+    @staticmethod
+    def defect_code(value: str | None) -> str | None:
+        value = (value or "").strip().upper()
+        return value if value in {"A", "B", "C"} else None
+
+    @staticmethod
+    def alpha(anomaly_count: int) -> float:
+        return round(min(1.5, 1 + anomaly_count / 100 * 0.3), 2)
+
+    @classmethod
+    def tier_for(cls, anomaly_count: int, a_count: int, b_count: int) -> str:
+        if anomaly_count == 0:
+            return "新供应商"
+        ab_ratio = (a_count + b_count) / anomaly_count
+        if anomaly_count > 120 or a_count > 0 or ab_ratio > 0.5:
+            return "C级"
+        if anomaly_count > 60 or ab_ratio >= 0.3:
+            return "B级"
+        if anomaly_count >= 20:
+            return "A级"
+        return "S级"
+
+    @staticmethod
+    def tier_condition(tier: str) -> str:
+        return {
+            "S级": "异常记录 < 20 次，且无 A 级缺陷",
+            "A级": "异常记录 20~60 次，且 A/B 级缺陷占比 < 30%",
+            "B级": "异常记录 60~120 次，或 A/B 级缺陷占比 30%~50%",
+            "C级": "异常记录 > 120 次，或存在 A 级缺陷，或 A/B 级缺陷占比 > 50%",
+            "新供应商": "近3年异常记录为 0",
+        }[tier]
+
+    @classmethod
+    def _categories(cls, rows: list[dict]) -> list[SupplierCategoryBreakdown]:
+        grouped: dict[str, list[dict]] = defaultdict(list)
+        for row in rows:
+            category = normalize_category(row.get("product_category")) or normalize_category(row.get("product_name"))
+            grouped[category or "待补标品类"].append(row)
+        result = []
+        for category, items in grouped.items():
+            materials = Counter(row.get("material_type") for row in items if row.get("material_type"))
+            problems = Counter(row.get("anomaly_type") for row in items if row.get("anomaly_type"))
+            result.append(
+                SupplierCategoryBreakdown(
+                    category=category,
+                    anomaly_count=len(items),
+                    anomaly_rate=None,
+                    top_material=materials.most_common(1)[0][0] if materials else None,
+                    top_problem=problems.most_common(1)[0][0] if problems else None,
+                )
+            )
+        return sorted(result, key=lambda item: (-item.anomaly_count, item.category))
+
+    def _monthly_trend(self, rows: list[dict]) -> list[dict[str, int | str]]:
+        start = self._month_offset(self.as_of.replace(day=1), -5)
+        counts = Counter(
+            row["date"].strftime("%Y-%m")
+            for row in rows
+            if row.get("date") and row["date"] >= start
+        )
+        output = []
+        cursor = start
+        for _ in range(self.recent_months):
+            key = cursor.strftime("%Y-%m")
+            output.append({"month": key, "count": counts.get(key, 0)})
+            cursor = date(cursor.year + (cursor.month == 12), 1 if cursor.month == 12 else cursor.month + 1, 1)
+        return output
+
+    @staticmethod
+    def _month_offset(value: date, months: int) -> date:
+        index = value.year * 12 + value.month - 1 + months
+        return date(index // 12, index % 12 + 1, 1)
+
+    def _recent_records(self, rows: list[dict]) -> list[SupplierRecentRecord]:
+        start = self._month_offset(self.as_of.replace(day=1), -5)
+        recent = sorted(
+            (row for row in rows if row.get("date") and row["date"] >= start),
+            key=lambda row: row["date"],
+            reverse=True,
+        )
+        return [
+            SupplierRecentRecord(
+                case_id=int(row["id"]),
+                month=row["date"].strftime("%Y-%m") if row.get("date") else None,
+                product=row.get("product_name"),
+                anomaly=row.get("anomaly_type"),
+                defect_level=self.defect_code(row.get("defect_level")),
+            )
+            for row in recent[: self.max_recent_records]
+        ]

migrations/versions/20260907_0015_add_supplier_read_permissions.py

@@ -0,0 +1,39 @@
+"""Align full supplier-profile access with the PRD's L2 role matrix."""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "20260907_0015"
+down_revision: str | None = "20260905_0014"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+PERMISSIONS = ("procurement", "reviewer", "manager")
+
+
+def upgrade() -> None:
+    connection = op.get_bind()
+    statement = sa.text(
+        """
+        INSERT INTO role_permissions (role, func_key, data_level, can_view, can_operate)
+        VALUES (:role, 'supplier.read', 'L2', true, false)
+        ON CONFLICT (role, func_key) DO UPDATE SET
+            data_level = EXCLUDED.data_level,
+            can_view = EXCLUDED.can_view,
+            can_operate = EXCLUDED.can_operate,
+            updated_at = now()
+        """
+    )
+    for role in PERMISSIONS:
+        connection.execute(statement, {"role": role})
+
+
+def downgrade() -> None:
+    connection = op.get_bind()
+    for role in PERMISSIONS:
+        connection.execute(
+            sa.text("DELETE FROM role_permissions WHERE role = :role AND func_key = 'supplier.read'"),
+            {"role": role},
+        )

tests/test_conversations.py

@@ -2,6 +2,7 @@ from datetime import UTC, datetime
 
 import pytest
 from app.models.entities import Conversation, ConversationMessage
+from app.schemas.risk_prediction import SupplierFrequency
 from app.services.conversations import ConversationService
 
 
@@ -31,6 +32,32 @@ def test_conversation_message_data_reads_persisted_risk_result() -> None:
     assert result.risk_result.level.value == "low"
 
 
+def test_conversation_message_data_reads_persisted_semantic_and_response_card() -> None:
+    message = ConversationMessage(
+        id=11,
+        conversation_id=3,
+        role="assistant",
+        content="已完成分析。",
+        structured_data={
+            "semantic": {"intent": "solution", "category": "洗发水"},
+            "responseCard": {
+                "type": "solution",
+                "title": "AI 案例推理生成",
+                "similarCases": [{"case_id": 101, "similarity": 0.82}],
+                "suggestions": ["完成包材相容性测试"],
+            },
+        },
+        created_at=datetime.now(UTC),
+    )
+
+    result = ConversationService._message(message)
+
+    assert result.semantic == {"intent": "solution", "category": "洗发水"}
+    assert result.response_card is not None
+    assert result.response_card["similarCases"][0]["case_id"] == 101
+    assert result.response_card["suggestions"] == ["完成包材相容性测试"]
+
+
 def test_conversation_data_includes_all_messages() -> None:
     conversation = Conversation(
         id=3,
@@ -144,6 +171,115 @@ def test_unknown_standalone_name_is_treated_as_new_supplier() -> None:
     assert ConversationService._response_type("新供应商A", {"intent": "supplier"}) == "general"
 
 
+@pytest.mark.parametrize(
+    ("count", "a_count", "b_count", "expected"),
+    [
+        (0, 0, 0, "新供应商"),
+        (12, 0, 1, "S级 · 战略"),
+        (40, 0, 4, "A级 · 优质"),
+        (70, 0, 10, "B级 · 关注"),
+        (40, 0, 16, "B级 · 关注"),
+        (10, 1, 0, "C级 · 高风险"),
+        (121, 0, 0, "C级 · 高风险"),
+    ],
+)
+def test_supplier_tier_uses_prd_frequency_and_defect_rules(
+    count: int,
+    a_count: int,
+    b_count: int,
+    expected: str,
+) -> None:
+    assert ConversationService._supplier_tier(
+        count,
+        a_count=a_count,
+        b_count=b_count,
+    ) == expected
+
+
+def test_supplier_management_suggestions_follow_prd_actions() -> None:
+    assert ConversationService._supplier_management_suggestions("B级 · 关注") == [
+        "加强巡检",
+        "新品打样加严",
+    ]
+    assert ConversationService._supplier_management_suggestions("新供应商") == [
+        "无历史数据",
+        "首单全检",
+        "安排驻厂巡检",
+    ]
+
+
+@pytest.mark.asyncio
+async def test_supplier_card_uses_prd_proxy_frequency_and_tier_advice(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    service = object.__new__(ConversationService)
+
+    async def history_stats(*_args, **_kwargs):
+        return 70, 0, 10
+
+    async def top_anomaly_types(*_args, **_kwargs):
+        return [{"name": "泵头异常", "count": 12}]
+
+    monkeypatch.setattr(service, "_supplier_history_stats", history_stats)
+    monkeypatch.setattr(service, "_top_anomaly_types", top_anomaly_types)
+
+    result = await service._supplier_card(
+        "祥印供应商风险怎么样",
+        {"supplier": "祥印"},
+    )
+
+    assert result["anomalyCount"] == 70
+    assert result["cooperationCount"] == 70
+    assert result["frequencyCoefficient"] == 1.21
+    assert result["tier"] == "B级 · 关注"
+    assert result["suggestions"] == ["加强巡检", "新品打样加严"]
+    assert result["frequencyBasis"] == "近3年异常记录数"
+    assert result["dataAvailable"] is True
+
+
+def test_new_supplier_frequency_serializes_zero_proxy_count() -> None:
+    result = SupplierFrequency(
+        supplier_name="新供应商",
+        cooperation_count=0,
+        anomaly_count=0,
+        tier="新供应商",
+        suggestion="无历史数据；首单全检；安排驻厂巡检",
+        note="合作次数为近3年异常记录数代理值，不代表真实采购订单合作次数。",
+    ).model_dump(by_alias=True)
+
+    assert result["cooperationCount"] == 0
+    assert result["anomalyCount"] == 0
+    assert result["frequencyBasis"] == "近3年异常记录数"
+
+
+def test_unspecified_supplier_returns_displayable_no_data_frequency_card() -> None:
+    result = SupplierFrequency(
+        note="未指定供应商，暂无供应商频次分析。",
+        frequency_basis="未提供供应商",
+    ).model_dump(by_alias=True)
+
+    assert result["dataAvailable"] is False
+    assert result["alpha"] is None
+    assert result["note"] == "未指定供应商，暂无供应商频次分析。"
+
+
+def test_similar_cases_are_typed_before_risk_result_serialization() -> None:
+    from app.schemas.risk_prediction import RiskAssessmentResult, SimilarCase
+
+    result = RiskAssessmentResult(
+        subject_id="test",
+        score=52,
+        level="medium",
+        model_version="test",
+        factors=[],
+        similar_cases=[SimilarCase(case_id=1, similarity=0.75)],
+    )
+
+    payload = result.model_dump(mode="json")
+
+    assert payload["similar_cases"][0]["case_id"] == 1
+
+
 @pytest.mark.asyncio
 async def test_any_product_without_history_uses_solution_card(monkeypatch: pytest.MonkeyPatch) -> None:
     service = object.__new__(ConversationService)

tests/test_risk_prediction.py

@@ -18,6 +18,12 @@ def test_preview_requires_authentication() -> None:
     assert response.status_code == 401
 
 
+def test_supplier_profile_requires_authentication() -> None:
+    response = TestClient(app).get("/api/v1/suppliers")
+
+    assert response.status_code == 401
+
+
 def test_baseline_engine_calculates_weighted_score() -> None:
     score = BaselineRiskEngine().predict(
         [

tests/test_suppliers.py

@@ -0,0 +1,66 @@
+from datetime import date
+
+from app.schemas.suppliers import SupplierListData, SupplierOption
+from app.services.suppliers import SupplierProfileService
+
+
+def test_supplier_tier_rules_follow_prd() -> None:
+    assert SupplierProfileService.tier_for(0, 0, 0) == "新供应商"
+    assert SupplierProfileService.tier_for(12, 0, 0) == "S级"
+    assert SupplierProfileService.tier_for(40, 0, 4) == "A级"
+    assert SupplierProfileService.tier_for(70, 0, 10) == "B级"
+    assert SupplierProfileService.tier_for(40, 0, 16) == "B级"
+    assert SupplierProfileService.tier_for(10, 1, 0) == "C级"
+    assert SupplierProfileService.tier_for(121, 0, 0) == "C级"
+
+
+def test_supplier_alpha_is_capped_and_rounded() -> None:
+    assert SupplierProfileService.alpha(0) == 1.0
+    assert SupplierProfileService.alpha(42) == 1.13
+    assert SupplierProfileService.alpha(148) == 1.44
+    assert SupplierProfileService.alpha(200) == 1.5
+
+
+def test_supplier_list_exposes_default_highest_anomaly_supplier() -> None:
+    result = SupplierListData(
+        items=[
+            SupplierOption(name="甲供应商", anomaly_count=2, tier="S级"),
+            SupplierOption(name="乙供应商", anomaly_count=1, tier="S级"),
+        ],
+        default_supplier="甲供应商",
+    )
+
+    assert result.default_supplier == result.items[0].name
+
+
+def test_supplier_category_breakdown_is_top_five_with_material_and_problem() -> None:
+    rows = [
+        {"product_category": "洗发水", "product_name": "洗发水", "material_type": "泵头", "anomaly_type": "漏液"},
+        {"product_category": "洗发水", "product_name": "洗发水", "material_type": "泵头", "anomaly_type": "漏液"},
+        {"product_category": "面霜", "product_name": "面霜", "material_type": "瓶子", "anomaly_type": "刮花"},
+    ]
+
+    result = SupplierProfileService._categories(rows)
+
+    assert result[0].category == "洗发水/洗护类"
+    assert result[0].top_material == "泵头"
+    assert result[0].top_problem == "漏液"
+
+
+def test_supplier_recent_records_are_limited_and_sorted() -> None:
+    service = SupplierProfileService(object(), as_of=date(2026, 9, 7))
+    rows = [
+        {
+            "id": index,
+            "date": date(2026, 4, index),
+            "product_name": "产品",
+            "anomaly_type": str(index),
+            "defect_level": "B",
+        }
+        for index in range(1, 13)
+    ]
+
+    result = service._recent_records(rows)
+
+    assert len(result) == 10
+    assert result[0].case_id == 12