合入持久化刷卡语音消息的代码，暂未测试

talkingq-url/banban/service/ pending_voice_message.py

@@ -0,0 +1,146 @@
+from dataclasses import dataclass
+
+from banban.dao.pending_voice_message import PendingVoiceMessageDAO
+from banban.service.device_audio_cache import device_audio_cache_service
+from services.database_service_base import DatabaseServiceBase
+from utils.logger import session_logger
+
+
+@dataclass(frozen=True)
+class PendingVoicePlaybackItem:
+    pending_id: int
+    audio_url: str
+    im_message_id: int
+    media_file_key: str
+    source: str
+
+
+class PendingVoiceMessageService(DatabaseServiceBase):
+    def __init__(self) -> None:
+        super().__init__(service_name="pending_voice_message")
+
+    async def add_pending_message(
+        self,
+        *,
+        target_device_id: str,
+        im_message_id: int,
+        media_file_key: str,
+        sender_device_id: str | None = None,
+        audio_url: str | None = None,
+        source: str = "unknown",
+    ) -> None:
+        normalized_target_device_id = (target_device_id or "").strip()
+        normalized_media_file_key = (media_file_key or "").strip()
+        if not normalized_target_device_id or not normalized_media_file_key:
+            raise ValueError("target_device_id and media_file_key are required")
+
+        db_session = await self.get_session()
+        try:
+            dao = PendingVoiceMessageDAO(db_session)
+            await dao.upsert_pending(
+                target_device_id=normalized_target_device_id,
+                sender_device_id=(sender_device_id or "").strip() or None,
+                im_message_id=int(im_message_id),
+                media_file_key=normalized_media_file_key,
+                audio_url=(audio_url or "").strip() or None,
+                source=source,
+            )
+            await db_session.commit()
+            session_logger.info(
+                normalized_target_device_id,
+                "pending_voice",
+                (
+                    "待收听留言已持久化: "
+                    f"im_message_id={im_message_id}, source={source}"
+                ),
+            )
+        except Exception:
+            await db_session.rollback()
+            raise
+        finally:
+            await db_session.close()
+
+    async def get_playback_items(
+        self,
+        *,
+        target_device_id: str,
+        limit: int = 10,
+    ) -> list[PendingVoicePlaybackItem]:
+        db_session = await self.get_session()
+        try:
+            dao = PendingVoiceMessageDAO(db_session)
+            rows = await dao.list_pending_for_delivery(
+                target_device_id=target_device_id,
+                limit=limit,
+            )
+        finally:
+            await db_session.close()
+
+        items: list[PendingVoicePlaybackItem] = []
+        for row in rows:
+            media_file_key = str(row["media_file_key"])
+            fallback_audio_url = (row["audio_url"] or "").strip()
+            try:
+                audio_url = await device_audio_cache_service.get_device_audio_url(
+                    media_file_key,
+                    device_id=target_device_id,
+                )
+            except Exception as exc:
+                if fallback_audio_url:
+                    audio_url = fallback_audio_url
+                else:
+                    session_logger.error(
+                        target_device_id,
+                        "pending_voice",
+                        f"待收听留言音频URL生成失败: media_file_key={media_file_key}, error={exc}",
+                        exc_info=True,
+                    )
+                    continue
+            items.append(
+                PendingVoicePlaybackItem(
+                    pending_id=int(row["id"]),
+                    audio_url=audio_url,
+                    im_message_id=int(row["im_message_id"]),
+                    media_file_key=media_file_key,
+                    source=str(row["source"] or "unknown"),
+                )
+            )
+        return items
+
+    async def mark_delivered(self, pending_ids: list[int]) -> None:
+        if not pending_ids:
+            return
+        db_session = await self.get_session()
+        try:
+            dao = PendingVoiceMessageDAO(db_session)
+            await dao.mark_delivered(pending_ids)
+            await db_session.commit()
+        except Exception:
+            await db_session.rollback()
+            raise
+        finally:
+            await db_session.close()
+
+    async def clear_pending_for_device(self, *, target_device_id: str) -> None:
+        db_session = await self.get_session()
+        try:
+            dao = PendingVoiceMessageDAO(db_session)
+            await dao.clear_pending_for_device(target_device_id=target_device_id)
+            await db_session.commit()
+        except Exception:
+            await db_session.rollback()
+            raise
+        finally:
+            await db_session.close()
+
+    async def list_devices_with_pending(self, *, limit: int = 1000) -> list[str]:
+        db_session = await self.get_session()
+        try:
+            dao = PendingVoiceMessageDAO(db_session)
+            rows = await dao.list_devices_with_pending(limit=limit)
+        finally:
+            await db_session.close()
+        return [str(row["target_device_id"]) for row in rows]
+
+
+pending_voice_message_service = PendingVoiceMessageService()

