Compare commits
5 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 848f23aaea | |||
| fdd3d8cf9b | |||
| 18185a557a | |||
| 0860eed92b | |||
| 4844e49161 |
@@ -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')
|
||||
@@ -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')
|
||||
@@ -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')
|
||||
@@ -0,0 +1,79 @@
|
||||
"""平台店铺映射内部上报端点(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.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,
|
||||
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)
|
||||
created, row_id = repo.upsert(db, payload)
|
||||
logger.info(
|
||||
"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=created, row_id=row_id)
|
||||
+3
-1
@@ -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。
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -0,0 +1,117 @@
|
||||
"""平台店铺表(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=shopId)、**美团**(meituan_poi_id_str,非稳定主键、单列存)、
|
||||
**京东**(id_jd=storeId + jd_vender_id,均稳定数字主键;3.cn 短链反查)。
|
||||
"""
|
||||
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 机制(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)
|
||||
|
||||
# ===== 地理(同名店异地区分 / 地理分桶匹配的主要燃料)=====
|
||||
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
|
||||
|
||||
# ===== 美团原料(同淘宝;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)
|
||||
|
||||
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"<StoreMapping id={self.id} taobao=({self.id_taobao!r},{self.name_taobao!r}) "
|
||||
f"source={self.source_platform} trace_id={self.trace_id}>"
|
||||
)
|
||||
@@ -0,0 +1,172 @@
|
||||
"""平台店铺映射落库:一次比价(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 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")
|
||||
|
||||
# 源平台 → 该平台店名所在列(缓存查询的匹配键)。镜像 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
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 缓存查询: 比价前按"源平台店名"反查已沉淀的各目标平台店铺 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
|
||||
@@ -0,0 +1,62 @@
|
||||
"""平台店铺映射内部上报的收发模型。
|
||||
|
||||
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
|
||||
|
||||
# 美团原料(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
|
||||
|
||||
# 京东原料(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
|
||||
|
||||
|
||||
class StoreMappingOut(BaseModel):
|
||||
"""上报结果。inserted=1 为新建该 trace 行,0 为合并进已存在行(填空,不覆盖);
|
||||
row_id 为该 trace 对应行 id。"""
|
||||
|
||||
inserted: int
|
||||
row_id: int | None = None
|
||||
Reference in New Issue
Block a user