修改音频保存逻辑，后续再完善，添加设备信息保存到数据库

talkingq-url/banban/dao/device_setting.py

@@ -0,0 +1,104 @@
+from collections.abc import Mapping
+from datetime import time
+from typing import Optional
+
+from sqlalchemy import text
+
+from banban.dao import BaseDAO
+
+
+class DeviceSettingDAO(BaseDAO):
+    async def create(
+        self,
+        device_id: str,
+        sleep_mode: int = 0,
+        disable_time_start: Optional[time] = None,
+        disable_time_end: Optional[time] = None,
+        timezone: str = "Asia/Shanghai",
+        volume: Optional[int] = None,
+        brightness: Optional[int] = None,
+        disable_weekdays: Optional[str] = None,
+    ) -> int:
+        result = await self.execute(
+            """
+                INSERT INTO device_settings (
+                    device_id, sleep_mode, disable_time_start, disable_time_end,
+                    timezone, volume, brightness, disable_weekdays
+                )
+                VALUES (
+                    :device_id, :sleep_mode, :disable_time_start, :disable_time_end,
+                    :timezone, :volume, :brightness, :disable_weekdays
+                )
+                """,
+            {
+                "device_id": device_id,
+                "sleep_mode": sleep_mode,
+                "disable_time_start": disable_time_start,
+                "disable_time_end": disable_time_end,
+                "timezone": timezone,
+                "volume": volume,
+                "brightness": brightness,
+                "disable_weekdays": disable_weekdays,
+            },
+        )
+        await self.commit()
+        return result.lastrowid
+
+    async def get_by_id(self, setting_id: int) -> Optional[Mapping]:
+        return (
+            await self.execute(
+                "SELECT * FROM device_settings WHERE setting_id = :setting_id",
+                {"setting_id": setting_id},
+            )
+        ).mappings().first()
+
+    async def get_by_device_id(self, device_id: str) -> Optional[Mapping]:
+        return (
+            await self.execute(
+                "SELECT * FROM device_settings WHERE device_id = :device_id",
+                {"device_id": device_id},
+            )
+        ).mappings().first()
+
+    async def update(
+        self,
+        device_id: str,
+        sleep_mode: Optional[int] = None,
+        disable_time_start: Optional[time] = None,
+        disable_time_end: Optional[time] = None,
+        timezone: Optional[str] = None,
+        volume: Optional[int] = None,
+        brightness: Optional[int] = None,
+        disable_weekdays: Optional[str] = None,
+    ) -> None:
+        await self.execute(
+            """
+                UPDATE device_settings
+                SET sleep_mode = COALESCE(:sleep_mode, sleep_mode),
+                    disable_time_start = COALESCE(:disable_time_start, disable_time_start),
+                    disable_time_end = COALESCE(:disable_time_end, disable_time_end),
+                    timezone = COALESCE(:timezone, timezone),
+                    volume = COALESCE(:volume, volume),
+                    brightness = COALESCE(:brightness, brightness),
+                    disable_weekdays = COALESCE(:disable_weekdays, disable_weekdays)
+                WHERE device_id = :device_id
+                """,
+            {
+                "device_id": device_id,
+                "sleep_mode": sleep_mode,
+                "disable_time_start": disable_time_start,
+                "disable_time_end": disable_time_end,
+                "timezone": timezone,
+                "volume": volume,
+                "brightness": brightness,
+                "disable_weekdays": disable_weekdays,
+            },
+        )
+        await self.commit()
+
+    async def delete_by_device_id(self, device_id: str) -> None:
+        await self.execute(
+            "DELETE FROM device_settings WHERE device_id = :device_id",
+            {"device_id": device_id},
+        )
+        await self.commit()

talkingq-url/banban/dao/im.py

@@ -91,6 +91,42 @@ class ImDAO(BaseDAO):
             child_name=row["child_name"],
         )
     
