From 4844e491611325f1e540196027a4c013d1fd5523 Mon Sep 17 00:00:00 2001 From: marco Date: Sat, 13 Jun 2026 20:34:53 +0800 Subject: [PATCH 1/5] =?UTF-8?q?feat(store=5Fmapping):=20=E8=B7=A8=E5=B9=B3?= =?UTF-8?q?=E5=8F=B0=E5=BA=97=E9=93=BA=E6=98=A0=E5=B0=84=E8=A1=A8=20+=20pr?= =?UTF-8?q?icebot=20=E5=86=85=E9=83=A8=E4=B8=8A=E6=8A=A5=E7=AB=AF=E7=82=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增「平台店铺表」资产层: 淘宝比价拿到 shopId 后, pricebot server→server 把跨平台 店铺身份(各平台 id/名 + 地理 + 来源)落库, 作为未来"我见过这家店→跳过重搜/匹配"的源头。 与 price_observation(价格事实)平行、独立。 - models/store_mapping.py: store_mapping 表 22 列 —— 跨平台身份(id/name_taobao/meituan/jd) + 地理(city/geohash/lng/lat/taobao_address) + 溯源(source_platform/trace_id/device/user) + 淘宝原料(share_url/resolved_url/deeplink) + attrs(JSONB) + created_at。 - schemas/store_mapping.py + repositories/store_mapping.py: append-only, trace_id 幂等 (pricebot 重试/replay 不重复写; 并发 IntegrityError 兜底返已存在行), 跨方言安全。 - api/internal/store.py: POST /internal/store-mapping(复用 price.py 共享密钥 X-Internal-Secret 校验)。 - 注册 model(__init__) + router(main.py); 迁移 store_mapping_table 接现 head coin_txn_task_ref_uq。 验证: alembic 单 head + 零模型/迁移漂移; TestClient 全链(无密钥401/有密钥inserted=1/幂等inserted=0/错密钥401)。 Co-Authored-By: Claude Fable 5 --- alembic/versions/store_mapping_table.py | 77 ++++++++++++++++++++ app/api/internal/store.py | 46 ++++++++++++ app/main.py | 4 +- app/models/__init__.py | 1 + app/models/store_mapping.py | 97 +++++++++++++++++++++++++ app/repositories/store_mapping.py | 65 +++++++++++++++++ app/schemas/store_mapping.py | 49 +++++++++++++ 7 files changed, 338 insertions(+), 1 deletion(-) create mode 100644 alembic/versions/store_mapping_table.py create mode 100644 app/api/internal/store.py create mode 100644 app/models/store_mapping.py create mode 100644 app/repositories/store_mapping.py create mode 100644 app/schemas/store_mapping.py diff --git a/alembic/versions/store_mapping_table.py b/alembic/versions/store_mapping_table.py new file mode 100644 index 0000000..7e3e709 --- /dev/null +++ b/alembic/versions/store_mapping_table.py @@ -0,0 +1,77 @@ +"""store_mapping table (平台店铺表:跨平台同店 id/名 映射,server 侧无条件落库) + +Revision ID: store_mapping_table +Revises: coin_txn_task_ref_uq +Create Date: 2026-06-13 00:00:00.000000 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = 'store_mapping_table' +down_revision: Union[str, Sequence[str], None] = 'coin_txn_task_ref_uq' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + 'store_mapping', + sa.Column('id', sa.Integer(), autoincrement=True, nullable=False), + # 跨平台身份 + sa.Column('id_taobao', sa.String(length=64), nullable=True), + sa.Column('name_taobao', sa.String(length=128), nullable=True), + sa.Column('id_meituan', sa.String(length=64), nullable=True), + sa.Column('name_meituan', sa.String(length=128), nullable=True), + sa.Column('id_jd', sa.String(length=64), nullable=True), + sa.Column('name_jd', sa.String(length=128), nullable=True), + # 地理 + sa.Column('city', sa.String(length=64), nullable=True), + sa.Column('geohash', sa.String(length=16), nullable=True), + sa.Column('lng', sa.Float(), nullable=True), + sa.Column('lat', sa.Float(), nullable=True), + sa.Column('taobao_address', sa.String(length=256), nullable=True), + # 溯源 + sa.Column('source_platform', sa.String(length=32), nullable=True), + sa.Column('business_type', sa.String(length=16), nullable=False), + sa.Column('trace_id', sa.String(length=64), nullable=False), + sa.Column('source_device_id', sa.String(length=64), nullable=True), + sa.Column('source_user_id', sa.Integer(), nullable=True), + # 淘宝原料(URL 可能很长 → Text) + sa.Column('taobao_share_url', sa.String(length=256), nullable=True), + sa.Column('taobao_resolved_url', sa.Text(), nullable=True), + sa.Column('taobao_deeplink', sa.Text(), nullable=True), + # PG 上为 JSONB,其它(SQLite)为 JSON——与模型层 with_variant 对齐 + sa.Column('attrs', sa.JSON().with_variant(sa.dialects.postgresql.JSONB(), 'postgresql'), nullable=True), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False), + sa.PrimaryKeyConstraint('id'), + # 一次比价一行,trace_id 幂等去重(防 pricebot 重试 / replay 重复写) + sa.UniqueConstraint('trace_id', name='uq_store_mapping_trace'), + ) + with op.batch_alter_table('store_mapping', schema=None) as batch_op: + batch_op.create_index(batch_op.f('ix_store_mapping_id_taobao'), ['id_taobao'], unique=False) + batch_op.create_index(batch_op.f('ix_store_mapping_id_meituan'), ['id_meituan'], unique=False) + batch_op.create_index(batch_op.f('ix_store_mapping_id_jd'), ['id_jd'], unique=False) + batch_op.create_index(batch_op.f('ix_store_mapping_geohash'), ['geohash'], unique=False) + batch_op.create_index(batch_op.f('ix_store_mapping_source_platform'), ['source_platform'], unique=False) + batch_op.create_index(batch_op.f('ix_store_mapping_source_device_id'), ['source_device_id'], unique=False) + batch_op.create_index(batch_op.f('ix_store_mapping_source_user_id'), ['source_user_id'], unique=False) + batch_op.create_index(batch_op.f('ix_store_mapping_created_at'), ['created_at'], unique=False) + + +def downgrade() -> None: + with op.batch_alter_table('store_mapping', schema=None) as batch_op: + batch_op.drop_index(batch_op.f('ix_store_mapping_created_at')) + batch_op.drop_index(batch_op.f('ix_store_mapping_source_user_id')) + batch_op.drop_index(batch_op.f('ix_store_mapping_source_device_id')) + batch_op.drop_index(batch_op.f('ix_store_mapping_source_platform')) + batch_op.drop_index(batch_op.f('ix_store_mapping_geohash')) + batch_op.drop_index(batch_op.f('ix_store_mapping_id_jd')) + batch_op.drop_index(batch_op.f('ix_store_mapping_id_meituan')) + batch_op.drop_index(batch_op.f('ix_store_mapping_id_taobao')) + + op.drop_table('store_mapping') diff --git a/app/api/internal/store.py b/app/api/internal/store.py new file mode 100644 index 0000000..91ef749 --- /dev/null +++ b/app/api/internal/store.py @@ -0,0 +1,46 @@ +"""平台店铺映射内部上报端点(pricebot → app-server)。 + +pricebot 在淘宝比价拿到 shopId 后,把这一行跨平台店铺映射 POST 到这里落库。 +**不是给客户端的接口**:不走用户 JWT,靠 server 间共享密钥头 `X-Internal-Secret` 校验 +(复用 price.py 的 _check_secret,与 price-observation 同一密钥)。 + +与 price.py 的 /internal/price-observation 平行:那个落价格事实,这个落店铺身份映射。 +""" +from __future__ import annotations + +import logging + +from typing import Annotated + +from fastapi import APIRouter, Header + +from app.api.deps import DbSession +from app.api.internal.price import _check_secret +from app.repositories import store_mapping as repo +from app.schemas.store_mapping import StoreMappingIn, StoreMappingOut + +logger = logging.getLogger("shagua.internal.store") + +router = APIRouter(prefix="/internal", tags=["internal"]) + + +@router.post( + "/store-mapping", + response_model=StoreMappingOut, + summary="平台店铺映射内部上报(pricebot→app-server,落 store_mapping)", +) +def report_store_mapping( + payload: StoreMappingIn, + db: DbSession, + x_internal_secret: Annotated[str | None, Header()] = None, +) -> StoreMappingOut: + _check_secret(x_internal_secret) + inserted, row_id = repo.insert(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, + payload.source_device_id, payload.source_user_id, + ) + return StoreMappingOut(inserted=inserted, row_id=row_id) diff --git a/app/main.py b/app/main.py index fe296f2..cea4568 100644 --- a/app/main.py +++ b/app/main.py @@ -21,6 +21,7 @@ from app.api.v1.compare_milestone import router as compare_milestone_router from app.api.v1.compare_record import router as compare_record_router from app.api.v1.coupon import router as coupon_router from app.api.internal.price import router as internal_price_router +from app.api.internal.store import router as internal_store_router from app.api.v1.feedback import router as feedback_router from app.api.v1.invite import router as invite_router from app.api.v1.meituan import router as meituan_router @@ -102,8 +103,9 @@ app.include_router(savings_router) app.include_router(ad_router) app.include_router(order_router) app.include_router(report_router) -# 内部(server→server)端点:pricebot 上报价格观测,靠共享密钥头校验,不对客户端开放。 +# 内部(server→server)端点:pricebot 上报价格观测 / 店铺映射,靠共享密钥头校验,不对客户端开放。 app.include_router(internal_price_router) +app.include_router(internal_store_router) app.include_router(platform_router) # 用户上传文件(头像)静态服务。生产可改由 nginx 直接 serve MEDIA_ROOT。 diff --git a/app/models/__init__.py b/app/models/__init__.py index 4ed0d1b..a089a8a 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -23,6 +23,7 @@ from app.models.price_observation import PriceObservation # noqa: F401 from app.models.price_report import PriceReport # noqa: F401 from app.models.savings import SavingsRecord # noqa: F401 from app.models.signin import SigninBoostRecord, SigninRecord # noqa: F401 +from app.models.store_mapping import StoreMapping # noqa: F401 from app.models.task import UserTask # noqa: F401 from app.models.user import User # noqa: F401 from app.models.wallet import ( # noqa: F401 diff --git a/app/models/store_mapping.py b/app/models/store_mapping.py new file mode 100644 index 0000000..632f5b5 --- /dev/null +++ b/app/models/store_mapping.py @@ -0,0 +1,97 @@ +"""平台店铺表(store_mapping)—— 跨平台"同一家店"的 id/名 映射资产层。 + +每完成一次淘宝比价(在目标淘宝店通过 更多操作→分享→复制链接 拿到分享短链、 +HTTP 解析出 shopId 后),pricebot server→server 内部上报落这里一行。**与登录无关、 +不依赖客户端鉴权**(比价透传链路当前不鉴权,user_id 客户端带上时一并记)。 + +与 price_observation 的区别: +- price_observation:平台/门店视角的**价格事实**(某店这单多少钱)。 +- store_mapping:平台/门店视角的**身份映射**(同一家物理店在 淘宝/美团/京东 各自的 + 店铺 id 与店名)。是未来"我见过这家店→跳过重新搜索/匹配"的源头。两表独立。 + +「先存下来、用法后说」:列尽量铺全(各平台 id/名 + 地理 + 溯源 + 淘宝原料 URL), +attrs(JSONB)兜底存灵活明细,免得每多记一个字段就迁移 schema。 + +⚠️ 数据质量:跨平台"同一家店"的连接来自 agent 的 LLM 店铺匹配,匹配错则一行里连错店。 +本表是 append-only 原始记录(每比价一行、trace_id 幂等防重试重复),清洗/归一二期再做。 +当前只接通**淘宝**(id_taobao 才有真值;美团/京东多为 name-only 或空)。 +""" +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import ( + JSON, + DateTime, + Float, + Integer, + String, + Text, + UniqueConstraint, + func, +) +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.orm import Mapped, mapped_column + +from app.db.base import Base + +# PG 上用 JSONB,SQLite(本地/测试)退化为通用 JSON(同 price_observation / comparison_record)。 +_JSON = JSON().with_variant(JSONB(), "postgresql") + + +class StoreMapping(Base): + __tablename__ = "store_mapping" + __table_args__ = ( + # 一次比价(trace)只记一条:pricebot 重试 / 客户端 replay 重复上报时幂等去重。 + # 一次淘宝比价 = 一个目标淘宝店 → 一行映射(源 + 各平台身份压在同一行)。 + UniqueConstraint("trace_id", name="uq_store_mapping_trace"), + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + + # ===== 跨平台店铺身份(同一家物理店在各平台的 id/名;按比价角色稀疏填充)===== + # id_taobao = 分享短链解析出的 shopId(淘宝当目标、走完取 id 流程才有); + # name_taobao = 店铺页 a11y content_desc "店铺标题：xxx" 剥前缀。 + id_taobao: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) + name_taobao: Mapped[str | None] = mapped_column(String(128), nullable=True) + # 美团/京东:当前无同款 share→id 机制,多为 name-only(源平台店名来自 intent / agent 匹配名)。 + id_meituan: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) + name_meituan: Mapped[str | None] = mapped_column(String(128), nullable=True) + id_jd: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) + name_jd: Mapped[str | None] = mapped_column(String(128), nullable=True) + + # ===== 地理(同名店异地区分 / 地理分桶匹配的主要燃料)===== + city: Mapped[str | None] = mapped_column(String(64), nullable=True) + geohash: Mapped[str | None] = mapped_column(String(16), index=True, nullable=True) + lng: Mapped[float | None] = mapped_column(Float, nullable=True) + lat: Mapped[float | None] = mapped_column(Float, nullable=True) + # 淘宝门店地址(店铺页 a11y 抓到才有;比经纬度更利于人工/LLM 匹配) + taobao_address: Mapped[str | None] = mapped_column(String(256), nullable=True) + + # ===== 溯源 / 用户画像 ===== + # 源平台(发起比价那家:meituan / taobao_flash / jd_waimai ...) + source_platform: Mapped[str | None] = mapped_column(String(32), index=True, nullable=True) + business_type: Mapped[str] = mapped_column(String(16), nullable=False, default="food") + # pricebot 侧 trace_id:回指原始 trace(溯源)+ 幂等去重键(uq_store_mapping_trace) + trace_id: Mapped[str] = mapped_column(String(64), nullable=False) + source_device_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) + # user_id 当前比价链路不鉴权拿不到,客户端带上时才有;先可空。 + source_user_id: Mapped[int | None] = mapped_column(Integer, index=True, nullable=True) + + # ===== 淘宝原料(可复跳 / 可重解析 / 调试;URL 可能很长 → Text)===== + taobao_share_url: Mapped[str | None] = mapped_column(String(256), nullable=True) # m.tb.cn 短链 + taobao_resolved_url: Mapped[str | None] = mapped_column(Text, nullable=True) # 解析出的目标 URL(含 shopId) + taobao_deeplink: Mapped[str | None] = mapped_column(Text, nullable=True) # 拼好的 et-store/search deeplink + + # 灵活字段兜底(免得加字段就迁移) + attrs: Mapped[dict | None] = mapped_column(_JSON, nullable=True) + + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), index=True, nullable=False + ) + + def __repr__(self) -> str: # pragma: no cover + return ( + f"" + ) diff --git a/app/repositories/store_mapping.py b/app/repositories/store_mapping.py new file mode 100644 index 0000000..f6566e8 --- /dev/null +++ b/app/repositories/store_mapping.py @@ -0,0 +1,65 @@ +"""平台店铺映射落库:一次比价一行,trace_id 幂等。 + +幂等键 = trace_id。pricebot 重试 / replay 会重复上报同一 trace,这里先查该 trace 是否 +已存在,存在即跳过(返回已存在行 id)——跨方言(PG / SQLite dev)都安全,不依赖 ON CONFLICT。 +极端并发下若仍撞唯一约束,IntegrityError 回滚后再查一次返回已存在行(本就幂等,不重试)。 +""" +from __future__ import annotations + +import logging + +from sqlalchemy import select +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") + + +def _existing_id(db: Session, trace_id: str) -> int | None: + return db.execute( + select(StoreMapping.id).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 + + 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() + logger.warning("store_mapping 并发幂等冲突 trace=%s,返回已存在行", payload.trace_id) + return 0, _existing_id(db, payload.trace_id) + + db.refresh(row) + return 1, row.id diff --git a/app/schemas/store_mapping.py b/app/schemas/store_mapping.py new file mode 100644 index 0000000..4143fe2 --- /dev/null +++ b/app/schemas/store_mapping.py @@ -0,0 +1,49 @@ +"""平台店铺映射内部上报的收发模型。 + +pricebot 在淘宝比价拿到 shopId 后 server→server POST 一行映射。所有字段可空(按比价 +角色稀疏填充),server 端只校验共享密钥 + 幂等(trace_id)落库。 +""" +from __future__ import annotations + +from pydantic import BaseModel + + +class StoreMappingIn(BaseModel): + """一次比价的跨平台店铺映射上报体(一行)。""" + + trace_id: str + business_type: str = "food" + source_platform: str | None = None + + # 跨平台身份(按角色稀疏填充) + id_taobao: str | None = None + name_taobao: str | None = None + id_meituan: str | None = None + name_meituan: str | None = None + id_jd: str | None = None + name_jd: str | None = None + + # 地理 + city: str | None = None + geohash: str | None = None + lng: float | None = None + lat: float | None = None + taobao_address: str | None = None + + # 溯源 + source_device_id: str | None = None + source_user_id: int | None = None + + # 淘宝原料(可复跳 / 可重解析 / 调试) + taobao_share_url: str | None = None + taobao_resolved_url: str | None = None + taobao_deeplink: str | None = None + + attrs: dict | None = None + + +class StoreMappingOut(BaseModel): + """上报结果。inserted=1 为新写入,0 为幂等跳过(trace 已存在);row_id 为该 trace 对应行 id。""" + + inserted: int + row_id: int | None = None -- 2.52.0 From 0860eed92b01ad3a8298ff354fe891cf7db8c392 Mon Sep 17 00:00:00 2001 From: marco Date: Sat, 13 Jun 2026 21:53:42 +0800 Subject: [PATCH 2/5] =?UTF-8?q?feat(store=5Fmapping):=20insert=E2=86=92ups?= =?UTF-8?q?ert=20=E5=A1=AB=E7=A9=BA=E5=90=88=E5=B9=B6,=20=E4=B8=80=20trace?= =?UTF-8?q?=20=E4=B8=80=E8=A1=8C=E5=A4=9A=E5=B9=B3=E5=8F=B0=E6=B8=90?= =?UTF-8?q?=E8=BF=9B=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 -- 2.52.0 From 18185a557a993109c047a10954b2622188161d3e Mon Sep 17 00:00:00 2001 From: marco Date: Sat, 13 Jun 2026 22:25:20 +0800 Subject: [PATCH 3/5] =?UTF-8?q?feat(store=5Fmapping):=20=E5=8A=A0=E7=BE=8E?= =?UTF-8?q?=E5=9B=A2=E5=8E=9F=E6=96=99=E5=88=97(poi=5Fid=5Fstr=20+=20?= =?UTF-8?q?=E5=88=86=E4=BA=AB=E9=93=BE/=E5=8F=8D=E6=9F=A5URL/deeplink)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 美团接入店内搜索 deeplink 链路, 表加 4 列存美团侧店铺身份/原料: - meituan_poi_id_str: 分享链 302 反查出的 poi_id_str(索引) - meituan_share_url / meituan_resolved_url / meituan_deeplink: 与 taobao_* 平行 poi_id_str 不进 id_meituan: 调研实证它每次分享重新加密、非稳定主键(同店两条分享链 两个值, 见 docs §八), 单列当"可复跳的一次性票据"存; id_meituan 留给将来 CPS API 拿到的稳定数字 poi_id。4 列均纳入 upsert 填空合并集。 迁移 store_mapping_meituan_cols(down=store_mapping_table); 已 upgrade, 单 head, 模型零漂移, 合并不变式 26列=22合并+4排除 仍成立。 Co-Authored-By: Claude Fable 5 --- .../versions/store_mapping_meituan_cols.py | 39 +++++++++++++++++++ app/models/store_mapping.py | 26 +++++++++++-- app/repositories/store_mapping.py | 1 + app/schemas/store_mapping.py | 6 +++ 4 files changed, 69 insertions(+), 3 deletions(-) create mode 100644 alembic/versions/store_mapping_meituan_cols.py diff --git a/alembic/versions/store_mapping_meituan_cols.py b/alembic/versions/store_mapping_meituan_cols.py new file mode 100644 index 0000000..e9494a2 --- /dev/null +++ b/alembic/versions/store_mapping_meituan_cols.py @@ -0,0 +1,39 @@ +"""store_mapping 加美团原料列(poi_id_str + 分享链/反查 URL/deeplink) + +Revision ID: store_mapping_meituan_cols +Revises: store_mapping_table +Create Date: 2026-06-13 00:00:00.000000 + +美团接入店内搜索 deeplink 链路: dpurl.cn 短链 → 302 反查 poi_id_str → imeituan:// deeplink。 +poi_id_str 每次分享重新加密、非稳定主键, 单列 meituan_poi_id_str 存(不占 id_meituan, +后者留给将来 CPS API 的稳定数字 poi_id)。其余三列与 taobao_* 原料列平行。 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = 'store_mapping_meituan_cols' +down_revision: Union[str, Sequence[str], None] = 'store_mapping_table' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + with op.batch_alter_table('store_mapping', schema=None) as batch_op: + batch_op.add_column(sa.Column('meituan_poi_id_str', sa.String(length=64), nullable=True)) + batch_op.add_column(sa.Column('meituan_share_url', sa.String(length=256), nullable=True)) + batch_op.add_column(sa.Column('meituan_resolved_url', sa.Text(), nullable=True)) + batch_op.add_column(sa.Column('meituan_deeplink', sa.Text(), nullable=True)) + batch_op.create_index(batch_op.f('ix_store_mapping_meituan_poi_id_str'), ['meituan_poi_id_str'], unique=False) + + +def downgrade() -> None: + with op.batch_alter_table('store_mapping', schema=None) as batch_op: + batch_op.drop_index(batch_op.f('ix_store_mapping_meituan_poi_id_str')) + batch_op.drop_column('meituan_deeplink') + batch_op.drop_column('meituan_resolved_url') + batch_op.drop_column('meituan_share_url') + batch_op.drop_column('meituan_poi_id_str') diff --git a/app/models/store_mapping.py b/app/models/store_mapping.py index 632f5b5..2849f75 100644 --- a/app/models/store_mapping.py +++ b/app/models/store_mapping.py @@ -9,12 +9,13 @@ HTTP 解析出 shopId 后),pricebot server→server 内部上报落这里一行 - store_mapping:平台/门店视角的**身份映射**(同一家物理店在 淘宝/美团/京东 各自的 店铺 id 与店名)。是未来"我见过这家店→跳过重新搜索/匹配"的源头。两表独立。 -「先存下来、用法后说」:列尽量铺全(各平台 id/名 + 地理 + 溯源 + 淘宝原料 URL), +「先存下来、用法后说」:列尽量铺全(各平台 id/名 + 地理 + 溯源 + 淘宝/美团原料 URL), attrs(JSONB)兜底存灵活明细,免得每多记一个字段就迁移 schema。 ⚠️ 数据质量:跨平台"同一家店"的连接来自 agent 的 LLM 店铺匹配,匹配错则一行里连错店。 本表是 append-only 原始记录(每比价一行、trace_id 幂等防重试重复),清洗/归一二期再做。 -当前只接通**淘宝**(id_taobao 才有真值;美团/京东多为 name-only 或空)。 +已接通**淘宝**(id_taobao=shopId)与**美团**(meituan_poi_id_str,非稳定主键、单列存); +京东暂为 name-only 或空。 """ from __future__ import annotations @@ -54,9 +55,12 @@ class StoreMapping(Base): # name_taobao = 店铺页 a11y content_desc "店铺标题：xxx" 剥前缀。 id_taobao: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) name_taobao: Mapped[str | None] = mapped_column(String(128), nullable=True) - # 美团/京东:当前无同款 share→id 机制,多为 name-only(源平台店名来自 intent / agent 匹配名)。 + # 美团:无同款 share→稳定id 机制(poi_id_str 每次变,见下方 meituan_poi_id_str),id_meituan + # 留给将来 CPS API 的稳定 poi_id;name-only 时仅 name_meituan(源平台店名来自 intent/agent 匹配名)。 id_meituan: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) name_meituan: Mapped[str | None] = mapped_column(String(128), nullable=True) + # 京东:已接通 share→id(3.cn 短链反查)。id_jd = storeId(门店稳定数字主键,同 taobao shopId→ + # id_taobao);venderId(deeplink 还需)单列存 jd_vender_id。name_jd = 店铺页店名。 id_jd: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) name_jd: Mapped[str | None] = mapped_column(String(128), nullable=True) @@ -83,6 +87,22 @@ class StoreMapping(Base): taobao_resolved_url: Mapped[str | None] = mapped_column(Text, nullable=True) # 解析出的目标 URL(含 shopId) taobao_deeplink: Mapped[str | None] = mapped_column(Text, nullable=True) # 拼好的 et-store/search deeplink + # ===== 美团原料(同淘宝;dpurl.cn 短链 → 302 反查 poi_id_str → imeituan:// deeplink)===== + # ⚠️ poi_id_str 每次分享重新加密、非稳定主键(调研文档 §八), 故单列存"可复跳的一次性票据", + # 不进 id_meituan —— 后者留给将来 CPS API 拿到的稳定数字 poi_id。 + meituan_poi_id_str: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) + meituan_share_url: Mapped[str | None] = mapped_column(String(256), nullable=True) # dpurl.cn 短链 + meituan_resolved_url: Mapped[str | None] = mapped_column(Text, nullable=True) # 302 落地 menu URL(含 poi_id_str) + meituan_deeplink: Mapped[str | None] = mapped_column(Text, nullable=True) # 拼好的 imeituan:// 店内搜索 deeplink + + # ===== 京东原料(秒送;3.cn 短链 → 跟随重定向反查 venderId+storeId → openapp.jdmobile:// deeplink)===== + # storeId 进 id_jd(稳定店主键);venderId 单列存(deeplink 模板 venderId+storeId 都要,且 venderId + # 是商家维度、可跨门店,与门店 storeId 分开记)。其余三列与 taobao_*/meituan_* 平行。 + jd_vender_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) + jd_share_url: Mapped[str | None] = mapped_column(String(256), nullable=True) # 3.cn 短链 + jd_resolved_url: Mapped[str | None] = mapped_column(Text, nullable=True) # 反查出的目标 openapp.jdmobile:// deeplink + jd_deeplink: Mapped[str | None] = mapped_column(Text, nullable=True) # 拼好的 pages/search 店内搜索 deeplink + # 灵活字段兜底(免得加字段就迁移) attrs: Mapped[dict | None] = mapped_column(_JSON, nullable=True) diff --git a/app/repositories/store_mapping.py b/app/repositories/store_mapping.py index b9b9472..5b64750 100644 --- a/app/repositories/store_mapping.py +++ b/app/repositories/store_mapping.py @@ -31,6 +31,7 @@ _MERGE_COLUMNS = ( "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", "attrs", ) diff --git a/app/schemas/store_mapping.py b/app/schemas/store_mapping.py index 0b448c5..a5eb9aa 100644 --- a/app/schemas/store_mapping.py +++ b/app/schemas/store_mapping.py @@ -39,6 +39,12 @@ class StoreMappingIn(BaseModel): taobao_resolved_url: str | None = None taobao_deeplink: str | None = None + # 美团原料(poi_id_str 非稳定主键, 单列存; 见 model 注释) + meituan_poi_id_str: str | None = None + meituan_share_url: str | None = None + meituan_resolved_url: str | None = None + meituan_deeplink: str | None = None + attrs: dict | None = None -- 2.52.0 From fdd3d8cf9b06cfa108adb4f882fc573ad6f91276 Mon Sep 17 00:00:00 2001 From: marco Date: Sat, 13 Jun 2026 22:35:39 +0800 Subject: [PATCH 4/5] =?UTF-8?q?feat(store=5Fmapping):=20=E5=8A=A0=E4=BA=AC?= =?UTF-8?q?=E4=B8=9C=E5=8E=9F=E6=96=99=E5=88=97=20+=20upsert=20=E9=A6=96?= =?UTF-8?q?=E5=86=99=E6=B3=9B=E5=8C=96(=E4=BB=BB=E6=84=8F=E5=B9=B3?= =?UTF-8?q?=E5=8F=B0=E9=A6=96=E5=86=99=E4=B8=8D=E4=B8=A2=E5=88=97)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 京东秒送接入: 店铺身份是两个稳定数字 id —— storeId 进 id_jd(同 taobao shopId→id_taobao), venderId(deeplink 还需、商家维度)单列存 jd_vender_id; 另加 jd_share_url/jd_resolved_url/ jd_deeplink 原料列(与 taobao_*/meituan_* 平行)。迁移 store_mapping_jd_cols 加这 4 列 + jd_vender_id 索引。 顺手修一个潜在 bug: upsert 的首写(INSERT)原来硬编码只灌 taobao+通用列, 首写的若是美团/ 京东腿, 它们的列不在构造器里会被丢。改成按 _MERGE_COLUMNS 逐列 setattr, 任意平台首写都全量 落库(美团也受益)。合并(填空不覆盖)逻辑不变。 验证: 迁移单 head、store_mapping 无新漂移; TestClient 打真 PG —— 京东首写 4 列全落 → 淘宝同 trace 合并填淘宝列、不覆盖京东列。 Co-Authored-By: Claude Fable 5 --- alembic/versions/store_mapping_jd_cols.py | 40 +++++++++++++++++++++++ app/models/store_mapping.py | 4 +-- app/repositories/store_mapping.py | 23 +++---------- app/schemas/store_mapping.py | 6 ++++ 4 files changed, 53 insertions(+), 20 deletions(-) create mode 100644 alembic/versions/store_mapping_jd_cols.py diff --git a/alembic/versions/store_mapping_jd_cols.py b/alembic/versions/store_mapping_jd_cols.py new file mode 100644 index 0000000..02acf32 --- /dev/null +++ b/alembic/versions/store_mapping_jd_cols.py @@ -0,0 +1,40 @@ +"""store_mapping 加京东原料列(jd_vender_id + 分享链/反查 URL/deeplink) + +Revision ID: store_mapping_jd_cols +Revises: store_mapping_meituan_cols +Create Date: 2026-06-13 00:00:00.000000 + +京东秒送接入店内搜索 deeplink 链路: 3.cn 短链 → 跟随重定向反查 venderId+storeId → +openapp.jdmobile:// deeplink。京东店铺身份是**两个**稳定数字 id: storeId 进 id_jd(稳定 +店主键, 同 taobao 的 shopId→id_taobao), venderId 进单列 jd_vender_id(deeplink 还需它)。 +其余三列与 taobao_*/meituan_* 原料列平行。 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = 'store_mapping_jd_cols' +down_revision: Union[str, Sequence[str], None] = 'store_mapping_meituan_cols' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + with op.batch_alter_table('store_mapping', schema=None) as batch_op: + batch_op.add_column(sa.Column('jd_vender_id', sa.String(length=64), nullable=True)) + batch_op.add_column(sa.Column('jd_share_url', sa.String(length=256), nullable=True)) + batch_op.add_column(sa.Column('jd_resolved_url', sa.Text(), nullable=True)) + batch_op.add_column(sa.Column('jd_deeplink', sa.Text(), nullable=True)) + batch_op.create_index(batch_op.f('ix_store_mapping_jd_vender_id'), ['jd_vender_id'], unique=False) + + +def downgrade() -> None: + with op.batch_alter_table('store_mapping', schema=None) as batch_op: + batch_op.drop_index(batch_op.f('ix_store_mapping_jd_vender_id')) + batch_op.drop_column('jd_deeplink') + batch_op.drop_column('jd_resolved_url') + batch_op.drop_column('jd_share_url') + batch_op.drop_column('jd_vender_id') diff --git a/app/models/store_mapping.py b/app/models/store_mapping.py index 2849f75..a76f8aa 100644 --- a/app/models/store_mapping.py +++ b/app/models/store_mapping.py @@ -14,8 +14,8 @@ attrs(JSONB)兜底存灵活明细,免得每多记一个字段就迁移 schema。 ⚠️ 数据质量:跨平台"同一家店"的连接来自 agent 的 LLM 店铺匹配,匹配错则一行里连错店。 本表是 append-only 原始记录(每比价一行、trace_id 幂等防重试重复),清洗/归一二期再做。 -已接通**淘宝**(id_taobao=shopId)与**美团**(meituan_poi_id_str,非稳定主键、单列存); -京东暂为 name-only 或空。 +已接通**淘宝**(id_taobao=shopId)、**美团**(meituan_poi_id_str,非稳定主键、单列存)、 +**京东**(id_jd=storeId + jd_vender_id,均稳定数字主键;3.cn 短链反查)。 """ from __future__ import annotations diff --git a/app/repositories/store_mapping.py b/app/repositories/store_mapping.py index 5b64750..12d4666 100644 --- a/app/repositories/store_mapping.py +++ b/app/repositories/store_mapping.py @@ -32,6 +32,7 @@ _MERGE_COLUMNS = ( "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", ) @@ -58,28 +59,14 @@ def upsert(db: Session, payload: StoreMappingIn) -> tuple[int, int | None]: 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, - 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, ) + for col in _MERGE_COLUMNS: + setattr(row, col, getattr(payload, col)) db.add(row) try: db.commit() diff --git a/app/schemas/store_mapping.py b/app/schemas/store_mapping.py index a5eb9aa..2d84c75 100644 --- a/app/schemas/store_mapping.py +++ b/app/schemas/store_mapping.py @@ -45,6 +45,12 @@ class StoreMappingIn(BaseModel): meituan_resolved_url: str | None = None meituan_deeplink: str | None = None + # 京东原料(venderId 单列存, storeId 进 id_jd; 见 model 注释) + jd_vender_id: str | None = None + jd_share_url: str | None = None + jd_resolved_url: str | None = None + jd_deeplink: str | None = None + attrs: dict | None = None -- 2.52.0 From 848f23aaeae65650c6e0d9622312f97e38a5af39 Mon Sep 17 00:00:00 2001 From: marco Date: Sun, 14 Jun 2026 00:44:55 +0800 Subject: [PATCH 5/5] =?UTF-8?q?feat(store=5Fmapping):=20=E5=8A=A0=20GET=20?= =?UTF-8?q?/internal/store-mapping/lookup=20=E7=BC=93=E5=AD=98=E6=9F=A5?= =?UTF-8?q?=E8=AF=A2(=E6=8C=89=E6=BA=90=E5=BA=97=E5=90=8D+geo=20=E5=8F=8D?= =?UTF-8?q?=E6=9F=A5=E7=9B=AE=E6=A0=87=E5=BA=97=E9=93=BA=20id)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 比价前 pricebot 调它反查各目标平台已沉淀的店铺 id, 命中则目标腿直接 deeplink 跳店内搜索。 - lookup_nearest: 按"源平台对应 name 列 == 源店名"匹配; 每个目标**分别**取"含该目标 id 的同名候选 里最近一条"(haversine, 淘宝 id / 京东 id 可能在不同行); 排除源平台自己; 有入参 geo 取最近、否则 取 created_at 最新。返 {target: {店铺id..., deeplink, row_id, dist_km}}。 - 端点日志打源/店名/geo/命中行/距离/deeplink, 与 pricebot 侧对账。 - 同次提交带上 upsert 首写改逐列灌 _MERGE_COLUMNS(任意平台首写不丢列)+ 美团/京东合并列。 Co-Authored-By: Claude Fable 5 --- app/api/internal/store.py | 33 +++++++++++++ app/repositories/store_mapping.py | 82 +++++++++++++++++++++++++++++++ 2 files changed, 115 insertions(+) diff --git a/app/api/internal/store.py b/app/api/internal/store.py index d220925..916398a 100644 --- a/app/api/internal/store.py +++ b/app/api/internal/store.py @@ -24,6 +24,39 @@ logger = logging.getLogger("shagua.internal.store") router = APIRouter(prefix="/internal", tags=["internal"]) +@router.get( + "/store-mapping/lookup", + summary="比价前按源平台店名反查各目标平台已沉淀的店铺 id(命中→pricebot 直接 deeplink)", +) +def lookup_store_mapping( + source_platform: str, + name: str, + db: DbSession, + lat: float | None = None, + lng: float | None = None, + x_internal_secret: Annotated[str | None, Header()] = None, +) -> dict: + _check_secret(x_internal_secret) + result = repo.lookup_nearest(db, source_platform, name, lat, lng) + if result: + hits = ", ".join( + f"{t}:row{v['row_id']}" + f"{'(' + str(v['dist_km']) + 'km)' if 'dist_km' in v else ''}" + f"→{v.get('deeplink') or '(无deeplink)'}" + for t, v in result.items() + ) + logger.info( + "store_mapping lookup source=%s name=%r geo=(%s,%s) → 命中 %s", + source_platform, name, lat, lng, hits, + ) + else: + logger.info( + "store_mapping lookup source=%s name=%r geo=(%s,%s) → MISS", + source_platform, name, lat, lng, + ) + return result + + @router.post( "/store-mapping", response_model=StoreMappingOut, diff --git a/app/repositories/store_mapping.py b/app/repositories/store_mapping.py index 12d4666..8ebc9fc 100644 --- a/app/repositories/store_mapping.py +++ b/app/repositories/store_mapping.py @@ -13,6 +13,7 @@ name-match 置信度、非已核实同一实体。作为 append-only 原始资 from __future__ import annotations import logging +import math from sqlalchemy import select from sqlalchemy.exc import IntegrityError @@ -23,6 +24,22 @@ 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 = ( @@ -88,3 +105,68 @@ def upsert(db: Session, payload: StoreMappingIn) -> tuple[int, int | None]: db.refresh(existing) logger.info("store_mapping 合并 trace=%s 填列=%s", payload.trace_id, filled) return 0, existing.id + + +# ============================================================ +# 缓存查询: 比价前按"源平台店名"反查已沉淀的各目标平台店铺 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 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 -- 2.52.0