1b4c47f77a
- 绑定模型新增 owner_user_id，支持 child_id 为空\n- 新增绑定后设置儿童接口 /bindings/{device_id}/child\n- 前端绑定页改为先绑设备后补充儿童\n- 补充数据库迁移脚本与 DAO 测试用例
52 lines
2.2 KiB
Python
52 lines
2.2 KiB
Python
from collections.abc import Mapping
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
from app.dao.binding import BindingDAO
|
|
|
|
|
|
class BindingService:
|
|
def __init__(self, db):
|
|
self.dao = BindingDAO(db)
|
|
|
|
def start_bind(self, user_id: int, device_id: str, child_id: int | None = None) -> tuple[str, datetime]:
|
|
bind_token = self.dao.start_bind(user_id, device_id, child_id)
|
|
return bind_token, datetime.utcnow()
|
|
|
|
def confirm_bind(self, bind_token: str, user_id: int) -> Mapping:
|
|
session = self.dao.get_session(bind_token, user_id)
|
|
if not session:
|
|
raise ValueError("Bind session not found")
|
|
if datetime.utcnow() > session["expires_at"]:
|
|
raise ValueError("Bind session expired")
|
|
if session["status"] != 1:
|
|
raise ValueError("Bind session already processed")
|
|
|
|
self.dao.confirm_bind(session["id"], session["device_id"], session["target_child_id"], user_id)
|
|
return {"device_id": session["device_id"], "child_id": session["target_child_id"]}
|
|
|
|
def get_binding(self, device_id: str, user_id: int) -> Optional[Mapping]:
|
|
return self.dao.get_by_device(device_id, user_id)
|
|
|
|
def get_current_binding(self, user_id: int) -> Optional[Mapping]:
|
|
return self.dao.get_current_by_user(user_id)
|
|
|
|
def direct_bind(self, device_id: str, child_id: int | None, user_id: int) -> Mapping:
|
|
self.dao.direct_bind(device_id, child_id, user_id)
|
|
return {"device_id": device_id, "child_id": child_id}
|
|
|
|
def set_binding_child(self, device_id: str, child_id: int, user_id: int) -> Mapping:
|
|
ok = self.dao.set_binding_child(device_id=device_id, child_id=child_id, user_id=user_id)
|
|
if not ok:
|
|
raise ValueError("binding not found")
|
|
return {"device_id": device_id, "child_id": child_id}
|
|
|
|
def unbind(self, device_id: str, user_id: int) -> bool:
|
|
return self.dao.unbind(device_id, user_id)
|
|
|
|
def list_history(self, device_id: str, limit: int = 20, cursor: datetime = None) -> tuple[list, bool]:
|
|
rows = self.dao.list_history(device_id, limit, cursor)
|
|
has_more = len(rows) > limit
|
|
rows = rows[:limit]
|
|
return rows, has_more
|