修改数据同步任务

supply_infra/db/models/global_tree_category.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from datetime import datetime
 
-from sqlalchemy import BigInteger, Integer, String, Text, func
+from sqlalchemy import BigInteger, Integer, String, Text, UniqueConstraint, func
 from sqlalchemy.orm import Mapped, mapped_column
 
 from supply_infra.db.base import Base
@@ -12,12 +12,15 @@ class GlobalTreeCategory(Base):
     """全局树分类 — 从 ODPS public_pattern_mining_category 同步。"""
 
     __tablename__ = "global_tree_category"
+    __table_args__ = (UniqueConstraint("source_id", name="uk_source_id"),)
 
     id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
     name: Mapped[str | None] = mapped_column(String(128), nullable=True, comment="名称")
     description: Mapped[str | None] = mapped_column(Text, nullable=True, comment="描述")
     level: Mapped[int | None] = mapped_column(Integer, nullable=True, comment="层级")
     parent_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True, comment="父节点")
+    source_id: Mapped[int] = mapped_column(BigInteger, nullable=False, comment="hive表id")
+    source_parent_id: Mapped[int] = mapped_column(BigInteger, nullable=False, comment="hive表父id")
     is_delete: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="是否删除0-正常 1-删除")
     create_time: Mapped[datetime | None] = mapped_column(
         nullable=True,

supply_infra/db/models/global_tree_element.py

@@ -20,6 +20,7 @@ class GlobalTreeElement(Base):
     id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
     name: Mapped[str | None] = mapped_column(String(128), nullable=True, comment="名称")
     category_id: Mapped[int] = mapped_column(BigInteger, nullable=False, comment="分类id")
+    source_category_id: Mapped[int] = mapped_column(BigInteger, nullable=False, comment="hive表category_id")
     is_delete: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="是否删除 0-正常 1-删除")
     create_time: Mapped[datetime] = mapped_column(
         nullable=False,

supply_infra/db/repositories/global_tree_category_repo.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from sqlalchemy import func, select
 from sqlalchemy.dialects.mysql import insert
 
 from supply_infra.db.models.global_tree_category import GlobalTreeCategory
@@ -9,12 +10,22 @@ _BATCH_SIZE = 1000
 
 
 class GlobalTreeCategoryRepository(BaseRepository[GlobalTreeCategory]):
-    """全局树分类表 — INSERT IGNORE 按 id 主键去重。"""
+    """全局树分类表 — 按 source_id 去重后增量写入。"""
 
     model = GlobalTreeCategory
 
+    def get_source_id_map(self) -> dict[int, int]:
+        """返回全量 source_id → MySQL id 映射，用于增量去重与 parent 关联。"""
+        stmt = select(GlobalTreeCategory.source_id, GlobalTreeCategory.id)
+        return {int(row.source_id): int(row.id) for row in self.session.execute(stmt).all()}
+
+    def get_max_id(self) -> int:
+        """返回当前表内最大 id，用于分配新 id（兼容无 source_id 的历史数据）。"""
+        max_id = self.session.scalar(select(func.max(GlobalTreeCategory.id)))
+        return int(max_id) if max_id is not None else 0
+
     def bulk_insert_ignore(self, rows: list[dict]) -> int:
-        """批量插入，MySQL 自动忽略 id 重复行。"""
+        """批量插入，MySQL 按 source_id 唯一索引忽略已存在行（历史数据不变）。"""
         if not rows:
             return 0
 

supply_infra/scheduler/jobs/sync_global_tree_odps_to_mysql.py

@@ -5,7 +5,9 @@
 1. 拉取 pattern_mining_element（name, category_id）
 2. 拉取 public_pattern_mining_category 全量分类
 3. 筛选元素关联的分类，并沿 parent_id 向上追溯至顶层
-4. 分别批量 INSERT IGNORE 写入 global_tree_element / global_tree_category
+4. 查询 MySQL 已有分类，按 source_id 去重（仅插入新增）
+5. 为新分类分配 MySQL id/parent_id，INSERT IGNORE 写入（历史分类不变）
+6. 重载 source_id 映射，将元素写入 global_tree_element（INSERT IGNORE 去重）
 """
 from __future__ import annotations
 
@@ -46,30 +48,141 @@ def _collect_categories_with_ancestors(
     return list(collected.values())
 
 
-def _to_element_rows(elements: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    return [
-        {"name": str(row["name"]), "category_id": int(row["category_id"])}
-        for row in elements
-        if row.get("name") is not None and row.get("category_id") is not None
-    ]
+def _dedupe_category_source_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """按 source_id 去重，保留最后一条（同日 ODPS 数据可能重复）。"""
+    by_source_id: dict[int, dict[str, Any]] = {}
+    for row in rows:
+        by_source_id[int(row["source_id"])] = row
+    return list(by_source_id.values())
 
 
-def _to_category_rows(categories: list[dict[str, Any]]) -> list[dict[str, Any]]:
+def _dedupe_element_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """按 (name, category_id) 去重，与表唯一键一致。"""
+    seen: set[tuple[str, int]] = set()
+    result: list[dict[str, Any]] = []
+    for row in rows:
+        key = (row["name"], row["category_id"])
+        if key in seen:
+            continue
+        seen.add(key)
+        result.append(row)
+    return result
+
+
+def _to_category_source_rows(categories: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """将 ODPS 分类转为含 source_id / source_parent_id 的中间结构。"""
     rows: list[dict[str, Any]] = []
     for cat in categories:
         parent_id = cat.get("parent_id")
         rows.append(
             {
-                "id": int(cat["id"]),
+                "source_id": int(cat["id"]),
+                "source_parent_id": int(parent_id) if parent_id is not None else 0,
                 "name": str(cat["name"]) if cat.get("name") is not None else None,
                 "description": cat.get("description"),
                 "level": int(cat["level"]) if cat.get("level") is not None else None,
-                "parent_id": int(parent_id) if parent_id is not None else None,
             }
         )
     return rows
 
 
+def _prepare_new_category_rows(
+    categories: list[dict[str, Any]],
+    source_id_to_mysql_id: dict[int, int],
+    next_id: int,
+) -> tuple[list[dict[str, Any]], dict[int, int], int]:
+    """
+    过滤已存在 source_id，多轮扫描为新分类分配 MySQL id 与 parent_id。
+
+    按 level 初排后多轮插入：仅当父节点已在映射中（库内已有或本轮已分配）才写入，
+    避免同批次内因 level 数据异常导致子节点被误跳过。
+
+    Returns:
+        (待插入行, 更新后的 source_id → mysql_id 映射, 下一可用 id)
+    """
+    existing_source_ids = set(source_id_to_mysql_id.keys())
+    pending = [c for c in categories if int(c["source_id"]) not in existing_source_ids]
+    pending.sort(key=lambda c: (c.get("level") or 0, int(c["source_id"])))
+
+    updated_map = dict(source_id_to_mysql_id)
+    rows: list[dict[str, Any]] = []
+
+    while pending:
+        still_pending: list[dict[str, Any]] = []
+        progress = False
+
+        for cat in pending:
+            source_id = int(cat["source_id"])
+            source_parent_id = int(cat["source_parent_id"])
+
+            if source_parent_id != 0 and source_parent_id not in updated_map:
+                still_pending.append(cat)
+                continue
+
+            mysql_parent_id: int | None = None
+            if source_parent_id != 0:
+                mysql_parent_id = updated_map[source_parent_id]
+
+            updated_map[source_id] = next_id
+            rows.append(
+                {
+                    "id": next_id,
+                    "name": cat["name"],
+                    "description": cat.get("description"),
+                    "level": cat.get("level"),
+                    "parent_id": mysql_parent_id,
+                    "source_id": source_id,
+                    "source_parent_id": source_parent_id,
+                }
+            )
+            next_id += 1
+            progress = True
+
+        if not progress and still_pending:
+            for cat in still_pending:
+                logger.warning(
+                    "Parent source_id=%s not found for category source_id=%s, skip",
+                    cat["source_parent_id"],
+                    cat["source_id"],
+                )
+            break
+
+        pending = still_pending
+
+    return rows, updated_map, next_id
+
+
+def _to_element_rows(
+    elements: list[dict[str, Any]],
+    source_id_to_mysql_id: dict[int, int],
+) -> list[dict[str, Any]]:
+    """将 ODPS 元素的 hive category_id 映射为 MySQL 分类 id。"""
+    rows: list[dict[str, Any]] = []
+    for row in elements:
+        if row.get("name") is None or row.get("category_id") is None:
+            continue
+
+        source_category_id = int(row["category_id"])
+        mysql_category_id = source_id_to_mysql_id.get(source_category_id)
+        if mysql_category_id is None:
+            logger.warning(
+                "Category source_id=%s not found for element %s, skip",
+                source_category_id,
+                row["name"],
+            )
+            continue
+
+        rows.append(
+            {
+                "name": str(row["name"]),
+                "category_id": mysql_category_id,
+                "source_category_id": source_category_id,
+            }
+        )
+
+    return rows
+
+
 def sync_global_tree_odps_to_mysql(partition_date: str | None = None) -> dict:
     """
     从 ODPS 拉取全局树元素与分类，写入 MySQL。
@@ -91,31 +204,58 @@ def sync_global_tree_odps_to_mysql(partition_date: str | None = None) -> dict:
         len(raw_categories),
     )
 
-    elements = _to_element_rows(raw_elements)
     category_by_id = {int(c["id"]): c for c in raw_categories if c.get("id") is not None}
 
-    seed_ids = {row["category_id"] for row in elements}
+    seed_ids = {
+        int(row["category_id"])
+        for row in raw_elements
+        if row.get("category_id") is not None
+    }
     categories = _collect_categories_with_ancestors(seed_ids, category_by_id)
-    category_rows = _to_category_rows(categories)
+    category_source_rows = _dedupe_category_source_rows(_to_category_source_rows(categories))
 
     logger.info(
-        "Resolved %d elements and %d categories (with ancestors) for insert",
-        len(elements),
-        len(category_rows),
+        "Resolved %d elements and %d categories (with ancestors) for sync",
+        len(raw_elements),
+        len(category_source_rows),
     )
 
     with get_session() as session:
         element_repo = GlobalTreeElementRepository(session)
         category_repo = GlobalTreeCategoryRepository(session)
+
+        source_id_to_mysql_id = category_repo.get_source_id_map()
+        logger.info("Loaded %d existing categories from MySQL", len(source_id_to_mysql_id))
+
+        existing_source_ids = set(source_id_to_mysql_id.keys())
+        categories_already_exist = sum(
+            1 for c in category_source_rows if int(c["source_id"]) in existing_source_ids
+        )
+
+        next_id = category_repo.get_max_id() + 1
+        new_category_rows, _, _ = _prepare_new_category_rows(
+            category_source_rows,
+            source_id_to_mysql_id,
+            next_id,
+        )
+        category_inserted = category_repo.bulk_insert_ignore(new_category_rows)
+
+        # 插入后从 DB 重载映射，确保元素 category_id 与库内真实 id 一致
+        source_id_to_mysql_id = category_repo.get_source_id_map()
+        elements = _dedupe_element_rows(_to_element_rows(raw_elements, source_id_to_mysql_id))
         element_inserted = element_repo.bulk_insert_ignore(elements)
-        category_inserted = category_repo.bulk_insert_ignore(category_rows)
+
+    categories_failed = len(category_source_rows) - categories_already_exist - len(new_category_rows)
 
     result = {
         "partition_date": partition_date,
-        "elements_fetched": len(elements),
+        "elements_fetched": len(raw_elements),
         "elements_inserted": element_inserted,
-        "categories_fetched": len(category_rows),
+        "elements_skipped": len(raw_elements) - len(elements),
+        "categories_fetched": len(category_source_rows),
         "categories_inserted": category_inserted,
+        "categories_already_exist": categories_already_exist,
+        "categories_failed": categories_failed,
         "synced_at": datetime.now().isoformat(),
     }
     logger.info("Global tree sync completed: %s", result)