补充设备语音归档与联调脚本

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()

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}] <- <binary:{len(message)}>", 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("<I", sequence_number)
+        + struct.pack("<B", packet_type)
+        + struct.pack("<I", len(audio_data))
+        + audio_data
+    )
+
+
+async def connect_device(ctx: DeviceContext, ws_url: str) -> 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())

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()