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>

.env.example

@@ -11,7 +11,7 @@ DATABASE_URL=sqlite:///./data/app.db
 
 # ===== JWT =====
 # 生产部署务必改成随机长字符串,可用:python -c "import secrets; print(secrets.token_urlsafe(64))"
-JWT_SECRET_KEY=change-me-in-prod-please-use-a-long-random-string
+JWT_SECRET_KEY=
 JWT_ALGORITHM=HS256
 # access token 有效期(分钟),默认 2 小时
 JWT_ACCESS_TOKEN_EXPIRE_MINUTES=120
@@ -27,18 +27,6 @@ JG_PRIVATE_KEY_PATH=./secrets/jverify_rsa_private.pem
 JG_VERIFY_ENDPOINT=https://api.verification.jpush.cn/v1/web/loginTokenVerify
 JG_REQUEST_TIMEOUT_SEC=15
 
-# ===== 极光推送 JPush(无障碍保护掉线告警)=====
-# 客户端 push appkey 已知 = 966b451a8d9cfe12d173ea9d。若推送与一键登录/短信是同一个极光应用,
-# JPUSH_* 留空即自动回退到上面的 JG_APP_KEY/JG_MASTER_SECRET;否则单独填那个 push 应用的密钥。
-# 运维清单(厂商通道等)见 spec/accessibility-liveness-push.md §7。
-JPUSH_APP_KEY=
-JPUSH_MASTER_SECRET=
-JPUSH_PUSH_ENDPOINT=https://api.jpush.cn/v3/push
-# 无障碍保护存活监控后台任务
-HEARTBEAT_MONITOR_ENABLED=true
-HEARTBEAT_TIMEOUT_MINUTES=10
-HEARTBEAT_SCAN_INTERVAL_SEC=60
-
 # ===== 短信 (mock 模式) =====
 # mock = true 时,任意 6 位数字均通过,且 /sms/send 不真发短信(只 log)。
 # 后续接阿里云/腾讯云短信时,改成 false 并填供应商相关 key。

alembic/versions/3a9941e76909_merge_coupon_engage_app_pkg_and_coin_.py

@@ -1,26 +0,0 @@
-"""merge coupon_engage_app_pkg and coin_txn_task_ref_uq heads
-
-Revision ID: 3a9941e76909
-Revises: coin_txn_task_ref_uq, coupon_engage_app_pkg
-Create Date: 2026-06-13 10:09:37.557466
-
-"""
-from typing import Sequence, Union
-
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision: str = '3a9941e76909'
-down_revision: Union[str, Sequence[str], None] = ('coin_txn_task_ref_uq', 'coupon_engage_app_pkg')
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    pass
-
-
-def downgrade() -> None:
-    pass

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/coupon_engagement_app_package.py

@@ -1,47 +0,0 @@
-"""coupon_prompt_engagement 加 app_package(弹窗频控改按 App 为单位)
-
-2026-06-12 产品确认:领券引导窗「一个 app 一天只能弹一次」,没领完进其他平台要再弹,
-彻底领完(coupon_daily_completion)才全局不弹。频控行从 (device, 日) 唯一改为
-(device, app_package, 日) 唯一;旧数据 app_package 回填 NULL(= 全局兜底行,
-按 App 查询时忽略,不带包名的旧式查询仍生效)。
-
-⚠️ downgrade 有损:同设备同日多 App 各一行时,重建 (device, 日) 唯一约束会撞;
-仅开发/测试库可降级(频控行本就是当日临时数据,coupon_claim_record 资产不受影响)。
-
-Revision ID: coupon_engage_app_pkg
-Revises: 9b894f5fff05
-Create Date: 2026-06-12 00:00:00.000000
-
-"""
-
-from collections.abc import Sequence
-
-import sqlalchemy as sa
-
-from alembic import op
-
-# revision identifiers, used by Alembic.
-revision: str = "coupon_engage_app_pkg"
-down_revision: str | Sequence[str] | None = "9b894f5fff05"
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-
-def upgrade() -> None:
-    # batch 模式兼容 SQLite(本地)与 PG(线上):SQLite 改约束要重建表,batch 自动处理。
-    with op.batch_alter_table("coupon_prompt_engagement", schema=None) as batch_op:
-        batch_op.add_column(sa.Column("app_package", sa.String(length=64), nullable=True))
-        batch_op.drop_constraint("uq_coupon_engage_device_date", type_="unique")
-        batch_op.create_unique_constraint(
-            "uq_coupon_engage_device_app_date",
-            ["device_id", "app_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_app_date", type_="unique")
-        batch_op.create_unique_constraint(
-            "uq_coupon_engage_device_date", ["device_id", "engage_date"]
-        )
-        batch_op.drop_column("app_package")

alembic/versions/device_table.py

@@ -1,53 +0,0 @@
-"""device table (无障碍保护存活检测 + 极光推送)
-
-Revision ID: device_liveness_table
-Revises: f3d0a16bb4c2
-Create Date: 2026-06-15 12:00:00.000000
-
-"""
-from typing import Sequence, Union
-
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision: str = 'device_liveness_table'
-down_revision: Union[str, Sequence[str], None] = 'f3d0a16bb4c2'
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.create_table(
-        'device',
-        sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
-        sa.Column('user_id', sa.Integer(), nullable=False),
-        sa.Column('device_id', sa.String(length=128), nullable=False),
-        sa.Column('registration_id', sa.String(length=64), nullable=True),
-        sa.Column('platform', sa.String(length=16), nullable=False),
-        sa.Column('app_version', sa.String(length=32), nullable=True),
-        sa.Column('ever_protected', sa.Boolean(), nullable=False),
-        sa.Column('last_heartbeat_at', sa.DateTime(timezone=True), nullable=True),
-        sa.Column('last_report_protection_on', sa.Boolean(), nullable=False),
-        sa.Column('liveness_state', sa.String(length=16), nullable=False),
-        sa.Column('notified_at', sa.DateTime(timezone=True), nullable=True),
-        sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
-        sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
-        sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
-        sa.PrimaryKeyConstraint('id'),
-        sa.UniqueConstraint('user_id', 'device_id', name='uq_device_user_device'),
-    )
-    with op.batch_alter_table('device', schema=None) as batch_op:
-        batch_op.create_index(batch_op.f('ix_device_user_id'), ['user_id'], unique=False)
-        batch_op.create_index(batch_op.f('ix_device_device_id'), ['device_id'], unique=False)
-        batch_op.create_index(batch_op.f('ix_device_last_heartbeat_at'), ['last_heartbeat_at'], unique=False)
-
-
-def downgrade() -> None:
-    with op.batch_alter_table('device', schema=None) as batch_op:
-        batch_op.drop_index(batch_op.f('ix_device_last_heartbeat_at'))
-        batch_op.drop_index(batch_op.f('ix_device_device_id'))
-        batch_op.drop_index(batch_op.f('ix_device_user_id'))
-
-    op.drop_table('device')

alembic/versions/f3d0a16bb4c2_merge_store_mapping_cols_branch_into_.py

@@ -1,26 +0,0 @@
-"""merge store_mapping cols branch into coupon/coin mergepoint
-
-Revision ID: f3d0a16bb4c2
-Revises: 3a9941e76909, store_mapping_jd_cols
-Create Date: 2026-06-14 14:31:25.504510
-
-"""
-from typing import Sequence, Union
-
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision: str = 'f3d0a16bb4c2'
-down_revision: Union[str, Sequence[str], None] = ('3a9941e76909', 'store_mapping_jd_cols')
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    pass
-
-
-def downgrade() -> None:
-    pass

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

@@ -50,7 +50,7 @@ def _reward_video_rows(
             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)
@@ -127,7 +127,7 @@ def _feed_rows(db: Session, *, date: str, user_id: int | None) -> list[dict]:
     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)
