feat(store_mapping): 淘宝缓存 deeplink 失效检测→标记→回退(app-server 侧)

加 taobao_deeplink_invalid_at 列(迁移+model);mark_taobao_deeplink_invalid 按 shopId 标记所有行;lookup_nearest 过滤失效淘宝候选;新增 POST /internal/store-mapping/invalidate。+ 7 单测。

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

alembic/versions/coupon_engage_per_package.py

@@ -0,0 +1,53 @@
+"""coupon_prompt_engagement 频控加 package 维度(美团/淘宝/京东各自独立弹)
+
+Revision ID: coupon_engage_per_package
+Revises: store_mapping_jd_cols
+Create Date: 2026-06-14 11:00:00.000000
+
+需求:领券引导窗按 (device, App, 自然日) 频控——在美团弹过/领过,不影响淘宝、京东今天
+仍各弹一次。原表唯一键是 (device_id, engage_date),缺 package → 任一 App 弹过就把整台
+设备当天标记 engage,其余 App 被压住不弹(bug)。
+
+本迁移:
+  1. 加 package 列(NOT NULL,旧行用 server_default "" 填占位,不影响新逻辑判断)。
+  2. 旧唯一约束 (device_id, engage_date) → 新 (device_id, package, engage_date)。
+
+SQLite 不支持直接 drop/add 约束,用 batch_alter_table(建临时表 + 拷数据 + 换名,
+与 store_mapping_* 同款)。
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'coupon_engage_per_package'
+down_revision: Union[str, Sequence[str], None] = 'store_mapping_jd_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('coupon_prompt_engagement', schema=None) as batch_op:
+        # 加 package 列。旧行(改造前的全局记录)填 "" 占位:它们对应的是"老的全局态",
+        # 新逻辑按 (device, package, 日) 判,占位 "" 不会与真实包名(com.xxx)碰撞。
+        batch_op.add_column(
+            sa.Column('package', sa.String(length=64), nullable=False, server_default='')
+        )
+        # 旧唯一约束 (device_id, engage_date) → 新三元组 (device_id, package, engage_date)。
+        batch_op.drop_constraint('uq_coupon_engage_device_date', type_='unique')
+        batch_op.create_unique_constraint(
+            'uq_coupon_engage_device_pkg_date',
+            ['device_id', 'package', 'engage_date'],
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table('coupon_prompt_engagement', schema=None) as batch_op:
+        batch_op.drop_constraint('uq_coupon_engage_device_pkg_date', type_='unique')
+        batch_op.create_unique_constraint(
+            'uq_coupon_engage_device_date',
+            ['device_id', 'engage_date'],
+        )
+        batch_op.drop_column('package')

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')

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')

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')

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/admin/repositories/ad_audit.py

@@ -1,16 +1,18 @@
 """看广告金币审计:复算 expected_coin 并与实发对比。
 
 只读。复用 [app.core.rewards] 的公式函数(不另写公式,避免与正式发奖口径漂移):
-- 看视频:每条 granted = 1 份,第 N 份 = 当日该用户 granted 的 reward_video 顺序号
+- 看视频:每条 granted = 1 份,第 N 份 = 该用户 granted 的 reward_video **账号累计**顺序号
-  (与 ad_reward.grant_ad_reward 里 `_granted_today + 1` 一致)。
+  (与 ad_reward.grant_ad_reward 里 `_granted_cumulative + 1` 一致;LT 因子不按天重置,
-- 信息流:每条按 unit_count 份逐份累加,LT 序号 = 当日该用户已 granted 份数累计
+  故复算时要把当日序号叠加上该用户在本日**之前**的累计已发份数)。
-  (与 ad_feed_reward._unit_reward_total 的 existing_units 一致)。
+- 信息流:每条按 unit_count 份逐份累加,LT 序号 = 该用户 granted 份数**账号累计**
+  (与 ad_feed_reward._unit_reward_total 的 existing_units 一致;同样不按天重置,
+  复算需叠加本日之前的累计份数)。
 
 非 granted(capped/ecpm_missing)不占用份序号、应发恒 0,据此校验闸口是否确实没发。
 """
 from __future__ import annotations
 
-from sqlalchemy import select
+from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
 from app.core import rewards
@@ -19,22 +21,42 @@ from app.models.ad_reward import AdRewardRecord
 from app.repositories.ad_feed_reward import FEED_REWARD_UNIT_SECONDS
 
 
+def _prior_granted_counts(
+    db: Session, *, date: str, user_id: int | None
+) -> dict[int, int]:
+    """各用户在 date **之前**已发奖的 reward_video 累计份数,作为当日复算的 LT 序号起点。
+    LT 因子改账号累计后,当日第 1 份并非全局第 1 份,需叠加历史累计。"""
+    stmt = (
+        select(AdRewardRecord.user_id, func.count())
+        .where(
+            AdRewardRecord.reward_date < date,
+            AdRewardRecord.reward_scene == "reward_video",
+            AdRewardRecord.status == "granted",
+        )
+        .group_by(AdRewardRecord.user_id)
+    )
+    if user_id is not None:
+        stmt = stmt.where(AdRewardRecord.user_id == user_id)
+    return {uid: n for uid, n in db.execute(stmt).all()}
+
+
 def _reward_video_rows(
     db: Session, *, date: str, user_id: int | None
 ) -> list[dict]:
-    """看视频记录复算。按 (user_id, created_at) 升序还原当日第 N 份。"""
+    """看视频记录复算。按 (user_id, created_at) 升序还原账号累计第 N 份(含本日之前的累计)。"""
     stmt = (
         select(AdRewardRecord)
         .where(
             AdRewardRecord.reward_date == date,
             AdRewardRecord.reward_scene == "reward_video",
         )
-        .order_by(AdRewardRecord.user_id, AdRewardRecord.created_at)
+        .order_by(AdRewardRecord.user_id, AdRewardRecord.created_at, AdRewardRecord.id)
     )
     if user_id is not None:
         stmt = stmt.where(AdRewardRecord.user_id == user_id)
 
-    granted_n: dict[int, int] = {}  # user_id -> 已 granted 份数
+    # 用本日之前的累计份数做起点,当日 granted 在其上继续递增 → 与 _granted_cumulative+1 对齐
+    granted_n: dict[int, int] = _prior_granted_counts(db, date=date, user_id=user_id)
     rows: list[dict] = []
     for rec in db.execute(stmt).scalars():
         if rec.status == "granted":
@@ -80,17 +102,38 @@ def _reward_video_rows(
     return rows
 
 
+def _feed_prior_granted_units(
+    db: Session, *, date: str, user_id: int | None
+) -> dict[int, int]:
+    """各用户在 date **之前** granted 的信息流份数累计,作为当日复算的 LT 序号起点。"""
+    stmt = (
+        select(
+            AdFeedRewardRecord.user_id,
+            func.coalesce(func.sum(AdFeedRewardRecord.unit_count), 0),
+        )
+        .where(
+            AdFeedRewardRecord.reward_date < date,
+            AdFeedRewardRecord.status == "granted",
+        )
+        .group_by(AdFeedRewardRecord.user_id)
+    )
+    if user_id is not None:
+        stmt = stmt.where(AdFeedRewardRecord.user_id == user_id)
+    return {uid: int(n) for uid, n in db.execute(stmt).all()}
+
+
 def _feed_rows(db: Session, *, date: str, user_id: int | None) -> list[dict]:
-    """信息流记录复算。granted 记录逐份累加,LT 序号沿用当日累计份数。"""
+    """信息流记录复算。granted 记录逐份累加,LT 序号沿用账号累计份数(含本日之前)。"""
     stmt = (
         select(AdFeedRewardRecord)
         .where(AdFeedRewardRecord.reward_date == date)
-        .order_by(AdFeedRewardRecord.user_id, AdFeedRewardRecord.created_at)
+        .order_by(AdFeedRewardRecord.user_id, AdFeedRewardRecord.created_at, AdFeedRewardRecord.id)
     )
     if user_id is not None:
         stmt = stmt.where(AdFeedRewardRecord.user_id == user_id)
 
-    granted_units: dict[int, int] = {}  # user_id -> 已 granted 份数累计
+    # 本日之前的累计份数做起点,与 _unit_reward_total 的 existing_units(累计)对齐
+    granted_units: dict[int, int] = _feed_prior_granted_units(db, date=date, user_id=user_id)
     rows: list[dict] = []
     for rec in db.execute(stmt).scalars():
         if rec.status == "granted":
@@ -142,20 +185,37 @@ def _feed_rows(db: Session, *, date: str, user_id: int | None) -> list[dict]:
 
 
 def ad_coin_audit(
-    db: Session, *, date: str, user_id: int | None, scene: str | None, limit: int
+    db: Session,
-) -> list[dict]:
+    *,
-    """返回当日发奖复算明细,按 created_at 倒序(最新在前)截断到 limit。
+    date: str,
+    user_id: int | None,
+    scene: str | None,
+    limit: int,
+    only_mismatch: bool = False,
+) -> dict:
+    """当日发奖复算。返回 {total, mismatch_count, truncated, items}。
 
-    scene: None=两类都要 / "reward_video" / "feed"。
+    scene: None=两类都要 / "reward_video" / "feed";only_mismatch=True 只展示不一致(✗)行。
-    份序号在截断前已基于全天数据算好,故 limit 只影响展示条数、不影响 expected 复算正确性。
+    关键:`total` 与 `mismatch_count` 在**全量**(截断前)上统计,故对账数字始终可信,不受 limit
+    影响;`items` 才是展示集(only_mismatch 时只取 ✗ 行)按 created_at 倒序截断到 limit。
+    份序号在全天数据上已算好,limit 只影响展示条数、不影响 expected 复算正确性。
     """
     rows: list[dict] = []
     if scene in (None, "reward_video"):
         rows.extend(_reward_video_rows(db, date=date, user_id=user_id))
     if scene in (None, "feed"):
         rows.extend(_feed_rows(db, date=date, user_id=user_id))
-    rows.sort(key=lambda r: r["created_at"], reverse=True)
+    rows.sort(key=lambda r: (r["created_at"], r["record_id"]), reverse=True)
-    return rows[:limit]
+
+    total = len(rows)
+    mismatch_count = sum(1 for r in rows if not r["matched"])
+    display = [r for r in rows if not r["matched"]] if only_mismatch else rows
+    return {
+        "total": total,
+        "mismatch_count": mismatch_count,
+        "truncated": len(display) > limit,
+        "items": display[:limit],
+    }
 
 
 def formula_snapshot() -> dict:

app/admin/repositories/audit_log.py

