"""平台店铺映射落库:一次比价(trace)一行,各目标平台解析出 id 就 upsert 进同一行。 合并键 = 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 import logging import math from sqlalchemy import func, select, update from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.models.store_mapping import StoreMapping from app.schemas.store_mapping import StoreMappingIn logger = logging.getLogger("shagua.store_mapping") # 源平台 → 该平台店名所在列(缓存查询的匹配键)。镜像 pricebot reporter 的同名表。 _SOURCE_NAME_COLUMN = { "meituan": "name_meituan", "meituan_waimai": "name_meituan", "jd_waimai": "name_jd", "jd_waimai_standalone": "name_jd", "taobao_flash": "name_taobao", } # 源平台 → 它自己对应的目标 key(查缓存时排除"源平台自己", 不会复用源平台的店铺 id)。 _SOURCE_TARGET_KEY = { "meituan": "meituan", "meituan_waimai": "meituan", "jd_waimai": "jd", "jd_waimai_standalone": "jd", "taobao_flash": "taobao", } # 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", "meituan_poi_id_str", "meituan_share_url", "meituan_resolved_url", "meituan_deeplink", "jd_vender_id", "jd_share_url", "jd_resolved_url", "jd_deeplink", "attrs", ) def _find(db: Session, trace_id: str) -> StoreMapping | None: return db.execute( select(StoreMapping).where(StoreMapping.trace_id == trace_id) ).scalar_one_or_none() 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 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: # 首写: 把 payload 全部可合并列灌进去(逐列 setattr 而非硬编码构造器, 否则首写的是 # 美团/京东腿时它们的列会漏 —— 不在硬编码列表里就丢)。trace_id/business_type 是键/默认, 显式给。 row = StoreMapping( trace_id=payload.trace_id, business_type=payload.business_type, ) for col in _MERGE_COLUMNS: setattr(row, col, getattr(payload, col)) 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() db.refresh(existing) logger.info("store_mapping 合并 trace=%s 填列=%s", payload.trace_id, filled) return 0, existing.id def mark_taobao_deeplink_invalid(db: Session, shop_id: str) -> int: """把所有 id_taobao=shop_id 的行标记淘宝 deeplink 失效(置 invalid_at=now)。返回本次新标记的行数。 按 shopId 标记**所有**行 —— 同一个坏 shopId(撞淘宝"页面出错了"降级页)可能散在多次比价的 多行里, 全标掉才能让后续 lookup 不再返回它。幂等: 已标记的行(invalid_at 非 NULL)跳过。""" result = db.execute( update(StoreMapping) .where( StoreMapping.id_taobao == shop_id, StoreMapping.taobao_deeplink_invalid_at.is_(None), ) .values(taobao_deeplink_invalid_at=func.now()) ) db.commit() return result.rowcount or 0 def mark_jd_deeplink_invalid(db: Session, store_id: str) -> int: """把所有 id_jd=store_id 的行标记京东 deeplink 失效(置 invalid_at=now)。返回本次新标记的行数。 同 mark_taobao_deeplink_invalid:按 storeId 标记**所有**行(撞京东"当前门店超出配送范围"页), 全标掉才能让后续 lookup 不再返回它。幂等:已标记的行(invalid_at 非 NULL)跳过。""" result = db.execute( update(StoreMapping) .where( StoreMapping.id_jd == store_id, StoreMapping.jd_deeplink_invalid_at.is_(None), ) .values(jd_deeplink_invalid_at=func.now()) ) db.commit() return result.rowcount or 0 # ============================================================ # 缓存查询: 比价前按"源平台店名"反查已沉淀的各目标平台店铺 id, 命中就让 pricebot 直接 # deeplink 跳店内搜索, 省掉"开平台→进店→分享反查"整段。 # ============================================================ def _haversine_km(lat1: float, lng1: float, lat2: float, lng2: float) -> float: """两点球面距离(km)。仅用于同名候选里挑最近, 精度够用。""" r = 6371.0 p1, p2 = math.radians(lat1), math.radians(lat2) dp = math.radians(lat2 - lat1) dl = math.radians(lng2 - lng1) a = math.sin(dp / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dl / 2) ** 2 return 2 * r * math.asin(math.sqrt(a)) def _pick_best(rows: list[StoreMapping], lat: float | None, lng: float | None) -> StoreMapping: """同名 + 含目标 id 的候选里挑一条:有入参 geo 且有候选带 geo → 取最近;否则取 created_at 最新。""" if lat is not None and lng is not None: geod = [r for r in rows if r.lat is not None and r.lng is not None] if geod: return min(geod, key=lambda r: _haversine_km(lat, lng, r.lat, r.lng)) return max(rows, key=lambda r: r.created_at) # 目标 key → (该平台店铺 id 列, 组装返回 payload 的函数)。pricebot 拿 id 现拼 deeplink。 _TARGETS = { "taobao": ("id_taobao", lambda r: {"shop_id": r.id_taobao, "deeplink": r.taobao_deeplink}), "jd": ("id_jd", lambda r: {"store_id": r.id_jd, "vender_id": r.jd_vender_id, "deeplink": r.jd_deeplink}), "meituan": ("meituan_poi_id_str", lambda r: {"poi_id_str": r.meituan_poi_id_str, "deeplink": r.meituan_deeplink}), } def lookup_nearest( db: Session, source_platform: str, store_name: str, lat: float | None = None, lng: float | None = None, ) -> dict: """按"源平台店名"反查各目标平台已沉淀的店铺 id。返回 {target_key: {id..., deeplink, row_id, ...}}。 - 匹配键 = 源平台对应的 name 列 == store_name(精确)。 - 每个目标**分别**取"含该目标 id 的同名候选里最近一条"(淘宝 id / 京东 id 可能在不同行)。 - 排除源平台自己(不复用源平台的店铺 id)。命中为空 = 没缓存, pricebot 走现场反查老路。""" name_col = _SOURCE_NAME_COLUMN.get(source_platform) if not name_col or not store_name: return {} rows = db.execute( select(StoreMapping).where(getattr(StoreMapping, name_col) == store_name) ).scalars().all() if not rows: return {} src_key = _SOURCE_TARGET_KEY.get(source_platform) out: dict = {} for tgt, (id_attr, make_payload) in _TARGETS.items(): if tgt == src_key: continue # 不返回源平台自己 cands = [r for r in rows if getattr(r, id_attr)] if tgt == "taobao": # 失效的淘宝 deeplink(撞过错误页被 invalidate)整条排除 = 当没缓存, pricebot 走正常搜店。 cands = [r for r in cands if r.taobao_deeplink_invalid_at is None] elif tgt == "jd": # 同上:失效的京东 deeplink(撞"当前门店超出配送范围"被 invalidate)整条排除。 cands = [r for r in cands if r.jd_deeplink_invalid_at is None] if not cands: continue best = _pick_best(cands, lat, lng) payload = make_payload(best) payload["row_id"] = best.id if best.lat is not None and best.lng is not None and lat is not None and lng is not None: payload["dist_km"] = round(_haversine_km(lat, lng, best.lat, best.lng), 3) out[tgt] = payload return out