统一供应商统计口径并优化后端查询性能

统一对话页与供应商画像的采购订单、异常数、基础分、频次系数、供应商得分和层级计算；补充供应商维度计算结果字段，保留未归类品类标记并优化供应商列表聚合查询；增加响应压缩并完善供应商统计测试。

app/api/router.py

@@ -7,6 +7,7 @@ from app.api.v1.conversations import router as conversations_router
 from app.api.v1.dashboard import router as dashboard_router
 from app.api.v1.health import router as health_router
 from app.api.v1.model_config import router as model_config_router
+from app.api.v1.monitoring import router as monitoring_router
 from app.api.v1.ocr import router as ocr_router
 from app.api.v1.reports import router as reports_router
 from app.api.v1.risk_items import router as risk_items_router
@@ -17,6 +18,7 @@ from app.api.v1.suppliers import router as suppliers_router
 api_router = APIRouter(prefix="/api/v1")
 api_router.include_router(health_router)
 api_router.include_router(model_config_router)
+api_router.include_router(monitoring_router)
 api_router.include_router(ocr_router)
 api_router.include_router(auth_router)
 api_router.include_router(conversations_router)

app/api/v1/monitoring.py

@@ -0,0 +1,28 @@
+from typing import Annotated
+
+from fastapi import APIRouter, Depends
+
+from app.api.dependencies import DatabaseSession, require_permission
+from app.models.entities import OperationAudit, UserAccount
+from app.schemas.common import ApiResponse
+
+router = APIRouter(prefix="/monitoring", tags=["monitoring"])
+
+
+@router.post("/risk")
+async def monitor_risk(
+    payload: dict,
+    session: DatabaseSession,
+    current_user: Annotated[UserAccount, Depends(require_permission("risk.assess", operate=True))],
+) -> ApiResponse[dict]:
+    session.add(
+        OperationAudit(
+            actor_user_id=current_user.id,
+            actor_emp_no=current_user.emp_no,
+            action="risk.monitor.add",
+            resource_type="risk_assessment",
+            resource_id=str(payload.get("subject_id") or ""),
+            audit_metadata=payload,
+        )
+    )
+    return ApiResponse(data={"monitored": True, "subject_id": payload.get("subject_id")})

app/main.py

@@ -2,6 +2,7 @@
 
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
+from starlette.middleware.gzip import GZipMiddleware
 
 from app.api.router import api_router
 from app.core.config import Settings, get_settings
@@ -25,6 +26,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
         allow_methods=["GET", "POST", "PATCH", "DELETE", "OPTIONS"],
         allow_headers=["Authorization", "Content-Type", "Accept"],
     )
+    application.add_middleware(GZipMiddleware, minimum_size=1000, compresslevel=5)
     application.add_exception_handler(AppException, app_exception_handler)
     application.include_router(api_router)
     return application

app/schemas/risk_prediction.py

@@ -79,6 +79,7 @@ class SimilarCase(ApiSchema):
     description: str | None = None
     root_cause: str | None = None
     defect_level: str | None = None
+    result: str | None = None
     problem: str | None = None
     solution: str | None = None
     similarity: float | None = None

app/schemas/suppliers.py

@@ -21,6 +21,8 @@ class SupplierMetric(ApiSchema):
     anomaly_total: int
     anomaly_rate: float | None = None
     frequency_alpha: float
+    supplier_base_score: float | None = None
+    supplier_score: float | None = None
     risk_score: float | None = None
     main_category_count: int
     cooperation_count_basis: str = "按近3年采购订单去重单据编号统计"
@@ -29,6 +31,7 @@ class SupplierMetric(ApiSchema):
 
 class SupplierCategoryBreakdown(ApiSchema):
     category: str
+    is_unclassified: bool = False
     anomaly_count: int
     anomaly_rate: float | None = None
     top_material: str | None = None

app/services/conversations.py

