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

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

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/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/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/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"