@@ -205,7 +205,7 @@ def ad_coin_audit(
         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)
 
     total = len(rows)
     mismatch_count = sum(1 for r in rows if not r["matched"])

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]:
-    """游标分页(id 倒序),与现有 list_* 约定一致。返回 (rows, next_cursor)。"""
+) -> tuple[list[AdminAuditLog], int | None, int]:
+    """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)
-    stmt = stmt.order_by(AdminAuditLog.id.desc())
-    rows = list(db.execute(stmt.limit(limit + 1)).scalars().all())
+
+    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(AdminAuditLog.id.desc()).offset(offset).limit(limit + 1)
+        ).scalars().all()
+    )
     has_more = len(rows) > limit
     items = rows[:limit]
-    # next_cursor 必须是"本页返回的最后一条"的 id(下一页查 id < 它),不能用 rows[limit]——
-    # rows[limit] 是探测下一页用的第 limit+1 条,它既不在本页也不在下页 → 每页边界丢一条。
-    next_cursor = items[-1].id if has_more else None
-    return items, next_cursor
+    next_cursor = offset + limit if has_more else None
+    return items, next_cursor, total

app/admin/repositories/queries.py

@@ -38,6 +38,28 @@ 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,
     *,
@@ -53,11 +75,11 @@ def list_users(
     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])。
-    日期入参统一转 UTC naive 比较(User 时间均为 UTC naive,见 _as_utc_naive)。"""
+    日期入参统一转 tz-aware UTC 比较(列为 timestamptz,见 _as_utc)。"""
     stmt = select(User)
     if phone:
         stmt = stmt.where(User.phone.like(f"{phone}%"))  # 前缀匹配
@@ -68,13 +90,13 @@ def list_users(
     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_naive(created_from))
+        stmt = stmt.where(User.created_at >= _as_utc(created_from))
     if created_to is not None:
-        stmt = stmt.where(User.created_at <= _as_utc_naive(created_to))
+        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_naive(last_login_from))
+        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_naive(last_login_to))
+        stmt = stmt.where(User.last_login_at <= _as_utc(last_login_to))
 
     sort_cols = {
         "id": User.id,
@@ -84,14 +106,7 @@ def list_users(
     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)
-    stmt = stmt.order_by(order_fn(sort_col), id_order)
-
-    offset = max(cursor or 0, 0)
-    rows = list(db.execute(stmt.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
+    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]:
@@ -160,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:
@@ -190,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(
@@ -244,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)
@@ -254,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(
@@ -463,14 +502,14 @@ def list_price_reports(
     user_id: int | None = None,
     limit: int = 20,
     cursor: int | None = None,
-) -> tuple[list[PriceReport], int | None]:
-    """上报更低价列表(admin 全量,可按状态/用户筛)。游标同 feedback:id 倒序。"""
+) -> tuple[list[PriceReport], int | None, int]:
+    """上报更低价列表(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/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:
+    if body.status is not None and body.status != target.status:
+        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

@@ -45,7 +45,7 @@ def list_users(
     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,
         nickname=nickname, created_from=created_from, created_to=created_to,
         last_login_from=last_login_from, last_login_to=last_login_to,
@@ -54,6 +54,7 @@ def list_users(
     return CursorPage(
         items=[AdminUserListItem.model_validate(u) for u in items],
         next_cursor=next_cursor,
+        total=total,
     )
 
 
@@ -126,7 +127,8 @@ def grant_user_coins(
     if body.mode == "set":
         if body.amount < 0:
             raise HTTPException(status_code=400, detail="目标金币值不能为负")
-        before = wallet_repo.get_or_create_account(db, user_id, commit=False).coin_balance
+        # lock=True:锁账户行,防连点/并发各读同一 before 算同一 delta 双写,余额错位
+        before = wallet_repo.get_or_create_account(db, user_id, commit=False, lock=True).coin_balance
         delta = body.amount - before
         if delta == 0:
             raise HTTPException(status_code=400, detail=f"当前金币已为 {body.amount},无需调整")
@@ -134,9 +136,9 @@ def grant_user_coins(
         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)
+            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})"
@@ -174,7 +176,10 @@ def grant_user_cash(
     if body.mode == "set":
         if body.amount_cents < 0:
             raise HTTPException(status_code=400, detail="目标现金值不能为负")
-        before = wallet_repo.get_or_create_account(db, user_id, commit=False).cash_balance_cents
+        # 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(
@@ -184,9 +189,9 @@ def grant_user_cash(
         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)
+            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} 分)"

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/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,6 +37,13 @@ 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):
     mode: Literal["delta", "set"] = Field(
         "delta", description="delta=增减(amount 为变动量) / set=设为(amount 为目标值,须≥0)"
@@ -47,6 +54,8 @@ class GrantCoinsRequest(BaseModel):
     )
     reason: str = Field(..., min_length=1, max_length=128, description="操作原因(必填,入审计)")
 
+    _v_reason = field_validator("reason")(_strip_reason)
+
 
 class GrantCashRequest(BaseModel):
     mode: Literal["delta", "set"] = Field(
@@ -58,6 +67,8 @@ class GrantCashRequest(BaseModel):
     )
     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

@@ -17,7 +17,12 @@ from fastapi import APIRouter, Header
 from app.api.deps import DbSession
 from app.api.internal.price import _check_secret
 from app.repositories import store_mapping as repo
-from app.schemas.store_mapping import StoreMappingIn, StoreMappingOut
+from app.schemas.store_mapping import (
+    StoreMappingIn,
+    StoreMappingInvalidateIn,
+    StoreMappingInvalidateOut,
+    StoreMappingOut,
+)
 
 logger = logging.getLogger("shagua.internal.store")
 
@@ -77,3 +82,26 @@ def report_store_mapping(
         payload.source_device_id, payload.source_user_id,
     )
     return StoreMappingOut(inserted=created, row_id=row_id)
+
+
+@router.post(
+    "/store-mapping/invalidate",
+    response_model=StoreMappingInvalidateOut,
+    summary="标记某平台 shopId 的缓存 deeplink 失效(pricebot 撞错误页回退时上报,lookup 不再返回)",
+)
+def invalidate_store_mapping(
+    payload: StoreMappingInvalidateIn,
+    db: DbSession,
+    x_internal_secret: Annotated[str | None, Header()] = None,
+) -> StoreMappingInvalidateOut:
+    _check_secret(x_internal_secret)
+    if payload.platform != "taobao":
+        # 当前只接淘宝; 其它平台先 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

