From 5fede9cda1533823bca24131ac003a78da36ed1b Mon Sep 17 00:00:00 2001 From: stu2not Date: Sat, 9 May 2026 10:55:06 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=AF=E6=8C=81=E8=AE=BE=E5=A4=87=E7=9F=AD?= =?UTF-8?q?=E6=8C=89=E5=90=91=E5=AE=B6=E9=95=BF=E7=95=99=E8=A8=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- talkingq-url/banban/service/im.py | 2 + talkingq-url/config.py | 4 + talkingq-url/handlers/mqtt_handler.py | 36 ++++++--- .../handlers/websocket_message_handler.py | 79 +++++++++++++++---- talkingq-url/services/device_target_cache.py | 38 ++++++++- 5 files changed, 130 insertions(+), 29 deletions(-) diff --git a/talkingq-url/banban/service/im.py b/talkingq-url/banban/service/im.py index 9b11433..72cf2da 100644 --- a/talkingq-url/banban/service/im.py +++ b/talkingq-url/banban/service/im.py @@ -140,6 +140,8 @@ async def present_message_item( ) -> ChildConversationMessageItem: item = row_to_message_item(row) if item.content_type == 2 and item.media_file_key: + if item.media_file_key.startswith(("http://", "https://")): + return item try: item.media_file_key = await audio_storage.get_audio_url(item.media_file_key) except MessageAudioStorageError: diff --git a/talkingq-url/config.py b/talkingq-url/config.py index 7b175e9..0f3227e 100644 --- a/talkingq-url/config.py +++ b/talkingq-url/config.py @@ -68,6 +68,10 @@ class Settings(BaseSettings): talkingq_mqtt_qos: int = Field(default=1, validation_alias="TALKINGQ_MQTT_QOS") talkingq_mqtt_keepalive: int = Field(default=60, validation_alias="TALKINGQ_MQTT_KEEPALIVE") talkingq_mqtt_nfc_notice_interval: int = Field(default=600, validation_alias="TALKINGQ_MQTT_NFC_NOTICE_INTERVAL") + talkingq_parent_message_uuid: str = Field( + default="TalkingQ_xxx", + validation_alias="TALKINGQ_PARENT_MESSAGE_UUID", + ) admin_api_key: str = Field(default="", validation_alias="ADMIN_API_KEY") client_api_key: str = Field(default="", validation_alias="CLIENT_API_KEY") diff --git a/talkingq-url/handlers/mqtt_handler.py b/talkingq-url/handlers/mqtt_handler.py index c3b288e..a6e6caa 100644 --- a/talkingq-url/handlers/mqtt_handler.py +++ b/talkingq-url/handlers/mqtt_handler.py @@ -335,8 +335,21 @@ class TalkingQMQTTService: async def _handle_short_press_message(self, device_id: str, payload: dict): params = payload.get("params", {}) - nfc_uuid = params.get("uuid") + nfc_uuid = str(params.get("uuid") or "").strip() logger.info(device_id, "", f"[短按留言] 设备 {device_id} 短按发送留言, UUID={nfc_uuid}") + if nfc_uuid == settings.talkingq_parent_message_uuid: + await device_target_cache.set_parent_target(device_id, nfc_uuid) + response_payload = { + "msg_id": "011", + "status": "success", + "type": 0, + "params": { + "url": "http://175.24.73.253:8080/assets/audio/message_ok.mp3" + }, + } + await self._publish(f"device/{device_id}/event_resp", response_payload) + return + media_file_key = str(params.get("media_file_key") or params.get("audio_url") or "").strip() if media_file_key: try: @@ -351,6 +364,18 @@ class TalkingQMQTTService: ext_json=params.get("ext_json") if isinstance(params.get("ext_json"), dict) else None, ) logger.info(device_id, "", f"[短按留言] 设备 {device_id} 留言已写入家长会话") + await self._publish( + f"device/{device_id}/event_resp", + { + "msg_id": "011", + "status": "success", + "type": 0, + "params": { + "url": "http://175.24.73.253:8080/assets/audio/message_ok.mp3" + }, + }, + ) + return except Exception as exc: logger.warning(device_id, "", f"[短按留言] 设备 {device_id} 留言写入失败: {exc}") await self._publish( @@ -363,14 +388,7 @@ class TalkingQMQTTService: ) return - payload = { - "msg_id": "011", - "status": "success", - "type": 0, - "params": { - "url": f"http://{settings.server_host}:{settings.server_port}/assets/audio/message_ok.mp3" - } - } + payload = {"msg_id": "011", "status": "failed", "message": "unsupported uuid"} await self._publish(f"device/{device_id}/event_resp", payload) async def _handle_device_identity_init(self, imei: str, payload: dict): diff --git a/talkingq-url/handlers/websocket_message_handler.py b/talkingq-url/handlers/websocket_message_handler.py index 66f875c..4454776 100644 --- a/talkingq-url/handlers/websocket_message_handler.py +++ b/talkingq-url/handlers/websocket_message_handler.py @@ -13,6 +13,7 @@ from services.device_target_cache import device_target_cache from services.offline_audio_cache import offline_audio_cache from services.target_audio_cache import target_audio_cache from services.card_service import card_service +from services.device_target_cache import VOICE_TARGET_KIND_PARENT from utils.logger import session_logger from handlers.prompt_sound_handler import handle_prompt_sound_request from handlers.session_cleanup_handler import handle_old_session_cleanup @@ -147,29 +148,28 @@ async def handle_binary_message(websocket: WebSocket, device_id: str, serial_num session_id = session_key[1] if packet_type == 1: # 开始包 - target_device_id = await device_target_cache.get_target(device_id) - if target_device_id: - # 清除音频缓存 - await target_audio_cache.clear_audio_data(target_device_id) + voice_target = await device_target_cache.get_voice_target(device_id) + if voice_target: + await target_audio_cache.clear_audio_data(voice_target.audio_cache_key) return None, None current_active_session = await handle_start_packet( device_id, session_id, session_key, session, current_active_session ) elif packet_type == 3: # 中断包 - target_device_id = await device_target_cache.get_target(device_id) - if target_device_id: + voice_target = await device_target_cache.get_voice_target(device_id) + if voice_target: return None, None current_active_session = await handle_interrupt_packet( device_id, session_id, session_key, session, current_active_session ) elif packet_type == 2: # 结束包 - target_device_id = await device_target_cache.get_target(device_id) - if target_device_id: - # 处理缓存的音频数据 + voice_target = await device_target_cache.get_voice_target(device_id) + if voice_target: session_logger.info(device_id, session_id, f"收到结束包，开始处理缓存音频数据") - await process_cached_audio(device_id, target_device_id, serial_number) - - # 移除目标设备关联 + if voice_target.kind == VOICE_TARGET_KIND_PARENT: + await process_parent_leave_message(device_id, voice_target.audio_cache_key) + else: + await process_cached_audio(device_id, voice_target.target_device_id, serial_number) await device_target_cache.remove_target(device_id) return None, None if current_active_session == session_key: @@ -243,20 +243,65 @@ async def handle_interrupt_packet(device_id, session_id, session_key, session, c async def handle_target_audio_packet(device_id: str, audio_data: bytes): """处理发送给目标设备的音频包""" try: - target_device_id = await device_target_cache.get_target(device_id) + voice_target = await device_target_cache.get_voice_target(device_id) - if not target_device_id: + if not voice_target: session_logger.warning(device_id, "target", "未设置目标设备，无法发送音频") return - # 缓存音频数据 - await target_audio_cache.add_audio_data(target_device_id, audio_data) - session_logger.info(device_id, "target", f"已缓存音频数据到目标设备 {target_device_id}") + await target_audio_cache.add_audio_data(voice_target.audio_cache_key, audio_data) + if voice_target.kind == VOICE_TARGET_KIND_PARENT: + session_logger.info(device_id, "target", "已缓存发给家长的留言音频数据") + else: + session_logger.info(device_id, "target", f"已缓存音频数据到目标设备 {voice_target.target_device_id}") except Exception as e: session_logger.error(device_id, "target", f"处理目标音频包时出错: {e}", exc_info=True) +async def process_parent_leave_message(device_id: str, audio_cache_key: str): + """处理设备发给家长的留言音频并写入家长会话。""" + try: + cached_audio = await target_audio_cache.get_audio_data(audio_cache_key) + if not cached_audio: + session_logger.info(device_id, "parent", "发给家长的留言没有缓存音频数据") + return + + audio_file_key = await save_audio_file(cached_audio, device_id) + audio_url = f"http://{settings.server_host}:{settings.server_port}/{audio_file_key}" + await im_conversation_service.create_device_parent_leave_message( + device_id=device_id, + media_file_key=audio_url, + media_mime_type="audio/mpeg", + media_size_bytes=len(cached_audio), + ext_json={"source": "device_ws_parent_leave_message"}, + ) + session_logger.info(device_id, "parent", "发给家长的留言已写入家长会话") + + websocket = await connection_manager.get_connection(device_id) + if websocket and websocket.client_state.name == "CONNECTED": + success_audio_url = f"http://{settings.server_host}:{settings.server_port}/assets/audio/message_ok.mp3" + await websocket.send_text(f"PROMPT_SOUND_URL:{success_audio_url}") + session_logger.info(device_id, "device", f"留言成功提示音已发送给设备 {device_id}") + else: + session_logger.warning(device_id, "device", f"设备 {device_id} 不在线，暂不发送留言成功提示音") + except Exception as e: + if isinstance(e, HTTPException): + session_logger.error(device_id, "parent", f"发给家长的留言保存失败，返回HTTPException: {e}") + websocket = await connection_manager.get_connection(device_id) + if e.status_code == 404 or e.status_code == 400: + if websocket and websocket.client_state.name == "CONNECTED": + fail_audio_url = f"http://{settings.server_host}:{settings.server_port}/assets/audio/save_audio_fail.mp3" + await websocket.send_text(f"PROMPT_SOUND_URL:{fail_audio_url}") + session_logger.info(device_id, "device", f"留言失败提示音已发送给设备 {device_id}") + else: + session_logger.warning(device_id, "device", f"设备 {device_id} 不在线，暂不发送留言失败提示音") + else: + session_logger.error(device_id, "parent", f"处理家长留言音频时出错: {e}", exc_info=True) + finally: + await target_audio_cache.clear_audio_data(audio_cache_key) + + async def process_cached_audio(device_id: str, target_device_id: str, serial_number: str): """处理缓存的音频数据并发送音频URL""" try: diff --git a/talkingq-url/services/device_target_cache.py b/talkingq-url/services/device_target_cache.py index 1f3b1c9..5eac15d 100644 --- a/talkingq-url/services/device_target_cache.py +++ b/talkingq-url/services/device_target_cache.py @@ -1,19 +1,51 @@ import asyncio +from dataclasses import dataclass from typing import Dict, Optional +VOICE_TARGET_KIND_DEVICE = "device" +VOICE_TARGET_KIND_PARENT = "parent" + + +@dataclass(frozen=True) +class VoiceTarget: + kind: str + target_device_id: Optional[str] = None + uuid: Optional[str] = None + audio_cache_key: Optional[str] = None + + class DeviceTargetCache: def __init__(self): - self.targets: Dict[str, str] = {} + self.targets: Dict[str, VoiceTarget] = {} self.lock = asyncio.Lock() async def get_target(self, device_id: str) -> Optional[str]: + async with self.lock: + target = self.targets.get(device_id) + if not target or target.kind != VOICE_TARGET_KIND_DEVICE: + return None + return target.target_device_id + + async def get_voice_target(self, device_id: str) -> Optional[VoiceTarget]: async with self.lock: return self.targets.get(device_id) async def set_target(self, device_id: str, target_device_id: str): async with self.lock: - self.targets[device_id] = target_device_id + self.targets[device_id] = VoiceTarget( + kind=VOICE_TARGET_KIND_DEVICE, + target_device_id=target_device_id, + audio_cache_key=target_device_id, + ) + + async def set_parent_target(self, device_id: str, uuid: str): + async with self.lock: + self.targets[device_id] = VoiceTarget( + kind=VOICE_TARGET_KIND_PARENT, + uuid=uuid, + audio_cache_key=f"parent:{device_id}", + ) async def remove_target(self, device_id: str): async with self.lock: @@ -26,4 +58,4 @@ class DeviceTargetCache: # 单例实例 -device_target_cache = DeviceTargetCache() \ No newline at end of file +device_target_cache = DeviceTargetCache()