From 5bdb2852e573353ebe3c631d26046face8cfb1d4 Mon Sep 17 00:00:00 2001
From: kx_guo <kx_guo@163.com>
Date: Thu, 10 Sep 2026 17:33:54 +0800
Subject: [PATCH] =?UTF-8?q?=E7=BB=9F=E4=B8=80=E4=BE=9B=E5=BA=94=E5=95=86?=
 =?UTF-8?q?=E4=B8=BB=E6=95=B0=E6=8D=AE=E5=B9=B6=E4=BC=98=E5=8C=96=E5=AF=B9?=
 =?UTF-8?q?=E8=AF=9D=E5=90=91=E9=87=8F=E6=A3=80=E7=B4=A2?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 README.md                                     |   9 ++
 app/models/__init__.py                        |   4 +
 app/models/entities.py                        |  60 +++++++-
 app/schemas/suppliers.py                      |   8 ++
 app/services/conversations.py                 |  36 ++++-
 app/services/database_risk_scoring.py         |  19 ++-
 app/services/rag.py                           |   9 +-
 app/services/supplier_cleaning.py             | 126 +++++++++++++++++
 app/services/supplier_identity.py             |  44 ++++++
 app/services/suppliers.py                     |  83 ++++++++++--
 ...20260910_0019_supplier_master_and_links.py |  62 +++++++++
 scripts/backfill_embeddings.py                |  52 ++++---
 scripts/clean_suppliers.py                    | 128 ++++++++++++++++++
 tests/test_models.py                          |   2 +
 tests/test_supplier_cleaning.py               |  29 ++++
 tests/test_suppliers.py                       |   4 +
 16 files changed, 635 insertions(+), 40 deletions(-)
 create mode 100644 app/services/supplier_cleaning.py
 create mode 100644 app/services/supplier_identity.py
 create mode 100644 migrations/versions/20260910_0019_supplier_master_and_links.py
 create mode 100644 scripts/clean_suppliers.py
 create mode 100644 tests/test_supplier_cleaning.py

diff --git a/README.md b/README.md
index d0ccabb..d3631d0 100644
--- a/README.md
+++ b/README.md
@@ -41,6 +41,15 @@ python scripts/import_excel_data.py `
 
 脚本只导入异常明细 Sheet 的 2,448 条记录，以及标准文件中的“包装/包材/法规/内料”四个分类 Sheet 共 219 条记录；“总表”不重复导入。脚本可重复执行，已导入行会按源文件、Sheet 和 Excel 行号自动跳过。原始异常文本和标准附件链接会完整保留，不做截断。
 
+供应商清洗采用“原始字段保留 + 主数据 + 多对多关联”方式，不覆盖 `anomaly_cases.supplier`。使用具备建表及对 `anomaly_cases` 建外键权限的数据库账号执行：
+
+```powershell
+python -m alembic upgrade head
+python scripts/clean_suppliers.py
+```
+
+脚本会拆分 `/`、`、`、逗号、分号及换行分隔的复合供应商，生成 `supplier_master` 和 `anomaly_case_suppliers`；自动结果默认处于“待确认”状态。管理员应补充主数据中的标准名称、别名、城市与省份后再标记为“已确认”。供应商搜索会优先匹配标准名称、简称、别名、城市和省份；迁移尚未执行时自动回退到原始字段搜索。
+
 风险评估预览：`POST http://127.0.0.1:18474/api/v1/risk-predictions/preview`。指标值表示风险程度（0-100），接口返回的综合分按 PRD 转换为安全得分，分数越高风险越低；等级阈值为 80/60/40/20。
 
 对话页真实 AI：配置 `RISK_LLM_API_KEY`、`RISK_LLM_BASE_URL` 和 `RISK_LLM_MODEL` 后，调用 `POST /api/v1/conversations/{id}/stream`。接口使用 OpenAI 兼容 Chat Completions 协议，先提取 `intent/category/specification/packaging/supplier/factory/stage/confidence` 语义 JSON，再通过 SSE 返回增量文本；最近 20 条消息作为上下文，完整助手消息、语义结果和风险评分会在流结束后持久化。也兼容参考项目的 `OPENAI_API_KEY`、`LLM_BASE_URL`、`LLM_MODEL` 配置名。未配置密钥时返回明确的 503，不生成模拟回复。
