设备首页接入真实状态与音量控制

talkingq-url/handlers/mqtt_handler.py

@@ -1,6 +1,7 @@
 import asyncio
 import json
 import time
+from datetime import datetime
 from typing import Awaitable, Callable, Dict, Optional
 
 import aiomqtt
@@ -9,10 +10,10 @@ from banban.service.binding import BindingService
 from banban.service.device_setting import device_setting_service
 from banban.service.location import location_service
 from config import settings
-from database.models import ChildLocationCurrent
 from services.card_service import card_service
 from services.device_target_cache import device_target_cache
 from services.offline_audio_cache import offline_audio_cache
+from services.task_manager import task_manager
 from utils.logger import session_logger as logger
 
 
@@ -90,14 +91,31 @@ class TalkingQMQTTService:
             logger.error("", "", f"mqtt loop failed: {exc}")
             self._connected = False
 
+    async def _schedule_persistence(self, device_id: str, label: str, coro) -> None:
+        async def _runner():
+            try:
+                await coro
+            except Exception as exc:
+                logger.error(device_id, "mqtt_persistence", f"{label} failed: {exc}", exc_info=True)
+
+        await task_manager.create_task(
+            _runner(),
+            device_id=device_id,
+            task_type="persistence",
+        )
+
     async def _handle_device_info(self, device_id: str, payload: dict):
         data = payload.get("data", {})
-        await device_setting_service.insert_or_update(
-            device_id=device_id,
-            power=data.get("power"),
-            signal_strength=data.get("signal"),
-            version_str=data.get("version"),
-            volume=data.get("voice"),
+        await self._schedule_persistence(
+            device_id,
+            "device_info",
+            device_setting_service.insert_or_update(
+                device_id=device_id,
+                power=data.get("power"),
+                signal_strength=data.get("signal"),
+                version_str=data.get("version"),
+                volume=data.get("voice"),
+            ),
         )
         await self._publish(f"device/{device_id}/event_resp", {"msg_id": "000", "status": "success"})
 
@@ -107,12 +125,31 @@ class TalkingQMQTTService:
             return
 
         data = payload.get("data", {})
-        location = ChildLocationCurrent(
-            device_id=device_id,
-            lat=data.get("latitude"),
-            lon=data.get("longitude"),
+        raw_device_time = data.get("device_time")
+        parsed_device_time = None
+        if isinstance(raw_device_time, str) and raw_device_time.strip():
+            try:
+                parsed_device_time = datetime.fromisoformat(raw_device_time.strip())
+            except ValueError:
+                parsed_device_time = None
+
+        await self._schedule_persistence(
+            device_id,
+            "gps",
+            location_service.report_mqtt_device_location(
+                device_id=device_id,
+                latitude=data.get("latitude"),
+                longitude=data.get("longitude"),
+                coord_type=data.get("coord_type"),
+                accuracy_m=data.get("accuracy_m"),
+                altitude_m=data.get("altitude_m"),
+                speed_mps=data.get("speed_mps"),
+                heading_deg=data.get("heading_deg"),
+                source=data.get("source"),
+                battery_pct=data.get("battery_pct"),
+                device_time=parsed_device_time,
+            ),
         )
-        await location_service.insert_or_update(device_id=device_id, location=location)
 
     async def _handle_volume_response(self, device_id: str, payload: dict):
         if payload.get("status") != "success":
@@ -120,13 +157,36 @@ class TalkingQMQTTService:
             return
 
         data = payload.get("data", {})
-        await device_setting_service.insert_or_update(device_id=device_id, volume=data.get("current_level"))
+        current_level = data.get("current_level")
+        if current_level is None:
+            return
+        await self._schedule_persistence(
+            device_id,
+            "volume",
+            device_setting_service.insert_or_update(
+                device_id=device_id,
+                power=None,
+                volume=current_level,
+                signal_strength=None,
+                version_str=None,
+            ),
+        )
 
     async def _handle_ota_response(self, device_id: str, payload: dict):
         status = payload.get("status")
         data = payload.get("data", {})
         if status == "success":
-            await device_setting_service.insert_or_update(device_id=device_id, version_str=data.get("new_version"))
+            await self._schedule_persistence(
+                device_id,
+                "ota",
+                device_setting_service.insert_or_update(
+                    device_id=device_id,
+                    power=None,
+                    volume=None,
+                    signal_strength=None,
+                    version_str=data.get("new_version"),
+                ),
+            )
         elif status != "accepted":
             logger.warning(device_id, "", f"[OTA] command failed: {payload}")