feat(store_mapping): insert→upsert 填空合并, 一 trace 一行多平台渐进填充

store_mapping 的语义是"一次比价(一个 trace)= 一个真实店铺一行", 各目标平台解析出自己的 shopId 就往同一行填自己那几列: 淘宝腿先建行(id_taobao/name_taobao), 将来京东腿解析出 id 再合并进同一行(id_jd/name_jd)。trace_id 唯一约束天然就是这一行的合并键。原 insert 是"撞 trace 就跳过", 第二条腿来上报会被丢掉, 撑不起渐进填充。改成 upsert: - 不存在 → INSERT(同原行为) - 已存在 → 填空合并(fill-the-blanks): 只写该行当前为 NULL 的列, 绝不覆盖已有非空值。保证后到平台只填自己的列、动不了先到平台的数据; 共享列(geo/source/溯源)先到先得。 - 并发撞唯一约束 → 回滚后转走同一填空合并路径(不依赖 ON CONFLICT, PG/SQLite 都安全)。返回值 (created, row_id): created=1 新建 / 0 合并; 端点日志区分"新建/合并"并带上 jd 列。注: 一行里 id_taobao 与 id_jd 共存只是 name-match 置信度(两腿搜同一源店名各自匹配到某店), 非已核实同一实体; 作 append-only 原始资产留存, 精确匹配下游做。验证: TestClient 打真实 PG —— 淘宝建行(inserted=1) → 京东同 trace 合并(只 UPDATE id_jd/ name_jd, geohash 不进 SET) → 淘宝重报不同 id 不覆盖(无 UPDATE)。最终一行各列正确, 已清理。 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-06-13 21:53:42 +08:00
parent 4844e49161
commit 0860eed92b
3 changed files with 88 additions and 50 deletions
@@ -35,12 +35,12 @@ def report_store_mapping(
    x_internal_secret: Annotated[str | None, Header()] = None,
 ) -> StoreMappingOut:
    _check_secret(x_internal_secret)
-    inserted, row_id = repo.insert(db, payload)
+    created, row_id = repo.upsert(db, payload)
    logger.info(
-        "store_mapping trace=%s inserted=%d row_id=%s source=%s "
-        "taobao=(%s,%s) device=%s user=%s",
-        payload.trace_id, inserted, row_id, payload.source_platform,
-        payload.id_taobao, payload.name_taobao,
+        "store_mapping trace=%s %s row_id=%s source=%s "
+        "taobao=(%s,%s) jd=(%s,%s) device=%s user=%s",
+        payload.trace_id, "新建" if created else "合并", row_id, payload.source_platform,
+        payload.id_taobao, payload.name_taobao, payload.id_jd, payload.name_jd,
        payload.source_device_id, payload.source_user_id,
    )
-    return StoreMappingOut(inserted=inserted, row_id=row_id)
+    return StoreMappingOut(inserted=created, row_id=row_id)
@@ -1,8 +1,14 @@
-"""平台店铺映射落库:一次比价一行,trace_id 幂等。
+"""平台店铺映射落库:一次比价(trace)一行,各目标平台解析出 id 就 upsert 进同一行。

