banban/talkingq-url/banban/service/pending_voice_message.py

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


@dataclass(frozen=True)
class PendingVoicePlaybackItem:
    pending_id: int
    audio_url: str
    im_message_id: int
    media_file_key: str
    source: str


@dataclass(frozen=True)
class LatestVoicePlaybackItem:
    audio_url: str
    im_message_id: int
    media_file_key: str
    source: str = "latest_history_fallback"


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

        await self.notify_device_pending_message(
            target_device_id=normalized_target_device_id,
        )

    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 get_latest_history_playback_item(
        self,
        *,
        target_device_id: str,
    ) -> LatestVoicePlaybackItem | None:
        db_session = await self.get_session()
        try:
            dao = ImDAO(db_session)
            row = await dao.get_latest_voice_message_received_by_device(
                device_id=target_device_id,
            )
        finally:
            await db_session.close()

        if not row:
            return None

        media_file_key = str(row["media_file_key"] or "").strip()
        if not media_file_key:
            return None
        try:
            audio_url = await device_audio_cache_service.get_device_audio_url(
                media_file_key,
                device_id=target_device_id,
            )
        except Exception as exc:
            session_logger.error(
                target_device_id,
                "pending_voice",
                (
                    "最近历史留言音频URL生成失败: "
                    f"message_id={row['id']}, media_file_key={media_file_key}, error={exc}"
                ),
                exc_info=True,
            )
            return None

        return LatestVoicePlaybackItem(
            audio_url=audio_url,
            im_message_id=int(row["id"]),
            media_file_key=media_file_key,
        )

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