From 7d4ae15fb50b6fdd10e19b12c028752785b49347 Mon Sep 17 00:00:00 2001 From: HycJack <772403255@qq.com> Date: Wed, 29 Apr 2026 04:24:44 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=94=B9=E9=9F=B3=E9=A2=91=E4=BF=9D?= =?UTF-8?q?=E5=AD=98=E9=80=BB=E8=BE=91=EF=BC=8C=E5=90=8E=E7=BB=AD=E5=86=8D?= =?UTF-8?q?=E5=AE=8C=E5=96=84=EF=BC=8C=E6=B7=BB=E5=8A=A0=E8=AE=BE=E5=A4=87?= =?UTF-8?q?=E4=BF=A1=E6=81=AF=E4=BF=9D=E5=AD=98=E5=88=B0=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E5=BA=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- talkingq-url/banban/dao/device_setting.py | 104 +++++++++++++++++ talkingq-url/banban/dao/im.py | 36 ++++++ talkingq-url/banban/dao/location.py | 61 +++++++++- talkingq-url/banban/service/device_setting.py | 107 ++++++++++++++++++ talkingq-url/banban/service/im.py | 49 ++++++++ talkingq-url/banban/service/location.py | 14 ++- talkingq-url/database/models.py | 3 + talkingq-url/handlers/mqtt_handler.py | 13 +++ .../handlers/websocket_auth_handler.py | 3 +- talkingq-url/handlers/websocket_handler.py | 4 +- .../handlers/websocket_message_handler.py | 12 +- 11 files changed, 397 insertions(+), 9 deletions(-) create mode 100644 talkingq-url/banban/dao/device_setting.py create mode 100644 talkingq-url/banban/service/device_setting.py diff --git a/talkingq-url/banban/dao/device_setting.py b/talkingq-url/banban/dao/device_setting.py new file mode 100644 index 0000000..2c41c84 --- /dev/null +++ b/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() \ No newline at end of file diff --git a/talkingq-url/banban/dao/im.py b/talkingq-url/banban/dao/im.py index d1b01b9..b4ab869 100644 --- a/talkingq-url/banban/dao/im.py +++ b/talkingq-url/banban/dao/im.py @@ -90,6 +90,42 @@ class ImDAO(BaseDAO): child_id=int(row["child_id"]), 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) diff --git a/talkingq-url/banban/dao/location.py b/talkingq-url/banban/dao/location.py index fcddb9c..56aacaa 100644 --- a/talkingq-url/banban/dao/location.py +++ b/talkingq-url/banban/dao/location.py @@ -289,4 +289,63 @@ class LocationDAO(BaseDAO): return result.mappings().first() async def _next_primary_key(self, table_name: str) -> int | None: - return None \ No newline at end of file + 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() \ No newline at end of file diff --git a/talkingq-url/banban/service/device_setting.py b/talkingq-url/banban/service/device_setting.py new file mode 100644 index 0000000..3052be5 --- /dev/null +++ b/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() \ No newline at end of file diff --git a/talkingq-url/banban/service/im.py b/talkingq-url/banban/service/im.py index 0d4d699..9c5574d 100644 --- a/talkingq-url/banban/service/im.py +++ b/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() diff --git a/talkingq-url/banban/service/location.py b/talkingq-url/banban/service/location.py index e85d37e..e053d84 100644 --- a/talkingq-url/banban/service/location.py +++ b/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() diff --git a/talkingq-url/database/models.py b/talkingq-url/database/models.py index 41ab0e7..f900c29 100644 --- a/talkingq-url/database/models.py +++ b/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, diff --git a/talkingq-url/handlers/mqtt_handler.py b/talkingq-url/handlers/mqtt_handler.py index ce895fc..44774e0 100644 --- a/talkingq-url/handlers/mqtt_handler.py +++ b/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}") diff --git a/talkingq-url/handlers/websocket_auth_handler.py b/talkingq-url/handlers/websocket_auth_handler.py index 3c39c01..e297bfb 100644 --- a/talkingq-url/handlers/websocket_auth_handler.py +++ b/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 diff --git a/talkingq-url/handlers/websocket_handler.py b/talkingq-url/handlers/websocket_handler.py index 1dfc89f..32fc339 100644 --- a/talkingq-url/handlers/websocket_handler.py +++ b/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( diff --git a/talkingq-url/handlers/websocket_message_handler.py b/talkingq-url/handlers/websocket_message_handler.py index 7e474e3..c3c4025 100644 --- a/talkingq-url/handlers/websocket_message_handler.py +++ b/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)