待收听留言入库后即时推送提醒

talkingq-url/banban/service/pending_voice_message.py

@@ -3,6 +3,7 @@ from dataclasses import dataclass
 from banban.dao.im import ImDAO
 from banban.dao.pending_voice_message import PendingVoiceMessageDAO
 from banban.service.device_audio_cache import device_audio_cache_service
+from config import settings
 from services.database_service_base import DatabaseServiceBase
 from utils.logger import session_logger
 
@@ -28,6 +29,43 @@ class PendingVoiceMessageService(DatabaseServiceBase):
     def __init__(self) -> None:
         super().__init__(service_name="pending_voice_message")
 
+    def get_pending_notice_audio_url(self) -> str:
+        return f"http://{settings.server_host}:{settings.server_port}/assets/audio/new_message.mp3"
+
+    async def notify_device_pending_message(self, *, target_device_id: str) -> bool:
+        normalized_target_device_id = (target_device_id or "").strip()
+        if not normalized_target_device_id:
+            return False
+
+        try:
+            from handlers.mqtt_handler import TalkingQMQTTService
+
+            service = await TalkingQMQTTService.get_instance()
+            if service is None:
+                session_logger.warning(
+                    normalized_target_device_id,
+                    "pending_voice",
+                    "MQTT服务未初始化，跳过待收听留言提醒",
+                )
+                return False
+
+            audio_url = self.get_pending_notice_audio_url()
+            await service.send_nfc_notice(normalized_target_device_id, audio_url)
+            session_logger.info(
+                normalized_target_device_id,
+                "pending_voice",
+                f"待收听留言提醒已发送: audio_url={audio_url}",
+            )
+            return True
+        except Exception as exc:
+            session_logger.error(
+                normalized_target_device_id,
+                "pending_voice",
+                f"待收听留言提醒发送失败: {exc}",
+                exc_info=True,
+            )
+            return False
+
     async def add_pending_message(
         self,
         *,
@@ -69,6 +107,10 @@ class PendingVoiceMessageService(DatabaseServiceBase):
         finally:
             await db_session.close()
 
+        await self.notify_device_pending_message(
+            target_device_id=normalized_target_device_id,
+        )
+
     async def get_playback_items(
         self,
         *,

talkingq-url/services/scheduler.py

@@ -1,13 +1,7 @@
-import logging
-from typing import Optional, Dict
-
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from apscheduler.triggers.interval import IntervalTrigger
 from banban.service.pending_voice_message import pending_voice_message_service
-from services.connection_manager import connection_manager
-from handlers.mqtt_handler import TalkingQMQTTService
 from services.offline_audio_cache import offline_audio_cache
-from config import settings
 from utils.logger import session_logger as logger
 
 class TaskScheduler:
@@ -40,7 +34,6 @@ class TaskScheduler:
 
     async def _execute_task(self):
         try:
-            service = await TalkingQMQTTService.get_instance()
             pending_devices = await pending_voice_message_service.list_devices_with_pending()
             audio_cache = await offline_audio_cache.get_all_audio_cache()
             fallback_devices = list(audio_cache.keys()) if audio_cache else []
@@ -49,13 +42,13 @@ class TaskScheduler:
                 logger.warning("", "", "无离线音频缓存，跳过本次执行")
                 return
             for device_id in device_ids:
-                websocket = await connection_manager.get_connection(device_id)
-                if websocket is not None:
-                    audio_url = f"http://{settings.server_host}:{settings.server_port}/assets/audio/new_message.mp3"
-                    await service.send_nfc_notice(device_id, audio_url)
-                    logger.info(device_id, "", f"[定时任务] 发送音频成功: device={device_id}, audio_url={audio_url}")
+                sent = await pending_voice_message_service.notify_device_pending_message(
+                    target_device_id=device_id,
+                )
+                if sent:
+                    logger.info(device_id, "", f"[定时任务] 发送待收听留言提醒成功: device={device_id}")
                 else:
-                    logger.info(device_id, "", f"[定时任务] 跳过音频发送: 设备 {device_id} 离线")
+                    logger.info(device_id, "", f"[定时任务] 待收听留言提醒未发送: device={device_id}")
         except Exception as e:
             logger.error("", "", f"[定时任务] 执行失败: {e}")
 

talkingq-url/tests/test_pending_voice_notice.py

@@ -0,0 +1,99 @@
+from types import SimpleNamespace
+
+import pytest
+
+from banban.service.pending_voice_message import PendingVoiceMessageService
+from services.scheduler import TaskScheduler
+
+
+class FakeSession:
+    def __init__(self):
+        self.commits = 0
+        self.rollbacks = 0
+        self.closed = False
+
+    async def execute(self, statement, params=None):
+        return SimpleNamespace(mappings=lambda: SimpleNamespace(all=lambda: []))
+
+    async def commit(self):
+        self.commits += 1
+
+    async def rollback(self):
+        self.rollbacks += 1
+
+    async def close(self):
+        self.closed = True
+
+
+@pytest.mark.asyncio
+async def test_add_pending_message_sends_mqtt_notice_after_commit(monkeypatch):
+    service = PendingVoiceMessageService()
+    session = FakeSession()
+    upserts = []
+    notices = []
+
+    async def fake_get_session():
+        return session
+
+    async def fake_upsert_pending(self, **kwargs):
+        upserts.append(kwargs)
+
+    async def fake_notify_device_pending_message(*, target_device_id):
+        notices.append(target_device_id)
+        return True
+
+    monkeypatch.setattr(service, "get_session", fake_get_session)
+    monkeypatch.setattr(
+        "banban.service.pending_voice_message.PendingVoiceMessageDAO.upsert_pending",
+        fake_upsert_pending,
+    )
+    monkeypatch.setattr(service, "notify_device_pending_message", fake_notify_device_pending_message)
+
+    await service.add_pending_message(
+        target_device_id=" TalkingQ_device001 ",
+        sender_device_id=" TalkingQ_device002 ",
+        im_message_id=123,
+        media_file_key="messages/audio/demo.mp3",
+        audio_url="http://example.test/demo.mp3",
+        source="parent_child_voice",
+    )
+
+    assert session.commits == 1
+    assert session.rollbacks == 0
+    assert session.closed is True
+    assert upserts == [
+        {
+            "target_device_id": "TalkingQ_device001",
+            "sender_device_id": "TalkingQ_device002",
+            "im_message_id": 123,
+            "media_file_key": "messages/audio/demo.mp3",
+            "audio_url": "http://example.test/demo.mp3",
+            "source": "parent_child_voice",
+        }
+    ]
+    assert notices == ["TalkingQ_device001"]
+
+
+@pytest.mark.asyncio
+async def test_scheduler_sends_pending_notice_without_websocket_check(monkeypatch):
+    scheduler = TaskScheduler()
+    calls = []
+
+    class FakePendingVoiceService:
+        async def list_devices_with_pending(self):
+            return ["TalkingQ_device001"]
+
+        async def notify_device_pending_message(self, *, target_device_id):
+            calls.append(target_device_id)
+            return True
+
+    class FakeOfflineAudioCache:
+        async def get_all_audio_cache(self):
+            return {}
+
+    monkeypatch.setattr("services.scheduler.pending_voice_message_service", FakePendingVoiceService())
+    monkeypatch.setattr("services.scheduler.offline_audio_cache", FakeOfflineAudioCache())
+
+    await scheduler._execute_task()
+
+    assert calls == ["TalkingQ_device001"]