+    async def get_device_by_id(self, *, device_id: str) -> DeviceIdentity:
+        row = (
+            await self.execute(
+                text(
+                    """
+                    SELECT
+                      da.device_id,
+                      db.child_id,
+                      c.child_name
+                    FROM device_auth AS da
+                    LEFT JOIN device_bindings AS db
+                      ON db.device_id = da.device_id
+                     AND db.status = 1
+                    LEFT JOIN children AS c
+                      ON c.child_id = db.child_id
+                     AND c.status = 1
+                    WHERE da.device_id = :device_id
+                      AND da.is_active = 1
+                    LIMIT 1
+                    """
+                ),
+                {"device_id": device_id},
+            )
+        ).mappings().first()
+        if not row:
+            from fastapi import HTTPException, status
+            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid device credentials")
+        if row["child_id"] is None:
+            from fastapi import HTTPException, status
+            raise HTTPException(status_code=404, detail="device not bound to a child")
+        return DeviceIdentity(
+            device_id=str(row["device_id"]),
+            child_id=int(row["child_id"]),
+            child_name=row["child_name"],
+        )
+
     async def assert_child_exists(self, *, child_id: int) -> Mapping[str, Any]:
         child_row = await self._get_child_row(child_id)
         if not child_row:

talkingq-url/banban/dao/location.py

@@ -290,3 +290,62 @@ class LocationDAO(BaseDAO):
 
     async def _next_primary_key(self, table_name: str) -> int | None:
         return None
+    
+    async def insert(self, device_id: str, latitude: float, longitude: float) -> None:
+        insert_current_sql = """
+        INSERT INTO child_location_current
+            (device_id, lat, lng)
+        VALUES
+            (:device_id, :lat, :lng)
+            )
+        """
+        params = {
+            "device_id": device_id,
+            "lat": latitude,
+            "lng": longitude,
+        }
+        await self.execute(text(insert_current_sql), params)
+        await self.commit()
+
+    async def update(self, device_id: str, latitude: float, longitude: float) -> None:
+        update_current_sql = """
+        UPDATE child_location_current
+        SET lat = :lat, lng = :lng
+        WHERE device_id = :device_id
+        """
+        params = {
+            "device_id": device_id,
+            "lat": latitude,
+            "lng": longitude,
+        }
+        await self.execute(text(update_current_sql), params)
+        await self.commit()
+    
+    async def get_device_current_location(self, *, device_id: str) -> Mapping[str, Any]:
+        # sql 查询
+        result = await self.execute(
+            text(
+                f"""
+                SELECT
+                  child_id,
+                  device_id,
+                  coord_type,
+                  lat,
+                  lng,
+                  accuracy_m,
+                  altitude_m,
+                  speed_mps,
+                  heading_deg,
+                  source,
+                  battery_pct,
+                  device_time,
+                  server_time,
+                  updated_at
+                FROM child_location_current
+                WHERE device_id = :device_id
+                LIMIT 1
+                """
+            ),
+            {"device_id": device_id},
+        )
+        return result.mappings().first()

talkingq-url/banban/service/device_setting.py