@@ -67,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(
@@ -112,16 +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)。
-    # 透传链路拿不到发起 App 的包名 → 写 app_package=NULL 的全局兜底行:只对旧式不带
-    # package 的 should-show 查询生效;按 App 频控(2026-06-12)由客户端 /prompt/shown
-    # 负责,本行不堵其他 App 的弹窗(领一半终止,其他平台还要弹)。写库失败绝不能
+    # 领券任务首帧(step=0)= 用户已发起领券 → 记一条今日 engagement(claim_started),
+    # 今天**这个 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)
@@ -191,30 +192,28 @@ async def coupon_step(
     return resp_json
 
 
-@router.post("/prompt/shown", summary="领券引导窗已对某 App 弹出(按 App 频控主判据)")
+@router.post("/prompt/shown", summary="领券引导窗弹出即上报(按 App 记 shown)")
 def coupon_prompt_shown(payload: CouponPromptShownIn, db: DbSession) -> dict[str, bool]:
-    """客户端弹窗一亮即调用 → 记一条今日该 App 的 engagement(shown)。
+    """客户端弹出引导窗那刻调 → 记一条今日 engagement(shown),今天**这个 App** 不再自动弹。
 
-    2026-06-12 频控按 App 为单位(方案文档「一个 app 一天只能弹一次,时机=每天第一次
-    进入」):以"弹出"为频控锚点,用户领/拒/无视都只算这一次;dismiss/claim_started
-    后续只升级同一行的 engage_type。MVP 不鉴权,按 device_id 记。
+    频控主判据(管跨重装):弹出即占用今天这个 App 的"一次"。用户领/拒/无视都算用掉。
+    后续点领取/拒绝再由 step/dismiss 把 type 升级。按 (device, package, 日) 记。
     """
     coupon_repo.mark_engagement(
-        db, payload.device_id, payload.user_id, "shown", payload.package
+        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),今天该 App 不再弹。
+    """客户端点关闭引导窗时调用 → 记一条今日 engagement(dismissed),今天**这个 App** 不再弹。
 
     server 在透传链路里看不到"用户拒绝"(拒绝不发起领券),故必须客户端通知。
-    频控以 /prompt/shown 为主判据,本端点把同一行升级成 dismissed(记录用);
-    旧客户端不带 package → 写 NULL 全局兜底行(旧语义)。MVP 不鉴权,按 device_id 记。
+    频控按 (device, package, 日),各 App 独立。MVP 不鉴权,按 device_id 记。
     """
     coupon_repo.mark_engagement(
-        db, payload.device_id, payload.user_id, "dismissed", payload.package
+        db, payload.device_id, payload.package, payload.user_id, "dismissed"
     )
     return {"ok": True}
 
@@ -222,35 +221,23 @@ def coupon_prompt_dismiss(payload: CouponPromptDismissIn, db: DbSession) -> dict
 @router.get(
     "/prompt/should-show",
     response_model=CouponPromptShouldShowOut,
-    summary="切到某外卖 App 时是否还应弹领券引导窗(按 App 频控)",
+    summary="切到外卖 App 时是否还应弹领券引导窗",
 )
 def coupon_prompt_should_show(
-    device_id: str, db: DbSession, package: str | None = None
+    device_id: str, db: DbSession, package: str = ""
 ) -> CouponPromptShouldShowOut:
-    """should_show = 今天没跑完整轮领券 AND 该 App 今天没弹过。
-
-    2026-06-12 按 App 为单位(产品确认):
-    - 该 App 今天弹过(shown/dismissed/claim_started 任一)→ false;
-    - **其他 App** 弹过/领了一半不影响本 App → 仍 true(没领完进其他平台要再弹);
-    - 今天已跑完整轮(coupon_daily_completion,"彻底领完")→ 全局 false。
-    package 不传 = 旧客户端,退回旧全局语义(任意一行算 engage)+ 同样吃 completion 闸。
-    客户端据此决定弹不弹(纯后台判据,客户端不做前台 SP 缓存判断)。"""
-    if coupon_repo.has_completed_today(db, device_id):
-        return CouponPromptShouldShowOut(should_show=False)
+    """今天这台设备**这个 App** 已 engage(弹/领/拒)过 → should_show=false。各 App 独立:
+    美团弹过不压淘宝/京东。客户端切到目标 App 时带 package 查(老客户端不带 → "" 全局态)。"""
     return CouponPromptShouldShowOut(
         should_show=not coupon_repo.has_engaged_today(db, device_id, package)
     )
 
 
-@router.post("/prompt/reset", summary="重置今日领券状态:弹窗 engagement + 完成记录(开发测试用)")
+@router.post("/prompt/reset", summary="重置今日领券引导窗 engagement(开发测频控用)")
 def coupon_prompt_reset(payload: CouponPromptDismissIn, db: DbSession) -> dict[str, bool]:
-    """删这台设备今天的 engagement + coupon_daily_completion:弹窗又能弹,且首页
-    「去领取」卡/弹窗「一键自动领取」CTA 恢复可点(只删 engagement 的话, 重置后 CTA
-    仍被"今日已跑完整轮"置灰, 2026-06-12)。配合客户端本地重置 = 等效重装。
-    领券记录(coupon_claim_record)是资产沉淀、不参与任何门控判断, 不删。
+    """删这台设备今天的 engagement → has_engaged_today 变 false,今天又能弹。
     开发设置「重置今日领券弹窗状态」按钮调。MVP 不鉴权,按 device_id。"""
     coupon_repo.reset_today_engagement(db, payload.device_id)
-    coupon_repo.reset_today_completion(db, payload.device_id)
     return {"ok": True}
 
 
@@ -268,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/device.py

@@ -1,66 +0,0 @@
-"""设备注册 / 心跳 endpoint(无障碍保护存活检测)。
-
-路由前缀 /api/v1/device,需 Bearer 鉴权(设备绑登录用户)。
-  POST /register   注册设备 / 更新 registration_id(App 前台、拿到 push token 时调)
-  POST /heartbeat  上报心跳(无障碍服务存活时周期调,刷新存活)
-
-后端 heartbeat_monitor_worker 据此发现心跳超时的设备并极光推送告警。
-见 spec: spec/accessibility-liveness-push.md。
-"""
-from __future__ import annotations
-
-import logging
-
-from fastapi import APIRouter
-
-from app.api.deps import CurrentUser, DbSession
-from app.repositories import device as device_repo
-from app.schemas.device import (
-    DeviceOut,
-    DeviceRegisterRequest,
-    HeartbeatRequest,
-    OkResponse,
-)
-
-logger = logging.getLogger("shagua.device")
-
-router = APIRouter(prefix="/api/v1/device", tags=["device"])
-
-
-@router.post("/register", response_model=DeviceOut, summary="注册设备/更新推送token")
-def register_device(
-    req: DeviceRegisterRequest,
-    user: CurrentUser,
-    db: DbSession,
-) -> DeviceOut:
-    device = device_repo.register_or_update(
-        db,
-        user_id=user.id,
-        device_id=req.device_id,
-        registration_id=req.registration_id,
-        platform=req.platform,
-        app_version=req.app_version,
-    )
-    logger.info(
-        "device register user_id=%d device_id=%s reg=%s",
-        user.id,
-        req.device_id,
-        bool(req.registration_id),
-    )
-    return DeviceOut.model_validate(device)
-
-
-@router.post("/heartbeat", response_model=OkResponse, summary="上报心跳")
-def report_heartbeat(
-    req: HeartbeatRequest,
-    user: CurrentUser,
-    db: DbSession,
-) -> OkResponse:
-    device_repo.touch_heartbeat(
-        db,
-        user_id=user.id,
-        device_id=req.device_id,
-        accessibility_enabled=req.accessibility_enabled,
-        registration_id=req.registration_id,
-    )
-    return OkResponse()

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(
@@ -61,32 +66,6 @@ class Settings(BaseSettings):
     JG_VERIFY_ENDPOINT: str = "https://api.verification.jpush.cn/v1/web/loginTokenVerify"
     JG_REQUEST_TIMEOUT_SEC: int = 15
 
-    # ===== 极光推送 JPush(无障碍保护掉线告警)=====
-    # 客户端 push appkey 已知 = 966b451a8d9cfe12d173ea9d(android build.gradle manifestPlaceholder)。
-    # 若推送与一键登录/短信是同一个极光应用(大概率),JPUSH_* 留空即自动回退到 JG_*。
-    # 否则在 .env 单独配 JPUSH_APP_KEY / JPUSH_MASTER_SECRET(对应那个 push appkey)。
-    JPUSH_APP_KEY: str = ""
-    JPUSH_MASTER_SECRET: str = ""
-    JPUSH_PUSH_ENDPOINT: str = "https://api.jpush.cn/v3/push"
-
-    # 无障碍保护存活监控后台任务
-    HEARTBEAT_MONITOR_ENABLED: bool = True       # 总开关
-    HEARTBEAT_TIMEOUT_MINUTES: int = 10          # 多久没心跳算掉线(≈3 个客户端心跳周期)
-    HEARTBEAT_SCAN_INTERVAL_SEC: int = 60        # 扫描周期
-
-    @property
-    def jpush_app_key(self) -> str:
-        return self.JPUSH_APP_KEY or self.JG_APP_KEY
-
-    @property
-    def jpush_master_secret(self) -> str:
-        return self.JPUSH_MASTER_SECRET or self.JG_MASTER_SECRET
-
-    @property
-    def jpush_configured(self) -> bool:
-        """推送凭证齐全(缺则 monitor 只扫不发,不报错)。"""
-        return bool(self.jpush_app_key and self.jpush_master_secret)
-
     # ===== 短信 =====
     SMS_MOCK: bool = True
     SMS_CODE_TTL_SEC: int = 300
@@ -224,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/heartbeat_monitor_worker.py

@@ -1,148 +0,0 @@
-"""无障碍保护存活监控后台任务。
-
-周期扫描「曾经保护过、当前 alive、心跳超时」的设备 = App 被彻底杀掉/无障碍已停(心跳断了),
-**命中即在服务器终端打印告警**(本期先不接推送,工程量大,用终端打印代替真实通知);并把状态机
-推进到 notified 防每轮重复打印(心跳恢复时由 repositories.device.touch_heartbeat 重置回 alive)。
-结构仿 withdraw_reconcile_worker(单实例锁 + asyncio 轮询 + 优雅退出)。
-
-见 spec: spec/accessibility-liveness-push.md。
-"""
-from __future__ import annotations
-
-import asyncio
-import contextlib
-import logging
-import os
-import time
-from collections.abc import Iterator
-from datetime import datetime, timezone
-from pathlib import Path
-
-from sqlalchemy.exc import SQLAlchemyError
-
-from app.core.config import settings
-from app.db.session import SessionLocal
-from app.repositories import device as device_repo
-
-logger = logging.getLogger("shagua.heartbeat_monitor")
-_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "heartbeat_monitor.lock"
-
-
-def _touch_lock() -> None:
-    with contextlib.suppress(FileNotFoundError):
-        os.utime(_LOCK_PATH, None)
-
-
-@contextlib.contextmanager
-def _single_instance_lock(stale_after_sec: int) -> Iterator[bool]:
-    """同机多进程保护:同一时间只允许一个监控 worker 运行。"""
-    _LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
-    fd: int | None = None
-    try:
-        try:
-            fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
-        except FileExistsError:
-            try:
-                age = time.time() - _LOCK_PATH.stat().st_mtime
-            except FileNotFoundError:
-                age = stale_after_sec + 1
-            if age > stale_after_sec:
-                with contextlib.suppress(FileNotFoundError):
-                    _LOCK_PATH.unlink()
-                try:
-                    fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
-                except FileExistsError:
-                    fd = None
-
-        if fd is None:
-            yield False
-            return
-
-        os.write(fd, f"pid={os.getpid()} started_at={int(time.time())}\n".encode("ascii"))
-        yield True
-    finally:
-        if fd is not None:
-            os.close(fd)
-            with contextlib.suppress(FileNotFoundError):
-                _LOCK_PATH.unlink()
-
-
-def _silent_seconds(last: datetime | None) -> int | None:
-    """距上次心跳的秒数(兼容 sqlite 取回的 naive datetime)。"""
-    if last is None:
-        return None
-    ref = datetime.now(timezone.utc) if last.tzinfo is not None else datetime.utcnow()
-    return int((ref - last).total_seconds())
-
-
-def _scan_once(timeout_minutes: int) -> dict:
-    """扫描一轮:找出心跳超时(App 被彻底杀掉/无障碍已停)的设备,在**服务器终端打印**告警代替真实推送。
-
-    本期不接推送(极光/厂商通道工程量大),只做服务端掉线检测:命中即 logger.warning 打印到终端,
-    并把状态机推进到 notified 防每轮重复打印(心跳恢复时 touch_heartbeat 会重置回 alive)。
-    """
-    notified = 0
-    with SessionLocal() as db:
-        overdue = device_repo.list_overdue(db, timeout_minutes=timeout_minutes)
-        for device in overdue:
-            silent = _silent_seconds(device.last_heartbeat_at)
-            logger.warning(
-                "🔴 [掉线检测] user_id=%s device_id=%s 已 %s 秒无心跳(阈值 %d 分钟)"
-                " → 判定 App 已被杀/无障碍已停。【本应推送通知提醒用户重开;推送暂未接,先终端打印代替】",
-                device.user_id,
-                device.device_id,
-                silent if silent is not None else "?",
-                timeout_minutes,
-            )
-            device_repo.mark_notified(db, device_id_pk=device.id)
-            notified += 1
-        return {"checked": len(overdue), "notified": notified}
-
-
-async def _run_loop() -> None:
-    interval = max(10, int(settings.HEARTBEAT_SCAN_INTERVAL_SEC))
-    timeout_minutes = max(1, int(settings.HEARTBEAT_TIMEOUT_MINUTES))
-    lock_stale_after = max(interval * 3, 600)
-    with _single_instance_lock(lock_stale_after) as lock_acquired:
-        if not lock_acquired:
-            logger.warning("heartbeat monitor skipped: another worker owns lock")
-            return
-        await _run_locked_loop(interval, timeout_minutes)
-
-
-async def _run_locked_loop(interval: int, timeout_minutes: int) -> None:
-    logger.info(
-        "heartbeat monitor started interval=%ss timeout=%sm",
-        interval,
-        timeout_minutes,
-    )
-    try:
-        while True:
-            try:
-                _touch_lock()
-                result = await asyncio.to_thread(_scan_once, timeout_minutes)
-                if result["notified"]:
-                    logger.info("heartbeat monitor result=%s", result)
-            except SQLAlchemyError:
-                logger.exception("heartbeat monitor db error")
-            except Exception:  # noqa: BLE001 - 后台任务不能因单次异常退出
-                logger.exception("heartbeat monitor unexpected error")
-            await asyncio.sleep(interval)
-    except asyncio.CancelledError:
-        logger.info("heartbeat monitor stopped")
-        raise
-
-
-def start_heartbeat_monitor() -> asyncio.Task | None:
-    if not settings.HEARTBEAT_MONITOR_ENABLED:
-        logger.info("heartbeat monitor disabled")
-        return None
-    return asyncio.create_task(_run_loop(), name="heartbeat-monitor")
-
-
-async def stop_heartbeat_monitor(task: asyncio.Task | None) -> None:
-    if task is None:
-        return
-    task.cancel()
-    with contextlib.suppress(asyncio.CancelledError):
-        await task

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。"""
@@ -187,8 +194,12 @@ def calculate_ad_reward_coin(ecpm: str | int | float | None, count_after_this: i
     eCPM 是穿山甲 getEcpm 原值,单位【分/千次展示】;先 ÷100 转成元(因子判档 + 收益换算都用元)。
     单次收益(元)= eCPM元 ÷ 1000(每千次→单次) × 因子1(eCPM 元档) × 因子2(LT);
     再按 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(count_after_this)
     return max(0, round(yuan * COIN_PER_YUAN))
 

app/integrations/jpush.py

@@ -1,94 +0,0 @@
-"""极光推送 JPush(无障碍保护掉线告警)。
-
-调极光 Push REST v3 /v3/push 给指定 registration_id 推一条通知。鉴权复用极光 Basic Auth
-(appKey:masterSecret),与一键登录/短信同模式(见 integrations/jiguang.py、sms.py)。
-
-凭证:settings.jpush_app_key / jpush_master_secret(JPUSH_* 留空时自动回退到 JG_*,
-若推送与一键登录是同一极光应用)。客户端 push appkey 已知 = 966b451a8d9cfe12d173ea9d。
-厂商通道(App 被杀也能到达)由极光后台 + 客户端插件负责,本服务只管调 push 接口。
-"""
-from __future__ import annotations
-
-import base64
-import logging
-
-import httpx
-
-from app.core.config import settings
-
-logger = logging.getLogger("shagua.jpush")
-
-
-class JPushError(Exception):
-    """极光推送调用失败。"""
-
-
-class JPushNotConfiguredError(JPushError):
-    """缺 appKey / masterSecret。"""
-
-
-def push_to_registration_ids(
-    registration_ids: list[str],
-    *,
-    title: str,
-    alert: str,
-    extras: dict | None = None,
-) -> dict:
-    """给一批 registration_id 推送通知 + 透传消息。失败抛 JPushError。
-
-    Returns: 极光响应 JSON(含 sendno / msg_id)。
-    """
-    if not settings.jpush_configured:
-        raise JPushNotConfiguredError(
-            "JPush 未配置(缺 JPUSH_APP_KEY/JPUSH_MASTER_SECRET,且 JG_* 也为空)"
-        )
-    reg_ids = [r for r in registration_ids if r]
-    if not reg_ids:
-        raise JPushError("registration_ids 为空")
-
-    auth_b64 = base64.b64encode(
-        f"{settings.jpush_app_key}:{settings.jpush_master_secret}".encode()
-    ).decode()
-    payload = {
-        "platform": ["android"],
-        "audience": {"registration_id": reg_ids},
-        "notification": {
-            "android": {
-                "alert": alert,
-                "title": title,
-                "priority": 1,
-                "extras": extras or {},
-            },
-        },
-        "message": {
-            "msg_content": alert,
-            "title": title,
-            "content_type": "text",
-            "extras": extras or {},
-        },
-        "options": {"time_to_live": 86400, "apns_production": True},
-    }
-
-    try:
-        resp = httpx.post(
-            settings.JPUSH_PUSH_ENDPOINT,
-            json=payload,
-            headers={
-                "Content-Type": "application/json",
-                "Authorization": f"Basic {auth_b64}",
-            },
-            timeout=settings.JG_REQUEST_TIMEOUT_SEC,
-        )
-    except httpx.HTTPError as e:
-        raise JPushError(f"jpush 网络错误: {e}") from e
-
-    if resp.status_code != 200:
-        body = resp.text[:300]
-        logger.error("[JPush] http=%s body=%s", resp.status_code, body)
-        raise JPushError(f"jpush http {resp.status_code}")
-
-    data = resp.json()
-    if not data.get("sendno") and not data.get("msg_id"):
-        logger.error("[JPush] unexpected response: %s", data)
-        raise JPushError(f"jpush 响应异常: {data}")
-    return data

app/main.py

@@ -20,7 +20,6 @@ from app.api.v1.compare import router as compare_router
 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.v1.device import router as device_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
@@ -36,10 +35,6 @@ from app.api.v1.user import router as user_router
 from app.api.v1.wallet import router as wallet_router
 from app.api.v1.wxpay import router as wxpay_router
 from app.core.config import settings
-from app.core.heartbeat_monitor_worker import (
-    start_heartbeat_monitor,
-    stop_heartbeat_monitor,
-)
 from app.core.logging import setup_logging
 from app.core.withdraw_reconcile_worker import (
     start_withdraw_reconcile_worker,
@@ -61,11 +56,9 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
         settings.DATABASE_URL.split("://", 1)[0],
     )
     reconcile_task = start_withdraw_reconcile_worker()
-    heartbeat_task = start_heartbeat_monitor()
     try:
         yield
     finally:
-        await stop_heartbeat_monitor(heartbeat_task)
         await stop_withdraw_reconcile_worker(reconcile_task)
         logger.info("shutting down")
 
@@ -98,7 +91,6 @@ app.include_router(user_router)
 app.include_router(feedback_router)
 app.include_router(invite_router)
 app.include_router(coupon_router)
-app.include_router(device_router)
 app.include_router(compare_router)
 app.include_router(compare_record_router)
 app.include_router(compare_milestone_router)

app/models/__init__.py

@@ -7,7 +7,6 @@ from app.models.admin import AdminAuditLog, AdminUser  # noqa: F401
 from app.models.app_config import AppConfig  # noqa: F401
 from app.models.comparison import ComparisonRecord  # noqa: F401
 from app.models.comparison_milestone import ComparisonMilestoneClaim  # noqa: F401
-from app.models.device import Device  # noqa: F401
 from app.models.coupon_state import (  # noqa: F401
     CouponClaimRecord,
     CouponDailyCompletion,

app/models/coupon_state.py

@@ -137,36 +137,35 @@ class CouponDailyCompletion(Base):
 
 
 class CouponPromptEngagement(Base):
-    """按 (device, App, 自然日) 记"今天该 App 是否弹过/engage 过领券引导窗"——弹窗频控源。
+    """按 (device, **App**, 自然日) 记"今天这个 App 是否对领券引导窗表达过意向"——弹窗频控源。
 
-    2026-06-12 改为**按 App 为单位**(对齐方案文档「一个 app 一天只能弹出弹窗一次」+ 产品确认:
-    没领完就进其他平台要再弹,彻底领完才全局不弹——"彻底领完"由 coupon_daily_completion 负责):
-    - app_package 非空:该 App 当天的频控行(shown/dismissed 由客户端带包名上报)。
-    - app_package 为 NULL:全局兜底行(step=0 透传链路自动记的 claim_started 拿不到包名 +
-      旧版客户端不带包名)。**按 App 查询时忽略 NULL 行**(领券中途终止 ≠ 其他 App engage 过);
-      不带包名的旧式查询仍把 NULL 行算作"今天 engage 过"(旧客户端行为不变)。
+    2026-06-14:频控维度从 (device, 日) 改为 (device, package, 日)。需求是美团/淘宝/京东
+    各自独立——在美团弹过/领过,不影响淘宝、京东今天仍各弹一次。原来缺 package → 任一 App
+    弹过就把整台设备当天标记 engage,其余 App 被压住不弹(bug)。
     """
 
     __tablename__ = "coupon_prompt_engagement"
     __table_args__ = (
-        # 一台设备、一个 App、一天一条(app_package=NULL 的全局兜底行不受唯一约束限制——
-        # SQL 标准里 NULL 互不相等;写入走 select-first upsert,正常不会堆重复行)。
+        # 一台设备、一个 App、一天一条:今天**这个 App** engage 过(领或拒)才不再弹该 App。
         UniqueConstraint(
-            "device_id", "app_package", "engage_date",
-            name="uq_coupon_engage_device_app_date",
+            "device_id", "package", "engage_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)
-    # 在哪个外卖 App 弹的窗(Android 包名,如 com.sankuai.meituan)。NULL = 全局兜底行(见类注释)。
-    app_package: Mapped[str | None] = mapped_column(String(64), nullable=True)
     # Asia/Shanghai 自然日。
     engage_date: Mapped[date] = mapped_column(Date, nullable=False)
-    # shown(弹窗已对该 App 弹出,频控主判据)/ claim_started(点了一键领取)/
-    # dismissed(点了拒绝/关闭)。判断只看"今天该 App 有没有这条",type 仅记录区分。
+    # claim_started(点了一键领取)/ dismissed(点了拒绝/关闭)/ shown(自动弹出即记)。
+    # 仅记录区分,判断只看"今天这个 App 有没有这条",type 不影响弹不弹。
     engage_type: Mapped[str] = mapped_column(String(16), nullable=False)
 
     created_at: Mapped[datetime] = mapped_column(

app/models/device.py

@@ -1,83 +0,0 @@
-"""设备表(无障碍保护存活检测 + 极光推送)。
-
-每条 = 一个用户的一台设备(per-install,device_id 由客户端 DeviceId.get() 生成)。
-客户端的无障碍服务存活时周期上报心跳刷新 last_heartbeat_at;App 前台/登录时上报
-registration_id(极光推送目标)。后端 heartbeat_monitor_worker 扫描「曾经保护过、
-现在心跳超时」的设备,通过极光推送提醒用户重开无障碍。
-
-liveness_state 状态机(防刷屏,一次掉线只推一条):
-  unknown → alive(收到 service 心跳)→ silent/notified(扫描发现超时并已推送)
-  心跳恢复时 handler 重置回 alive。
-见 spec: spec/accessibility-liveness-push.md。
-"""
-from __future__ import annotations
-
-from datetime import datetime
-
-from sqlalchemy import (
-    Boolean,
-    DateTime,
-    ForeignKey,
-    Integer,
-    String,
-    UniqueConstraint,
-    func,
-)
-from sqlalchemy.orm import Mapped, mapped_column
-
-from app.db.base import Base
-
-
-class Device(Base):
-    __tablename__ = "device"
-    __table_args__ = (
-        UniqueConstraint("user_id", "device_id", name="uq_device_user_device"),
-    )
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
-    user_id: Mapped[int] = mapped_column(
-        Integer, ForeignKey("user.id"), index=True, nullable=False
-    )
-    # 客户端 DeviceId.get() 生成的 per-install id(如 device_Pixel_ab12cd34)
-    device_id: Mapped[str] = mapped_column(String(128), index=True, nullable=False)
-    # 极光推送 registration id;拿到才填(JCollectionAuth 同意后才下发)
-    registration_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
-    platform: Mapped[str] = mapped_column(String(16), nullable=False, default="android")
-    app_version: Mapped[str | None] = mapped_column(String(32), nullable=True)
-
-    # 收到过 service 心跳即 true(=该设备开过无障碍,功能对它有意义)
-    ever_protected: Mapped[bool] = mapped_column(
-        Boolean, nullable=False, default=False
-    )
-    # 最近一次 service 心跳时间(存活证明);超时即视为保护掉线
-    last_heartbeat_at: Mapped[datetime | None] = mapped_column(
-        DateTime(timezone=True), index=True, nullable=True
-    )
-    # 最近一次上报的无障碍开关状态(观测用)
-    last_report_protection_on: Mapped[bool] = mapped_column(
-        Boolean, nullable=False, default=False
-    )
-    # unknown / alive / silent / notified
-    liveness_state: Mapped[str] = mapped_column(
-        String(16), nullable=False, default="unknown"
-    )
-    # 最近一次推送告警时间
-    notified_at: Mapped[datetime | None] = mapped_column(
-        DateTime(timezone=True), nullable=True
-    )
-
-    created_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), server_default=func.now(), nullable=False
-    )
-    updated_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True),
-        server_default=func.now(),
-        onupdate=func.now(),
-        nullable=False,
-    )
-
-    def __repr__(self) -> str:  # pragma: no cover
-        return (
-            f"<Device id={self.id} user_id={self.user_id} "
-            f"device_id={self.device_id} state={self.liveness_state}>"
-        )

app/models/feedback.py

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

app/models/store_mapping.py

@@ -86,6 +86,9 @@ class StoreMapping(Base):
     taobao_share_url: Mapped[str | None] = mapped_column(String(256), nullable=True)   # m.tb.cn 短链
     taobao_resolved_url: Mapped[str | None] = mapped_column(Text, nullable=True)        # 解析出的目标 URL(含 shopId)
     taobao_deeplink: Mapped[str | None] = mapped_column(Text, nullable=True)            # 拼好的 et-store/search deeplink
+    # 淘宝 deeplink 失效标记:比价撞"页面出错了"降级页时被置(pricebot server→server invalidate),
+    # NULL=有效。lookup 反查过滤掉非 NULL 的淘宝候选,不再返回坏 deeplink(重搜会写新行覆盖)。
+    taobao_deeplink_invalid_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
 
     # ===== 美团原料(同淘宝;dpurl.cn 短链 → 302 反查 poi_id_str → imeituan:// deeplink)=====
     # ⚠️ poi_id_str 每次分享重新加密、非稳定主键(调研文档 §八), 故单列存"可复跳的一次性票据",

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:
@@ -66,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):