@@ -576,6 +576,16 @@ class ConversationService:
         )
         party_type = "加工厂" if is_factory else "供应商"
         tier = self._supplier_tier(anomaly_count, a_count=a_count, b_count=b_count)
+        supplier_dimension = None
+        if not is_factory and hasattr(self, "session"):
+            database_score = await DatabaseRiskScoringService(
+                self.session,
+                as_of=datetime.now(UTC).date(),
+            ).score(supplier=name)
+            supplier_dimension = next(
+                (item for item in database_score.dimensions if item.name == "供应商表现"),
+                None,
+            )
         return {
             "type": "supplier",
             "title": "供应商深度画像",
@@ -584,9 +594,15 @@ class ConversationService:
             "anomalyCount": anomaly_count,
             "cooperationCount": cooperation_display,
             "tier": tier,
-            "frequencyCoefficient": round(min(1.5, 1 + cooperation_display / 100 * 0.3), 2)
-            if cooperation_display is not None
-            else 1.0,
+            "frequencyCoefficient": (
+                supplier_dimension.frequency_alpha
+                if supplier_dimension and supplier_dimension.frequency_alpha is not None
+                else round(min(1.5, 1 + cooperation_display / 100 * 0.3), 2)
+                if cooperation_display is not None
+                else 1.0
+            ),
+            "supplierBaseScore": supplier_dimension.base_score if supplier_dimension else None,
+            "supplierScore": supplier_dimension.score if supplier_dimension else None,
             "topProblems": problems,
             "suggestions": self._supplier_management_suggestions(tier),
             "frequencyBasis": "近3年采购订单去重单据数" if has_order_data else "暂无匹配采购订单",
@@ -958,7 +974,10 @@ class ConversationService:
             supplier = None
         supplier_or_factory = supplier or factory
 
-        database_score = await DatabaseRiskScoringService(self.session).score(
+        database_score = await DatabaseRiskScoringService(
+            self.session,
+            as_of=datetime.now(UTC).date(),
+        ).score(
             category=category,
             material=self._first_match(
                 text,
@@ -1006,7 +1025,7 @@ class ConversationService:
         alpha = round(min(1.5, 1 + (alpha_count / 100) * 0.3), 2) if cooperation_count else 1.0
         tier = (
             self._supplier_tier(
-                alpha_count,
+                supplier_count,
                 a_count=supplier_a_count,
                 b_count=supplier_b_count,
             )
@@ -1033,6 +1052,10 @@ class ConversationService:
             )
             for item in database_score.dimensions
         ]
+        supplier_dimension = next(
+            (item for item in database_score.dimensions if item.name == "供应商表现"),
+            None,
+        )
         factors = [
             RiskFactorResult(
                 name=item.name,
@@ -1079,7 +1102,11 @@ class ConversationService:
                         supplier_name=supplier,
                         cooperation_count=cooperation_count,
                         anomaly_count=supplier_count,
-                        alpha=alpha,
+                        alpha=(
+                            supplier_dimension.frequency_alpha
+                            if supplier_dimension and supplier_dimension.frequency_alpha is not None
+                            else alpha
+                        ),
                         tier=tier,
                         suggestion="；".join(self._supplier_management_suggestions(tier)),
                         frequency_basis="近3年采购订单去重单据数" if cooperation_count else "暂无匹配采购订单",

app/services/database_risk_scoring.py

@@ -26,6 +26,8 @@ class DimensionScore:
     reason: str | None = None
     sample_size: int | None = None
     raw_value: float | int | None = None
+    base_score: float | None = None
+    frequency_alpha: float | None = None
 
 
 @dataclass(frozen=True, slots=True)
@@ -59,19 +61,25 @@ class DatabaseRiskScoringService:
         factory: str | None = None,
         stage: str | None = None,
         defect_level: str | None = None,
+        rows: list[dict[str, Any]] | None = None,
+        cooperation_count: int | None = None,
     ) -> DatabaseRiskScore:
         await self._load_model_config()
-        rows = await self._load_rows()
+        source_rows = rows if rows is not None else await self._load_rows()
         category_name = normalize_category(category)
         material_name = normalize_material(material)
         stage_name = normalize_stage(stage)
         party_name = clean_value(supplier)
 
-        cooperation_count = await self._supplier_order_count(party_name)
+        resolved_cooperation_count = (
+            cooperation_count
+            if cooperation_count is not None
+            else await self._supplier_order_count(party_name)
+        )
         dimensions = (
-            self._supplier_dimension(rows, party_name, cooperation_count),
-            self._material_dimension(rows, material_name),
-            self._category_dimension(rows, category_name),
+            self._supplier_dimension(source_rows, party_name, resolved_cooperation_count),
+            self._material_dimension(source_rows, material_name),
+            self._category_dimension(source_rows, category_name),
             DimensionScore(
                 name="加工厂质量",
                 score=50.0,
@@ -86,8 +94,8 @@ class DatabaseRiskScoringService:
                 status="default",
                 reason="当前数据库没有检测结果与标准判定记录，按 PRD 中性值 50 分计入",
             ),
-            self._stage_dimension(rows, stage_name),
-            self._defect_dimension(rows, category_name, material_name, party_name, defect_level),
+            self._stage_dimension(source_rows, stage_name),
+            self._defect_dimension(source_rows, category_name, material_name, party_name, defect_level),
         )
         score = round(sum(item.score * item.weight / 100 for item in dimensions), 2)
         return DatabaseRiskScore(score=score, level=classify_score(score), dimensions=dimensions)
@@ -198,6 +206,8 @@ class DatabaseRiskScoringService:
             ),
             sample_size=anomaly_count,
             raw_value=count,
+            base_score=base,
+            frequency_alpha=alpha,
         )
 
     def _material_dimension(self, rows: list[dict[str, Any]], material: str | None) -> DimensionScore:

app/services/rag.py

@@ -123,6 +123,7 @@ class AnomalyCaseRAG:
             AnomalyCase.solution,
             AnomalyCase.root_cause,
             AnomalyCase.defect_level,
+            AnomalyCase.result,
         )
         vector_ready = await self._vector_column_ready()
         has_vectors = await self._has_stored_vectors() if vector_ready else False
@@ -135,7 +136,7 @@ class AnomalyCaseRAG:
             """
             SELECT id, date, product_name, product_category, material_type, supplier,
                    factory, stage_new, anomaly_type, description, solution, root_cause,
-                   defect_level, 1 - (embedding_vector <=> CAST(:embedding AS halfvec)) AS vector_score
+                   defect_level, result, 1 - (embedding_vector <=> CAST(:embedding AS halfvec)) AS vector_score
             FROM anomaly_cases
             WHERE embedding_vector IS NOT NULL
             ORDER BY embedding_vector <=> CAST(:embedding AS halfvec)
@@ -306,6 +307,7 @@ class AnomalyCaseRAG:
             description=row.get("description"),
             root_cause=row.get("root_cause"),
             defect_level=row.get("defect_level"),
+            result=row.get("result"),
             problem=(row.get("anomaly_type") or row.get("description") or "历史异常未填写"),
             solution=(row.get("solution") or row.get("root_cause") or "请查看原始案例记录"),
         )

app/services/suppliers.py

@@ -2,10 +2,11 @@
 
 from __future__ import annotations
 
+import re
 from collections import Counter, defaultdict
 from datetime import UTC, date, datetime, timedelta
 
-from sqlalchemy import select
+from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.exceptions import AppException
@@ -33,19 +34,20 @@ class SupplierProfileService:
         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)
+        grouped = await self._load_supplier_summary(keyword=keyword)
         items = [
             SupplierOption(
                 name=name,
-                anomaly_count=len(items),
+                anomaly_count=summary["total"],
                 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),
+                    summary["total"],
+                    summary["a"],
+                    summary["b"],
                 ),
             )
-            for name, items in sorted(grouped.items(), key=lambda item: (-len(item[1]), item[0]))[:limit]
+            for name, summary in sorted(
+                grouped.items(), key=lambda item: (-item[1]["total"], item[0])
+            )[:limit]
         ]
         return SupplierListData(
             items=items,
@@ -53,7 +55,8 @@ class SupplierProfileService:
         )
 
     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)]
+        all_rows = await self._load_rows()
+        rows = [row for row in all_rows if self.matches(row.get("supplier"), supplier_name)]
         if not rows:
             raise AppException("供应商不存在或暂无近3年异常记录", code=404, status_code=404)
 
@@ -69,31 +72,95 @@ class SupplierProfileService:
         tier = self.tier_for(anomaly_total, a_count, b_count)
         categories = self._categories(rows)
         scoring = DatabaseRiskScoringService(self.session, as_of=self.as_of)
-        score = await scoring.score(supplier=supplier_name)
+        score = await scoring.score(
+            supplier=supplier_name,
+            rows=all_rows,
+            cooperation_count=order_count,
+        )
+        supplier_dimension = next(
+            (item for item in score.dimensions if item.name == "供应商表现"),
+            None,
+        )
+        cooperation_display = order_count if order_count > 0 else None
+        display_tier = {
+            "S级": "S级 · 战略",
+            "A级": "A级 · 优质",
+            "B级": "B级 · 关注",
+            "C级": "C级 · 高风险",
+        }.get(tier, tier)
         return SupplierProfileData(
             supplier_name=supplier_name,
-            tier=tier,
+            tier=display_tier,
             tier_condition=self.tier_condition(tier),
             metric=SupplierMetric(
-                cooperation_total=order_count,
+                cooperation_total=cooperation_display,
                 anomaly_total=anomaly_total,
                 anomaly_rate=None,
-                frequency_alpha=self.alpha(
-                    order_count,
-                    coefficient=scoring.frequency_coefficient,
-                    cap=scoring.frequency_cap,
+                frequency_alpha=(
+                    supplier_dimension.frequency_alpha
+                    if supplier_dimension and supplier_dimension.frequency_alpha is not None
+                    else self.alpha(
+                        cooperation_display or anomaly_total,
+                        coefficient=scoring.frequency_coefficient,
+                        cap=scoring.frequency_cap,
+                    )
                 ),
+                supplier_base_score=(
+                    supplier_dimension.base_score if supplier_dimension else None
+                ),
+                supplier_score=supplier_dimension.score if supplier_dimension else None,
                 risk_score=score.score,
                 main_category_count=len(categories),
             ),
-            category_breakdown=categories[: self.max_categories],
+            category_breakdown=self._display_categories(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]:
+    async def _load_supplier_summary(
+        self, *, keyword: str | None = None
+    ) -> dict[str, dict[str, int]]:
+        """Load only supplier aggregates for the selector.
+
+        The selector used to fetch every anomaly row and group it in Python.
+        This grouped query returns a handful of rows even when the anomaly
+        table grows, which materially reduces first-navigation latency over
+        the remote PostgreSQL connection.
+        """
+
+        start = self.as_of - timedelta(days=self.history_days)
+        statement = (
+            select(AnomalyCase.supplier, AnomalyCase.defect_level, func.count())
+            .where(
+                AnomalyCase.supplier.is_not(None),
+                AnomalyCase.date.is_not(None),
+                AnomalyCase.date >= start,
+                AnomalyCase.date <= self.as_of,
+            )
+            .group_by(AnomalyCase.supplier, AnomalyCase.defect_level)
+        )
+        if keyword and keyword.strip():
+            statement = statement.where(AnomalyCase.supplier.ilike(f"%{keyword.strip()}%"))
+        grouped: dict[str, dict[str, int]] = defaultdict(
+            lambda: {"total": 0, "a": 0, "b": 0}
+        )
+        for supplier, defect_level, count in (await self.session.execute(statement)).all():
+            name = (supplier or "").strip()
+            if not name:
+                continue
+            value = grouped[name]
+            amount = int(count or 0)
+            value["total"] += amount
+            code = self.defect_code(defect_level)
+            if code == "A":
+                value["a"] += amount
+            elif code == "B":
+                value["b"] += amount
+        return grouped
+
+    async def _load_rows(self, *, supplier_name: str | None = None) -> list[dict]:
         start = self.as_of - timedelta(days=self.history_days)
         statement = select(
             AnomalyCase.id,
@@ -110,6 +177,8 @@ class SupplierProfileService:
             AnomalyCase.date >= start,
             AnomalyCase.date <= self.as_of,
         )
+        if supplier_name:
+            statement = statement.where(AnomalyCase.supplier.ilike(f"%{supplier_name}%"))
         return [dict(row) for row in (await self.session.execute(statement)).mappings().all()]
 
     @staticmethod
@@ -127,8 +196,8 @@ class SupplierProfileService:
 
     @staticmethod
     def defect_code(value: str | None) -> str | None:
-        value = (value or "").strip().upper()
-        return value if value in {"A", "B", "C"} else None
+        match = re.search(r"[ABCD]", (value or "").upper())
+        return match.group(0) if match else None
 
     @staticmethod
     def alpha(anomaly_count: int, *, coefficient: float = 0.3, cap: float = 1.5) -> float:
@@ -165,18 +234,30 @@ class SupplierProfileService:
             grouped[category or "待补标品类"].append(row)
         result = []
         for category, items in grouped.items():
+            is_unclassified = category == "待补标品类"
             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,
+                    is_unclassified=is_unclassified,
                     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))
+        return sorted(result, key=lambda item: (item.is_unclassified, -item.anomaly_count, item.category))
+
+    @classmethod
+    def _display_categories(
+        cls, categories: list[SupplierCategoryBreakdown]
+    ) -> list[SupplierCategoryBreakdown]:
+        """Keep up to five real categories and always retain the unclassified row."""
+
+        real = [item for item in categories if not item.is_unclassified][: cls.max_categories]
+        unclassified = next((item for item in categories if item.is_unclassified), None)
+        return real + ([unclassified] if unclassified else [])
 
     def _monthly_trend(self, rows: list[dict]) -> list[dict[str, int | str]]:
         start = self._month_offset(self.as_of.replace(day=1), -5)

tests/test_suppliers.py

@@ -47,6 +47,32 @@ def test_supplier_category_breakdown_is_top_five_with_material_and_problem() ->
     assert result[0].top_problem == "漏液"
 
 
+def test_unclassified_category_is_marked_and_sorted_after_real_categories() -> None:
+    rows = [
+        {"product_category": None, "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].is_unclassified is False
+    assert result[1].category == "待补标品类"
+    assert result[1].is_unclassified is True
+
+
+def test_display_categories_keeps_unclassified_after_five_real_categories() -> None:
+    rows = [
+        {"product_category": category, "product_name": category, "material_type": "瓶子", "anomaly_type": "异常"}
+        for category in ["洗发水", "面霜", "沐浴露", "牙膏", "清洁剂", "脱毛膏"]
+    ] + [{"product_category": None, "product_name": "未命名", "material_type": "瓶子", "anomaly_type": "异常"}]
+
+    result = SupplierProfileService._display_categories(SupplierProfileService._categories(rows))
+
+    assert len(result) == 6
+    assert result[-1].is_unclassified is True
+
+
 def test_supplier_recent_records_are_limited_and_sorted() -> None:
     service = SupplierProfileService(object(), as_of=date(2026, 9, 7))
     rows = [