diff --git a/talkingq-url/banban/service/device_voice_archive.py b/talkingq-url/banban/service/device_voice_archive.py new file mode 100644 index 0000000..89a1b97 --- /dev/null +++ b/talkingq-url/banban/service/device_voice_archive.py @@ -0,0 +1,427 @@ +import asyncio +import os +import shutil +from dataclasses import dataclass +from uuid import uuid4 + +from sqlalchemy import text + +from banban.dao.im import ImDAO +from banban.schemas.im import DeviceMessageCreateRequest +from banban.service.message_audio_storage import ( + MessageAudioStorageError, + message_audio_storage_service, +) +from config import settings +from services.database_service_base import DatabaseServiceBase +from utils.audio_format import ( + DEFAULT_CHANNELS, + DEFAULT_SAMPLE_RATE, + DEFAULT_SAMPLE_WIDTH, + detect_audio_format, + wrap_pcm_as_wav, +) +from utils.logger import session_logger + + +CHILD_PARTICIPANT_TYPE = 2 +CHILD_PEER_CONVERSATION_TYPE = 1 + + +@dataclass(frozen=True) +class PreparedArchiveAudio: + filepath: str + mime_type: str + size_bytes: int + source_format: str + archive_format: str + + +class DeviceVoiceArchiveService(DatabaseServiceBase): + def __init__(self) -> None: + super().__init__(service_name="device_voice_archive") + + async def archive_peer_voice_message( + self, + *, + sender_device_id: str, + receiver_device_id: str, + local_audio_path: str, + ) -> bool: + archive_id = f"voice_archive:{uuid4().hex[:12]}" + session_logger.info( + sender_device_id, + archive_id, + ( + "开始归档设备语音消息: " + f"sender_device_id={sender_device_id}, receiver_device_id={receiver_device_id}, " + f"local_audio_path={local_audio_path}" + ), + ) + + if not local_audio_path or not os.path.exists(local_audio_path): + session_logger.warning( + sender_device_id, + archive_id, + f"跳过语音归档: 本地音频文件不存在, local_audio_path={local_audio_path}", + ) + return False + + prepared_audio = None + db_session = await self.get_session() + try: + prepared_audio = await self._prepare_archive_audio( + sender_device_id=sender_device_id, + archive_id=archive_id, + local_audio_path=local_audio_path, + ) + if prepared_audio is None: + return False + + dao = ImDAO(db_session) + sender_identity = await self._get_child_identity_by_device_id( + db_session=db_session, + device_id=sender_device_id, + ) + receiver_identity = await self._get_child_identity_by_device_id( + db_session=db_session, + device_id=receiver_device_id, + ) + + if sender_identity is None: + session_logger.warning( + sender_device_id, + archive_id, + f"跳过语音归档: 发送设备未绑定有效 child, device_id={sender_device_id}", + ) + return False + if receiver_identity is None: + session_logger.warning( + sender_device_id, + archive_id, + f"跳过语音归档: 接收设备未绑定有效 child, device_id={receiver_device_id}", + ) + return False + if sender_identity["child_id"] == receiver_identity["child_id"]: + session_logger.warning( + sender_device_id, + archive_id, + ( + "跳过语音归档: 发送和接收设备映射到了同一个 child, " + f"child_id={sender_identity['child_id']}" + ), + ) + return False + + session_logger.info( + sender_device_id, + archive_id, + ( + "设备绑定解析完成: " + f"sender_child_id={sender_identity['child_id']}, " + f"sender_child_name={sender_identity['child_name'] or 'unknown'}, " + f"receiver_child_id={receiver_identity['child_id']}, " + f"receiver_child_name={receiver_identity['child_name'] or 'unknown'}" + ), + ) + + with open(prepared_audio.filepath, "rb") as archive_file: + archive_audio_data = archive_file.read() + + session_logger.info( + sender_device_id, + archive_id, + ( + "开始上传语音到 COS: " + f"bucket={settings.cos_bucket_message or 'unset'}, " + f"media_mime_type={prepared_audio.mime_type}, " + f"audio_size_bytes={prepared_audio.size_bytes}, " + f"archive_format={prepared_audio.archive_format}" + ), + ) + stored_audio = await message_audio_storage_service.upload_audio( + sender_device_id=sender_device_id, + receiver_device_id=receiver_device_id, + content=archive_audio_data, + content_type=prepared_audio.mime_type, + ) + session_logger.info( + sender_device_id, + archive_id, + f"COS 上传成功: file_key={stored_audio.file_key}", + ) + + participant_a_id, participant_b_id, pair_key = dao._build_child_peer_pair( + int(sender_identity["child_id"]), + int(receiver_identity["child_id"]), + ) + client_msg_id = f"device_voice_{uuid4().hex[:24]}" + payload = DeviceMessageCreateRequest( + conversation_type=CHILD_PEER_CONVERSATION_TYPE, + peer_child_id=int(receiver_identity["child_id"]), + content_type=2, + media_file_key=stored_audio.file_key, + media_mime_type=prepared_audio.mime_type, + media_size_bytes=prepared_audio.size_bytes, + client_msg_id=client_msg_id, + ext_json={ + "archive_source": "device_voice", + "sender_device_id": sender_device_id, + "receiver_device_id": receiver_device_id, + "local_audio_path": local_audio_path, + "archive_format": prepared_audio.archive_format, + "source_format": prepared_audio.source_format, + }, + ) + session_logger.info( + sender_device_id, + archive_id, + ( + "开始写入 IM 消息: " + f"conversation_type={CHILD_PEER_CONVERSATION_TYPE}, " + f"pair_key={pair_key}, client_msg_id={client_msg_id}" + ), + ) + conversation_id, idempotent = await dao.create_message( + conversation_type=CHILD_PEER_CONVERSATION_TYPE, + participant_a_type=CHILD_PARTICIPANT_TYPE, + participant_a_id=participant_a_id, + participant_b_type=CHILD_PARTICIPANT_TYPE, + participant_b_id=participant_b_id, + pair_key=pair_key, + sender_type=CHILD_PARTICIPANT_TYPE, + sender_id=str(sender_identity["child_id"]), + receiver_type=CHILD_PARTICIPANT_TYPE, + receiver_id=str(receiver_identity["child_id"]), + sender_name_snapshot=sender_identity["child_name"], + sender_avatar_snapshot=None, + receiver_name_snapshot=receiver_identity["child_name"], + receiver_avatar_snapshot=None, + payload=payload, + ) + message_row = await dao._get_message_by_conversation_client_id( + conversation_id=conversation_id, + client_msg_id=client_msg_id, + ) + session_logger.info( + sender_device_id, + archive_id, + ( + "IM 消息写入完成: " + f"conversation_id={conversation_id}, " + f"message_id={message_row['id'] if message_row else 'unknown'}, " + f"seq={message_row['seq'] if message_row else 'unknown'}, " + f"idempotent={idempotent}" + ), + ) + session_logger.info( + sender_device_id, + archive_id, + ( + "设备语音归档成功: " + f"sender_device_id={sender_device_id}, " + f"receiver_device_id={receiver_device_id}, " + f"conversation_id={conversation_id}, " + f"file_key={stored_audio.file_key}, " + f"archive_format={prepared_audio.archive_format}" + ), + ) + return True + except MessageAudioStorageError as exc: + session_logger.error( + sender_device_id, + archive_id, + f"设备语音归档失败: COS 上传异常: {exc}", + ) + return False + except Exception as exc: + session_logger.error( + sender_device_id, + archive_id, + f"设备语音归档失败: {exc}", + exc_info=True, + ) + return False + finally: + if prepared_audio and prepared_audio.archive_format == "mp3" and prepared_audio.filepath != local_audio_path: + if os.path.exists(prepared_audio.filepath): + os.remove(prepared_audio.filepath) + await db_session.close() + + async def _prepare_archive_audio( + self, + *, + sender_device_id: str, + archive_id: str, + local_audio_path: str, + ) -> PreparedArchiveAudio | None: + with open(local_audio_path, "rb") as local_file: + local_audio_data = local_file.read() + + source_format = detect_audio_format(local_audio_data) + local_size = len(local_audio_data) + session_logger.info( + sender_device_id, + archive_id, + ( + "异步归档开始判断音频格式: " + f"local_audio_path={local_audio_path}, source_format={source_format}, " + f"local_size_bytes={local_size}" + ), + ) + + if source_format == "mp3": + session_logger.info( + sender_device_id, + archive_id, + "异步归档判断结果: 已是 MP3，无需转码", + ) + return PreparedArchiveAudio( + filepath=local_audio_path, + mime_type="audio/mpeg", + size_bytes=local_size, + source_format=source_format, + archive_format="mp3", + ) + + wav_path = f"{local_audio_path}.archive.wav" + mp3_path = f"{local_audio_path}.archive.mp3" + try: + if source_format == "wav": + with open(wav_path, "wb") as wav_file: + wav_file.write(local_audio_data) + session_logger.info( + sender_device_id, + archive_id, + f"异步归档检测到 WAV，准备转 MP3: wav_path={wav_path}", + ) + else: + wrapped_wav = wrap_pcm_as_wav( + local_audio_data, + sample_rate=DEFAULT_SAMPLE_RATE, + channels=DEFAULT_CHANNELS, + sample_width=DEFAULT_SAMPLE_WIDTH, + ) + with open(wav_path, "wb") as wav_file: + wav_file.write(wrapped_wav) + session_logger.info( + sender_device_id, + archive_id, + ( + "异步归档将原始字节封装为 WAV: " + f"wav_path={wav_path}, sample_rate={DEFAULT_SAMPLE_RATE}, " + f"channels={DEFAULT_CHANNELS}, sample_width={DEFAULT_SAMPLE_WIDTH}" + ), + ) + + await self._convert_to_mp3( + sender_device_id=sender_device_id, + archive_id=archive_id, + source_path=wav_path, + target_path=mp3_path, + ) + mp3_size = os.path.getsize(mp3_path) + session_logger.info( + sender_device_id, + archive_id, + ( + "异步归档转码完成: " + f"mp3_path={mp3_path}, mp3_size_bytes={mp3_size}, " + f"source_format={source_format}" + ), + ) + return PreparedArchiveAudio( + filepath=mp3_path, + mime_type="audio/mpeg", + size_bytes=mp3_size, + source_format=source_format, + archive_format="mp3", + ) + finally: + if os.path.exists(wav_path): + os.remove(wav_path) + + async def _convert_to_mp3( + self, + *, + sender_device_id: str, + archive_id: str, + source_path: str, + target_path: str, + ) -> None: + ffmpeg_path = shutil.which("ffmpeg") + if not ffmpeg_path: + raise RuntimeError("ffmpeg not found in PATH") + + command = [ + ffmpeg_path, + "-y", + "-loglevel", + "error", + "-i", + source_path, + "-codec:a", + "libmp3lame", + "-b:a", + "32k", + target_path, + ] + session_logger.info( + sender_device_id, + archive_id, + f"异步归档开始转码为 MP3: command={' '.join(command)}", + ) + process = await asyncio.create_subprocess_exec( + *command, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout, stderr = await process.communicate() + if process.returncode != 0: + stderr_text = stderr.decode("utf-8", errors="ignore").strip() + raise RuntimeError( + f"ffmpeg convert failed, returncode={process.returncode}, stderr={stderr_text}" + ) + session_logger.info( + sender_device_id, + archive_id, + ( + "异步归档 FFmpeg 转码完成: " + f"source_path={source_path}, target_path={target_path}, " + f"ffmpeg_stdout={stdout.decode('utf-8', errors='ignore').strip() or 'empty'}" + ), + ) + + async def _get_child_identity_by_device_id(self, *, db_session, device_id: str): + result = await db_session.execute( + text( + """ + SELECT + da.device_id, + db.child_id, + c.child_name + FROM device_auth AS da + LEFT JOIN device_bindings AS db + ON db.device_id = da.device_id + AND db.status = 1 + LEFT JOIN children AS c + ON c.child_id = db.child_id + AND c.status = 1 + WHERE da.device_id = :device_id + AND da.is_active = 1 + LIMIT 1 + """ + ), + {"device_id": device_id}, + ) + row = result.mappings().first() + if not row or row["child_id"] is None: + return None + return { + "device_id": str(row["device_id"]), + "child_id": int(row["child_id"]), + "child_name": row["child_name"], + } + + +device_voice_archive_service = DeviceVoiceArchiveService() diff --git a/talkingq-url/scripts/simulate_device_voice_exchange.py b/talkingq-url/scripts/simulate_device_voice_exchange.py new file mode 100644 index 0000000..8c297ce --- /dev/null +++ b/talkingq-url/scripts/simulate_device_voice_exchange.py @@ -0,0 +1,348 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import asyncio +import json +import struct +import sys +import uuid +from dataclasses import dataclass, field +from pathlib import Path +from typing import Callable + +import pymysql +import websockets +from pymysql.cursors import DictCursor + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from config import settings + + +DEFAULT_WS_URL = "ws://127.0.0.1:8080/ws" + + +@dataclass +class DeviceContext: + device_id: str + serial_number: str + child_id: int + card_uuid: str + inbox: asyncio.Queue[str] = field(default_factory=asyncio.Queue) + websocket: websockets.WebSocketClientProtocol | None = None + receiver_task: asyncio.Task | None = None + + +def build_runtime_args() -> argparse.Namespace: + repo_root = Path(__file__).resolve().parents[2] + audio_dir = repo_root / "tmp_voice_sim" + + parser = argparse.ArgumentParser(description="Simulate real websocket voice exchange between two devices") + parser.add_argument("--ws-url", default=DEFAULT_WS_URL) + parser.add_argument("--device-a", default="TalkingQ_device001") + parser.add_argument("--device-b", default="TalkingQ_device002") + parser.add_argument("--a1-audio", default=str(audio_dir / "device001_to_device002_1.mp3")) + parser.add_argument("--b1-audio", default=str(audio_dir / "device002_to_device001_1.mp3")) + parser.add_argument("--a2-audio", default=str(audio_dir / "device001_to_device002_2.mp3")) + parser.add_argument("--b2-audio", default=str(audio_dir / "device002_to_device001_2.mp3")) + return parser.parse_args() + + +def connect_db(): + return pymysql.connect( + host=settings.db_host, + port=settings.db_port, + user=settings.db_user, + password=settings.db_password, + database=settings.db_name, + charset="utf8mb4", + cursorclass=DictCursor, + autocommit=True, + ) + + +def load_device_contexts(device_ids: list[str]) -> dict[str, DeviceContext]: + placeholders = ", ".join(["%s"] * len(device_ids)) + sql = f""" + SELECT + da.device_id, + da.serial_number, + db.child_id, + c.card_uuid + FROM device_auth AS da + LEFT JOIN device_bindings AS db + ON db.device_id = da.device_id + AND db.status = 1 + LEFT JOIN cards AS c + ON c.device_id = da.device_id + AND c.status = 1 + WHERE da.device_id IN ({placeholders}) + AND da.is_active = 1 + """ + + contexts: dict[str, DeviceContext] = {} + with connect_db() as connection: + with connection.cursor() as cursor: + cursor.execute(sql, device_ids) + rows = cursor.fetchall() + + for row in rows: + if row["child_id"] is None: + raise RuntimeError(f"device {row['device_id']} is not bound to any child") + if not row["card_uuid"]: + raise RuntimeError(f"device {row['device_id']} does not have an active card") + contexts[row["device_id"]] = DeviceContext( + device_id=str(row["device_id"]), + serial_number=str(row["serial_number"]), + child_id=int(row["child_id"]), + card_uuid=str(row["card_uuid"]), + ) + + missing = [device_id for device_id in device_ids if device_id not in contexts] + if missing: + raise RuntimeError(f"device context not found: {', '.join(missing)}") + return contexts + + +def fetch_conversation_snapshot(child_a_id: int, child_b_id: int) -> tuple[int | None, int, int]: + pair_key = f"{min(child_a_id, child_b_id)}:{max(child_a_id, child_b_id)}" + with connect_db() as connection: + with connection.cursor() as cursor: + cursor.execute( + """ + SELECT id, message_count + FROM im_conversations + WHERE conversation_type = 1 + AND pair_key = %s + LIMIT 1 + """, + (pair_key,), + ) + row = cursor.fetchone() + if not row: + return None, 0, 0 + + conversation_id = int(row["id"]) + message_count = int(row["message_count"]) + cursor.execute( + "SELECT COALESCE(MAX(id), 0) AS max_message_id FROM im_messages WHERE conversation_id = %s", + (conversation_id,), + ) + max_row = cursor.fetchone() or {"max_message_id": 0} + return conversation_id, message_count, int(max_row["max_message_id"] or 0) + + +def fetch_new_messages(conversation_id: int, min_message_id: int) -> list[dict]: + with connect_db() as connection: + with connection.cursor() as cursor: + cursor.execute( + """ + SELECT + id, + conversation_id, + seq, + sender_type, + sender_id, + receiver_type, + receiver_id, + content_type, + media_file_key, + client_msg_id, + created_at + FROM im_messages + WHERE conversation_id = %s + AND id > %s + ORDER BY id ASC + """, + (conversation_id, min_message_id), + ) + return cursor.fetchall() + + +async def receiver_loop(ctx: DeviceContext) -> None: + assert ctx.websocket is not None + async for message in ctx.websocket: + if isinstance(message, str): + print(f"[{ctx.device_id}] <- {message}", flush=True) + await ctx.inbox.put(message) + else: + print(f"[{ctx.device_id}] <- ", flush=True) + + +async def wait_for_text( + ctx: DeviceContext, + predicate: Callable[[str], bool], + *, + timeout: float, + description: str, +) -> str: + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while True: + remaining = deadline - loop.time() + if remaining <= 0: + raise TimeoutError(f"{ctx.device_id} timed out waiting for {description}") + message = await asyncio.wait_for(ctx.inbox.get(), timeout=remaining) + if predicate(message): + return message + + +def create_packet(device_id: str, session_id: str, sequence_number: int, packet_type: int, audio_data: bytes) -> bytes: + if len(session_id) != 32: + raise ValueError("session_id must be exactly 32 ascii chars") + return ( + device_id.encode("ascii") + + b"\x00" + + session_id.encode("ascii") + + b"\x00" + + struct.pack(" None: + websocket = await websockets.connect(ws_url, ping_interval=None, max_size=None) + ctx.websocket = websocket + ctx.receiver_task = asyncio.create_task(receiver_loop(ctx)) + await websocket.send(json.dumps({"device_id": ctx.device_id, "serial_number": ctx.serial_number}, ensure_ascii=False)) + await wait_for_text( + ctx, + lambda text: json.loads(text).get("status") == "authenticated", + timeout=10, + description="authentication response", + ) + print(f"[{ctx.device_id}] authenticated", flush=True) + + +async def send_voice_message( + sender: DeviceContext, + target: DeviceContext, + audio_path: Path, + *, + label: str, +) -> None: + if sender.websocket is None: + raise RuntimeError(f"{sender.device_id} is not connected") + if not audio_path.exists(): + raise FileNotFoundError(f"audio file not found: {audio_path}") + + audio_bytes = audio_path.read_bytes() + session_id = uuid.uuid4().hex + + print(f"[{label}] register target card {target.card_uuid}", flush=True) + await sender.websocket.send(f"REGISTER_TARGET_DEVICE:{target.card_uuid}") + await wait_for_text( + sender, + lambda text: text.startswith("TARGET_DEVICE_REGISTERED_URL:"), + timeout=5, + description="target registration response", + ) + + await sender.websocket.send(create_packet(sender.device_id, session_id, 0, 1, b"")) + print(f"[{label}] start session {session_id}", flush=True) + + chunk_size = 2048 + sequence_number = 1 + for offset in range(0, len(audio_bytes), chunk_size): + chunk = audio_bytes[offset : offset + chunk_size] + await sender.websocket.send(create_packet(sender.device_id, session_id, sequence_number, 4, chunk)) + sequence_number += 1 + await asyncio.sleep(0.03) + + await sender.websocket.send(create_packet(sender.device_id, session_id, sequence_number, 2, b"")) + print(f"[{label}] finish session {session_id}, bytes={len(audio_bytes)}", flush=True) + + await wait_for_text( + sender, + lambda text: text.startswith("PROMPT_SOUND_URL:"), + timeout=20, + description="message stored response", + ) + await asyncio.sleep(0.5) + + +async def close_device(ctx: DeviceContext) -> None: + if ctx.websocket is not None: + await ctx.websocket.close() + if ctx.receiver_task is not None: + try: + await asyncio.wait_for(ctx.receiver_task, timeout=2) + except Exception: + ctx.receiver_task.cancel() + + +async def main() -> None: + args = build_runtime_args() + audio_paths = { + "a1": Path(args.a1_audio), + "b1": Path(args.b1_audio), + "a2": Path(args.a2_audio), + "b2": Path(args.b2_audio), + } + + contexts = load_device_contexts([args.device_a, args.device_b]) + device_a = contexts[args.device_a] + device_b = contexts[args.device_b] + + conversation_id, message_count_before, max_message_id_before = fetch_conversation_snapshot( + device_a.child_id, + device_b.child_id, + ) + print( + f"[snapshot-before] conversation_id={conversation_id} message_count={message_count_before} max_message_id={max_message_id_before}", + flush=True, + ) + + await connect_device(device_a, args.ws_url) + await connect_device(device_b, args.ws_url) + + try: + await send_voice_message(device_a, device_b, audio_paths["a1"], label="A->B #1") + await send_voice_message(device_b, device_a, audio_paths["b1"], label="B->A #1") + await send_voice_message(device_a, device_b, audio_paths["a2"], label="A->B #2") + await send_voice_message(device_b, device_a, audio_paths["b2"], label="B->A #2") + finally: + await close_device(device_a) + await close_device(device_b) + + await asyncio.sleep(2) + + conversation_id_after, message_count_after, max_message_id_after = fetch_conversation_snapshot( + device_a.child_id, + device_b.child_id, + ) + print( + f"[snapshot-after] conversation_id={conversation_id_after} message_count={message_count_after} max_message_id={max_message_id_after}", + flush=True, + ) + if conversation_id_after is None: + raise RuntimeError("child-peer conversation was not created") + + new_messages = fetch_new_messages(conversation_id_after, max_message_id_before) + print(f"[new-messages] count={len(new_messages)}", flush=True) + for row in new_messages: + print( + json.dumps( + { + "id": int(row["id"]), + "seq": int(row["seq"]), + "sender_id": row["sender_id"], + "receiver_id": row["receiver_id"], + "content_type": int(row["content_type"]), + "media_file_key": row["media_file_key"], + "client_msg_id": row["client_msg_id"], + "created_at": row["created_at"].isoformat(sep=" "), + }, + ensure_ascii=False, + ), + flush=True, + ) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/talkingq-url/utils/audio_format.py b/talkingq-url/utils/audio_format.py new file mode 100644 index 0000000..8468df8 --- /dev/null +++ b/talkingq-url/utils/audio_format.py @@ -0,0 +1,33 @@ +import io +import wave + + +DEFAULT_SAMPLE_RATE = 16000 +DEFAULT_CHANNELS = 1 +DEFAULT_SAMPLE_WIDTH = 2 + + +def detect_audio_format(audio_data: bytes) -> str: + if len(audio_data) >= 12 and audio_data[:4] == b"RIFF" and audio_data[8:12] == b"WAVE": + return "wav" + if audio_data.startswith(b"ID3"): + return "mp3" + if len(audio_data) >= 2 and audio_data[0] == 0xFF and (audio_data[1] & 0xE0) == 0xE0: + return "mp3" + return "pcm_s16le_16k_mono" + + +def wrap_pcm_as_wav( + audio_data: bytes, + *, + sample_rate: int = DEFAULT_SAMPLE_RATE, + channels: int = DEFAULT_CHANNELS, + sample_width: int = DEFAULT_SAMPLE_WIDTH, +) -> bytes: + wav_buffer = io.BytesIO() + with wave.open(wav_buffer, "wb") as wav_file: + wav_file.setnchannels(channels) + wav_file.setsampwidth(sample_width) + wav_file.setframerate(sample_rate) + wav_file.writeframes(audio_data) + return wav_buffer.getvalue()