Andrey Lutsenko
3d36faa49d
При повторной привязке Seaca переиспользуется запись device_uuid вместо INSERT.
Добавлены POST /devices/{id}/revoke и DELETE для полного удаления записи в настройках SAC.
105 lines
3.2 KiB
Python
105 lines
3.2 KiB
Python
"""Управление зарегистрированными устройствами Seaca."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.mobile_device import MobileDevice
|
|
from app.models.user import User
|
|
from app.services.mobile_tokens import revoke_refresh_tokens_for_device
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MobileDeviceSummary:
|
|
id: int
|
|
user_id: int
|
|
username: str
|
|
device_uuid: str
|
|
display_name: str
|
|
platform: str
|
|
app_version: str | None
|
|
has_fcm_token: bool
|
|
enrolled_at: datetime
|
|
last_seen_at: datetime | None
|
|
revoked_at: datetime | None
|
|
is_active: bool
|
|
|
|
|
|
def list_mobile_devices(db: Session) -> list[MobileDeviceSummary]:
|
|
rows = db.scalars(select(MobileDevice).order_by(MobileDevice.enrolled_at.desc())).all()
|
|
user_ids = {r.user_id for r in rows}
|
|
usernames: dict[int, str] = {}
|
|
if user_ids:
|
|
for user in db.scalars(select(User).where(User.id.in_(user_ids))).all():
|
|
usernames[user.id] = user.username
|
|
return [_to_summary(row, usernames.get(row.user_id, "?")) for row in rows]
|
|
|
|
|
|
def _to_summary(row: MobileDevice, username: str) -> MobileDeviceSummary:
|
|
return MobileDeviceSummary(
|
|
id=row.id,
|
|
user_id=row.user_id,
|
|
username=username,
|
|
device_uuid=row.device_uuid,
|
|
display_name=row.display_name or "",
|
|
platform=row.platform or "android",
|
|
app_version=row.app_version,
|
|
has_fcm_token=bool((row.fcm_token or "").strip()),
|
|
enrolled_at=row.enrolled_at,
|
|
last_seen_at=row.last_seen_at,
|
|
revoked_at=row.revoked_at,
|
|
is_active=row.revoked_at is None,
|
|
)
|
|
|
|
|
|
def get_device_or_raise(db: Session, device_id: int) -> MobileDevice:
|
|
row = db.get(MobileDevice, device_id)
|
|
if row is None:
|
|
raise ValueError("device not found")
|
|
return row
|
|
|
|
|
|
def assert_device_active(device: MobileDevice) -> None:
|
|
if device.revoked_at is not None:
|
|
raise PermissionError("device has been revoked")
|
|
|
|
|
|
def touch_device(db: Session, device: MobileDevice, *, commit: bool = False) -> None:
|
|
device.last_seen_at = datetime.now(timezone.utc)
|
|
if commit:
|
|
db.commit()
|
|
|
|
|
|
def revoke_device(db: Session, device_id: int) -> MobileDeviceSummary:
|
|
device = get_device_or_raise(db, device_id)
|
|
if device.revoked_at is None:
|
|
device.revoked_at = datetime.now(timezone.utc)
|
|
revoke_refresh_tokens_for_device(db, device.id)
|
|
db.commit()
|
|
db.refresh(device)
|
|
user = db.get(User, device.user_id)
|
|
username = user.username if user else "?"
|
|
return _to_summary(device, username)
|
|
|
|
|
|
def delete_device_record(db: Session, device_id: int) -> None:
|
|
device = get_device_or_raise(db, device_id)
|
|
revoke_refresh_tokens_for_device(db, device.id)
|
|
db.delete(device)
|
|
db.commit()
|
|
|
|
|
|
def update_fcm_token(db: Session, device: MobileDevice, fcm_token: str) -> None:
|
|
cleaned = (fcm_token or "").strip()
|
|
if not cleaned:
|
|
raise ValueError("fcm_token is required")
|
|
now = datetime.now(timezone.utc)
|
|
device.fcm_token = cleaned
|
|
device.fcm_token_updated_at = now
|
|
device.last_seen_at = now
|
|
db.commit()
|