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

Merge branch 'main' into test-clean

Browse Source
This commit is contained in:
stu2not
2026-05-07 09:32:00 +08:00
parent 3385c882ec 3866cfc84a
commit cfa6a77134
84 changed files with 1945 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
talkingq-url/handlers/audio_file_handler.py
View File

@@ -5,15 +5,53 @@ from utils.logger import session_logger
message_audio_storage_service = MessageAudioStorageService()
# async def save_audio_file(audio_data: bytes, device_id: str) -> str:
# """Upload device audio to COS and return its object key."""
# try:
# stored = await message_audio_storage_service.upload_audio(
# device_id=device_id,
# content=audio_data,
# )
# session_logger.info(device_id, "audio", f"audio uploaded to COS: {stored.file_key}")
# return stored.file_key
# except Exception as e:
# session_logger.error(device_id, "audio", f"failed to store audio: {e}", exc_info=True)
# raise
import os
import uuid
from config import settings
from utils.logger import session_logger
# from utils.audio_denoiser import reduce_background_noise
async def save_audio_file(audio_data: bytes, device_id: str) -> str:
"""Upload device audio to COS and return its object key."""
"""
保存音频数据到 assets/audio 目录
Args:
audio_data: 音频二进制数据
device_id: 设备ID
Returns:
音频文件的相对路径
"""
try:
stored = await message_audio_storage_service.upload_audio(
device_id=device_id,
content=audio_data,
)
session_logger.info(device_id, "audio", f"audio uploaded to COS: {stored.file_key}")
return stored.file_key
audio_dir = os.path.join(settings.assets_dir, "audio")
os.makedirs(audio_dir, exist_ok=True)
filename = f"{device_id}_{uuid.uuid4().hex[:8]}.mp3"
filepath = os.path.join(audio_dir, filename)
with open(filepath, 'wb') as f:
f.write(audio_data)
# relative_path = f"assets/audio/{filename}"
session_logger.info(device_id, "audio", f"音频文件已保存: {filepath}")
# reduce_background_noise(filepath, relative_path,noise_path='assets/audio/noise_sample.wav',normalize_volume=True)
return filepath
except Exception as e:
session_logger.error(device_id, "audio", f"failed to store audio: {e}", exc_info=True)
raise
session_logger.error(device_id, "audio", f"保存音频文件时出错: {e}", exc_info=True)
raise

157
talkingq-url/handlers/mqtt_handler.py
View File

@@ -11,11 +11,16 @@ from banban.service.device_setting import device_setting_service
from banban.service.location import location_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 datetime import datetime
from services.device_target_cache import device_target_cache
from services.offline_audio_cache import offline_audio_cache
from services.task_manager import task_manager
from utils.logger import session_logger as logger
from services.task_manager import task_manager
class TalkingQMQTTService:
_instance = None
@@ -45,6 +50,10 @@ class TalkingQMQTTService:
"005": self._handle_nfc_listen_report,
"006": self._handle_bind_response,
"007": self._handle_open_response,
"008": self._handle_sleep_schedule_response,
"009": self._handle_remote_sleep_wake_response,
"010": self._handle_alarm_report,
"011": self._handle_short_press_message,
}
@classmethod
@@ -189,6 +198,8 @@ class TalkingQMQTTService:
)
elif status != "accepted":
logger.warning(device_id, "", f"[OTA] command failed: {payload}")
else:
logger.warning(device_id, "", f"[OTA] 设备 {device_id} 升级异常: {payload}")
async def _handle_nfc_notice_response(self, device_id: str, payload: dict):
status = payload.get("status")
@@ -202,19 +213,13 @@ class TalkingQMQTTService:
await self._send_nfc_listen_response(device_id, nfc_uuid)
async def _handle_bind_response(self, device_id: str, payload: dict):
params = payload.get("params", {})
data = payload.get("data", {})
status = payload.get("status") or params.get("status")
logger.info(device_id, "", f"[NFC bind] device={device_id} status={status}")
if status not in (None, "success", "accepted"):
logger.warning(device_id, "", f"[NFC bind] bind failed: {payload}")
return
nfc_uuid = data.get("uuid") or params.get("uuid")
if not nfc_uuid:
logger.warning(device_id, "", f"[NFC bind] missing uuid: {payload}")
return
params = payload
status = params.get("status")
logger.info(device_id, "", f"[NFC绑定] 设备 {device_id} 请求绑定卡片返回状态: {status}")
data = params.get("data", {})
nfc_uuid = data.get("uuid")
# 卡片与设备绑定
# await card_service.activate_card(device_id=device_id, card_uuid=nfc_uuid)
service = BindingService()
result = await service.finalize_nfc_bind(device_id=device_id, card_uuid=nfc_uuid)
if result is None:
@@ -224,8 +229,43 @@ class TalkingQMQTTService:
logger.info(device_id, "", f"[NFC bind] bind completed, uuid={nfc_uuid}, result={result}")
async def _handle_open_response(self, device_id: str, payload: dict):
params = payload
status = params.get("status")
logger.info(device_id, "", f"[设备状态] 设备 {device_id} 请求返回设备状态: {status}")
data = params.get("data", {})
device_status = data.get("status")
logger.info(device_id, "", f"[设备状态] 设备 {device_id} 请求设备状态: {device_status}")
async def _handle_sleep_schedule_response(self, device_id: str, payload: dict):
status = payload.get("status")
if status == "success":
logger.info(device_id, "", f"[定时休眠] 设备 {device_id} 休眠时间设置成功")
else:
logger.warning(device_id, "", f"[定时休眠] 设备 {device_id} 休眠时间设置失败: {payload}")
async def _handle_remote_sleep_wake_response(self, device_id: str, payload: dict):
status = payload.get("status")
if status == "success":
logger.info(device_id, "", f"[远程休眠唤醒] 设备 {device_id} 操作成功")
else:
logger.warning(device_id, "", f"[远程休眠唤醒] 设备 {device_id} 操作失败: {payload}")
async def _handle_alarm_report(self, device_id: str, payload: dict):
logger.info(device_id, "", f"[告警] 设备 {device_id} 发送紧急报警")
await self._publish(f"device/{device_id}/event_resp", {"msg_id": "010", "status": "success"})
async def _handle_short_press_message(self, device_id: str, payload: dict):
params = payload.get("params", {})
logger.info(device_id, "", f"[device open] response={params}")
nfc_uuid = params.get("uuid")
logger.info(device_id, "", f"[短按留言] 设备 {device_id} 短按发送留言, UUID={nfc_uuid}")
payload = {
"msg_id": "011",
"type": 0,
"params": {
"url": f"http://{settings.server_host}:{settings.server_port}/assets/audio/message_ok.mp3"
}
}
await self._publish(f"device/{device_id}/event_resp", payload)
async def _send_nfc_listen_response(self, device_id: str, nfc_uuid: str):
topic = f"device/{device_id}/event_resp"
@@ -244,7 +284,7 @@ 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 == "53D92B6DA20001":
if nfc_uuid == "53C22B6DA20001":
payload = {
"msg_id": "005",
"type": 0,
@@ -257,6 +297,17 @@ class TalkingQMQTTService:
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",
@@ -372,25 +423,69 @@ class TalkingQMQTTService:
return "003"
async def send_nfc_notice(self, device_id: str, url: str) -> str:
await self._publish(
f"device/{device_id}/command",
{"msg_id": "004", "params": {"url_1": url}},
)
topic = f"device/{device_id}/command"
payload = {
"msg_id": "004",
"params": {"url": url}
}
await self._publish(topic, payload)
return "004"
async def send_bind_nfc_command(self, device_id: str, uuid: Optional[str] = None) -> str:
del uuid
await self._publish(f"device/{device_id}/command", {"msg_id": "006"})
async def send_bind_nfc_command(self, device_id: str, uuid: str) -> str:
topic = f"device/{device_id}/command"
payload = {
"msg_id": "006",
"params": {
"url": f"http://{settings.server_host}:{settings.server_port}/assets/audio/bind_nfc_ready_zh.mp3"
}
}
await self._publish(topic, payload)
return "006"
async def send_open_command(self, device_id: str, open_type: int, is_open: int) -> str:
if open_type not in [0, 1]:
raise ValueError("open_type must be 0 or 1")
topic = f"device/{device_id}/command"
if open_type not in [0, 1, 2]:
raise ValueError("open_type must be 0, 1 or 2")
if open_type == 0:
payload = {"msg_id": "007", "type": open_type}
else:
payload = {"msg_id": "007", "type": open_type, "status": is_open}
await self._publish(f"device/{device_id}/command", payload)
payload = {
"msg_id": "007",
"type": open_type
}
elif open_type == 1:
payload = {
"msg_id": "007",
"type": open_type
}
elif open_type == 2:
payload = {
"msg_id": "007",
"type": open_type
}
await self._publish(topic, payload)
return "007"
async def send_sleep_schedule_command(self, device_id: str, start: str, end: str) -> str:
topic = f"device/{device_id}/command"
payload = {
"msg_id": "008",
"params": {
"start": start,
"end": end
}
}
await self._publish(topic, payload)
return "008"
async def send_remote_sleep_wake_command(self, device_id: str, switch: str) -> str:
if switch not in ["on", "off"]:
raise ValueError("switch must be 'on' or 'off'")
topic = f"device/{device_id}/command"
payload = {
"msg_id": "009",
"params": {
"switch": switch
}
}
await self._publish(topic, payload)
return "009"

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

@@ -18,6 +18,8 @@ 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
from handlers.audio_file_handler import message_audio_storage_service, save_audio_file
from fastapi import HTTPException
async def handle_websocket_messages(websocket: WebSocket, device_id: str, serial_number: str):
"""
@@ -266,13 +268,13 @@ async def process_cached_audio(device_id: str, target_device_id: str, serial_num
# 保存音频文件
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
# 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":
@@ -293,7 +295,7 @@ async def process_cached_audio(device_id: str, target_device_id: str, serial_num
await websocket.send_text(f"PROMPT_SOUND_URL:{success_audio_url}")
session_logger.info(device_id, "device", f"留言已收到音频URL给设备 {device_id}")
else:
session_logger.warning(device_id, "device", f"设备 {device_id} 不在线，发送留言已收到音频URL失败")
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到离线缓存")
@@ -305,7 +307,19 @@ async def process_cached_audio(device_id: str, target_device_id: str, serial_num
# else:
# session_logger.warning(device_id, "device", f"设备 {device_id} 不在线，发送留言成功音频失败")
except Exception as e:
session_logger.error(device_id, "target", f"处理缓存音频时出错: {e}", exc_info=True)
# 处理HTTPException异常
if isinstance(e, HTTPException):
session_logger.error(device_id, "target", f"保存音频文件时出错，返回HTTPException: {e}")
websocket = await connection_manager.get_connection(device_id)
if e.status_code == 404 or e.status_code == 400:
if websocket and websocket.client_state.name == "CONNECTED":
success_audio_url = f"http://{settings.server_host}:{settings.server_port}/assets/audio/save_audio_fail.mp3"
await websocket.send_text(f"PROMPT_SOUND_URL:{success_audio_url}")
session_logger.info(device_id, "device", f"留言已收到音频URL给设备 {device_id}")
else:
session_logger.warning(device_id, "device", f"设备 {device_id} 不在线，暂不发送留言已收到音频URL")
else:
session_logger.error(device_id, "target", f"处理缓存音频时出错: {e}", exc_info=True)
finally:
# 清除缓存
await target_audio_cache.clear_audio_data(target_device_id)