feat(ota): support forced firmware dispatch

talkingq-url/banban/routers/devices.py

@@ -190,6 +190,7 @@ class DeviceFirmwareStatusResponse(BaseModel):
     firmware_id: int | None = None
     firmware_url: str | None = None
     update_available: bool
+    force_update: bool = False
     can_update: bool
     update_status: str
     progress: float

talkingq-url/banban/service/device.py

@@ -1,3 +1,4 @@
+import re
 from collections.abc import Mapping
 from datetime import datetime, timezone, time, timedelta
 from typing import Any, List, Literal
@@ -398,6 +399,7 @@ class DeviceService(DatabaseServiceBase):
             "current_version": current_version,
             "latest_version": latest_version,
             "update_available": update_available,
+            "force_update": bool(firmware_selection.force_update) if firmware_selection else False,
             "update_status": update_status or "idle",
             "progress": float(progress or 0.0),
             "target_version": target_version,
@@ -408,16 +410,13 @@ class DeviceService(DatabaseServiceBase):
             "firmware_url": firmware_url,
         }
 
-    async def start_firmware_update(
+    async def _dispatch_firmware_update(
         self,
         *,
         device_id: str,
-        user_id: int,
-    ) -> dict[str, Any]:
-        firmware_status = await self.get_firmware_status(device_id=device_id, user_id=user_id)
-        if not firmware_status.get("update_available"):
-            raise HTTPException(status_code=409, detail="device firmware is already up to date")
-
+        firmware_status: Mapping[str, Any],
+        source: str,
+    ) -> dict[str, Any] | None:
         latest_version = firmware_status.get("latest_version")
         if not latest_version:
             raise HTTPException(status_code=404, detail="latest firmware version not configured")
@@ -432,18 +431,41 @@ class DeviceService(DatabaseServiceBase):
         if service is None:
             raise HTTPException(status_code=503, detail="MQTT 服务未初始化")
 
-        await device_firmware_update_manager.update_firmware_update(
-            device_id=device_id,
-            firmware_version=latest_version,
-            update_status="sent",
-            progress=0.0,
-            ota_channel=firmware_status.get("ota_channel"),
-            target_version=latest_version,
-            firmware_url=firmware_url,
-            firmware_id=firmware_status.get("firmware_id"),
-            source="manual",
-        )
-        msg_id = await service.send_ota_command(device_id, firmware_url, latest_version)
+        if source == "forced":
+            claimed = await device_firmware_update_manager.claim_forced_update(
+                device_id=device_id,
+                target_version=latest_version,
+                firmware_url=firmware_url,
+                firmware_id=firmware_status.get("firmware_id"),
+                ota_channel=firmware_status["ota_channel"],
+                retry_seconds=max(300, int(settings.ota_sent_timeout_seconds or 0)),
+            )
+            if not claimed:
+                return None
+        else:
+            saved = await device_firmware_update_manager.update_firmware_update(
+                device_id=device_id,
+                firmware_version=latest_version,
+                update_status="sent",
+                progress=0.0,
+                ota_channel=firmware_status.get("ota_channel"),
+                target_version=latest_version,
+                firmware_url=firmware_url,
+                firmware_id=firmware_status.get("firmware_id"),
+                source=source,
+            )
+            if not saved:
+                raise HTTPException(status_code=503, detail="firmware update status unavailable")
+        try:
+            msg_id = await service.send_ota_command(device_id, firmware_url, latest_version)
+        except Exception:
+            await device_firmware_update_manager.update_firmware_update(
+                device_id=device_id,
+                firmware_version=latest_version,
+                update_status="failed",
+                progress=0.0,
+            )
+            raise
 
         return {
             **firmware_status,
@@ -455,6 +477,66 @@ class DeviceService(DatabaseServiceBase):
             "can_update": bool(firmware_status.get("can_update")),
         }
 
