feat(store-mapping): JD 缓存 deeplink 失效标记(对标淘宝)

store_mapping 加 jd_deeplink_invalid_at 列 + 迁移(down_revision=store_mapping_tb_dl_invalid);
mark_jd_deeplink_invalid 按 id_jd 标记 + lookup_nearest 过滤已失效 JD 候选;
/internal/store-mapping/invalidate 接 jd 分支。配 pricebot 京东缓存 deeplink 失效回退。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

alembic/versions/store_mapping_jd_dl_invalid.py

@@ -0,0 +1,32 @@
+"""store_mapping 加京东 deeplink 失效标记列 jd_deeplink_invalid_at
+
+Revision ID: store_mapping_jd_dl_invalid
+Revises: store_mapping_tb_dl_invalid
+Create Date: 2026-06-15 00:00:00.000000
+
+京东同淘宝: 缓存的店内搜索 deeplink 会失效(打开是"当前门店超出配送范围"页)。pricebot 比价撞到
+失效页时回退正常搜店, 并 server→server 通知把该 storeId 的 deeplink 标记失效。本列记失效时刻
+(NULL=有效); lookup 反查时过滤掉已失效的京东候选, 不再返回坏 deeplink。
+与淘宝 taobao_deeplink_invalid_at 对称; 用时间戳而非布尔: 留痕可审计、可统计失效率, 不销毁原 deeplink。
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'store_mapping_jd_dl_invalid'
+down_revision: Union[str, Sequence[str], None] = 'store_mapping_tb_dl_invalid'
+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_deeplink_invalid_at', sa.DateTime(timezone=True), nullable=True))
+
+
+def downgrade() -> None:
+    with op.batch_alter_table('store_mapping', schema=None) as batch_op:
+        batch_op.drop_column('jd_deeplink_invalid_at')

alembic/versions/store_mapping_tb_dl_invalid.py

@@ -0,0 +1,33 @@
+"""store_mapping 加淘宝 deeplink 失效标记列 taobao_deeplink_invalid_at
+
+Revision ID: store_mapping_tb_dl_invalid
+Revises: coupon_engage_per_package
+Create Date: 2026-06-15 00:00:00.000000
+
+缓存的淘宝店内搜索 deeplink 会失效(打开是"页面出错了"降级页)。pricebot 比价撞到错误页时
+回退正常搜店, 并 server→server 通知把该 shopId 的 deeplink 标记失效。本列记失效时刻
+(NULL=有效); lookup 反查时过滤掉已失效的淘宝候选, 不再返回坏 deeplink。
+只加淘宝一列(当前只接淘宝); 用时间戳而非布尔: 留痕可审计、可统计失效率, 且不销毁原 deeplink。
+无需索引: 过滤总叠在 name_taobao== 之后, 候选集已小。
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'store_mapping_tb_dl_invalid'
+down_revision: Union[str, Sequence[str], None] = 'coupon_engage_per_package'
+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('taobao_deeplink_invalid_at', sa.DateTime(timezone=True), nullable=True))
+
+
+def downgrade() -> None:
+    with op.batch_alter_table('store_mapping', schema=None) as batch_op:
+        batch_op.drop_column('taobao_deeplink_invalid_at')

app/api/internal/store.py

@@ -17,7 +17,12 @@ 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
+from app.schemas.store_mapping import (
+    StoreMappingIn,
+    StoreMappingInvalidateIn,
+    StoreMappingInvalidateOut,
+    StoreMappingOut,
+)
 
 logger = logging.getLogger("shagua.internal.store")
 
@@ -77,3 +82,29 @@ def report_store_mapping(
         payload.source_device_id, payload.source_user_id,
     )
     return StoreMappingOut(inserted=created, row_id=row_id)