app/repositories/coupon_state.py

@@ -31,46 +31,33 @@ def today_cn() -> date:
 
 # ===== 弹窗频控(coupon_prompt_engagement)=====
 
-def has_engaged_today(
-    db: Session, device_id: str, app_package: str | None = None
-) -> bool:
-    """这台设备今天是否已弹过/engage 过领券引导窗。有 = 不再弹。
-
-    2026-06-12 按 App 为单位:
-    - app_package 给定(新客户端):只看**该 App** 当天的行;app_package=NULL 的全局兜底行
-      (step=0 claim_started / 旧客户端写的)**不算**——领券中途终止不该堵死其他 App 的弹窗,
-      "彻底领完才全局不弹"由 has_completed_today 在 should-show 端点单独把关。
-    - app_package=None(旧客户端不带包名):保持旧全局语义,当天任意一行都算 engage 过。
-    """
-    cond = [
-        CouponPromptEngagement.device_id == device_id,
-        CouponPromptEngagement.engage_date == today_cn(),
-    ]
-    if app_package is not None:
-        cond.append(CouponPromptEngagement.app_package == app_package)
-    row = db.execute(select(CouponPromptEngagement.id).where(*cond)).first()
+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()
     return row is not None
 
 
 def mark_engagement(
-    db: Session,
-    device_id: str,
-    user_id: int | None,
-    engage_type: str,
-    app_package: str | None = None,
+    db: Session, device_id: str, package: str, user_id: int | None, engage_type: str
 ) -> None:
