From 0860eed92b01ad3a8298ff354fe891cf7db8c392 Mon Sep 17 00:00:00 2001 From: marco Date: Sat, 13 Jun 2026 21:53:42 +0800 Subject: [PATCH] =?UTF-8?q?feat(store=5Fmapping):=20insert=E2=86=92upsert?= =?UTF-8?q?=20=E5=A1=AB=E7=A9=BA=E5=90=88=E5=B9=B6,=20=E4=B8=80=20trace=20?= =?UTF-8?q?=E4=B8=80=E8=A1=8C=E5=A4=9A=E5=B9=B3=E5=8F=B0=E6=B8=90=E8=BF=9B?= =?UTF-8?q?=E5=A1=AB=E5=85=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- app/api/internal/store.py | 12 +-- app/repositories/store_mapping.py | 123 +++++++++++++++++++----------- app/schemas/store_mapping.py | 3 +- 3 files changed, 88 insertions(+), 50 deletions(-) diff --git a/app/api/internal/store.py b/app/api/internal/store.py index 91ef749..d220925 100644 --- a/app/api/internal/store.py +++ b/app/api/internal/store.py @@ -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) diff --git a/app/repositories/store_mapping.py b/app/repositories/store_mapping.py index f6566e8..b9b9472 100644 --- a/app/repositories/store_mapping.py +++ b/app/repositories/store_mapping.py @@ -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 diff --git a/app/schemas/store_mapping.py b/app/schemas/store_mapping.py index 4143fe2..0b448c5 100644 --- a/app/schemas/store_mapping.py +++ b/app/schemas/store_mapping.py @@ -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