告警时主动刷新定位地址

2026-06-08 14:10:34 +08:00
parent ac9d669107
commit 32b91936ad
5 changed files with 206 additions and 11 deletions
--- a/talkingq-url/banban/dao/device_alarm.py
+++ b/talkingq-url/banban/dao/device_alarm.py
@@ -59,6 +59,39 @@ class DeviceAlarmDAO(BaseDAO):
        await self.commit()
        return int(result.lastrowid)

+    async def get_by_id(self, *, alarm_id: int, device_id: str | None = None) -> Mapping[str, Any] | None:
+        where_device = "AND device_id = :device_id" if device_id is not None else ""
+        params: dict[str, Any] = {"alarm_id": alarm_id}
+        if device_id is not None:
+            params["device_id"] = device_id
+
+        result = await self.execute(
+            text(
+                f"""
+                SELECT
+                  alarm_id,
+                  device_id,
+                  owner_user_id,
+                  child_id,
+                  source_msg_id,
+                  coord_type,
+                  lat,
+                  lng,
+                  location_updated_at,
+                  address,
+                  address_resolved_at,
+                  address_resolve_status,
+                  created_at
+                FROM device_alarm_events
+                WHERE alarm_id = :alarm_id
+                  {where_device}
+                LIMIT 1
+                """
+            ),
+            params,
+        )
+        return result.mappings().first()
+
    async def get_current_location_snapshot(self, *, child_id: int) -> Mapping[str, Any] | None:
        result = await self.execute(
            text(
@@ -103,6 +136,40 @@ class DeviceAlarmDAO(BaseDAO):
        await self.commit()
        return bool(result.rowcount)

+    async def update_location_snapshot(
+        self,
+        *,
+        alarm_id: int,
+        location: Mapping[str, Any],
+    ) -> bool:
+        lat = location.get("lat")
+        lng = location.get("lng")
+        result = await self.execute(
+            text(
+                """
+                UPDATE device_alarm_events
+                SET coord_type = :coord_type,
+                    lat = :lat,
+                    lng = :lng,
+                    location_updated_at = :location_updated_at,
+                    address = NULL,
+                    address_resolved_at = NULL,
+                    address_resolve_status = :address_resolve_status
+                WHERE alarm_id = :alarm_id
+                """
+            ),
+            {
+                "alarm_id": alarm_id,
+                "coord_type": location.get("coord_type"),
+                "lat": lat,
+                "lng": lng,
+                "location_updated_at": location.get("updated_at"),
+                "address_resolve_status": 0 if lat is not None and lng is not None else None,
+            },
+        )
+        await self.commit()
+        return bool(result.rowcount)
+
    async def get_active_binding_context(self, *, device_id: str) -> Mapping[str, Any] | None:
        result = await self.execute(
            text(
--- a/talkingq-url/banban/routers/devices.py
+++ b/talkingq-url/banban/routers/devices.py
@@ -342,8 +342,7 @@ def _row_to_alarm_item(row: Mapping) -> DeviceAlarmItem:
        location_age_seconds = _seconds_between(created_at, location_updated_at)
        location_stale = (
            location_age_seconds is None
-            or location_age_seconds < 0
-            or location_age_seconds > settings.alarm_location_stale_seconds
+            or abs(location_age_seconds) > settings.alarm_location_stale_seconds
        )

    return DeviceAlarmItem(
--- a/talkingq-url/banban/service/device_alarm.py
+++ b/talkingq-url/banban/service/device_alarm.py
@@ -17,7 +17,13 @@ class DeviceAlarmService(DatabaseServiceBase):
    def __init__(self):
        super().__init__(service_name="device_alarm_service")

-    async def record_alarm_event(self, *, device_id: str, source_msg_id: str = "010") -> int | None:
+    async def record_alarm_event(
+        self,
+        *,
+        device_id: str,
+        source_msg_id: str = "010",
+        resolve_address: bool = True,
+    ) -> int | None:
        db_session = await self.get_session()
        try:
            dao = DeviceAlarmDAO(db_session)
@@ -36,7 +42,7 @@ class DeviceAlarmService(DatabaseServiceBase):
                source_msg_id=source_msg_id,
                location=location,
            )
-            if alarm_id and location and location.get("lat") is not None and location.get("lng") is not None:
+            if resolve_address and alarm_id and location and location.get("lat") is not None and location.get("lng") is not None:
                await task_manager.create_task(
                    self.resolve_alarm_address(
                        alarm_id=alarm_id,
@@ -51,6 +57,76 @@ class DeviceAlarmService(DatabaseServiceBase):
        finally:
            await db_session.close()

+    async def get_alarm_event(
+        self,
+        *,
+        alarm_id: int,
+        device_id: str | None = None,
+    ) -> Mapping[str, Any] | None:
+        db_session = await self.get_session()
+        try:
+            dao = DeviceAlarmDAO(db_session)
+            return await dao.get_by_id(alarm_id=alarm_id, device_id=device_id)
+        finally:
+            await db_session.close()
+
+    async def update_alarm_location(
+        self,
+        *,
+        alarm_id: int,
+        location: Mapping[str, Any],
+    ) -> bool:
+        db_session = await self.get_session()
+        try:
+            dao = DeviceAlarmDAO(db_session)
+            return await dao.update_location_snapshot(alarm_id=alarm_id, location=location)
+        finally:
+            await db_session.close()
+
+    async def apply_alarm_location(
+        self,
+        *,
+        alarm_id: int,
+        device_id: str,
+        location: Mapping[str, Any],
+    ) -> bool:
+        if location.get("lat") is None or location.get("lng") is None:
+            return False
+
+        updated = await self.update_alarm_location(alarm_id=alarm_id, location=location)
+        if not updated:
+            return False
+
+        address = str(location.get("address") or "").strip()
+        if address:
+            await self.update_alarm_address(
+                alarm_id=alarm_id,
+                address=address,
+                status=ADDRESS_RESOLVE_SUCCESS,
+            )
+            return True
+
+        await self.resolve_alarm_address(
+            alarm_id=alarm_id,
+            device_id=device_id,
+            lat=float(location["lat"]),
+            lng=float(location["lng"]),
+        )
+        return True
+
+    async def resolve_alarm_current_location(self, *, alarm_id: int, device_id: str) -> bool:
+        alarm = await self.get_alarm_event(alarm_id=alarm_id, device_id=device_id)
+        if not alarm or alarm.get("lat") is None or alarm.get("lng") is None:
+            return False
+
+        await self.resolve_alarm_address(
+            alarm_id=alarm_id,
+            device_id=device_id,
+            lat=float(alarm["lat"]),
+            lng=float(alarm["lng"]),
+        )
+        return True
+
    async def resolve_alarm_address(
        self,
        *,
--- a/talkingq-url/config.py
+++ b/talkingq-url/config.py
@@ -48,6 +48,7 @@ class Settings(BaseSettings):
    )
    amap_http_timeout_seconds: float = Field(default=5.0, validation_alias="AMAP_HTTP_TIMEOUT_SECONDS")
    alarm_location_stale_seconds: int = Field(default=600, validation_alias="ALARM_LOCATION_STALE_SECONDS")
+    alarm_gps_query_timeout_seconds: float = Field(default=5.0, validation_alias="ALARM_GPS_QUERY_TIMEOUT_SECONDS")

    aliyun_api_key: str = Field(default="", validation_alias="ALIYUN_API_KEY")
    aliyun_vocabulary_id: str = Field(default="", validation_alias="ALIYUN_VOCABULARY_ID")
--- a/talkingq-url/handlers/mqtt_handler.py
+++ b/talkingq-url/handlers/mqtt_handler.py
@@ -51,6 +51,8 @@ class TalkingQMQTTService:
        self._volume_command_lock = asyncio.Lock()
        self._volume_command_states: Dict[tuple[str, int], dict] = {}
        self._last_volume_command_time_ms = 0
+        self._gps_waiter_lock = asyncio.Lock()
+        self._gps_waiters: Dict[str, list[asyncio.Future]] = {}

        self._msg_handlers: Dict[str, Callable[[str, dict], Awaitable[None]]] = {
            "000": self._handle_device_info,
@@ -189,10 +191,8 @@ class TalkingQMQTTService:
        lon = data.get("longitude")
        logger.info(device_id, "", f"[GPS] 设备 {device_id} 位置: 纬度={lat}, 经度={lon}")
        
-        await self._schedule_persistence(
-            device_id,
-            "gps",
-            location_service.report_mqtt_device_location(
+        async def _persist_gps_and_notify():
+            row = await location_service.report_mqtt_device_location(
                device_id=device_id,
                latitude=data.get("latitude"),
                longitude=data.get("longitude"),
@@ -204,8 +204,10 @@ class TalkingQMQTTService:
                source=data.get("source"),
                battery_pct=data.get("battery_pct"),
                device_time=parsed_device_time,
-            ),
-        )
+            )
+            await self._notify_gps_waiters(device_id, row)
+
+        await self._schedule_persistence(device_id, "gps", _persist_gps_and_notify())

    async def _handle_volume_response(self, device_id: str, payload: dict):
        command_time = self._parse_command_time(payload.get("time"))
@@ -399,7 +401,27 @@ class TalkingQMQTTService:
    async def _handle_alarm_report(self, device_id: str, payload: dict):
        logger.info(device_id, "", f"[告警] 设备 {device_id} 发送紧急报警")
        async def _record_alarm_and_notify():
-            alarm_id = await device_alarm_service.record_alarm_event(device_id=device_id, source_msg_id="010")
+            alarm_id = await device_alarm_service.record_alarm_event(
+                device_id=device_id,
+                source_msg_id="010",
+                resolve_address=False,
+            )
+            if alarm_id:
+                location = await self.query_gps_and_wait_for_location(
+                    device_id,
+                    timeout_seconds=settings.alarm_gps_query_timeout_seconds,
+                )
+                if location and location.get("lat") is not None and location.get("lng") is not None:
+                    await device_alarm_service.apply_alarm_location(
+                        alarm_id=alarm_id,
+                        device_id=device_id,
+                        location=location,
+                    )
+                else:
+                    await device_alarm_service.resolve_alarm_current_location(
+                        alarm_id=alarm_id,
+                        device_id=device_id,
+                    )
            sms_notification_service.schedule_alarm_notification(device_id=device_id, alarm_id=alarm_id)

        await self._schedule_persistence(device_id, "alarm_event", _record_alarm_and_notify())
@@ -647,6 +669,36 @@ class TalkingQMQTTService:
        await self._publish(f"device/{device_id}/command", {"msg_id": "001"})
        return "001"

+    async def query_gps_and_wait_for_location(self, device_id: str, *, timeout_seconds: float) -> dict | None:
+        timeout = max(0.1, float(timeout_seconds or 0))
+        loop = asyncio.get_running_loop()
+        future = loop.create_future()
+        async with self._gps_waiter_lock:
+            self._gps_waiters.setdefault(device_id, []).append(future)
+
+        try:
+            await self.send_gps_query(device_id)
+            return await asyncio.wait_for(future, timeout=timeout)
+        except asyncio.TimeoutError:
+            return None
+        finally:
+            async with self._gps_waiter_lock:
+                waiters = self._gps_waiters.get(device_id)
+                if waiters and future in waiters:
+                    waiters.remove(future)
+                if waiters == []:
+                    self._gps_waiters.pop(device_id, None)
+
+    async def _notify_gps_waiters(self, device_id: str, row: dict | None) -> None:
+        if row is None:
+            return
+        async with self._gps_waiter_lock:
+            waiters = self._gps_waiters.pop(device_id, [])
+
+        for future in waiters:
+            if not future.done():
+                future.set_result(dict(row))
+
    def _parse_command_time(self, value) -> int | None:
        if value is None:
            return None