小程序后端接入 talkingq 共享库并新增消息与定位能力

mini-program/app/service/binding.py

@@ -5,11 +5,33 @@ from typing import Optional
 from app.dao.binding import BindingDAO
 
 
+class BindingError(ValueError):
+    def __init__(self, message: str, status_code: int = 400) -> None:
+        super().__init__(message)
+        self.status_code = status_code
+
+
 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]:
+    def _ensure_bindable_device(self, device_id: str, serial_number: str) -> None:
+        row = self.dao.get_device_auth(device_id)
+        if row is None:
+            raise BindingError("device not found in device_auth", status_code=404)
+        if str(row["serial_number"]) != serial_number:
+            raise BindingError("serial_number does not match device_id", status_code=400)
+        if int(row["is_active"]) != 1:
+            raise BindingError("device is inactive", status_code=400)
+
+    def start_bind(
+        self,
+        user_id: int,
+        device_id: str,
+        serial_number: str,
+        child_id: int | None = None,
+    ) -> tuple[str, datetime]:
+        self._ensure_bindable_device(device_id, serial_number)
         bind_token = self.dao.start_bind(user_id, device_id, child_id)
         return bind_token, datetime.utcnow()
 
@@ -37,7 +59,14 @@ class BindingService:
         rows = rows[:limit]
         return rows, has_more
 
-    def direct_bind(self, device_id: str, child_id: int | None, user_id: int) -> Mapping:
+    def direct_bind(
+        self,
+        device_id: str,
+        serial_number: str,
+        child_id: int | None,
+        user_id: int,
+    ) -> Mapping:
+        self._ensure_bindable_device(device_id, serial_number)
         self.dao.direct_bind(device_id, child_id, user_id)
         return {"device_id": device_id, "child_id": child_id}