@@ -0,0 +1,107 @@
+from collections.abc import Mapping
+from datetime import time
+from typing import Optional
+
+from services.database_service_base import DatabaseServiceBase
+
+from banban.dao.device_setting import DeviceSettingDAO
+
+
+class DeviceSettingService(DatabaseServiceBase):
+    def __init__(self):
+        super().__init__(service_name="device_setting_service")
+
+    async def create_setting(
+        self,
+        device_id: str,
+        sleep_mode: int = 0,
+        disable_time_start: Optional[time] = None,
+        disable_time_end: Optional[time] = None,
+        timezone: str = "Asia/Shanghai",
+        power: Optional[int] = None,
+        volume: Optional[int] = None,
+        brightness: Optional[int] = None,
+        disable_weekdays: Optional[str] = None,
+        signal_strength: Optional[int] = None,
+        version_str: Optional[str] = None,
+    ) -> int:
+        db_session = await self.get_session()
+        try:
+            dao = DeviceSettingDAO(db_session)
+            return await dao.create(
+                device_id=device_id,
+                sleep_mode=sleep_mode,
+                disable_time_start=disable_time_start,
+                disable_time_end=disable_time_end,
+                timezone=timezone,
+                power=power,
+                volume=volume,
+                brightness=brightness,
+                disable_weekdays=disable_weekdays,
+                signal_strength=signal_strength,
+                version_str=version_str,
+            )
+        finally:
+            await db_session.close()
+
+    async def get_setting_by_device_id(self, device_id: str) -> Optional[Mapping]:
+        db_session = await self.get_session()
+        try:
+            dao = DeviceSettingDAO(db_session)
+            return await dao.get_by_device_id(device_id=device_id)
+        finally:
+            await db_session.close()
+
+    async def update_setting(
+        self,
+        device_id: str,
+        sleep_mode: Optional[int] = None,
+        disable_time_start: Optional[time] = None,
+        disable_time_end: Optional[time] = None,
+        timezone: Optional[str] = None,
+        power: Optional[int] = None,
+        volume: Optional[int] = None,
+        brightness: Optional[int] = None,
+        disable_weekdays: Optional[str] = None,
+        signal_strength: Optional[int] = None,
+        version_str: Optional[str] = None,
+    ) -> None:
+        db_session = await self.get_session()
+        try:
+            dao = DeviceSettingDAO(db_session)
+            await dao.update(
+                device_id=device_id,
+                sleep_mode=sleep_mode,
+                disable_time_start=disable_time_start,
+                disable_time_end=disable_time_end,
+                timezone=timezone,
+                volume=volume,
+                brightness=brightness,
+                disable_weekdays=disable_weekdays,
+                power=power,
+                signal_strength=signal_strength,
+                version_str=version_str,
+            )
+        finally:
+            await db_session.close()
+
+    async def delete_setting(self, device_id: str) -> None:
+        db_session = await self.get_session()
+        try:
+            dao = DeviceSettingDAO(db_session)
+            await dao.delete_by_device_id(device_id=device_id)
+        finally:
+            await db_session.close()
+
+    async def insert_or_update(self, device_id: str, power: int, volume: int, signal_strength: int, version_str: str) -> None:
+        try:
+            current_row = await self.get_setting_by_device_id(device_id=device_id)
+            if current_row:
+                await self.update_setting(device_id=device_id, power=power, volume=volume, signal_strength=signal_strength, version_str=version_str)
+            else:
+                await self.create_setting(device_id=device_id, power=power, volume=volume, signal_strength=signal_strength, version_str=version_str)
+        finally:
+            pass
+
+# 创建全局 DeviceSettingService 实例
+device_setting_service = DeviceSettingService()

talkingq-url/banban/service/im.py

@@ -251,6 +251,55 @@ class ImService(DatabaseServiceBase):
         finally:
             await db_session.close()
 
+    '''
+    Todo 创建设备消息, 还不完善
+    '''
+    async def create_device_message(
+        self,
+        *,
+        device_id: str,
+        serial_number: str,
+        target_device_id: str,
+        audio_url: str,
+    ):
+        db_session = await self.get_session()
+        try:
+            dao = ImDAO(db_session)
+            device_identity = await dao.authenticate_device_identity(
+                device_id=device_id,
+                serial_number=serial_number,
+            )
+            target_device_identity = await dao.get_device_by_id(device_id=target_device_id)
+
+            sender_child_row = await dao.assert_child_exists(child_id=device_identity.child_id)
+            receiver_child_row = await dao.assert_child_exists(child_id=target_device_identity.child_id)
+
+            conversation_id, idempotent = await dao.create_message(
+                conversation_type=PARENT_CHILD_CONVERSATION_TYPE,
+                participant_a_type=CHILD_PARTICIPANT_TYPE,
+                participant_a_id=str(device_identity.child_id),
+                participant_b_type=PARENT_PARTICIPANT_TYPE,
+                participant_b_id=str(receiver_child_row.child_id),
+                pair_key=f"{device_identity.child_id}:{target_device_identity.child_id}",
+                sender_type=CHILD_PARTICIPANT_TYPE,
+                sender_id=str(device_identity.child_id),
+                receiver_type=PARENT_PARTICIPANT_TYPE,
+                receiver_id=str(receiver_child_row.child_id),
+                sender_name_snapshot=sender_child_row["child_name"],
+                sender_avatar_snapshot=None,
+                receiver_name_snapshot=receiver_child_row["child_name"],
+                receiver_avatar_snapshot=None,
+                payload=None,
+            )
+
+            
+            return device_identity
+        except Exception:
+            await db_session.rollback()
+            raise
+        finally:
+            await db_session.close()
+
 
 # 创建全局 ImService 实例
 im_service = ImService()

talkingq-url/banban/service/location.py

@@ -4,7 +4,7 @@ from datetime import datetime
 from typing import Any
 
 from services.database_service_base import DatabaseServiceBase
-
+from database.models import ChildLocationCurrent
 try:
     from banban.dao.location import LocationDAO, ParentDeviceAccess
     from banban.schemas.location import DeviceLocationReportRequest
@@ -85,6 +85,18 @@ class LocationService(DatabaseServiceBase):
         finally:
             await db_session.close()
 
+    async def insert_or_update(self, device_id: str, location: ChildLocationCurrent) -> None:
+        db_session = await self.get_session()
+        try:
+            dao = LocationDAO(db_session)
+            current_row = await dao.get_device_current_location(device_id=device_id)
+            if current_row:
+                await dao.update(device_id=device_id, latitude=location.lat, longitude=location.lng)
+                # 更新成功加到历史记录表
+            else:
+                await dao.insert(device_id=device_id, latitude=location.lat, longitude=location.lng)
+        finally:
+            await db_session.close()
 
 # 创建全局 LocationService 实例
 location_service = LocationService()

talkingq-url/database/models.py

@@ -257,6 +257,9 @@ class DeviceSetting(Base):
     volume: Mapped[Optional[int]] = mapped_column(Integer)
     brightness: Mapped[Optional[int]] = mapped_column(Integer)
     disable_weekdays: Mapped[Optional[str]] = mapped_column(String(32))
+    power: Mapped[Optional[int]] = mapped_column(Integer)
+    signal_strength: Mapped[Optional[int]] = mapped_column(Integer)
+    version_str: Mapped[Optional[str]] = mapped_column(String(64))
     created_at: Mapped[Optional[datetime]] = mapped_column(DateTime, server_default=text("CURRENT_TIMESTAMP"))
     updated_at: Mapped[Optional[datetime]] = mapped_column(
         DateTime,

talkingq-url/handlers/mqtt_handler.py

@@ -2,10 +2,13 @@ import json
 import time
 import asyncio
 from typing import Optional, Dict, Callable, Awaitable
+from banban.service.device_setting import device_setting_service
 from config import settings
 from services.card_service import card_service
 from services.offline_audio_cache import offline_audio_cache
 import aiomqtt
+from banban.service.location import location_service
+from database.models import ChildLocationCurrent
 
 from services.device_target_cache import device_target_cache
 from utils.logger import session_logger as logger
@@ -97,6 +100,8 @@ class TalkingQMQTTService:
         version = data.get("version")
         voice = data.get("voice")
         logger.info(device_id, "", f"[设备信息] 设备 {d_id} 信息: 电量={power}, 信号强度={signal}, 版本号={version}, 音量={voice}")
+        # 插入到数据库
+        await device_setting_service.insert_or_update(device_id=device_id, power=power, signal_strength=signal, version_str=version, volume=voice)
         # else:
         #     logger.warning(device_id, "", f"[设备信息] 设备 {device_id} 查询失败: {payload}")
         await self._publish(f"device/{device_id}/event_resp", {"msg_id": "000", "status": "success"})
@@ -108,6 +113,9 @@ class TalkingQMQTTService:
             lat = data.get("latitude")
             lon = data.get("longitude")
             logger.info(device_id, "", f"[GPS] 设备 {device_id} 位置: 纬度={lat}, 经度={lon}")
+            # 插入到数据库
+            location = ChildLocationCurrent(device_id=device_id, lat=lat, lon=lon)
+            await location_service.insert_or_update(device_id=device_id, location=location)
         else:
             logger.warning(device_id, "", f"[GPS] 设备 {device_id} 查询失败: {payload}")
 
@@ -117,6 +125,9 @@ class TalkingQMQTTService:
         if status == "success":
             level = data.get("current_level")
             logger.info(device_id, "", f"[音量] 设备 {device_id} 当前音量: {level}")
+            # 更新设备音量
+            await device_setting_service.insert_or_update(device_id=device_id, volume=level)
+
         else:
             logger.warning(device_id, "", f"[音量] 设备 {device_id} 调节失败: {payload}")
 
@@ -130,6 +141,8 @@ class TalkingQMQTTService:
         elif status == "success":
             new_ver = data.get("new_version")
             logger.info(device_id, "", f"[OTA] 设备 {device_id} 升级完成: {new_ver}")
+            # 更新设备版本号
+            await device_setting_service.insert_or_update(device_id=device_id, version_str=new_ver)
         else:
             logger.warning(device_id, "", f"[OTA] 设备 {device_id} 升级异常: {payload}")
 

talkingq-url/handlers/websocket_auth_handler.py

@@ -17,6 +17,7 @@ async def authenticate_websocket(websocket: WebSocket):
         tuple: (认证状态, 设备ID) - (是否认证成功, 设备ID)
     """
     device_id = None
+    serial_number = None
     authenticated = False
     
     auth_timeout = 10  # 10秒认证超时
@@ -63,4 +64,4 @@ async def authenticate_websocket(websocket: WebSocket):
             session_logger.warning("unknown", "auth", "WebSocket认证超时")
             break
     
-    return authenticated, device_id
+    return authenticated, device_id, serial_number

talkingq-url/handlers/websocket_handler.py

@@ -21,7 +21,7 @@ async def websocket_endpoint(websocket: WebSocket):
     device_id = None
     
     try:
-        authenticated, device_id = await authenticate_websocket(websocket)
+        authenticated, device_id, serial_number = await authenticate_websocket(websocket)
         
         if not authenticated:
             session_logger.warning(
@@ -44,7 +44,7 @@ async def websocket_endpoint(websocket: WebSocket):
         #     await offline_audio_cache.clear_audio_urls(device_id)
         #     session_logger.info(device_id, "offline", f"已清空设备的离线音频URL缓存，共 {len(audio_urls)} 个")
         
-        await handle_websocket_messages(websocket, device_id)
+        await handle_websocket_messages(websocket, device_id, serial_number)
         
     except WebSocketDisconnect:
         session_logger.info(

talkingq-url/handlers/websocket_message_handler.py

@@ -17,8 +17,9 @@ from utils.logger import session_logger
 from handlers.prompt_sound_handler import handle_prompt_sound_request
 from handlers.session_cleanup_handler import handle_old_session_cleanup
 from config import settings
+from banban.service.im import im_service as im_conversation_service
 
-async def handle_websocket_messages(websocket: WebSocket, device_id: str):
+async def handle_websocket_messages(websocket: WebSocket, device_id: str, serial_number: str):
     """
     处理WebSocket连接中的所有消息
     
@@ -49,6 +50,7 @@ async def handle_websocket_messages(websocket: WebSocket, device_id: str):
                 current_active_session, first_audio_received_time = await handle_binary_message(
                     websocket, 
                     device_id, 
+                    serial_number,
                     data, 
                     current_active_session,
                     first_audio_received_time
@@ -134,7 +136,7 @@ async def handle_text_message(websocket: WebSocket, device_id: str, text_data: s
         )
 
 
-async def handle_binary_message(websocket: WebSocket, device_id: str, binary_data, current_active_session, first_audio_received_time):
+async def handle_binary_message(websocket: WebSocket, device_id: str, serial_number: str, binary_data, current_active_session, first_audio_received_time):
     """处理二进制音频消息"""
     session_key, session, packet_type, audio_data, sample_rate = await parse_packet(binary_data, websocket)
     
@@ -163,7 +165,7 @@ async def handle_binary_message(websocket: WebSocket, device_id: str, binary_dat
             if target_device_id:
                 # 处理缓存的音频数据
                 session_logger.info(device_id, session_id, f"收到结束包，开始处理缓存音频数据")
-                await process_cached_audio(device_id, target_device_id)
+                await process_cached_audio(device_id, target_device_id, serial_number)
                 
                 # 移除目标设备关联
                 await device_target_cache.remove_target(device_id)
@@ -253,7 +255,7 @@ async def handle_target_audio_packet(device_id: str, audio_data: bytes):
         session_logger.error(device_id, "target", f"处理目标音频包时出错: {e}", exc_info=True)
 
 
-async def process_cached_audio(device_id: str, target_device_id: str):
+async def process_cached_audio(device_id: str, target_device_id: str, serial_number: str):
     """处理缓存的音频数据并发送音频URL"""
     try:
         # 获取缓存的音频数据
@@ -265,6 +267,8 @@ async def process_cached_audio(device_id: str, target_device_id: str):
         # 保存音频文件
         audio_path = await save_audio_file(cached_audio, device_id)
         audio_url = f"http://{settings.server_host}:{settings.server_port}/{audio_path}"
+        # 将音频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_url)
         
         # 发送URL给目标设备
         # target_websocket = await connection_manager.get_connection(target_device_id)