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/api/v1/auth.py

@@ -12,11 +12,11 @@ from __future__ import annotations
 
 import logging
 
-from fastapi import APIRouter, Depends, HTTPException, Request, status
+from fastapi import APIRouter, Depends, HTTPException, status
 
 from app.api.deps import CurrentUser, DbSession
 from app.core import test_account
-from app.core.ratelimit import enforce_rate_limit, rate_limit
+from app.core.ratelimit import rate_limit
 from app.core.security import TokenError, decode_token, issue_token_pair
 from app.integrations.jiguang import JiguangError, mask_phone, verify_and_get_phone
 from app.integrations.sms import SmsError, send_code, verify_code
@@ -38,12 +38,6 @@ logger = logging.getLogger("shagua.auth")
 
 router = APIRouter(prefix="/api/v1/auth", tags=["auth"])
 
-# 手机号登录防刷:同一设备(device_id) + 同一 IP 每小时最多的登录尝试次数(成功/失败都计)。
-SMS_LOGIN_MAX_PER_HOUR = 5
-# 发码防刷:同一设备(device_id) + 同一 IP 每小时最多的发码次数。比登录略宽(发码含正常重发);
-# 堵「换手机号绕开单号 60s 冷却 / 单号每日上限」的洞 —— 那两道是单号维度,一机换号能绕开。
-SMS_SEND_MAX_PER_HOUR_PER_DEVICE = 10
-
 
 def _login_response(
     user, *, onboarding_completed: bool, force_onboarding: bool = False
@@ -92,26 +86,13 @@ def jverify_login(req: JverifyLoginRequest, db: DbSession) -> TokenWithUser:
     summary="发送短信验证码",
     dependencies=[Depends(rate_limit(10, 60, "sms-send"))],  # 同 IP 每分钟≤10 次(防一 IP 刷不同号)
 )
-def sms_send(req: SmsSendRequest, request: Request) -> SmsSendResponse:
+def sms_send(req: SmsSendRequest) -> SmsSendResponse:
     # 测试账号:不真发短信(号码非真实手机号,真发会失败/浪费),直接放行让客户端进入填码界面。
     # 真正的"免验证码"在 /sms/login 跳过校验;每日上限也在 login 处算(send 不耗额度)。
-    # (也不受下面设备发码限流约束:QA 联调要反复发码。)
     if test_account.is_test_account(req.phone):
         logger.info("test_account sms_send short-circuit (不真发)")
         return SmsSendResponse(sent=True, mock=True, cooldown_sec=0)
 
-    # 防刷:同一设备(device_id) + 同一 IP 每小时最多 SMS_SEND_MAX_PER_HOUR_PER_DEVICE 次发码。
-    # 补「换手机号绕开单号 60s 冷却 / 单号每日上限」的洞(那两道是单号维度,一机换号能绕);设备维度按机器封顶,
-    # 挡短信轰炸/烧钱。放在真发(send_code)之前 → 超限直接拦下、不真发短信。与路由上 IP 维度(10次/分钟)互补。
-    enforce_rate_limit(
-        request,
-        scope="sms-send-device",
-        subject=req.device_id,
-        limit=SMS_SEND_MAX_PER_HOUR_PER_DEVICE,
-        window_sec=3600,
-        detail="操作过于频繁,请稍后再试",
-    )
-
     try:
         cooldown = send_code(req.phone)
     except SmsError as e:
@@ -128,10 +109,9 @@ def sms_send(req: SmsSendRequest, request: Request) -> SmsSendResponse:
     summary="手机号+验证码登录",
     dependencies=[Depends(rate_limit(20, 60, "sms-login"))],  # 防撞库爆破(另有单码失败次数上限)
 )
-def sms_login(req: SmsLoginRequest, request: Request, db: DbSession) -> TokenWithUser:
+def sms_login(req: SmsLoginRequest, db: DbSession) -> TokenWithUser:
     # 测试账号:免验证码登录 + 每日上限 + 每次都走新手引导(详见 app/core/test_account.py)。
     # 放在最前面:命中即不校验验证码;先扣当日额度,超限直接拒,挡住有人猜到号后脚本刷。
-    # 测试账号走自己的每日额度、不受下面 (设备+IP) 每小时限流约束(QA 需在一小时内反复登录联调)。
     if test_account.is_test_account(req.phone):
         if not test_account.try_consume_quota():
             raise HTTPException(status_code=429, detail="测试账号今日使用次数已达上限,请明天再试")
@@ -143,19 +123,6 @@ def sms_login(req: SmsLoginRequest, request: Request, db: DbSession) -> TokenWit
         # 不读/不写 onboarding_completion 表。
         return _login_response(user, onboarding_completed=False, force_onboarding=True)
 
