From 46cc5272f31a6c117b2b7ab95afd64d83c9222ea Mon Sep 17 00:00:00 2001 From: HycJack <772403255@qq.com> Date: Fri, 8 May 2026 17:33:03 +0800 Subject: [PATCH] =?UTF-8?q?add=20=E5=91=8A=E8=AD=A6=E5=92=8C=E8=81=94?= =?UTF-8?q?=E8=B0=83=E9=97=AE=E9=A2=98=E4=BF=AE=E6=94=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- talkingq-url/banban/dao/device.py | 4 + talkingq-url/banban/dao/device_alarm.py | 86 ++++++++ talkingq-url/banban/routers/devices.py | 181 +++++++++++++++- talkingq-url/banban/service/device.py | 49 ++++- talkingq-url/banban/service/device_alarm.py | 47 +++++ talkingq-url/banban/service/device_setting.py | 39 +++- talkingq-url/database/models.py | 16 ++ talkingq-url/handlers/mqtt_handler.py | 194 +++++++++--------- 8 files changed, 516 insertions(+), 100 deletions(-) create mode 100644 talkingq-url/banban/dao/device_alarm.py create mode 100644 talkingq-url/banban/service/device_alarm.py diff --git a/talkingq-url/banban/dao/device.py b/talkingq-url/banban/dao/device.py index 517884d..73852e1 100644 --- a/talkingq-url/banban/dao/device.py +++ b/talkingq-url/banban/dao/device.py @@ -77,6 +77,10 @@ class DeviceDAO(BaseDAO): db.device_id, db.child_id, c.child_name, + ds.sleep_mode, + ds.disable_time_start, + ds.disable_time_end, + ds.timezone, ds.power, ds.volume, ds.`signal` AS signal_strength, diff --git a/talkingq-url/banban/dao/device_alarm.py b/talkingq-url/banban/dao/device_alarm.py new file mode 100644 index 0000000..ea7faa5 --- /dev/null +++ b/talkingq-url/banban/dao/device_alarm.py @@ -0,0 +1,86 @@ +from collections.abc import Mapping +from typing import Any + +from sqlalchemy import text + +from banban.dao import BaseDAO + + +class DeviceAlarmDAO(BaseDAO): + async def create( + self, + *, + device_id: str, + owner_user_id: int | None, + child_id: int | None, + source_msg_id: str = "010", + ) -> int: + result = await self.execute( + """ + INSERT INTO device_alarm_events ( + device_id, owner_user_id, child_id, source_msg_id, created_at + ) + VALUES ( + :device_id, :owner_user_id, :child_id, :source_msg_id, CURRENT_TIMESTAMP + ) + """, + { + "device_id": device_id, + "owner_user_id": owner_user_id, + "child_id": child_id, + "source_msg_id": source_msg_id, + }, + ) + await self.commit() + return int(result.lastrowid) + + async def get_active_binding_context(self, *, device_id: str) -> Mapping[str, Any] | None: + result = await self.execute( + text( + """ + SELECT owner_user_id, child_id + FROM device_bindings + WHERE device_id = :device_id + AND status = 1 + LIMIT 1 + """ + ), + {"device_id": device_id}, + ) + return result.mappings().first() + + async def list_by_device( + self, + *, + device_id: str, + owner_user_id: int, + limit: int, + ) -> list[Mapping[str, Any]]: + result = await self.execute( + text( + """ + SELECT + dae.alarm_id, + dae.device_id, + dae.owner_user_id, + dae.child_id, + dae.source_msg_id, + dae.created_at, + c.child_name + FROM device_alarm_events AS dae + LEFT JOIN children AS c + ON c.child_id = dae.child_id + AND c.status = 1 + WHERE dae.device_id = :device_id + AND dae.owner_user_id = :owner_user_id + ORDER BY dae.created_at DESC, dae.alarm_id DESC + LIMIT :limit + """ + ), + { + "device_id": device_id, + "owner_user_id": owner_user_id, + "limit": limit, + }, + ) + return list(result.mappings().all()) diff --git a/talkingq-url/banban/routers/devices.py b/talkingq-url/banban/routers/devices.py index edcdd40..95b5406 100644 --- a/talkingq-url/banban/routers/devices.py +++ b/talkingq-url/banban/routers/devices.py @@ -1,6 +1,6 @@ import logging from collections.abc import Mapping -from datetime import datetime +from datetime import datetime, time from services.connection_manager import connection_manager from fastapi import APIRouter, Depends, HTTPException, Query, Request from pydantic import BaseModel, Field @@ -15,6 +15,7 @@ try: DeviceLocationTrajectoryItem, DeviceLocationTrajectoryResponse, ) + from banban.service.device_alarm import device_alarm_service from banban.service.location import location_service from banban.service.device import device_service except ModuleNotFoundError: @@ -24,6 +25,7 @@ except ModuleNotFoundError: DeviceLocationTrajectoryItem, DeviceLocationTrajectoryResponse, ) + from banban.service.device_alarm import device_alarm_service from banban.service.location import location_service from banban.service.device import device_service @@ -53,6 +55,10 @@ class DeviceStatusResponse(BaseModel): device_id: str child_id: int | None = None child_name: str | None = None + sleep_mode: int | None = None + disable_time_start: str | None = None + disable_time_end: str | None = None + timezone: str | None = None power: int | None = None volume: int | None = None signal: int | None = None @@ -72,6 +78,20 @@ class DeviceStatusResponse(BaseModel): location_updated_at: datetime | None = None +class DeviceAlarmItem(BaseModel): + alarm_id: int + device_id: str + child_id: int | None = None + child_name: str | None = None + source_msg_id: str + created_at: datetime + + +class DeviceAlarmListResponse(BaseModel): + items: list[DeviceAlarmItem] + total: int + + class DeviceVolumeUpdateRequest(BaseModel): level: int = Field(ge=0, le=100) @@ -82,6 +102,31 @@ class DeviceVolumeUpdateResponse(BaseModel): msg_id: str +class DeviceSleepScheduleUpdateRequest(BaseModel): + start: str = Field(..., pattern=r"^\d{2}:\d{2}$") + end: str = Field(..., pattern=r"^\d{2}:\d{2}$") + timezone: str = Field(default="Asia/Shanghai", min_length=1, max_length=32) + + +class DeviceSleepScheduleResponse(BaseModel): + device_id: str + sleep_mode: int + start: str | None = None + end: str | None = None + timezone: str + msg_id: str | None = None + + +class DeviceRemoteSleepWakeRequest(BaseModel): + switch: str = Field(..., pattern=r"^(on|off)$") + + +class DeviceRemoteSleepWakeResponse(BaseModel): + device_id: str + switch: str + msg_id: str + + @@ -140,11 +185,31 @@ def _row_to_trajectory_item(row: Mapping, *, child_name: str | None) -> DeviceLo ) +def _format_time_value(value: object) -> str | None: + if value is None: + return None + if isinstance(value, time): + return value.strftime("%H:%M") + text_value = str(value) + return text_value[:5] if len(text_value) >= 5 else text_value + + +def _parse_time_value(value: str, field_name: str) -> time: + try: + return datetime.strptime(value, "%H:%M").time() + except ValueError as exc: + raise HTTPException(status_code=422, detail=f"{field_name} must use HH:MM format") from exc + + def _row_to_device_status_response(row: Mapping) -> DeviceStatusResponse: return DeviceStatusResponse( device_id=str(row["device_id"]), child_id=int(row["child_id"]) if row["child_id"] is not None else None, child_name=row.get("child_name"), + sleep_mode=row.get("sleep_mode"), + disable_time_start=_format_time_value(row.get("disable_time_start")), + disable_time_end=_format_time_value(row.get("disable_time_end")), + timezone=row.get("timezone"), power=row["power"], volume=row["volume"], signal=row["signal_strength"], @@ -165,6 +230,17 @@ def _row_to_device_status_response(row: Mapping) -> DeviceStatusResponse: ) +def _row_to_alarm_item(row: Mapping) -> DeviceAlarmItem: + return DeviceAlarmItem( + alarm_id=int(row["alarm_id"]), + device_id=str(row["device_id"]), + child_id=int(row["child_id"]) if row["child_id"] is not None else None, + child_name=row.get("child_name"), + source_msg_id=str(row["source_msg_id"]), + created_at=row["created_at"], + ) + + @router.get("/{device_id}/messages", response_model=DeviceMessageListResponse) async def list_device_messages( device_id: str, @@ -226,6 +302,36 @@ async def get_device_status( return _row_to_device_status_response(row) +@router.get("/{device_id}/alarms", response_model=DeviceAlarmListResponse) +async def list_device_alarms( + device_id: str, + request: Request, + limit: int = Query(default=20, ge=1, le=100), + current_user_id: int = Depends(get_current_user_id), +) -> DeviceAlarmListResponse: + await device_service.ensure_device_access(device_id=device_id, user_id=current_user_id) + rows = await device_alarm_service.list_device_alarms( + device_id=device_id, + owner_user_id=current_user_id, + limit=limit, + ) + + logger.info( + "device alarms fetched", + extra={ + "event": "device_alarm_list", + "request_id": getattr(request.state, "request_id", None), + "user_id": current_user_id, + "device_id": device_id, + "returned_count": len(rows), + }, + ) + return DeviceAlarmListResponse( + items=[_row_to_alarm_item(row) for row in rows], + total=len(rows), + ) + + @router.post("/{device_id}/volume", response_model=DeviceVolumeUpdateResponse) async def set_device_volume( device_id: str, @@ -253,6 +359,77 @@ async def set_device_volume( return DeviceVolumeUpdateResponse(device_id=device_id, level=payload.level, msg_id=msg_id) +@router.post("/{device_id}/sleep-schedule", response_model=DeviceSleepScheduleResponse) +async def set_device_sleep_schedule( + device_id: str, + payload: DeviceSleepScheduleUpdateRequest, + request: Request, + current_user_id: int = Depends(get_current_user_id), +) -> DeviceSleepScheduleResponse: + start_time = _parse_time_value(payload.start, "start") + end_time = _parse_time_value(payload.end, "end") + msg_id = await device_service.set_sleep_schedule( + device_id=device_id, + user_id=current_user_id, + start=start_time, + end=end_time, + timezone=payload.timezone, + ) + + logger.info( + "device sleep schedule command sent", + extra={ + "event": "device_sleep_schedule_set", + "request_id": getattr(request.state, "request_id", None), + "user_id": current_user_id, + "device_id": device_id, + "start": payload.start, + "end": payload.end, + "timezone": payload.timezone, + "msg_id": msg_id, + }, + ) + return DeviceSleepScheduleResponse( + device_id=device_id, + sleep_mode=1, + start=payload.start, + end=payload.end, + timezone=payload.timezone, + msg_id=msg_id, + ) + + +@router.post("/{device_id}/sleep-wake", response_model=DeviceRemoteSleepWakeResponse) +async def set_device_remote_sleep_wake( + device_id: str, + payload: DeviceRemoteSleepWakeRequest, + request: Request, + current_user_id: int = Depends(get_current_user_id), +) -> DeviceRemoteSleepWakeResponse: + msg_id = await device_service.set_remote_sleep_wake( + device_id=device_id, + user_id=current_user_id, + switch=payload.switch, + ) + + logger.info( + "device remote sleep wake command sent", + extra={ + "event": "device_remote_sleep_wake_set", + "request_id": getattr(request.state, "request_id", None), + "user_id": current_user_id, + "device_id": device_id, + "switch": payload.switch, + "msg_id": msg_id, + }, + ) + return DeviceRemoteSleepWakeResponse( + device_id=device_id, + switch=payload.switch, + msg_id=msg_id, + ) + + @router.get("/{device_id}/location", response_model=DeviceLocationCurrentResponse) async def get_current_device_location( @@ -358,4 +535,4 @@ async def get_device_location_trajectory( total=len(rows), start_at=start_at, end_at=end_at, - ) \ No newline at end of file + ) diff --git a/talkingq-url/banban/service/device.py b/talkingq-url/banban/service/device.py index 8e63add..135ca0c 100644 --- a/talkingq-url/banban/service/device.py +++ b/talkingq-url/banban/service/device.py @@ -1,10 +1,12 @@ from collections.abc import Mapping -from typing import Any, List +from datetime import time +from typing import Any, List, Literal from services.database_service_base import DatabaseServiceBase from fastapi import HTTPException from banban.dao.device import DeviceDAO +from banban.service.device_setting import device_setting_service class DeviceService(DatabaseServiceBase): @@ -68,6 +70,51 @@ class DeviceService(DatabaseServiceBase): raise HTTPException(status_code=503, detail="MQTT 服务未初始化") return await service.send_volume_command(device_id, level) + async def set_sleep_schedule( + self, + *, + device_id: str, + user_id: int, + start: time, + end: time, + timezone: str, + ) -> str: + await self.ensure_device_access(device_id=device_id, user_id=user_id) + + from handlers.mqtt_handler import TalkingQMQTTService + + service = await TalkingQMQTTService.get_instance() + if service is None: + raise HTTPException(status_code=503, detail="MQTT 服务未初始化") + + start_text = start.strftime("%H:%M") + end_text = end.strftime("%H:%M") + msg_id = await service.send_sleep_schedule_command(device_id, start_text, end_text) + await device_setting_service.upsert_sleep_schedule( + device_id=device_id, + sleep_mode=1, + disable_time_start=start, + disable_time_end=end, + timezone=timezone, + ) + return msg_id + + async def set_remote_sleep_wake( + self, + *, + device_id: str, + user_id: int, + switch: Literal["on", "off"], + ) -> str: + await self.ensure_device_access(device_id=device_id, user_id=user_id) + + from handlers.mqtt_handler import TalkingQMQTTService + + service = await TalkingQMQTTService.get_instance() + if service is None: + raise HTTPException(status_code=503, detail="MQTT 服务未初始化") + return await service.send_remote_sleep_wake_command(device_id, switch) + # 创建全局 DeviceService 实例 device_service = DeviceService() diff --git a/talkingq-url/banban/service/device_alarm.py b/talkingq-url/banban/service/device_alarm.py new file mode 100644 index 0000000..efabfdb --- /dev/null +++ b/talkingq-url/banban/service/device_alarm.py @@ -0,0 +1,47 @@ +from collections.abc import Mapping +from typing import Any + +from banban.dao.device_alarm import DeviceAlarmDAO +from services.database_service_base import DatabaseServiceBase + + +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: + db_session = await self.get_session() + try: + dao = DeviceAlarmDAO(db_session) + binding = await dao.get_active_binding_context(device_id=device_id) + if not binding: + return None + return await dao.create( + device_id=device_id, + owner_user_id=int(binding["owner_user_id"]) if binding.get("owner_user_id") is not None else None, + child_id=int(binding["child_id"]) if binding.get("child_id") is not None else None, + source_msg_id=source_msg_id, + ) + finally: + await db_session.close() + + async def list_device_alarms( + self, + *, + device_id: str, + owner_user_id: int, + limit: int, + ) -> list[Mapping[str, Any]]: + db_session = await self.get_session() + try: + dao = DeviceAlarmDAO(db_session) + return await dao.list_by_device( + device_id=device_id, + owner_user_id=owner_user_id, + limit=limit, + ) + finally: + await db_session.close() + + +device_alarm_service = DeviceAlarmService() diff --git a/talkingq-url/banban/service/device_setting.py b/talkingq-url/banban/service/device_setting.py index 6e0a289..ec87791 100644 --- a/talkingq-url/banban/service/device_setting.py +++ b/talkingq-url/banban/service/device_setting.py @@ -93,7 +93,14 @@ class DeviceSettingService(DatabaseServiceBase): finally: await db_session.close() - async def insert_or_update(self, device_id: str, power: int, volume: int, signal_strength: int, version_str: str) -> None: + async def insert_or_update( + self, + device_id: str, + power: Optional[int], + volume: Optional[int], + signal_strength: Optional[int], + version_str: Optional[str], + ) -> None: try: current_row = await self.get_setting_by_device_id(device_id=device_id) if current_row: @@ -103,5 +110,33 @@ class DeviceSettingService(DatabaseServiceBase): finally: pass + async def upsert_sleep_schedule( + self, + *, + device_id: str, + sleep_mode: int, + disable_time_start: time, + disable_time_end: time, + timezone: str, + ) -> None: + current_row = await self.get_setting_by_device_id(device_id=device_id) + if current_row: + await self.update_setting( + device_id=device_id, + sleep_mode=sleep_mode, + disable_time_start=disable_time_start, + disable_time_end=disable_time_end, + timezone=timezone, + ) + return + + await self.create_setting( + device_id=device_id, + sleep_mode=sleep_mode, + disable_time_start=disable_time_start, + disable_time_end=disable_time_end, + timezone=timezone, + ) + # 创建全局 DeviceSettingService 实例 -device_setting_service = DeviceSettingService() \ No newline at end of file +device_setting_service = DeviceSettingService() diff --git a/talkingq-url/database/models.py b/talkingq-url/database/models.py index 4147097..8b141e6 100644 --- a/talkingq-url/database/models.py +++ b/talkingq-url/database/models.py @@ -267,6 +267,22 @@ class DeviceSetting(Base): onupdate=datetime.utcnow, ) + +class DeviceAlarmEvent(Base): + __tablename__ = "device_alarm_events" + __table_args__ = ( + Index("idx_device_alarm_device_created", "device_id", "created_at"), + Index("idx_device_alarm_owner_created", "owner_user_id", "created_at"), + Index("idx_device_alarm_child_created", "child_id", "created_at"), + ) + + alarm_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + device_id: Mapped[str] = mapped_column(String(64), nullable=False) + owner_user_id: Mapped[Optional[int]] = mapped_column(Integer) + child_id: Mapped[Optional[int]] = mapped_column(Integer) + source_msg_id: Mapped[str] = mapped_column(String(8), server_default=text("'010'")) + created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) + class IMConversation(Base): __tablename__ = "im_conversations" __table_args__ = ( diff --git a/talkingq-url/handlers/mqtt_handler.py b/talkingq-url/handlers/mqtt_handler.py index b396205..a5801e1 100644 --- a/talkingq-url/handlers/mqtt_handler.py +++ b/talkingq-url/handlers/mqtt_handler.py @@ -1,9 +1,15 @@ +import asyncio import json import time -import asyncio -from typing import Optional, Dict, Callable, Awaitable -from banban.service.device_setting import device_setting_service +from datetime import datetime +from typing import Awaitable, Callable, Dict, Optional + +import aiomqtt + from banban.service.binding import BindingService +from banban.service.device_alarm import device_alarm_service +from banban.service.device_setting import device_setting_service +from banban.service.location import location_service from config import settings from services.card_service import card_service from services.offline_audio_cache import offline_audio_cache @@ -12,6 +18,8 @@ from banban.service.location import location_service from database.models import ChildLocationCurrent from datetime import datetime 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 from services.task_manager import task_manager @@ -69,30 +77,28 @@ class TalkingQMQTTService: try: topic = str(message.topic) parts = topic.split("/") - if parts[0] == "device": - device_id = parts[1] if len(parts) >= 2 else "unknown" + if not parts or parts[0] != "device": + continue - if not device_id.startswith(f"{self.device_prefix}_"): - continue + device_id = parts[1] if len(parts) >= 2 else "unknown" + if not device_id.startswith(f"{self.device_prefix}_"): + continue - payload = json.loads(message.payload.decode("utf-8")) - - msg_id = payload.get("msg_id") - - handler = self._msg_handlers.get(msg_id) - if handler: - await handler(device_id, payload) - else: - logger.warning(device_id, "", f"未知 msg_id: {msg_id}, 设备: {device_id}") - - except json.JSONDecodeError as e: - logger.error("", "", f"消息解析失败: {e}") - except Exception as e: - logger.error("", "", f"消息处理异常: {e}") + payload = json.loads(message.payload.decode("utf-8")) + msg_id = payload.get("msg_id") + handler = self._msg_handlers.get(msg_id) + if handler: + await handler(device_id, payload) + else: + logger.warning(device_id, "", f"unknown msg_id={msg_id}") + except json.JSONDecodeError as exc: + logger.error("", "", f"invalid mqtt payload: {exc}") + except Exception as exc: + logger.error("", "", f"mqtt message handling failed: {exc}") except asyncio.CancelledError: pass - except Exception as e: - logger.error("", "", f"消息循环异常: {e}") + except Exception as exc: + logger.error("", "", f"mqtt loop failed: {exc}") self._connected = False async def _schedule_persistence(self, device_id: str, label: str, coro) -> None: @@ -109,7 +115,6 @@ class TalkingQMQTTService: ) async def _handle_device_info(self, device_id: str, payload: dict): - # status = payload.get("status") data = payload.get("data", {}) await self._schedule_persistence( device_id, @@ -199,17 +204,15 @@ class TalkingQMQTTService: async def _handle_nfc_notice_response(self, device_id: str, payload: dict): status = payload.get("status") - if status == "success": - logger.info(device_id, "", f"[NFC通知] 设备 {device_id} 已收到留言提示") - else: - logger.warning(device_id, "", f"[NFC通知] 设备 {device_id} 留言提示失败: {payload}") + if status != "success": + logger.warning(device_id, "", f"[NFC notice] command failed: {payload}") async def _handle_nfc_listen_report(self, device_id: str, payload: dict): params = payload.get("params", {}) nfc_uuid = params.get("uuid") logger.info(device_id, "", f"[NFC收听] 设备 {device_id} 请求收听留言, UUID={nfc_uuid}") await self._send_nfc_listen_response(device_id, nfc_uuid) - + async def _handle_bind_response(self, device_id: str, payload: dict): params = payload status = params.get("status") @@ -223,7 +226,8 @@ class TalkingQMQTTService: if result is None: logger.warning(device_id, "", "[NFC bind] no pending bind session found") return - logger.info(device_id, "", f"[NFC绑定] 设备 {device_id} 请求绑定卡片, UUID={nfc_uuid}, result={result}") + + logger.info(device_id, "", f"[NFC bind] bind completed, uuid={nfc_uuid}, result={result}") async def _handle_open_response(self, device_id: str, payload: dict): params = payload @@ -237,6 +241,25 @@ class TalkingQMQTTService: status = payload.get("status") if status == "success": logger.info(device_id, "", f"[定时休眠] 设备 {device_id} 休眠时间设置成功") + data = payload.get("data", {}) or {} + start = data.get("start") + end = data.get("end") + timezone = data.get("timezone") or "Asia/Shanghai" + if start and end: + try: + await self._schedule_persistence( + device_id, + "sleep_schedule", + device_setting_service.upsert_sleep_schedule( + device_id=device_id, + sleep_mode=1, + disable_time_start=datetime.strptime(start, "%H:%M").time(), + disable_time_end=datetime.strptime(end, "%H:%M").time(), + timezone=timezone, + ), + ) + except ValueError: + logger.warning(device_id, "", f"[定时休眠] 无法解析设备返回的时间: {payload}") else: logger.warning(device_id, "", f"[定时休眠] 设备 {device_id} 休眠时间设置失败: {payload}") @@ -249,6 +272,11 @@ class TalkingQMQTTService: async def _handle_alarm_report(self, device_id: str, payload: dict): logger.info(device_id, "", f"[告警] 设备 {device_id} 发送紧急报警") + await self._schedule_persistence( + device_id, + "alarm_event", + device_alarm_service.record_alarm_event(device_id=device_id, source_msg_id="010"), + ) await self._publish(f"device/{device_id}/event_resp", {"msg_id": "010", "status": "success"}) async def _handle_short_press_message(self, device_id: str, payload: dict): @@ -268,7 +296,6 @@ class TalkingQMQTTService: topic = f"device/{device_id}/event_resp" card = await card_service.get_card_by_uuid(nfc_uuid) if not card: - logger.warning("", "", f"[NFC收听] 设备 {device_id} 没有找到卡片{nfc_uuid}，无法发送留言提示") payload = { "msg_id": "005", "type": 0, @@ -291,7 +318,8 @@ class TalkingQMQTTService: } } await self._publish(topic, payload) - return + return + if has_pending: audio_urls = await offline_audio_cache.get_audio_urls(device_id) # 53D92B6DA20001 测试卡片 @@ -314,49 +342,38 @@ class TalkingQMQTTService: } } else: - params = {} - for k, audio_url in enumerate(audio_urls, start=1): - params[f"url_{k}"] = audio_url - payload = { - "msg_id": "005", - "type": 0, - "params": params - } + params = {f"url_{k}": audio_url for k, audio_url in enumerate(audio_urls, start=1)} + payload = {"msg_id": "005", "type": 0, "params": params} await offline_audio_cache.clear_audio_urls(device_id) await self._publish(topic, payload) - else: - payload = { - "msg_id": "005", - "type": 0, - "params": { - "url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/no_message.mp3" - } - } - await self._publish(topic, payload) - else: - # 检查卡片是否存在 - existing_card = await card_service.get_card_by_uuid(nfc_uuid) - - if existing_card: - # 卡片已存在，使用卡片绑定的设备ID作为目标设备ID - target_device_id = existing_card.device_id - logger.info(device_id, "card", f"卡片已存在，绑定的设备ID: {target_device_id}") - else: - # 卡片不存在，创建新卡片并绑定到当前设备 - new_card = await card_service.activate_card(nfc_uuid, device_id) - logger.info(device_id, "card", f"新卡片{nfc_uuid}已创建并激活，绑定到设备: {device_id}") return - - # 设置目标设备 - await device_target_cache.set_target(device_id, target_device_id) + payload = { "msg_id": "005", - "type": 1, + "type": 0, "params": { - "url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/welcome.mp3" - } + "url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/no_message.mp3" + }, } await self._publish(topic, payload) + return + + existing_card = await card_service.get_card_by_uuid(nfc_uuid) + if existing_card: + target_device_id = existing_card.device_id + else: + await card_service.activate_card(nfc_uuid, device_id) + return + + await device_target_cache.set_target(device_id, target_device_id) + payload = { + "msg_id": "005", + "type": 1, + "params": { + "url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/welcome.mp3" + }, + } + await self._publish(topic, payload) async def connect(self): async with self._connect_lock: @@ -373,19 +390,15 @@ class TalkingQMQTTService: ) await self._client.__aenter__() self._connected = True - logger.system_info("", f"TalkingQ MQTT 已连接: {self.broker}:{self.port}") await self._client.subscribe("device/+/response", qos=self.qos) - logger.system_info("", "已订阅: device/+/response") - await self._client.subscribe("device/+/event", qos=self.qos) - logger.system_info("", "已订阅: device/+/event") self._message_task = asyncio.create_task(self._message_loop()) - logger.info("", "", "TalkingQ MQTT 服务启动中...") - except Exception as e: + logger.info("", "", f"mqtt connected: {self.broker}:{self.port}") + except Exception as exc: self._connected = False - logger.error("", "", f"TalkingQ MQTT 连接失败: {e}") + logger.error("", "", f"mqtt connect failed: {exc}") async def disconnect(self): if self._message_task and not self._message_task.done(): @@ -412,35 +425,26 @@ class TalkingQMQTTService: await self._ensure_connected() try: await self._client.publish(topic, json.dumps(payload, ensure_ascii=False), qos=self.qos) - logger.info("", "", f"下发命令 -> {topic}: {payload}") - except Exception as e: - logger.error("", "", f"命令下发失败: {e}") + logger.info("", "", f"mqtt publish topic={topic} payload={payload}") + except Exception as exc: + logger.error("", "", f"mqtt publish failed: {exc}") async def send_gps_query(self, device_id: str) -> str: - topic = f"device/{device_id}/command" - payload = {"msg_id": "001"} - await self._publish(topic, payload) + await self._publish(f"device/{device_id}/command", {"msg_id": "001"}) return "001" async def send_volume_command(self, device_id: str, level: int) -> str: - topic = f"device/{device_id}/command" - payload = { - "msg_id": "002", - "params": {"level": level} - } - await self._publish(topic, payload) + await self._publish( + f"device/{device_id}/command", + {"msg_id": "002", "params": {"level": level}}, + ) return "002" async def send_ota_command(self, device_id: str, url: str, version: str) -> str: - topic = f"device/{device_id}/command" - payload = { - "msg_id": "003", - "params": { - "url": url, - "version": version, - } - } - await self._publish(topic, payload) + await self._publish( + f"device/{device_id}/command", + {"msg_id": "003", "params": {"url": url, "version": version}}, + ) return "003" async def send_nfc_notice(self, device_id: str, url: str) -> str: