feat(admin): 设备存活监控接口 + 首次无障碍开启时间(first_protected_at)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

alembic/versions/bcfcaf07152b_merge_meituan_coupon_image_meta_and_.py

@@ -0,0 +1,26 @@
+"""merge meituan_coupon_image_meta and comparison/ad heads
+
+Revision ID: bcfcaf07152b
+Revises: 4dc2af7ebe74, meituan_coupon_image_meta
+Create Date: 2026-06-24 20:34:09.475641
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'bcfcaf07152b'
+down_revision: Union[str, Sequence[str], None] = ('4dc2af7ebe74', 'meituan_coupon_image_meta')
+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/device_first_protected_at.py

@@ -0,0 +1,31 @@
+"""device_liveness 加 first_protected_at(首次开无障碍时刻)
+
+Revision ID: device_first_protected_at
+Revises: bcfcaf07152b
+Create Date: 2026-06-25 00:00:00.000000
+
+admin 设备存活页要展示「首次无障碍开启时间」。touch_heartbeat 在 ever_protected 首次翻 true
+时记一次(后续心跳不覆盖)。仅新增可空列,SQLite 原生支持 add_column、不用 batch;downgrade
+的 drop_column 在 SQLite 走 batch_alter_table 兜底。老设备无此时刻 → 留 NULL(无法准确回填)。
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "device_first_protected_at"
+down_revision: Union[str, Sequence[str], None] = "bcfcaf07152b"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "device_liveness",
+        sa.Column("first_protected_at", sa.DateTime(timezone=True), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("device_liveness") as batch_op:
+        batch_op.drop_column("first_protected_at")

app/admin/main.py

@@ -23,6 +23,7 @@ from app.admin.routers.comparison import router as comparison_router
 from app.admin.routers.config import router as config_router
 from app.admin.routers.cps import router as cps_router
 from app.admin.routers.dashboard import router as dashboard_router
+from app.admin.routers.device_liveness import router as device_liveness_router
 from app.admin.routers.ops_stat_config import router as ops_stat_config_router
 from app.admin.routers.feedback import router as feedback_router
 from app.admin.routers.feedback_qr import router as feedback_qr_router
@@ -83,6 +84,7 @@ def health() -> dict[str, str]:
 
 admin_app.include_router(auth_router)
 admin_app.include_router(dashboard_router)
+admin_app.include_router(device_liveness_router)
 admin_app.include_router(ops_stat_config_router)
 admin_app.include_router(ops_marquee_seed_router)
 admin_app.include_router(users_router)

app/admin/repositories/queries.py

@@ -8,14 +8,16 @@ from __future__ import annotations
 from datetime import datetime, timedelta, timezone
 from zoneinfo import ZoneInfo
 
-from sqlalchemy import Select, asc, desc, func, or_, select
+from sqlalchemy import Select, asc, case, desc, func, or_, select
 from sqlalchemy.orm import Session
 
 from app.core import rewards
+from app.core.config import settings
 from app.models.ad_feed_reward import AdFeedRewardRecord
 from app.models.ad_reward import AdRewardRecord
 from app.models.admin import AdminAuditLog
 from app.models.comparison import ComparisonRecord
+from app.models.device import DeviceLiveness
 from app.models.feedback import Feedback
 from app.models.onboarding import OnboardingCompletion
 from app.models.price_report import PriceReport
@@ -199,6 +201,156 @@ def list_onboarding_devices(db: Session, *, limit: int = 500) -> list[dict]:
     ]
 
 
