合入音频支持http和https兼容，后端支持角色切换

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

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

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

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"]
+__all__ = ["banban_router"]

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

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}")  

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

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

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

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

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

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:

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

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

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
 

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)

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