-    """记今日意向(shown / claim_started / dismissed)。(device, App, 今天) 唯一,幂等 upsert。
+    """记今日意向(shown / claim_started / dismissed)。(device, package, 今天) 唯一,幂等 upsert。
 
-    app_package=None = 全局兜底行(step=0 透传链路拿不到包名 / 旧客户端),与各 App 行互不覆盖。
-    同 (device, App, 日) 重复上报走更新(shown → dismissed 升级 engage_type)。
+    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,
-            # SQLAlchemy 的 == None 会生成 IS NULL,NULL 兜底行与 App 行各自独立 upsert。
-            CouponPromptEngagement.app_package == app_package,
         )
     ).scalar_one_or_none()
     if row is not None:
@@ -79,20 +66,20 @@ def mark_engagement(
             row.user_id = user_id
     else:
         db.add(CouponPromptEngagement(
-            device_id=device_id, user_id=user_id, app_package=app_package,
+            device_id=device_id, package=package, user_id=user_id,
             engage_date=today, engage_type=engage_type,
         ))
     try:
         db.commit()
     except IntegrityError:
-        # 并发下另一请求刚插了同 (device, App, 日) → 唯一约束撞,回滚忽略(本就幂等)。
+        # 并发下另一请求刚插了同 (device, package, 日) → 唯一约束撞,回滚忽略(本就幂等)。
         db.rollback()
 
 
 def reset_today_engagement(db: Session, device_id: str) -> int:
-    """删这台设备今天的 engagement(开发设置「重置今日领券弹窗状态」调,测频控用)。
-    按 device+日 删,**所有 App 的频控行 + NULL 兜底行一并清**(频控按 App 拆行后语义不变:
-    重置 = 等效重装,每个 App 今天都又能弹)。返回删除行数。"""
+    """删这台设备今天**所有 App** 的 engagement(开发设置「重置今日领券弹窗状态」调,测频控用)。
+    删后各 App has_engaged_today → false,今天又都能弹。返回删除行数。
+    (不按 package 过滤:重置是"把今天清干净从头测",清全部 App 最符合预期。)"""
     result = db.execute(
         delete(CouponPromptEngagement).where(
             CouponPromptEngagement.device_id == device_id,
@@ -105,21 +92,6 @@ def reset_today_engagement(db: Session, device_id: str) -> int:
 
 # ===== 今日跑完整轮(coupon_daily_completion)=====
 
-def reset_today_completion(db: Session, device_id: str) -> int:
-    """删这台设备今天的「已跑完整轮」记录(开发设置「重置今日领券弹窗状态」调)。
-    删后 has_completed_today → false:首页「去领取」卡恢复可点、领券弹窗 CTA 不再置灰
-    (只清 engagement 不清这条的话, 重置后弹窗能弹但「一键自动领取」仍是灰的, 2026-06-12)。
-    返回删除行数。"""
-    result = db.execute(
-        delete(CouponDailyCompletion).where(
-            CouponDailyCompletion.device_id == device_id,
-            CouponDailyCompletion.complete_date == today_cn(),
-        )
-    )
-    db.commit()
-    return result.rowcount or 0
-
-
 def has_completed_today(db: Session, device_id: str) -> bool:
     """这台设备今天是否已跑完整轮领券(到 done 帧)。有 = 首页置灰、不能再领。"""
     row = db.execute(
@@ -159,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/device.py

@@ -1,106 +0,0 @@
-"""device 表读写(设备注册 / 心跳 / 超时扫描)。"""
-from __future__ import annotations
-
-from datetime import datetime, timedelta, timezone
-
-from sqlalchemy import select
-from sqlalchemy.orm import Session
-
-from app.models.device import Device
-
-
-def _get(db: Session, *, user_id: int, device_id: str) -> Device | None:
-    stmt = select(Device).where(
-        Device.user_id == user_id, Device.device_id == device_id
-    )
-    return db.execute(stmt).scalar_one_or_none()
-
-
-def register_or_update(
-    db: Session,
-    *,
-    user_id: int,
-    device_id: str,
-    registration_id: str | None,
-    platform: str = "android",
-    app_version: str | None = None,
-) -> Device:
-    """注册设备或更新其 registration_id / 元信息。upsert by (user_id, device_id)。"""
-    device = _get(db, user_id=user_id, device_id=device_id)
-    if device is None:
-        device = Device(
-            user_id=user_id,
-            device_id=device_id,
-            registration_id=registration_id,
-            platform=platform or "android",
-            app_version=app_version,
-        )
-        db.add(device)
-    else:
-        if registration_id:
-            device.registration_id = registration_id
-        if platform:
-            device.platform = platform
-        if app_version:
-            device.app_version = app_version
-    db.commit()
-    db.refresh(device)
-    return device
-
-
-def touch_heartbeat(
-    db: Session,
-    *,
-    user_id: int,
-    device_id: str,
-    accessibility_enabled: bool,
-    registration_id: str | None,
-) -> Device:
-    """处理一次心跳(心跳也能自注册)。
-
-    service 心跳或 accessibility_enabled=true 时,刷新存活并把状态机重置回 alive、
-    清掉 notified_at(掉线恢复 → 下次再断才会再推一条)。
-    """
-    now = datetime.now(timezone.utc)
-    device = _get(db, user_id=user_id, device_id=device_id)
-    if device is None:
-        device = Device(user_id=user_id, device_id=device_id)
-        db.add(device)
-
-    if registration_id:
-        device.registration_id = registration_id
-    device.last_report_protection_on = accessibility_enabled
-
-    if accessibility_enabled:
-        device.last_heartbeat_at = now
-        device.ever_protected = True
-        device.liveness_state = "alive"
-        device.notified_at = None
-
-    db.commit()
-    db.refresh(device)
-    return device
-
-
-def list_overdue(db: Session, *, timeout_minutes: int) -> list[Device]:
-    """掉线设备:曾经保护过、当前 alive、心跳超时。
-
-    本期只做终端打印检测、不推送 → 不再要求有 registration_id(没接极光 token 的设备也要检出)。
-    """
-    cutoff = datetime.now(timezone.utc) - timedelta(minutes=timeout_minutes)
-    stmt = select(Device).where(
-        Device.ever_protected.is_(True),
-        Device.liveness_state == "alive",
-        Device.last_heartbeat_at.is_not(None),
-        Device.last_heartbeat_at < cutoff,
-    )
-    return list(db.execute(stmt).scalars().all())
-
-
-def mark_notified(db: Session, *, device_id_pk: int) -> None:
-    """标记已推送告警(状态机进入 notified,避免重复推送)。"""
-    device = db.get(Device, device_id_pk)
-    if device is not None:
-        device.liveness_state = "notified"
-        device.notified_at = datetime.now(timezone.utc)
-        db.commit()

app/repositories/store_mapping.py

@@ -15,7 +15,7 @@ from __future__ import annotations
 import logging
 import math
 
-from sqlalchemy import select
+from sqlalchemy import func, select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -107,6 +107,23 @@ def upsert(db: Session, payload: StoreMappingIn) -> tuple[int, int | None]:
     return 0, existing.id
 
 
+def mark_taobao_deeplink_invalid(db: Session, shop_id: str) -> int:
+    """把所有 id_taobao=shop_id 的行标记淘宝 deeplink 失效(置 invalid_at=now)。返回本次新标记的行数。
+
+    按 shopId 标记**所有**行 —— 同一个坏 shopId(撞淘宝"页面出错了"降级页)可能散在多次比价的
+    多行里, 全标掉才能让后续 lookup 不再返回它。幂等: 已标记的行(invalid_at 非 NULL)跳过。"""
+    result = db.execute(
+        update(StoreMapping)
+        .where(
+            StoreMapping.id_taobao == shop_id,
+            StoreMapping.taobao_deeplink_invalid_at.is_(None),
+        )
+        .values(taobao_deeplink_invalid_at=func.now())
+    )
+    db.commit()
+    return result.rowcount or 0
+
+
 # ============================================================
 # 缓存查询: 比价前按"源平台店名"反查已沉淀的各目标平台店铺 id, 命中就让 pricebot 直接
 # deeplink 跳店内搜索, 省掉"开平台→进店→分享反查"整段。
@@ -161,6 +178,9 @@ def lookup_nearest(
         if tgt == src_key:
             continue  # 不返回源平台自己
         cands = [r for r in rows if getattr(r, id_attr)]
+        if tgt == "taobao":
+            # 失效的淘宝 deeplink(撞过错误页被 invalidate)整条排除 = 当没缓存, pricebot 走正常搜店。
+            cands = [r for r in cands if r.taobao_deeplink_invalid_at is None]
         if not cands:
             continue
         best = _pick_best(cands, lat, lng)

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:
-    """取用户金币账户,不存在则建一个空账户。"""
-    acc = db.get(CoinAccount, user_id)
+def get_or_create_account(
+    db: Session, user_id: int, *, commit: bool = True, lock: bool = False
+) -> 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/coupon_state.py

@@ -7,34 +7,30 @@ from pydantic import BaseModel
 class CouponPromptDismissIn(BaseModel):
     """客户端拒绝/关闭领券引导窗的通知体。
 
-    server 据此记一条今日 engagement(dismissed)。2026-06-12 频控按 App 为单位:
-    package = 在哪个外卖 App 关的窗(Android 包名);旧客户端不带 → None(全局兜底行)。
+    server 据此记一条今日 engagement(dismissed)→ 今天**这个 App** 不再弹引导窗。
+    频控按 (device, package, 日):各 App 独立。package 缺省 ""(老客户端兼容,退化为全局态)。
     MVP 不鉴权,按 device_id 判断;user_id 登录态带上就一并记(资产),可空。
     """
 
     device_id: str
+    package: str = ""
     user_id: int | None = None
-    package: str | None = None
 
 
 class CouponPromptShownIn(BaseModel):
-    """领券引导窗已对某 App 弹出(shown)的通知体——按 App 频控的主判据。
+    """客户端弹出领券引导窗即上报(记 shown)。
 
-    客户端弹窗一亮即上报:该 App 今天不再弹(「一个 app 一天只能弹一次,
-    时机=每天第一次进入」);领/拒后续上报只升级 engage_type,不影响频控。
+    弹出那刻就记一条今日 engagement(shown)→ 今天**这个 App** 不再自动弹(频控主判据,
+    管跨重装;本地 SP 兜后台抖动)。后续用户点领取/拒绝再把 type 升级成 claim_started/dismissed。
     """
 
     device_id: str
+    package: str
     user_id: int | None = None
-    package: str | None = None
 
 
 class CouponPromptShouldShowOut(BaseModel):
-    """切到外卖 App 时是否还应弹领券引导窗。
-
-    false 的两种来源:该 App 今天已弹过(per-App engagement)/ 今天已跑完整轮领券
-    (completion,彻底领完全局不弹)。
-    """
+    """切到外卖 App 时是否还应弹领券引导窗。今天**这个 App** 已 engage(弹/领/拒)过 → false。"""
 
     should_show: bool
 

app/schemas/device.py

@@ -1,36 +0,0 @@
-"""设备注册 / 心跳相关 schema(无障碍保护存活检测)。"""
-from __future__ import annotations
-
-from datetime import datetime
-
-from pydantic import BaseModel, ConfigDict
-
-
-class DeviceRegisterRequest(BaseModel):
-    device_id: str
-    registration_id: str | None = None
-    platform: str = "android"
-    app_version: str | None = None
-
-
-class HeartbeatRequest(BaseModel):
-    device_id: str
-    source: str = "service"          # service | app
-    accessibility_enabled: bool = True
-    registration_id: str | None = None
-
-
-class DeviceOut(BaseModel):
-    model_config = ConfigDict(from_attributes=True)
-
-    id: int
-    device_id: str
-    registration_id: str | None
-    ever_protected: bool
-    liveness_state: str
-    last_heartbeat_at: datetime | None
-    updated_at: datetime | None
-
-
-class OkResponse(BaseModel):
-    ok: bool = True

app/schemas/store_mapping.py

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

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` 即末页。
-- `status` / `user_id` 均为精确匹配,可叠加。
+- **offset 分页**(同 [admin-users-list](./admin-users-list.md)):`cursor` 即 offset,传上一页返回的 `next_cursor`;`next_cursor=null` 即末页。改用 offset 是为了在任意列排序下游标语义统一,代价是翻页期间数据变动可能错位一条(admin 低频可接受)。
+- 排序:`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/feedback.md

@@ -7,8 +7,8 @@
 | 字段 | 类型 | 必填 | 说明 |
 |---|---|---|---|
 | `content` | string | ✓ | 反馈正文,**1-2000 字**(strip 后) |
-| `contact` | string | ✓ | 联系方式(微信/QQ/手机号),**1-128 字**,便于回访 |
-| `images` | file[] | ✗ | 截图,**最多 4 张**,每张走头像同款校验(JPEG/PNG/WebP,≤ 5 MB,魔数嗅探) |
+| `contact` | string | ✗ | 联系方式(微信/QQ/手机号),**≤128 字**。原型改版后客户端已不再采集、不传该字段(后端默认空串);保留字段兼容旧端 |
+| `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/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 | 提交时间 |

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_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