-    # 防刷:同一设备(device_id) + 同一 IP 每小时最多 SMS_LOGIN_MAX_PER_HOUR 次登录尝试。放在验证码校验
-    # **之前** → 输错验证码的失败尝试也计数,才挡得住撞库/爆破。与路由上 IP 维度的 sms-login 限流(同 IP)互补。
-    # ⚠️ 按设备而非手机号 → 一台机器换不同手机号刷登录也受限(防一机狂登多号);device_id 空(老客户端)时
-    #    退化为该 IP 下所有空设备聚一桶,仍受限。
-    enforce_rate_limit(
-        request,
-        scope="sms-login-device",
-        subject=req.device_id,
-        limit=SMS_LOGIN_MAX_PER_HOUR,
-        window_sec=3600,
-        detail="登录尝试过于频繁,请稍后再试",
-    )
-
     if not verify_code(req.phone, req.code):
         raise HTTPException(status_code=400, detail="invalid sms code")
 

app/core/ratelimit.py

@@ -57,29 +57,3 @@ def rate_limit(limit: int, window_sec: float, scope: str):
             )
 
     return _dep
-
-
-def enforce_rate_limit(
-    request: Request,
-    scope: str,
-    subject: str,
-    limit: int,
-    window_sec: float,
-    *,
-    detail: str = "操作过于频繁,请稍后再试",
-) -> None:
-    """在路由内部手动限流,按 (subject, 客户端 IP) 计数。
-
-    用于限流 key 需要请求体字段(如手机号)、Depends 阶段还拿不到 body 时
-    —— 此时无法用 [rate_limit] 依赖,改在 handler 解析完 body 后调用本函数。
-    key = `scope:subject:client_ip`;同一 (subject, IP) 在 window_sec 内超过 limit 次 → 抛 429。
-    受 [settings.RATE_LIMIT_ENABLED] 总开关控制(与 [rate_limit] 一致)。
-    """
-    if not settings.RATE_LIMIT_ENABLED:
-        return
-    key = f"{scope}:{subject}:{_client_ip(request)}"
-    if not _hit(key, limit, window_sec):
-        raise HTTPException(
-            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
-            detail=detail,
-        )

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"

app/schemas/auth.py

@@ -71,10 +71,6 @@ class JverifyLoginRequest(BaseModel):
 
 class SmsSendRequest(BaseModel):
     phone: str = Field(..., min_length=11, max_length=11, pattern=r"^1\d{10}$")
-    device_id: str = Field(
-        "", max_length=64,
-        description="硬件级设备标识(Android ANDROID_ID),用于发码防刷按 设备+IP 限流;空=按 IP 聚一桶",
-    )
 
 
 class SmsSendResponse(BaseModel):

tests/test_auth.py

@@ -74,73 +74,6 @@ def test_sms_send_too_frequent(client) -> None:
     assert client.post("/api/v1/auth/sms/send", json={"phone": phone}).status_code == 429
 
 
-def test_sms_send_device_ip_rate_limit(client, monkeypatch) -> None:
-    """发码防刷:同一设备(device_id) + 同一 IP 每小时最多 N 次发码,超出 429。
-    用不同手机号(绕开单号 60s 冷却)证明限流按设备封顶 —— 堵「换号绕开单号冷却/每日上限」的洞。
-    conftest 默认关限流;本用例临时打开 + 调小阈值(避开同 IP 10/分钟那道)+ 清计数隔离。"""
-    from app.api.v1 import auth
-    from app.core import ratelimit
-
-    monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True)
-    monkeypatch.setattr(auth, "SMS_SEND_MAX_PER_HOUR_PER_DEVICE", 3)
-    ratelimit._buckets.clear()
-
-    device = "dev-send-A"
-    # 同设备、每次换不同手机号(绕开单号冷却)发码,前 3 次 200
-    for i in range(3):
-        r = client.post(
-            "/api/v1/auth/sms/send",
-            json={"phone": f"137001370{i:02d}", "device_id": device},
-        )
-        assert r.status_code == 200, f"第 {i + 1} 次应放行: {r.text}"
-    # 第 4 次:同设备超限 → 429(即便又换了手机号)
-    r = client.post(
-        "/api/v1/auth/sms/send",
-        json={"phone": "13700137003", "device_id": device},
-    )
-    assert r.status_code == 429, r.text
-
-    # 换个设备 → 独立计数(限流键含 device_id),放行
-    r = client.post(
-        "/api/v1/auth/sms/send",
-        json={"phone": "13700137004", "device_id": "dev-send-B"},
-    )
-    assert r.status_code == 200, r.text
-
-
-def test_sms_login_device_ip_rate_limit(client, monkeypatch) -> None:
-    """防刷:同一设备(device_id) + 同一 IP 每小时最多 SMS_LOGIN_MAX_PER_HOUR 次登录尝试,超出 429。
-    conftest 默认 RATE_LIMIT_ENABLED=false(内存计数跨用例累加),本用例临时打开并清空计数隔离。"""
-    from app.api.v1 import auth
-    from app.core import ratelimit
-
-    monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True)
-    ratelimit._buckets.clear()  # 隔离:清掉跨用例累加的内存计数
-
-    device = "dev-rl-A"
-    # 前 N 次故意每次换手机号、固定设备:mock 校验放行 → 200。证明限流按设备而非手机号,同设备共用一个桶
-    for i in range(auth.SMS_LOGIN_MAX_PER_HOUR):
-        r = client.post(
-            "/api/v1/auth/sms/login",
-            json={"phone": f"135331533{i:02d}", "code": "123456", "device_id": device},
-        )
-        assert r.status_code == 200, f"第 {i + 1} 次应放行: {r.text}"
-    # 第 N+1 次:同设备同 IP 超限 → 429(即便又换了个手机号)
-    r = client.post(
-        "/api/v1/auth/sms/login",
-        json={"phone": "13533153399", "code": "123456", "device_id": device},
-    )
-    assert r.status_code == 429, r.text
-    assert "频繁" in r.json()["detail"]
-
-    # 同 IP 换个设备 → 独立计数(限流键含 device_id、非纯 IP/手机号),仍放行
-    r = client.post(
-        "/api/v1/auth/sms/login",
-        json={"phone": "13533153300", "code": "123456", "device_id": "dev-rl-B"},
-    )
-    assert r.status_code == 200, r.text
-
-
 def test_sms_login_bad_code(client) -> None:
     """mock 下校验位数:5 位过 schema 但业务要 6 位 → 400。"""
     phone = "13700137000"