+
+
+@router.post(
+    "/store-mapping/invalidate",
+    response_model=StoreMappingInvalidateOut,
+    summary="标记某平台 shopId 的缓存 deeplink 失效(pricebot 撞错误页回退时上报,lookup 不再返回)",
+)
+def invalidate_store_mapping(
+    payload: StoreMappingInvalidateIn,
+    db: DbSession,
+    x_internal_secret: Annotated[str | None, Header()] = None,
+) -> StoreMappingInvalidateOut:
+    _check_secret(x_internal_secret)
+    if payload.platform == "taobao":
+        affected = repo.mark_taobao_deeplink_invalid(db, payload.shop_id)
+    elif payload.platform == "jd":
+        affected = repo.mark_jd_deeplink_invalid(db, payload.shop_id)
+    else:
+        # 当前只接淘宝/京东; 其它平台先 no-op(affected=0), 不报错 — 向后兼容 pricebot 将来扩展。
+        logger.info("store_mapping invalidate 跳过: platform=%s 暂不支持", payload.platform)
+        return StoreMappingInvalidateOut(ok=True, affected=0)
+    logger.info(
+        "store_mapping invalidate platform=%s shop_id=%s → 标记失效 %d 行",
+        payload.platform, payload.shop_id, affected,
+    )
+    return StoreMappingInvalidateOut(ok=True, affected=affected)

app/models/store_mapping.py

@@ -86,6 +86,9 @@ class StoreMapping(Base):
     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
+    # 淘宝 deeplink 失效标记:比价撞"页面出错了"降级页时被置(pricebot server→server invalidate),
+    # NULL=有效。lookup 反查过滤掉非 NULL 的淘宝候选,不再返回坏 deeplink(重搜会写新行覆盖)。
+    taobao_deeplink_invalid_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
 
     # ===== 美团原料(同淘宝;dpurl.cn 短链 → 302 反查 poi_id_str → imeituan:// deeplink)=====
     # ⚠️ poi_id_str 每次分享重新加密、非稳定主键(调研文档 §八), 故单列存"可复跳的一次性票据",
@@ -102,6 +105,8 @@ class StoreMapping(Base):
     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
+    # 京东 deeplink 失效标记(同 taobao_deeplink_invalid_at):撞"当前门店超出配送范围"页时被置,NULL=有效。
+    jd_deeplink_invalid_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
 
     # 灵活字段兜底(免得加字段就迁移)
     attrs: Mapped[dict | None] = mapped_column(_JSON, nullable=True)

app/repositories/store_mapping.py

@@ -15,7 +15,7 @@ from __future__ import annotations
 import logging
 import math
 
-from sqlalchemy import select
+from sqlalchemy import func, select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -107,6 +107,40 @@ def upsert(db: Session, payload: StoreMappingIn) -> tuple[int, int | None]:
     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 跳店内搜索, 省掉"开平台→进店→分享反查"整段。
@@ -161,6 +195,12 @@ def lookup_nearest(
         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)

app/schemas/store_mapping.py

@@ -60,3 +60,17 @@ class StoreMappingOut(BaseModel):
 
     inserted: int
     row_id: int | None = None
+
+
+class StoreMappingInvalidateIn(BaseModel):
+    """标记某平台某 shopId 的缓存 deeplink 失效(pricebot 撞错误页回退时上报)。"""
+
+    platform: str  # 目前只支持 "taobao"
+    shop_id: str  # 失效的店铺 id(淘宝 = shopId = id_taobao)
+
+
+class StoreMappingInvalidateOut(BaseModel):
+    """标记结果。affected = 本次新标记失效的行数(按 shopId 标记所有匹配行)。"""
+
+    ok: bool
+    affected: int

tests/test_store_mapping_invalidate.py

