diff --git a/talkingq-url/.gitignore b/talkingq-url/.gitignore index d1abc05..09a9cd9 100644 --- a/talkingq-url/.gitignore +++ b/talkingq-url/.gitignore @@ -31,6 +31,7 @@ fullcode.md all.md *.sql.gz database_backups +runtime/ # Docker *.log @@ -49,4 +50,4 @@ database_backups .Spotlight-V100 .Trashes ehthumbs.db -Thumbs.db \ No newline at end of file +Thumbs.db diff --git a/talkingq-url/api/assets.py b/talkingq-url/api/assets.py index 8259248..7ee7904 100644 --- a/talkingq-url/api/assets.py +++ b/talkingq-url/api/assets.py @@ -12,5 +12,13 @@ def configure_static_assets(app): firmware_directory = os.path.join(settings.assets_dir, "firmware") os.makedirs(firmware_directory, exist_ok=True) + + device_audio_directory = settings.device_audio_cache_dir + os.makedirs(device_audio_directory, exist_ok=True) - app.mount("/assets", StaticFiles(directory=settings.assets_dir), name="assets") \ No newline at end of file + app.mount("/assets", StaticFiles(directory=settings.assets_dir), name="assets") + app.mount( + "/device-audio", + StaticFiles(directory=device_audio_directory), + name="device-audio", + ) diff --git a/talkingq-url/banban/routers/device_im.py b/talkingq-url/banban/routers/device_im.py index d923149..f7071a5 100644 --- a/talkingq-url/banban/routers/device_im.py +++ b/talkingq-url/banban/routers/device_im.py @@ -22,7 +22,7 @@ try: ConversationMessageCreateResponse, DeviceMessageCreateRequest, ) - from banban.service.im import im_service + from banban.service.im import im_service, present_device_message_item except ModuleNotFoundError: from banban.service import get_db_session from banban.routers.im import ( @@ -41,7 +41,7 @@ except ModuleNotFoundError: ConversationMessageCreateResponse, DeviceMessageCreateRequest, ) - from banban.service.im import im_service + from banban.service.im import im_service, present_device_message_item router = APIRouter(prefix="/device-im", tags=["device-im"]) @@ -247,7 +247,10 @@ async def list_device_conversation_messages( rows = rows[:limit] rows = list(rows) rows.reverse() - items = [_row_to_message_item(row) for row in rows] + items = [ + await present_device_message_item(row, device_id=device_id) + for row in rows + ] next_cursor_seq = items[0].seq if has_more and items else None logger.info( @@ -312,4 +315,4 @@ async def create_message_from_device( conversation_type=result.conversation_type, conversation_type_name=result.conversation_type_name, message=result.message, - ) \ No newline at end of file + ) diff --git a/talkingq-url/banban/routers/mqtt_router.py b/talkingq-url/banban/routers/mqtt_router.py index 254ffc5..0f39a8a 100644 --- a/talkingq-url/banban/routers/mqtt_router.py +++ b/talkingq-url/banban/routers/mqtt_router.py @@ -1,6 +1,7 @@ import logging from typing import Optional from fastapi import APIRouter, HTTPException, Query +from banban.service.device_audio_cache import device_audio_cache_service from services.offline_audio_cache import offline_audio_cache from banban.schemas.mqtt_models import ( GPSQueryRequest, @@ -95,7 +96,11 @@ async def set_nfc_unread( req: NFCUnreadRequest, # current_user_id: int = Depends(get_current_user_id) ): - await offline_audio_cache.add_audio_url(req.device_id, req.url) + audio_url = await device_audio_cache_service.get_device_audio_url( + req.url, + device_id=req.device_id, + ) + await offline_audio_cache.add_audio_url(req.device_id, audio_url) return CommandResponse(msg_id="", device_id=req.device_id, message="已设置未读留言") diff --git a/talkingq-url/banban/service/device_audio_cache.py b/talkingq-url/banban/service/device_audio_cache.py new file mode 100644 index 0000000..e08ba78 --- /dev/null +++ b/talkingq-url/banban/service/device_audio_cache.py @@ -0,0 +1,174 @@ +import asyncio +import hashlib +import mimetypes +from pathlib import Path +from urllib.parse import urlparse +from urllib.request import Request, urlopen + +from banban.service.message_audio_storage import ( + MessageAudioStorageError, + MessageAudioStorageService, +) +from config import settings +from utils.logger import session_logger + + +_AUDIO_CONTENT_TYPE_TO_EXT = { + "audio/mpeg": "mp3", + "audio/mp3": "mp3", + "audio/aac": "aac", + "audio/wav": "wav", + "audio/x-wav": "wav", + "audio/x-m4a": "m4a", + "audio/mp4": "m4a", + "audio/webm": "webm", +} + + +class DeviceAudioCacheError(Exception): + pass + + +class DeviceAudioCacheService: + def __init__(self, audio_storage: MessageAudioStorageService | None = None) -> None: + self.audio_storage = audio_storage or MessageAudioStorageService() + + def cache_dir(self) -> Path: + return Path(settings.device_audio_cache_dir) + + def public_base_url(self) -> str: + configured = settings.device_audio_public_base_url.strip().rstrip("/") + if configured: + return configured + return f"http://{settings.server_host}:{settings.server_port}/device-audio" + + async def get_device_audio_url( + self, + file_key_or_url: str, + *, + device_id: str | None = None, + ) -> str: + source = (file_key_or_url or "").strip() + if not source: + raise DeviceAudioCacheError("audio source is required") + if source.startswith("http://"): + return source + + cache_key = hashlib.sha256(source.encode("utf-8")).hexdigest() + cached = self._find_cached_file(cache_key) + if cached is not None: + return self._public_url(cached.name) + + download_url = await self._resolve_download_url(source) + content, content_type = await asyncio.to_thread(self._download_audio, download_url) + extension = self._resolve_extension(source=source, content_type=content_type) + target_path = self.cache_dir() / f"{cache_key}.{extension}" + target_path.parent.mkdir(parents=True, exist_ok=True) + target_path.write_bytes(content) + + session_logger.info( + device_id or "", + "device_audio", + f"cached device audio: source={source} file={target_path}", + ) + return self._public_url(target_path.name) + + async def save_device_audio( + self, + content: bytes, + *, + device_id: str, + extension: str = "mp3", + ) -> tuple[str, Path]: + if not content: + raise DeviceAudioCacheError("audio content is empty") + + normalized_extension = extension.strip().lower().lstrip(".") or "mp3" + if normalized_extension not in {"mp3", "aac", "m4a", "wav", "webm"}: + normalized_extension = "mp3" + + digest = hashlib.sha256( + f"{device_id}:".encode("utf-8") + content + ).hexdigest() + target_path = self.cache_dir() / f"{digest}.{normalized_extension}" + target_path.parent.mkdir(parents=True, exist_ok=True) + await asyncio.to_thread(target_path.write_bytes, content) + + session_logger.info( + device_id, + "device_audio", + f"saved device audio cache: file={target_path}", + ) + return self._public_url(target_path.name), target_path + + async def _resolve_download_url(self, source: str) -> str: + if source.startswith("https://"): + if not self._is_allowed_https_source(source): + raise DeviceAudioCacheError("unsupported https audio source") + return source + try: + return await self.audio_storage.get_audio_url(source) + except MessageAudioStorageError as exc: + raise DeviceAudioCacheError(str(exc)) from exc + + def _is_allowed_https_source(self, source: str) -> bool: + parsed = urlparse(source) + if not parsed.scheme == "https" or not parsed.netloc: + return False + + configured_base = settings.cos_public_base_url.strip() + if configured_base: + configured_host = urlparse(configured_base).netloc + if configured_host and parsed.netloc == configured_host: + return True + + bucket = settings.cos_bucket_message.strip() + region = settings.cos_region.strip() + if bucket and region and parsed.netloc == f"{bucket}.cos.{region}.myqcloud.com": + return True + return False + + def _download_audio(self, url: str) -> tuple[bytes, str | None]: + request = Request(url, headers={"User-Agent": "banban-device-audio-cache/1.0"}) + with urlopen(request, timeout=15) as response: + content_type = response.headers.get("Content-Type") + content_length = response.headers.get("Content-Length") + if content_length and int(content_length) > settings.device_audio_max_bytes: + raise DeviceAudioCacheError("audio file is too large") + content = response.read(settings.device_audio_max_bytes + 1) + if len(content) > settings.device_audio_max_bytes: + raise DeviceAudioCacheError("audio file is too large") + if not content: + raise DeviceAudioCacheError("audio file is empty") + return content, content_type + + def _find_cached_file(self, cache_key: str) -> Path | None: + cache_dir = self.cache_dir() + if not cache_dir.exists(): + return None + matches = list(cache_dir.glob(f"{cache_key}.*")) + if not matches: + return None + return matches[0] + + def _resolve_extension(self, *, source: str, content_type: str | None) -> str: + normalized_content_type = (content_type or "").split(";", 1)[0].strip().lower() + if normalized_content_type in _AUDIO_CONTENT_TYPE_TO_EXT: + return _AUDIO_CONTENT_TYPE_TO_EXT[normalized_content_type] + + suffix = Path(urlparse(source).path).suffix.lower().lstrip(".") + if suffix in {"mp3", "aac", "m4a", "wav", "webm"}: + return suffix + + guessed = mimetypes.guess_extension(normalized_content_type or "") + if guessed: + extension = guessed.lstrip(".") + if extension in {"mp3", "aac", "m4a", "wav", "webm"}: + return extension + return "mp3" + + def _public_url(self, filename: str) -> str: + return f"{self.public_base_url()}/{filename}" + + +device_audio_cache_service = DeviceAudioCacheService() diff --git a/talkingq-url/banban/service/device_voice_archive.py b/talkingq-url/banban/service/device_voice_archive.py index 9b5b17a..35fbe59 100644 --- a/talkingq-url/banban/service/device_voice_archive.py +++ b/talkingq-url/banban/service/device_voice_archive.py @@ -125,8 +125,10 @@ class DeviceVoiceArchiveService(DatabaseServiceBase): ), ) - with open(prepared_audio.filepath, "rb") as archive_file: - archive_audio_data = archive_file.read() + archive_audio_data = await asyncio.to_thread( + self._read_file_bytes, + prepared_audio.filepath, + ) session_logger.info( sender_device_id, @@ -245,7 +247,7 @@ class DeviceVoiceArchiveService(DatabaseServiceBase): finally: if prepared_audio and prepared_audio.archive_format == "mp3" and prepared_audio.filepath != local_audio_path: if os.path.exists(prepared_audio.filepath): - os.remove(prepared_audio.filepath) + await asyncio.to_thread(os.remove, prepared_audio.filepath) await db_session.close() async def _prepare_archive_audio( @@ -255,8 +257,10 @@ class DeviceVoiceArchiveService(DatabaseServiceBase): archive_id: str, local_audio_path: str, ) -> PreparedArchiveAudio | None: - with open(local_audio_path, "rb") as local_file: - local_audio_data = local_file.read() + local_audio_data = await asyncio.to_thread( + self._read_file_bytes, + local_audio_path, + ) source_format = detect_audio_format(local_audio_data) local_size = len(local_audio_data) @@ -288,8 +292,11 @@ class DeviceVoiceArchiveService(DatabaseServiceBase): mp3_path = f"{local_audio_path}.archive.mp3" try: if source_format == "wav": - with open(wav_path, "wb") as wav_file: - wav_file.write(local_audio_data) + await asyncio.to_thread( + self._write_file_bytes, + wav_path, + local_audio_data, + ) session_logger.info( sender_device_id, archive_id, @@ -302,8 +309,11 @@ class DeviceVoiceArchiveService(DatabaseServiceBase): channels=DEFAULT_CHANNELS, sample_width=DEFAULT_SAMPLE_WIDTH, ) - with open(wav_path, "wb") as wav_file: - wav_file.write(wrapped_wav) + await asyncio.to_thread( + self._write_file_bytes, + wav_path, + wrapped_wav, + ) session_logger.info( sender_device_id, archive_id, @@ -320,7 +330,7 @@ class DeviceVoiceArchiveService(DatabaseServiceBase): source_path=wav_path, target_path=mp3_path, ) - mp3_size = os.path.getsize(mp3_path) + mp3_size = await asyncio.to_thread(os.path.getsize, mp3_path) session_logger.info( sender_device_id, archive_id, @@ -339,7 +349,7 @@ class DeviceVoiceArchiveService(DatabaseServiceBase): ) finally: if os.path.exists(wav_path): - os.remove(wav_path) + await asyncio.to_thread(os.remove, wav_path) async def _convert_to_mp3( self, @@ -423,5 +433,13 @@ class DeviceVoiceArchiveService(DatabaseServiceBase): "child_name": row["child_name"], } + def _read_file_bytes(self, filepath: str) -> bytes: + with open(filepath, "rb") as file: + return file.read() + + def _write_file_bytes(self, filepath: str, content: bytes) -> None: + with open(filepath, "wb") as file: + file.write(content) + device_voice_archive_service = DeviceVoiceArchiveService() diff --git a/talkingq-url/banban/service/im.py b/talkingq-url/banban/service/im.py index fa69068..d622aed 100644 --- a/talkingq-url/banban/service/im.py +++ b/talkingq-url/banban/service/im.py @@ -7,6 +7,7 @@ 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 try: @@ -19,7 +20,6 @@ try: except ModuleNotFoundError: from banban.dao.im import ImDAO, DeviceIdentity, ConversationMessageCreateResult from banban.schemas.im import ChildConversationMessageItem, DeviceMessageCreateRequest, ParentChildMessageCreateRequest -from handlers.audio_file_handler import message_audio_storage_service from utils.logger import session_logger @@ -152,6 +152,28 @@ async def present_message_item( return item +async def present_device_message_item( + row: Mapping[str, Any], + *, + device_id: str, +) -> ChildConversationMessageItem: + item = row_to_message_item(row) + if item.content_type == 2 and item.media_file_key: + try: + item.media_file_key = await device_audio_cache_service.get_device_audio_url( + item.media_file_key, + device_id=device_id, + ) + except Exception as exc: + session_logger.error( + device_id, + "device_audio", + f"failed to prepare device audio url: {exc}", + exc_info=True, + ) + return item + + class ImService(DatabaseServiceBase): def __init__(self): super().__init__(service_name="im_service") @@ -292,9 +314,12 @@ class ImService(DatabaseServiceBase): binding_service = BindingService() device = await binding_service.get_current_binding(parent_user_id) try: - audio_url = await message_audio_storage_service.get_audio_url(stored_audio.file_key) + audio_url = await device_audio_cache_service.get_device_audio_url( + stored_audio.file_key, + device_id=device.device_id, + ) except Exception: - session_logger.error(device.device_id, "audio", f"failed to get audio url: {stored_audio.file_key}", exc_info=True) + 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}") except Exception: diff --git a/talkingq-url/config.py b/talkingq-url/config.py index 7b175e9..f72ca2a 100644 --- a/talkingq-url/config.py +++ b/talkingq-url/config.py @@ -99,6 +99,18 @@ class Settings(BaseSettings): default=2 * 1024 * 1024, validation_alias="COS_AVATAR_MAX_BYTES", ) + device_audio_cache_dir: str = Field( + default="runtime/device-audio", + validation_alias="DEVICE_AUDIO_CACHE_DIR", + ) + device_audio_public_base_url: str = Field( + default="", + validation_alias="DEVICE_AUDIO_PUBLIC_BASE_URL", + ) + device_audio_max_bytes: int = Field( + default=20 * 1024 * 1024, + validation_alias="DEVICE_AUDIO_MAX_BYTES", + ) jwt_secret: str = Field( default="dev_only_change_jwt_secret", diff --git a/talkingq-url/handlers/websocket_message_handler.py b/talkingq-url/handlers/websocket_message_handler.py index d53b950..a5ee544 100644 --- a/talkingq-url/handlers/websocket_message_handler.py +++ b/talkingq-url/handlers/websocket_message_handler.py @@ -4,7 +4,9 @@ import asyncio from fastapi import WebSocket from handlers.audio_packet_parser import parse_packet from handlers.audio_session_handler import handle_websocket_data -from handlers.audio_file_handler import message_audio_storage_service, save_audio_file, upload_message_audio +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 services.audio_session import audio_session_manager from services.interrupt_handler import interrupt_handler from services.task_manager import task_manager @@ -329,35 +331,32 @@ async def process_parent_leave_message(device_id: str, audio_cache_key: str): async def process_cached_audio(device_id: str, target_device_id: str, serial_number: str): """处理缓存的音频数据并发送音频URL""" try: - # 获取缓存的音频数据 cached_audio = await target_audio_cache.get_audio_data(target_device_id) if not cached_audio: session_logger.info(device_id, "target", f"目标设备 {target_device_id} 没有缓存的音频数据") return - - # 保存音频文件 - audio_file_key = await save_audio_file(cached_audio, device_id) - audio_url = f"http://{settings.server_host}:{settings.server_port}/{audio_file_key}" - # 将音频URL保存到数据库 im_conversation和im_message - # await im_conversation_service.create_device_message(device_id=device_id, serial_number=serial_number, target_device_id=target_device_id, audio_url=audio_file_key) - # try: - # audio_url = await message_audio_storage_service.get_audio_url(audio_file_key) - # except Exception: - # audio_url = audio_file_key - # 发送URL给目标设备 - # target_websocket = await connection_manager.get_connection(target_device_id) - # if target_websocket and target_websocket.client_state.name == "CONNECTED": - # await target_websocket.send_text("TTS_START") - # session_logger.info(device_id, "target", "已发送 TTS_START 给客户端") - # await target_websocket.send_text(f"NFC_SOUND_URL:{audio_url}") - # session_logger.info(device_id, "target", f"已发送音频URL给目标设备 {target_device_id}: {audio_url}") - # await target_websocket.send_text("TTS_END") - # session_logger.info(device_id, "target", "已发送 TTS_END 给客户端") - # else: - - # # 目标设备不在线，保存到离线缓存 + + audio_url, local_audio_path = await device_audio_cache_service.save_device_audio( + cached_audio, + device_id=target_device_id, + ) await offline_audio_cache.add_audio_url(target_device_id, audio_url) - session_logger.warning(device_id, "target", f"目标设备 {target_device_id} 不在线，保存音频URL到离线缓存") + session_logger.info( + device_id, + "target", + f"设备留言已加入待收听队列: target_device_id={target_device_id}, audio_url={audio_url}", + ) + + await task_manager.create_task( + 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), + ), + device_id=device_id, + task_type="device_voice_archive", + ) + websocket = await connection_manager.get_connection(device_id) if websocket and websocket.client_state.name == "CONNECTED": success_audio_url = f"http://{settings.server_host}:{settings.server_port}/assets/audio/message_ok.mp3" @@ -365,18 +364,7 @@ async def process_cached_audio(device_id: str, target_device_id: str, serial_num session_logger.info(device_id, "device", f"留言已收到音频URL给设备 {device_id}") else: session_logger.warning(device_id, "device", f"设备 {device_id} 不在线，暂不发送留言已收到音频URL") - # # # 目标设备不在线，保存到离线缓存 - # await offline_audio_cache.add_audio_url(target_device_id, audio_url) - # session_logger.warning(device_id, "target", f"目标设备 {target_device_id} 不在线，保存音频URL到离线缓存") - # websocket = await connection_manager.get_connection(device_id) - # if websocket and websocket.client_state.name == "CONNECTED": - # success_audio_url = f"http://{settings.server_host}:{settings.server_port}/assets/audio/success_zh.mp3" - # await websocket.send_text(f"NFC_MESSAGE_SUCCESS_URL:{success_audio_url}") - # session_logger.info(device_id, "device", f"已发送留言成功音频URL给设备 {device_id}") - # else: - # session_logger.warning(device_id, "device", f"设备 {device_id} 不在线，发送留言成功音频失败") except Exception as e: - # 处理HTTPException异常 if isinstance(e, HTTPException): session_logger.error(device_id, "target", f"保存音频文件时出错，返回HTTPException: {e}") websocket = await connection_manager.get_connection(device_id) diff --git a/talkingq-url/services/offline_audio_cache.py b/talkingq-url/services/offline_audio_cache.py index c0d8dc1..30814d4 100644 --- a/talkingq-url/services/offline_audio_cache.py +++ b/talkingq-url/services/offline_audio_cache.py @@ -15,7 +15,11 @@ class OfflineAudioCache: async def get_audio_urls(self, device_id: str) -> List[str]: async with self.lock: - return self.offline_audio.get(device_id, []) + return list(self.offline_audio.get(device_id, [])) + + async def pop_audio_urls(self, device_id: str) -> List[str]: + async with self.lock: + return self.offline_audio.pop(device_id, []) async def clear_audio_urls(self, device_id: str): async with self.lock: @@ -31,4 +35,4 @@ class OfflineAudioCache: return self.offline_audio # 单例实例 -offline_audio_cache = OfflineAudioCache() \ No newline at end of file +offline_audio_cache = OfflineAudioCache()