+def _heartbeat_seconds_ago(last: datetime | None) -> int | None:
+    """距上次心跳的秒数(兼容 SQLite 取回的 naive datetime,按 UTC 处理)。None = 从没心跳。"""
+    if last is None:
+        return None
+    if last.tzinfo is None:
+        last = last.replace(tzinfo=timezone.utc)
+    return int((datetime.now(timezone.utc) - last).total_seconds())
+
+
+def _device_model_from_id(device_id: str) -> str:
+    """从 device_id(格式 device_<机型>_<hash>)解析机型;不规范则回退原 id。"""
+    parts = device_id.split("_")
+    if len(parts) >= 3 and parts[0] == "device":
+        return "_".join(parts[1:-1]).replace("_", " ")
+    return device_id
+
+
+def _attach_device_derived(devices: list[DeviceLiveness]) -> None:
+    """给每台设备瞬态挂 device_model / online / display_state / offline_seconds(供 schema 读)。
+
+    在线判定:开过无障碍(ever_protected)且距上次心跳 ≤ HEARTBEAT_TIMEOUT_MINUTES。
+    从没开过无障碍(ever_protected=False)= never(不算掉线,避免把装了没用的设备误报掉线)。
+    offline_seconds 仅 offline 时给值(=掉线时长),在线/从未启用为 None。"""
+    timeout_sec = max(1, int(settings.HEARTBEAT_TIMEOUT_MINUTES)) * 60
+    for d in devices:
+        d.device_model = _device_model_from_id(d.device_id)
+        secs = _heartbeat_seconds_ago(d.last_heartbeat_at)
+        if not d.ever_protected:
+            d.online = False
+            d.display_state = "never"
+            d.offline_seconds = None
+        elif secs is not None and secs <= timeout_sec:
+            d.online = True
+            d.display_state = "online"
+            d.offline_seconds = None
+        else:
+            d.online = False
+            d.display_state = "offline"
+            d.offline_seconds = secs
+
+
+def _attach_device_user_info(db: Session, devices: list[DeviceLiveness]) -> None:
+    """给每台设备瞬态挂归属用户 phone/nickname(同 _attach_user_info,供 admin schema 读)。"""
+    uids = {d.user_id for d in devices}
+    if not uids:
+        return
+    rows = db.execute(
+        select(User.id, User.phone, User.nickname).where(User.id.in_(uids))
+    ).all()
+    umap = {uid: (phone, nick) for uid, phone, nick in rows}
+    for d in devices:
+        phone, nick = umap.get(d.user_id, (None, None))
+        d.phone = phone
+        d.nickname = nick
+
+
+def _liveness_cutoff() -> datetime:
+    """掉线判定分界:此刻 - HEARTBEAT_TIMEOUT_MINUTES。心跳早于它 = 掉线(同 list_overdue 口径)。"""
+    timeout_min = max(1, int(settings.HEARTBEAT_TIMEOUT_MINUTES))
+    return datetime.now(timezone.utc) - timedelta(minutes=timeout_min)
+
+
+def list_device_liveness(
+    db: Session,
+    *,
+    status: str | None = None,
+    device_id: str | None = None,
+    phone: str | None = None,
+    user_id: int | None = None,
+    sort_by: str = "status",
+    sort_order: str = "desc",
+    limit: int = 20,
+    cursor: int | None = None,
+) -> tuple[list[DeviceLiveness], int | None, int]:
+    """设备存活列表(admin 全量)。按 在线情况(online/offline/never)/ 设备id(包含)/
+    归属用户(手机号前缀 或 user_id)筛,offset 分页。join user 取 phone/nickname,派生
+    device_model/online/display_state/offline_seconds 挂行上。
+    默认排序 status=掉线置顶(offline → online → never;掉线组内掉得最久在前)。"""
+    cutoff = _liveness_cutoff()
+    stmt = select(DeviceLiveness)
+    # 在线情况派生筛选(口径同 _attach_device_derived:ever_protected + 心跳是否过阈值)
+    if status == "online":
+        stmt = stmt.where(
+            DeviceLiveness.ever_protected.is_(True),
+            DeviceLiveness.last_heartbeat_at.is_not(None),
+            DeviceLiveness.last_heartbeat_at >= cutoff,
+        )
+    elif status == "offline":
+        stmt = stmt.where(
+            DeviceLiveness.ever_protected.is_(True),
+            DeviceLiveness.last_heartbeat_at.is_not(None),
+            DeviceLiveness.last_heartbeat_at < cutoff,
+        )
+    elif status == "never":
+        stmt = stmt.where(DeviceLiveness.ever_protected.is_(False))
+    if device_id and device_id.strip():
+        stmt = stmt.where(DeviceLiveness.device_id.like(f"%{device_id.strip()}%"))
+    if user_id is not None:
+        stmt = stmt.where(DeviceLiveness.user_id == user_id)
+    if phone:
+        stmt = stmt.where(
+            DeviceLiveness.user_id.in_(select(User.id).where(User.phone.like(f"{phone}%")))
+        )
+
+    if sort_by in ("last_heartbeat_at", "created_at"):
+        col = (
+            DeviceLiveness.last_heartbeat_at
+            if sort_by == "last_heartbeat_at"
+            else DeviceLiveness.created_at
+        )
+        order_fn = asc if sort_order == "asc" else desc
+        id_order = asc(DeviceLiveness.id) if sort_order == "asc" else desc(DeviceLiveness.id)
+        sort_clause: tuple = (order_fn(col), id_order)
+    else:
+        # 默认「掉线置顶」:offline(0) → online(1) → never(2);掉线组内按心跳最旧(掉得最久)在前
+        rank = case(
+            (DeviceLiveness.ever_protected.is_(False), 2),
+            (DeviceLiveness.last_heartbeat_at < cutoff, 0),
+            else_=1,
+        )
+        sort_clause = (asc(rank), asc(DeviceLiveness.last_heartbeat_at), desc(DeviceLiveness.id))
+
+    items, next_cursor, total = offset_paginate(db, stmt, sort_clause, limit=limit, cursor=cursor)
+    _attach_device_user_info(db, items)
+    _attach_device_derived(items)
+    return items, next_cursor, total
+
+
+def device_liveness_stats(db: Session) -> dict:
+    """顶部卡片:总设备数 + 在线 / 已掉线 / 未启用(按心跳阈值派生,口径同列表)。"""
+    cutoff = _liveness_cutoff()
+
+    def _count(*conds) -> int:
+        return int(db.execute(select(func.count(DeviceLiveness.id)).where(*conds)).scalar_one())
+
+    total = int(db.execute(select(func.count(DeviceLiveness.id))).scalar_one())
+    never = _count(DeviceLiveness.ever_protected.is_(False))
+    online = _count(
+        DeviceLiveness.ever_protected.is_(True),
+        DeviceLiveness.last_heartbeat_at.is_not(None),
+        DeviceLiveness.last_heartbeat_at >= cutoff,
+    )
+    offline = _count(
+        DeviceLiveness.ever_protected.is_(True),
+        DeviceLiveness.last_heartbeat_at.is_not(None),
+        DeviceLiveness.last_heartbeat_at < cutoff,
+    )
+    return {"total": total, "online": online, "offline": offline, "never": never}
+
+
 def list_all_coin_transactions(
     db: Session,
     *,

app/admin/routers/device_liveness.py

@@ -0,0 +1,63 @@
+"""admin 设备存活监控(只读):列设备心跳/在线掉线 + 顶部统计卡片。
+
+数据源 device_liveness 表(心跳 last_heartbeat_at + liveness_state + kill_alert_pending,
+见 app/models/device.py)。在线/掉线、掉线时长由 repo 按 HEARTBEAT_TIMEOUT_MINUTES 阈值派生。
+纯读:无写、无审计。任意登录管理员可看(同大盘/设备管理,无角色门)。
+"""
+from __future__ import annotations
+
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Query
+
+from app.admin.deps import AdminDb, get_current_admin
+from app.admin.repositories import queries
+from app.admin.schemas.common import CursorPage
+from app.admin.schemas.device import DeviceLivenessItem, DeviceLivenessStats
+
+router = APIRouter(
+    prefix="/admin/api/device-liveness",
+    tags=["admin-device-liveness"],
+    dependencies=[Depends(get_current_admin)],
+)
+
+
+@router.get("/stats", response_model=DeviceLivenessStats, summary="设备存活统计(顶部卡片)")
+def device_stats(db: AdminDb) -> DeviceLivenessStats:
+    return DeviceLivenessStats(**queries.device_liveness_stats(db))
+
+
+@router.get(
+    "",
+    response_model=CursorPage[DeviceLivenessItem],
+    summary="设备存活列表(在线情况/设备id/归属用户 筛选 + 排序 + 分页,默认掉线置顶)",
+)
+def list_devices(
+    db: AdminDb,
+    status: Annotated[str | None, Query(pattern="^(online|offline|never)$")] = None,
+    device_id: Annotated[str | None, Query(max_length=128)] = None,
+    phone: Annotated[str | None, Query(max_length=20)] = None,
+    user_id: Annotated[int | None, Query()] = None,
+    sort_by: Annotated[
+        str, Query(pattern="^(status|last_heartbeat_at|created_at)$")
+    ] = "status",
+    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[DeviceLivenessItem]:
+    items, next_cursor, total = queries.list_device_liveness(
+        db,
+        status=status,
+        device_id=device_id,
+        phone=phone,
+        user_id=user_id,
+        sort_by=sort_by,
+        sort_order=sort_order,
+        limit=limit,
+        cursor=cursor,
+    )
+    return CursorPage(
+        items=[DeviceLivenessItem.model_validate(d) for d in items],
+        next_cursor=next_cursor,
+        total=total,
+    )

app/admin/schemas/ad_config.py

@@ -15,6 +15,7 @@ class AdConfigOut(BaseModel):
     reward_enabled: bool
     compare_ad_enabled: bool
     coupon_ad_enabled: bool
+    withdrawal_ad_enabled: bool
 
 
 class AdConfigUpdate(BaseModel):
@@ -28,3 +29,4 @@ class AdConfigUpdate(BaseModel):
     reward_enabled: bool | None = None
     compare_ad_enabled: bool | None = None
     coupon_ad_enabled: bool | None = None
+    withdrawal_ad_enabled: bool | None = None

app/admin/schemas/device.py

@@ -0,0 +1,54 @@
+"""admin 设备存活监控 schemas(只读)。
+
+数据源 device_liveness 表(见 app/models/device.py)。online / display_state /
+offline_seconds 为后端按 HEARTBEAT_TIMEOUT_MINUTES 阈值派生的瞬态字段(非 DB 列),
+在 repo 里算好挂到行上,供 from_attributes 读出。
+"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict
+
+
+class DeviceLivenessItem(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    user_id: int
+    # join user 取(瞬态;理论上 user 恒存在,留空仅防脏数据)
+    phone: str | None = None
+    nickname: str | None = None
+
+    device_id: str
+    device_model: str | None = None      # 由 device_id 解析(device_<机型>_<hash>);非 DB 列
+    platform: str
+    app_version: str | None = None
+    registration_id: str | None = None  # 非空 = 拿到极光 token、可推送
+
+    ever_protected: bool                 # 是否开过无障碍(=该设备对功能有意义)
+    first_protected_at: datetime | None = None   # 首次开无障碍时刻(老设备为 null)
+    last_heartbeat_at: datetime | None = None
+    last_report_protection_on: bool      # 心跳里上报的无障碍开关(恒 true,仅观测)
+    liveness_state: str                  # unknown / alive / silent(当前未用) / notified
+    notified_at: datetime | None = None
+    kill_alert_pending: bool             # 掉线召回待客户端 ack(与 state 解耦)
+    created_at: datetime
+    updated_at: datetime
+
+    # ===== 后端派生(非 DB 列)=====
+    online: bool = False                 # ever_protected 且距上次心跳 ≤ 阈值
+    # 显示态:online=在线 / offline=已掉线 / never=从未启用(没开过无障碍)
+    display_state: str = "never"
+    # 掉线时长(秒):仅 offline 时有值;在线 / 从未启用为 None
+    offline_seconds: int | None = None
+
+
+class DeviceLivenessStats(BaseModel):
+    """顶部卡片:总设备数 + 在线 / 已掉线 / 未启用(按心跳阈值派生的 operator 口径,
+    不暴露内部状态机 unknown/alive/silent/notified)。"""
+
+    total: int
+    online: int
+    offline: int
+    never: int

app/api/v1/ad.py

@@ -53,6 +53,9 @@ REASON_UNKNOWN_USER = 2  # user_id 不存在(可能伪造)
 
 REWARD_SCENE_REWARD_VIDEO = "reward_video"
 REWARD_SCENE_SIGNIN_BOOST = "signin_boost"
+# 提现看视频:看完才能提现的「硬门槛」广告,**不发金币**,只记一条幂等记录(收益由 eCPM 上报口径
+# ad_type="withdrawal_video" 单独统计)。故意不放进 SUPPORTED_REWARD_SCENES——它不走发币分支。
+REWARD_SCENE_WITHDRAWAL_AD = "withdrawal_ad"
 SUPPORTED_REWARD_SCENES = {REWARD_SCENE_REWARD_VIDEO, REWARD_SCENE_SIGNIN_BOOST}
 
 
@@ -126,6 +129,19 @@ def pangle_callback(request: Request, db: DbSession) -> PangleCallbackOut:
         return PangleCallbackOut(is_verify=True, reason=REASON_OK)
 
     try:
+        if reward_scene == REWARD_SCENE_WITHDRAWAL_AD:
+            # 提现看视频:已知合法场景,只记一条幂等记录、**不动钱包**(发不发币由 reward_scene 决定)。
+            # is_verify=true 让穿山甲不重试(区别于 unknown_scene 的 false)。
+            crud_ad.record_external_reward(
+                db, user_id, trans_id, coin=0, reward_scene=REWARD_SCENE_WITHDRAWAL_AD,
+                ad_session_id=ad_session_id, ecpm=ecpm,
+                reward_name=params.get("reward_name"), raw=raw[:1024],
+                status="withdrawal_ad",
+            )
+            logger.info(
+                "pangle callback withdrawal_ad user_id=%d trans_id=%s", user_id, trans_id,
+            )
+            return PangleCallbackOut(is_verify=True, reason=REASON_OK)
         if reward_scene not in SUPPORTED_REWARD_SCENES:
             rec = crud_ad.record_external_reward(
                 db, user_id, trans_id, coin=0, reward_scene=reward_scene[:32],

app/api/v1/cps_redirect.py

@@ -83,7 +83,10 @@ def cps_landing(code: str, request: Request, db: Session = Depends(get_db)):
         redirect_uri = f"{settings.CPS_REDIRECT_BASE.rstrip('/')}/wx/oauth/cb"
         auth_url = wx_oauth.build_authorize_url(redirect_uri, "snsapi_base", f"base:{code}")
         return RedirectResponse(auth_url, status_code=302)
-    _record(db, link, request, "visit", openid=openid)
+    # ?authed=1 是 userinfo 授权后的自动回跳重载,不是用户主动打开:visit 已在授权前那次
+    # 落地(base 静默授权回来那次)记过,这里再记会让每个走"领券授权"的用户 visit 虚增 1。
+    if request.query_params.get("authed") != "1":
+        _record(db, link, request, "visit", openid=openid)
     if link.platform == "taobao":
         activity = db.get(CpsActivity, link.activity_id)
         image_url = (activity.image_url if activity else None) or _DEFAULT_TAOBAO_IMAGE
@@ -201,8 +204,9 @@ function copyToken(){
   if(UINFO_URL){location.href=UINFO_URL;return}
   doCopy();
 }
-// userinfo 授权回流(?authed=1):自动复制,把"点领券→授权→复制"衔接成一步无感
-if(location.search.indexOf('authed=1')>=0){doCopy()}
+// userinfo 授权回流(?authed=1):自动复制,把"点领券→授权→复制"衔接成一步无感;
+// 复制后用 replaceState 清掉 authed=1,避免刷新/微信返回(BFCache)重载时重复触发复制上报。
+if(location.search.indexOf('authed=1')>=0){doCopy();history.replaceState(null,'',location.pathname)}
 </script>
 </body>
 </html>"""

app/api/v1/platform.py

@@ -68,6 +68,7 @@ def ad_config(db: DbSession) -> AdConfigPublicOut:
         reward_enabled=c["reward_enabled"],
         compare_ad_enabled=c["compare_ad_enabled"],
         coupon_ad_enabled=c["coupon_ad_enabled"],
+        withdrawal_ad_enabled=c["withdrawal_ad_enabled"],
     )
 
 

app/models/device.py

@@ -51,6 +51,11 @@ class DeviceLiveness(Base):
     ever_protected: Mapped[bool] = mapped_column(
         Boolean, nullable=False, default=False
     )
+    # 首次开无障碍(首次收到 accessibility_enabled 心跳)的时刻;ever_protected 第一次翻 true 时记一次,
+    # 后续心跳不覆盖。老设备(迁移前已 protected)无此值 → NULL。
+    first_protected_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
     # 最近一次 service 心跳时间(存活证明);超时即视为保护掉线
     last_heartbeat_at: Mapped[datetime | None] = mapped_column(
         DateTime(timezone=True), index=True, nullable=True

app/repositories/app_config.py

@@ -109,6 +109,7 @@ _AD_CONFIG_DEFAULTS: dict[str, Any] = {
     "reward_enabled": True,               # 福利激励视频开关
     "compare_ad_enabled": True,           # 比价广告开关
     "coupon_ad_enabled": True,            # 领券广告开关
+    "withdrawal_ad_enabled": True,        # 提现激励视频开关(关=直接放行提现,不挡用户的钱)
 }
 
 

app/repositories/device.py

@@ -72,6 +72,8 @@ def touch_heartbeat(
     device.last_report_protection_on = accessibility_enabled
 
     if accessibility_enabled:
+        if not device.ever_protected:
+            device.first_protected_at = now  # 首次开无障碍记一次,后续心跳不覆盖
         device.last_heartbeat_at = now
         device.ever_protected = True
         device.liveness_state = "alive"

app/schemas/platform.py

@@ -43,6 +43,7 @@ class AdConfigPublicOut(BaseModel):
     reward_enabled: bool  # 福利激励视频开关
     compare_ad_enabled: bool  # 比价广告开关
     coupon_ad_enabled: bool  # 领券广告开关
+    withdrawal_ad_enabled: bool  # 提现激励视频开关(关=客户端直接放行提现)
 
 
 class AppVersionOut(BaseModel):