From d93bba10b91374a2c9b563b1e9afeda9b410a192 Mon Sep 17 00:00:00 2001 From: HycJack <772403255@qq.com> Date: Mon, 18 May 2026 19:56:32 +0800 Subject: [PATCH] =?UTF-8?q?=E5=90=88=E5=85=A5=E9=9F=B3=E9=A2=91=E6=94=AF?= =?UTF-8?q?=E6=8C=81http=E5=92=8Chttps=E5=85=BC=E5=AE=B9=EF=BC=8C=E5=90=8E?= =?UTF-8?q?=E7=AB=AF=E6=94=AF=E6=8C=81=E8=A7=92=E8=89=B2=E5=88=87=E6=8D=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- talkingq-url/.gitignore | 3 +- talkingq-url/api/assets.py | 10 +- talkingq-url/banban/dao/device.py | 50 +++++ talkingq-url/banban/routers/__init__.py | 4 +- talkingq-url/banban/routers/device_im.py | 11 +- talkingq-url/banban/routers/devices.py | 108 ++++++++--- talkingq-url/banban/routers/mqtt_router.py | 7 +- talkingq-url/banban/routers/roles.py | 134 ++++++++++++++ talkingq-url/banban/service/device.py | 20 ++ .../banban/service/device_audio_cache.py | 174 ++++++++++++++++++ .../banban/service/device_voice_archive.py | 40 ++-- talkingq-url/banban/service/im.py | 31 +++- talkingq-url/banban/service/location.py | 26 ++- talkingq-url/config.py | 12 ++ talkingq-url/handlers/mqtt_handler.py | 41 +---- .../handlers/websocket_message_handler.py | 60 +++--- talkingq-url/services/offline_audio_cache.py | 8 +- 17 files changed, 613 insertions(+), 126 deletions(-) create mode 100644 talkingq-url/banban/routers/roles.py create mode 100644 talkingq-url/banban/service/device_audio_cache.py 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/dao/device.py b/talkingq-url/banban/dao/device.py index 73852e1..467ae70 100644 --- a/talkingq-url/banban/dao/device.py +++ b/talkingq-url/banban/dao/device.py @@ -62,6 +62,56 @@ class DeviceDAO(BaseDAO): ) return result.mappings().all() + async def list_device_ai_conversations( + self, + *, + device_id: str, + cursor: int | None, + limit: int, + ) -> List[Mapping[str, Any]]: + params = {"device_id": device_id, "limit": limit + 1} + where = "ch.device_id = :device_id" + if cursor is not None: + where += " AND ch.id < :cursor" + params["cursor"] = cursor + + result = await self.execute( + text( + f""" + SELECT + ch.id AS conversation_id, + ch.role_key, + COALESCE(r.name, ch.role_key) AS role_name, + r.description AS role_description, + COALESCE(stats.message_count, 0) AS message_count, + latest.content AS last_message_preview, + latest.created_at AS last_message_at, + ch.last_interaction_time, + ch.created_at, + ch.updated_at + FROM conversation_histories AS ch + LEFT JOIN roles AS r + ON r.role_key = ch.role_key + LEFT JOIN ( + SELECT + conversation_id, + COUNT(*) AS message_count, + MAX(id) AS latest_message_id + FROM conversation_messages + GROUP BY conversation_id + ) AS stats + ON stats.conversation_id = ch.id + LEFT JOIN conversation_messages AS latest + ON latest.id = stats.latest_message_id + WHERE {where} + ORDER BY COALESCE(latest.created_at, ch.updated_at, ch.created_at) DESC, ch.id DESC + LIMIT :limit + """ + ), + params, + ) + return result.mappings().all() + async def get_device_status( self, *, diff --git a/talkingq-url/banban/routers/__init__.py b/talkingq-url/banban/routers/__init__.py index e05a878..6bc45f9 100644 --- a/talkingq-url/banban/routers/__init__.py +++ b/talkingq-url/banban/routers/__init__.py @@ -7,6 +7,7 @@ from banban.routers.device_location import router as device_location_router from banban.routers.devices import router as devices_router from banban.routers.im import router as im_router from banban.routers.parents import router as parents_router +from banban.routers.roles import router as roles_router from banban.routers.wechat_auth import router as wechat_auth_router from banban.routers.mqtt_router import router as mqtt_router @@ -22,7 +23,8 @@ banban_router.include_router(device_im_router, tags=["banban-device-im"]) banban_router.include_router(device_location_router, tags=["banban-device-location"]) banban_router.include_router(im_router, tags=["banban-im"]) banban_router.include_router(parents_router, tags=["banban-parents"]) +banban_router.include_router(roles_router, tags=["banban-roles"]) banban_router.include_router(mqtt_router, tags=["banban-mqtt"]) -__all__ = ["banban_router"] \ No newline at end of file +__all__ = ["banban_router"] 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/devices.py b/talkingq-url/banban/routers/devices.py index d8336db..6dd9ab4 100644 --- a/talkingq-url/banban/routers/devices.py +++ b/talkingq-url/banban/routers/devices.py @@ -1,7 +1,6 @@ import logging from collections.abc import Mapping from datetime import datetime, time -from services.connection_manager import connection_manager from fastapi import APIRouter, Depends, HTTPException, Query, Request from pydantic import BaseModel, Field from sqlalchemy import text @@ -51,6 +50,24 @@ class DeviceMessageListResponse(BaseModel): next_cursor: int | None = None +class DeviceAiConversationItem(BaseModel): + conversation_id: int + role_key: str + role_name: str + role_description: str | None = None + message_count: int + last_message_preview: str | None = None + last_message_at: datetime | None = None + created_at: datetime + updated_at: datetime + + +class DeviceAiConversationListResponse(BaseModel): + items: list[DeviceAiConversationItem] + total: int + next_cursor: int | None = None + + class DeviceStatusResponse(BaseModel): device_id: str child_id: int | None = None @@ -159,6 +176,20 @@ def _row_to_message_item(row: Mapping) -> DeviceMessageItem: ) +def _row_to_ai_conversation_item(row: Mapping) -> DeviceAiConversationItem: + return DeviceAiConversationItem( + conversation_id=int(row["conversation_id"]), + role_key=str(row["role_key"]), + role_name=str(row["role_name"] or row["role_key"]), + role_description=row.get("role_description"), + message_count=int(row["message_count"] or 0), + last_message_preview=row.get("last_message_preview"), + last_message_at=row.get("last_message_at"), + created_at=row["created_at"], + updated_at=row["updated_at"], + ) + + def _row_to_current_location_response(row: Mapping) -> DeviceLocationCurrentResponse: return DeviceLocationCurrentResponse( child_id=int(row["child_id"]), @@ -293,6 +324,43 @@ async def list_device_messages( ) +@router.get("/{device_id}/ai-conversations", response_model=DeviceAiConversationListResponse) +async def list_device_ai_conversations( + device_id: str, + request: Request, + cursor: int | None = Query(default=None, ge=1), + limit: int = Query(default=20, ge=1, le=100), + current_user_id: int = Depends(get_current_user_id), +) -> DeviceAiConversationListResponse: + rows = await device_service.list_device_ai_conversations( + device_id=device_id, + user_id=current_user_id, + cursor=cursor, + limit=limit, + ) + + has_more = len(rows) > limit + rows = rows[:limit] + next_cursor = int(rows[-1]["conversation_id"]) if has_more and rows else None + + logger.info( + "listed device ai conversations", + extra={ + "event": "device_ai_conversations", + "request_id": getattr(request.state, "request_id", None), + "user_id": current_user_id, + "device_id": device_id, + "returned_count": len(rows), + }, + ) + + return DeviceAiConversationListResponse( + items=[_row_to_ai_conversation_item(row) for row in rows], + total=len(rows), + next_cursor=next_cursor, + ) + + @router.get("/{device_id}/status", response_model=DeviceStatusResponse) async def get_device_status( device_id: str, @@ -504,15 +572,12 @@ async def get_current_device_location( service = await TalkingQMQTTService.get_instance() if service is None: raise HTTPException(status_code=503, detail="MQTT 服务未初始化") + query_started_at = datetime.now() await service.send_gps_query(device_id) try: - # 每隔1s 获取一次GPS数据，最多3次 - # 先判断设备是否在线，不在线直接提示设备没有在线，通过websocket判断 - target_websocket = await connection_manager.get_connection(device_id) - if not target_websocket or target_websocket.client_state.name != "CONNECTED": - raise HTTPException(status_code=408, detail="设备未在线") - i=0 - while i<1: + # MQTT 设备没有 WebSocket 连接时，也可以通过 GPS 回包刷新定位。 + row = None + for _ in range(3): try: row = await location_service.get_device_current_location( device_id=device_id, @@ -521,12 +586,10 @@ async def get_current_device_location( except Exception as e: await asyncio.sleep(1) continue - - if row is None: - await asyncio.sleep(1) - else: - if row["updated_at"] is not None \ - and (datetime.now() - row['updated_at']) <= timedelta(seconds=10): + + if row is not None and row["updated_at"] is not None: + updated_at = row["updated_at"] + if updated_at >= query_started_at - timedelta(seconds=1): logger.info( "device current location fetched", extra={ @@ -538,22 +601,13 @@ async def get_current_device_location( }, ) return _row_to_current_location_response(row) - - i+=1 + + await asyncio.sleep(1) + if row is None: raise HTTPException(status_code=408, detail="GPS数据上报超时") - logger.info( - "device location reported", - extra={ - "event": "device_location_report", - "request_id": getattr(request.state, "request_id", None), - "device_id": device_id, - "child_id": int(row["child_id"]), - "lat": float(row["lat"]), - "lng": float(row["lng"]), - }, - ) + raise HTTPException(status_code=408, detail="GPS数据未刷新") except Exception as e: raise HTTPException(status_code=408, detail=f"GPS数据上报失败: {e}") 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/routers/roles.py b/talkingq-url/banban/routers/roles.py new file mode 100644 index 0000000..af11547 --- /dev/null +++ b/talkingq-url/banban/routers/roles.py @@ -0,0 +1,134 @@ +from collections.abc import Mapping +from typing import List + +from fastapi import APIRouter, Depends, HTTPException +from pydantic import BaseModel + +from banban.security import get_current_user_id +from banban.service.device import device_service +from handlers.prompt_sound_handler import send_welcome_sound +from services.conversation_history import DeviceConversationHistory, conversation_history_manager +from services.device_config import DeviceConfig, device_config_manager +from services.role_manager import role_manager + + +router = APIRouter(prefix="/roles", tags=["banban-roles"]) + + +class RoleSummaryResponse(BaseModel): + role_key: str + name: str + description: str | None = None + default_language: str | None = None + languages: List[str] = [] + + +class DeviceRoleResponse(BaseModel): + device_id: str + role_key: str + name: str + description: str | None = None + preferred_language: str | None = None + languages: List[str] = [] + + +class DeviceRoleUpdateRequest(BaseModel): + role_key: str + language: str | None = None + play_welcome: bool = False + + +def _role_languages(role: Mapping) -> list[str]: + multilingual = role.get("multilingual") + if isinstance(multilingual, dict): + return list(multilingual.keys()) + default_language = role.get("default_language") + return [str(default_language)] if default_language else [] + + +def _role_to_summary(role: Mapping) -> RoleSummaryResponse: + return RoleSummaryResponse( + role_key=str(role.get("role_key") or ""), + name=str(role.get("name") or role.get("role_key") or ""), + description=role.get("description"), + default_language=role.get("default_language"), + languages=_role_languages(role), + ) + + +@router.get("", response_model=list[RoleSummaryResponse]) +async def list_roles( + current_user_id: int = Depends(get_current_user_id), +) -> list[RoleSummaryResponse]: + del current_user_id + roles = await role_manager.get_all_roles() + return [ + _role_to_summary(role) + for role in sorted(roles.values(), key=lambda item: str(item.get("role_key") or "")) + ] + + +@router.get("/devices/{device_id}", response_model=DeviceRoleResponse) +async def get_device_role( + device_id: str, + current_user_id: int = Depends(get_current_user_id), +) -> DeviceRoleResponse: + await device_service.ensure_device_access(device_id=device_id, user_id=current_user_id) + config = await device_config_manager.get_config(device_id, force_refresh=True) + role = await role_manager.get_role(config.selected_role_key) + if not role: + raise HTTPException(status_code=404, detail="role not found") + + return DeviceRoleResponse( + device_id=device_id, + role_key=config.selected_role_key, + name=str(role.get("name") or config.selected_role_key), + description=role.get("description"), + preferred_language=config.preferred_language, + languages=_role_languages(role), + ) + + +@router.put("/devices/{device_id}", response_model=DeviceRoleResponse) +async def update_device_role( + device_id: str, + payload: DeviceRoleUpdateRequest, + current_user_id: int = Depends(get_current_user_id), +) -> DeviceRoleResponse: + await device_service.ensure_device_access(device_id=device_id, user_id=current_user_id) + + role_key = payload.role_key.strip() + if not role_key: + raise HTTPException(status_code=422, detail="role_key is required") + + role = await role_manager.get_role(role_key) + if not role: + raise HTTPException(status_code=404, detail="role not found") + + current_config = await device_config_manager.get_config(device_id, force_refresh=True) + language = payload.language or current_config.preferred_language or role.get("default_language") + available_languages = _role_languages(role) + if language and available_languages and language not in available_languages: + raise HTTPException(status_code=422, detail="language is not supported by role") + + await device_config_manager.set_config( + device_id, + DeviceConfig(selected_role_key=role_key, preferred_language=language), + ) + await conversation_history_manager.set_history( + device_id, + DeviceConversationHistory(), + role_key, + ) + + if payload.play_welcome: + await send_welcome_sound(device_id, role_key, language) + + return DeviceRoleResponse( + device_id=device_id, + role_key=role_key, + name=str(role.get("name") or role_key), + description=role.get("description"), + preferred_language=language, + languages=available_languages, + ) diff --git a/talkingq-url/banban/service/device.py b/talkingq-url/banban/service/device.py index 6c28ee3..0cfe1e0 100644 --- a/talkingq-url/banban/service/device.py +++ b/talkingq-url/banban/service/device.py @@ -43,6 +43,26 @@ class DeviceService(DatabaseServiceBase): finally: await db_session.close() + async def list_device_ai_conversations( + self, + *, + device_id: str, + user_id: int, + cursor: int | None, + limit: int, + ) -> List[Mapping[str, Any]]: + db_session = await self.get_session() + try: + dao = DeviceDAO(db_session) + await dao.ensure_device_access(device_id=device_id, user_id=user_id) + return await dao.list_device_ai_conversations( + device_id=device_id, + cursor=cursor, + limit=limit, + ) + finally: + await db_session.close() + async def get_device_status( self, *, 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/banban/service/location.py b/talkingq-url/banban/service/location.py index a60b5a8..7c94057 100644 --- a/talkingq-url/banban/service/location.py +++ b/talkingq-url/banban/service/location.py @@ -109,7 +109,7 @@ class LocationService(DatabaseServiceBase): altitude_m: float | None = None, speed_mps: float | None = None, heading_deg: int | None = None, - source: int | None = None, + source: int | str | None = None, battery_pct: int | None = None, device_time: datetime | None = None, ) -> Mapping[str, Any] | None: @@ -129,6 +129,8 @@ class LocationService(DatabaseServiceBase): battery_pct: int | None device_time: datetime + source_value = self._normalize_location_source(source) + db_session = await self.get_session() try: dao = LocationDAO(db_session) @@ -144,7 +146,7 @@ class LocationService(DatabaseServiceBase): altitude_m=altitude_m, speed_mps=speed_mps, heading_deg=heading_deg, - source=source if source is not None else 0, + source=source_value, battery_pct=battery_pct, device_time=device_time or datetime.now(), ) @@ -156,5 +158,25 @@ class LocationService(DatabaseServiceBase): finally: await db_session.close() + @staticmethod + def _normalize_location_source(source: int | str | None) -> int: + if source is None: + return 0 + if isinstance(source, int): + return source + text = str(source).strip().lower() + if not text: + return 0 + if text.isdigit(): + return int(text) + return { + "gps": 1, + "wifi": 2, + "cell": 3, + "base_station": 3, + "manual": 4, + "mock": 9, + }.get(text, 0) + # 创建全局 LocationService 实例 location_service = LocationService() 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/mqtt_handler.py b/talkingq-url/handlers/mqtt_handler.py index 6ef24cc..3507718 100644 --- a/talkingq-url/handlers/mqtt_handler.py +++ b/talkingq-url/handlers/mqtt_handler.py @@ -449,43 +449,10 @@ class TalkingQMQTTService: is_owner = await card_service.check_card_ownership(nfc_uuid, device_id) if is_owner: - has_pending = await offline_audio_cache.has_pending_audio(device_id) - if nfc_uuid == "53C22B6DA20001": - payload = { - "msg_id": "005", - "type": 0, - "params": { - "url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/test_zh.mp3" - } - } - await self._publish(topic, payload) - return - - if has_pending: - audio_urls = await offline_audio_cache.get_audio_urls(device_id) - # 53D92B6DA20001 测试卡片 - if nfc_uuid == "53C22B6DA20001": - payload = { - "msg_id": "005", - "type": 0, - "params": { - "url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/TalkingQ_XQSN00001005_2f654480.mp3" - } - } - await self._publish(topic, payload) - return - if len(audio_urls) == 0: - payload = { - "msg_id": "005", - "type": 0, - "params": { - "url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/no_message.mp3" - } - } - else: - params = {f"url_{k}": audio_url for k, audio_url in enumerate(audio_urls, start=1)} - payload = {"msg_id": "005", "type": 0, "params": params} - await offline_audio_cache.clear_audio_urls(device_id) + 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)} + payload = {"msg_id": "005", "type": 0, "params": params} await self._publish(topic, payload) return 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()