diff --git a/app/models/__init__.py b/app/models/__init__.py
index 4eafc59..c07312e 100644
--- a/app/models/__init__.py
+++ b/app/models/__init__.py
@@ -2,6 +2,7 @@
 
 from app.models.entities import (
     AnomalyCase,
+    AnomalyCaseSupplier,
     Conversation,
     ConversationMessage,
     OperationAudit,
@@ -13,11 +14,13 @@ from app.models.entities import (
     RolePermission,
     StandardRule,
     SupplierProfile,
+    SupplierMaster,
     UserAccount,
 )
 
 __all__ = [
     "AnomalyCase",
+    "AnomalyCaseSupplier",
     "Conversation",
     "ConversationMessage",
     "OperationAudit",
@@ -29,5 +32,6 @@ __all__ = [
     "RolePermission",
     "StandardRule",
     "SupplierProfile",
+    "SupplierMaster",
     "UserAccount",
 ]
diff --git a/app/models/entities.py b/app/models/entities.py
index 253ca86..ea87d86 100644
--- a/app/models/entities.py
+++ b/app/models/entities.py
@@ -4,7 +4,7 @@ from datetime import date, datetime
 from decimal import Decimal
 from typing import Any
 
-from sqlalchemy import BigInteger, Boolean, Date, DateTime, ForeignKey, Index, Integer, Numeric, String, Text, func
+from sqlalchemy import BigInteger, Boolean, Date, DateTime, ForeignKey, Index, Integer, Numeric, String, Text, UniqueConstraint, func
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
@@ -59,6 +59,64 @@ class AnomalyCase(Base):
 
     standard_rule: Mapped["StandardRule | None"] = relationship(back_populates="anomaly_cases")
     review_records: Mapped[list["ReviewRecord"]] = relationship(back_populates="case")
+    supplier_links: Mapped[list["AnomalyCaseSupplier"]] = relationship(
+        back_populates="case", cascade="all, delete-orphan"
+    )
+
+
+class SupplierMaster(Base):
+    """Canonical supplier entity independent from raw workbook text."""
+
+    __tablename__ = "supplier_master"
+    __table_args__ = (
+        Index("ix_supplier_master_city", "city"),
+        Index("ix_supplier_master_normalized_name", "normalized_name", unique=True),
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    canonical_name: Mapped[str] = mapped_column(String(200), nullable=False)
+    short_name: Mapped[str | None] = mapped_column(String(100))
+    normalized_name: Mapped[str] = mapped_column(String(200), nullable=False)
+    city: Mapped[str | None] = mapped_column(String(50))
+    province: Mapped[str | None] = mapped_column(String(50))
+    supplier_type: Mapped[str | None] = mapped_column(String(50))
+    aliases: Mapped[list[str]] = mapped_column(JSONB, default=list, server_default="[]", nullable=False)
+    status: Mapped[str] = mapped_column(String(20), default="启用", server_default="启用", nullable=False)
+    source: Mapped[str | None] = mapped_column(String(100))
+    review_status: Mapped[str] = mapped_column(String(20), default="待确认", server_default="待确认", nullable=False)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), nullable=False)
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False
+    )
+
+    anomaly_links: Mapped[list["AnomalyCaseSupplier"]] = relationship(
+        back_populates="supplier", cascade="all, delete-orphan"
+    )
+
+
+class AnomalyCaseSupplier(Base):
+    """Many-to-many mapping from an anomaly case to canonical suppliers."""
+
+    __tablename__ = "anomaly_case_suppliers"
+    __table_args__ = (
+        UniqueConstraint("case_id", "supplier_id", "supplier_role", name="uq_case_supplier_role"),
+        Index("ix_case_suppliers_supplier", "supplier_id"),
+        Index("ix_case_suppliers_case", "case_id"),
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    case_id: Mapped[int] = mapped_column(BigInteger, ForeignKey("anomaly_cases.id", ondelete="CASCADE"), nullable=False)
+    supplier_id: Mapped[int] = mapped_column(BigInteger, ForeignKey("supplier_master.id", ondelete="CASCADE"), nullable=False)
+    supplier_role: Mapped[str | None] = mapped_column(String(50))
+    material_type: Mapped[str | None] = mapped_column(String(200))
+    source_text: Mapped[str | None] = mapped_column(Text)
+    match_method: Mapped[str] = mapped_column(String(30), default="manual", server_default="manual", nullable=False)
+    match_confidence: Mapped[Decimal | None] = mapped_column(Numeric(5, 4))
+    review_status: Mapped[str] = mapped_column(String(20), default="待确认", server_default="待确认", nullable=False)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), nullable=False)
+
+    case: Mapped[AnomalyCase] = relationship(back_populates="supplier_links")
+    supplier: Mapped[SupplierMaster] = relationship(back_populates="anomaly_links")
 
 
 class SupplierProfile(Base):
diff --git a/app/schemas/suppliers.py b/app/schemas/suppliers.py
index fb2ead1..eb640a3 100644
--- a/app/schemas/suppliers.py
+++ b/app/schemas/suppliers.py
@@ -9,6 +9,14 @@ class SupplierOption(ApiSchema):
     name: str
     anomaly_count: int
     tier: str
+    city: str | None = None
+    province: str | None = None
+    short_name: str | None = None
+    review_status: str | None = None
+    city: str | None = None
+    province: str | None = None
+    short_name: str | None = None
+    review_status: str | None = None
 
 
 class SupplierListData(ApiSchema):
diff --git a/app/services/conversations.py b/app/services/conversations.py
index 8408c3b..f8f523c 100644
--- a/app/services/conversations.py
+++ b/app/services/conversations.py
@@ -37,6 +37,7 @@ from app.schemas.risk_prediction import (
 from app.services.database_risk_scoring import DatabaseRiskScoringService
 from app.services.llm_client import LLMClient
 from app.services.purchase_orders import calendar_year_window, supplier_order_counts
+from app.services.supplier_identity import resolve_supplier
 from app.services.rag import AnomalyCaseRAG
 
 logger = logging.getLogger("dieya_risk_prediction.conversations")
@@ -412,6 +413,9 @@ class ConversationService:
             ).all()
         )
         supplier_name = self._infer_supplier_follow_up(text, supplier_names)
+        master_identity = await resolve_supplier(self.session, text)
+        if not supplier_name and master_identity:
+            supplier_name = master_identity.canonical_name
         supplier_name = supplier_name or self._infer_unknown_supplier_follow_up(text)
         if not supplier_name:
             return normalized
@@ -421,7 +425,9 @@ class ConversationService:
                 "risk_requested": True,
                 "supplier": supplier_name,
                 "confidence": max(float(normalized.get("confidence") or 0), 0.75),
-                "semantic_source": "database_supplier_match"
+                "semantic_source": "supplier_master_match"
+                if master_identity
+                else "database_supplier_match"
                 if supplier_name in supplier_names
                 else "standalone_new_supplier_fallback",
             }
@@ -575,12 +581,19 @@ class ConversationService:
             match = re.search(r"(?:供应商|加工厂|工厂)\s*(?:是|用|为)?\s*([^，。！？?]+)", text)
             name = match.group(1).strip() if match else "未指定"
         is_factory = any(word in text for word in ("加工厂", "工厂", "生产厂")) and "供应商" not in text
+        identity = (
+            None
+            if is_factory or not hasattr(self, "session")
+            else await resolve_supplier(self.session, name)
+        )
+        lookup_name = identity.short_name or identity.aliases[0] if identity else name
+        display_name = identity.canonical_name if identity else name
         column = AnomalyCase.factory if is_factory else AnomalyCase.supplier