+    async def maybe_dispatch_forced_update(
+        self,
+        *,
+        device_id: str,
+        current_version: str | None,
+        device_type: Any = None,
+        ota_channel: str | None = None,
+    ) -> dict[str, Any] | None:
+        """Send a published forced OTA once a device reports an older version."""
+        if not isinstance(current_version, str) or not re.fullmatch(r"\d+(?:\.\d+)*", current_version.strip()):
+            return None
+
+        status_row = {
+            "device_id": device_id,
+            "version": current_version,
+            "device_type": device_type,
+            "ota_channel": ota_channel,
+        }
+        selection = await ota_firmware_service.select_latest_for_device(status_row)
+        if not selection or not selection.force_update or not selection.firmware_id:
+            return None
+        if not re.fullmatch(r"\d+(?:\.\d+)*", selection.version.strip()):
+            return None
+        if not self._compare_versions(current_version, selection.version):
+            return None
+
+        return await self._dispatch_firmware_update(
+            device_id=device_id,
+            firmware_status={
+                "device_id": device_id,
+                "current_version": current_version,
+                "latest_version": selection.version,
+                "update_available": True,
+                "force_update": True,
+                "can_update": True,
+                "ota_channel": selection.channel_key,
+                "firmware_id": selection.firmware_id,
+                "firmware_url": selection.firmware_url,
+            },
+            source="forced",
+        )
+
+    async def start_firmware_update(
+        self,
+        *,
+        device_id: str,
+        user_id: int,
+    ) -> dict[str, Any]:
+        firmware_status = await self.get_firmware_status(device_id=device_id, user_id=user_id)
+        if not firmware_status.get("update_available"):
+            raise HTTPException(status_code=409, detail="device firmware is already up to date")
+
+        result = await self._dispatch_firmware_update(
+            device_id=device_id,
+            firmware_status=firmware_status,
+            source="manual",
+        )
+        assert result is not None
+        return result
+
 
 # 创建全局 DeviceService 实例
 device_service = DeviceService()

talkingq-url/banban/service/ota_firmware.py

@@ -18,6 +18,7 @@ class OtaFirmwareSelection:
     firmware_url: str
     firmware_id: int | None = None
     source: str = "channel"
+    force_update: bool = False
 
 
 class OtaFirmwareService(DatabaseServiceBase):
@@ -55,7 +56,8 @@ class OtaFirmwareService(DatabaseServiceBase):
                       f.id,
                       f.channel_key,
                       f.version,
-                      f.firmware_url
+                      f.firmware_url,
+                      f.force_update
                     FROM device_ota_firmwares AS f
                     JOIN device_ota_channels AS c
                       ON c.channel_key = f.channel_key
@@ -78,6 +80,7 @@ class OtaFirmwareService(DatabaseServiceBase):
                 version=str(row["version"]),
                 firmware_url=str(row["firmware_url"]),
                 source="channel",
+                force_update=bool(row["force_update"]),
             )
         finally:
             await db_session.close()

talkingq-url/database/init_db.py

@@ -152,6 +152,7 @@ async def _ensure_device_ota_tables(conn) -> None:
               release_status VARCHAR(32) NOT NULL DEFAULT 'draft',
               release_channel VARCHAR(32) NOT NULL DEFAULT 'stable',
               is_latest TINYINT(1) NOT NULL DEFAULT 0,
+              force_update TINYINT(1) NOT NULL DEFAULT 0,
               notes TEXT NULL,
               created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
               published_at DATETIME NULL,
@@ -163,6 +164,11 @@ async def _ensure_device_ota_tables(conn) -> None:
             """
         )
     )
+    await _ensure_columns(
+        conn,
+        table_name="device_ota_firmwares",
+        columns=[("force_update", "force_update TINYINT(1) NOT NULL DEFAULT 0 AFTER is_latest")],
+    )
     await conn.execute(
         text(
             """

talkingq-url/database/models.py

@@ -169,6 +169,7 @@ class DeviceOtaFirmware(Base):
     release_status = Column(String(32), nullable=False, default="draft", index=True)
     release_channel = Column(String(32), nullable=False, default="stable")
     is_latest = Column(Boolean, nullable=False, default=False, index=True)
+    force_update = Column(Boolean, nullable=False, default=False, server_default=text("0"))
     notes = Column(Text, nullable=True)
     created_at = Column(DateTime, nullable=False, server_default=func.now())
     published_at = Column(DateTime, nullable=True)

talkingq-url/handlers/mqtt_handler.py

@@ -172,10 +172,8 @@ class TalkingQMQTTService:
             ),
         )
         
