huangyicao/banban
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: cache device voice audio for playback

Browse Source
This commit is contained in:
stu2not
2026-05-15 15:34:34 +08:00
parent b71911d8c2
commit ac14bfc47f
10 changed files with 297 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
talkingq-url/.gitignore vendored
View File

@@ -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
Thumbs.db

10
talkingq-url/api/assets.py
View File

@@ -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")
app.mount("/assets", StaticFiles(directory=settings.assets_dir), name="assets")
app.mount(
"/device-audio",
StaticFiles(directory=device_audio_directory),
name="device-audio",
)

11
talkingq-url/banban/routers/device_im.py
View File

@@ -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,
)
)

7
talkingq-url/banban/routers/mqtt_router.py
View File

@@ -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="已设置未读留言")

174
talkingq-url/banban/service/device_audio_cache.py Normal file
View File

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

40
talkingq-url/banban/service/device_voice_archive.py
View File

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

31
talkingq-url/banban/service/im.py
View File

@@ -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:

12
talkingq-url/config.py
View File

@@ -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",

60
talkingq-url/handlers/websocket_message_handler.py
View File

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

8
talkingq-url/services/offline_audio_cache.py
View File

@@ -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()
offline_audio_cache = OfflineAudioCache()