shaguabijia-app-server/app/repositories/device.py

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


def _get(db: Session, *, user_id: int, device_id: str) -> DeviceLiveness | None:
    stmt = select(DeviceLiveness).where(
        DeviceLiveness.user_id == user_id, DeviceLiveness.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,
) -> DeviceLiveness:
    """注册设备或更新其 registration_id / 元信息。upsert by (user_id, device_id)。"""
    device = _get(db, user_id=user_id, device_id=device_id)
    if device is None:
        device = DeviceLiveness(
            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,
) -> DeviceLiveness:
    """处理一次心跳(心跳也能自注册)。

    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 = DeviceLiveness(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:
        if not device.ever_protected:
            device.first_protected_at = now  # 首次开无障碍记一次,后续心跳不覆盖
        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[DeviceLiveness]:
    """掉线设备:曾经保护过、当前 alive、心跳超时。

    本期只做终端打印检测、不推送 → 不再要求有 registration_id(没接极光 token 的设备也要检出)。
    """
    cutoff = datetime.now(timezone.utc) - timedelta(minutes=timeout_minutes)
    stmt = select(DeviceLiveness).where(
        DeviceLiveness.ever_protected.is_(True),
        DeviceLiveness.liveness_state == "alive",
        DeviceLiveness.last_heartbeat_at.is_not(None),
        DeviceLiveness.last_heartbeat_at < cutoff,
    )
    return list(db.execute(stmt).scalars().all())


def mark_notified(db: Session, *, device_id_pk: int) -> None:
    """标记掉线已检出(状态机进入 notified,避免每轮重复处理)+ 置「待客户端提醒」标记。

    kill_alert_pending 与 state 解耦(见 spec accessibility-liveness-pull-prompt.md):
    心跳恢复(touch_heartbeat)只重置 state、不动此标记,故客户端下次进 App 必能拉到这次掉线。
    """
    device = db.get(DeviceLiveness, device_id_pk)
    if device is not None:
        device.liveness_state = "notified"
        device.notified_at = datetime.now(timezone.utc)
        device.kill_alert_pending = True
        db.commit()


def get_device(db: Session, *, user_id: int, device_id: str) -> DeviceLiveness | None:
    """按 (user_id, device_id) 取设备(liveness 查询用)。"""
    return _get(db, user_id=user_id, device_id=device_id)


def ack_kill_alert(db: Session, *, user_id: int, device_id: str) -> None:
    """客户端已弹过「开启自启动」引导 → 清「待提醒」标记(幂等; 下次真掉线 worker 再置)。"""
    device = _get(db, user_id=user_id, device_id=device_id)
    if device is not None and device.kill_alert_pending:
        device.kill_alert_pending = False
        db.commit()