-幂等键 = trace_id。pricebot 重试 / replay 会重复上报同一 trace,这里先查该 trace 是否
-已存在,存在即跳过(返回已存在行 id)——跨方言(PG / SQLite dev)都安全,不依赖 ON CONFLICT。
-极端并发下若仍撞唯一约束,IntegrityError 回滚后再查一次返回已存在行(本就幂等,不重试)。
+合并键 = trace_id(唯一约束)。一次跨平台比价 = 一个 trace = 一个真实店铺:淘宝腿解析出
+shopId 先建行(填淘宝列),京东腿(将来)解析出 id 再 upsert 进**同一行**(填京东列)。
+
+合并策略 = 填空(fill-the-blanks):只写该行当前为 NULL 的列,绝不覆盖已有非空值。保证后到
+的平台只填自己那几列、动不了先到平台的数据;共享列(geo / source / 溯源)先到先得。
+不依赖 ON CONFLICT,跨方言(PG / SQLite dev)都安全;并发撞唯一约束则回滚后转走合并路径。
+
+⚠️ 一行里 id_taobao 与 id_jd 共存只表示"两条腿搜同一个源店名各自匹配到了某家店",是
+name-match 置信度、非已核实同一实体。作为 append-only 原始资产留存,跨平台精确匹配由下游做。
 """
 from __future__ import annotations

@@ -17,49 +23,80 @@ from app.schemas.store_mapping import StoreMappingIn

 logger = logging.getLogger("shagua.store_mapping")

+# upsert 填空时可写的列。不含: trace_id(合并键)/ business_type(非空默认)/
+# id(主键)/ created_at(server_default)。各平台只会带自己那几列非空, 其余为 None 不动。
+_MERGE_COLUMNS = (
+    "source_platform",
+    "id_taobao", "name_taobao", "id_meituan", "name_meituan", "id_jd", "name_jd",
+    "city", "geohash", "lng", "lat", "taobao_address",
+    "source_device_id", "source_user_id",
+    "taobao_share_url", "taobao_resolved_url", "taobao_deeplink",
+    "attrs",
+)

-def _existing_id(db: Session, trace_id: str) -> int | None:
+
+def _find(db: Session, trace_id: str) -> StoreMapping | None:
    return db.execute(
-        select(StoreMapping.id).where(StoreMapping.trace_id == trace_id)
+        select(StoreMapping).where(StoreMapping.trace_id == trace_id)
    ).scalar_one_or_none()


-def insert(db: Session, payload: StoreMappingIn) -> tuple[int, int | None]:
-    """写入一行映射,返回 (inserted, row_id)。inserted=1 新写入 / 0 幂等跳过。"""
-    existing = _existing_id(db, payload.trace_id)
-    if existing is not None:
-        return 0, existing
+def _merge_fill_blanks(existing: StoreMapping, payload: StoreMappingIn) -> list[str]:
+    """把 payload 里非空、且 existing 当前为 NULL 的列填进去。返回被填的列名(空=无变化)。"""
+    filled: list[str] = []
+    for col in _MERGE_COLUMNS:
+        new = getattr(payload, col)
+        if new is not None and getattr(existing, col) is None:
+            setattr(existing, col, new)
+            filled.append(col)
+    return filled

-    row = StoreMapping(
-        trace_id=payload.trace_id,
-        business_type=payload.business_type,
-        source_platform=payload.source_platform,
-        id_taobao=payload.id_taobao,
-        name_taobao=payload.name_taobao,
-        id_meituan=payload.id_meituan,
-        name_meituan=payload.name_meituan,
-        id_jd=payload.id_jd,
-        name_jd=payload.name_jd,
-        city=payload.city,
-        geohash=payload.geohash,
-        lng=payload.lng,
-        lat=payload.lat,
-        taobao_address=payload.taobao_address,
-        source_device_id=payload.source_device_id,
-        source_user_id=payload.source_user_id,
-        taobao_share_url=payload.taobao_share_url,
-        taobao_resolved_url=payload.taobao_resolved_url,
-        taobao_deeplink=payload.taobao_deeplink,
-        attrs=payload.attrs,
-    )
-    db.add(row)
-    try:
+
+def upsert(db: Session, payload: StoreMappingIn) -> tuple[int, int | None]:
+    """落一行跨平台店铺映射,返回 (created, row_id)。
+    created=1 新建该 trace 的行 / 0 合并进已存在行(填空,不覆盖)。"""
+    existing = _find(db, payload.trace_id)
+    if existing is None:
+        row = StoreMapping(
+            trace_id=payload.trace_id,
+            business_type=payload.business_type,
+            source_platform=payload.source_platform,
+            id_taobao=payload.id_taobao,
+            name_taobao=payload.name_taobao,
+            id_meituan=payload.id_meituan,
+            name_meituan=payload.name_meituan,
+            id_jd=payload.id_jd,
+            name_jd=payload.name_jd,
+            city=payload.city,
+            geohash=payload.geohash,
+            lng=payload.lng,
+            lat=payload.lat,
+            taobao_address=payload.taobao_address,
+            source_device_id=payload.source_device_id,
+            source_user_id=payload.source_user_id,
+            taobao_share_url=payload.taobao_share_url,
+            taobao_resolved_url=payload.taobao_resolved_url,
+            taobao_deeplink=payload.taobao_deeplink,
+            attrs=payload.attrs,
+        )
+        db.add(row)
+        try:
+            db.commit()
+        except IntegrityError:
+            # 并发: 另一个请求刚插了同 trace → 撞唯一约束。回滚后转合并路径填空。
+            db.rollback()
+            existing = _find(db, payload.trace_id)
+            if existing is None:
+                raise
+            logger.warning("store_mapping 并发冲突 trace=%s, 转填空合并", payload.trace_id)
+        else:
+            db.refresh(row)
+            return 1, row.id
+
+    # 已存在(或并发回退到此): 填空合并, 只写当前 NULL 的列
+    filled = _merge_fill_blanks(existing, payload)
+    if filled:
        db.commit()
-    except IntegrityError:
-        # 并发下另一个请求刚插了同 trace → 唯一约束撞了。回滚后返回已存在行(幂等)。
-        db.rollback()
-        logger.warning("store_mapping 并发幂等冲突 trace=%s,返回已存在行", payload.trace_id)
-        return 0, _existing_id(db, payload.trace_id)
-
-    db.refresh(row)
-    return 1, row.id
+        db.refresh(existing)
+        logger.info("store_mapping 合并 trace=%s 填列=%s", payload.trace_id, filled)
+    return 0, existing.id
@@ -43,7 +43,8 @@ class StoreMappingIn(BaseModel):


 class StoreMappingOut(BaseModel):
-    """上报结果。inserted=1 为新写入,0 为幂等跳过(trace 已存在);row_id 为该 trace 对应行 id。"""
+    """上报结果。inserted=1 为新建该 trace 行,0 为合并进已存在行(填空,不覆盖);
+    row_id 为该 trace 对应行 id。"""

    inserted: int
    row_id: int | None = None