From e5ba92ba138a5cdbc988ac1d24bf28dcfc0daf2b Mon Sep 17 00:00:00 2001 From: stu2not Date: Tue, 22 Sep 2026 14:58:53 +0800 Subject: [PATCH] feat(ota): support forced firmware dispatch --- talkingq-url/banban/routers/devices.py | 1 + talkingq-url/banban/service/device.py | 120 ++++++++++--- talkingq-url/banban/service/ota_firmware.py | 5 +- talkingq-url/database/init_db.py | 6 + talkingq-url/database/models.py | 1 + talkingq-url/handlers/mqtt_handler.py | 34 +++- .../services/device_update_manager.py | 76 +++++++++ talkingq-url/tests/test_forced_ota.py | 159 ++++++++++++++++++ .../tests/test_mqtt_device_info_report.py | 24 +++ .../tests/test_ota_channel_selection.py | 2 + 10 files changed, 401 insertions(+), 27 deletions(-) create mode 100644 talkingq-url/tests/test_forced_ota.py diff --git a/talkingq-url/banban/routers/devices.py b/talkingq-url/banban/routers/devices.py index 6455b1f..ce7734d 100644 --- a/talkingq-url/banban/routers/devices.py +++ b/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 diff --git a/talkingq-url/banban/service/device.py b/talkingq-url/banban/service/device.py index 5d777b2..c048b72 100644 --- a/talkingq-url/banban/service/device.py +++ b/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() diff --git a/talkingq-url/banban/service/ota_firmware.py b/talkingq-url/banban/service/ota_firmware.py index e6210ee..c8036e9 100644 --- a/talkingq-url/banban/service/ota_firmware.py +++ b/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() diff --git a/talkingq-url/database/init_db.py b/talkingq-url/database/init_db.py index d445634..9ea1b09 100644 --- a/talkingq-url/database/init_db.py +++ b/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( """ diff --git a/talkingq-url/database/models.py b/talkingq-url/database/models.py index 407d3b8..1cdb8d5 100644 --- a/talkingq-url/database/models.py +++ b/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) diff --git a/talkingq-url/handlers/mqtt_handler.py b/talkingq-url/handlers/mqtt_handler.py index 879555d..1c6a904 100644 --- a/talkingq-url/handlers/mqtt_handler.py +++ b/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: diff --git a/talkingq-url/services/device_update_manager.py b/talkingq-url/services/device_update_manager.py index 9c8856a..5555cfc 100644 --- a/talkingq-url/services/device_update_manager.py +++ b/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, diff --git a/talkingq-url/tests/test_forced_ota.py b/talkingq-url/tests/test_forced_ota.py new file mode 100644 index 0000000..f2a80c9 --- /dev/null +++ b/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 diff --git a/talkingq-url/tests/test_mqtt_device_info_report.py b/talkingq-url/tests/test_mqtt_device_info_report.py index ad454c5..fd19511 100644 --- a/talkingq-url/tests/test_mqtt_device_info_report.py +++ b/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, diff --git a/talkingq-url/tests/test_ota_channel_selection.py b/talkingq-url/tests/test_ota_channel_selection.py index a409a01..d05ddc5 100644 --- a/talkingq-url/tests/test_ota_channel_selection.py +++ b/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