diff --git a/talkingq-url/banban/service/pending_voice_message.py b/talkingq-url/banban/service/pending_voice_message.py index b756458..5471aa4 100644 --- a/talkingq-url/banban/service/pending_voice_message.py +++ b/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, *, diff --git a/talkingq-url/services/scheduler.py b/talkingq-url/services/scheduler.py index 7ec7762..717a861 100644 --- a/talkingq-url/services/scheduler.py +++ b/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}") diff --git a/talkingq-url/tests/test_pending_voice_notice.py b/talkingq-url/tests/test_pending_voice_notice.py new file mode 100644 index 0000000..fb71c03 --- /dev/null +++ b/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"]