tests/test_test_account.py

@@ -142,54 +142,6 @@ def test_send_does_not_consume_quota(client, enabled, monkeypatch) -> None:
     assert client.post("/api/v1/auth/sms/login", json={"phone": TEST_PHONE, "code": "1234"}).status_code == 429
 
 
-# ============================ 登录限流豁免 ============================
-
-def test_exempt_from_device_ip_login_rate_limit(client, enabled, monkeypatch) -> None:
-    """测试号豁免 (设备+IP) 每小时登录限流。
-
-    普通设备同设备同 IP 到 SMS_LOGIN_MAX_PER_HOUR 次即被拦(见 test_auth.test_sms_login_device_ip_rate_limit);
-    测试号走自己的每日额度、不受这道限流——这里临时打开 RATE_LIMIT_ENABLED,同一设备连登远超该上限仍全 200。
-    回归用:防有人把 enforce_rate_limit 挪到测试账号 early-return 之前而破坏豁免。"""
-    from app.api.v1 import auth
-    from app.core import ratelimit
-
-    monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True)
-    ratelimit._buckets.clear()  # 隔离:清掉跨用例累加的内存计数
-
-    # 同一设备连登 (每小时上限 + 3) 次,远超普通设备会被拦的阈值;测试号豁免 → 全部放行
-    for i in range(auth.SMS_LOGIN_MAX_PER_HOUR + 3):
-        r = client.post(
-            "/api/v1/auth/sms/login",
-            json={"phone": TEST_PHONE, "code": "000000", "device_id": "dev-test-exempt"},
-        )
-        assert r.status_code == 200, (
-            f"测试号第 {i + 1} 次登录应放行(豁免每小时限流),实际 {r.status_code}: {r.text}"
-        )
-
-
-def test_exempt_from_device_ip_send_rate_limit(client, enabled, monkeypatch) -> None:
-    """测试号豁免 (设备+IP) 每小时**发码**限流(与登录那道同样靠早返回豁免)。
-
-    普通设备同设备到 SMS_SEND_MAX_PER_HOUR_PER_DEVICE 次发码即被拦(见 test_auth.test_sms_send_device_ip_rate_limit);
-    测试号发码本就短路(不真发)、且在限流之前早返回 → 同一设备连发远超上限仍全 200。
-    回归用:防有人把发码限流挪到测试账号 early-return 之前而把 QA 联调也风控了。"""
-    from app.api.v1 import auth
-    from app.core import ratelimit
-
-    monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True)
-    monkeypatch.setattr(auth, "SMS_SEND_MAX_PER_HOUR_PER_DEVICE", 2)  # 调小:不豁免的话第 3 次就该被拦
-    ratelimit._buckets.clear()
-
-    for i in range(auth.SMS_SEND_MAX_PER_HOUR_PER_DEVICE + 2):
-        r = client.post(
-            "/api/v1/auth/sms/send",
-            json={"phone": TEST_PHONE, "device_id": "dev-test-send"},
-        )
-        assert r.status_code == 200, (
-            f"测试号第 {i + 1} 次发码应放行(豁免每小时限流),实际 {r.status_code}: {r.text}"
-        )
-
-
 # ============================ 计数单元逻辑 ============================
 
 def test_quota_resets_across_day(enabled, monkeypatch) -> None: