陈世睿
8adad30ff2
- 新增 device 表 + /api/v1/device/{register,heartbeat} + 迁移 device_table
- heartbeat_monitor_worker 周期扫描心跳超时(App 被杀/无障碍停)→ 服务器终端打印告警
(推送本期未接,先用 logger 终端打印代替真实通知;integrations/jpush.py 已备,后续直接替换)
- config / .env.example 增 JPUSH_* / HEARTBEAT_*
- 见 spec(仓库外 e:\codes\spec\accessibility-liveness-push.md)
注:本提交同时快照了并行 session 未提交的 coupon_state 改动 + 相关 migration
(与本功能同迁移链耦合,无法单独拆分,经确认一并提交)。
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
107 lines
3.2 KiB
Python
107 lines
3.2 KiB
Python
"""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()
|