@@ -0,0 +1,156 @@
+"""淘宝 deeplink 失效标记 + lookup 过滤 + invalidate 端点测试。
+
+覆盖:按 shopId 标记所有行(幂等)、lookup 过滤失效淘宝候选、失效后新行仍可命中、
+invalidate 端点鉴权(503 未配 / 401 头错 / 200)、端到端标记后 lookup MISS、非淘宝 no-op。
+"""
+from __future__ import annotations
+
+import pytest
+
+from app.core.config import settings
+from app.db.session import SessionLocal
+from app.models.store_mapping import StoreMapping
+from app.repositories import store_mapping as repo
+
+_SECRET = "test-internal-secret-only-for-pytest"
+
+
+def _mk_row(db, *, trace_id, name_meituan=None, name_taobao=None,
+            id_taobao=None, deeplink=None, lat=None, lng=None,
+            name_jd=None, id_jd=None, jd_deeplink=None):
+    row = StoreMapping(
+        trace_id=trace_id, business_type="food",
+        name_meituan=name_meituan, name_taobao=name_taobao,
+        id_taobao=id_taobao, taobao_deeplink=deeplink, lat=lat, lng=lng,
+        name_jd=name_jd, id_jd=id_jd, jd_deeplink=jd_deeplink,
+    )
+    db.add(row)
+    db.commit()
+    db.refresh(row)
+    return row
+
+
+@pytest.fixture()
+def db():
+    """每个用例前后清空 store_mapping(表由 conftest 的 create_all 建好,session 级共享)。"""
+    s = SessionLocal()
+    s.query(StoreMapping).delete()
+    s.commit()
+    yield s
+    s.query(StoreMapping).delete()
+    s.commit()
+    s.close()
+
+
+# ---------- repo 层 ----------
+
+def test_mark_invalid_marks_all_rows_with_shop_id(db):
+    # 同一个坏 shopId 散在两行(两次比价),另一行不同 shopId
+    _mk_row(db, trace_id="t1", name_meituan="绝味鸭脖", name_taobao="绝味鸭脖", id_taobao="BAD1", deeplink="dl_a")
+    _mk_row(db, trace_id="t2", name_meituan="绝味鸭脖", name_taobao="绝味鸭脖", id_taobao="BAD1", deeplink="dl_b")
+    _mk_row(db, trace_id="t3", name_meituan="绝味鸭脖", name_taobao="绝味鸭脖", id_taobao="GOOD9", deeplink="dl_g")
+
+    # 按 shopId 标记所有行
+    assert repo.mark_taobao_deeplink_invalid(db, "BAD1") == 2
+    # 幂等:已标记的不重复
+    assert repo.mark_taobao_deeplink_invalid(db, "BAD1") == 0
+
+    rows = {r.trace_id: r for r in db.query(StoreMapping).all()}
+    assert rows["t1"].taobao_deeplink_invalid_at is not None
+    assert rows["t2"].taobao_deeplink_invalid_at is not None
+    assert rows["t3"].taobao_deeplink_invalid_at is None  # 不同 shopId 不动
+
+
+def test_lookup_filters_invalid_taobao(db):
+    _mk_row(db, trace_id="t1", name_meituan="绝味鸭脖", name_taobao="绝味鸭脖", id_taobao="BAD1", deeplink="dl")
+    # 标记前命中
+    assert repo.lookup_nearest(db, "meituan", "绝味鸭脖")["taobao"]["shop_id"] == "BAD1"
+    # 标记失效后淘宝候选被过滤 → MISS(仅此一条淘宝)
+    repo.mark_taobao_deeplink_invalid(db, "BAD1")
+    assert "taobao" not in repo.lookup_nearest(db, "meituan", "绝味鸭脖")
+
+
+def test_lookup_picks_new_valid_row_after_invalidate(db):
+    # 失效旧行 + 重搜写的新行(新 shopId,invalid_at=NULL)共存 → lookup 选到新行
+    _mk_row(db, trace_id="t1", name_meituan="店A", name_taobao="店A", id_taobao="BAD1", deeplink="dl_bad")
+    repo.mark_taobao_deeplink_invalid(db, "BAD1")
+    _mk_row(db, trace_id="t2", name_meituan="店A", name_taobao="店A", id_taobao="NEW2", deeplink="dl_new")
+    assert repo.lookup_nearest(db, "meituan", "店A")["taobao"]["shop_id"] == "NEW2"
+
+
+# ---------- invalidate 端点 ----------
+
+def test_invalidate_endpoint_secret_unset_503(client, monkeypatch):
+    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", "")  # 未配 = 端点关闭
+    r = client.post("/internal/store-mapping/invalidate",
+                    json={"platform": "taobao", "shop_id": "X"})
+    assert r.status_code == 503
+
+
+def test_invalidate_endpoint_auth(client, monkeypatch):
+    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", _SECRET)
+    body = {"platform": "taobao", "shop_id": "X"}
+    assert client.post("/internal/store-mapping/invalidate", json=body).status_code == 401
+    assert client.post("/internal/store-mapping/invalidate", json=body,
+                       headers={"X-Internal-Secret": "wrong"}).status_code == 401
+    r = client.post("/internal/store-mapping/invalidate", json=body,
+                    headers={"X-Internal-Secret": _SECRET})
+    assert r.status_code == 200 and r.json()["ok"] is True
+
+
+def test_invalidate_endpoint_marks_and_lookup_miss(client, db, monkeypatch):
+    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", _SECRET)
+    _mk_row(db, trace_id="t1", name_meituan="店B", name_taobao="店B", id_taobao="BADX", deeplink="dl")
+    _mk_row(db, trace_id="t2", name_meituan="店B", name_taobao="店B", id_taobao="BADX", deeplink="dl2")
+    r = client.post("/internal/store-mapping/invalidate",
+                    json={"platform": "taobao", "shop_id": "BADX"},
+                    headers={"X-Internal-Secret": _SECRET})
+    assert r.status_code == 200 and r.json()["affected"] == 2
+    lk = client.get("/internal/store-mapping/lookup",
+                    params={"source_platform": "meituan", "name": "店B"},
+                    headers={"X-Internal-Secret": _SECRET})
+    assert "taobao" not in lk.json()
+
+
+def test_invalidate_endpoint_unsupported_platform_noop(client, monkeypatch):
+    # 当前支持 taobao/jd;其它平台(如 meituan)no-op 返 affected=0
+    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", _SECRET)
+    r = client.post("/internal/store-mapping/invalidate",
+                    json={"platform": "meituan", "shop_id": "X"},
+                    headers={"X-Internal-Secret": _SECRET})
+    assert r.status_code == 200 and r.json()["affected"] == 0
+
+
+# ---------- 京东(对称淘宝)----------
+
+def test_mark_jd_invalid_marks_all_rows_with_store_id(db):
+    _mk_row(db, trace_id="j1", name_meituan="兰州拉面", name_jd="兰州拉面", id_jd="JBAD", jd_deeplink="dl_a")
+    _mk_row(db, trace_id="j2", name_meituan="兰州拉面", name_jd="兰州拉面", id_jd="JBAD", jd_deeplink="dl_b")
+    _mk_row(db, trace_id="j3", name_meituan="兰州拉面", name_jd="兰州拉面", id_jd="JGOOD", jd_deeplink="dl_g")
+    assert repo.mark_jd_deeplink_invalid(db, "JBAD") == 2
+    assert repo.mark_jd_deeplink_invalid(db, "JBAD") == 0  # 幂等
+    rows = {r.trace_id: r for r in db.query(StoreMapping).all()}
+    assert rows["j1"].jd_deeplink_invalid_at is not None
+    assert rows["j2"].jd_deeplink_invalid_at is not None
+    assert rows["j3"].jd_deeplink_invalid_at is None  # 不同 storeId 不动
+
+
+def test_lookup_filters_invalid_jd(db):
+    _mk_row(db, trace_id="j1", name_meituan="兰州拉面", name_jd="兰州拉面", id_jd="JBAD", jd_deeplink="dl")
+    assert repo.lookup_nearest(db, "meituan", "兰州拉面")["jd"]["store_id"] == "JBAD"
+    repo.mark_jd_deeplink_invalid(db, "JBAD")
+    assert "jd" not in repo.lookup_nearest(db, "meituan", "兰州拉面")
+
+
+def test_invalidate_endpoint_jd_marks_and_lookup_miss(client, db, monkeypatch):
+    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", _SECRET)
+    _mk_row(db, trace_id="j1", name_meituan="店J", name_jd="店J", id_jd="JBADX", jd_deeplink="dl")
+    _mk_row(db, trace_id="j2", name_meituan="店J", name_jd="店J", id_jd="JBADX", jd_deeplink="dl2")
+    r = client.post("/internal/store-mapping/invalidate",
+                    json={"platform": "jd", "shop_id": "JBADX"},
+                    headers={"X-Internal-Secret": _SECRET})
+    assert r.status_code == 200 and r.json()["affected"] == 2
+    lk = client.get("/internal/store-mapping/lookup",
+                    params={"source_platform": "meituan", "name": "店J"},
+                    headers={"X-Internal-Secret": _SECRET})
+    assert "jd" not in lk.json()