@@ -4,7 +4,7 @@
 """
 from __future__ import annotations
 
-from sqlalchemy import select
+from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
 from app.models.admin import AdminAuditLog
@@ -49,8 +49,9 @@ def list_audit_logs(
     admin_id: int | None = None,
     limit: int = 50,
     cursor: int | None = None,
-) -> tuple[list[AdminAuditLog], int | None]:
+) -> tuple[list[AdminAuditLog], int | None, int]:
-    """游标分页(id 倒序),与现有 list_* 约定一致。返回 (rows, next_cursor)。"""
+    """offset 分页(id 倒序)+ total。cursor 即 offset((page-1)*pageSize),支持页码跳页。
+    返回 (rows, next_cursor, total)。"""
     stmt = select(AdminAuditLog)
     if action:
         stmt = stmt.where(AdminAuditLog.action == action)
@@ -58,13 +59,15 @@ def list_audit_logs(
         stmt = stmt.where(AdminAuditLog.target_type == target_type)
     if admin_id is not None:
         stmt = stmt.where(AdminAuditLog.admin_id == admin_id)
-    if cursor is not None:
+
-        stmt = stmt.where(AdminAuditLog.id < cursor)
+    total = int(db.execute(select(func.count()).select_from(stmt.subquery())).scalar_one())
-    stmt = stmt.order_by(AdminAuditLog.id.desc())
+    offset = max(cursor or 0, 0)
-    rows = list(db.execute(stmt.limit(limit + 1)).scalars().all())
+    rows = list(
+        db.execute(
+            stmt.order_by(AdminAuditLog.id.desc()).offset(offset).limit(limit + 1)
+        ).scalars().all()
+    )
     has_more = len(rows) > limit
     items = rows[:limit]
-    # next_cursor 必须是"本页返回的最后一条"的 id(下一页查 id < 它),不能用 rows[limit]——
+    next_cursor = offset + limit if has_more else None
-    # rows[limit] 是探测下一页用的第 limit+1 条,它既不在本页也不在下页 → 每页边界丢一条。
+    return items, next_cursor, total
-    next_cursor = items[-1].id if has_more else None
-    return items, next_cursor

app/admin/repositories/queries.py

@@ -38,15 +38,48 @@ def cursor_paginate(
     return items, next_cursor
 
 
+def offset_paginate(
+    db: Session, stmt: Select, sort_clause: tuple, *, limit: int, cursor: int | None
+) -> tuple[list, int | None, int]:
+    """offset 分页 + 总数。stmt 只含 where/join,不要预先带 order_by/offset/limit。
+
+    cursor 即 offset(页码分页:offset=(page-1)*pageSize)。返回 (items, next_cursor, total):
+    - total:符合筛选条件的总条数(供 antd pagination 渲染页码/共 N 条),count 在 P0 量级开销可忽略;
+    - next_cursor:下一页 offset(兼容「加载更多」),末页为 None。
+    多取 1 条探测下一页。sort_clause 为 order_by 表达式元组(末位应含 id 保证稳定排序)。"""
+    total = int(
+        db.execute(select(func.count()).select_from(stmt.subquery())).scalar_one()
+    )
+    offset = max(cursor or 0, 0)
+    rows = list(
+        db.execute(stmt.order_by(*sort_clause).offset(offset).limit(limit + 1)).scalars().all()
+    )
+    has_more = len(rows) > limit
+    items = rows[:limit]
+    next_cursor = offset + limit if has_more else None
+    return items, next_cursor, total
+
+
 def list_users(
     db: Session,
     *,
     phone: str | None = None,
     register_channel: str | None = None,
     status: str | None = None,
+    nickname: str | None = None,
+    created_from: datetime | None = None,
+    created_to: datetime | None = None,
+    last_login_from: datetime | None = None,
+    last_login_to: datetime | None = None,
+    sort_by: str = "id",
+    sort_order: str = "desc",
     limit: int = 20,
     cursor: int | None = None,
-) -> tuple[list[User], int | None]:
+) -> tuple[list[User], int | None, int]:
+    """用户列表(admin 全量)。支持手机号前缀 / 渠道 / 状态 / 昵称模糊 / 注册·最近登录时间范围筛选,
+    按 id·注册时间·最近登录排序。**offset 分页**(cursor=offset):任意列排序下游标语义统一,
+    代价是翻页期间数据变动可能错位一条——admin 低频场景可接受(同 [list_all_withdraw_orders])。
+    日期入参统一转 tz-aware UTC 比较(列为 timestamptz,见 _as_utc)。"""
     stmt = select(User)
     if phone:
         stmt = stmt.where(User.phone.like(f"{phone}%"))  # 前缀匹配
@@ -54,7 +87,26 @@ def list_users(
         stmt = stmt.where(User.register_channel == register_channel)
     if status:
         stmt = stmt.where(User.status == status)
-    return cursor_paginate(db, stmt, User.id, limit=limit, cursor=cursor)
+    if nickname and nickname.strip():
+        stmt = stmt.where(User.nickname.ilike(f"%{nickname.strip()}%"))
+    if created_from is not None:
+        stmt = stmt.where(User.created_at >= _as_utc(created_from))
+    if created_to is not None:
+        stmt = stmt.where(User.created_at <= _as_utc(created_to))
+    if last_login_from is not None:
+        stmt = stmt.where(User.last_login_at >= _as_utc(last_login_from))
+    if last_login_to is not None:
+        stmt = stmt.where(User.last_login_at <= _as_utc(last_login_to))
+
+    sort_cols = {
+        "id": User.id,
+        "created_at": User.created_at,
+        "last_login_at": User.last_login_at,
+    }
+    sort_col = sort_cols.get(sort_by, User.id)
+    order_fn = asc if sort_order == "asc" else desc
+    id_order = asc(User.id) if sort_order == "asc" else desc(User.id)
+    return offset_paginate(db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor)
 
 
 def list_onboarding_devices(db: Session, *, limit: int = 500) -> list[dict]:
@@ -123,7 +175,7 @@ def list_all_withdraw_orders(
     quick_filter: str | None = None,
     limit: int = 20,
     cursor: int | None = None,
-) -> tuple[list[WithdrawOrder], int | None]:
+) -> tuple[list[WithdrawOrder], int | None, int]:
     stmt = select(WithdrawOrder)
     needs_user_join = bool(keyword and keyword.strip()) or quick_filter == "high_risk"
     if needs_user_join:
@@ -153,16 +205,16 @@ def list_all_withdraw_orders(
 
     date_col = WithdrawOrder.updated_at if date_field == "updated_at" else WithdrawOrder.created_at
     if date_from is not None:
-        stmt = stmt.where(date_col >= _as_utc_naive(date_from))
+        stmt = stmt.where(date_col >= _as_utc(date_from))
     if date_to is not None:
-        stmt = stmt.where(date_col <= _as_utc_naive(date_to))
+        stmt = stmt.where(date_col <= _as_utc(date_to))
 
-    now = datetime.now(timezone.utc).replace(tzinfo=None)
+    # tz-aware:列为 timestamptz,比较绝对时刻、与 DB 会话时区无关(同 _as_utc / stats.py)
+    now = datetime.now(timezone.utc)
     today_start = (
         datetime.now(ZoneInfo("Asia/Shanghai"))
         .replace(hour=0, minute=0, second=0, microsecond=0)
         .astimezone(timezone.utc)
-        .replace(tzinfo=None)
     )
     if quick_filter == "abnormal":
         stmt = stmt.where(
@@ -207,6 +259,53 @@ def list_all_withdraw_orders(
     sort_col = sort_cols.get(sort_by, WithdrawOrder.created_at)
     order_fn = asc if sort_order == "asc" else desc
     id_order = asc(WithdrawOrder.id) if sort_order == "asc" else desc(WithdrawOrder.id)
+    return offset_paginate(db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor)
+
+
+def _as_utc(value: datetime) -> datetime:
+    """前端传 ISO 时间 → 统一成 tz-aware UTC 再比较。
+
+    所有时间列均为 `DateTime(timezone=True)`(Postgres timestamptz);用 tz-aware 绑定参数
+    比较的是绝对时刻,与 DB 会话时区无关、恒正确。曾用 naive UTC,正确性依赖会话 TimeZone=UTC,
+    生产会话非 UTC 时筛选边界会整体偏移——故统一 tz-aware(与 stats.py / withdraw_summary 一致)。
+    无时区入参按 UTC 解释。"""
+    if value.tzinfo is None:
+        return value.replace(tzinfo=timezone.utc)
+    return value.astimezone(timezone.utc)
+
+
+def list_feedbacks(
+    db: Session,
+    *,
+    status: str | None = None,
+    user_id: int | None = None,
+    content: str | None = None,
+    created_from: datetime | None = None,
+    created_to: datetime | None = None,
+    sort_by: str = "id",
+    sort_order: str = "desc",
+    limit: int = 20,
+    cursor: int | None = None,
+) -> tuple[list[Feedback], int | None]:
+    """反馈工单列表。支持 状态 / 用户ID / 内容模糊 / 提交时间范围 筛选,按 id·提交时间排序。
+    **offset 分页**(cursor=offset):任意列排序下游标语义统一(同 [list_users]),代价是翻页期间
+    数据变动可能错位一条——admin 低频场景可接受。created_at 为 timestamptz,日期入参统一转 tz-aware UTC 比较。"""
+    stmt = select(Feedback)
+    if status:
+        stmt = stmt.where(Feedback.status == status)
+    if user_id is not None:
+        stmt = stmt.where(Feedback.user_id == user_id)
+    if content and content.strip():
+        stmt = stmt.where(Feedback.content.ilike(f"%{content.strip()}%"))
+    if created_from is not None:
+        stmt = stmt.where(Feedback.created_at >= _as_utc(created_from))
+    if created_to is not None:
+        stmt = stmt.where(Feedback.created_at <= _as_utc(created_to))
+
+    sort_cols = {"id": Feedback.id, "created_at": Feedback.created_at}
+    sort_col = sort_cols.get(sort_by, Feedback.id)
+    order_fn = asc if sort_order == "asc" else desc
+    id_order = asc(Feedback.id) if sort_order == "asc" else desc(Feedback.id)
     stmt = stmt.order_by(order_fn(sort_col), id_order)
 
     offset = max(cursor or 0, 0)
@@ -217,29 +316,6 @@ def list_all_withdraw_orders(
     return items, next_cursor
 
 
-def _as_utc_naive(value: datetime) -> datetime:
-    """前端传 ISO 时间;DB 当前按 UTC naive 比较最稳(SQLite/本地开发一致)。"""
-    if value.tzinfo is None:
-        return value
-    return value.astimezone(timezone.utc).replace(tzinfo=None)
-
-
-def list_feedbacks(
-    db: Session,
-    *,
-    status: str | None = None,
-    user_id: int | None = None,
-    limit: int = 20,
-    cursor: int | None = None,
-) -> tuple[list[Feedback], int | None]:
-    stmt = select(Feedback)
-    if status:
-        stmt = stmt.where(Feedback.status == status)
-    if user_id is not None:
-        stmt = stmt.where(Feedback.user_id == user_id)
-    return cursor_paginate(db, stmt, Feedback.id, limit=limit, cursor=cursor)
-
-
 def get_withdraw_by_out_bill_no(db: Session, out_bill_no: str) -> WithdrawOrder | None:
     """按商户单号查提现单(admin 重试打款先拿 user_id 用,M3)。"""
     return db.execute(
@@ -426,14 +502,14 @@ def list_price_reports(
     user_id: int | None = None,
     limit: int = 20,
     cursor: int | None = None,
-) -> tuple[list[PriceReport], int | None]:
+) -> tuple[list[PriceReport], int | None, int]:
-    """上报更低价列表(admin 全量,可按状态/用户筛)。游标同 feedback:id 倒序。"""
+    """上报更低价列表(admin 全量,可按状态/用户筛)。offset 分页 + total,id 倒序。"""
     stmt = select(PriceReport)
     if status:
         stmt = stmt.where(PriceReport.status == status)
     if user_id is not None:
         stmt = stmt.where(PriceReport.user_id == user_id)
-    return cursor_paginate(db, stmt, PriceReport.id, limit=limit, cursor=cursor)
+    return offset_paginate(db, stmt, (PriceReport.id.desc(),), limit=limit, cursor=cursor)
 
 
 def price_report_summary(db: Session) -> dict:

app/admin/routers/ad_audit.py

@@ -29,16 +29,20 @@ def get_ad_coin_audit(
         str | None, Query(description="reward_video / feed;不传=两类都要")
     ] = None,
     limit: Annotated[int, Query(ge=1, le=500)] = 100,
+    only_mismatch: Annotated[
+        bool, Query(description="只看不一致(✗)行;统计数仍按全量,不受影响")
+    ] = False,
 ) -> AdCoinAuditOut:
     audit_date = date or cn_today().isoformat()
-    rows = ad_audit.ad_coin_audit(
+    result = ad_audit.ad_coin_audit(
         db, date=audit_date, user_id=user_id, scene=scene, limit=limit,
+        only_mismatch=only_mismatch,
     )
-    items = [AdCoinAuditRow(**r) for r in rows]
     return AdCoinAuditOut(
         date=audit_date,
         formula=AdCoinFormulaOut(**ad_audit.formula_snapshot()),
-        total=len(items),
+        total=result["total"],
-        mismatch_count=sum(1 for it in items if not it.matched),
+        mismatch_count=result["mismatch_count"],
-        items=items,
+        truncated=result["truncated"],
+        items=[AdCoinAuditRow(**r) for r in result["items"]],
     )

app/admin/routers/admins.py

@@ -17,6 +17,12 @@ router = APIRouter(
 )
 
 
+def _active_super_count(db: AdminDb) -> int:
+    return sum(
+        1 for a in admin_repo.list_admins(db) if a.role == "super_admin" and a.status == "active"
+    )
+
+
 @router.get("", response_model=list[AdminOut], summary="管理员列表")
 def list_admins(db: AdminDb) -> list[AdminOut]:
     return [AdminOut.model_validate(a) for a in admin_repo.list_admins(db)]
@@ -48,16 +54,29 @@ def update_admin(
     if admin_id == admin.id and body.status == "disabled":
         raise HTTPException(status_code=400, detail="不能禁用自己")
 
+    # 防自锁:降级 / 禁用某个 super_admin 前,确认操作后仍至少剩 1 个 active super_admin,
+    # 否则会进入「零可用超管」死局——本路由仅 super 可进,只能改库恢复。
+    demotes_super = (
+        target.role == "super_admin"
+        and target.status == "active"
+        and (
+            (body.role is not None and body.role != "super_admin")
+            or body.status == "disabled"
+        )
+    )
+    if demotes_super and _active_super_count(db) <= 1:
+        raise HTTPException(status_code=400, detail="不能降级/禁用最后一个超级管理员")
+
     changes: dict = {}
-    if body.role is not None:
+    if body.role is not None and body.role != target.role:
+        changes["role"] = {"before": target.role, "after": body.role}
         target.role = body.role
-        changes["role"] = body.role
+    if body.status is not None and body.status != target.status:
-    if body.status is not None:
+        changes["status"] = {"before": target.status, "after": body.status}
         target.status = body.status
-        changes["status"] = body.status
     if body.password is not None:
-        target.password_hash = hash_password(body.password)
         changes["password"] = "reset"
+        target.password_hash = hash_password(body.password)
     if not changes:
         raise HTTPException(status_code=400, detail="无任何变更字段")
     db.commit()

app/admin/routers/audit.py

@@ -26,9 +26,11 @@ def list_audit_logs(
     limit: Annotated[int, Query(ge=1, le=100)] = 50,
     cursor: Annotated[int | None, Query()] = None,
 ) -> CursorPage[AdminAuditLogOut]:
-    items, next_cursor = audit_repo.list_audit_logs(
+    items, next_cursor, total = audit_repo.list_audit_logs(
         db, action=action, target_type=target_type, admin_id=admin_id, limit=limit, cursor=cursor,
     )
     return CursorPage(
-        items=[AdminAuditLogOut.model_validate(x) for x in items], next_cursor=next_cursor,
+        items=[AdminAuditLogOut.model_validate(x) for x in items],
+        next_cursor=next_cursor,
+        total=total,
     )

app/admin/routers/feedback.py

@@ -1,6 +1,7 @@
-"""admin 反馈工单:列表(读)+ 标记已处理(写,带审计)。"""
+"""admin 反馈工单:列表(读,支持 状态/用户ID/内容/时间 筛选 + 排序)+ 标记已处理(写,带审计)。"""
 from __future__ import annotations
 
+from datetime import datetime
 from typing import Annotated
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Request
@@ -25,11 +26,25 @@ def list_feedbacks(
     db: AdminDb,
     status: Annotated[str | None, Query()] = None,
     user_id: Annotated[int | None, Query()] = None,
+    content: Annotated[str | None, Query(max_length=100)] = None,
+    created_from: Annotated[datetime | None, Query()] = None,
+    created_to: Annotated[datetime | None, Query()] = None,
+    sort_by: Annotated[str, Query(pattern="^(id|created_at)$")] = "id",
+    sort_order: Annotated[str, Query(pattern="^(asc|desc)$")] = "desc",
     limit: Annotated[int, Query(ge=1, le=100)] = 20,
     cursor: Annotated[int | None, Query()] = None,
 ) -> CursorPage[FeedbackOut]:
     items, next_cursor = queries.list_feedbacks(
-        db, status=status, user_id=user_id, limit=limit, cursor=cursor,
+        db,
+        status=status,
+        user_id=user_id,
+        content=content,
+        created_from=created_from,
+        created_to=created_to,
+        sort_by=sort_by,
+        sort_order=sort_order,
+        limit=limit,
+        cursor=cursor,
     )
     return CursorPage(
         items=[FeedbackOut.model_validate(f) for f in items], next_cursor=next_cursor,

app/admin/routers/price_report.py

@@ -40,11 +40,13 @@ def list_price_reports(
     limit: Annotated[int, Query(ge=1, le=100)] = 20,
     cursor: Annotated[int | None, Query()] = None,
 ) -> CursorPage[PriceReportOut]:
-    items, next_cursor = queries.list_price_reports(
+    items, next_cursor, total = queries.list_price_reports(
         db, status=status, user_id=user_id, limit=limit, cursor=cursor,
     )
     return CursorPage(
-        items=[PriceReportOut.model_validate(r) for r in items], next_cursor=next_cursor,
+        items=[PriceReportOut.model_validate(r) for r in items],
+        next_cursor=next_cursor,
+        total=total,
     )
 
 
@@ -60,7 +62,9 @@ def approve_price_report(
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
 ) -> OkResponse:
-    rep = db.get(PriceReport, report_id)
+    # 行锁(SELECT FOR UPDATE):并发/连点双请求会都读到 pending → 各发一次金币双倍发奖,
+    # 锁住该行串行化,第二个请求拿锁后看到 approved → 走 400。SQLite 下 FOR UPDATE 为 no-op。
+    rep = db.get(PriceReport, report_id, with_for_update=True)
     if rep is None:
         raise HTTPException(status_code=404, detail="上报记录不存在")
     if rep.status != "pending":
@@ -88,7 +92,7 @@ def reject_price_report(
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
 ) -> OkResponse:
-    rep = db.get(PriceReport, report_id)
+    rep = db.get(PriceReport, report_id, with_for_update=True)  # 行锁,同 approve(防并发重复审核)
     if rep is None:
         raise HTTPException(status_code=404, detail="上报记录不存在")
     if rep.status != "pending":

app/admin/routers/users.py

@@ -1,6 +1,7 @@
 """admin 用户管理:列表 + 360 详情(读)+ 封禁/解封 + 手动调金币(写,带审计)。"""
 from __future__ import annotations
 
+from datetime import datetime
 from typing import Annotated
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Request
@@ -28,22 +29,32 @@ router = APIRouter(
 )
 
 
-@router.get("", response_model=CursorPage[AdminUserListItem], summary="用户列表(筛选+分页)")
+@router.get("", response_model=CursorPage[AdminUserListItem], summary="用户列表(筛选+排序+分页)")
 def list_users(
     db: AdminDb,
     phone: Annotated[str | None, Query()] = None,
     register_channel: Annotated[str | None, Query()] = None,
     status: Annotated[str | None, Query()] = None,
+    nickname: Annotated[str | None, Query(max_length=100)] = None,
+    created_from: Annotated[datetime | None, Query()] = None,
+    created_to: Annotated[datetime | None, Query()] = None,
+    last_login_from: Annotated[datetime | None, Query()] = None,
+    last_login_to: Annotated[datetime | None, Query()] = None,
+    sort_by: Annotated[str, Query(pattern="^(id|created_at|last_login_at)$")] = "id",
+    sort_order: Annotated[str, Query(pattern="^(asc|desc)$")] = "desc",
     limit: Annotated[int, Query(ge=1, le=100)] = 20,
     cursor: Annotated[int | None, Query()] = None,
 ) -> CursorPage[AdminUserListItem]:
-    items, next_cursor = queries.list_users(
+    items, next_cursor, total = queries.list_users(
         db, phone=phone, register_channel=register_channel, status=status,
-        limit=limit, cursor=cursor,
+        nickname=nickname, created_from=created_from, created_to=created_to,
+        last_login_from=last_login_from, last_login_to=last_login_to,
+        sort_by=sort_by, sort_order=sort_order, limit=limit, cursor=cursor,
     )
     return CursorPage(
         items=[AdminUserListItem.model_validate(u) for u in items],
         next_cursor=next_cursor,
+        total=total,
     )
 
 
@@ -101,7 +112,7 @@ def set_user_debug_trace(
     return OkResponse()
 
 
-@router.post("/{user_id}/coins", response_model=OkResponse, summary="手动增减金币(带审计)")
+@router.post("/{user_id}/coins", response_model=OkResponse, summary="手动增减/设值金币(带审计)")
 def grant_user_coins(
     user_id: int,
     body: GrantCoinsRequest,
@@ -109,33 +120,46 @@ def grant_user_coins(
     admin: Annotated[AdminUser, Depends(require_role("finance"))],
     db: AdminDb,
 ) -> OkResponse:
-    if body.amount == 0:
-        raise HTTPException(status_code=400, detail="amount 不能为 0")
     user = user_repo.get_user_by_id(db, user_id)
     if user is None:
         raise HTTPException(status_code=404, detail="用户不存在")
-    # 负数扣减时不允许扣成负余额(运营误操作保护)
+    # set=设为目标值:读当前余额算出要写的差值,仍复用 grant_coins 写一笔流水(沿用原子/审计/扣负保护)
-    if body.amount < 0:
+    if body.mode == "set":
-        acc_now = wallet_repo.get_or_create_account(db, user_id, commit=False)
+        if body.amount < 0:
-        if acc_now.coin_balance + body.amount < 0:
+            raise HTTPException(status_code=400, detail="目标金币值不能为负")
-            raise HTTPException(
+        # lock=True:锁账户行,防连点/并发各读同一 before 算同一 delta 双写,余额错位
-                status_code=400, detail=f"扣减后金币为负(当前余额 {acc_now.coin_balance})"
+        before = wallet_repo.get_or_create_account(db, user_id, commit=False, lock=True).coin_balance
-            )
+        delta = body.amount - before
-    biz_type = "admin_grant" if body.amount > 0 else "admin_deduct"
+        if delta == 0:
+            raise HTTPException(status_code=400, detail=f"当前金币已为 {body.amount},无需调整")
+    else:
+        if body.amount == 0:
+            raise HTTPException(status_code=400, detail="amount 不能为 0")
+        delta = body.amount
+        # 负数扣减时不允许扣成负余额(运营误操作保护);lock=True 防并发扣穿
+        if delta < 0:
+            acc_now = wallet_repo.get_or_create_account(db, user_id, commit=False, lock=True)
+            if acc_now.coin_balance + delta < 0:
+                raise HTTPException(
+                    status_code=400, detail=f"扣减后金币为负(当前余额 {acc_now.coin_balance})"
+                )
+    biz_type = "admin_grant" if delta > 0 else "admin_deduct"
     # grant_coins 只 flush 不 commit;审计同 commit=False;最后一起 commit → 原子(改钱+留痕)
     acc, _ = wallet_repo.grant_coins(
-        db, user_id, body.amount, biz_type=biz_type, remark=f"admin:{body.reason}"[:128],
+        db, user_id, delta, biz_type=biz_type, remark=f"admin:{body.reason}"[:128],
     )
+    detail = {"amount": delta, "balance_after": acc.coin_balance, "reason": body.reason}
+    if body.mode == "set":
+        detail.update({"mode": "set", "target": body.amount, "before": before})
     write_audit(
         db, admin, action="user.coins.grant", target_type="user", target_id=user_id,
-        detail={"amount": body.amount, "balance_after": acc.coin_balance, "reason": body.reason},
+        detail=detail, ip=get_client_ip(request), commit=False,
-        ip=get_client_ip(request), commit=False,
     )
     db.commit()
     return OkResponse()
 
 
-@router.post("/{user_id}/cash", response_model=OkResponse, summary="手动增减现金(带审计)")
+@router.post("/{user_id}/cash", response_model=OkResponse, summary="手动增减/设值现金(带审计)")
 def grant_user_cash(
     user_id: int,
     body: GrantCashRequest,
@@ -143,32 +167,50 @@ def grant_user_cash(
     admin: Annotated[AdminUser, Depends(require_role("finance"))],
     db: AdminDb,
 ) -> OkResponse:
-    """给指定用户增/减现金(分)。正=发放、负=扣减;主要用于让无现金用户直接测试提现。"""
+    """给指定用户增/减或设值现金(分)。delta:正=发放、负=扣减;set:直接设为目标值。
-    if body.amount_cents == 0:
+    主要用于让无现金用户直接测试提现。"""
-        raise HTTPException(status_code=400, detail="amount_cents 不能为 0")
     user = user_repo.get_user_by_id(db, user_id)
     if user is None:
         raise HTTPException(status_code=404, detail="用户不存在")
-    # 负数扣减时不允许扣成负余额(运营误操作保护)
+    # set=设为目标值:读当前余额算差值,仍复用 grant_cash 写一笔流水(沿用原子/审计/扣负保护)
-    if body.amount_cents < 0:
+    if body.mode == "set":
-        acc_now = wallet_repo.get_or_create_account(db, user_id, commit=False)
+        if body.amount_cents < 0:
-        if acc_now.cash_balance_cents + body.amount_cents < 0:
+            raise HTTPException(status_code=400, detail="目标现金值不能为负")
+        # lock=True:锁账户行,防连点/并发各读同一 before 算同一 delta 双写,余额错位
+        before = wallet_repo.get_or_create_account(
+            db, user_id, commit=False, lock=True
+        ).cash_balance_cents
+        delta = body.amount_cents - before
+        if delta == 0:
             raise HTTPException(
-                status_code=400, detail=f"扣减后现金为负(当前余额 {acc_now.cash_balance_cents} 分)"
+                status_code=400, detail=f"当前现金已为 {body.amount_cents} 分,无需调整"
             )
-    biz_type = "admin_grant" if body.amount_cents > 0 else "admin_deduct"
+    else:
+        if body.amount_cents == 0:
+            raise HTTPException(status_code=400, detail="amount_cents 不能为 0")
+        delta = body.amount_cents
+        # 负数扣减时不允许扣成负余额(运营误操作保护);lock=True 防并发扣穿
+        if delta < 0:
+            acc_now = wallet_repo.get_or_create_account(db, user_id, commit=False, lock=True)
+            if acc_now.cash_balance_cents + delta < 0:
+                raise HTTPException(
+                    status_code=400, detail=f"扣减后现金为负(当前余额 {acc_now.cash_balance_cents} 分)"
+                )
+    biz_type = "admin_grant" if delta > 0 else "admin_deduct"
     # grant_cash 只 flush 不 commit;审计同 commit=False;最后一起 commit → 原子(改钱+留痕)
     acc, _ = wallet_repo.grant_cash(
-        db, user_id, body.amount_cents, biz_type=biz_type, remark=f"admin:{body.reason}"[:128],
+        db, user_id, delta, biz_type=biz_type, remark=f"admin:{body.reason}"[:128],
     )
+    detail = {
+        "amount_cents": delta,
+        "balance_after_cents": acc.cash_balance_cents,
+        "reason": body.reason,
+    }
+    if body.mode == "set":
+        detail.update({"mode": "set", "target_cents": body.amount_cents, "before_cents": before})
     write_audit(
         db, admin, action="user.cash.grant", target_type="user", target_id=user_id,
-        detail={
+        detail=detail, ip=get_client_ip(request), commit=False,
-            "amount_cents": body.amount_cents,
-            "balance_after_cents": acc.cash_balance_cents,
-            "reason": body.reason,
-        },
-        ip=get_client_ip(request), commit=False,
     )
     db.commit()
     return OkResponse()

app/admin/routers/withdraw.py

@@ -63,7 +63,7 @@ def list_withdraws(
     limit: Annotated[int, Query(ge=1, le=100)] = 20,
     cursor: Annotated[int | None, Query()] = None,
 ) -> CursorPage[WithdrawOrderOut]:
-    items, next_cursor = queries.list_all_withdraw_orders(
+    items, next_cursor, total = queries.list_all_withdraw_orders(
         db,
         user_id=user_id,
         status=status,
@@ -78,7 +78,9 @@ def list_withdraws(
         cursor=cursor,
     )
     return CursorPage(
-        items=[WithdrawOrderOut.model_validate(o) for o in items], next_cursor=next_cursor,
+        items=[WithdrawOrderOut.model_validate(o) for o in items],
+        next_cursor=next_cursor,
+        total=total,
     )
 
 
@@ -87,7 +89,12 @@ def withdraws_summary(db: AdminDb) -> WithdrawSummaryOut:
     return WithdrawSummaryOut(**queries.withdraw_summary(db))
 
 
-@router.get("/health-check", response_model=WxpayHealthCheckOut, summary="提现配置健康检查")
+@router.get(
+    "/health-check",
+    response_model=WxpayHealthCheckOut,
+    summary="提现配置健康检查",
+    dependencies=[Depends(require_role("finance"))],  # 暴露密钥路径/配置,限财务+super
+)
 def withdraw_health_check() -> WxpayHealthCheckOut:
     private_path = wxpay._resolve_config_path(settings.WXPAY_MCH_PRIVATE_KEY_PATH)  # noqa: SLF001
     public_path = wxpay._resolve_config_path(settings.WXPAY_PUBLIC_KEY_PATH)  # noqa: SLF001
@@ -160,7 +167,7 @@ def withdraw_detail(out_bill_no: str, db: AdminDb) -> WithdrawDetailOut:
             withdraw_success_cents=overview["withdraw_success_cents"],
         )
 
-    recent_withdraws, _ = queries.list_all_withdraw_orders(
+    recent_withdraws, _, _ = queries.list_all_withdraw_orders(
         db, user_id=order.user_id, limit=5, cursor=None,
     )
     recent_cash_transactions, _ = queries.list_all_cash_transactions(

app/admin/schemas/ad_audit.py

@@ -48,10 +48,15 @@ class AdCoinFormulaOut(BaseModel):
 
 
 class AdCoinAuditOut(BaseModel):
-    """审计响应:公式参照 + 命中条数 + 明细。"""
+    """审计响应:公式参照 + 全量统计 + 明细。"""
 
     date: str = Field(..., description="审计日期(北京时间 YYYY-MM-DD)")
     formula: AdCoinFormulaOut
-    total: int = Field(..., description="返回的明细条数")
+    total: int = Field(..., description="该筛选下复算总条数(全量,不受 limit/only_mismatch 影响)")
-    mismatch_count: int = Field(..., description="其中 matched=false 的条数(=0 说明公式全部生效)")
+    mismatch_count: int = Field(
-    items: list[AdCoinAuditRow]
+        ..., description="全量不一致条数(=0 说明公式全部生效;在截断前统计,可信)"
+    )
+    truncated: bool = Field(
+        ..., description="展示集是否被 limit 截断(true=还有未返回的明细,请缩小范围或调大 limit)"
+    )
+    items: list[AdCoinAuditRow] = Field(..., description="展示明细;only_mismatch=true 时只含 ✗ 行")

app/admin/schemas/common.py

@@ -9,10 +9,15 @@ T = TypeVar("T")
 
 
 class CursorPage(BaseModel, Generic[T]):
-    """游标分页响应:items + 下一页游标(next_cursor=None 表示末页)。"""
+    """分页响应:items + 下一页游标(next_cursor=None 表示末页)+ 可选 total。
+
+    next_cursor:offset 分页时即下一页 offset,「加载更多」用;末页为 None。
+    total:符合筛选条件的总条数,页码分页(antd pagination)用;不需要总数的接口可不传(None)。
+    """
 
     items: list[T]
     next_cursor: int | None = None
+    total: int | None = None
 
 
 class OkResponse(BaseModel):

app/admin/schemas/price_report.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 
 from datetime import datetime
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 
 class PriceReportOut(BaseModel):
@@ -36,6 +36,14 @@ class PriceReportOut(BaseModel):
 class PriceReportRejectRequest(BaseModel):
     reason: str = Field(min_length=1, max_length=256, description="拒绝理由,用户端记录页会看到")
 
+    @field_validator("reason")
+    @classmethod
+    def _reason_not_blank(cls, v: str) -> str:
+        # min_length=1 放过纯空白(" "),trim 后再校验非空,避免审计/用户端记录到空理由
+        if not v.strip():
+            raise ValueError("拒绝理由不能为空")
+        return v.strip()
+
 
 class PriceReportSummary(BaseModel):
     """审核台顶部各状态计数。"""

app/admin/schemas/user.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 from datetime import datetime
 from typing import Literal
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 
 class AdminUserListItem(BaseModel):
@@ -37,15 +37,38 @@ class AdminUserOverview(BaseModel):
     feedback_total: int
 
 
+def _strip_reason(v: str) -> str:
+    # min_length=1 放过纯空白(" "),trim 后再校验非空,避免审计记到空原因
+    if not v.strip():
+        raise ValueError("操作原因不能为空")
+    return v.strip()
+
+
 class GrantCoinsRequest(BaseModel):
-    amount: int = Field(..., description="金币变动:正=增加,负=扣减(不可为 0)")
+    mode: Literal["delta", "set"] = Field(
+        "delta", description="delta=增减(amount 为变动量) / set=设为(amount 为目标值,须≥0)"
+    )
+    amount: int = Field(
+        ...,
+        description="delta 模式:金币变动(正=增加,负=扣减,不可为 0);set 模式:目标金币值(须≥0)",
+    )
     reason: str = Field(..., min_length=1, max_length=128, description="操作原因(必填,入审计)")
 
+    _v_reason = field_validator("reason")(_strip_reason)
+
 
 class GrantCashRequest(BaseModel):
-    amount_cents: int = Field(..., description="现金变动(分):正=增加,负=扣减(不可为 0)")
+    mode: Literal["delta", "set"] = Field(
+        "delta", description="delta=增减(amount_cents 为变动量) / set=设为(amount_cents 为目标值,须≥0)"
+    )
+    amount_cents: int = Field(
+        ...,
+        description="delta 模式:现金变动(分,正=增加,负=扣减,不可为 0);set 模式:目标现金值(分,须≥0)",
+    )
     reason: str = Field(..., min_length=1, max_length=128, description="操作原因(必填,入审计)")
 
+    _v_reason = field_validator("reason")(_strip_reason)
+
 
 class SetUserStatusRequest(BaseModel):
     status: Literal["active", "disabled"] = Field(

app/api/internal/store.py

@@ -0,0 +1,107 @@
+"""平台店铺映射内部上报端点(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,
+    StoreMappingInvalidateIn,
+    StoreMappingInvalidateOut,
+    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)
+
+
+@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":
+        # 当前只接淘宝; 其它平台先 no-op(affected=0), 不报错 — 向后兼容 pricebot 将来扩展。
+        logger.info("store_mapping invalidate 跳过: platform=%s 暂不支持", payload.platform)
+        return StoreMappingInvalidateOut(ok=True, affected=0)
+    affected = repo.mark_taobao_deeplink_invalid(db, payload.shop_id)
+    logger.info(
+        "store_mapping invalidate platform=%s shop_id=%s → 标记失效 %d 行",
+        payload.platform, payload.shop_id, affected,
+    )
+    return StoreMappingInvalidateOut(ok=True, affected=affected)