talkingq-url/banban/service/binding.py

@@ -239,3 +239,6 @@ class BindingService(DatabaseServiceBase):
             return rows, has_more
         finally:
             await db_session.close()
+
+
+binding_service = BindingService()

talkingq-url/banban/service/device_voice_archive.py

@@ -37,6 +37,15 @@ class PreparedArchiveAudio:
     archive_format: str
 
 
+@dataclass(frozen=True)
+class DeviceVoiceArchiveResult:
+    sender_device_id: str
+    receiver_device_id: str
+    conversation_id: int
+    message_id: int
+    media_file_key: str
+
+
 class DeviceVoiceArchiveService(DatabaseServiceBase):
     def __init__(self) -> None:
         super().__init__(service_name="device_voice_archive")
@@ -47,7 +56,7 @@ class DeviceVoiceArchiveService(DatabaseServiceBase):
         sender_device_id: str,
         receiver_device_id: str,
         local_audio_path: str,
-    ) -> bool:
+    ) -> DeviceVoiceArchiveResult | None:
         archive_id = f"voice_archive:{uuid4().hex[:12]}"
         session_logger.info(
             sender_device_id,
@@ -65,7 +74,7 @@ class DeviceVoiceArchiveService(DatabaseServiceBase):
                 archive_id,
                 f"跳过语音归档: 本地音频文件不存在, local_audio_path={local_audio_path}",
             )
-            return False
+            return None
 
         prepared_audio = None
         db_session = await self.get_session()
@@ -76,7 +85,7 @@ class DeviceVoiceArchiveService(DatabaseServiceBase):
                 local_audio_path=local_audio_path,
             )
             if prepared_audio is None:
-                return False
+                return None
 
             dao = ImDAO(db_session)
             sender_identity = await self._get_child_identity_by_device_id(
@@ -94,14 +103,14 @@ class DeviceVoiceArchiveService(DatabaseServiceBase):
                     archive_id,
                     f"跳过语音归档: 发送设备未绑定有效 child, device_id={sender_device_id}",
                 )
-                return False
+                return None
             if receiver_identity is None:
                 session_logger.warning(
                     sender_device_id,
                     archive_id,
                     f"跳过语音归档: 接收设备未绑定有效 child, device_id={receiver_device_id}",
                 )
-                return False
+                return None
             if sender_identity["child_id"] == receiver_identity["child_id"]:
                 session_logger.warning(
                     sender_device_id,
@@ -111,7 +120,7 @@ class DeviceVoiceArchiveService(DatabaseServiceBase):
                         f"child_id={sender_identity['child_id']}"
                     ),
                 )
-                return False
+                return None
 
             session_logger.info(
                 sender_device_id,
@@ -228,14 +237,22 @@ class DeviceVoiceArchiveService(DatabaseServiceBase):
                     f"archive_format={prepared_audio.archive_format}"
                 ),
             )
-            return True
+            if not message_row:
+                raise RuntimeError("message was not found after insert")
+            return DeviceVoiceArchiveResult(
+                sender_device_id=sender_device_id,
+                receiver_device_id=receiver_device_id,
+                conversation_id=int(conversation_id),
+                message_id=int(message_row["id"]),
+                media_file_key=stored_audio.file_key,
+            )
         except MessageAudioStorageError as exc:
             session_logger.error(
                 sender_device_id,
                 archive_id,
                 f"设备语音归档失败: COS 上传异常: {exc}",
             )
