From 1ffad03801684f3eb99709ee2d20388678e0887b Mon Sep 17 00:00:00 2001 From: HycJack <772403255@qq.com> Date: Tue, 26 May 2026 02:47:59 +0800 Subject: [PATCH] =?UTF-8?q?=E5=90=88=E5=85=A5=E6=8C=81=E4=B9=85=E5=8C=96?= =?UTF-8?q?=E5=88=B7=E5=8D=A1=E8=AF=AD=E9=9F=B3=E6=B6=88=E6=81=AF=E7=9A=84?= =?UTF-8?q?=E4=BB=A3=E7=A0=81=EF=BC=8C=E6=9A=82=E6=9C=AA=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../banban/dao/pending_voice_message.py | 175 ++++++ talkingq-url/banban/routers/bindings.py | 32 +- talkingq-url/banban/routers/mqtt_router.py | 4 + .../banban/service/ pending_voice_message.py | 146 +++++ talkingq-url/banban/service/binding.py | 3 + .../banban/service/device_voice_archive.py | 35 +- talkingq-url/banban/service/im.py | 60 +- talkingq-url/database/models.py | 28 + talkingq-url/handlers/mqtt_handler.py | 50 +- .../handlers/websocket_message_handler.py | 39 +- talkingq-url/main.py | 13 +- talkingq-url/mysql/init/02-init.sql | 534 ++++++++++++++++++ .../services/database_service_base.py | 6 +- talkingq-url/services/scheduler.py | 8 +- 14 files changed, 1065 insertions(+), 68 deletions(-) create mode 100644 talkingq-url/banban/dao/pending_voice_message.py create mode 100644 talkingq-url/banban/service/ pending_voice_message.py create mode 100644 talkingq-url/mysql/init/02-init.sql diff --git a/talkingq-url/banban/dao/pending_voice_message.py b/talkingq-url/banban/dao/pending_voice_message.py new file mode 100644 index 0000000..abf6622 --- /dev/null +++ b/talkingq-url/banban/dao/pending_voice_message.py @@ -0,0 +1,175 @@ +from collections.abc import Mapping +from typing import Any + +from banban.dao import BaseDAO + + +PENDING_VOICE_STATUS_PENDING = "pending" +PENDING_VOICE_STATUS_DELIVERED = "delivered" +PENDING_VOICE_STATUS_LISTENED = "listened" +PENDING_VOICE_STATUS_EXPIRED = "expired" + + +class PendingVoiceMessageDAO(BaseDAO): + async def upsert_pending( + 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: + await self.execute( + """ + INSERT INTO device_pending_voice_messages ( + target_device_id, + sender_device_id, + im_message_id, + media_file_key, + audio_url, + source, + status + ) VALUES ( + :target_device_id, + :sender_device_id, + :im_message_id, + :media_file_key, + :audio_url, + :source, + :status + ) + ON DUPLICATE KEY UPDATE + sender_device_id = COALESCE(sender_device_id, VALUES(sender_device_id)), + audio_url = COALESCE(audio_url, VALUES(audio_url)), + source = VALUES(source), + updated_at = CURRENT_TIMESTAMP + """, + { + "target_device_id": target_device_id, + "sender_device_id": sender_device_id, + "im_message_id": im_message_id, + "media_file_key": media_file_key, + "audio_url": audio_url, + "source": source, + "status": PENDING_VOICE_STATUS_PENDING, + }, + ) + + async def list_pending_for_delivery( + self, + *, + target_device_id: str, + limit: int = 10, + ) -> list[Mapping[str, Any]]: + rows = ( + await self.execute( + """ + SELECT + id, + target_device_id, + sender_device_id, + im_message_id, + media_file_key, + audio_url, + source, + status, + delivery_count, + created_at + FROM device_pending_voice_messages + WHERE target_device_id = :target_device_id + AND status = :status + ORDER BY created_at ASC, id ASC + LIMIT :limit + """, + { + "target_device_id": target_device_id, + "status": PENDING_VOICE_STATUS_PENDING, + "limit": limit, + }, + ) + ).mappings().all() + return rows + + async def list_devices_with_pending(self, *, limit: int = 1000) -> list[Mapping[str, Any]]: + rows = ( + await self.execute( + """ + SELECT + target_device_id, + COUNT(*) AS pending_count, + MIN(created_at) AS first_pending_at + FROM device_pending_voice_messages + WHERE status = :status + GROUP BY target_device_id + ORDER BY first_pending_at ASC + LIMIT :limit + """, + { + "status": PENDING_VOICE_STATUS_PENDING, + "limit": limit, + }, + ) + ).mappings().all() + return rows + + async def mark_delivered(self, message_ids: list[int]) -> None: + if not message_ids: + return + placeholders, params = self._build_id_params(message_ids) + await self.execute( + f""" + UPDATE device_pending_voice_messages + SET status = :status, + delivery_count = delivery_count + 1, + delivered_at = COALESCE(delivered_at, CURRENT_TIMESTAMP), + updated_at = CURRENT_TIMESTAMP + WHERE id IN ({placeholders}) + AND status = :pending_status + """, + { + **params, + "status": PENDING_VOICE_STATUS_DELIVERED, + "pending_status": PENDING_VOICE_STATUS_PENDING, + }, + ) + + async def mark_listened(self, message_ids: list[int]) -> None: + if not message_ids: + return + placeholders, params = self._build_id_params(message_ids) + await self.execute( + f""" + UPDATE device_pending_voice_messages + SET status = :status, + listened_at = COALESCE(listened_at, CURRENT_TIMESTAMP), + updated_at = CURRENT_TIMESTAMP + WHERE id IN ({placeholders}) + """, + { + **params, + "status": PENDING_VOICE_STATUS_LISTENED, + }, + ) + + async def clear_pending_for_device(self, *, target_device_id: str) -> None: + await self.execute( + """ + UPDATE device_pending_voice_messages + SET status = :status, + updated_at = CURRENT_TIMESTAMP + WHERE target_device_id = :target_device_id + AND status = :pending_status + """, + { + "target_device_id": target_device_id, + "status": PENDING_VOICE_STATUS_EXPIRED, + "pending_status": PENDING_VOICE_STATUS_PENDING, + }, + ) + + def _build_id_params(self, message_ids: list[int]) -> tuple[str, dict[str, int]]: + params = {f"id_{index}": int(message_id) for index, message_id in enumerate(message_ids)} + placeholders = ", ".join(f":{key}" for key in params) + return placeholders, params diff --git a/talkingq-url/banban/routers/bindings.py b/talkingq-url/banban/routers/bindings.py index 3104c04..d880297 100644 --- a/talkingq-url/banban/routers/bindings.py +++ b/talkingq-url/banban/routers/bindings.py @@ -12,7 +12,7 @@ from banban.dao.binding import ( SESSION_STATUS_PENDING, ) from banban.security import get_current_user_id -from banban.service.binding import BindingError, BindingService +from banban.service.binding import BindingError, binding_service router = APIRouter(prefix="/bindings", tags=["bindings"]) @@ -91,9 +91,8 @@ async def start_bind( current_user_id: int = Depends(get_current_user_id), ) -> BindStartResponse: del request - service = BindingService() try: - bind_token, expires_at = await service.start_bind( + bind_token, expires_at = await binding_service.start_bind( current_user_id, payload.device_id, payload.serial_number, @@ -115,8 +114,7 @@ async def get_bind_session( current_user_id: int = Depends(get_current_user_id), ) -> BindSessionResponse: del request - service = BindingService() - session = await service.get_bind_session(bind_token, current_user_id) + session = await binding_service.get_bind_session(bind_token, current_user_id) if not session: raise HTTPException(status_code=404, detail="bind session not found") @@ -137,9 +135,8 @@ async def confirm_bind( current_user_id: int = Depends(get_current_user_id), ) -> BindConfirmResponse: del request - service = BindingService() try: - result = await service.confirm_bind(payload.bind_token, current_user_id) + result = await binding_service.confirm_bind(payload.bind_token, current_user_id) except BindingError as exc: raise HTTPException(status_code=exc.status_code, detail=str(exc)) except ValueError as exc: @@ -169,9 +166,8 @@ async def direct_bind( current_user_id: int = Depends(get_current_user_id), ) -> DirectBindResponse: del request - service = BindingService() try: - result = await service.direct_bind( + result = await binding_service.direct_bind( payload.device_id, payload.serial_number, payload.child_id, @@ -190,9 +186,8 @@ async def set_binding_child( current_user_id: int = Depends(get_current_user_id), ) -> DirectBindResponse: del request - service = BindingService() try: - result = await service.set_binding_child(device_id=device_id, child_id=payload.child_id, user_id=current_user_id) + result = await binding_service.set_binding_child(device_id=device_id, child_id=payload.child_id, user_id=current_user_id) except BindingError as exc: raise HTTPException(status_code=exc.status_code, detail=str(exc)) except ValueError as exc: @@ -206,8 +201,7 @@ async def get_current_binding( current_user_id: int = Depends(get_current_user_id), ): del request - service = BindingService() - binding = await service.get_current_binding(current_user_id) + binding = await binding_service.get_current_binding(current_user_id) if not binding: raise HTTPException(status_code=404, detail="no binding found") return binding @@ -221,8 +215,7 @@ async def list_bindings( current_user_id: int = Depends(get_current_user_id), ) -> BindingListResponse: del request - service = BindingService() - rows, has_more = await service.list_bindings(current_user_id, limit, cursor) + rows, has_more = await binding_service.list_bindings(current_user_id, limit, cursor) next_cursor = int(rows[-1]["id"]) if has_more and rows else None items = [ BindingListItem( @@ -244,8 +237,7 @@ async def get_binding( current_user_id: int = Depends(get_current_user_id), ) -> BindingGetResponse: del request - service = BindingService() - binding = await service.get_binding(device_id, current_user_id) + binding = await binding_service.get_binding(device_id, current_user_id) if not binding: raise HTTPException(status_code=404, detail="binding not found") return BindingGetResponse(**binding) @@ -258,8 +250,7 @@ async def unbind_device( current_user_id: int = Depends(get_current_user_id), ) -> None: del request - service = BindingService() - if not await service.unbind(device_id, current_user_id): + if not await binding_service.unbind(device_id, current_user_id): raise HTTPException(status_code=404, detail="binding not found") @@ -273,8 +264,7 @@ async def get_bind_history( ) -> BindHistoryResponse: del request, current_user_id cursor_dt = datetime.fromisoformat(cursor) if cursor else None - service = BindingService() - rows, has_more = await service.list_history(device_id, limit, cursor_dt) + rows, has_more = await binding_service.list_history(device_id, limit, cursor_dt) next_cursor = rows[-1]["bound_at"].isoformat() if has_more and rows else None items = [BindHistoryItem(**row) for row in rows] return BindHistoryResponse(items=items, total=len(items), next_cursor=next_cursor) diff --git a/talkingq-url/banban/routers/mqtt_router.py b/talkingq-url/banban/routers/mqtt_router.py index 0f39a8a..7000fb6 100644 --- a/talkingq-url/banban/routers/mqtt_router.py +++ b/talkingq-url/banban/routers/mqtt_router.py @@ -2,6 +2,8 @@ import logging from typing import Optional from fastapi import APIRouter, HTTPException, Query from banban.service.device_audio_cache import device_audio_cache_service +from banban.service.pending_voice_message import pending_voice_message_service +from services.device_volume_manager import device_volume_manager from services.offline_audio_cache import offline_audio_cache from banban.schemas.mqtt_models import ( GPSQueryRequest, @@ -51,6 +53,7 @@ async def set_volume( req: VolumeRequest, # current_user_id: int = Depends(get_current_user_id) ): + await device_volume_manager.set_volume(req.device_id, req.level) service = await _get_service() msg_id = await service.send_volume_command(req.device_id, req.level) return CommandResponse(msg_id=msg_id, device_id=req.device_id) @@ -110,6 +113,7 @@ async def clear_nfc_unread( # current_user_id: int = Depends(get_current_user_id) ): await offline_audio_cache.clear_audio_urls(device_id) + await pending_voice_message_service.clear_pending_for_device(target_device_id=device_id) return CommandResponse(msg_id="", device_id=device_id, message="已清除未读留言") diff --git a/talkingq-url/banban/service/ pending_voice_message.py b/talkingq-url/banban/service/ pending_voice_message.py new file mode 100644 index 0000000..f7729ba --- /dev/null +++ b/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() diff --git a/talkingq-url/banban/service/binding.py b/talkingq-url/banban/service/binding.py index 338de38..eb9fd37 100644 --- a/talkingq-url/banban/service/binding.py +++ b/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() diff --git a/talkingq-url/banban/service/device_voice_archive.py b/talkingq-url/banban/service/device_voice_archive.py index 35fbe59..26407a8 100644 --- a/talkingq-url/banban/service/device_voice_archive.py +++ b/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): diff --git a/talkingq-url/banban/service/im.py b/talkingq-url/banban/service/im.py index d622aed..696e90f 100644 --- a/talkingq-url/banban/service/im.py +++ b/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 diff --git a/talkingq-url/database/models.py b/talkingq-url/database/models.py index c9eb83b..e29d3c3 100644 --- a/talkingq-url/database/models.py +++ b/talkingq-url/database/models.py @@ -365,6 +365,34 @@ class IMMessage(Base): deleted_at: Mapped[Optional[datetime]] = mapped_column(DateTime) +class DevicePendingVoiceMessage(Base): + __tablename__ = "device_pending_voice_messages" + __table_args__ = ( + UniqueConstraint("target_device_id", "im_message_id", name="uq_device_pending_voice_target_msg"), + Index("idx_device_pending_voice_target_status_created", "target_device_id", "status", "created_at"), + Index("idx_device_pending_voice_sender", "sender_device_id"), + Index("idx_device_pending_voice_im_message", "im_message_id"), + ) + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + target_device_id: Mapped[str] = mapped_column(String(64), nullable=False) + sender_device_id: Mapped[Optional[str]] = mapped_column(String(64)) + im_message_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + media_file_key: Mapped[str] = mapped_column(String(255), nullable=False) + audio_url: Mapped[Optional[str]] = mapped_column(String(512)) + source: Mapped[str] = mapped_column(String(32), nullable=False, server_default=text("'unknown'")) + status: Mapped[str] = mapped_column(String(32), nullable=False, server_default=text("'pending'")) + delivery_count: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0")) + delivered_at: Mapped[Optional[datetime]] = mapped_column(DateTime) + listened_at: Mapped[Optional[datetime]] = mapped_column(DateTime) + created_at: Mapped[datetime] = mapped_column(DateTime, server_default=text("CURRENT_TIMESTAMP")) + updated_at: Mapped[datetime] = mapped_column( + DateTime, + server_default=text("CURRENT_TIMESTAMP"), + onupdate=datetime.utcnow, + ) + + class ChildLocationCurrent(Base): __tablename__ = "child_location_current" diff --git a/talkingq-url/handlers/mqtt_handler.py b/talkingq-url/handlers/mqtt_handler.py index 1cf365c..d6c50d4 100644 --- a/talkingq-url/handlers/mqtt_handler.py +++ b/talkingq-url/handlers/mqtt_handler.py @@ -6,11 +6,12 @@ from typing import Awaitable, Callable, Dict, Optional import aiomqtt -from banban.service.binding import BindingService +from banban.service.binding import binding_service from banban.service.device_alarm import device_alarm_service from banban.service.device_setting import device_setting_service from banban.service.im import im_service from banban.service.location import location_service +from banban.service.pending_voice_message import pending_voice_message_service from config import settings from services.card_service import card_service from services.device_target_cache import device_target_cache @@ -19,6 +20,7 @@ from services.device_identity_initializer import ( device_identity_initializer, ) from services.device_update_manager import device_firmware_update_manager +from services.device_volume_manager import device_volume_manager from services.offline_audio_cache import offline_audio_cache from services.task_manager import task_manager from utils.logger import session_logger as logger @@ -119,6 +121,26 @@ class TalkingQMQTTService: task_type="persistence", ) + # async def _sync_desired_volume(self, device_id: str, current_level) -> None: + # if current_level is None: + # return + # try: + # current_level_int = int(current_level) + # except (TypeError, ValueError): + # logger.warning(device_id, "volume", f"invalid current volume level: {current_level}") + # return + + # desired_level = await device_volume_manager.get_configured_volume(device_id) + # if desired_level is None or desired_level == current_level_int: + # return + + # logger.info( + # device_id, + # "volume", + # f"当前音量 {current_level_int} 与期望音量 {desired_level} 不一致，补发音量指令", + # ) + # await self.send_volume_command(device_id, desired_level) + async def _handle_device_info(self, device_id: str, payload: dict): data = payload.get("data", {}) d_id = data.get("id") @@ -139,6 +161,7 @@ class TalkingQMQTTService: volume=data.get("voice"), ), ) + # await self._sync_desired_volume(device_id, current_volume) await self._publish(f"device/{device_id}/event_resp", {"msg_id": "000", "status": "success"}) async def _handle_gps_response(self, device_id: str, payload: dict): @@ -198,6 +221,7 @@ class TalkingQMQTTService: version_str=None, ), ) + # await self._sync_desired_volume(device_id, current_level) async def _handle_ota_response(self, device_id: str, payload: dict): status = payload.get("status") @@ -293,8 +317,7 @@ class TalkingQMQTTService: nfc_uuid = data.get("uuid") # 卡片与设备绑定 # await card_service.activate_card(device_id=device_id, card_uuid=nfc_uuid) - service = BindingService() - result = await service.finalize_nfc_bind(device_id=device_id, card_uuid=nfc_uuid) + result = await binding_service.finalize_nfc_bind(device_id=device_id, card_uuid=nfc_uuid) if result is None: logger.warning(device_id, "", "[NFC bind] no pending bind session found") return @@ -449,6 +472,23 @@ class TalkingQMQTTService: is_owner = await card_service.check_card_ownership(nfc_uuid, device_id) if is_owner: + playback_items = await pending_voice_message_service.get_playback_items( + target_device_id=device_id, + ) + if playback_items: + params = { + f"url_{k}": item.audio_url + for k, item in enumerate(playback_items, start=1) + } + payload = {"msg_id": "005", "type": 0, "params": params} + published = await self._publish(topic, payload) + if published: + await pending_voice_message_service.mark_delivered( + [item.pending_id for item in playback_items] + ) + await offline_audio_cache.clear_audio_urls(device_id) + return + audio_urls = await offline_audio_cache.pop_audio_urls(device_id) if audio_urls: params = {f"url_{k}": audio_url for k, audio_url in enumerate(audio_urls, start=1)} @@ -530,13 +570,15 @@ class TalkingQMQTTService: if not self._connected: await self.connect() - async def _publish(self, topic: str, payload: dict): + async def _publish(self, topic: str, payload: dict) -> bool: await self._ensure_connected() try: await self._client.publish(topic, json.dumps(payload, ensure_ascii=False), qos=self.qos) logger.info("", "", f"mqtt publish topic={topic} payload={payload}") + return True except Exception as exc: logger.error("", "", f"mqtt publish failed: {exc}") + return False async def send_gps_query(self, device_id: str) -> str: await self._publish(f"device/{device_id}/command", {"msg_id": "001"}) diff --git a/talkingq-url/handlers/websocket_message_handler.py b/talkingq-url/handlers/websocket_message_handler.py index a5ee544..bea8f05 100644 --- a/talkingq-url/handlers/websocket_message_handler.py +++ b/talkingq-url/handlers/websocket_message_handler.py @@ -7,6 +7,7 @@ from handlers.audio_session_handler import handle_websocket_data from handlers.audio_file_handler import upload_message_audio from banban.service.device_audio_cache import device_audio_cache_service from banban.service.device_voice_archive import device_voice_archive_service +from banban.service.pending_voice_message import pending_voice_message_service from services.audio_session import audio_session_manager from services.interrupt_handler import interrupt_handler from services.task_manager import task_manager @@ -340,22 +341,32 @@ async def process_cached_audio(device_id: str, target_device_id: str, serial_num cached_audio, device_id=target_device_id, ) - await offline_audio_cache.add_audio_url(target_device_id, audio_url) - session_logger.info( - device_id, - "target", - f"设备留言已加入待收听队列: target_device_id={target_device_id}, audio_url={audio_url}", + archive_result = await device_voice_archive_service.archive_peer_voice_message( + sender_device_id=device_id, + receiver_device_id=target_device_id, + local_audio_path=str(local_audio_path), ) - - await task_manager.create_task( - device_voice_archive_service.archive_peer_voice_message( + if archive_result is not None: + await pending_voice_message_service.add_pending_message( + target_device_id=target_device_id, sender_device_id=device_id, - receiver_device_id=target_device_id, - local_audio_path=str(local_audio_path), - ), - device_id=device_id, - task_type="device_voice_archive", - ) + im_message_id=archive_result.message_id, + media_file_key=archive_result.media_file_key, + audio_url=audio_url, + source="device_peer_voice", + ) + session_logger.info( + device_id, + "target", + f"设备留言已加入待收听队列: target_device_id={target_device_id}, audio_url={audio_url}", + ) + else: + await offline_audio_cache.add_audio_url(target_device_id, audio_url) + session_logger.warning( + device_id, + "target", + "设备留言已写入内存待收听队列，但归档失败，重启后无法恢复这条留言", + ) websocket = await connection_manager.get_connection(device_id) if websocket and websocket.client_state.name == "CONNECTED": diff --git a/talkingq-url/main.py b/talkingq-url/main.py index 734b6af..37fe7d6 100644 --- a/talkingq-url/main.py +++ b/talkingq-url/main.py @@ -35,6 +35,13 @@ async def lifespan(app: FastAPI): ) init_directories() + try: + session_logger.system_info("startup", "初始化数据库...") + await init_db() + session_logger.system_info("startup", "数据库初始化和迁移完成") + except Exception as e: + session_logger.system_info("startup", f"数据库初始化失败: {str(e)}") + scheduler = TaskScheduler.get_instance() scheduler.start() scheduler.add_interval_task( @@ -57,12 +64,6 @@ async def lifespan(app: FastAPI): else: session_logger.system_info("startup", "未配置 TalkingQ_MQTT，MQTT 服务链接未启动") exit(1) - try: - session_logger.system_info("startup", "初始化数据库...") - await init_db() - session_logger.system_info("startup", "数据库初始化和迁移完成") - except Exception as e: - session_logger.system_info("startup", f"数据库初始化失败: {str(e)}") await role_manager.initialize() await register_services() diff --git a/talkingq-url/mysql/init/02-init.sql b/talkingq-url/mysql/init/02-init.sql new file mode 100644 index 0000000..c901dce --- /dev/null +++ b/talkingq-url/mysql/init/02-init.sql @@ -0,0 +1,534 @@ +-- Banban shared schema +-- Sources: +-- 1. talkingq-url/mysql/init/01-init.sql +-- 2. mini-program/database/init.sql +-- Target database: talkingq +-- +-- Notes: +-- 1. talkingq-url base tables are kept as the device-domain foundation. +-- 2. mini-program business tables are merged into the same database. +-- 3. device_auth.device_id is the shared device identity key. +-- 4. Device <-> AI conversation records remain in +-- conversation_histories / conversation_messages. +-- 5. The database owner is shared, but table ownership is explicit: +-- talkingq-url owns device-domain base tables; mini-program owns +-- parent/child/binding/im/location extension tables. +-- 6. During migration to the shared database, talkingq-url data in +-- `talkingq` is the source of truth. Legacy `mini_program` data is +-- not merged into `talkingq`. + +CREATE DATABASE IF NOT EXISTS `talkingq` + DEFAULT CHARACTER SET utf8mb4 + COLLATE utf8mb4_unicode_ci; + +USE `talkingq`; + +-- --------------------------------------------------------------------------- +-- talkingq-url base tables +-- --------------------------------------------------------------------------- + +CREATE TABLE IF NOT EXISTS `device_configs` ( + `id` INT NOT NULL AUTO_INCREMENT, + `device_id` VARCHAR(64) NOT NULL, + `selected_role_key` VARCHAR(64) NOT NULL, + `preferred_language` VARCHAR(10) NULL, + `volume` INT NULL, + `last_update_time` FLOAT NOT NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE INDEX `device_id_UNIQUE` (`device_id` ASC) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `conversation_histories` ( + `id` INT NOT NULL AUTO_INCREMENT, + `device_id` VARCHAR(64) NOT NULL, + `role_key` VARCHAR(64) NOT NULL, + `last_interaction_time` FLOAT NOT NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + INDEX `idx_device_id` (`device_id` ASC), + INDEX `idx_role_key` (`role_key` ASC) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `conversation_messages` ( + `id` INT NOT NULL AUTO_INCREMENT, + `conversation_id` INT NOT NULL, + `is_user` TINYINT(1) NOT NULL DEFAULT 0, + `content` TEXT NOT NULL, + `timestamp` FLOAT NOT NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + INDEX `idx_conversation_id` (`conversation_id` ASC), + CONSTRAINT `fk_messages_conversation` + FOREIGN KEY (`conversation_id`) + REFERENCES `conversation_histories` (`id`) + ON DELETE CASCADE +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `roles` ( + `id` INT NOT NULL AUTO_INCREMENT, + `role_key` VARCHAR(64) NOT NULL, + `name` VARCHAR(128) NOT NULL, + `description` TEXT NULL, + `content` TEXT NOT NULL, + `default_language` VARCHAR(10) NULL, + `asr_provider` VARCHAR(64) NULL, + `llm_provider` VARCHAR(64) NULL, + `tts_provider` VARCHAR(64) NULL, + `competitive_llm_mode` TINYINT(1) NULL, + `volcano_model_id` VARCHAR(64) NULL, + `volcano_voice_type` VARCHAR(64) NULL, + `tencent_voice_type` VARCHAR(64) NULL, + `aliyun_voice_name` VARCHAR(64) NULL, + `minimax_voice_id` VARCHAR(64) NULL, + `url` VARCHAR(255) NULL, + `homophones` JSON NULL, + `enabled` TINYINT(1) NOT NULL DEFAULT 1, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE INDEX `role_key_UNIQUE` (`role_key` ASC) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `role_languages` ( + `id` INT NOT NULL AUTO_INCREMENT, + `role_id` INT NOT NULL, + `language_code` VARCHAR(10) NOT NULL, + `name` VARCHAR(128) NULL, + `content` TEXT NULL, + `asr_provider` VARCHAR(64) NULL, + `llm_provider` VARCHAR(64) NULL, + `tts_provider` VARCHAR(64) NULL, + `volcano_voice_type` VARCHAR(64) NULL, + `tencent_voice_type` VARCHAR(64) NULL, + `aliyun_voice_name` VARCHAR(64) NULL, + `minimax_voice_id` VARCHAR(64) NULL, + `url` VARCHAR(255) NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE INDEX `uix_role_language` (`role_id`, `language_code`), + CONSTRAINT `fk_role_languages_role` + FOREIGN KEY (`role_id`) + REFERENCES `roles` (`id`) + ON DELETE CASCADE +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `device_auth` ( + `id` INT NOT NULL AUTO_INCREMENT, + `device_id` VARCHAR(64) NOT NULL, + `serial_number` VARCHAR(64) NOT NULL, + `batch_id` VARCHAR(20) NULL, + `is_active` TINYINT(1) NOT NULL DEFAULT 1, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE INDEX `device_id_UNIQUE` (`device_id` ASC), + INDEX `idx_batch_id` (`batch_id` ASC) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `device_imei_mapping` ( + `id` INT NOT NULL AUTO_INCREMENT, + `imei` VARCHAR(64) NOT NULL, + `device_id` VARCHAR(64) NOT NULL, + `serial_number` VARCHAR(64) NOT NULL, + `status` VARCHAR(32) NOT NULL DEFAULT 'pending', + `activated_at` DATETIME NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE INDEX `imei_UNIQUE` (`imei` ASC), + UNIQUE INDEX `device_id_UNIQUE` (`device_id` ASC), + INDEX `idx_device_imei_mapping_status` (`status` ASC) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `device_firmware_update` ( + `id` INT NOT NULL AUTO_INCREMENT, + `device_id` VARCHAR(64) NOT NULL, + `serial_number` VARCHAR(64) NOT NULL, + `mac_address` VARCHAR(512) NULL, + `firmware_version` VARCHAR(64) NOT NULL, + `update_status` VARCHAR(32) NOT NULL DEFAULT 'success', + `progress` FLOAT NULL DEFAULT 0.0, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE INDEX `device_id_UNIQUE` (`device_id` ASC) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `system_config` ( + `id` INT NOT NULL AUTO_INCREMENT, + `config_key` VARCHAR(128) NOT NULL, + `config_value` TEXT NULL, + `description` VARCHAR(255) NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE INDEX `config_key_UNIQUE` (`config_key` ASC) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +INSERT INTO `device_configs` (`device_id`, `selected_role_key`, `preferred_language`, `last_update_time`) +VALUES ('default', 'assistant', 'zh', UNIX_TIMESTAMP()) +ON DUPLICATE KEY UPDATE `selected_role_key` = `selected_role_key`; + +INSERT INTO `system_config` (`config_key`, `config_value`, `description`) +VALUES + ('latest_firmware_version', '1.0.0', '最新固件版本'), + ('update_firmware_url', 'https://example.com/firmware/latest.bin', '固件更新URL') +ON DUPLICATE KEY UPDATE `config_value` = `config_value`; + +INSERT INTO `roles` (`role_key`, `name`, `description`, `content`, `default_language`, `enabled`) +VALUES ( + 'assistant', + '智能助手', + '默认智能助手角色', + '你是一个友好的智能助手，乐于帮助用户解答问题。', + 'zh', + 1 +) +ON DUPLICATE KEY UPDATE `role_key` = `role_key`; + +-- --------------------------------------------------------------------------- +-- mini-program extension tables +-- --------------------------------------------------------------------------- + +CREATE TABLE IF NOT EXISTS `parents` ( + `user_id` BIGINT NOT NULL AUTO_INCREMENT, + `openid` VARCHAR(64) NOT NULL, + `unionid` VARCHAR(64) NULL, + `nickname` VARCHAR(64) NULL, + `avatar_url` VARCHAR(255) NULL, + `avatar_file_key` VARCHAR(255) NULL, + `phone` VARCHAR(20) NULL, + `status` TINYINT NOT NULL DEFAULT 1, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`user_id`), + UNIQUE KEY `uq_parents_openid` (`openid`), + KEY `idx_parents_unionid` (`unionid`), + KEY `idx_parents_phone` (`phone`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `children` ( + `child_id` BIGINT NOT NULL AUTO_INCREMENT, + `child_name` VARCHAR(32) NOT NULL, + `child_gender` TINYINT NOT NULL DEFAULT 2, + `child_birthday` DATE NULL, + `status` TINYINT NOT NULL DEFAULT 1, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`child_id`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `parent_child_relations` ( + `id` BIGINT NOT NULL AUTO_INCREMENT, + `user_id` BIGINT NOT NULL, + `child_id` BIGINT NOT NULL, + `relation_type` TINYINT NOT NULL DEFAULT 9, + `is_primary` TINYINT(1) NOT NULL DEFAULT 0, + `status` TINYINT NOT NULL DEFAULT 1, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + CONSTRAINT `uq_user_child` UNIQUE (`user_id`, `child_id`), + KEY `idx_pcr_user_id` (`user_id`), + KEY `idx_pcr_child_id` (`child_id`), + CONSTRAINT `fk_pcr_user` + FOREIGN KEY (`user_id`) + REFERENCES `parents` (`user_id`) + ON DELETE CASCADE, + CONSTRAINT `fk_pcr_child` + FOREIGN KEY (`child_id`) + REFERENCES `children` (`child_id`) + ON DELETE CASCADE +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `device_bindings` ( + `id` BIGINT NOT NULL AUTO_INCREMENT, + `device_id` VARCHAR(64) NOT NULL, + `owner_user_id` BIGINT NOT NULL, + `child_id` BIGINT NULL, + `status` TINYINT NOT NULL DEFAULT 1, + `bound_at` DATETIME NOT NULL, + `unbound_at` DATETIME NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + CONSTRAINT `uq_device_binding_device` UNIQUE (`device_id`), + CONSTRAINT `uq_device_binding_child` UNIQUE (`child_id`), + KEY `idx_device_bindings_owner_user_id` (`owner_user_id`), + CONSTRAINT `fk_device_bindings_device` + FOREIGN KEY (`device_id`) + REFERENCES `device_auth` (`device_id`), + CONSTRAINT `fk_device_bindings_owner` + FOREIGN KEY (`owner_user_id`) + REFERENCES `parents` (`user_id`), + CONSTRAINT `fk_device_bindings_child` + FOREIGN KEY (`child_id`) + REFERENCES `children` (`child_id`) + ON DELETE SET NULL +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `device_bind_sessions` ( + `id` BIGINT NOT NULL AUTO_INCREMENT, + `bind_token` CHAR(36) NOT NULL, + `device_id` VARCHAR(64) NOT NULL, + `initiator_user_id` BIGINT NOT NULL, + `target_child_id` BIGINT NULL, + `challenge_code_hash` CHAR(64) NULL, + `challenge_set_at` DATETIME NULL, + `expires_at` DATETIME NOT NULL, + `max_attempt_count` TINYINT UNSIGNED NOT NULL DEFAULT 5, + `attempt_count` TINYINT UNSIGNED NOT NULL DEFAULT 0, + `status` TINYINT NOT NULL DEFAULT 1, + `confirmed_at` DATETIME NULL, + `consumed_at` DATETIME NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE KEY `uq_device_bind_sessions_token` (`bind_token`), + KEY `idx_device_bind_sessions_device_id` (`device_id`), + KEY `idx_device_bind_sessions_initiator_user_id` (`initiator_user_id`), + KEY `idx_device_bind_sessions_target_child_id` (`target_child_id`), + KEY `idx_device_bind_sessions_expires_at` (`expires_at`), + CONSTRAINT `fk_device_bind_sessions_device` + FOREIGN KEY (`device_id`) + REFERENCES `device_auth` (`device_id`), + CONSTRAINT `fk_device_bind_sessions_initiator` + FOREIGN KEY (`initiator_user_id`) + REFERENCES `parents` (`user_id`), + CONSTRAINT `fk_device_bind_sessions_target_child` + FOREIGN KEY (`target_child_id`) + REFERENCES `children` (`child_id`) + ON DELETE SET NULL +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `device_bind_history` ( + `id` BIGINT NOT NULL AUTO_INCREMENT, + `device_id` VARCHAR(64) NOT NULL, + `child_id` BIGINT NULL, + `bound_by_user_id` BIGINT NOT NULL, + `unbound_by_user_id` BIGINT NULL, + `bind_source` TINYINT NOT NULL DEFAULT 1, + `bound_at` DATETIME NOT NULL, + `unbound_at` DATETIME NULL, + `unbind_reason` VARCHAR(191) NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + KEY `idx_device_bind_history_device_id` (`device_id`), + KEY `idx_device_bind_history_child_id` (`child_id`), + KEY `idx_device_bind_history_bound_by_user_id` (`bound_by_user_id`), + KEY `idx_device_bind_history_unbound_by_user_id` (`unbound_by_user_id`), + KEY `idx_device_bind_history_bound_at` (`bound_at`), + CONSTRAINT `fk_device_bind_history_device` + FOREIGN KEY (`device_id`) + REFERENCES `device_auth` (`device_id`), + CONSTRAINT `fk_device_bind_history_child` + FOREIGN KEY (`child_id`) + REFERENCES `children` (`child_id`) + ON DELETE SET NULL, + CONSTRAINT `fk_device_bind_history_bound_by` + FOREIGN KEY (`bound_by_user_id`) + REFERENCES `parents` (`user_id`), + CONSTRAINT `fk_device_bind_history_unbound_by` + FOREIGN KEY (`unbound_by_user_id`) + REFERENCES `parents` (`user_id`) + ON DELETE SET NULL +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `cards` ( + `card_id` BIGINT NOT NULL AUTO_INCREMENT, + `card_uuid` VARCHAR(64) NOT NULL, + `device_id` VARCHAR(64) NULL, + `card_name` VARCHAR(64) NULL, + `status` TINYINT NOT NULL DEFAULT 0, + `total_swaps` INT NOT NULL DEFAULT 0, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`card_id`), + UNIQUE KEY `uq_cards_card_uuid` (`card_uuid`), + UNIQUE KEY `uq_cards_device_id` (`device_id`), + KEY `idx_cards_status` (`status`), + CONSTRAINT `fk_cards_device` + FOREIGN KEY (`device_id`) + REFERENCES `device_auth` (`device_id`) + ON DELETE SET NULL +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +-- Shared device volume is recommended to use device_configs.volume as source of truth. +-- device_settings.volume is retained for mini-program compatibility and UI storage. +CREATE TABLE IF NOT EXISTS `device_settings` ( + `setting_id` BIGINT NOT NULL AUTO_INCREMENT, + `device_id` VARCHAR(64) NOT NULL, + `sleep_mode` TINYINT NOT NULL DEFAULT 0, + `disable_time_start` TIME NULL, + `disable_time_end` TIME NULL, + `timezone` VARCHAR(32) NOT NULL DEFAULT 'Asia/Shanghai', + `volume` TINYINT UNSIGNED NULL, + `brightness` TINYINT UNSIGNED NULL, + `power` TINYINT UNSIGNED NULL, + `signal` TINYINT UNSIGNED NULL, + `version` VARCHAR(64) NULL, + `disable_weekdays` VARCHAR(32) NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`setting_id`), + UNIQUE KEY `uq_device_settings_device_id` (`device_id`), + CONSTRAINT `fk_device_settings_device` + FOREIGN KEY (`device_id`) + REFERENCES `device_auth` (`device_id`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +-- Device alarm events are reported by the device side through MQTT msg_id=010. +CREATE TABLE IF NOT EXISTS `device_alarm_events` ( + `alarm_id` BIGINT NOT NULL AUTO_INCREMENT, + `device_id` VARCHAR(64) NOT NULL, + `owner_user_id` BIGINT NULL, + `child_id` BIGINT NULL, + `source_msg_id` VARCHAR(8) NOT NULL DEFAULT '010', + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (`alarm_id`), + KEY `idx_device_alarm_device_created` (`device_id`, `created_at`), + KEY `idx_device_alarm_owner_created` (`owner_user_id`, `created_at`), + KEY `idx_device_alarm_child_created` (`child_id`, `created_at`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `im_conversations` ( + `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT, + `conversation_type` TINYINT UNSIGNED NOT NULL, + `participant_a_type` TINYINT UNSIGNED NOT NULL, + `participant_a_id` VARCHAR(64) NOT NULL, + `participant_b_type` TINYINT UNSIGNED NOT NULL, + `participant_b_id` VARCHAR(64) NOT NULL, + `pair_key` VARCHAR(191) NOT NULL, + `status` TINYINT NOT NULL DEFAULT 1, + `last_seq` BIGINT UNSIGNED NOT NULL DEFAULT 0, + `message_count` BIGINT UNSIGNED NOT NULL DEFAULT 0, + `last_message_preview` VARCHAR(255) NULL, + `last_message_at` DATETIME NULL, + `ext_json` JSON NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + CONSTRAINT `uq_conv_type_pair` UNIQUE (`conversation_type`, `pair_key`), + KEY `idx_im_conv_participant_a` (`participant_a_type`, `participant_a_id`), + KEY `idx_im_conv_participant_b` (`participant_b_type`, `participant_b_id`), + KEY `idx_im_conv_last_message_at` (`last_message_at`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `im_messages` ( + `id` BIGINT NOT NULL AUTO_INCREMENT, + `conversation_id` BIGINT UNSIGNED NOT NULL, + `seq` BIGINT UNSIGNED NOT NULL, + `sender_type` TINYINT UNSIGNED NOT NULL, + `sender_id` VARCHAR(64) NOT NULL, + `receiver_type` TINYINT UNSIGNED NOT NULL, + `receiver_id` VARCHAR(64) NOT NULL, + `content_type` TINYINT UNSIGNED NOT NULL, + `content_text` MEDIUMTEXT NULL, + `content_json` JSON NULL, + `media_file_key` VARCHAR(255) NULL, + `media_duration_ms` INT UNSIGNED NULL, + `media_mime_type` VARCHAR(64) NULL, + `media_size_bytes` BIGINT UNSIGNED NULL, + `media_transcript_text` TEXT NULL, + `client_msg_id` VARCHAR(64) NULL, + `sender_name_snapshot` VARCHAR(64) NULL, + `sender_avatar_snapshot` VARCHAR(255) NULL, + `receiver_name_snapshot` VARCHAR(64) NULL, + `receiver_avatar_snapshot` VARCHAR(255) NULL, + `ext_json` JSON NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `deleted_at` DATETIME NULL, + PRIMARY KEY (`id`), + CONSTRAINT `uq_im_msg_conv_seq` UNIQUE (`conversation_id`, `seq`), + CONSTRAINT `uq_im_msg_client` UNIQUE (`conversation_id`, `client_msg_id`), + KEY `idx_im_msg_conversation_created_at` (`conversation_id`, `created_at`), + KEY `idx_im_msg_sender` (`sender_type`, `sender_id`, `created_at`), + KEY `idx_im_msg_receiver` (`receiver_type`, `receiver_id`, `created_at`), + CONSTRAINT `fk_im_messages_conversation` + FOREIGN KEY (`conversation_id`) + REFERENCES `im_conversations` (`id`) + ON DELETE CASCADE +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +-- Persistent unread voice-message index for device NFC owner-card playback. +-- Audio content is still stored by im_messages/media_file_key; this table only stores delivery state. +CREATE TABLE IF NOT EXISTS `device_pending_voice_messages` ( + `id` BIGINT NOT NULL AUTO_INCREMENT, + `target_device_id` VARCHAR(64) NOT NULL, + `sender_device_id` VARCHAR(64) NULL, + `im_message_id` BIGINT NOT NULL, + `media_file_key` VARCHAR(255) NOT NULL, + `audio_url` VARCHAR(512) NULL, + `source` VARCHAR(32) NOT NULL DEFAULT 'unknown', + `status` VARCHAR(32) NOT NULL DEFAULT 'pending', + `delivery_count` INT NOT NULL DEFAULT 0, + `delivered_at` DATETIME NULL, + `listened_at` DATETIME NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE KEY `uq_device_pending_voice_target_msg` (`target_device_id`, `im_message_id`), + KEY `idx_device_pending_voice_target_status_created` (`target_device_id`, `status`, `created_at`), + KEY `idx_device_pending_voice_sender` (`sender_device_id`), + KEY `idx_device_pending_voice_im_message` (`im_message_id`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `child_location_current` ( + `child_id` BIGINT NOT NULL, + `device_id` VARCHAR(64) NOT NULL, + `coord_type` VARCHAR(16) NOT NULL DEFAULT 'gcj02', + `lat` DECIMAL(10, 7) NOT NULL, + `lng` DECIMAL(10, 7) NOT NULL, + `accuracy_m` INT UNSIGNED NULL, + `altitude_m` DECIMAL(8, 2) NULL, + `speed_mps` DECIMAL(8, 2) NULL, + `heading_deg` SMALLINT UNSIGNED NULL, + `source` TINYINT UNSIGNED NOT NULL, + `battery_pct` TINYINT UNSIGNED NULL, + `device_time` DATETIME NOT NULL, + `server_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`child_id`), + KEY `idx_child_loc_current_device` (`device_id`), + KEY `idx_child_loc_current_server_time` (`server_time`), + CONSTRAINT `fk_child_loc_current_child` + FOREIGN KEY (`child_id`) + REFERENCES `children` (`child_id`) + ON DELETE CASCADE, + CONSTRAINT `fk_child_loc_current_device` + FOREIGN KEY (`device_id`) + REFERENCES `device_auth` (`device_id`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `child_location_history` ( + `id` BIGINT NOT NULL AUTO_INCREMENT, + `child_id` BIGINT NOT NULL, + `device_id` VARCHAR(64) NOT NULL, + `coord_type` VARCHAR(16) NOT NULL DEFAULT 'gcj02', + `lat` DECIMAL(10, 7) NOT NULL, + `lng` DECIMAL(10, 7) NOT NULL, + `accuracy_m` INT UNSIGNED NULL, + `altitude_m` DECIMAL(8, 2) NULL, + `speed_mps` DECIMAL(8, 2) NULL, + `heading_deg` SMALLINT UNSIGNED NULL, + `source` TINYINT UNSIGNED NOT NULL, + `battery_pct` TINYINT UNSIGNED NULL, + `device_time` DATETIME NOT NULL, + `server_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + KEY `idx_child_loc_hist_child` (`child_id`, `created_at`), + KEY `idx_child_loc_hist_device` (`device_id`, `created_at`), + CONSTRAINT `fk_child_loc_hist_child` + FOREIGN KEY (`child_id`) + REFERENCES `children` (`child_id`) + ON DELETE CASCADE, + CONSTRAINT `fk_child_loc_hist_device` + FOREIGN KEY (`device_id`) + REFERENCES `device_auth` (`device_id`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; diff --git a/talkingq-url/services/database_service_base.py b/talkingq-url/services/database_service_base.py index 23c329a..61a2c1e 100644 --- a/talkingq-url/services/database_service_base.py +++ b/talkingq-url/services/database_service_base.py @@ -3,6 +3,8 @@ from utils.logger import session_logger from database.connection import get_db_manager class DatabaseServiceBase: + _logged_initialized_services = set() + """ 数据库服务基类，提供通用的数据库初始化和会话管理功能 所有需要访问数据库的服务类都应该继承此基类 @@ -19,7 +21,9 @@ class DatabaseServiceBase: try: self.db_manager = await get_db_manager() self._db_initialized = True - session_logger.info("system", self.service_name, f"{self.service_name}数据库连接初始化成功") + if self.service_name not in self._logged_initialized_services: + self._logged_initialized_services.add(self.service_name) + session_logger.info("system", self.service_name, f"{self.service_name}数据库连接初始化成功") except Exception as e: session_logger.error("system", self.service_name, f"{self.service_name}数据库连接初始化失败: {str(e)}") raise diff --git a/talkingq-url/services/scheduler.py b/talkingq-url/services/scheduler.py index 834a268..7ec7762 100644 --- a/talkingq-url/services/scheduler.py +++ b/talkingq-url/services/scheduler.py @@ -3,6 +3,7 @@ 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 @@ -40,11 +41,14 @@ 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() - if not audio_cache: + fallback_devices = list(audio_cache.keys()) if audio_cache else [] + device_ids = list(dict.fromkeys([*pending_devices, *fallback_devices])) + if not device_ids: logger.warning("", "", "无离线音频缓存，跳过本次执行") return - for device_id, audio_urls in audio_cache.items(): + 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"