app/api/v1/coupon.py

@@ -27,6 +27,7 @@ from app.schemas.coupon_state import (
     CouponCompletedTodayOut,
     CouponPromptDismissIn,
     CouponPromptShouldShowOut,
+    CouponPromptShownIn,
 )
 
 logger = logging.getLogger("shagua.coupon")
@@ -66,11 +67,11 @@ def _extract_coupon_results(resp_json: dict) -> list[dict]:
 
 
 def _mark_engagement_blocking(
-    device_id: str, user_id: int | None, engage_type: str
+    device_id: str, package: str, user_id: int | None, engage_type: str
 ) -> None:
     """独立 session 写 engagement(async 端点经 run_in_threadpool 调,不阻塞事件循环)。"""
     with SessionLocal() as db:
-        coupon_repo.mark_engagement(db, device_id, user_id, engage_type)
+        coupon_repo.mark_engagement(db, device_id, package, user_id, engage_type)
 
 
 def _record_claims_blocking(
@@ -111,14 +112,17 @@ async def coupon_step(
     device_id = meta.get("device_id")
     user_id = _to_int(meta.get("user_id"))  # 登录态才带;判断不靠它,资产留痕用
     trace_id = meta.get("trace_id")
+    # 发起领券时前台 App 包名(step body 带 "package")。频控按 App,这条 engagement 要记到
+    # 对应 App 上。App 内「去领取」发起时 package 可能缺/为空 → 退化为 "" 占位(全局态)。
+    pkg = meta.get("package") or ""
 
     # 领券任务首帧(step=0)= 用户已发起领券 → 记一条今日 engagement(claim_started),
-    # 今天这台设备不再弹引导窗(对齐前台"点一键领取即 markEngaged")。写库失败绝不能
+    # 今天**这个 App** 不再弹引导窗(对齐前台"点一键领取即 markEngaged")。写库失败绝不能
     # 连累领券主流程,整段吞掉。
     if device_id and meta.get("step") == 0:
         try:
             await run_in_threadpool(
-                _mark_engagement_blocking, device_id, user_id, "claim_started"
+                _mark_engagement_blocking, device_id, pkg, user_id, "claim_started"
             )
         except Exception as e:  # noqa: BLE001
             logger.warning("coupon engagement write failed: %s", e)
@@ -188,14 +192,29 @@ async def coupon_step(
     return resp_json
 
 
+@router.post("/prompt/shown", summary="领券引导窗弹出即上报(按 App 记 shown)")
+def coupon_prompt_shown(payload: CouponPromptShownIn, db: DbSession) -> dict[str, bool]:
+    """客户端弹出引导窗那刻调 → 记一条今日 engagement(shown),今天**这个 App** 不再自动弹。
+
+    频控主判据(管跨重装):弹出即占用今天这个 App 的"一次"。用户领/拒/无视都算用掉。
+    后续点领取/拒绝再由 step/dismiss 把 type 升级。按 (device, package, 日) 记。
+    """
+    coupon_repo.mark_engagement(
+        db, payload.device_id, payload.package, payload.user_id, "shown"
+    )
+    return {"ok": True}
+
+
 @router.post("/prompt/dismiss", summary="用户拒绝/关闭领券引导窗(记今日已 engage)")
 def coupon_prompt_dismiss(payload: CouponPromptDismissIn, db: DbSession) -> dict[str, bool]:
-    """客户端点关闭引导窗时调用 → 记一条今日 engagement(dismissed),今天不再弹。
+    """客户端点关闭引导窗时调用 → 记一条今日 engagement(dismissed),今天**这个 App** 不再弹。
 
     server 在透传链路里看不到"用户拒绝"(拒绝不发起领券),故必须客户端通知。
-    MVP 不鉴权,按 device_id 记。
+    频控按 (device, package, 日),各 App 独立。MVP 不鉴权,按 device_id 记。
     """
-    coupon_repo.mark_engagement(db, payload.device_id, payload.user_id, "dismissed")
+    coupon_repo.mark_engagement(
+        db, payload.device_id, payload.package, payload.user_id, "dismissed"
+    )
     return {"ok": True}
 
 
@@ -205,12 +224,12 @@ def coupon_prompt_dismiss(payload: CouponPromptDismissIn, db: DbSession) -> dict
     summary="切到外卖 App 时是否还应弹领券引导窗",
 )
 def coupon_prompt_should_show(
-    device_id: str, db: DbSession
+    device_id: str, db: DbSession, package: str = ""
 ) -> CouponPromptShouldShowOut:
-    """今天这台设备已 engage(领或拒)过 → should_show=false。客户端据此决定弹不弹
+    """今天这台设备**这个 App** 已 engage(弹/领/拒)过 → should_show=false。各 App 独立:
-    (纯后台判据,客户端不再做前台 SP 缓存判断)。"""
+    美团弹过不压淘宝/京东。客户端切到目标 App 时带 package 查(老客户端不带 → "" 全局态)。"""
     return CouponPromptShouldShowOut(
-        should_show=not coupon_repo.has_engaged_today(db, device_id)
+        should_show=not coupon_repo.has_engaged_today(db, device_id, package)
     )
 
 
@@ -236,3 +255,16 @@ def coupon_completed_today(
     return CouponCompletedTodayOut(
         completed=coupon_repo.has_completed_today(db, device_id)
     )
+
+
+@router.post(
+    "/completed-today/reset",
+    summary="重置今日已完成(开发设置全重置用,恢复首页「去领取」卡可点)",
+)
+def coupon_completed_today_reset(
+    payload: CouponPromptDismissIn, db: DbSession
+) -> dict[str, bool]:
+    """删这台设备今天的 completion → has_completed_today 变 false,首页「去领取」卡恢复可点。
+    与 /prompt/reset 配套:开发设置「重置今日领券弹窗状态」一键把今日状态全清。MVP 不鉴权。"""
+    coupon_repo.reset_today_completion(db, payload.device_id)
+    return {"ok": True}

app/api/v1/feedback.py

@@ -1,7 +1,7 @@
 """帮助与反馈 endpoint。
 
 路由前缀 `/api/v1/feedback`,需 Bearer 鉴权(反馈绑到登录用户,便于回访)。
-  POST /  提交反馈(multipart:content / contact 必填,images 可选 ≤4 张)
+  POST /  提交反馈(multipart:content 必填;contact 可选(原型改版后客户端已不再采集);images 可选 ≤6 张)
 
 截图复用 [app.core.media] 落盘到 /media/feedback/。
 """
@@ -20,7 +20,7 @@ logger = logging.getLogger("shagua.feedback")
 
 router = APIRouter(prefix="/api/v1/feedback", tags=["feedback"])
 
-_MAX_IMAGES = 4
+_MAX_IMAGES = 6
 _CONTENT_MAX = 2000
 _CONTACT_MAX = 128
 
@@ -30,7 +30,8 @@ async def submit_feedback(
     user: CurrentUser,
     db: DbSession,
     content: str = Form(...),
-    contact: str = Form(...),
+    # 原型改版后客户端不再采集联系方式;保留字段以兼容旧端 + 后续可能复用,默认空串。
+    contact: str = Form(default=""),
     images: list[UploadFile] = File(default=[]),
 ) -> FeedbackOut:
     content = content.strip()
@@ -39,8 +40,6 @@ async def submit_feedback(
         raise HTTPException(status_code=400, detail="反馈内容不能为空")
     if len(content) > _CONTENT_MAX:
         raise HTTPException(status_code=400, detail="反馈内容过长")
-    if not contact:
-        raise HTTPException(status_code=400, detail="联系方式不能为空")
     if len(contact) > _CONTACT_MAX:
         raise HTTPException(status_code=400, detail="联系方式过长")
 

app/core/config.py

@@ -9,11 +9,16 @@ from functools import lru_cache
 from pathlib import Path
 from typing import Literal
 
-from pydantic import Field
+from pydantic import Field, model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 _PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
 
+# 生产环境 JWT secret 的最小可接受长度(字节)。HS256 推荐高熵随机串;<16 视为弱密钥。
+_MIN_PROD_SECRET_LEN = 16
+# 已知的占位默认值(代码里写死的 default),prod 下绝不能沿用。
+_INSECURE_SECRET_DEFAULTS = frozenset({"change-me", "change-me-admin", ""})
+
 
 class Settings(BaseSettings):
     model_config = SettingsConfigDict(
@@ -198,6 +203,29 @@ class Settings(BaseSettings):
     def is_prod(self) -> bool:
         return self.APP_ENV == "prod"
 
+    @model_validator(mode="after")
+    def _enforce_prod_secrets(self) -> "Settings":
+        """prod 下强校验 JWT secret,弱/默认/空即启动报错(fail-fast,挡住 token 被伪造)。
+
+        只校验两个签发凭证:App 用户的 JWT_SECRET_KEY、后台的 ADMIN_JWT_SECRET——它们沿用默认值
+        时任何人都能伪造 access/admin token → 账号与后台失陷。INTERNAL_API_SECRET 默认空 = 内部端点
+        关闭(返 503),是安全的默认态,故不在此强制。dev 不触发,便于本地直接起。
+        """
+        if not self.is_prod:
+            return self
+        weak: list[str] = []
+        for name in ("JWT_SECRET_KEY", "ADMIN_JWT_SECRET"):
+            value = getattr(self, name)
+            if value in _INSECURE_SECRET_DEFAULTS or len(value) < _MIN_PROD_SECRET_LEN:
+                weak.append(name)
+        if weak:
+            raise ValueError(
+                f"APP_ENV=prod 但检测到弱/默认密钥: {', '.join(weak)} —— 必须改成 "
+                f"≥{_MIN_PROD_SECRET_LEN} 位高熵随机串(否则 JWT 可被伪造 → 用户/后台账号失陷)。"
+                f"生成示例: python -c \"import secrets; print(secrets.token_urlsafe(48))\""
+            )
+        return self
+
 
 @lru_cache(maxsize=1)
 def get_settings() -> Settings:

app/core/rewards.py

@@ -143,6 +143,13 @@ AD_LT_FACTOR_TABLE: tuple[tuple[float, int, int | None], ...] = (
     (1.0, 11, None),
 )
 
+# 客户端可影响的 eCPM 可信上限(分/千次展示):信息流广告一期由客户端上报 eCPM,伪造天价 eCPM
+# 可铸出天量金币(见 calculate_ad_reward_coin)。真实 eCPM 一般 <¥100 CPM(=10000 分),档位表顶档
+# 为 >¥400(=40000 分);取 ¥500 CPM=50000 分,留足真实头部余量又封死伪造值。钳在唯一计算口
+# calculate_ad_reward_coin,故 feed 与 reward_video(回退客户端上报 eCPM 时)一并护住;阈值设在所有
+# 真实值之上,不会少发正规奖励。
+AD_ECPM_MAX_FEN: int = 50_000
+
 
 def parse_ecpm_fen(ecpm: str | int | float | None) -> float:
     """解析 eCPM 原始值(穿山甲 getEcpm 原值,单位=分/千次展示)。非法/缺失→0。"""
@@ -172,24 +179,28 @@ def ad_ecpm_factor(ecpm_yuan: float) -> float:
     return 0.1
 
 
-def ad_lt_factor(today_count_after_this: int) -> float:
+def ad_lt_factor(count_after_this: int) -> float:
-    """LT 因子。today_count_after_this 是当天累计第 N 条/份广告奖励。"""
+    """LT 因子。count_after_this 是该账号累计第 N 条/份看视频奖励(不按天重置)。"""
-    count = max(1, today_count_after_this)
+    count = max(1, count_after_this)
     for factor, lo, hi in AD_LT_FACTOR_TABLE:
         if count >= lo and (hi is None or count <= hi):
             return factor
     return 1.0
 
 
-def calculate_ad_reward_coin(ecpm: str | int | float | None, today_count_after_this: int) -> int:
+def calculate_ad_reward_coin(ecpm: str | int | float | None, count_after_this: int) -> int:
     """按金币数值体系计算单份广告奖励金币。
 
     eCPM 是穿山甲 getEcpm 原值,单位【分/千次展示】;先 ÷100 转成元(因子判档 + 收益换算都用元)。
     单次收益(元)= eCPM元 ÷ 1000(每千次→单次) × 因子1(eCPM 元档) × 因子2(LT);
-    再按 1 元=10000 金币取整。
+    再按 1 元=10000 金币取整。count_after_this 为账号累计第 N 次看视频(LT 因子用,不按天重置)。
+
+    eCPM 在此先钳到 AD_ECPM_MAX_FEN(¥500 CPM):信息流广告一期 eCPM 由客户端上报,伪造天价值
+    会铸天量金币;钳在这唯一入口,feed 与 reward_video 回退客户端 eCPM 的路径都护住,且阈值高于
+    所有真实值,不影响正规发奖。
     """
-    ecpm_yuan = parse_ecpm_yuan(ecpm)
+    ecpm_yuan = min(parse_ecpm_yuan(ecpm), AD_ECPM_MAX_FEN / 100.0)
-    yuan = (ecpm_yuan / 1000.0) * ad_ecpm_factor(ecpm_yuan) * ad_lt_factor(today_count_after_this)
+    yuan = (ecpm_yuan / 1000.0) * ad_ecpm_factor(ecpm_yuan) * ad_lt_factor(count_after_this)
     return max(0, round(yuan * COIN_PER_YUAN))
 
 

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。

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

app/models/comparison.py

@@ -94,7 +94,7 @@ class ComparisonRecord(Base):
     # ===== 明细(JSON,越详细越好)=====
     # 下单菜品 [{name, qty, specs?}]
     items: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
-    # 逐平台对比 [{platform_id, platform_name, package, price, is_source, rank, coupon_saved, coupon_name}](price/coupon_saved 单位:元,原样存;coupon_name=优惠来源名)
+    # 逐平台对比 [{platform_id, platform_name, package, price, is_source, rank, coupon_saved, coupon_name, applied_coupons}](price/coupon_saved 单位:元,原样存;coupon_name=优惠来源名;applied_coupons=[{name,amount}] 多券明细)
     comparison_results: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
     # 目标平台未找到、跳过的菜名
     skipped_dish_names: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)

app/models/coupon_state.py

@@ -137,25 +137,35 @@ class CouponDailyCompletion(Base):
 
 
 class CouponPromptEngagement(Base):
-    """按 (device, 自然日) 记"今天是否对领券引导窗表达过意向"——弹窗频控源。"""
+    """按 (device, **App**, 自然日) 记"今天这个 App 是否对领券引导窗表达过意向"——弹窗频控源。
+
+    2026-06-14:频控维度从 (device, 日) 改为 (device, package, 日)。需求是美团/淘宝/京东
+    各自独立——在美团弹过/领过,不影响淘宝、京东今天仍各弹一次。原来缺 package → 任一 App
+    弹过就把整台设备当天标记 engage,其余 App 被压住不弹(bug)。
+    """
 
     __tablename__ = "coupon_prompt_engagement"
     __table_args__ = (
-        # 一台设备一天一条:今天 engage 过(领或拒)就不再弹。
+        # 一台设备、一个 App、一天一条:今天**这个 App** engage 过(领或拒)才不再弹该 App。
         UniqueConstraint(
-            "device_id", "engage_date",
+            "device_id", "package", "engage_date",
-            name="uq_coupon_engage_device_date",
+            name="uq_coupon_engage_device_pkg_date",
         ),
     )
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
 
     device_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    # 触发弹窗的目标 App 包名(com.sankuai.meituan / com.taobao.taobao / com.jingdong.app.mall)。
+    # 频控维度,各 App 独立。旧行(改造前)无此值 → 迁移用占位 "" 填,不影响新逻辑判断。
+    package: Mapped[str] = mapped_column(
+        String(64), nullable=False, server_default=""
+    )
     user_id: Mapped[int | None] = mapped_column(Integer, index=True, nullable=True)
     # Asia/Shanghai 自然日。
     engage_date: Mapped[date] = mapped_column(Date, nullable=False)
-    # claim_started(点了一键领取)/ dismissed(点了拒绝/关闭)。仅记录区分,
+    # claim_started(点了一键领取)/ dismissed(点了拒绝/关闭)/ shown(自动弹出即记)。
-    # 判断只看"今天有没有这条",type 不影响弹不弹。
+    # 仅记录区分,判断只看"今天这个 App 有没有这条",type 不影响弹不弹。
     engage_type: Mapped[str] = mapped_column(String(16), nullable=False)
 
     created_at: Mapped[datetime] = mapped_column(

app/models/feedback.py

@@ -1,7 +1,8 @@
 """用户反馈表(帮助与反馈)。
 
-每条 = 用户一次提交。content 必填,contact 必填(微信/QQ/手机,便于回访),images 为可选的
+每条 = 用户一次提交。content 必填;contact 原为必填(微信/QQ/手机),原型改版后客户端不再采集,
-截图 URL 列表(/media/feedback/...,JSON 存)。status: new(待处理)/ handled(已处理)。
+新数据存空串(列保持 NOT NULL,免迁移;历史数据仍有值);images 为可选的截图 URL 列表
+(/media/feedback/...,JSON 存)。status: new(待处理)/ handled(已处理)。
 """
 from __future__ import annotations
 

app/models/store_mapping.py

@@ -0,0 +1,120 @@
+"""平台店铺表(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
+    # 淘宝 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 每次分享重新加密、非稳定主键(调研文档 §八), 故单列存"可复跳的一次性票据",
+    # 不进 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}>"
+        )

app/repositories/ad_feed_reward.py

@@ -16,6 +16,10 @@ from app.repositories import wallet as crud_wallet
 
 
 FEED_REWARD_UNIT_SECONDS = 10
+# 单个 feed 事件的时长上限(秒):一期 duration_seconds 由客户端上报,伪造超长时长会刷份数
+# (每 10 秒 1 份)。真实单条信息流视频远小于此;取 120s=12 份封顶,挡刷量、不影响正规单。
+# 与 rewards.AD_ECPM_MAX_FEN(eCPM 钳顶)合起来,把单事件可铸金币锁进有限区间。
+FEED_MAX_DURATION_SECONDS = 120
 
 
 def _find_by_event(db: Session, client_event_id: str) -> AdFeedRewardRecord | None:
@@ -38,15 +42,14 @@ def _granted_today(db: Session, user_id: int, reward_date: str) -> int:
     ).scalar_one()
 
 
-def _unit_reward_total(db: Session, user_id: int, ecpm: str, unit_count: int, today: str) -> int:
+def _unit_reward_total(db: Session, user_id: int, ecpm: str, unit_count: int) -> int:
-    """按每个 10 秒单位逐份计算奖励,LT 使用当天累计奖励份序号。"""
+    """按每个 10 秒单位逐份计算奖励,LT 使用**账号累计**奖励份序号(不按天重置)。"""
     if unit_count <= 0:
         return 0
     existing_units = db.execute(
         select(func.coalesce(func.sum(AdFeedRewardRecord.unit_count), 0))
         .where(
             AdFeedRewardRecord.user_id == user_id,
-            AdFeedRewardRecord.reward_date == today,
             AdFeedRewardRecord.status == "granted",
         )
     ).scalar_one()
@@ -67,13 +70,19 @@ def grant_feed_reward(
     adn: str | None = None,
     slot_id: str | None = None,
 ) -> AdFeedRewardRecord:
-    """完成一条信息流广告后结算奖励。client_event_id 幂等,同号重试不重复发。"""
+    """完成一条信息流广告后结算奖励。client_event_id 幂等,同号重试不重复发。
+
+    一期 eCPM/时长均由客户端上报,故服务端两道硬闸防刷:时长钳到 FEED_MAX_DURATION_SECONDS
+    限单事件份数,eCPM 在 rewards.calculate_ad_reward_coin 内钳到 AD_ECPM_MAX_FEN 限单份金额;
+    叠加每日 get_ad_daily_limit 条数上限,把单用户日产出锁进有限区间。
+    """
     existing = _find_by_event(db, client_event_id)
     if existing is not None:
         return existing
 
     today = cn_today().isoformat()
-    safe_duration = max(0, min(duration_seconds, 24 * 60 * 60))
+    # 客户端上报时长先钳到 FEED_MAX_DURATION_SECONDS,防伪造超长时长刷份数(见常量注释)。
+    safe_duration = max(0, min(duration_seconds, FEED_MAX_DURATION_SECONDS))
     unit_count = safe_duration // FEED_REWARD_UNIT_SECONDS
 
     if _granted_today(db, user_id, today) >= rewards.get_ad_daily_limit(db):
@@ -92,7 +101,7 @@ def grant_feed_reward(
         )
         return _commit_record(db, rec, client_event_id)
 
-    coin = _unit_reward_total(db, user_id, ecpm, unit_count, today)
+    coin = _unit_reward_total(db, user_id, ecpm, unit_count)
     if coin > 0:
         crud_wallet.grant_coins(
             db, user_id, coin,

app/repositories/ad_reward.py

@@ -54,6 +54,20 @@ def _granted_today(db: Session, user_id: int, reward_date: str) -> int:
     ).scalar_one()
 
 
+def _granted_cumulative(db: Session, user_id: int) -> int:
+    """账号累计已发奖的看视频次数(不按天重置)——LT 因子(因子2)用它定"第 N 次广告"。
+    与 _granted_today 区别仅在去掉 reward_date 过滤;每日次数上限/冷却仍按当日统计。"""
+    return db.execute(
+        select(func.count())
+        .select_from(AdRewardRecord)
+        .where(
+            AdRewardRecord.user_id == user_id,
+            AdRewardRecord.status == "granted",
+            AdRewardRecord.reward_scene == "reward_video",
+        )
+    ).scalar_one()
+
+
 def grant_ad_reward(
     db: Session,
     user_id: int,
@@ -104,7 +118,9 @@ def grant_ad_reward(
         )
         return _commit_record(db, rec, trans_id)
 
-    coin = rewards.calculate_ad_reward_coin(ecpm_raw, _granted_today(db, user_id, today) + 1)
+    # LT 因子(因子2)按"账号累计第 N 次看广告"递减(2.0→1.0),不再按天重置;
+    # 每日次数上限/冷却仍按当日统计(over_count 用 _granted_today)。
+    coin = rewards.calculate_ad_reward_coin(ecpm_raw, _granted_cumulative(db, user_id) + 1)
 
     # 发金币 + 记一笔,同事务
     crud_wallet.grant_coins(

app/repositories/coupon_state.py

@@ -31,11 +31,13 @@ def today_cn() -> date:
 
 # ===== 弹窗频控(coupon_prompt_engagement)=====
 
-def has_engaged_today(db: Session, device_id: str) -> bool:
+def has_engaged_today(db: Session, device_id: str, package: str) -> bool:
-    """这台设备今天是否已对领券引导窗表达过意向(领或拒)。有 = 不再弹。"""
+    """这台设备今天**这个 App** 是否已对领券引导窗表达过意向(领/拒/弹出)。有 = 该 App 不再弹。
+    频控按 (device, package, 日):美团弹过不影响淘宝/京东今天各自仍弹一次。"""
     row = db.execute(
         select(CouponPromptEngagement.id).where(
             CouponPromptEngagement.device_id == device_id,
+            CouponPromptEngagement.package == package,
             CouponPromptEngagement.engage_date == today_cn(),
         )
     ).first()
@@ -43,13 +45,18 @@ def has_engaged_today(db: Session, device_id: str) -> bool:
 
 
 def mark_engagement(
-    db: Session, device_id: str, user_id: int | None, engage_type: str
+    db: Session, device_id: str, package: str, user_id: int | None, engage_type: str
 ) -> None:
-    """记今日意向(claim_started / dismissed)。(device, 今天) 唯一,幂等 upsert。"""
+    """记今日意向(shown / claim_started / dismissed)。(device, package, 今天) 唯一,幂等 upsert。
+
+    engage_type 升级口径(同一 (device,package,日) 多次调,只覆盖 type,不新增行):
+    shown(自动弹出)→ claim_started(点一键领取)/ dismissed(点关闭)。判断只看"有没有这条"。
+    """
     today = today_cn()
     row = db.execute(
         select(CouponPromptEngagement).where(
             CouponPromptEngagement.device_id == device_id,
+            CouponPromptEngagement.package == package,
             CouponPromptEngagement.engage_date == today,
         )
     ).scalar_one_or_none()
@@ -59,19 +66,20 @@ def mark_engagement(
             row.user_id = user_id
     else:
         db.add(CouponPromptEngagement(
-            device_id=device_id, user_id=user_id,
+            device_id=device_id, package=package, user_id=user_id,
             engage_date=today, engage_type=engage_type,
         ))
     try:
         db.commit()
     except IntegrityError:
-        # 并发下另一请求刚插了同 (device, 日) → 唯一约束撞,回滚忽略(本就幂等)。
+        # 并发下另一请求刚插了同 (device, package, 日) → 唯一约束撞,回滚忽略(本就幂等)。
         db.rollback()
 
 
 def reset_today_engagement(db: Session, device_id: str) -> int:
-    """删这台设备今天的 engagement(开发设置「重置今日领券弹窗状态」调,测频控用)。
+    """删这台设备今天**所有 App** 的 engagement(开发设置「重置今日领券弹窗状态」调,测频控用)。
-    删后 has_engaged_today → false,今天又能弹。返回删除行数。"""
+    删后各 App has_engaged_today → false,今天又都能弹。返回删除行数。
+    (不按 package 过滤:重置是"把今天清干净从头测",清全部 App 最符合预期。)"""
     result = db.execute(
         delete(CouponPromptEngagement).where(
             CouponPromptEngagement.device_id == device_id,
@@ -123,6 +131,19 @@ def mark_completed_today(
         db.rollback()
 
 
+def reset_today_completion(db: Session, device_id: str) -> int:
+    """删这台设备今天的"已完成"记录(开发设置「重置今日领券弹窗状态」全重置时调)。
+    删后 has_completed_today → false,首页「去领取」卡恢复可点。返回删除行数。"""
+    result = db.execute(
+        delete(CouponDailyCompletion).where(
+            CouponDailyCompletion.device_id == device_id,
+            CouponDailyCompletion.complete_date == today_cn(),
+        )
+    )
+    db.commit()
+    return result.rowcount or 0
+
+
 # ===== 领券记录(coupon_claim_record)=====
 
 def record_claims(

app/repositories/ops_marquee.py

@@ -6,9 +6,16 @@ feed 取最近的成功且省到钱的比价记录(脱敏用户名);不足 N 条
 种子是「生成规则」:用户名可空(空→随机合成脱敏名,避开同屏撞名)、金额是区间(每次随机取值)、
 feed 公平随机抽取(不看 sort_order),所有启用种子都有机会露出。
 
-脱敏名风格:手机尾号(138****5678)+ 中文昵称(省钱**、小张**、吃货**达人)混合,看起来像真实
+脱敏名规则(全 feed 统一,按字符算、中英文皆适用):
-异质用户群。真实条按 user_id 确定性合成(同用户恒定);种子留空 / 旧「用户****xxx」模板名一律随机
+  - 有昵称 → 昵称脱敏:≥3 字「首+隐藏字数个星+末」(省钱小能手→省***手 / SaveKing→S******g),
-合成(自愈历史种子,无需迁移)。
+    2 字「首+星」(阿强→阿*),1 字「用户+该字」(喵→用户喵);
+  - 没昵称 → 合成一个假名(见下)再脱敏,少数露「用户+id 后 3 位」(用户********618)。
+真实条:设过昵称的按真实昵称脱敏;当前多数用户没设昵称,直接展示「用户****」会清一色,故给它们合成
+假名——后期真实昵称多了,真实条会直接用真实昵称,合成占比自然下降。种子 / 兜底同样合成、避开同屏撞名;
+种子留空 / 旧「用户****xxx」模板名一律重新合成(自愈历史种子,无需迁移)。
+假名合成:纯本地语料**组合生成**(中文姓池×名字池 拼真名 / 中文网络昵称 / 英文昵称 / 英文名带数字尾),
+组合空间上万、中文为主英文为辅,脱敏后像真实异质用户群。**真实条按 `user_id` 播种确定性合成**(同一用户
+每次展示恒定、不同用户各异),刷新/翻页不变脸;种子 / 兜底用运行时随机源出多样填充。不依赖外部库。
 """
 from __future__ import annotations
 
@@ -21,6 +28,7 @@ from sqlalchemy.orm import Session
 from app.core.rewards import CN_TZ
 from app.models.comparison import ComparisonRecord
 from app.models.ops_marquee_seed import OpsMarqueeSeed
+from app.models.user import User
 
 # feed 运行时随机源(每次请求结果不同 = 轮播想要的「鲜活感」)
 _rng = random.Random()
@@ -36,59 +44,111 @@ _REAL_ROWS_TTL_SECONDS = 30
 _REAL_ROWS_FETCH_CAP = 600  # 一次多取些,够 limit≤30 去重后取数;命中缓存后复用
 _real_rows_cache: dict = {"at": None, "rows": None}
 
-# ===== 脱敏名生成:手机尾号 + 中文昵称 混合,看起来像真实异质用户群(替代旧的「用户********xxx」统一模板)=====
+# ===== 用户标识脱敏 + 没真实昵称时的假名合成 =====
-_PHONE_2ND = "3456789"  # 手机号第 2 位(13x~19x)
+# 脱敏规则(中英文皆适用,按字符算):有昵称→≥3 字「首+隐藏字数个星+末」(省钱小能手→省***手 /
-_SURNAMES = "王李张刘陈杨黄赵周吴徐孙马朱胡郭何高林罗郑梁谢宋唐许韩冯邓曹彭曾肖田董袁潘蒋蔡余杜叶程苏魏吕丁任沈姚卢"
+# SaveKing→S******g)、2 字「首+星」(阿强→阿*)、1 字「用户+该字」(喵→用户喵);没昵称→用户+8星+id 后 3 位。
-_NICK_WORDS = [
+# 没真实昵称时(当前多数用户 / 种子 / 兜底)合成假名:本地姓/名字池组合出真名 + 网络昵称语料(中/英)
-    "省钱", "吃货", "羊毛", "薅羊毛", "爱比价", "精打细算", "持家", "干饭",
+# 混播再走同一套脱敏,像真实异质用户群。后期真实昵称多了,真实条直接用真实昵称,这里占比自然下降。
-    "美食控", "外卖", "剁手", "捡漏", "搞钱", "打工人", "摸鱼", "养生",
+_ANON_STARS = "*" * 8  # 无昵称匿名串固定 8 星(用户********xxx)
-    "佛系", "躺平", "暴富", "锦鲤", "夜宵", "奶茶", "热爱生活", "会过日子",
+
+# 中文真名组合池:姓(百家姓常见 ~100)× 名字单字(~120)拼「姓 + 1~2 字」,组合空间上万。
+# 脱敏后只露「姓*」或「姓*末」,故无需真实姓名库——组合够多即看着各异(替代原 Faker zh_CN)。
+_SURNAME_ZH = list(
+    "王李张刘陈杨赵黄周吴徐孙胡朱高林何郭马罗梁宋郑谢韩唐冯于董萧程曹袁邓许傅沈曾彭吕苏卢蒋蔡贾丁"
+    "魏薛叶阎余潘杜戴夏钟汪田任姜范方石姚谭廖邹熊金陆郝孔白崔康毛邱秦江史顾侯邵孟龙万段钱汤尹黎"
+    "易常武乔贺赖龚文施洪丰房邢"
+)
+_GIVEN_ZH = list(
+    "伟芳娜秀英敏静丽强磊军洋勇艳杰娟涛明超霞平刚桂兰华健世广义兴良海山仁波宁贵福生龙元全国胜学祥"
+    "才发新利清飞彬富顺信子昌成康星光天达安岩中茂进有坚和彪博诚先敬震振壮会思群豪心邦承乐绍功松善"
+    "厚庆民友裕河哲亮政谦亨奇之轮翰朗伯宏言若鸣朋斌栋维启克伦翔旭鹏泽晨辰士以建家致树炎德行时泰盛雄"
+)
+# 英文名池:与英文昵称互补,偶尔带数字尾(jay87)更像注册用户名(替代原 Faker en_US)。
+_FIRST_EN = [
+    "Alex", "Chris", "Sam", "Jamie", "Taylor", "Jordan", "Casey", "Morgan", "Riley", "Jessie",
+    "Robin", "Drew", "Lee", "Max", "Kim", "Ray", "Dana", "Pat", "Terry", "Jay",
+    "Kelly", "Sky", "Ash", "Nico", "Remy", "Quinn", "Reese", "Sage", "Toni", "Val",
+]
+
+# 网络昵称语料:补一份「省钱小能手 / SaveKing」式网络昵称更像真实 app 用户。想更丰富往里加即可。
+_NICK_ZH = [
+    "省钱小能手", "薅羊毛专业户", "吃货一枚", "干饭人", "奶茶续命", "精打细算过日子",
+    "会过日子的人", "捡漏王", "外卖救星", "养生少年", "持家有道", "比价狂魔",
+    "月光族逆袭", "摸鱼达人", "暴富锦鲤", "美食猎人", "折扣猎手", "隐形贫困人口",
+    "打工不打烊", "佛系青年", "元气满满", "一只小馋猫", "爱吃的小朋友", "省钱才是硬道理",
+]
+_NICK_EN = [
+    "SaveKing", "DealHunter", "FoodieLife", "BudgetBoss", "CouponQueen", "ThriftyOne",
+    "SnackLover", "BargainHero", "PennyWise", "SmartShopper", "DiscountNinja", "HungryPanda",
+    "MoneyMaster", "CheapEats", "SaverPro", "FrugalFox", "NoodleKing", "BubbleTeaFan",
+    "GoldenSaver", "ValueHunter", "SnackAttack", "LazyFoodie",
 ]
-_NICK_PREFIX = ["小", "大", "老", "阿", "一只", "爱吃的", "快乐的"]
-_NICK_SUFFIX = ["达人", "小能手", "侠", "党", "一族", "星人", "鸭", "酱"]
-_LETTER_U = "ABCDEFGHJKLMNPQRSTUVWXYZ"
-_LETTER_L = "abcdefghijkmnpqrstuvwxyz"
 
 
-def _phone_name(rng: random.Random) -> str:
+def _mask_nickname(nick: str) -> str:
-    """手机尾号式:1[3-9]X****XXXX(合成、非真实号)。"""
+    """昵称脱敏(按字符,中英文皆可):≥3 字→首+(隐藏字数个)星+末(省钱小能手→省***手 / SaveKing→S******g);
-    head = "1" + rng.choice(_PHONE_2ND) + str(rng.randint(0, 9))
+    2 字→首+星(阿强→阿*);1 字→「用户」+该字(喵→用户喵)。调用方需保证 nick 已 strip 且非空。"""
-    return f"{head}****{rng.randint(0, 9999):04d}"
+    n = len(nick)
+    if n >= 3:
+        return nick[0] + "*" * (n - 2) + nick[-1]
+    if n == 2:
+        return nick[0] + "*"
+    return "用户" + nick
 
 
-def _nickname(rng: random.Random) -> str:
+def _anon_by_id(user_id: int) -> str:
-    """昵称打码式,中文为主(约 11/12),掺少量中英混 / 纯字母。
+    """没昵称匿名:固定「用户」+ 8 星 + id 后 3 位(不足 3 位前补 0):用户********618。"""
-    ⚠️ 脱敏星号一律夹在中间(xx**yy),星号前后都要有可见字——不出现结尾打码(省钱**)的形态。
+    return "用户" + _ANON_STARS + f"{user_id % 1000:03d}"
-    """
-    p = rng.randint(0, 11)
-    if p in (0, 1, 2, 3, 4):
-        return rng.choice(_NICK_WORDS) + "**" + rng.choice(_NICK_SUFFIX)  # 省钱**党 / 吃货**达人
-    if p in (5, 6):
-        return rng.choice(_SURNAMES) + "**" + rng.choice(_NICK_SUFFIX)  # 张**侠
-    if p in (7, 8):
-        return rng.choice(_NICK_PREFIX) + rng.choice(_SURNAMES) + "**" + rng.choice(_NICK_SUFFIX)  # 小张**达人
-    if p in (9, 10):
-        return rng.choice(_NICK_WORDS) + "**" + rng.choice(_LETTER_U)  # 中英混 省钱**A(字母移到星号后)
-    return rng.choice(_LETTER_U) + "**" + rng.choice(_LETTER_L)  # 纯字母(约 1/12) A**b
 
 
-def _realistic_name(rng: random.Random) -> str:
+def _synth_full_name(rng: random.Random) -> str:
-    """~45% 手机尾号 + ~55% 中文昵称,混合出真实异质感。"""
+    """合成一个完整(未脱敏)假名:中文真名(姓+1~2字) / 中文网络昵称 / 英文昵称 / 英文名(可带数字尾) 混播,
-    return _phone_name(rng) if rng.random() < 0.45 else _nickname(rng)
+    中文为主(~70%)英文为辅(~30%)。**全程用传入 rng** → 同种子复现、不同种子各异、组合空间上万。"""
+    r = rng.random()
+    if r < 0.35:
+        # 中文真名:姓 + 1~2 个名字字(60% 概率两字),脱敏后露「姓*」或「姓*末」
+        given = rng.choice(_GIVEN_ZH) + (rng.choice(_GIVEN_ZH) if rng.random() < 0.6 else "")
+        return rng.choice(_SURNAME_ZH) + given   # 王伟 / 李秀兰
+    if r < 0.70:
+        return rng.choice(_NICK_ZH)              # 中文网络昵称:省钱小能手
+    if r < 0.85:
+        return rng.choice(_NICK_EN)              # 英文昵称:SaveKing
+    base = rng.choice(_FIRST_EN)                 # 英文名,半数带数字尾:Jay87 / Alex
+    return base + (str(rng.randint(1, 99)) if rng.random() < 0.5 else "")
 
 
-def _mask_user(user_id: int) -> str:
+def _synth_masked_name(rng: random.Random) -> str:
-    """真实用户脱敏:按 user_id 确定性合成一个名(同用户恒定、不暴露真实信息)。"""
+    """合成一条「已脱敏」假名(供没真实昵称的真实用户 / 种子 / 兜底用)。用传入 rng 决定一切随机。"""
-    return _realistic_name(random.Random(user_id))
+    full = _synth_full_name(rng).strip()
+    return _mask_nickname(full) if full else "用户" + _ANON_STARS + f"{rng.randint(0, 999):03d}"
+
+
+def _mask_real(nickname: str | None, user_id: int) -> str:
+    """真实用户脱敏:设过昵称→昵称脱敏(中英文皆可);没昵称→**按 user_id 播种确定性合成假名**再脱敏
+    (同一用户每次展示恒定、刷新不变脸,不同用户各异),避免无昵称用户清一色「用户****xxx」;
+    少数(~15%)露「用户+真实 id 后 3 位」。当前多数用户没昵称,后期真实昵称多了占比下降。"""
+    nick = (nickname or "").strip()
+    if nick:
+        return _mask_nickname(nick)
+    # 关键:按 user_id 播种本地 rng → 同一用户的合成名跨请求/刷新恒定(0.15 抽签也用它)。
+    seeded = random.Random(user_id)
+    return _anon_by_id(user_id) if seeded.random() < 0.15 else _synth_masked_name(seeded)
+
+
+def _synth_name(rng: random.Random) -> str:
+    """合成一条脱敏名(种子留空 / 旧模板名 / 兜底用):绝大多数=假名脱敏,少数=用户+随机 3 位(用户********618)。"""
+    if rng.random() < 0.15:
+        return "用户" + _ANON_STARS + f"{rng.randint(0, 999):03d}"
+    return _synth_masked_name(rng)
 
 
 def _unique_name(used: set[str]) -> str:
-    """随机合成一个不与 used 撞的脱敏名(种子留空 / 旧模板名用),防同屏撞名。"""
+    """随机合成一个不与 used 撞的脱敏名(种子留空 / 旧模板名 / 兜底用),防同屏撞名。"""
     for _ in range(60):
-        n = _realistic_name(_rng)
+        n = _synth_name(_rng)
         if n not in used:
             return n
-    return _realistic_name(_rng) + str(_rng.randint(0, 99))  # 兜底(几乎不会到)
+    return "用户" + _ANON_STARS + f"{_rng.randint(0, 999):03d}"  # 兜底:数字尾号天然好去重
 
 
 def _validate_amount(min_cents: int, max_cents: int) -> None:
@@ -116,8 +176,9 @@ _FALLBACK_MAX_CENTS = 3500
 
 
 # ===== 用户侧:轮播 feed(真实 + 种子混播) =====
-def _recent_real_rows(db: Session) -> list[tuple[int, int]]:
+def _recent_real_rows(db: Session) -> list[tuple[int, int, str | None]]:
-    """近期 success 且省到钱的 (user_id, saved_cents),按 created_at desc;带 ~30s 进程内缓存。
+    """近期 success 且省到钱的 (user_id, saved_cents, nickname),按 created_at desc;带 ~30s 进程内缓存。
+    join user 取真实昵称用于脱敏(无昵称则展示层走「用户+id 后 3 位」)。
     返回纯元组(脱离 session),可安全跨请求复用。极端并发下偶尔多查一次(无锁、幂等),纯门面无副作用。
     """
     now = datetime.now(CN_TZ)
@@ -125,7 +186,8 @@ def _recent_real_rows(db: Session) -> list[tuple[int, int]]:
     if cached is not None and at is not None and (now - at).total_seconds() < _REAL_ROWS_TTL_SECONDS:
         return cached
     rows = db.execute(
-        select(ComparisonRecord.user_id, ComparisonRecord.saved_amount_cents)
+        select(ComparisonRecord.user_id, ComparisonRecord.saved_amount_cents, User.nickname)
+        .join(User, User.id == ComparisonRecord.user_id)
         .where(
             ComparisonRecord.status == "success",
             ComparisonRecord.saved_amount_cents > 0,
@@ -134,7 +196,7 @@ def _recent_real_rows(db: Session) -> list[tuple[int, int]]:
         .order_by(ComparisonRecord.created_at.desc())
         .limit(_REAL_ROWS_FETCH_CAP)
     ).all()
-    out = [(int(uid), int(sc)) for uid, sc in rows]
+    out = [(int(uid), int(sc), nick) for uid, sc, nick in rows]
     _real_rows_cache["rows"], _real_rows_cache["at"] = out, now
     return out
 
@@ -155,12 +217,12 @@ def get_feed(db: Session, limit: int = 8) -> list[dict]:
     items: list[dict] = []
     used_names: set[str] = set()
     seen_users: set[int] = set()
-    for uid, sc in rows:
+    for uid, sc, nick in rows:
         if uid in seen_users:
             continue
         seen_users.add(uid)
-        name = _mask_user(uid)
+        name = _mask_real(nick, uid)
-        used_names.add(name)  # 真实名按 user_id 稳定;偶发撞名可接受
+        used_names.add(name)  # 真实名按昵称/id 稳定;偶发撞名可接受
         items.append({"masked_user": name, "saved_amount_cents": int(sc)})
         if len(items) >= limit:
             break

app/repositories/store_mapping.py

@@ -0,0 +1,192 @@
+"""平台店铺映射落库:一次比价(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
+
+
+# ============================================================
+# 缓存查询: 比价前按"源平台店名"反查已沉淀的各目标平台店铺 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]
+        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

app/repositories/wallet.py

@@ -78,9 +78,15 @@ class WithdrawNotReviewable(Exception):
     """提现单当前状态不可审核(非 reviewing,可能已被处理过)。"""
 
 
-def get_or_create_account(db: Session, user_id: int, *, commit: bool = True) -> CoinAccount:
+def get_or_create_account(
-    """取用户金币账户,不存在则建一个空账户。"""
+    db: Session, user_id: int, *, commit: bool = True, lock: bool = False
-    acc = db.get(CoinAccount, user_id)
+) -> CoinAccount:
+    """取用户金币账户,不存在则建一个空账户。
+
+    lock=True 时对已存在的账户行加 SELECT FOR UPDATE(读-算-写余额的调用方串行化,防并发
+    双写余额错位,如 admin set 模式连点);默认 False 不改 C 端发奖行为。SQLite 下为 no-op。
+    """
+    acc = db.get(CoinAccount, user_id, with_for_update=True) if lock else db.get(CoinAccount, user_id)
     if acc is None:
         acc = CoinAccount(
             user_id=user_id,

app/schemas/compare_record.py

@@ -24,6 +24,13 @@ class ComparisonItemIn(BaseModel):
     specs: list[str] | None = None
 
 
+class AppliedCouponIn(BaseModel):
+    """单笔已用优惠(来自 comparison_results[].applied_coupons)。amount 单位:元、正数。"""
+
+    name: str
+    amount: float
+
+
 class ComparisonResultIn(BaseModel):
     """逐平台对比项(来自 done.params.comparison_results)。price 单位:元。"""
 
@@ -40,6 +47,10 @@ class ComparisonResultIn(BaseModel):
     # 优惠**来源名**(展示用, best-effort): 美团"外卖大额神券"/京东"百亿补贴"/淘宝"平台红包"。
     # None=没抠到 → 前端走通用"红包"。同样必须显式声明否则上报边界被 pydantic 静默丢弃(pricebot#38 引入)。
     coupon_name: str | None = None
+    # 多券明细 [{name, amount}](全口径: 平台红包+商家券+满减+配送减免, amount 单位元正数)。
+    # 跟 coupon_saved 并存, 是更丰富的明细; 空=没抠到 → 前端回退单券路径。
+    # 必须显式声明: 落库走 model_dump(), pydantic 默认丢未知字段, 不声明这行会被悄悄吞掉。
+    applied_coupons: list[AppliedCouponIn] = Field(default_factory=list)
 
 
 class ComparisonRecordIn(BaseModel):

app/schemas/coupon_state.py

@@ -7,16 +7,30 @@ from pydantic import BaseModel
 class CouponPromptDismissIn(BaseModel):
     """客户端拒绝/关闭领券引导窗的通知体。
 
-    server 据此记一条今日 engagement(dismissed)→ 今天这台设备不再弹引导窗。
+    server 据此记一条今日 engagement(dismissed)→ 今天**这个 App** 不再弹引导窗。
+    频控按 (device, package, 日):各 App 独立。package 缺省 ""(老客户端兼容,退化为全局态)。
     MVP 不鉴权,按 device_id 判断;user_id 登录态带上就一并记(资产),可空。
     """
 
     device_id: str
+    package: str = ""
+    user_id: int | None = None
+
+
+class CouponPromptShownIn(BaseModel):
+    """客户端弹出领券引导窗即上报(记 shown)。
+
+    弹出那刻就记一条今日 engagement(shown)→ 今天**这个 App** 不再自动弹(频控主判据,
+    管跨重装;本地 SP 兜后台抖动)。后续用户点领取/拒绝再把 type 升级成 claim_started/dismissed。
+    """
+
+    device_id: str
+    package: str
     user_id: int | None = None
 
 
 class CouponPromptShouldShowOut(BaseModel):
-    """切到外卖 App 时是否还应弹领券引导窗。今天已 engage(领或拒)过 → false。"""
+    """切到外卖 App 时是否还应弹领券引导窗。今天**这个 App** 已 engage(弹/领/拒)过 → false。"""
 
     should_show: bool
 

app/schemas/store_mapping.py

@@ -0,0 +1,76 @@
+"""平台店铺映射内部上报的收发模型。
+
+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
+
+
+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

docs/api/README.md

@@ -83,26 +83,27 @@
 | A5 | `GET /admin/api/users/{user_id}` | admin | [详情](./admin-user-detail.md) |
 | A6 | `POST /admin/api/users/{user_id}/status` | operator | [详情](./admin-user-status.md) |
 | A7 | `POST /admin/api/users/{user_id}/coins` | finance | [详情](./admin-user-coins.md) |
-| A8 | `GET /admin/api/wallet/coin-transactions` | admin | [详情](./admin-wallet-coin-transactions.md) |
+| A8 | `POST /admin/api/users/{user_id}/cash` | finance | [详情](./admin-user-cash.md) |
-| A9 | `GET /admin/api/wallet/cash-transactions` | admin | [详情](./admin-wallet-cash-transactions.md) |
+| A9 | `GET /admin/api/wallet/coin-transactions` | admin | [详情](./admin-wallet-coin-transactions.md) |
-| A10 | `GET /admin/api/withdraws` | admin | [详情](./admin-withdraws-list.md) |
+| A10 | `GET /admin/api/wallet/cash-transactions` | admin | [详情](./admin-wallet-cash-transactions.md) |
-| A11 | `POST /admin/api/withdraws/reconcile` | finance | [详情](./admin-withdraw-reconcile.md) |
+| A11 | `GET /admin/api/withdraws` | admin | [详情](./admin-withdraws-list.md) |
-| A12 | `POST /admin/api/withdraws/{out_bill_no}/refresh` | finance | [详情](./admin-withdraw-refresh.md) |
+| A12 | `POST /admin/api/withdraws/reconcile` | finance | [详情](./admin-withdraw-reconcile.md) |
-| A13 | `GET /admin/api/feedbacks` | admin | [详情](./admin-feedbacks-list.md) |
+| A13 | `POST /admin/api/withdraws/{out_bill_no}/refresh` | finance | [详情](./admin-withdraw-refresh.md) |
-| A14 | `POST /admin/api/feedbacks/{feedback_id}/handle` | operator | [详情](./admin-feedback-handle.md) |
+| A14 | `GET /admin/api/feedbacks` | admin | [详情](./admin-feedbacks-list.md) |
-| A15 | `GET /admin/api/admins` | super_admin | [详情](./admin-admins-list.md) |
+| A15 | `POST /admin/api/feedbacks/{feedback_id}/handle` | operator | [详情](./admin-feedback-handle.md) |
-| A16 | `POST /admin/api/admins` | super_admin | [详情](./admin-admin-create.md) |
+| A16 | `GET /admin/api/admins` | super_admin | [详情](./admin-admins-list.md) |
-| A17 | `PATCH /admin/api/admins/{admin_id}` | super_admin | [详情](./admin-admin-update.md) |
+| A17 | `POST /admin/api/admins` | super_admin | [详情](./admin-admin-create.md) |
-| A18 | `GET /admin/api/audit-logs` | admin | [详情](./admin-audit-logs.md) |
+| A18 | `PATCH /admin/api/admins/{admin_id}` | super_admin | [详情](./admin-admin-update.md) |
-| A19 | `GET /admin/api/dashboard-display` | admin | [详情](./admin-dashboard-display.md) |
+| A19 | `GET /admin/api/audit-logs` | admin | [详情](./admin-audit-logs.md) |
-| A20 | `PATCH /admin/api/dashboard-display/{metric}` | operator | [详情](./admin-dashboard-display.md) |
+| A20 | `GET /admin/api/dashboard-display` | admin | [详情](./admin-dashboard-display.md) |
-| A21 | `GET /admin/api/marquee-seeds` | admin | [详情](./admin-marquee-seeds.md) |
+| A21 | `PATCH /admin/api/dashboard-display/{metric}` | operator | [详情](./admin-dashboard-display.md) |
-| A22 | `POST /admin/api/marquee-seeds` | operator | [详情](./admin-marquee-seeds.md) |
+| A22 | `GET /admin/api/marquee-seeds` | admin | [详情](./admin-marquee-seeds.md) |
-| A23 | `PATCH /admin/api/marquee-seeds/{seed_id}` | operator | [详情](./admin-marquee-seeds.md) |
+| A23 | `POST /admin/api/marquee-seeds` | operator | [详情](./admin-marquee-seeds.md) |
-| A24 | `DELETE /admin/api/marquee-seeds/{seed_id}` | operator | [详情](./admin-marquee-seeds.md) |
+| A24 | `PATCH /admin/api/marquee-seeds/{seed_id}` | operator | [详情](./admin-marquee-seeds.md) |
-| A25 | `POST /admin/api/marquee-seeds/bulk` | operator | [详情](./admin-marquee-seeds.md) |
+| A25 | `DELETE /admin/api/marquee-seeds/{seed_id}` | operator | [详情](./admin-marquee-seeds.md) |
-| A26 | `GET /admin/api/marquee-seeds/preview` | admin | [详情](./admin-marquee-seeds.md) |
+| A26 | `POST /admin/api/marquee-seeds/bulk` | operator | [详情](./admin-marquee-seeds.md) |
-| A27 | `GET /admin/api/ad-coin-audit` | admin | [详情](./admin-ad-coin-audit.md)(看广告金币公式复算对账,只读) |
+| A27 | `GET /admin/api/marquee-seeds/preview` | admin | [详情](./admin-marquee-seeds.md) |
+| A28 | `GET /admin/api/ad-coin-audit` | admin | [详情](./admin-ad-coin-audit.md)(看广告金币公式复算对账,只读) |
 | - | `GET /admin/api/health` | 无 | admin 健康检查(无单独文档) |
 
 > ⚠️ 美团三个接口当前**无鉴权**,且 `referral-link` 的 `sid` 允许客户端传值覆盖默认渠道——见各接口"备注"。

docs/api/admin-ad-coin-audit.md

@@ -25,16 +25,18 @@ eCPM元 = getEcpm分 ÷ 100
   | `date` | string | 今天 | 北京时间 `YYYY-MM-DD`,审计某天 |
   | `user_id` | int | 全部 | 只看某用户;不传=所有用户 |
   | `scene` | string | 两类 | `reward_video` / `feed`;不传=两类都返回 |
-  | `limit` | int(1~500) | 100 | 返回明细条数(按时间倒序截断;**份序号在截断前已按全天数据算好**,不影响复算正确性) |
+  | `limit` | int(1~500) | 100 | **展示**明细条数(按时间倒序截断;**份序号在截断前已按全天数据算好**,不影响复算正确性) |
+  | `only_mismatch` | bool | false | 只展示不一致(✗)行;统计数仍按全量,不受影响 |
 
 - 出参 `200`:`AdCoinAuditOut`
   | 字段 | 类型 | 说明 |
   |---|---|---|
   | `date` | string | 审计日期 |
   | `formula` | object | 当前公式参数快照(见下) |
-  | `total` | int | 返回明细条数 |
+  | `total` | int | 该筛选下复算**总条数**(全量,不受 `limit`/`only_mismatch` 影响) |
-  | `mismatch_count` | int | 其中 `matched=false` 的条数;**=0 说明全部按公式发放** |
+  | `mismatch_count` | int | **全量**不一致条数(截断前统计,可信);**=0 说明全部按公式发放** |
-  | `items` | `AdCoinAuditRow[]` | 明细(见下) |
+  | `truncated` | bool | 展示集是否被 `limit` 截断(true=还有未返回的明细,请缩小范围或调大 `limit`) |
+  | `items` | `AdCoinAuditRow[]` | 展示明细(见下);`only_mismatch=true` 时只含 ✗ 行 |
 
 ### AdCoinFormulaOut(`formula`)
 | 字段 | 类型 | 说明 |

docs/api/admin-feedbacks-list.md

@@ -1,4 +1,4 @@
-# GET /admin/api/feedbacks — 反馈工单列表(游标分页)
+# GET /admin/api/feedbacks — 反馈工单列表(offset 分页 + 筛选/排序)
 
 > 所属:Admin·反馈 组（前缀 `/admin/api/feedbacks`） | 鉴权:Bearer admin_token（角色:任意已登录管理员,无 `require_role`,仅 `get_current_admin`） | [← 返回 API 索引](./README.md)
 
@@ -7,8 +7,13 @@
 |---|---|---|---|---|
 | `status` | string | ❌ | null | 反馈状态,精确匹配:`new`(待处理) / `handled`(已处理);传空/不传则不筛 |
 | `user_id` | int | ❌ | null | 按提交用户 id 精确筛 |
+| `content` | string | ❌ | null | 反馈内容模糊匹配(ilike,≤100 字) |
+| `created_from` | datetime | ❌ | null | 提交时间 ≥(ISO,统一按 UTC 比较) |
+| `created_to` | datetime | ❌ | null | 提交时间 ≤(ISO,统一按 UTC 比较) |
+| `sort_by` | string | ❌ | `id` | 排序列:`id` / `created_at` |
+| `sort_order` | string | ❌ | `desc` | `asc` / `desc` |
 | `limit` | int | ❌ | 20 | 1–100 |
-| `cursor` | int | ❌ | null | 上一页 next_cursor(按 feedback id 倒序,查 `id < cursor`) |
+| `cursor` | int | ❌ | null | 上一页 next_cursor(**offset 分页**,cursor=offset) |
 
 ## 出参
 响应 `200`:`{ items: FeedbackOut[], next_cursor: int|null }`(`next_cursor=null` 表示末页)
@@ -26,9 +31,10 @@
 
 ## 错误码
 - `401` 未带/无效/过期 admin token、管理员被禁用(头带 `WWW-Authenticate: Bearer`)
-- `422` `limit` 超出 1–100 范围 / 字段类型不合法
+- `422` `limit` 超出 1–100 范围 / `sort_by`·`sort_order` 不在允许集 / 字段类型不合法
 
 ## 说明
-- 游标分页约定:结果按 feedback `id` 倒序;`cursor` 传上一页返回的 `next_cursor`;`next_cursor=null` 即末页。
+- **offset 分页**(同 [admin-users-list](./admin-users-list.md)):`cursor` 即 offset,传上一页返回的 `next_cursor`;`next_cursor=null` 即末页。改用 offset 是为了在任意列排序下游标语义统一,代价是翻页期间数据变动可能错位一条(admin 低频可接受)。
-- `status` / `user_id` 均为精确匹配,可叠加。
+- 排序:`sort_by`(id/created_at)× `sort_order`(asc/desc),恒以 `id` 同向兜底次序。
+- `status` / `user_id` 精确匹配、`content` 模糊、`created_from`/`created_to` 时间范围,均可叠加。
 - 关联表 [feedback](../database/feedback.md);截图为相对路径,经 `GET /media/feedback/<file>` 静态读。

docs/api/admin-user-cash.md

@@ -0,0 +1,34 @@
+# POST /admin/api/users/{user_id}/cash — 手动增减/设值现金(带审计)
+
+> 所属:Admin·用户 组（前缀 `/admin/api/users`） | 鉴权:Bearer admin_token（角色:`finance`,`super_admin` 恒通过) | [← 返回 API 索引](./README.md)
+
+主要用于给无现金用户直接发钱、好让其测试提现链路。
+
+## 入参
+- 路径:`user_id`(int)
+- **application/json**:
+| 字段 | 类型 | 必填 | 说明 |
+|---|---|---|---|
+| `mode` | string | ✗ | `delta`(默认)=增减 / `set`=设为指定值 |
+| `amount_cents` | int | ✓ | `delta` 模式:现金变动(分,正=发放,负=扣减,不可为 0);`set` 模式:目标现金值(分,须 ≥ 0) |
+| `reason` | string | ✓ | 操作原因,1–128 字(必填,入审计与流水备注) |
+
+## 出参
+响应 `200`:`OkResponse` = `{ "ok": true }`
+
+## 错误码
+- `400`(`delta`)`amount_cents == 0`(`detail: "amount_cents 不能为 0"`);或负数扣减后余额会变负(`detail: "扣减后现金为负(当前余额 N 分)"`)
+- `400`(`set`)目标值为负(`detail: "目标现金值不能为负"`);或目标值等于当前余额(`detail: "当前现金已为 N 分,无需调整"`)
+- `401` 未带/无效/过期 admin token、管理员被禁用(头带 `WWW-Authenticate: Bearer`)
+- `403` 角色不足(需 `finance` 或 `super_admin`)
+- `404` 用户不存在(`detail: "用户不存在"`)
+- `422` 缺 `amount_cents`/`reason`、`reason` 长度不在 1–128、`mode` 非 `delta`/`set`、字段类型不合法 / `user_id` 非整数
+
+## 说明
+- 金额单位一律为**分**(`*_cents`);本接口只动现金余额,不涉及金币。
+- **set 模式**:读当前余额算出差值 `delta = target - 当前余额`,再复用同一套写入逻辑(故只写一笔差值流水)。目标值须 ≥ 0;差值为 0(已等于目标)直接拒绝。
+- 扣减保护:实际写入的 `delta < 0` 时若扣减后现金余额 < 0 直接拒绝(运营误操作保护);set 模式目标值 ≥ 0 天然不会扣成负。
+- 现金变动写流水 [cash_transaction](../database/cash_transaction.md):`biz_type` 实际差值为正记 `admin_grant`、为负记 `admin_deduct`(set 模式同理,不新增流水类型),`remark = admin:<reason>`(截断至 128 字)。
+- 写操作记审计 [admin_audit_log](../database/admin_audit_log.md):`action = user.cash.grant`,`target_type = user`,`target_id = user_id`,`detail = {amount_cents(=实际差值), balance_after_cents, reason}`;set 模式额外带 `{mode:"set", target_cents, before_cents}`。并记录操作 IP。
+- 现金变动 + 审计在同一事务原子提交(改钱必留痕)。
+- 关联用户表 [user](../database/user.md);现金账户 [coin_account](../database/coin_account.md);现金流水 [cash_transaction](../database/cash_transaction.md)。

docs/api/admin-user-coins.md

@@ -1,4 +1,4 @@
-# POST /admin/api/users/{user_id}/coins — 手动增减金币(带审计)
+# POST /admin/api/users/{user_id}/coins — 手动增减/设值金币(带审计)
 
 > 所属:Admin·用户 组（前缀 `/admin/api/users`） | 鉴权:Bearer admin_token（角色:`finance`,`super_admin` 恒通过) | [← 返回 API 索引](./README.md)
 
@@ -7,23 +7,26 @@
 - **application/json**:
 | 字段 | 类型 | 必填 | 说明 |
 |---|---|---|---|
-| `amount` | int | ✓ | 金币变动(个数):正=增加,负=扣减;不可为 0 |
+| `mode` | string | ✗ | `delta`(默认)=增减 / `set`=设为指定值 |
+| `amount` | int | ✓ | `delta` 模式:金币变动(正=增加,负=扣减,不可为 0);`set` 模式:目标金币值(须 ≥ 0) |
 | `reason` | string | ✓ | 操作原因,1–128 字(必填,入审计与流水备注) |
 
 ## 出参
 响应 `200`:`OkResponse` = `{ "ok": true }`
 
 ## 错误码
-- `400` `amount == 0`(`detail: "amount 不能为 0"`);或负数扣减后余额会变负(`detail: "扣减后金币为负(当前余额 N)"`)
+- `400`(`delta`)`amount == 0`(`detail: "amount 不能为 0"`);或负数扣减后余额会变负(`detail: "扣减后金币为负(当前余额 N)"`)
+- `400`(`set`)目标值为负(`detail: "目标金币值不能为负"`);或目标值等于当前余额(`detail: "当前金币已为 N,无需调整"`)
 - `401` 未带/无效/过期 admin token、管理员被禁用(头带 `WWW-Authenticate: Bearer`)
 - `403` 角色不足(需 `finance` 或 `super_admin`)
 - `404` 用户不存在(`detail: "用户不存在"`)
-- `422` 缺 `amount`/`reason`、`reason` 长度不在 1–128、字段类型不合法 / `user_id` 非整数
+- `422` 缺 `amount`/`reason`、`reason` 长度不在 1–128、`mode` 非 `delta`/`set`、字段类型不合法 / `user_id` 非整数
 
 ## 说明
 - 金币 = 个数(非现金);本接口只动金币余额,不涉及现金(`*_cents`)。
-- 扣减保护:`amount < 0` 时若扣减后金币余额 < 0 直接拒绝(运营误操作保护)。
+- **set 模式**:读当前余额算出差值 `delta = target - 当前余额`,再复用同一套写入逻辑(故只写一笔差值流水)。目标值须 ≥ 0;差值为 0(已等于目标)直接拒绝。
-- 金币变动写流水 [coin_transaction](../database/coin_transaction.md):`biz_type` 增加为 `admin_grant`、扣减为 `admin_deduct`,`remark = admin:<reason>`(截断至 128 字)。
+- 扣减保护:实际写入的 `delta < 0` 时若扣减后金币余额 < 0 直接拒绝(运营误操作保护);set 模式目标值 ≥ 0 天然不会扣成负。
-- 写操作记审计 [admin_audit_log](../database/admin_audit_log.md):`action = user.coins.grant`,`target_type = user`,`target_id = user_id`,`detail = {amount, balance_after, reason}`,并记录操作 IP。
+- 金币变动写流水 [coin_transaction](../database/coin_transaction.md):`biz_type` 实际差值为正记 `admin_grant`、为负记 `admin_deduct`(set 模式同理,不新增流水类型),`remark = admin:<reason>`(截断至 128 字)。
+- 写操作记审计 [admin_audit_log](../database/admin_audit_log.md):`action = user.coins.grant`,`target_type = user`,`target_id = user_id`,`detail = {amount(=实际差值), balance_after, reason}`;set 模式额外带 `{mode:"set", target, before}`。并记录操作 IP。
 - 金币变动 + 审计在同一事务原子提交(改钱必留痕)。
 - 关联用户表 [user](../database/user.md);金币账户 [coin_account](../database/coin_account.md);金币流水 [coin_transaction](../database/coin_transaction.md)。

docs/api/admin-users-list.md

@@ -1,4 +1,4 @@
-# GET /admin/api/users — 用户列表(筛选+分页)
+# GET /admin/api/users — 用户列表(筛选+排序+分页)
 
 > 所属:Admin·用户 组（前缀 `/admin/api/users`） | 鉴权:Bearer admin_token（角色:任意已登录管理员,无 require_role） | [← 返回 API 索引](./README.md)
 
@@ -8,8 +8,13 @@
 | `phone` | string | ❌ | null | 手机号**前缀**匹配(`phone LIKE '<值>%'`) |
 | `register_channel` | string | ❌ | null | 注册渠道,精确匹配 |
 | `status` | string | ❌ | null | 用户状态,精确匹配:`active` / `disabled` / `deleted` |
+| `nickname` | string | ❌ | null | 昵称**模糊**匹配(`ILIKE '%<值>%'`) |
+| `created_from` / `created_to` | datetime | ❌ | null | 注册时间范围(ISO 8601;按 UTC 比较) |
+| `last_login_from` / `last_login_to` | datetime | ❌ | null | 最近登录时间范围(ISO 8601;按 UTC 比较) |
+| `sort_by` | string | ❌ | `id` | 排序列:`id` / `created_at` / `last_login_at` |
+| `sort_order` | string | ❌ | `desc` | `asc` / `desc` |
 | `limit` | int | ❌ | 20 | 1–100 |
-| `cursor` | int | ❌ | null | 上一页 next_cursor(按 user id 倒序,查 `id < cursor`) |
+| `cursor` | int | ❌ | null | 上一页 next_cursor(**offset 偏移量**) |
 
 ## 出参
 响应 `200`:`{ items: AdminUserListItem[], next_cursor: int|null }`(`next_cursor=null` 表示末页)
@@ -28,10 +33,11 @@
 
 ## 错误码
 - `401` 未带/无效/过期 admin token、管理员被禁用(头带 `WWW-Authenticate: Bearer`)
-- `422` `limit` 超出 1–100 范围 / 字段类型不合法
+- `422` `limit` 超出 1–100 范围 / `sort_by`·`sort_order` 不在白名单 / 日期格式不合法 / 字段类型不合法
 
 ## 说明
-- 游标分页约定:结果按 user `id` 倒序;`cursor` 传上一页返回的 `next_cursor`;`next_cursor=null` 即末页。
+- **分页为 offset 偏移式**(为支持任意列排序):`cursor` 传上一页返回的 `next_cursor`(实为 offset);`next_cursor=null` 即末页。代价:翻页期间数据变动可能错位一条,admin 低频场景可接受(同提现列表)。
-- `phone` 为前缀匹配(`LIKE '<值>%'`),`register_channel` / `status` 为精确匹配;三者可叠加。
+- 排序:`sort_by` 限 `id`/`created_at`/`last_login_at`,非法值回落 `id`;同值再按 `id` 同向兜底,次序稳定。
+- 筛选:`phone` 前缀、`nickname` 模糊(`ILIKE`)、`register_channel`/`status` 精确、`created_*`/`last_login_*` 时间范围;均可叠加。
 - 关联用户表 [user](../database/user.md)。
 - 历史明细(金币流水、提现、比价、反馈等)不在本列表,走各自带 `user_id` 过滤的分页接口。

docs/api/admin-wallet-cash-transactions.md

@@ -25,7 +25,7 @@
 | `biz_type` | string | 业务类型 |
 | `ref_id` | string \| null | 关联业务 ID |
 | `remark` | string \| null | 备注 |
-| `created_at` | datetime | 创建时间（ISO 8601 UTC） |
+| `created_at` | datetime | 创建时间（北京 wall-clock naive,非 UTC;前端按字面显示,不做时区换算） |
 
 ## 错误码
 - `401` 未带/无效/过期 admin token、管理员被禁用（响应头带 `WWW-Authenticate: Bearer`）

docs/api/admin-wallet-coin-transactions.md

@@ -25,7 +25,7 @@
 | `biz_type` | string | 业务类型 |
 | `ref_id` | string \| null | 关联业务 ID |
 | `remark` | string \| null | 备注 |
-| `created_at` | datetime | 创建时间（ISO 8601 UTC） |
+| `created_at` | datetime | 创建时间（北京 wall-clock naive,非 UTC;前端按字面显示,不做时区换算） |
 
 ## 错误码
 - `401` 未带/无效/过期 admin token、管理员被禁用（响应头带 `WWW-Authenticate: Bearer`）

docs/api/feedback.md

@@ -7,8 +7,8 @@
 | 字段 | 类型 | 必填 | 说明 |
 |---|---|---|---|
 | `content` | string | ✓ | 反馈正文,**1-2000 字**(strip 后) |
-| `contact` | string | ✓ | 联系方式(微信/QQ/手机号),**1-128 字**,便于回访 |
+| `contact` | string | ✗ | 联系方式(微信/QQ/手机号),**≤128 字**。原型改版后客户端已不再采集、不传该字段(后端默认空串);保留字段兼容旧端 |
-| `images` | file[] | ✗ | 截图,**最多 4 张**,每张走头像同款校验(JPEG/PNG/WebP,≤ 5 MB,魔数嗅探) |
+| `images` | file[] | ✗ | 截图,**最多 6 张**,每张走头像同款校验(JPEG/PNG/WebP,≤ 5 MB,魔数嗅探) |
 
 ## 出参
 响应 `200`:
@@ -29,9 +29,9 @@
 > 不返回上传的 image URL——这是给运营后台看的,客户端通常不需要。
 
 ## 错误码
-- `400` 内容为空 / 内容超 2000 字 / 联系方式为空 / 联系方式超 128 字 / 图片超 4 张 / 单图非法(空/过大/格式不对)
+- `400` 内容为空 / 内容超 2000 字 / 联系方式超 128 字 / 图片超 6 张 / 单图非法(空/过大/格式不对)
 - `401` 未带 token / token 无效或过期 / 用户被禁用
-- `422` 缺 `content` 或 `contact` 字段
+- `422` 缺 `content` 字段
 
 ## 说明
 - **反馈绑用户**:`feedback.user_id = current_user.id`,便于回访

docs/api/platform-savings-feed.md

@@ -2,7 +2,11 @@
 
 > 所属:Platform 组（前缀 `/api/v1/platform`） | 鉴权:**无** | [← 返回 API 索引](./README.md)
 
-客户端首页顶部「用户****xxx 比价后节省 xx 元」滚动条数据源。全平台真实比价记录优先;不足时用运营配的种子([ops_marquee_seed](../database/ops_marquee_seed.md))补齐到 `limit` 条「混播」,保证轮播不空。
+客户端首页顶部「某用户 比价后节省 xx 元」滚动条数据源。全平台真实比价记录优先;不足时用运营配的种子([ops_marquee_seed](../database/ops_marquee_seed.md))补齐到 `limit` 条「混播」,保证轮播不空。
+
+**用户标识脱敏规则(全 feed 统一)**:① 有昵称 → 昵称脱敏:≥3 字「首+隐藏字数个星+末」(省钱小能手→`省***手`)、2 字「首+星」(阿强→`阿*`)、1 字「用户+该字」(喵→`用户喵`);② 没昵称 → 按 `user_id` **确定性合成一个假昵称再脱敏**(同用户恒定,避免无昵称用户清一色 `用户****`),少数露「用户」+ 8 星 + id 后 3 位(`用户********618`)。
+
+> 合成假名走**本地语料组合生成**(无外部库):中文真名(姓池 ×名字字池 拼「姓+1~2字」) / 中文网络昵称 / 英文昵称 / 英文名(可带数字尾) 混播,中文为主英文为辅,组合空间上万。真实条按 `user_id` 播种 → 同一用户名字恒定、不同用户各异(刷新/翻页不变脸,同屏几乎不撞名);种子/兜底用运行时随机源出多样。当前真实用户大多没设昵称,故走合成。
 
 ## 入参
 | 参数 | 类型 | 说明 |
@@ -15,15 +19,15 @@
 | 字段 | 类型 | 说明 |
 |---|---|---|
 | `items` | list | 轮播条目数组(最多 `limit` 条) |
-| `items[].masked_user` | string | 脱敏用户名,手机尾号(`138****5678`)/ 中文昵称(`省钱**`)混合风格 |
+| `items[].masked_user` | string | 脱敏用户名:昵称/合成名脱敏(`省***手` / `王*` / `SaveK**g`)或匿名 `用户********618` |
 | `items[].saved_amount_cents` | int | 节省（分），客户端 ÷100 显示「x.xx 元」 |
 | `items[].time` | string | 北京时间 `HH:MM:SS` |
 
 ## 说明
-- 真实条:`comparison_record` 中 `status='success'` 且 `0 < saved_amount_cents ≤ 300 元` 的近期记录(金额超 300 元视为异常 / bug 值剔除,防「节省 999 元」穿帮),**按 `user_id` 去重**(同一用户只取最新一条,避免单人刷屏);用户名按 `user_id` **确定性合成**手机尾号 / 中文昵称混合脱敏名(同用户恒定)。
+- 真实条:`comparison_record` 中 `status='success'` 且 `0 < saved_amount_cents ≤ 300 元` 的近期记录(金额超 300 元视为异常 / bug 值剔除,防「节省 999 元」穿帮),**JOIN `user` 表取昵称 → 仅 user 表内用户的记录入选**(孤儿/已删用户记录不计,设计如此;当前用户量少,缺口由种子补足),**按 `user_id` 去重**(同一用户只取最新一条,避免单人刷屏);有昵称按真实昵称脱敏,无昵称按 `user_id` 播种**确定性合成**脱敏名(同用户恒定、刷新不变脸)。
 - 种子条:`ops_marquee_seed`(`enabled=true`),仅在真实去重后不足 `limit` 时补齐;**从启用种子中公平随机抽取**(不再固定取前 N,所有种子都有机会露出)。每条种子:用户名留空或旧 `用户****xxx` 模板名则**随机合成混合风格名并避开同屏撞名**(自愈历史种子),金额在 `[min_cents, max_cents]` 取**长尾随机值**(小额居多、偶尔大额,固定金额则 min==max)。
 - 兜底:真实 + 种子仍凑不满 `limit`(种子被全停用 / 太少)时,用**内置合成条补满**,保证轮播既不空也不稀疏。
 - **`time` 为合成的「最近」时间**(从当前北京时间往前**随机抖动**递减,首条几十秒前,其余多数 1~5 分钟、偶尔扎堆或较长):社会证明轮播保证永远像刚发生且节奏自然不机械,不受旧测试数据 / 低谷期记录影响。真实的用户/金额不变,只换展示时间。
-- 因含随机(抽取 / 金额 / 合成名),**每次请求结果都不同**——这是轮播想要的鲜活感。
+- 因含随机(种子抽取 / 金额 / 展示时间 / 种子合成名),**每次请求结果都不同**——这是轮播想要的鲜活感;但**真实用户的脱敏名按 `user_id` 恒定**,同一用户跨请求不变脸(避免「同一人换张脸」露馅)。
 - **性能**:首页人人都看、真实数据变化慢,真实记录原始行带 **~30s 进程内缓存**(展示层随机仍每次重算,只省 DB 往返;新记录最多晚 30s 进轮播)。查询走复合索引 `ix_comparison_status_created (status, created_at)`,避免随数据量增大全表扫。
 - 逻辑见 `app/repositories/ops_marquee.py`;运营管理种子见 [admin-marquee-seeds](./admin-marquee-seeds.md)。

docs/database/OVERVIEW.md

@@ -2,7 +2,7 @@
 
 > 跨表视角。单表字段级细节看同目录 `<表名>.md`(索引见 [README](./README.md))。
 > 本文专门回答三件「跨表」的事:**① 每块 App 功能用到哪些表 ② 什么操作往哪张表写 ③ 表和表怎么连(join key,含没有外键约束、靠业务字段对齐的语义关联)**。
-> **范围**:业务表全部在 `shaguabijia-app-server`(SQLAlchemy 2.0 + SQLite 开发 / PostgreSQL 生产)。`pricebot-backend`(比价/领券 Agent)是纯内存态、**无任何表**;Android 客户端只有 EncryptedSharedPreferences / SharedPreferences、**无关系库**。共 **23 张业务表** + `alembic_version`(框架的迁移版本指针)。
+> **范围**:业务表全部在 `shaguabijia-app-server`(SQLAlchemy 2.0 + SQLite 开发 / PostgreSQL 生产)。`pricebot-backend`(比价/领券 Agent)是纯内存态、**无任何表**;Android 客户端只有 EncryptedSharedPreferences / SharedPreferences、**无关系库**。共 **28 张业务表** + `alembic_version`(框架的迁移版本指针)。领券联动的「今日状态」三张表(`coupon_*`)同理:领券过程在 pricebot 内存态跑、**不落库**,只有结果回到 app-server 才落这三张表。
 
 ---
 
@@ -17,6 +17,14 @@
 | profile「累计省了 / 省钱战绩 / 省钱明细」 | [`savings_record`](./savings_record.md) | 真实下单归因(source=compare)+ 无真实数据时 demo 兜底 |
 | 「上报更低价」提交 / 列表 | [`price_report`](./price_report.md) | 众包纠偏:用户举证某平台更便宜,人工审核发奖 |
 
+### 领券(每日领券联动 · 今日状态)
+| App 位置 / 动作 | 表 | 说明 |
+|---|---|---|
+| 领券**过程**(看屏→领券) | （无） | 在 pricebot-backend 内存态跑,**过程不落库**;结果回 app-server 才落下面三张表 |
+| 切外卖 App 时是否弹领券引导窗 | [`coupon_prompt_engagement`](./coupon_state.md) | 今天 engage 过(点领/点拒)就不再弹;判断维度 device_id |
+| 首页「去领取」卡是否置灰 | [`coupon_daily_completion`](./coupon_state.md) | 今天跑完整轮(到 done)就置灰;判断维度 device_id |
+| 每张券领取结果留痕 | [`coupon_claim_record`](./coupon_state.md) | 资产/画像/排查/CPS;当前**不参与**判断 |
+
 ### 钱包 / 福利(看广告赚钱闭环)
 | App 位置 / 动作 | 表 | 说明 |
 |---|---|---|
@@ -76,6 +84,13 @@
 | 首次进 profile 省钱页且无真实记录 | `savings_record`(C `source=demo`) | 懒种子,`ensure_seeded` 按 user 幂等 |
 | 上报更低价 `POST /report` | `price_report`(C) | 读 `comparison_record.best_price_cents` 校验 |
 | 提交反馈 `POST /feedback` | `feedback`(C) | |
+| 领券首帧 `POST /api/v1/coupon/step`(step=0) | `coupon_prompt_engagement`(C/U `claim_started`) | `(device_id, 北京日)` 幂等;best-effort |
+| 领券每帧结果 `POST /api/v1/coupon/step` | `coupon_claim_record`(C/U) | `(device_id, coupon_id, 北京日)` 幂等;best-effort |
+| 领券跑完 `POST /api/v1/coupon/step`(action.command=done) | `coupon_daily_completion`(C/U) | `(device_id, 北京日)` 幂等;best-effort |
+| 拒绝领券引导窗 `POST /api/v1/coupon/prompt/dismiss` | `coupon_prompt_engagement`(C/U `dismissed`) | 同上;客户端通知(透传链路看不到拒绝) |
+| 重置今日弹窗 `POST /api/v1/coupon/prompt/reset`(开发) | `coupon_prompt_engagement`(**D** 今日条) | 删后今天又能弹 |
+
+> 领券三表写库**全 best-effort**:`/coupon/step` 里写失败只 `logger.warning`、不连累领券返回;**判断只看 `device_id`**,`user_id` 可空旁路(资产留痕)。
 
 ### admin 端(管理员触发,均额外写一条 `admin_audit_log`)
 | 后台操作 | 写入 | 操作 |
@@ -122,6 +137,7 @@
 - **`comparison_record.store_name` ≈ `savings_record.shop_name`**:无 id 关联,按**店名字符串相等**给比价记录打「已下单」标记(瞬态,不写库)。两边店名同源 = 比价意图识别阶段的门店 query,语义=**店级**(同店比价多次会一并标已下单)。
 - **广告流会话关联**:`ad_reward_record.ad_session_id` 可与 `ad_ecpm_record.ad_session_id` 对齐;`ad_watch_log` 仍是旧版兼容统计,不逐条参与发奖。
 - **里程碑解锁进度不存库**:`comparison_milestone_claim` 只记「哪几档已领」;进度 = `comparison_record` 里 `status='success'` 的 `count`。
+- **领券三表无硬 FK,全靠软关联**:`coupon_prompt_engagement` / `coupon_daily_completion` / `coupon_claim_record` 的 `user_id` **软指** `user.id`(可空、有登录态才记、不进唯一键、不阻塞判断);`trace_id` **软指** pricebot work_logs(排查回指);唯一键都以 `device_id` + 北京自然日为主(详见 [`coupon_state.md`](./coupon_state.md))。
 - **`onboarding_completion.(user_id, device_id)`**:`user_id` 语义关联 `user.id`(无硬 FK,同 `coupon_*` 设备表),`device_id` = 客户端硬件级 `ANDROID_ID`(≠ 领券 per-install `device_id`)。登录读、走完引导写,决定是否再展示新手引导。
 
 ### ER 关系(文字版)
@@ -136,6 +152,8 @@ user ─1:N─ onboarding_completion             (user_id, 无硬 FK; (user_id,d
 comparison_record ─1:N─ price_report        (comparison_record_id, 可空)
 admin_user ─1:N─ admin_audit_log
 app_config                                   (独立, 无外键, key 为主键)
+coupon_prompt_engagement / coupon_daily_completion / coupon_claim_record
+                                             (独立, 无硬 FK; 维度=device_id+北京日, user_id/trace_id 仅软关联)
 ```
 
 ---

docs/database/README.md

@@ -3,16 +3,13 @@
 > 数据库:SQLite 起步(`data/app.db`),生产可切 PostgreSQL(改 `DATABASE_URL`)。
 > ORM:SQLAlchemy 2.0(`app/models/`),迁移:Alembic(`alembic/versions/`,`render_as_batch` 兼容 SQLite)。
 > 金额字段一律存**整数**:金币=个数,现金=**分**(`*_cents`)。时间列 `DateTime(timezone=True)`。
-
+> 最后更新:2026-06-11(合并:新增 3 张领券今日状态表 `coupon_*` + `onboarding_completion` 新手引导完成表;含 [OVERVIEW 总览](./OVERVIEW.md))
-> 最后更新:2026-06-10(新增 `onboarding_completion` 新手引导完成表 → 23 张业务表)
-=======
-
 
 > 🧭 **先看 [OVERVIEW.md — 表 × 功能 × 关系](./OVERVIEW.md)**:跨表的「每块功能用哪些表 / 什么操作写哪张表 / 表间 join key」都在那;本页只做**单表索引**,点进每张表的详情看字段级说明。
 
 ---
 
-## 表总览(23 张业务表 + `alembic_version` 框架表)
+## 表总览(28 张业务表 + `alembic_version` 框架表)
 
 ### 账号 / 反馈
 | 表 | 用途 | 模型 | 文档 |
@@ -45,6 +42,13 @@
 | `savings_record` | 省钱记录(profile 省钱战绩源;真实下单归因 + demo) | `models/savings.py` | [详情](./savings_record.md) |
 | `price_report` | 上报更低价(众包纠偏,人工审核发奖) | `models/price_report.py` | [详情](./price_report.md) |
 
+### 领券(每日领券联动 · 今日状态)
+| 表 | 用途 | 模型 | 文档 |
+|---|---|---|---|
+| `coupon_prompt_engagement` | 领券引导窗频控源(今日是否已 engage,按 device+日) | `models/coupon_state.py` | [详情](./coupon_state.md) |
+| `coupon_daily_completion` | 首页「去领取」置灰源(今日是否已跑完整轮) | `models/coupon_state.py` | [详情](./coupon_state.md) |
+| `coupon_claim_record` | 每张券领取结果沉淀(资产/画像/排查,不参与判断) | `models/coupon_state.py` | [详情](./coupon_state.md) |
+
 ### 美团 CPS 券缓存
 | 表 | 用途 | 模型 | 文档 |
 |---|---|---|---|

docs/database/coupon_state.md

@@ -0,0 +1,123 @@
+# coupon_state — 领券今日状态三张表(弹窗频控 / 首页置灰 / 领券记录)
+
+> 模型 `app/models/coupon_state.py` · 仓库 `app/repositories/coupon_state.py` · 接口 `app/api/v1/coupon.py`(prefix `/api/v1/coupon`) · [← 索引](./README.md) · [总览](./OVERVIEW.md)
+
+领券（优惠券自动化）联动产生的三张「今日状态」表，都挂在领券透传端点 `POST /api/v1/coupon/step` 这条链路上（pricebot 跑领券，结果回 app-server 落库；**领券过程本身在 pricebot 内存态跑、不落库**）。三表各管一件事：
+
+- **`coupon_prompt_engagement`** — 弹窗频控源。按 `(device, 自然日)` 记「今天是否对领券引导窗表达过**意向**」（点「一键领取」=`claim_started` / 点拒绝关闭=`dismissed` 都算）。切到外卖 App 时据此决定弹不弹：今天 engage 过就不再弹。
+- **`coupon_daily_completion`** — 首页置灰源。按 `(device, 自然日)` 记「今天是否已**跑完整轮**领券（到 done 帧）」。首页「去领取」卡据此置灰：今天跑完了就不能再领。
+- **`coupon_claim_record`** — 资产沉淀层。按 `(device, 券, 自然日)` 记每张券的领取结果（success/already_claimed/failed/skipped），**纯沉淀**（资产/画像/排查/CPS 归因），当前**不参与**「要不要领 / 弹不弹」的判断。
+
+三表共同口径：
+- **判断维度是 `device_id`，不是 `user_id`**：券发到的是设备上登录的那个外卖账号，device 比 user 更贴近「哪个登录环境」，且 `device_id` 全链路现成、不依赖领券鉴权（领券 MVP 阶段 `/coupon/step` 不鉴权）。客户端 `getOrCreateDeviceId` 生成存 SP，**卸载重装会变 → 当新设备重新弹一次**（产品预期）。
+- **日期 = `Asia/Shanghai` 自然日**（`claim_date` / `engage_date` / `complete_date`，`repositories/coupon_state.today_cn()`）。每日可领的券（签到/天天红包）靠这天然每天一条。
+- **`user_id` 可空**：领券登录态有就记（资产/画像），可空、**不进唯一键、不阻塞判断**。
+- **`trace_id` 可空**：回指 pricebot work_logs，供排查（哪次任务领的）。
+- **engagement vs completion vs claim 的区别**：engagement = 用户**表达过意向**（点了领或拒，不管跑没跑完）；completion = 这一轮**真跑到了 done**（整套流程走完）；claim = **每张券一条**的结果留痕。
+
+> 写库全部 **best-effort**：在 `/coupon/step` 里写库失败只 `logger.warning`、**绝不连累领券主流程/返回**（见 `app/api/v1/coupon.py`）。
+
+---
+
+## coupon_prompt_engagement — 弹窗频控（今日是否已对引导窗表达意向）
+
+`(device_id, engage_date)` 唯一，一台设备一天一条；今天 engage 过（领或拒）就不再弹。
+
+### 用在哪 / 增删改查
+- **C / U（幂等 upsert）**：`mark_engagement`。两条触发：
+  - `POST /api/v1/coupon/step` 且 `step==0`（领券首帧=用户已发起领券）→ 记 `claim_started`；
+  - `POST /api/v1/coupon/prompt/dismiss`（用户点关闭引导窗；server 在透传链路看不到「拒绝」，必须客户端通知）→ 记 `dismissed`。
+  - 已有今天那条则覆盖 `engage_type`（并补 `user_id`），否则插入。
+- **D**：`POST /api/v1/coupon/prompt/reset`（`reset_today_engagement`）—— 删这台设备今天那条，开发设置「重置今日领券弹窗状态」按钮调，测频控用；删后今天又能弹。
+- **R**：`GET /api/v1/coupon/prompt/should-show?device_id=…`（`has_engaged_today`）→ `should_show = not 今天已 engage`。客户端切外卖 App 前查，纯后台判据。
+
+### 字段
+| 列 | 类型 | 约束 / 默认 | 说明(取值 / join) |
+|---|---|---|---|
+| `id` | Integer | PK, autoincrement | |
+| `device_id` | String(64) | NOT NULL | 判断/聚合维度；客户端 `getOrCreateDeviceId`，重装会变 |
+| `user_id` | Integer | index, 可空 | 登录态有就记（资产）；不进唯一键、不阻塞判断 |
+| `engage_date` | **Date** | NOT NULL | **北京时间**自然日（`today_cn()`） |
+| `engage_type` | String(16) | NOT NULL | `claim_started`（点一键领取）/ `dismissed`（点拒绝关闭）；仅记录区分，**判断只看「今天有没有这条」，type 不影响弹不弹** |
+| `created_at` | DateTime(tz) | server_default now() | |
+| `updated_at` | DateTime(tz) | server_default now(), onupdate now() | |
+
+### 索引与约束
+- PK `id`；index `user_id`；UNIQUE(`device_id`, `engage_date`) = `uq_coupon_engage_device_date`（一台设备一天一条）。
+
+### 注意
+- `device_id` 重装会变 → 重装当新设备，今天重新弹一次（产品预期）。
+- 判断只看「今天这台设备有没有这条」，不看 `engage_type`（领或拒都算 engage 过、都不再弹）。
+
+---
+
+## coupon_daily_completion — 首页置灰（今日是否已跑完整轮领券）
+
+`(device_id, complete_date)` 唯一，一台设备一天一条；今天跑完整轮（到 done 帧）就把首页「去领取」卡置灰。
+
+### 用在哪 / 增删改查
+- **C / U（幂等 upsert）**：`mark_completed_today`，由 `POST /api/v1/coupon/step` 在 pricebot 返回 `action.command == "done"` 那帧调。pricebot 把中途单券 done 改写成 `wait+continue=true`，只有整套全跑完那帧才保留 `command=="done"`，故 **done 已等价「整轮完成」**。已有今天那条则补 `user_id`/`trace_id`，否则插入。
+- **U / D**：无业务删除。
+- **R**：`GET /api/v1/coupon/completed-today?device_id=…`（`has_completed_today`）→ `completed`。客户端据此把首页「去领取」卡置灰、不可点。
+
+### 字段
+| 列 | 类型 | 约束 / 默认 | 说明(取值 / join) |
+|---|---|---|---|
+| `id` | Integer | PK, autoincrement | |
+| `device_id` | String(64) | NOT NULL | 判断维度，与 engagement/claim 一致；客户端两端都用 ANDROID_ID |
+| `user_id` | Integer | index, 可空 | 登录态有就记（资产） |
+| `complete_date` | **Date** | NOT NULL | **北京时间**自然日（`today_cn()`） |
+| `trace_id` | String(64) | index, 可空 | 哪次任务跑到 done，回指 pricebot work_logs / 排查 |
+| `created_at` | DateTime(tz) | server_default now() | |
+| `updated_at` | DateTime(tz) | server_default now(), onupdate now() | |
+
+### 索引与约束
+- PK `id`；index `user_id`、`trace_id`；UNIQUE(`device_id`, `complete_date`) = `uq_coupon_completion_device_date`（一台设备一天一条）。
+
+### 注意
+- 口径（用户决策 2026-06-10 A 方案）：**到 done 即算完成，不管单券成败**——失败/跳过常是无障碍/环境问题，重复点也补不回来。
+- 与 engagement 区别：engagement 是「表达过意向」（点了就记，不管跑没跑完）；completion 是「真跑到了 done」。
+
+---
+
+## coupon_claim_record — 领券记录（每张券一天一条，资产沉淀层）
+
+`(device_id, coupon_id, claim_date)` 唯一，同设备同券同一天只一条；纯沉淀，**当前不参与判断**，留作以后按券去重 / CPS 归因 / 用户画像的数据源。
+
+### 用在哪 / 增删改查
+- **C / U（幂等 upsert）**：`record_claims`，由 `POST /api/v1/coupon/step` 写入。一帧的券结果来自 pricebot 的 `last_coupon_result`（最后一张）+ `action.params.coupon_results`（全量）——**会重复带同一张券**，端点 `_extract_coupon_results` 先**按 `coupon_id` 去重**（全量覆盖单张），仓库再靠唯一键幂等：已有则更新 `status`/`reason`/`claimed_count`/`extra`（以最后一次为准），否则插入。
+- **U / D**：无业务删除。
+- **R**：**当前无读取端点**（纯写入沉淀，未来做去重/归因/画像时再用）。
+
+### 字段
+| 列 | 类型 | 约束 / 默认 | 说明(取值 / join) |
+|---|---|---|---|
+| `id` | Integer | PK, autoincrement | |
+| `device_id` | String(64) | NOT NULL | 聚合维度；客户端 `getOrCreateDeviceId`，重装会变 |
+| `user_id` | Integer | index, 可空 | 登录态有就记（资产/画像）；不进唯一键 |
+| `coupon_id` | String(64) | NOT NULL | 券标识（取自 pricebot 结果） |
+| `claim_date` | **Date** | NOT NULL | **北京时间**自然日（`today_cn()`）；每日可领的券靠它天然每天一条 |
+| `status` | String(24) | NOT NULL | `success` / `already_claimed` / `failed` / `skipped`（原样取 pricebot coupon 结果） |
+| `vendor` | String(48) | 可空 | 券提供方 |
+| `coupon_name` | String(128) | 可空 | 取 pricebot `name` |
+| `claimed_count` | Integer | 可空 | 这张领到几张（pricebot `display_count`，给不出时 None；兼容 `claimed_count`） |
+| `trace_id` | String(64) | index, 可空 | 哪次任务领的，回指 pricebot work_logs / 排查 |
+| `reason` | String(255) | 可空 | failed / skipped 原因 |
+| `extra` | JSON（PG JSONB） | 可空 | 杂项兜底：券的结构化信息（面额/入口/关键节点摘要等），免得加字段就迁移；当前直接存这帧 pricebot 单券结果 dict |
+| `created_at` | DateTime(tz) | server_default now() | |
+| `updated_at` | DateTime(tz) | server_default now(), onupdate now() | |
+
+### 索引与约束
+- PK `id`；index `user_id`、`trace_id`；UNIQUE(`device_id`, `coupon_id`, `claim_date`) = `uq_coupon_claim_device_coupon_date`；Index(`device_id`, `claim_date`) = `ix_coupon_claim_device_date`（按 device+日 取一天所有券）。
+
+### 注意
+- **`extra` 别塞原始无障碍树**（几十 KB → 行膨胀）；原始大树看 `trace_id` 指过去的 work_logs。
+- 同批去重很关键：`autoflush=False` 下同 `coupon_id` 两次 `add` 会撞唯一约束、`IntegrityError` 回滚整批（done/单券记录全丢），故端点 `_extract_coupon_results` + 仓库 `seen` 集合双重防御。
+- `extra` 是 `JSON().with_variant(JSONB(), "postgresql")`：PG 用 JSONB（可建 GIN 索引），SQLite 退化通用 JSON（同 `price_observation` / `comparison_record`）。
+
+---
+
+## 三表共性小结
+- 数据流向：客户端 → `POST /api/v1/coupon/step`（透传给 pricebot）→ 结果回写这三张表（best-effort，写库失败不影响领券）。
+- 唯一键都含 `device_id` + 某个北京自然日列；`user_id` 永远是可空旁路（资产留痕，不进唯一键、不阻塞判断）。
+- 无硬外键：`user_id` 软指 `user.id`、`trace_id` 软指 pricebot work_logs（详见 [OVERVIEW → 表间关系 & Join Key](./OVERVIEW.md)）。

docs/database/feedback.md

@@ -2,10 +2,10 @@
 
 > 模型 `app/models/feedback.py` · 仓库 `app/repositories/feedback.py` · 接口 [feedback](../api/feedback.md) · admin [admin-feedbacks-list](../api/admin-feedbacks-list.md) / [admin-feedback-handle](../api/admin-feedback-handle.md) · [← 索引](./README.md) · [总览](./OVERVIEW.md)
 
-App「帮助与反馈」每次提交写一行。`content` 与 `contact` 必填,`images` 为可选截图。后台人工处理后置 `handled`。与 `price_report`(结构化上报更低价)不同,本表是**自由文本**反馈。
+App「帮助与反馈」每次提交写一行。`content` 必填;`contact` 原必填,**原型改版后客户端不再采集,新数据存空串**(列保持 NOT NULL、免迁移,历史数据仍有值);`images` 为可选截图(≤6 张)。后台人工处理后置 `handled`。与 `price_report`(结构化上报更低价)不同,本表是**自由文本**反馈。
 
 ## 用在哪 / 增删改查
-- **C(插入)**:`POST /api/v1/feedback`(multipart:`content` + `contact` + 可选 `images`;`create_feedback`)。截图先经 `core.media` 落 `/media/feedback/` 拿相对路径,再随反馈写入,`status='new'`。
+- **C(插入)**:`POST /api/v1/feedback`(multipart:`content` + 可选 `contact` + 可选 `images`;`create_feedback`)。截图先经 `core.media` 落 `/media/feedback/` 拿相对路径,再随反馈写入,`status='new'`。
 - **U(更新)**:admin 处理反馈 `update_feedback_status` → `status='handled'`(同事务写 `admin_audit_log`)。
 - **D**:无。
 - **R**:admin 反馈列表(可按 `status` 筛)。C 端当前无"我的反馈列表"读接口。
@@ -16,7 +16,7 @@ App「帮助与反馈」每次提交写一行。`content` 与 `contact` 必填,`
 | `id` | Integer | PK, autoincrement | |
 | `user_id` | Integer | FK→user.id, index, NOT NULL | 提交用户 |
 | `content` | Text | NOT NULL | 反馈正文 |
-| `contact` | String(128) | NOT NULL | 联系方式(微信/QQ/手机,便于回访) |
+| `contact` | String(128) | NOT NULL | 联系方式(微信/QQ/手机)。客户端改版后不再采集,新数据为空串;列仍 NOT NULL |
 | `images` | JSON | nullable | 截图相对 URL 列表 `/media/feedback/...`;无图为 NULL |
 | `status` | String(16) | NOT NULL, default `new` | 取值:`new`(待处理)/ `handled`(已处理) |
 | `created_at` | DateTime(tz) | server_default now(), index | 提交时间 |

docs/database/ops_marquee_seed.md

@@ -8,7 +8,7 @@
 | 列 | 类型 | 约束 / 默认 | 说明 |
 |---|---|---|---|
 | `id` | Integer | PK, autoincrement | |
-| `masked_user` | String(64) | NULL 可空 | 脱敏用户名,手机尾号(`138****5678`)/ 中文昵称(`省钱**`)风格(整串存)。**留空或旧 `用户****xxx` 模板名 → feed 展示时随机合成混合风格名**(避开同屏撞名、自愈历史种子),风格与真实条一致、永不穿帮 |
+| `masked_user` | String(64) | NULL 可空 | 脱敏用户名(整串存)。**留空或旧 `用户****xxx` 模板名 → feed 展示时按脱敏规则随机合成**(本地姓/名字池组合的真名 + 中英网络昵称语料,首末字+星;避开同屏撞名、自愈历史种子),风格与真实条一致、永不穿帮。详见 [platform-savings-feed](../api/platform-savings-feed.md) |
 | `min_cents` | Integer | NOT NULL | 节省金额区间下限(分) |
 | `max_cents` | Integer | NOT NULL | 节省金额区间上限(分);feed 每次在 `[min,max]` 随机取值,**固定金额则 min==max** |
 | `enabled` | Boolean | NOT NULL, default true | 停用的不参与混播 |

tests/test_admin.py

@@ -125,7 +125,7 @@ def test_long_password_does_not_crash(admin_client: TestClient) -> None:
 
 
 def test_audit_log_pagination_no_gap() -> None:
-    """审计游标分页跨页不丢/不重(回归 next_cursor off-by-one)。"""
+    """审计分页跨页不丢/不重(offset 分页:cursor 即 offset,翻完覆盖全部)。"""
     from app.admin.repositories import admin_user as admin_repo
     from app.admin.repositories import audit_log as audit_repo
 
@@ -142,12 +142,13 @@ def test_audit_log_pagination_no_gap() -> None:
             )
             created_ids.append(log.id)
 
-        # limit=2 翻 5 条,收集所有 id,应正好覆盖创建的 5 条(无丢无重)
+        # limit=2 翻 5 条,收集所有 id,应正好覆盖创建的 5 条(无丢无重);total 恒为符合条件总数
         seen: list[int] = []
         cursor = None
         for _ in range(10):  # 上限防死循环
-            rows, cursor = audit_repo.list_audit_logs(db, action=action, limit=2, cursor=cursor)
+            rows, cursor, total = audit_repo.list_audit_logs(db, action=action, limit=2, cursor=cursor)
             seen.extend(r.id for r in rows)
+            assert total == len(created_ids), f"total 应为 {len(created_ids)},得 {total}"
             if cursor is None:
                 break
         assert sorted(seen) == sorted(created_ids), f"分页丢/重: want={created_ids} got={seen}"

tests/test_admin_read.py

@@ -91,6 +91,80 @@ def test_user_filter_by_status(admin_client: TestClient, admin_token: str) -> No
     assert all(u["status"] == "active" for u in r.json()["items"])
 
 
+def test_user_list_sort_filter_range(admin_client: TestClient, admin_token: str) -> None:
+    """用户列表:渠道/昵称筛选 + id 升降序 + 时间范围 + 非法 sort_by 422。"""
+    db = SessionLocal()
+    try:
+        u1 = user_repo.upsert_user_for_login(db, phone="13811110001", register_channel="sms")
+        u1.nickname = "排序测试甲"
+        u2 = user_repo.upsert_user_for_login(db, phone="13811110002", register_channel="wechat")
+        u2.nickname = "排序测试乙"
+        db.commit()
+        id1, id2 = u1.id, u2.id
+    finally:
+        db.close()
+
+    # 渠道精确筛选
+    r = admin_client.get(
+        "/admin/api/users", params={"register_channel": "wechat"}, headers=_auth(admin_token)
+    )
+    assert r.status_code == 200, r.text
+    assert all(u["register_channel"] == "wechat" for u in r.json()["items"])
+
+    # 昵称模糊筛选
+    r = admin_client.get(
+        "/admin/api/users", params={"nickname": "排序测试", "limit": 100}, headers=_auth(admin_token)
+    )
+    ids = {u["id"] for u in r.json()["items"]}
+    assert id1 in ids and id2 in ids
+
+    # id 升序 / 降序
+    asc_ids = [
+        u["id"]
+        for u in admin_client.get(
+            "/admin/api/users",
+            params={"sort_by": "id", "sort_order": "asc", "limit": 100},
+            headers=_auth(admin_token),
+        ).json()["items"]
+    ]
+    assert asc_ids == sorted(asc_ids)
+    desc_ids = [
+        u["id"]
+        for u in admin_client.get(
+            "/admin/api/users",
+            params={"sort_by": "id", "sort_order": "desc", "limit": 100},
+            headers=_auth(admin_token),
+        ).json()["items"]
+    ]
+    assert desc_ids == sorted(desc_ids, reverse=True)
+
+    # 时间范围:2000 年之前无人;之后有人(验证范围条件确实生效)
+    assert (
+        admin_client.get(
+            "/admin/api/users", params={"created_to": "2000-01-01T00:00:00Z"},
+            headers=_auth(admin_token),
+        ).json()["items"]
+        == []
+    )
+    assert (
+        len(
+            admin_client.get(
+                "/admin/api/users", params={"created_from": "2000-01-01T00:00:00Z", "limit": 100},
+                headers=_auth(admin_token),
+            ).json()["items"]
+        )
+        >= 1
+    )
+
+    # 非法 sort_by → 422
+    assert (
+        admin_client.get(
+            "/admin/api/users", params={"sort_by": "phone"}, headers=_auth(admin_token)
+        ).status_code
+        == 422
+    )
+
+
 def test_wallet_and_withdraw_lists(admin_client: TestClient, admin_token: str) -> None:
     uid = _seed_user_with_data("13800000004")
     r = admin_client.get(
@@ -112,6 +186,65 @@ def test_feedback_list(admin_client: TestClient, admin_token: str) -> None:
     assert all(f["status"] == "new" for f in r.json()["items"])
 
 
+def test_ad_coin_audit_full_count_truncate_and_only_mismatch(
+    admin_client: TestClient, admin_token: str
+) -> None:
+    """A+B:total/mismatch_count 按全量统计(不受 limit 影响),truncated 旗标 + only_mismatch 过滤。
+
+    用 capped 行造确定性数据(应发恒 0,coin==0 即一致),不依赖发奖公式:
+    3 条 coin=0(一致) + 2 条 coin=7(不一致) → 全量 total=5、mismatch=2。
+    """
+    from app.models.ad_reward import AdRewardRecord
+
+    d = "2020-01-15"  # 固定历史日 + 独立 user,隔离其它用例数据
+    db = SessionLocal()
+    try:
+        uid = user_repo.upsert_user_for_login(db, phone="13800009999", register_channel="sms").id
+        for i in range(3):
+            db.add(AdRewardRecord(
+                trans_id=f"adaudit-ok-{i}", user_id=uid, coin=0, status="capped",
+                reward_scene="reward_video", reward_date=d,
+            ))
+        for i in range(2):
+            db.add(AdRewardRecord(
+                trans_id=f"adaudit-bad-{i}", user_id=uid, coin=7, status="capped",
+                reward_scene="reward_video", reward_date=d,
+            ))
+        db.commit()
+    finally:
+        db.close()
+
+    base = {"date": d, "user_id": uid}
+
+    # 全量:total=5、mismatch=2、不截断、返回 5 条
+    body = admin_client.get(
+        "/admin/api/ad-coin-audit", params={**base, "limit": 100}, headers=_auth(admin_token)
+    ).json()
+    assert body["total"] == 5
+    assert body["mismatch_count"] == 2
+    assert body["truncated"] is False
+    assert len(body["items"]) == 5
+
+    # 截断:limit=2 → 统计仍全量、truncated=True、只回 2 条
+    body = admin_client.get(
+        "/admin/api/ad-coin-audit", params={**base, "limit": 2}, headers=_auth(admin_token)
+    ).json()
+    assert body["total"] == 5 and body["mismatch_count"] == 2
+    assert body["truncated"] is True
+    assert len(body["items"]) == 2
+
+    # only_mismatch:只回 ✗ 行(2 条),统计仍全量、不截断
+    body = admin_client.get(
+        "/admin/api/ad-coin-audit",
+        params={**base, "limit": 100, "only_mismatch": True},
+        headers=_auth(admin_token),
+    ).json()
+    assert body["total"] == 5 and body["mismatch_count"] == 2
+    assert body["truncated"] is False
+    assert len(body["items"]) == 2
+    assert all(it["matched"] is False for it in body["items"])
+
+
 def test_read_apis_require_auth(admin_client: TestClient) -> None:
     """所有 M2 读接口未带 token → 401(router 级 get_current_admin 守卫)。"""
     for path in [

tests/test_admin_write.py

@@ -136,6 +136,102 @@ def test_grant_zero_rejected(admin_client: TestClient, finance_token: str) -> No
     assert r.status_code == 400
 
 
+def test_set_coins_writes_delta_txn(admin_client: TestClient, finance_token: str) -> None:
+    """set 模式:先有余额,再设为目标值,只写一笔差值流水,审计带 mode/target/before。"""
+    uid = _seed_user("13900000009")
+    # 先增到 300
+    admin_client.post(
+        f"/admin/api/users/{uid}/coins", json={"amount": 300, "reason": "底"},
+        headers=_auth(finance_token),
+    )
+    # 设为 100 → 差值 -200(扣减)
+    r = admin_client.post(
+        f"/admin/api/users/{uid}/coins", json={"mode": "set", "amount": 100, "reason": "设值"},
+        headers=_auth(finance_token),
+    )
+    assert r.status_code == 200, r.text
+    db = SessionLocal()
+    try:
+        assert db.get(CoinAccount, uid).coin_balance == 100
+        txns = db.execute(
+            select(CoinTransaction).where(CoinTransaction.user_id == uid)
+            .order_by(CoinTransaction.id.desc())
+        ).scalars().all()
+        # 两笔:+300(admin_grant)、-200(admin_deduct)
+        assert txns[0].amount == -200 and txns[0].biz_type == "admin_deduct"
+        logs = db.execute(
+            select(AdminAuditLog).where(
+                AdminAuditLog.action == "user.coins.grant", AdminAuditLog.target_id == str(uid)
+            ).order_by(AdminAuditLog.id.desc())
+        ).scalars().all()
+        assert logs[0].detail["mode"] == "set"
+        assert logs[0].detail["target"] == 100
+        assert logs[0].detail["before"] == 300
+        assert logs[0].detail["amount"] == -200
+    finally:
+        db.close()
+
+
+def test_set_coins_equal_balance_rejected(admin_client: TestClient, finance_token: str) -> None:
+    """set 为当前余额(差值 0)→ 拒绝,不写流水。"""
+    uid = _seed_user("13900000010")
+    admin_client.post(
+        f"/admin/api/users/{uid}/coins", json={"amount": 50, "reason": "底"},
+        headers=_auth(finance_token),
+    )
+    r = admin_client.post(
+        f"/admin/api/users/{uid}/coins", json={"mode": "set", "amount": 50, "reason": "x"},
+        headers=_auth(finance_token),
+    )
+    assert r.status_code == 400
+    db = SessionLocal()
+    try:
+        txns = db.execute(
+            select(CoinTransaction).where(CoinTransaction.user_id == uid)
+        ).scalars().all()
+        assert len(txns) == 1  # 仅初始那笔,设值被拒后无新流水
+    finally:
+        db.close()
+
+
+def test_set_coins_negative_target_rejected(admin_client: TestClient, finance_token: str) -> None:
+    uid = _seed_user("13900000011")
+    r = admin_client.post(
+        f"/admin/api/users/{uid}/coins", json={"mode": "set", "amount": -1, "reason": "x"},
+        headers=_auth(finance_token),
+    )
+    assert r.status_code == 400
+
+
+def test_set_cash_writes_delta_txn(admin_client: TestClient, finance_token: str) -> None:
+    """现金 set 模式:设为目标分值,写一笔差值流水。"""
+    uid = _seed_user("13900000012")
+    admin_client.post(
+        f"/admin/api/users/{uid}/cash", json={"amount_cents": 500, "reason": "底"},
+        headers=_auth(finance_token),
+    )
+    r = admin_client.post(
+        f"/admin/api/users/{uid}/cash",
+        json={"mode": "set", "amount_cents": 200, "reason": "设值"},
+        headers=_auth(finance_token),
+    )
+    assert r.status_code == 200, r.text
+    db = SessionLocal()
+    try:
+        assert db.get(CoinAccount, uid).cash_balance_cents == 200
+        logs = db.execute(
+            select(AdminAuditLog).where(
+                AdminAuditLog.action == "user.cash.grant", AdminAuditLog.target_id == str(uid)
+            ).order_by(AdminAuditLog.id.desc())
+        ).scalars().all()
+        assert logs[0].detail["mode"] == "set"
+        assert logs[0].detail["target_cents"] == 200
+        assert logs[0].detail["before_cents"] == 500
+        assert logs[0].detail["amount_cents"] == -300
+    finally:
+        db.close()
+
+
 # ===== 封号 =====
 
 def test_set_user_status_and_audit(admin_client: TestClient, operator_token: str) -> None:

tests/test_store_mapping_invalidate.py

@@ -0,0 +1,118 @@
+"""淘宝 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):
+    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,
+    )
+    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_non_taobao_noop(client, monkeypatch):
+    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", _SECRET)
+    r = client.post("/internal/store-mapping/invalidate",
+                    json={"platform": "jd", "shop_id": "X"},
+                    headers={"X-Internal-Secret": _SECRET})
+    assert r.status_code == 200 and r.json()["affected"] == 0