-            return False
+            return None
         except Exception as exc:
             session_logger.error(
                 sender_device_id,
@@ -243,7 +260,7 @@ class DeviceVoiceArchiveService(DatabaseServiceBase):
                 f"设备语音归档失败: {exc}",
                 exc_info=True,
             )
-            return False
+            return None
         finally:
             if prepared_audio and prepared_audio.archive_format == "mp3" and prepared_audio.filepath != local_audio_path:
                 if os.path.exists(prepared_audio.filepath):

talkingq-url/banban/service/im.py

@@ -4,12 +4,12 @@ import json
 from collections.abc import Mapping
 from pathlib import Path
 from typing import Any
-from services.offline_audio_cache import offline_audio_cache
 from fastapi import HTTPException
 from services.database_service_base import DatabaseServiceBase
 from banban.service.device_audio_cache import device_audio_cache_service
 from banban.service.message_audio_storage import MessageAudioStorageService, MessageAudioStorageError
-from banban.service.binding import BindingService
+from banban.service.binding import binding_service
+from banban.service.pending_voice_message import pending_voice_message_service
 try:
     from banban.dao.im import ImDAO, DeviceIdentity, ConversationMessageCreateResult
     from banban.schemas.im import (
@@ -223,6 +223,25 @@ class ImService(DatabaseServiceBase):
             message=presented_message,
         )
 
+    async def _get_raw_message_media_file_key(
+        self,
+        *,
+        conversation_id: int,
+        client_msg_id: str,
+    ) -> str | None:
+        db_session = await self.get_session()
+        try:
+            dao = ImDAO(db_session)
+            message_row = await dao._get_message_by_conversation_client_id(
+                conversation_id=conversation_id,
+                client_msg_id=client_msg_id,
+            )
+            if not message_row:
+                return None
+            return message_row["media_file_key"]
+        finally:
+            await db_session.close()
+
     async def create_parent_child_message(
         self,
         *,
@@ -305,28 +324,47 @@ class ImService(DatabaseServiceBase):
             ext_json=ext_json,
         )
 
+        result = None
         try:
             result = await self.create_parent_child_message(
                 parent_user_id=parent_user_id,
                 child_id=child_id,
                 payload=payload,
             )
-            binding_service = BindingService()
+            pending_media_file_key = await self._get_raw_message_media_file_key(
+                conversation_id=result.conversation_id,
+                client_msg_id=client_msg_id,
+            )
+            if not pending_media_file_key:
+                pending_media_file_key = stored_audio.file_key
+            if result.idempotent:
+                try:
+                    await self.audio_storage.delete_audio(stored_audio.file_key)
+                except MessageAudioStorageError:
+                    pass
             device = await binding_service.get_current_binding(parent_user_id)
             try:
                 audio_url = await device_audio_cache_service.get_device_audio_url(
-                    stored_audio.file_key,
+                    pending_media_file_key,
                     device_id=device.device_id,
                 )
             except Exception:
-                session_logger.error(device.device_id, "audio", f"failed to get device audio url: {stored_audio.file_key}", exc_info=True)
-                audio_url = stored_audio.file_key
-            await offline_audio_cache.add_audio_url(device.device_id, f"{audio_url}")
+                session_logger.error(device.device_id, "audio", f"failed to get device audio url: {pending_media_file_key}", exc_info=True)
+                audio_url = pending_media_file_key
+            await pending_voice_message_service.add_pending_message(
+                target_device_id=device.device_id,
+                sender_device_id=None,
+                im_message_id=result.message.id,
+                media_file_key=pending_media_file_key,
+                audio_url=f"{audio_url}",
+                source="parent_child_voice",
+            )
         except Exception:
-            try:
-                await self.audio_storage.delete_audio(stored_audio.file_key)
-            except MessageAudioStorageError:
-                pass
+            if result is None:
+                try:
+                    await self.audio_storage.delete_audio(stored_audio.file_key)
+                except MessageAudioStorageError:
+                    pass
             raise
 
         return result