-        anomaly_count, a_count, b_count = await self._supplier_history_stats(column, name)
+        anomaly_count, a_count, b_count = await self._supplier_history_stats(column, lookup_name)
         cooperation_count = (
             await supplier_order_counts(
                 self.session,
-                name,
+                display_name,
                 self._supplier_history_start(),
                 self._supplier_history_end(),
             )
@@ -592,7 +605,7 @@ class ConversationService:
         cooperation_display = cooperation_count if has_order_data else None
         problems = await self._top_anomaly_types(
             column,
-            name,
+            lookup_name,
             since=self._supplier_history_start(),
         )
         party_type = "加工厂" if is_factory else "供应商"
@@ -602,7 +615,7 @@ class ConversationService:
             database_score = await DatabaseRiskScoringService(
                 self.session,
                 as_of=datetime.now(UTC).date(),
-            ).score(supplier=name)
+            ).score(supplier=display_name)
             supplier_dimension = next(
                 (item for item in database_score.dimensions if item.name == "供应商表现"),
                 None,
@@ -610,7 +623,7 @@ class ConversationService:
         return {
             "type": "supplier",
             "title": "供应商深度画像",
-            "name": name,
+            "name": display_name,
             "partyType": party_type,
             "anomalyCount": anomaly_count,
             "cooperationCount": cooperation_display,
@@ -1039,6 +1052,15 @@ class ConversationService:
             factory = factory or supplier
             supplier = None
         supplier_or_factory = supplier or factory
+        supplier_identity = await resolve_supplier(self.session, supplier)
+        supplier_lookup = (
+            supplier_identity.short_name or supplier_identity.aliases[0]
+            if supplier_identity
+            else supplier
+        )
+        if supplier_identity:
+            supplier = supplier_identity.canonical_name
+            supplier_or_factory = supplier
 
         database_score = await DatabaseRiskScoringService(
             self.session,
@@ -1059,7 +1081,7 @@ class ConversationService:
         category_count = await self._count_category_anomalies(category)
         supplier_count, supplier_a_count, supplier_b_count = await self._supplier_history_stats(
             AnomalyCase.supplier,
-            supplier,
+            supplier_lookup,
         )
         factory_count = await self._count_anomalies(AnomalyCase.factory, factory)
         anomaly_count = supplier_count if supplier else factory_count
diff --git a/app/services/database_risk_scoring.py b/app/services/database_risk_scoring.py
index b495a6a..b627017 100644
--- a/app/services/database_risk_scoring.py
+++ b/app/services/database_risk_scoring.py
@@ -13,6 +13,7 @@ from app.domain.model_config import DEFAULT_FREQUENCY_CAP, DEFAULT_FREQUENCY_COE
 from app.domain.risk_engine import RiskLevel
 from app.models.entities import AnomalyCase, RiskModelConfig
 from app.services.purchase_orders import calendar_year_window, supplier_order_counts
+from app.services.supplier_identity import resolve_supplier
 
 STAGES = ("需求发现", "产品立项", "产品开发", "量产上市", "产品生命管理")
 
@@ -69,7 +70,9 @@ class DatabaseRiskScoringService:
         category_name = normalize_category(category)
         material_name = normalize_material(material)
         stage_name = normalize_stage(stage)
-        party_name = clean_value(supplier)
+        identity = await resolve_supplier(self.session, clean_value(supplier))
+        party_name = identity.canonical_name if identity else clean_value(supplier)
+        party_names = identity.match_names if identity else ((party_name,) if party_name else ())
 
         resolved_cooperation_count = (
             cooperation_count
@@ -77,7 +80,7 @@ class DatabaseRiskScoringService:
             else await self._supplier_order_count(party_name)
         )
         dimensions = (
-            self._supplier_dimension(source_rows, party_name, resolved_cooperation_count),
+            self._supplier_dimension(source_rows, party_name, resolved_cooperation_count, party_names),
             self._material_dimension(source_rows, material_name),
             self._category_dimension(source_rows, category_name),
             DimensionScore(
@@ -95,7 +98,7 @@ class DatabaseRiskScoringService:
                 reason="当前数据库没有检测结果与标准判定记录，按 PRD 中性值 50 分计入",
             ),
             self._stage_dimension(source_rows, stage_name),
-            self._defect_dimension(source_rows, category_name, material_name, party_name, defect_level),
+            self._defect_dimension(source_rows, category_name, material_name, party_name, defect_level, party_names),
         )
         score = round(sum(item.score * item.weight / 100 for item in dimensions), 2)
         return DatabaseRiskScore(score=score, level=classify_score(score), dimensions=dimensions)
@@ -148,6 +151,7 @@ class DatabaseRiskScoringService:
         rows: list[dict[str, Any]],
         party_name: str | None,
         cooperation_count: int | None = None,
+        party_names: tuple[str, ...] = (),
     ) -> DimensionScore:
         weight = self.weights["供应商表现"]
         if not party_name:
@@ -158,7 +162,7 @@ class DatabaseRiskScoringService:
                 status="default",
                 reason="未提供供应商/加工厂，按 PRD 缺失维度默认 50 分",
             )
-        matched = [row for row in rows if party_matches(row, party_name)]
+        matched = [row for row in rows if party_matches_any(row, party_names or (party_name,))]
         anomaly_count = len(matched)
         count = cooperation_count if cooperation_count is not None and cooperation_count > 0 else anomaly_count
         if count == 0:
@@ -312,6 +316,7 @@ class DatabaseRiskScoringService:
         material: str | None,
         party_name: str | None,
         defect_level: str | None,
+        party_names: tuple[str, ...] = (),
     ) -> DimensionScore:
         weight = self.weights["缺陷等级"]
         if not category and not material and not party_name and not defect_level:
@@ -331,7 +336,7 @@ class DatabaseRiskScoringService:
                 or normalize_category(row.get("product_name")) == category
             )
             and (not material or normalize_material(row.get("material_type")) == material)
-            and (not party_name or party_matches(row, party_name))
+            and (not party_name or party_matches_any(row, party_names or (party_name,)))
         ]
         counts = {
             level: sum(
@@ -433,6 +438,10 @@ def party_matches(row: dict[str, Any], party_name: str) -> bool:
     return party_name.lower() in supplier.lower()
 
 
+def party_matches_any(row: dict[str, Any], party_names: tuple[str, ...]) -> bool:
+    return any(party_matches(row, name) for name in party_names)
+
+
 def classify_score(score: float) -> RiskLevel:
     if score >= 80:
         return RiskLevel.LOW
diff --git a/app/services/rag.py b/app/services/rag.py
index 29c70cc..ade10fc 100644
--- a/app/services/rag.py
+++ b/app/services/rag.py
@@ -16,6 +16,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from app.core.config import get_settings
 from app.models.entities import AnomalyCase
 from app.schemas.risk_prediction import SimilarCase
+from app.services.supplier_identity import resolve_supplier
 
 logger = logging.getLogger("dieya_risk_prediction.rag")
 _embedding_semaphore = asyncio.Semaphore(1)
@@ -53,6 +54,12 @@ class AnomalyCaseRAG:
         stage: str | None = None,
         anomaly_type: str | None = None,
     ) -> list[RetrievedCase]:
+        identity = await resolve_supplier(self.session, supplier)
+        supplier_match = (
+            identity.short_name or identity.aliases[0]
+            if identity
+            else supplier
+        )
         terms = self._terms(query, category, material, supplier, factory, stage, anomaly_type)
         rows = await self._load_rows(
             query,
@@ -71,7 +78,7 @@ class AnomalyCaseRAG:
                     query=query,
                     category=category,
                     material=material,
-                    supplier=supplier,
+                    supplier=supplier_match,
                     factory=factory,
                     stage=stage,
                     anomaly_type=anomaly_type,
diff --git a/app/services/supplier_cleaning.py b/app/services/supplier_cleaning.py
new file mode 100644
index 0000000..041b6f1
--- /dev/null
+++ b/app/services/supplier_cleaning.py
@@ -0,0 +1,126 @@
+"""Pure helpers for splitting and normalizing raw supplier cells."""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+
+
+_SEPARATOR_RE = re.compile(r"[/／、,，;；|\\\n]+")
+_LABEL_RE = re.compile(r"(?P<label>[\u4e00-\u9fffA-Za-z0-9]+)\s*[:：]")
+
+
+@dataclass(frozen=True)
+class SupplierMention:
+    name: str
+    role: str | None
+    material_type: str | None
+    source_text: str
+    method: str
+    confidence: float
+
+
+def compact_text(value: str) -> str:
+    return re.sub(r"[\s()（）【】\[\]{}]+", "", value.casefold()).strip()
+
+
+def normalize_name(value: str) -> str:
+    value = re.sub(r"\s+", "", value).strip(" ：:、,，;；/\\")
+    for suffix in ("有限责任公司", "有限公司", "公司"):
+        if value.endswith(suffix) and len(value) > len(suffix) + 1:
+            value = value[: -len(suffix)]
+            break
+    if value.endswith("市") and len(value) > 2:
+        # Keep city in canonical display names; this is only a normalized key.
+        pass
+    return value
+
+
+def normalized_key(value: str) -> str:
+    return compact_text(normalize_name(value))
+
+
+def supplier_match_key(value: str) -> str:
+    key = normalized_key(value)
+    return re.sub(r"(佛山|广州|深圳|东莞|中山|珠海)市", r"\1", key)
+
+
+def resolve_canonical_name(name: str, purchase_order_names: list[str]) -> tuple[str, bool]:
+    """Resolve a short/raw name to one unambiguous procurement full name."""
+    key = supplier_match_key(name)
+    if not key:
+        return name, False
+    exact = [item for item in purchase_order_names if supplier_match_key(item) == key]
+    if len(exact) == 1:
+        return exact[0], True
+    contains = [item for item in purchase_order_names if key in supplier_match_key(item)]
+    if len(contains) == 1:
+        return contains[0], True
+    return name, False
+
+
+def infer_location(value: str) -> tuple[str | None, str | None]:
+    """Infer only explicit locations present in a supplier name."""
+    compact = compact_text(value)
+    city_map = {
+        "佛山": ("佛山", "广东"),
+        "广州": ("广州", "广东"),
+        "深圳": ("深圳", "广东"),
+        "东莞": ("东莞", "广东"),
+        "中山": ("中山", "广东"),
+        "珠海": ("珠海", "广东"),
+    }
+    for token, location in city_map.items():
+        if token in compact:
+            return location
+    return None, None
+
+
+def split_supplier_cell(value: str | None, *, material_type: str | None = None) -> list[SupplierMention]:
+    if not value or value.strip() in {"/", "-", "—", "无", "未知"}:
+        return []
+    text = value.strip()
+    mentions: list[SupplierMention] = []
+    current_role: str | None = None
+    current_material = material_type
+    for chunk in _SEPARATOR_RE.split(text):
+        chunk = chunk.strip()
+        if not chunk:
+            continue
+        bare_label = chunk.rstrip(" ：:")
+        if bare_label in {"内料", "包材", "包装", "瓶子", "泵头", "盖子", "软管", "彩盒", "物料", "加工厂", "供应商"}:
+            current_role = "加工厂" if "工厂" in bare_label else "供应商"
+            if bare_label not in {"加工厂", "供应商"}:
+                current_material = bare_label
+            continue
+        if chunk.endswith(("：", ":")) and bare_label:
+            current_role = "加工厂" if "工厂" in bare_label else "供应商"
+            material_label = re.sub(r"(?:加工厂|工厂|供应商)$", "", bare_label)
+            if material_label:
+                current_material = material_label
+            continue
+        labels = list(_LABEL_RE.finditer(chunk))
+        if labels:
+            label = labels[-1].group("label")
+            current_role = "加工厂" if "工厂" in label else "供应商"
+            material_label = re.sub(r"(?:加工厂|工厂|供应商)$", "", label)
+            if material_label:
+                current_material = material_label
+            chunk = chunk[labels[-1].end() :].strip()
+        # A comma can separate material labels from the actual name.
+        if not chunk or chunk in {"/", "-", "—"}:
+            continue
+        name = normalize_name(chunk)
+        if not name:
+            continue
+        mentions.append(
+            SupplierMention(
+                name=name,
+                role=current_role,
+                material_type=current_material,
+                source_text=chunk,
+                method="delimiter" if len(_SEPARATOR_RE.findall(text)) else "raw",
+                confidence=0.95 if len(_SEPARATOR_RE.findall(text)) else 0.75,
+            )
+        )
+    return mentions
diff --git a/app/services/supplier_identity.py b/app/services/supplier_identity.py
new file mode 100644
index 0000000..b781a1a
--- /dev/null
+++ b/app/services/supplier_identity.py
@@ -0,0 +1,44 @@
+"""Resolve supplier aliases to canonical master-data identities."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from sqlalchemy import select
+from sqlalchemy.exc import ProgrammingError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.entities import SupplierMaster
+from app.services.supplier_cleaning import supplier_match_key
+
+
+@dataclass(frozen=True, slots=True)
+class SupplierIdentity:
+    id: int
+    canonical_name: str
+    short_name: str | None
+    aliases: tuple[str, ...]
+
+    @property
+    def match_names(self) -> tuple[str, ...]:
+        return tuple(dict.fromkeys(filter(None, (self.short_name, *self.aliases, self.canonical_name))))
+
+
+async def resolve_supplier(session: AsyncSession, value: str | None) -> SupplierIdentity | None:
+    if not value or not isinstance(session, AsyncSession):
+        return None
+    key = supplier_match_key(value)
+    try:
+        async with session.begin_nested():
+            rows = (await session.execute(select(SupplierMaster).where(SupplierMaster.status == "启用"))).scalars().all()
+    except ProgrammingError:
+        return None
+    matches = []
+    for row in rows:
+        names = (row.canonical_name, row.short_name, *(row.aliases or []))
+        if any(name and supplier_match_key(name) == key for name in names):
+            matches.append(row)
+    if len(matches) != 1:
+        return None
+    row = matches[0]
+    return SupplierIdentity(row.id, row.canonical_name, row.short_name, tuple(row.aliases or []))
diff --git a/app/services/suppliers.py b/app/services/suppliers.py
index 49fc95f..83b2ca8 100644
--- a/app/services/suppliers.py
+++ b/app/services/suppliers.py
@@ -6,11 +6,12 @@ import re
 from collections import Counter, defaultdict
 from datetime import UTC, date, datetime
 
-from sqlalchemy import func, select
+from sqlalchemy import String, func, or_, select
+from sqlalchemy.exc import ProgrammingError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.exceptions import AppException
-from app.models.entities import AnomalyCase
+from app.models.entities import AnomalyCase, AnomalyCaseSupplier, SupplierMaster
 from app.schemas.suppliers import (
     SupplierCategoryBreakdown,
     SupplierListData,
@@ -34,6 +35,9 @@ class SupplierProfileService:
         self.as_of = as_of or datetime.now(UTC).date()
 
     async def list(self, *, keyword: str | None = None, limit: int = 100) -> SupplierListData:
+        canonical = await self._load_master_summary(keyword=keyword, limit=limit)
+        if canonical:
+            return SupplierListData(items=canonical, default_supplier=canonical[0].name)
         grouped = await self._load_supplier_summary(keyword=keyword)
         # Some imported legacy rows have no anomaly date.  They are excluded
         # from the normal three-year risk window, but must still be discoverable
@@ -63,8 +67,9 @@ class SupplierProfileService:
         )
 
     async def detail(self, supplier_name: str) -> SupplierProfileData:
-        all_rows = await self._load_rows()
-        rows = [row for row in all_rows if self.matches(row.get("supplier"), supplier_name)]
+        linked_rows = await self._load_linked_rows(supplier_name)
+        all_rows = linked_rows or await self._load_rows()
+        rows = linked_rows or [row for row in all_rows if self.matches(row.get("supplier"), supplier_name)]
         used_legacy_fallback = False
         if not rows:
             all_rows = await self._load_rows(include_undated=True)
@@ -119,9 +124,7 @@ class SupplierProfileService:
                         cap=scoring.frequency_cap,
                     )
                 ),
-                supplier_base_score=(
-                    supplier_dimension.base_score if supplier_dimension else None
-                ),
+                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),
@@ -134,10 +137,74 @@ class SupplierProfileService:
                 "该供应商仅命中历史导入记录，部分异常缺少发生日期；本页展示可识别的历史记录，"
                 "不将日期缺失数据混入近3年趋势。"
                 if used_legacy_fallback
-                else "合作次数按自然年度边界采购订单去重单据编号统计；异常数和供应商层级仍按异常案例库统计。"
+                else "供应商画像按标准供应商关联统计；合作次数按采购订单去重单据编号统计。"
             ),
         )
 
+    async def _load_master_summary(self, *, keyword: str | None, limit: int) -> list[SupplierOption]:
+        filters = [SupplierMaster.status == "启用"]
+        if keyword:
+            pattern = f"%{keyword.strip()}%"
+            filters.append(or_(
+                SupplierMaster.canonical_name.ilike(pattern),
+                SupplierMaster.short_name.ilike(pattern),
+                SupplierMaster.city.ilike(pattern),
+                SupplierMaster.province.ilike(pattern),
+                SupplierMaster.aliases.cast(String).ilike(pattern),
+            ))
+        statement = (
+            select(SupplierMaster, func.count(AnomalyCaseSupplier.id).label("anomaly_count"))
+            .outerjoin(AnomalyCaseSupplier, AnomalyCaseSupplier.supplier_id == SupplierMaster.id)
+            .where(*filters)
+            .group_by(SupplierMaster.id)
+            .order_by(func.count(AnomalyCaseSupplier.id).desc(), SupplierMaster.canonical_name)
+            .limit(limit)
+        )
+        try:
+            async with self.session.begin_nested():
+                rows = (await self.session.execute(statement)).all()
+        except ProgrammingError:
+            # Allows a rolling deployment: the legacy raw supplier search stays
+            # available until the privileged migration has created master tables.
+            return []
+        return [SupplierOption(
+            name=master.canonical_name,
+            anomaly_count=int(count or 0),
+            tier="待确认" if master.review_status != "已确认" else self.tier_for(int(count or 0), 0, 0),
+            city=master.city,
+            province=master.province,
+            short_name=master.short_name,
+            review_status=master.review_status,
+        ) for master, count in rows]
+
+    async def _load_linked_rows(self, supplier_name: str) -> list[dict]:
+        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,
+            )
+            .join(AnomalyCaseSupplier, AnomalyCaseSupplier.case_id == AnomalyCase.id)
+            .join(SupplierMaster, SupplierMaster.id == AnomalyCaseSupplier.supplier_id)
+            .where(or_(
+                SupplierMaster.canonical_name == supplier_name,
+                SupplierMaster.short_name == supplier_name,
+                SupplierMaster.aliases.cast(String).ilike(f"%{supplier_name}%"),
+            ))
+        )
+        try:
+            async with self.session.begin_nested():
+                result = await self.session.execute(statement)
+                return [dict(row) for row in result.mappings().all()]
+        except ProgrammingError:
+            return []
+
     async def _load_supplier_summary(
         self,
         *,
diff --git a/migrations/versions/20260910_0019_supplier_master_and_links.py b/migrations/versions/20260910_0019_supplier_master_and_links.py
new file mode 100644
index 0000000..aff58f3
--- /dev/null
+++ b/migrations/versions/20260910_0019_supplier_master_and_links.py
@@ -0,0 +1,62 @@
+"""Add canonical supplier master data and anomaly relationships."""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects.postgresql import JSONB
+
+revision: str = "20260910_0019"
+down_revision: str | None = "20260908_0018"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "supplier_master",
+        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
+        sa.Column("canonical_name", sa.String(length=200), nullable=False),
+        sa.Column("short_name", sa.String(length=100), nullable=True),
+        sa.Column("normalized_name", sa.String(length=200), nullable=False),
+        sa.Column("city", sa.String(length=50), nullable=True),
+        sa.Column("province", sa.String(length=50), nullable=True),
+        sa.Column("supplier_type", sa.String(length=50), nullable=True),
+        sa.Column("aliases", JSONB(), server_default="[]", nullable=False),
+        sa.Column("status", sa.String(length=20), server_default="启用", nullable=False),
+        sa.Column("source", sa.String(length=100), nullable=True),
+        sa.Column("review_status", sa.String(length=20), server_default="待确认", nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index("ix_supplier_master_city", "supplier_master", ["city"])
+    op.create_index("ix_supplier_master_normalized_name", "supplier_master", ["normalized_name"], unique=True)
+    op.create_table(
+        "anomaly_case_suppliers",
+        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
+        sa.Column("case_id", sa.BigInteger(), nullable=False),
+        sa.Column("supplier_id", sa.BigInteger(), nullable=False),
+        sa.Column("supplier_role", sa.String(length=50), nullable=True),
+        sa.Column("material_type", sa.String(length=200), nullable=True),
+        sa.Column("source_text", sa.Text(), nullable=True),
+        sa.Column("match_method", sa.String(length=30), server_default="manual", nullable=False),
+        sa.Column("match_confidence", sa.Numeric(5, 4), nullable=True),
+        sa.Column("review_status", sa.String(length=20), server_default="待确认", nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(["case_id"], ["anomaly_cases.id"], ondelete="CASCADE"),
+        sa.ForeignKeyConstraint(["supplier_id"], ["supplier_master.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("case_id", "supplier_id", "supplier_role", name="uq_case_supplier_role"),
+    )
+    op.create_index("ix_case_suppliers_supplier", "anomaly_case_suppliers", ["supplier_id"])
+    op.create_index("ix_case_suppliers_case", "anomaly_case_suppliers", ["case_id"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_case_suppliers_case", table_name="anomaly_case_suppliers")
+    op.drop_index("ix_case_suppliers_supplier", table_name="anomaly_case_suppliers")
+    op.drop_table("anomaly_case_suppliers")
+    op.drop_index("ix_supplier_master_normalized_name", table_name="supplier_master")
+    op.drop_index("ix_supplier_master_city", table_name="supplier_master")
+    op.drop_table("supplier_master")
diff --git a/scripts/backfill_embeddings.py b/scripts/backfill_embeddings.py
index 2da7356..27d4351 100644
--- a/scripts/backfill_embeddings.py
+++ b/scripts/backfill_embeddings.py
@@ -28,7 +28,7 @@ def _case_text(row: dict[str, Any]) -> str:
         ("产品", row.get("product_name")),
         ("品类", row.get("product_category")),
         ("包材", row.get("material_type")),
-        ("供应商", row.get("supplier")),
+        ("标准供应商", row.get("canonical_supplier") or row.get("supplier")),
         ("加工厂", row.get("factory")),
         ("阶段", row.get("stage_new")),
         ("异常类型", row.get("anomaly_type")),
@@ -47,11 +47,19 @@ async def _embedding(
     model: str,
 ) -> list[list[float]]:
     for attempt in range(4):
-        response = await client.post(
-            endpoint,
-            headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
-            json={"model": model, "input": texts},
-        )
+        try:
+            response = await client.post(
+                endpoint,
+                headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
+                json={"model": model, "input": texts},
+            )
+        except httpx.RequestError:
+            if attempt == 3:
+                raise
+            delay = 2.0 * (2**attempt) + random.uniform(0, 1)
+            logger.warning("Embedding 网络异常 attempt=%s delay=%.1fs", attempt + 1, delay)
+            await asyncio.sleep(delay)
+            continue
         if response.status_code not in {429, 500, 502, 503, 504}:
             response.raise_for_status()
             data = response.json().get("data") or []
@@ -75,7 +83,7 @@ async def _embedding(
     raise RuntimeError("Embedding 请求重试次数已用尽")
 
 
-async def main(batch_size: int) -> None:
+async def main(batch_size: int, *, force: bool = False) -> None:
     settings = get_settings()
     api_key = (
         settings.rag_embedding_api_key.get_secret_value()
@@ -108,21 +116,26 @@ async def main(batch_size: int) -> None:
             await session.execute(
                 text(
                     """
-                    SELECT id, product_name, product_category, material_type, supplier,
-                           factory, stage_new, anomaly_type, description, root_cause, solution
-                    FROM anomaly_cases
-                    WHERE embedding_vector IS NULL
+                    SELECT ac.id, ac.product_name, ac.product_category, ac.material_type, ac.supplier,
+                           ac.factory, ac.stage_new, ac.anomaly_type, ac.description, ac.root_cause, ac.solution,
+                           string_agg(DISTINCT sm.canonical_name, ' / ') AS canonical_supplier
+                    FROM anomaly_cases ac
+                    LEFT JOIN anomaly_case_suppliers acs ON acs.case_id = ac.id
+                    LEFT JOIN supplier_master sm ON sm.id = acs.supplier_id
+                    WHERE (:force OR ac.embedding_vector IS NULL)
+                    GROUP BY ac.id
                     ORDER BY id
                     """
-                )
+                ),
+                {"force": force},
             )
         ).mappings().all()
 
     logger.info("待生成案例向量数量=%s", len(rows))
     print(f"待生成案例向量数量：{len(rows)}，批次大小：{batch_size}", flush=True)
     timeout = httpx.Timeout(
-        settings.rag_embedding_timeout_seconds,
-        connect=min(5, settings.rag_embedding_timeout_seconds),
+        max(30, settings.rag_embedding_timeout_seconds),
+        connect=min(10, max(30, settings.rag_embedding_timeout_seconds)),
     )
     async with httpx.AsyncClient(timeout=timeout) as client:
         success_count = 0
@@ -163,13 +176,16 @@ async def main(batch_size: int) -> None:
                 f"成功：{success_count}，失败：{failure_count}，当前案例：{batch[-1]['id']}",
                 flush=True,
             )
-            await asyncio.sleep(1.5)
+            # The provider already returns 429 with Retry-After when throttled;
+            # keep only a small pacing gap between successful batches.
+            await asyncio.sleep(0.2)
 
 
 if __name__ == "__main__":
     parser = argparse.ArgumentParser()
     parser.add_argument("--batch-size", type=int, default=1)
+    parser.add_argument("--force", action="store_true", help="recompute existing vectors after source normalization")
     args = parser.parse_args()
-    if not 1 <= args.batch_size <= 8:
-        raise SystemExit("--batch-size 必须在 1 到 8 之间")
-    asyncio.run(main(args.batch_size))
+    if not 1 <= args.batch_size <= 64:
+        raise SystemExit("--batch-size 必须在 1 到 64 之间")
+    asyncio.run(main(args.batch_size, force=args.force))
diff --git a/scripts/clean_suppliers.py b/scripts/clean_suppliers.py
new file mode 100644
index 0000000..8370ce2
--- /dev/null
+++ b/scripts/clean_suppliers.py
@@ -0,0 +1,128 @@
+"""Build canonical supplier entities and anomaly links from raw anomaly data.
+
+The script is intentionally additive: it never overwrites anomaly_cases.supplier.
+Run after ``alembic upgrade head``.  Existing master rows are reused by normalized
+name; low-confidence mentions remain marked ``待确认`` for manual review.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+from collections import defaultdict
+
+from sqlalchemy import delete, distinct, select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+
+from app.core.config import get_settings
+from app.models.entities import AnomalyCase, AnomalyCaseSupplier, PurchaseOrderLine, SupplierMaster
+from app.services.supplier_cleaning import infer_location, normalized_key, resolve_canonical_name, split_supplier_cell
+
+
+async def clean(*, rebuild: bool = False) -> dict[str, int]:
+    settings = get_settings()
+    engine = create_async_engine(settings.database_url, pool_pre_ping=True)
+    factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+    stats = defaultdict(int)
+    try:
+        async with factory() as session:
+            if rebuild:
+                await session.execute(delete(AnomalyCaseSupplier))
+                await session.execute(delete(SupplierMaster))
+                stats["derived_tables_rebuilt"] = 1
+            masters = {row.normalized_name: row for row in (await session.execute(select(SupplierMaster))).scalars()}
+            order_names = [
+                value.strip()
+                for value in (await session.execute(select(distinct(PurchaseOrderLine.supplier)))).scalars()
+                if value and value.strip()
+            ]
+            cases = (await session.execute(select(AnomalyCase))).scalars().all()
+            parsed: list[tuple[AnomalyCase, object]] = []
+            for case in cases:
+                mentions = split_supplier_cell(case.supplier, material_type=case.material_type)
+                for mention in mentions:
+                    canonical_name, confirmed = resolve_canonical_name(mention.name, order_names)
+                    key = normalized_key(canonical_name)
+                    master = masters.get(key)
+                    if master is None:
+                        city, province = infer_location(canonical_name)
+                        master = SupplierMaster(
+                            canonical_name=canonical_name,
+                            short_name=mention.name,
+                            normalized_name=key,
+                            city=city,
+                            province=province,
+                            supplier_type="供应商",
+                            aliases=[mention.name],
+                            source="采购订单匹配" if confirmed else "异常数据清洗",
+                            review_status="已确认" if confirmed else "待确认",
+                        )
+                        session.add(master)
+                        masters[key] = master
+                        stats["masters_created"] += 1
+                        if confirmed:
+                            stats["masters_confirmed_from_orders"] += 1
+                    elif mention.name not in master.aliases:
+                        master.aliases = [*master.aliases, mention.name]
+                        if len(mention.name) < len(master.short_name or master.canonical_name):
+                            master.short_name = mention.name
+                    parsed.append((case, mention))
+                stats["cases_scanned"] += 1
+
+            # Assign all master IDs in one database round trip.
+            await session.flush()
+            existing_links = {
+                (case_id, supplier_id, role)
+                for case_id, supplier_id, role in (
+                    await session.execute(
+                        select(
+                            AnomalyCaseSupplier.case_id,
+                            AnomalyCaseSupplier.supplier_id,
+                            AnomalyCaseSupplier.supplier_role,
+                        )
+                    )
+                ).all()
+            }
+            pending_links: set[tuple[int, int, str | None]] = set()
+            for case, mention in parsed:
+                canonical_name, _ = resolve_canonical_name(mention.name, order_names)
+                master = masters[normalized_key(canonical_name)]
+                link_key = (case.id, master.id, mention.role)
+                if link_key in existing_links or link_key in pending_links:
+                    stats["links_skipped"] += 1
+                    continue
+                session.add(
+                    AnomalyCaseSupplier(
+                        case_id=case.id,
+                        supplier_id=master.id,
+                        supplier_role=mention.role,
+                        material_type=mention.material_type,
+                        source_text=mention.source_text,
+                        match_method=mention.method,
+                        match_confidence=mention.confidence,
+                        review_status="待确认",
+                    )
+                )
+                pending_links.add(link_key)
+                stats["links_created"] += 1
+            await session.commit()
+    finally:
+        await engine.dispose()
+    return dict(stats)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--rebuild",
+        action="store_true",
+        help="clear and rebuild only the derived supplier master/link tables",
+    )
+    args = parser.parse_args()
+    result = asyncio.run(clean(rebuild=args.rebuild))
+    for key, value in result.items():
+        print(f"{key}={value}")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/test_models.py b/tests/test_models.py
index eacd546..4f2bcfa 100644
--- a/tests/test_models.py
+++ b/tests/test_models.py
@@ -19,6 +19,7 @@ from sqlalchemy import BigInteger
 def test_prd_core_tables_are_registered() -> None:
     assert set(Base.metadata.tables) == {
         "anomaly_cases",
+        "anomaly_case_suppliers",
         "conversations",
         "conversation_messages",
         "operation_audits",
@@ -30,6 +31,7 @@ def test_prd_core_tables_are_registered() -> None:
         "role_permissions",
         "standard_rules",
         "supplier_profiles",
+        "supplier_master",
         "user_accounts",
     }
 
diff --git a/tests/test_supplier_cleaning.py b/tests/test_supplier_cleaning.py
new file mode 100644
index 0000000..a146840
--- /dev/null
+++ b/tests/test_supplier_cleaning.py
@@ -0,0 +1,29 @@
+from app.services.supplier_cleaning import infer_location, normalized_key, resolve_canonical_name, split_supplier_cell
+
+
+def test_split_supplier_cell_preserves_multiple_supplier_mentions() -> None:
+    mentions = split_supplier_cell("内料：博元\n瓶子，盖子：皓盛/佛山执得")
+    assert [item.name for item in mentions] == ["博元", "皓盛", "佛山执得"]
+    assert mentions[0].material_type == "内料"
+    assert mentions[1].material_type == "盖子"
+
+
+def test_empty_supplier_markers_are_ignored() -> None:
+    assert split_supplier_cell("/") == []
+    assert normalized_key("佛山市执得包装有限公司") == "佛山市执得包装"
+    assert infer_location("佛山市执得包装有限公司") == ("佛山", "广东")
+
+
+def test_purchase_order_full_name_wins_when_short_name_is_unambiguous() -> None:
+    names = ["佛山市佳亿包装有限公司", "广州祥印包装有限公司"]
+    assert resolve_canonical_name("佳亿", names) == ("佛山市佳亿包装有限公司", True)
+    assert resolve_canonical_name("佛山执得", ["佛山市执得塑料制品有限公司"]) == (
+        "佛山市执得塑料制品有限公司",
+        True,
+    )
+
+
+def test_material_and_factory_labels_are_not_supplier_names() -> None:
+    mentions = split_supplier_cell("内料工厂：众绘\n软管工厂：永禾\n彩盒工厂：佳亿")
+    assert [item.name for item in mentions] == ["众绘", "永禾", "佳亿"]
+    assert all(item.role == "加工厂" for item in mentions)
diff --git a/tests/test_suppliers.py b/tests/test_suppliers.py
index fa9f89f..89cd863 100644
--- a/tests/test_suppliers.py
+++ b/tests/test_suppliers.py
@@ -29,6 +29,9 @@ def test_supplier_list_falls_back_to_undated_legacy_rows(monkeypatch) -> None:
     service = SupplierProfileService.__new__(SupplierProfileService)
     calls = []
 
+    async def fake_master_summary(*, keyword=None, limit=100):
+        return []
+
     async def fake_summary(*, keyword=None, include_undated=False):
         calls.append(include_undated)
         if not include_undated:
@@ -36,6 +39,7 @@ def test_supplier_list_falls_back_to_undated_legacy_rows(monkeypatch) -> None:
         return {"皓盛/佛山执得": {"total": 1, "a": 0, "b": 0}}
 
     service._load_supplier_summary = fake_summary
+    service._load_master_summary = fake_master_summary
     result = __import__("asyncio").run(service.list(keyword="佛山市", limit=8))
     assert calls == [False, True]
     assert result.items[0].name == "皓盛/佛山执得"