diff --git a/.DS_Store b/.DS_Store index a9b2e23..fcd63ea 100644 Binary files a/.DS_Store and b/.DS_Store differ diff --git a/banban.pem b/banban.pem new file mode 100644 index 0000000..b12e62c --- /dev/null +++ b/banban.pem @@ -0,0 +1,27 @@ +-----BEGIN RSA PRIVATE KEY----- +MIIEowIBAAKCAQEAskd8DRSgZIF9IqD/bEU8YMhWiLiyoP1Yl+MpFxxdyIJnIqXe +6PRAgaZVPiy1vzYbr+acVBp4ZueBsVR5Tmd+u9009fzMi/VZkc2FBXrpbkBkyBE9 +ftiYiK7BcKJEQ2YqiFEIpGg6Jn009dvTyK35GLyfNM34XKzkhOjNElgcRpCTbFh5 +QhHrm6Wt3uv6Txb6rnoO+mroox6Pe/jd4XEh6vcwNOq324dCuTEeazv3OUkFnRzl +7+WC55mzDU0ZevLkxQqGJYKnjUsmW4f2YmDGylRbbgVv4q6ycfg8VUyn2HiCJjJk +4Q02KiVRmiYlZBJug2P7ZIyhrTNSOK2lit/ACwIDAQABAoIBABim04D4NAY3H5IN +GB9PuD3GcKzJeVh937pqR1wc6IeZfIsTy5iDxHMJCIAnQjRqq3ZtxCUfCedqJgqw +HHFaf0bum2xTpLHKXzR5lOYWl7ZOl4D7XU5abfEh1Q7EPqiL3+AR8HN8vrm5a3wp +sFsaBLfxG8Sl9yyoHZXr87C7LUxrupOOB0gyqU2sdufA26yMEJXao/oes2DyxFLx +P9k4tRdcZDfUF5jftkyQNvZ/XIZ/yavQ26GyHtts8hm9buBl/m83K44MjlHEIFv6 +KpkxLMICzxCNxsu13XKqQOpesgpOH7UNmS/jd9aqQKdRNj+wJIy29Y55aqYPzuFQ +Lbp1Kl0CgYEAtbclUeP1WSJwXCFV7330Mzj1VxlOa61zC1imprSco+JGoSrRhpGo +xiDvI9ZAh5M0jmfkDzQjbtGwj2F5UujxXeyKeSJngiQlDhMMgr2SfdKkDxm0Tmx+ +WpZnXvUBINIg+8OsgUA1Z/TM57ud3F/B657PAVNq07wIMRUlvmjG4s0CgYEA+yi8 +ju/1r9t/uJcVkjWLxzFnJ8qC8swmF3djmEJpM7Z4dUUVaMMB7hQfBQJmxM0DRWkt +KMD47+gFQsOEn1w/GJxn/5GVQeyNUasvQt2sYe9HJIVCNbxSmIFhA0m3Sqy/z8fv +InbtEOy/K7zVsPG3I5kNbJsLxDrIXxmpKRnVHjcCgYB+uWKFYXxQ1PuWxIixpB0R +O7+dJkDSRvvcBc7yozI5+CtZagsE1b/lrEIZs+j8o2Qbi8g38hxjxEhlNYzujRUG +c1d+csfMsnhFAHPRGXN329Yd0cc0ieT0N7+PMT3ALcpiyWscGDMmdEoRsX29meoa +731dZ1cwogj0cdMInvlUYQKBgQDonFTuaS15nugOdNdEn5UCei3Yu4VGC28oAqna +BX/bph6wNbhbW2h5MGd+QzgdAucJrRxnBzpHLvNYXy6ATXYefBURrKq48LX9snbG +Dfouhea02zpz/CPfHMxVuDsqzQ2lCb3fhJeROkLf5jdfdq6wKHs3X+2o4uxar7Bs +4YDxRwKBgAmqv2YkmxG9zP9keB1AWYiIZxiH7KiNy6Jl9jrdcuLvlTr5UzodkcUX +8uIK77PWjiJ5SuXry4KYxb4QU1spc7xKUiKMhzzHx3IQOZco+isLrxxRldK5ZZu0 +H9WhfMh67rfyJ0kxtK6PXeBtAPcsoD3I2oblDdZg0RbXsS+VK8Rh +-----END RSA PRIVATE KEY----- \ No newline at end of file diff --git a/talkingq-url/api/__init__.py b/talkingq-url/api/__init__.py index fc295de..34f9fc4 100644 --- a/talkingq-url/api/__init__.py +++ b/talkingq-url/api/__init__.py @@ -4,6 +4,7 @@ from api.roles import router as roles_router from api.auth import router as auth_router from api.device_control import router as device_control_router from api.ota import router as ota_router # 新增OTA路由 +from api.mqtt_router import router as mqtt_router # 新增MQTT路由 api_router = APIRouter() api_router.include_router(websocket_router, tags=["WebSocket"]) @@ -11,4 +12,5 @@ api_router.include_router(roles_router, tags=["Roles"]) api_router.include_router(auth_router, tags=["Auth"]) api_router.include_router(device_control_router, tags=["Device Control"]) api_router.include_router(ota_router, tags=["OTA"]) # 注册OTA路由 +api_router.include_router(mqtt_router, tags=["MQTT"]) # 注册MQTT路由 diff --git a/talkingq-url/api/mqtt_models.py b/talkingq-url/api/mqtt_models.py new file mode 100644 index 0000000..10c5486 --- /dev/null +++ b/talkingq-url/api/mqtt_models.py @@ -0,0 +1,57 @@ +from typing import Optional +from pydantic import BaseModel, Field + + +class GPSQueryRequest(BaseModel): + device_id: str = Field(..., description="设备ID，如 TalkingQ_001") + + +class VolumeRequest(BaseModel): + device_id: str = Field(..., description="设备ID") + level: int = Field(..., ge=0, le=100, description="音量等级 0-100") + + +class OTARequest(BaseModel): + device_id: str = Field(..., description="设备ID") + url: str = Field(..., description="固件下载地址") + version: str = Field(..., description="目标版本号") + + +class NFCNoticeRequest(BaseModel): + device_id: str = Field(..., description="设备ID") + url: str = Field(..., description="提示音地址") + + +class NFCListenRequest(BaseModel): + device_id: str = Field(..., description="设备ID") + uuid: str = Field(..., description="NFC卡片UUID") + + +class NFCUnreadRequest(BaseModel): + device_id: str = Field(..., description="设备ID") + url: str = Field(..., description="未读留言音频地址") + + +class NFCBindRequest(BaseModel): + device_id: str = Field(..., description="设备ID") + uuid: str = Field(..., description="NFC卡片UUID") + + +class CommandResponse(BaseModel): + code: int = 0 + message: str = "success" + msg_id: str = Field(..., description="消息ID") + device_id: str = Field(..., description="设备ID") + + +class DeviceResponseData(BaseModel): + msg_id: str + status: Optional[str] = None + data: Optional[dict] = None + timestamp: Optional[float] = None + + +class DeviceStatusResponse(BaseModel): + device_id: str + online: bool + last_seen: Optional[float] = None diff --git a/talkingq-url/api/mqtt_router.py b/talkingq-url/api/mqtt_router.py new file mode 100644 index 0000000..d6d9a8c --- /dev/null +++ b/talkingq-url/api/mqtt_router.py @@ -0,0 +1,126 @@ +import logging +from typing import Optional +from fastapi import APIRouter, HTTPException, Query +from services.offline_audio_cache import offline_audio_cache +from api.mqtt_models import ( + GPSQueryRequest, + VolumeRequest, + OTARequest, + NFCNoticeRequest, + NFCListenRequest, + NFCUnreadRequest, + NFCBindRequest, + CommandResponse, + DeviceResponseData, + DeviceStatusResponse, +) +from handlers.mqtt_handler import TalkingQMQTTService + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/api/mqtt", tags=["MQTT 服务"]) + + +async def _get_service() -> TalkingQMQTTService: + service = await TalkingQMQTTService.get_instance() + if service is None: + raise HTTPException(status_code=503, detail="MQTT 服务未初始化") + return service + + +@router.post("/gps", response_model=CommandResponse, summary="GPS位置查询") +async def query_gps(req: GPSQueryRequest): + service = await _get_service() + msg_id = await service.send_gps_query(req.device_id) + return CommandResponse(msg_id=msg_id, device_id=req.device_id) + + +@router.post("/volume", response_model=CommandResponse, summary="音量调节") +async def set_volume(req: VolumeRequest): + service = await _get_service() + msg_id = await service.send_volume_command(req.device_id, req.level) + return CommandResponse(msg_id=msg_id, device_id=req.device_id) + + +@router.post("/ota", response_model=CommandResponse, summary="OTA升级") +async def start_ota(req: OTARequest): + service = await _get_service() + msg_id = await service.send_ota_command(req.device_id, req.url, req.version) + return CommandResponse(msg_id=msg_id, device_id=req.device_id) + + +@router.post("/nfc/notice", response_model=CommandResponse, summary="NFC留言下发提示") +async def send_nfc_notice(req: NFCNoticeRequest): + service = await _get_service() + msg_id = await service.send_nfc_notice(req.device_id, req.url) + return CommandResponse(msg_id=msg_id, device_id=req.device_id) + + +@router.post("/nfc/listen", response_model=CommandResponse, summary="NFC收听留言") +async def nfc_listen(req: NFCListenRequest): + service = await _get_service() + topic = f"device/{req.device_id}/event" + payload = { + "msg_id": "005", + "params": {"uuid": req.uuid} + } + await service._publish(topic, payload) + return CommandResponse(msg_id="005", device_id=req.device_id) + + +@router.post("/nfc/unread", response_model=CommandResponse, summary="设置NFC未读留言") +async def set_nfc_unread(req: NFCUnreadRequest): + await offline_audio_cache.add_audio_url(req.device_id, req.url) + return CommandResponse(msg_id="", device_id=req.device_id, message="已设置未读留言") + + +@router.delete("/nfc/unread/{device_id}", response_model=CommandResponse, summary="清除NFC未读留言定时推送") +async def clear_nfc_unread(device_id: str): + await offline_audio_cache.clear_audio_urls(device_id) + return CommandResponse(msg_id="", device_id=device_id, message="已清除未读留言") + + +@router.post("/nfc/bind", response_model=CommandResponse, summary="NFC绑定卡片") +async def bind_nfc(req: NFCBindRequest): + service = await _get_service() + msg_id = await service.send_bind_nfc_command(req.device_id, req.uuid) + return CommandResponse(msg_id=msg_id, device_id=req.device_id) + + +# @router.get("/nfc/unread", summary="查询所有未读留言设备列表") +# async def get_nfc_unread_devices(): +# service = await _get_service() +# devices = service.get_nfc_unread_devices() +# return {"count": len(devices), "devices": devices} + + +# @router.get("/device/{device_id}/status", response_model=DeviceStatusResponse, summary="查询设备在线状态") +# async def get_device_status(device_id: str): +# service = await _get_service() +# online = service.is_device_online(device_id) +# last_seen = service.get_device_last_seen(device_id) +# return DeviceStatusResponse(device_id=device_id, online=online, last_seen=last_seen) + + +# @router.get("/device/{device_id}/responses", summary="查询设备上报记录") +# async def get_device_responses( +# device_id: str, +# msg_id: Optional[str] = Query(None, description="按msg_id过滤"), +# limit: int = Query(20, ge=1, le=100, description="返回条数"), +# ): +# service = await _get_service() +# responses = service.get_device_responses(device_id, msg_id=msg_id, limit=limit) +# return {"device_id": device_id, "count": len(responses), "responses": responses} + + +# @router.get("/device/{device_id}/wait_response", summary="等待设备响应(同步阻塞)") +# async def wait_device_response( +# device_id: str, +# msg_id: str = Query(..., description="等待的消息ID"), +# timeout: float = Query(10.0, ge=1.0, le=60.0, description="超时时间(秒)"), +# ): +# service = await _get_service() +# response = service.wait_for_response(device_id, msg_id, timeout=timeout) +# if response is None: +# raise HTTPException(status_code=408, detail=f"等待设备 {device_id} 的 msg_id={msg_id} 响应超时") +# return response diff --git a/talkingq-url/assets/audio/error_card.mp3 b/talkingq-url/assets/audio/error_card.mp3 new file mode 100644 index 0000000..c3193d6 Binary files /dev/null and b/talkingq-url/assets/audio/error_card.mp3 differ diff --git a/talkingq-url/assets/audio/message_ok_zh.mp3 b/talkingq-url/assets/audio/message_ok_zh.mp3 new file mode 100644 index 0000000..712a3e8 Binary files /dev/null and b/talkingq-url/assets/audio/message_ok_zh.mp3 differ diff --git a/talkingq-url/assets/audio/new_message.mp3 b/talkingq-url/assets/audio/new_message.mp3 new file mode 100644 index 0000000..87383b9 Binary files /dev/null and b/talkingq-url/assets/audio/new_message.mp3 differ diff --git a/talkingq-url/assets/audio/no_message.mp3 b/talkingq-url/assets/audio/no_message.mp3 new file mode 100644 index 0000000..83bedc3 Binary files /dev/null and b/talkingq-url/assets/audio/no_message.mp3 differ diff --git a/talkingq-url/assets/audio/success_zh.mp3 b/talkingq-url/assets/audio/success_zh.mp3 new file mode 100644 index 0000000..eb07858 Binary files /dev/null and b/talkingq-url/assets/audio/success_zh.mp3 differ diff --git a/talkingq-url/assets/audio/test_zh.mp3 b/talkingq-url/assets/audio/test_zh.mp3 new file mode 100644 index 0000000..fdf084b Binary files /dev/null and b/talkingq-url/assets/audio/test_zh.mp3 differ diff --git a/talkingq-url/assets/audio/tts_error_zh.mp3 b/talkingq-url/assets/audio/tts_error_zh.mp3 new file mode 100644 index 0000000..67e4907 Binary files /dev/null and b/talkingq-url/assets/audio/tts_error_zh.mp3 differ diff --git a/talkingq-url/assets/firmware/EG800AKCN_91LC-beta1-20260418225651-fota.bin b/talkingq-url/assets/firmware/EG800AKCN_91LC-beta1-20260418225651-fota.bin new file mode 100644 index 0000000..96486ed Binary files /dev/null and b/talkingq-url/assets/firmware/EG800AKCN_91LC-beta1-20260418225651-fota.bin differ diff --git a/talkingq-url/config.py b/talkingq-url/config.py index 0a99485..7e9b0ef 100644 --- a/talkingq-url/config.py +++ b/talkingq-url/config.py @@ -36,10 +36,10 @@ class Settings(BaseSettings): db_password: str db_name: str = "talkingq" db_echo: bool = False # 是否打印SQL语句 - + # TalkingQ设备MQTT命令服务配置 talkingq_mqtt_broker: str = "broker.emqx.io" - talkingq_mqtt_port: int = 1883 + talkingq_mqtt_port: int = 1884 talkingq_mqtt_username: str = "" talkingq_mqtt_password: str = "" talkingq_mqtt_device_prefix: str = "TalkingQ" diff --git a/talkingq-url/handlers/audio_file_handler.py b/talkingq-url/handlers/audio_file_handler.py index cce0e60..efd34ab 100644 --- a/talkingq-url/handlers/audio_file_handler.py +++ b/talkingq-url/handlers/audio_file_handler.py @@ -2,7 +2,7 @@ import os import uuid from config import settings from utils.logger import session_logger -from utils.audio_denoiser import reduce_background_noise +# from utils.audio_denoiser import reduce_background_noise async def save_audio_file(audio_data: bytes, device_id: str) -> str: diff --git a/talkingq-url/handlers/mqtt_handler.py b/talkingq-url/handlers/mqtt_handler.py new file mode 100644 index 0000000..ce895fc --- /dev/null +++ b/talkingq-url/handlers/mqtt_handler.py @@ -0,0 +1,378 @@ +import json +import time +import asyncio +from typing import Optional, Dict, Callable, Awaitable +from config import settings +from services.card_service import card_service +from services.offline_audio_cache import offline_audio_cache +import aiomqtt + +from services.device_target_cache import device_target_cache +from utils.logger import session_logger as logger + + +class TalkingQMQTTService: + _instance = None + _lock = asyncio.Lock() + + def __init__(self, config: dict): + self.broker = config.get("broker", "broker.emqx.io") + self.port = config.get("port", 1883) + self.username = config.get("username") + self.password = config.get("password") + self.device_prefix = config.get("device_prefix", "TalkingQ") + self.qos = config.get("qos", 0) + self.keepalive = config.get("keepalive", 60) + self.nfc_notice_interval = config.get("nfc_notice_interval", 600) + + self._client: Optional[aiomqtt.Client] = None + self._connected = False + self._connect_lock = asyncio.Lock() + self._message_task: Optional[asyncio.Task] = None + + self._msg_handlers: Dict[str, Callable[[str, dict], Awaitable[None]]] = { + "000": self._handle_device_info, + "001": self._handle_gps_response, + "002": self._handle_volume_response, + "003": self._handle_ota_response, + "004": self._handle_nfc_notice_response, + "005": self._handle_nfc_listen_report, + "006": self._handle_bind_response, + "007": self._handle_open_response, + } + + @classmethod + async def get_instance(cls, config: dict = None) -> Optional["TalkingQMQTTService"]: + async with cls._lock: + if cls._instance is None and config is not None: + cls._instance = cls(config) + return cls._instance + + @classmethod + async def reset_instance(cls): + async with cls._lock: + if cls._instance is not None: + await cls._instance.disconnect() + cls._instance = None + + async def _message_loop(self): + try: + async for message in self._client.messages: + try: + topic = str(message.topic) + parts = topic.split("/") + if parts[0] == "device": + device_id = parts[1] if len(parts) >= 2 else "unknown" + + if not device_id.startswith(f"{self.device_prefix}_"): + continue + + payload = json.loads(message.payload.decode("utf-8")) + + msg_id = payload.get("msg_id") + + handler = self._msg_handlers.get(msg_id) + if handler: + await handler(device_id, payload) + else: + logger.warning(device_id, "", f"未知 msg_id: {msg_id}, 设备: {device_id}") + + except json.JSONDecodeError as e: + logger.error("", "", f"消息解析失败: {e}") + except Exception as e: + logger.error("", "", f"消息处理异常: {e}") + except asyncio.CancelledError: + pass + except Exception as e: + logger.error("", "", f"消息循环异常: {e}") + self._connected = False + + async def _handle_device_info(self, device_id: str, payload: dict): + # status = payload.get("status") + data = payload.get("data", {}) + # if status == "success": + d_id = data.get("id") + power = data.get("power") + signal = data.get("signal") + version = data.get("version") + voice = data.get("voice") + logger.info(device_id, "", f"[设备信息] 设备 {d_id} 信息: 电量={power}, 信号强度={signal}, 版本号={version}, 音量={voice}") + # else: + # logger.warning(device_id, "", f"[设备信息] 设备 {device_id} 查询失败: {payload}") + await self._publish(f"device/{device_id}/event_resp", {"msg_id": "000", "status": "success"}) + + async def _handle_gps_response(self, device_id: str, payload: dict): + status = payload.get("status") + data = payload.get("data", {}) + if status == "success": + lat = data.get("latitude") + lon = data.get("longitude") + logger.info(device_id, "", f"[GPS] 设备 {device_id} 位置: 纬度={lat}, 经度={lon}") + else: + logger.warning(device_id, "", f"[GPS] 设备 {device_id} 查询失败: {payload}") + + async def _handle_volume_response(self, device_id: str, payload: dict): + status = payload.get("status") + data = payload.get("data", {}) + if status == "success": + level = data.get("current_level") + logger.info(device_id, "", f"[音量] 设备 {device_id} 当前音量: {level}") + else: + logger.warning(device_id, "", f"[音量] 设备 {device_id} 调节失败: {payload}") + + async def _handle_ota_response(self, device_id: str, payload: dict): + status = payload.get("status") + data = payload.get("data", {}) + if status == "accepted": + current = data.get("current_version") + target = data.get("target_version") + logger.info(device_id, "", f"[OTA] 设备 {device_id} 已接受升级: {current} -> {target}") + elif status == "success": + new_ver = data.get("new_version") + logger.info(device_id, "", f"[OTA] 设备 {device_id} 升级完成: {new_ver}") + 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") + if status == "success": + logger.info(device_id, "", f"[NFC通知] 设备 {device_id} 已收到留言提示") + else: + logger.warning(device_id, "", f"[NFC通知] 设备 {device_id} 留言提示失败: {payload}") + + async def _handle_nfc_listen_report(self, device_id: str, payload: dict): + params = payload.get("params", {}) + nfc_uuid = params.get("uuid") + logger.info(device_id, "", f"[NFC收听] 设备 {device_id} 请求收听留言, UUID={nfc_uuid}") + 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", {}) + status = params.get("status") + logger.info(device_id, "", f"[NFC绑定] 设备 {device_id} 请求绑定卡片返回状态: {status}") + + nfc_uuid = params.get("uuid") + logger.info(device_id, "", f"[NFC绑定] 设备 {device_id} 请求绑定卡片, UUID={nfc_uuid}") + + async def _handle_open_response(self, device_id: str, payload: dict): + params = payload.get("params", {}) + 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 _send_nfc_listen_response(self, device_id: str, nfc_uuid: str): + topic = f"device/{device_id}/event_resp" + card = await card_service.get_card_by_uuid(nfc_uuid) + if not card: + logger.warning("", "", f"[NFC收听] 设备 {device_id} 没有找到卡片{nfc_uuid}，无法发送留言提示") + payload = { + "msg_id": "005", + "type": 0, + "params": { + "url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/error_card.mp3" + } + } + await self._publish(topic, payload) + return + + 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": + 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 == "53D92B6DA20001": + 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 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 = {} + for k, audio_url in enumerate(audio_urls, start=1): + params[f"url_{k}"] = audio_url + payload = { + "msg_id": "005", + "type": 0, + "params": params + } + await offline_audio_cache.clear_audio_urls(device_id) + await self._publish(topic, payload) + else: + payload = { + "msg_id": "005", + "type": 0, + "params": { + "url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/no_message.mp3" + } + } + await self._publish(topic, payload) + else: + # 检查卡片是否存在 + existing_card = await card_service.get_card_by_uuid(nfc_uuid) + + if existing_card: + # 卡片已存在，使用卡片绑定的设备ID作为目标设备ID + target_device_id = existing_card.device_id + logger.info(device_id, "card", f"卡片已存在，绑定的设备ID: {target_device_id}") + else: + # 卡片不存在，创建新卡片并绑定到当前设备 + new_card = await card_service.activate_card(nfc_uuid, device_id) + logger.info(device_id, "card", f"新卡片{nfc_uuid}已创建并激活，绑定到设备: {device_id}") + return + + # 设置目标设备 + await device_target_cache.set_target(device_id, target_device_id) + payload = { + "msg_id": "005", + "type": 1, + "params": { + "url_1": f"http://{settings.server_host}:{settings.server_port}/assets/audio/welcome.mp3" + } + } + await self._publish(topic, payload) + + async def connect(self): + async with self._connect_lock: + if self._connected: + return + try: + self._client = aiomqtt.Client( + hostname=self.broker, + port=self.port, + username=self.username, + password=self.password, + identifier=f"talkingq_server_{int(time.time())}", + keepalive=self.keepalive, + ) + await self._client.__aenter__() + self._connected = True + logger.system_info("", f"TalkingQ MQTT 已连接: {self.broker}:{self.port}") + + await self._client.subscribe("device/+/response", qos=self.qos) + logger.system_info("", "已订阅: device/+/response") + + await self._client.subscribe("device/+/event", qos=self.qos) + logger.system_info("", "已订阅: device/+/event") + + self._message_task = asyncio.create_task(self._message_loop()) + logger.info("", "", "TalkingQ MQTT 服务启动中...") + except Exception as e: + self._connected = False + logger.error("", "", f"TalkingQ MQTT 连接失败: {e}") + + async def disconnect(self): + if self._message_task and not self._message_task.done(): + self._message_task.cancel() + try: + await self._message_task + except asyncio.CancelledError: + pass + self._message_task = None + + if self._client is not None: + try: + await self._client.__aexit__(None, None, None) + except Exception: + pass + self._client = None + self._connected = False + + async def _ensure_connected(self): + if not self._connected: + await self.connect() + + async def _publish(self, topic: str, payload: dict): + await self._ensure_connected() + try: + await self._client.publish(topic, json.dumps(payload, ensure_ascii=False), qos=self.qos) + logger.info("", "", f"下发命令 -> {topic}: {payload}") + except Exception as e: + logger.error("", "", f"命令下发失败: {e}") + + async def send_gps_query(self, device_id: str) -> str: + topic = f"device/{device_id}/command" + payload = {"msg_id": "001"} + await self._publish(topic, payload) + return "001" + + async def send_volume_command(self, device_id: str, level: int) -> str: + topic = f"device/{device_id}/command" + payload = { + "msg_id": "002", + "params": {"level": level} + } + await self._publish(topic, payload) + return "002" + + async def send_ota_command(self, device_id: str, url: str, version: str) -> str: + topic = f"device/{device_id}/command" + payload = { + "msg_id": "003", + "params": { + "url": url, + "version": version, + } + } + await self._publish(topic, payload) + return "003" + + async def send_nfc_notice(self, device_id: str, url: str) -> str: + topic = f"device/{device_id}/command" + payload = { + "msg_id": "004", + "params": {"url_1": url} + } + await self._publish(topic, payload) + return "004" + + + async def send_bind_nfc_command(self, device_id: str, uuid: str) -> str: + topic = f"device/{device_id}/command" + payload = { + "msg_id": "006" + } + await self._publish(topic, payload) + return "006" + + async def send_open_command(self, device_id: str, open_type: int, is_open: int) -> str: + topic = f"device/{device_id}/command" + if open_type not in [0, 1]: + raise ValueError("open_type must be 0 or 1") + if open_type == 0: + payload = { + "msg_id": "007", + "type": open_type + } + elif open_type == 1: + payload = { + "msg_id": "007", + "type": open_type, + "status": is_open + } + await self._publish(topic, payload) + return "007" diff --git a/talkingq-url/handlers/websocket_handler.py b/talkingq-url/handlers/websocket_handler.py index aa624b9..1dfc89f 100644 --- a/talkingq-url/handlers/websocket_handler.py +++ b/talkingq-url/handlers/websocket_handler.py @@ -30,19 +30,19 @@ async def websocket_endpoint(websocket: WebSocket): await websocket.send_text('{"status": "error", "message": "Not authenticated"}') return - # 检查设备是否有离线音频URL需要发送 - has_pending = await offline_audio_cache.has_pending_audio(device_id) - if has_pending: - audio_urls = await offline_audio_cache.get_audio_urls(device_id) - for audio_url in audio_urls: - try: - await websocket.send_text(f"SOUND_URL:{audio_url}") - session_logger.info(device_id, "offline", f"发送离线音频URL: {audio_url}") - except Exception as e: - session_logger.error(device_id, "offline", f"发送离线音频URL失败: {e}") - # 清空已发送的离线音频URL - await offline_audio_cache.clear_audio_urls(device_id) - session_logger.info(device_id, "offline", f"已清空设备的离线音频URL缓存，共 {len(audio_urls)} 个") + # # 检查设备是否有离线音频URL需要发送 + # has_pending = await offline_audio_cache.has_pending_audio(device_id) + # if has_pending: + # audio_urls = await offline_audio_cache.get_audio_urls(device_id) + # for audio_url in audio_urls: + # try: + # await websocket.send_text(f"SOUND_URL:{audio_url}") + # session_logger.info(device_id, "offline", f"发送离线音频URL: {audio_url}") + # except Exception as e: + # session_logger.error(device_id, "offline", f"发送离线音频URL失败: {e}") + # # 清空已发送的离线音频URL + # 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) diff --git a/talkingq-url/handlers/websocket_message_handler.py b/talkingq-url/handlers/websocket_message_handler.py index 25ce8c0..7e474e3 100644 --- a/talkingq-url/handlers/websocket_message_handler.py +++ b/talkingq-url/handlers/websocket_message_handler.py @@ -267,18 +267,36 @@ async def process_cached_audio(device_id: str, target_device_id: str): audio_url = f"http://{settings.server_host}:{settings.server_port}/{audio_path}" # 发送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 给客户端") + # 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: + + # # 目标设备不在线，保存到离线缓存 + 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/message_ok.mp3" + await websocket.send_text(f"PROMPT_SOUND_URL:{success_audio_url}") + session_logger.info(device_id, "device", f"留言已收到音频URL给设备 {device_id}") else: - # 目标设备不在线，保存到离线缓存 - 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.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: session_logger.error(device_id, "target", f"处理缓存音频时出错: {e}", exc_info=True) finally: diff --git a/talkingq-url/main.py b/talkingq-url/main.py index e93635c..3ab668b 100644 --- a/talkingq-url/main.py +++ b/talkingq-url/main.py @@ -2,6 +2,7 @@ import asyncio import os from contextlib import asynccontextmanager from fastapi import FastAPI +from services import offline_audio_cache from services.device_config import DeviceConfigManager from services.schedule_session_cleanup import background_cleanup_task from services.role_manager import role_manager @@ -14,6 +15,8 @@ from api.assets import configure_static_assets from database.init_db import init_db from database.connection import get_db_manager from services.firmware_scanner import firmware_scanner +from handlers.mqtt_handler import TalkingQMQTTService +from services.scheduler import TaskScheduler device_config_manager = DeviceConfigManager() @@ -29,6 +32,28 @@ async def lifespan(app: FastAPI): ) init_directories() + scheduler = TaskScheduler.get_instance() + scheduler.start() + scheduler.add_interval_task( + interval_seconds=settings.talkingq_mqtt_nfc_notice_interval + ) + if settings.talkingq_mqtt_broker: + mqtt_config = { + "broker": settings.talkingq_mqtt_broker, + "port": settings.talkingq_mqtt_port, + "username": settings.talkingq_mqtt_username, + "password": settings.talkingq_mqtt_password, + "device_prefix": settings.talkingq_mqtt_device_prefix, + "qos": settings.talkingq_mqtt_qos, + "keepalive": settings.talkingq_mqtt_keepalive, + "nfc_notice_interval": settings.talkingq_mqtt_nfc_notice_interval, + } + service = await TalkingQMQTTService.get_instance(mqtt_config) + await service.connect() + session_logger.system_info("startup", "TalkingQ MQTT 设备命令服务链接已启动") + else: + session_logger.system_info("startup", "未配置 TalkingQ_MQTT，MQTT 服务链接未启动") + exit(1) try: session_logger.system_info("startup", "初始化数据库...") await init_db() @@ -93,6 +118,10 @@ async def lifespan(app: FastAPI): if is_main_process: session_logger.system_info("shutdown", f"关闭数据库连接时出错: {str(e)}") + TaskScheduler.reset_instance() + await TalkingQMQTTService.reset_instance() + session_logger.system_info("shutdown", "TalkingQ MQTT 服务链接已关闭") + if is_main_process: session_logger.system_info("shutdown", "应用关闭") diff --git a/talkingq-url/requirements.txt b/talkingq-url/requirements.txt index 42a6f64..76552cf 100644 --- a/talkingq-url/requirements.txt +++ b/talkingq-url/requirements.txt @@ -1,7 +1,7 @@ aiofiles==24.1.0 -aiohttp==3.11.13 +aiohttp fastapi==0.115.11 -pycld2==0.41 +# pycld2==0.41 pydantic-settings==2.8.1 python-multipart==0.0.20 PyYAML==6.0.2 @@ -17,5 +17,6 @@ numpy==1.26.4 scipy==1.11.4 librosa==0.10.1 setuptools==69.5.1 - -paho-mqtt>=2.0.0 \ No newline at end of file +pycld2 +aiomqtt>=2.0.0 +apscheduler>=3.10.0 \ No newline at end of file diff --git a/talkingq-url/services/offline_audio_cache.py b/talkingq-url/services/offline_audio_cache.py index 4654b78..c0d8dc1 100644 --- a/talkingq-url/services/offline_audio_cache.py +++ b/talkingq-url/services/offline_audio_cache.py @@ -25,7 +25,10 @@ class OfflineAudioCache: async def has_pending_audio(self, device_id: str) -> bool: async with self.lock: return device_id in self.offline_audio and len(self.offline_audio[device_id]) > 0 - + + async def get_all_audio_cache(self) -> List[str]: + async with self.lock: + return self.offline_audio # 单例实例 offline_audio_cache = OfflineAudioCache() \ No newline at end of file diff --git a/talkingq-url/services/scheduler.py b/talkingq-url/services/scheduler.py new file mode 100644 index 0000000..834a268 --- /dev/null +++ b/talkingq-url/services/scheduler.py @@ -0,0 +1,69 @@ +import logging +from typing import Optional, Dict + +from apscheduler.schedulers.asyncio import AsyncIOScheduler +from apscheduler.triggers.interval import IntervalTrigger +from services.connection_manager import connection_manager +from handlers.mqtt_handler import TalkingQMQTTService +from services.offline_audio_cache import offline_audio_cache +from config import settings +from utils.logger import session_logger as logger + +class TaskScheduler: + _instance = None + + def __init__(self): + self._scheduler = AsyncIOScheduler(timezone="Asia/Shanghai") + + @classmethod + def get_instance(cls) -> "TaskScheduler": + if cls._instance is None: + cls._instance = cls() + return cls._instance + + @classmethod + def reset_instance(cls): + if cls._instance is not None: + cls._instance.shutdown() + cls._instance = None + + def start(self): + if not self._scheduler.running: + self._scheduler.start() + logger.system_info("", "定时任务调度器已启动") + + def shutdown(self): + if self._scheduler.running: + self._scheduler.shutdown(wait=False) + logger.system_info("", "定时任务调度器已关闭") + + async def _execute_task(self): + try: + service = await TalkingQMQTTService.get_instance() + audio_cache = await offline_audio_cache.get_all_audio_cache() + if not audio_cache: + logger.warning("", "", "无离线音频缓存，跳过本次执行") + return + for device_id, audio_urls in audio_cache.items(): + websocket = await connection_manager.get_connection(device_id) + if websocket is not None: + audio_url = f"http://{settings.server_host}:{settings.server_port}/assets/audio/new_message.mp3" + await service.send_nfc_notice(device_id, audio_url) + logger.info(device_id, "", f"[定时任务] 发送音频成功: device={device_id}, audio_url={audio_url}") + else: + logger.info(device_id, "", f"[定时任务] 跳过音频发送: 设备 {device_id} 离线") + except Exception as e: + logger.error("", "", f"[定时任务] 执行失败: {e}") + + def add_interval_task(self, interval_seconds: int = 600): + task_id = "Voice_Message_Task" + task_name = "Voice_Message_Task" + job = self._scheduler.add_job( + self._execute_task, + trigger=IntervalTrigger(seconds=interval_seconds), + id=task_id, + name=task_name, + replace_existing=True, + ) + + logger.system_info("", f"[定时任务] 已添加: {task_id} | 间隔={interval_seconds}s | 下次执行={job.next_run_time}")