-        await self._schedule_persistence(
-            device_id,
-            "device_info",
-            device_setting_service.insert_or_update(
+        async def persist_info_and_check_forced_ota():
+            await device_setting_service.insert_or_update(
                 device_id=device_id,
                 power=data.get("power"),
                 signal_strength=data.get("signal"),
@@ -183,8 +181,28 @@ class TalkingQMQTTService:
                 volume=data.get("voice"),
                 imei=str(imei).strip() if imei is not None and str(imei).strip() else None,
                 device_type=device_type,
-            ),
-        )
+            )
+            if version is not None:
+                try:
+                    from banban.service.device import device_service
+
+                    setting = await device_setting_service.get_setting_by_device_id(device_id)
+                    forced_update = await device_service.maybe_dispatch_forced_update(
+                        device_id=device_id,
+                        current_version=version,
+                        device_type=setting.get("device_type") if setting else device_type,
+                        ota_channel=setting.get("ota_channel") if setting else data.get("ota_channel"),
+                    )
+                    if forced_update:
+                        logger.info(
+                            device_id,
+                            "ota",
+                            f"已自动下发强制 OTA: target={forced_update.get('target_version')}, msg_id={forced_update.get('msg_id')}",
+                        )
+                except Exception as exc:
+                    logger.error(device_id, "ota", f"强制 OTA 自动下发失败: {exc}")
+
+        await self._schedule_persistence(device_id, "device_info", persist_info_and_check_forced_ota())
         wechat_mp_notification_service.schedule_low_battery_notification(
             device_id=device_id,
             power=power,
@@ -854,10 +872,12 @@ class TalkingQMQTTService:
         return state
 
     async def send_ota_command(self, device_id: str, url: str, version: str) -> str:
-        await self._publish(
+        published = await self._publish(
             f"device/{device_id}/command",
             {"msg_id": "003", "params": {"url": url, "version": version}},
         )
+        if not published:
+            raise RuntimeError(f"OTA command publish failed: {device_id}")
         return "003"
 
     async def send_nfc_notice(self, device_id: str, url: str) -> str:

talkingq-url/services/device_update_manager.py

@@ -1,8 +1,10 @@
+from datetime import datetime, timedelta, timezone
 from typing import Dict, Optional
 import time
 from collections.abc import Mapping
 
 from sqlalchemy import select, update, insert, delete, text
+from sqlalchemy.exc import IntegrityError
 from database.models import DeviceFirmwareUpdate
 from services.database_service_base import DatabaseServiceBase
 from utils.logger import session_logger
@@ -44,6 +46,80 @@ class DeviceFirmwareUpdateManager(DatabaseServiceBase):
             "updated_at": row.updated_at,
         }
 
+    @staticmethod
+    def _can_claim_forced_update(record, target_version: str, retry_seconds: int) -> bool:
+        if record is None:
+            return True
+        status = str(record.update_status or "").strip().lower()
+        target = record.target_version or record.firmware_version
+        if target != target_version:
+            return True
+        if status in {"sent", "accepted", "updating"}:
+            return False
+        updated_at = record.updated_at
+        if not isinstance(updated_at, datetime):
+            return False
+        now = datetime.now(timezone.utc) if updated_at.tzinfo else datetime.now()
+        return now - updated_at >= timedelta(seconds=retry_seconds)
+
+    async def claim_forced_update(
+        self,
+        *,
+        device_id: str,
+        target_version: str,
+        firmware_url: str,
+        firmware_id: int | None,
+        ota_channel: str,
+        retry_seconds: int,
+    ) -> bool:
+        """Atomically reserve one MQTT send per device and retry window."""
+        db_session = await self.get_session()
+        try:
+            result = await db_session.execute(
+                select(DeviceFirmwareUpdate)
+                .where(DeviceFirmwareUpdate.device_id == device_id)
+                .with_for_update()
+            )
+            record = result.scalar_one_or_none()
+            if not self._can_claim_forced_update(record, target_version, retry_seconds):
+                return False
+
+            values = {
+                "firmware_version": target_version,
+                "target_version": target_version,
+                "firmware_url": firmware_url,
+                "firmware_id": firmware_id,
+                "ota_channel": ota_channel,
+                "update_status": "sent",
+                "progress": 0.0,
+                "source": "forced",
+            }
+            if record:
+                await db_session.execute(
+                    update(DeviceFirmwareUpdate)
+                    .where(DeviceFirmwareUpdate.device_id == device_id)
+                    .values(**values)
+                )
+            else:
+                serial_number = await self._get_serial_number(db_session, device_id)
+                await db_session.execute(
+                    insert(DeviceFirmwareUpdate).values(
+                        device_id=device_id, serial_number=serial_number, **values
+                    )
+                )
+            await db_session.commit()
+            self._clear_cache(device_id)
+            return True
+        except IntegrityError:
+            await db_session.rollback()
+            return False
+        except Exception as exc:
+            await db_session.rollback()
+            session_logger.error(device_id, "firmware_update", f"强制 OTA 抢占发送资格失败: {exc}")
+            return False
+        finally:
+            await db_session.close()
+
     async def create_firmware_update(
         self,
         device_id: str,

talkingq-url/tests/test_forced_ota.py

@@ -0,0 +1,159 @@
+from datetime import datetime
+
+import pytest
+
+from banban.service.device import DeviceService
+from banban.service.ota_firmware import OtaFirmwareSelection
+from handlers.mqtt_handler import TalkingQMQTTService
+
+
+@pytest.mark.asyncio
+async def test_simulated_device_force_ota_dispatches_once_and_completes(monkeypatch):
+    device_id = "TalkingQ_XQSN00001005"
+    service = DeviceService()
+    mqtt_service = TalkingQMQTTService({"broker": "127.0.0.1"})
+    record = None
+    ota_commands = []
+    published = []
+
+    async def fake_select_latest_for_device(status_row):
+        assert status_row["device_type"] == 2
+        return OtaFirmwareSelection(
+            channel_key="gps",
+            version="1.8.5",
+            firmware_url="https://example.com/gps-1.8.5.bin",
+            firmware_id=15,
+            force_update=True,
+        )
+
+    async def fake_get_firmware_update_dict(requested_device_id):
+        assert requested_device_id == device_id
+        return dict(record) if record else None
+
+    async def fake_update_firmware_update(*args, **kwargs):
+        nonlocal record
+        if args:
+            kwargs = {"device_id": args[0], **kwargs}
+        record = dict(record or {})
+        record.update(kwargs)
+        record.setdefault("device_id", device_id)
+        record.setdefault("updated_at", datetime.now())
+        return True
+
+    async def fake_claim_forced_update(**kwargs):
+        nonlocal record
+        if record and record.get("update_status") in {"sent", "accepted", "updating"}:
+            return False
+        record = dict(record or {})
+        record.update(
+            {
+                "device_id": device_id,
+                "firmware_version": kwargs["target_version"],
+                "target_version": kwargs["target_version"],
+                "firmware_url": kwargs["firmware_url"],
+                "firmware_id": kwargs["firmware_id"],
+                "ota_channel": kwargs["ota_channel"],
+                "update_status": "sent",
+                "progress": 0.0,
+                "source": "forced",
+                "updated_at": datetime.now(),
+            }
+        )
+        return True
+
+    class FakeMqttService:
+        async def send_ota_command(self, requested_device_id, url, version):
+            ota_commands.append({"device_id": requested_device_id, "url": url, "version": version})
+            return "003"
+
+    async def fake_get_instance():
+        return FakeMqttService()
+
+    async def fake_publish(topic, payload):
+        published.append((topic, payload))
+        return True
+
+    async def fake_schedule_persistence(requested_device_id, label, coro):
+        assert requested_device_id == device_id
+        await coro
+
+    async def fake_insert_or_update(**kwargs):
+        assert kwargs["device_id"] == device_id
+
+    async def fake_get_setting_by_device_id(requested_device_id):
+        return {"device_id": requested_device_id, "device_type": 2, "ota_channel": None}
+
+    monkeypatch.setattr(
+        "banban.service.device.ota_firmware_service.select_latest_for_device",
+        fake_select_latest_for_device,
+    )
+    monkeypatch.setattr(
+        "banban.service.device.device_firmware_update_manager.get_firmware_update_dict",
+        fake_get_firmware_update_dict,
+    )
+    monkeypatch.setattr(
+        "banban.service.device.device_firmware_update_manager.update_firmware_update",
+        fake_update_firmware_update,
+    )
+    monkeypatch.setattr(
+        "banban.service.device.device_firmware_update_manager.claim_forced_update",
+        fake_claim_forced_update,
+    )
+    monkeypatch.setattr("handlers.mqtt_handler.TalkingQMQTTService.get_instance", fake_get_instance)
+    monkeypatch.setattr(mqtt_service, "_publish", fake_publish)
+    monkeypatch.setattr(mqtt_service, "_schedule_persistence", fake_schedule_persistence)
+    monkeypatch.setattr("handlers.mqtt_handler.device_setting_service.insert_or_update", fake_insert_or_update)
+    monkeypatch.setattr("handlers.mqtt_handler.device_setting_service.get_setting_by_device_id", fake_get_setting_by_device_id)
+    monkeypatch.setattr(
+        "handlers.mqtt_handler.wechat_mp_notification_service.schedule_low_battery_notification",
+        lambda **kwargs: None,
+    )
+
+    simulated_device_info = {
+        "msg_id": "000",
+        "data": {
+            "id": device_id,
+            "version": "1.8.2",
+            "device_type": "2",
+            "power": 80,
+            "signal": 4,
+        },
+    }
+    await mqtt_service._handle_device_info(device_id, simulated_device_info)
+    await mqtt_service._handle_device_info(device_id, simulated_device_info)
+
+    assert ota_commands == [{
+        "device_id": device_id,
+        "url": "https://example.com/gps-1.8.5.bin",
+        "version": "1.8.5",
+    }]
+    assert record["update_status"] == "sent"
+    assert record["target_version"] == "1.8.5"
+    assert record["source"] == "forced"
+    assert published == [
+        (f"device/{device_id}/event_resp", {"msg_id": "000", "status": "success"}),
+        (f"device/{device_id}/event_resp", {"msg_id": "000", "status": "success"}),
+    ]
+
+    await mqtt_service._handle_ota_response(
+        device_id,
+        {
+            "status": "accepted",
+            "data": {"current_version": "1.8.2", "target_version": "1.8.5"},
+        },
+    )
+    assert record["update_status"] == "accepted"
+
+    await mqtt_service._handle_ota_response(
+        device_id,
+        {"status": "updating", "data": {"version": "1.8.5", "progress": 65}},
+    )
+    assert record["update_status"] == "updating"
+    assert record["progress"] == 65.0
+
+    await mqtt_service._handle_ota_response(
+        device_id,
+        {"status": "success", "data": {"new_version": "1.8.5"}},
+    )
+    assert record["update_status"] == "success"
+    assert record["progress"] == 100.0

talkingq-url/tests/test_mqtt_device_info_report.py

@@ -24,10 +24,22 @@ async def test_device_info_report_persists_metadata_and_periodic_location(monkey
         calls.append(("location", kwargs))
         return {"device_id": kwargs["device_id"], "lat": kwargs["latitude"], "lng": kwargs["longitude"]}
 
+    async def fake_maybe_dispatch_forced_update(**kwargs):
+        del kwargs
+        return None
+
+    async def fake_get_setting_by_device_id(device_id):
+        return {"device_id": device_id, "device_type": 2, "ota_channel": None}
+
     monkeypatch.setattr(service, "_publish", fake_publish)
     monkeypatch.setattr(service, "_schedule_persistence", fake_schedule_persistence)
     monkeypatch.setattr("handlers.mqtt_handler.device_setting_service.insert_or_update", fake_insert_or_update)
+    monkeypatch.setattr("handlers.mqtt_handler.device_setting_service.get_setting_by_device_id", fake_get_setting_by_device_id)
     monkeypatch.setattr("handlers.mqtt_handler.location_service.report_mqtt_device_location", fake_report_mqtt_device_location)
+    monkeypatch.setattr(
+        "banban.service.device.device_service.maybe_dispatch_forced_update",
+        fake_maybe_dispatch_forced_update,
+    )
     monkeypatch.setattr(
         "handlers.mqtt_handler.wechat_mp_notification_service.schedule_low_battery_notification",
         lambda **kwargs: None,
@@ -103,9 +115,21 @@ async def test_device_info_report_without_location_does_not_write_location(monke
     async def fake_insert_or_update(**kwargs):
         calls.append(("settings", kwargs["device_type"]))
 
+    async def fake_maybe_dispatch_forced_update(**kwargs):
+        del kwargs
+        return None
+
+    async def fake_get_setting_by_device_id(device_id):
+        return {"device_id": device_id, "device_type": 1, "ota_channel": None}
+
     monkeypatch.setattr(service, "_publish", fake_publish)
     monkeypatch.setattr(service, "_schedule_persistence", fake_schedule_persistence)
     monkeypatch.setattr("handlers.mqtt_handler.device_setting_service.insert_or_update", fake_insert_or_update)
+    monkeypatch.setattr("handlers.mqtt_handler.device_setting_service.get_setting_by_device_id", fake_get_setting_by_device_id)
+    monkeypatch.setattr(
+        "banban.service.device.device_service.maybe_dispatch_forced_update",
+        fake_maybe_dispatch_forced_update,
+    )
     monkeypatch.setattr(
         "handlers.mqtt_handler.wechat_mp_notification_service.schedule_low_battery_notification",
         lambda **kwargs: None,

talkingq-url/tests/test_ota_channel_selection.py

@@ -28,6 +28,7 @@ async def test_firmware_status_selects_gps_channel(monkeypatch):
             version="1.1.0",
             firmware_url="https://example.com/gps-1.1.0.bin",
             firmware_id=10,
+            force_update=True,
         )
 
     async def fake_get_firmware_update_dict(device_id):
@@ -44,6 +45,7 @@ async def test_firmware_status_selects_gps_channel(monkeypatch):
     assert result["firmware_id"] == 10
     assert result["latest_version"] == "1.1.0"
     assert result["firmware_url"] == "https://example.com/gps-1.1.0.bin"
+    assert result["force_update"] is True
     assert result["update_available"] is True
     assert result["can_update"] is True