From: luozheng <1732036298@qq.com> Date: Wed, 27 Aug 2025 04:01:29 +0000 (+0800) Subject: 雷鸟api协议Python验证 X-Git-Tag: leiniao_debug_1.0.0_20250922_020716~135^2~11 X-Git-Url: http://git.snicker.com.cn/css/static/gitweb.css?a=commitdiff_plain;h=3c037a5bf1e67002dce3b854089adb42720842d5;p=mirrors%2FFD230.git 雷鸟api协议Python验证 --- diff --git a/code/sdk/zhihui/leiniao/leiniao_python_test/device_info_apis_4.py b/code/sdk/zhihui/leiniao/leiniao_python_test/device_info_apis_4.py new file mode 100644 index 000000000..3a0cb0937 --- /dev/null +++ b/code/sdk/zhihui/leiniao/leiniao_python_test/device_info_apis_4.py @@ -0,0 +1,187 @@ +import requests +import time +import hashlib +import hmac +import random +import string +import json +from typing import Dict, Optional + +# -------------------------- 基础配置 -------------------------- +BASE_URL = "https://ambybox-api.test.leiniao.com/ambybox-device-api" +# 接口路径 +TOKEN_INTERFACE = "/v1/auth/getDeviceToken" +GET_CHARACTER_INTERFACE = "/v1/device/getCharacter" +GET_USER_ID_INTERFACE = "/v1/device/getUserId" + +# 完整URL +TOKEN_URL = BASE_URL + TOKEN_INTERFACE +GET_CHARACTER_URL = BASE_URL + GET_CHARACTER_INTERFACE +GET_USER_ID_URL = BASE_URL + GET_USER_ID_INTERFACE + +# 测试签名密钥 +SIGN_SECRET = "vDdXVXgPUEUM6_piUjz5AEhMvYTSFzQdPaN-ogKnSus=" + +# 设备信息 +DEVICE_ID = "DEVICE_PC_TEST_001" +DEVICE_ICCID = "86123456789012345678" + +# -------------------------- 工具函数 -------------------------- +def get_device_token() -> Optional[str]: + """获取设备Token""" + try: + timestamp = str(int(time.time() * 1000)) + nonce = ''.join(random.choices(string.ascii_letters + string.digits, k=12)) + + # 生成签名 + sign_data = f"{DEVICE_ID}:{DEVICE_ICCID}:{timestamp}:{nonce}" + hmac_obj = hmac.new(SIGN_SECRET.encode("utf-8"), sign_data.encode("utf-8"), hashlib.sha256) + signature = hmac_obj.hexdigest().lower() + + headers = { + "deviceId": DEVICE_ID, + "deviceIccid": DEVICE_ICCID, + "timestamp": timestamp, + "nonce": nonce, + "signature": signature, + "deviceModel": "abmybox_1", + "deviceVersion": "1.0.0", + "Content-Type": "application/json" + } + + response = requests.get(TOKEN_URL, headers=headers, timeout=10) + response_data = response.json() + + if response_data.get("errorCode") is None and "data" in response_data: + print("✅ 成功获取设备Token") + return response_data["data"]["token"] + else: + print(f"❌ 获取Token失败：{response_data.get('message', '未知错误')}") + return None + + except Exception as e: + print(f"❌ 获取Token异常：{str(e)}") + return None + +def create_auth_headers(token: str) -> Dict[str, str]: + """创建通用认证请求头""" + return { + "deviceId": DEVICE_ID, + "deviceToken": token, + "Content-Type": "application/json" + } + +# -------------------------- 4.1 获取设备角色信息接口 -------------------------- +def test_get_character(token: str): + """测试获取设备角色信息接口""" + if not token: + print("❌ Token为空，无法调用接口") + return + + try: + headers = create_auth_headers(token) + + print("📤 发送获取设备角色信息请求...") + response = requests.get(GET_CHARACTER_URL, headers=headers, timeout=10) + response_data = response.json() + + print("=" * 50) + print("【响应结果】") + print(json.dumps(response_data, ensure_ascii=False, indent=2)) + + error_code = response_data.get("errorCode") + if error_code is None: + data = response_data.get("data") + if data and isinstance(data, dict): + print("\n✅ 获取设备角色信息成功") + print(f"角色ID: {data.get('characterId')}") + print(f"角色名称: {data.get('characterName')}") + + # 打印角色形象信息 + avatar = data.get('characterAvatar', {}) + print("\n角色形象配置:") + print(f" 发色: {avatar.get('hairColor')}") + print(f" 瞳色: {avatar.get('eyesColor')}") + print(f" 发型: {avatar.get('hairStyle')}") + print(f" 屏保图片: {avatar.get('screenImage')}") + print(f" 屏保图片是否作为对话表情: {avatar.get('screenImageAsEmoji')}") + + # 打印表情信息 + emojis = data.get('emotionEmojis', []) + if emojis: + print("\n角色表情:") + for emoji in emojis: + print(f" {emoji.get('emotion')}: {emoji.get('emojiUrl')}") + else: + print("\n⚠️ 获取设备角色信息成功，但返回数据为空") + else: + if error_code == "AUTH_FAILURE": + print(f"\n❌ 认证失败：{response_data.get('message', '设备Token无效')}") + else: + print(f"\n❌ 请求失败：{response_data.get('message', '未知错误')} (错误码: {error_code})") + + except Exception as e: + print(f"❌ 获取设备角色信息接口异常：{str(e)}") + +# -------------------------- 4.2 获取设备绑定用户ID接口 -------------------------- +def test_get_user_id(token: str): + """测试获取设备绑定用户ID接口""" + if not token: + print("❌ Token为空，无法调用接口") + return + + try: + headers = create_auth_headers(token) + + print("📤 发送获取设备绑定用户ID请求...") + response = requests.get(GET_USER_ID_URL, headers=headers, timeout=10) + response_data = response.json() + + print("=" * 50) + print("【响应结果】") + print(json.dumps(response_data, ensure_ascii=False, indent=2)) + + error_code = response_data.get("errorCode") + if error_code is None: + user_id = response_data.get("data") + if user_id: + print(f"\n✅ 设备已绑定用户，用户ID: {user_id}") + else: + print("\n✅ 设备未绑定用户") + else: + if error_code == "AUTH_FAILURE": + print(f"\n❌ 认证失败：{response_data.get('message', '设备Token无效')}") + else: + print(f"\n❌ 请求失败：{response_data.get('message', '未知错误')} (错误码: {error_code})") + + except Exception as e: + print(f"❌ 获取设备绑定用户ID接口异常：{str(e)}") + +# -------------------------- 主流程 -------------------------- +def main(): + # 1. 获取Token + token = get_device_token() + if not token: + return + + print("\n" + "=" * 60) + print("请选择要测试的设备信息接口：") + print("1 - 4.1 获取设备角色信息接口") + print("2 - 4.2 获取设备绑定用户ID接口") + print("0 - 退出") + print("=" * 60) + + choice = input("请输入选项（0-2）：") + + if choice == "1": + test_get_character(token) + elif choice == "2": + test_get_user_id(token) + elif choice == "0": + print("退出程序") + return + else: + print("无效选项") + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/code/sdk/zhihui/leiniao/leiniao_python_test/device_token.txt b/code/sdk/zhihui/leiniao/leiniao_python_test/device_token.txt new file mode 100644 index 000000000..ff35c4aae --- /dev/null +++ b/code/sdk/zhihui/leiniao/leiniao_python_test/device_token.txt @@ -0,0 +1 @@ +eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJXWVNaIiwiZGV2aWNlTW9kZWwiOiJhYm15Ym94XzEiLCJkZXZpY2VWZXJzaW9uIjoiMi4wLjAiLCJkZXZpY2VJZCI6IkRFVklDRV8wMDEiLCJleHAiOjE3NTg4NTE0NDV9.joDdP7owmZf51cwaY16g9VM-J0Egdi6lt-4X7ZqNid0 \ No newline at end of file diff --git a/code/sdk/zhihui/leiniao/leiniao_python_test/device_websocket_client_6.py b/code/sdk/zhihui/leiniao/leiniao_python_test/device_websocket_client_6.py new file mode 100644 index 000000000..54d961839 --- /dev/null +++ b/code/sdk/zhihui/leiniao/leiniao_python_test/device_websocket_client_6.py @@ -0,0 +1,417 @@ +import websocket +import threading +import time +import json +import ssl +import urllib.parse +import requests +from typing import Optional, Callable, Dict, Any, Tuple, NoReturn +import hashlib +import hmac +import random +import string +import logging +from enum import Enum + +# 配置日志 +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(levelname)s - %(message)s', + datefmt='%Y-%m-%d %H:%M:%S' +) +logger = logging.getLogger(__name__) + +# 常量定义 +class Constants(Enum): + # WS_BASE_URL = "wss://ambybox-api.test.leiniao.com/ambybox-device-api/v1/websocket" + WS_BASE_URL = "ws://127.0.0.1:8765" # 临时本地测试,到时候需要替换为正式环境 + TOKEN_API_URL = "https://ambybox-api.test.leiniao.com/ambybox-device-api/v1/auth/getDeviceToken" + UPGRADE_TEST_URL = "https://ambybox-api.test.leiniao.com/ambybox-device-api/v1/upgrade/getTask" + SIGN_SECRET = "vDdXVXgPUEUM6_piUjz5AEhMvYTSFzQdPaN-ogKnSus=" + DEVICE_MODEL = "abmybox_1" + DEVICE_VERSION = "1.0.0" + HEARTBEAT_INTERVAL = 30 # 心跳间隔(秒) + HEARTBEAT_TIMEOUT = 10 # 心跳超时(秒) + CONNECT_TIMEOUT = 20 # 连接超时(秒) + MAX_RECONNECT_DELAY = 60 # 最大重连延迟(秒) + + +class DeviceWebSocketClient: + """设备WebSocket客户端，支持自动重连、心跳检测和命令处理""" + + def __init__(self, device_id: str, device_token: str, device_iccid: str): + self.device_id = device_id.strip() + self.device_token = device_token.strip() + self.device_iccid = device_iccid.strip() + + self.ws_url = Constants.WS_BASE_URL.value + self.ws: Optional[websocket.WebSocketApp] = None + self.connected = False + self.reconnect_count = 0 + self.running = False + self.last_heartbeat_response = time.time() + + # 命令处理器映射 + self.command_handlers: Dict[str, Callable[[Dict[str, Any]], None]] = { + "CharacterChange": self.handle_character_change, + "CharacterConfigChange": self.handle_character_config_change, + "SleepModeStart": self.handle_sleep_mode_start, + "SleepModeEnd": self.handle_sleep_mode_end, + "DeviceBind": self.handle_device_bind, + "DeviceUnbind": self.handle_device_unbind, + "ScheduleReminder": self.handle_schedule_reminder + } + + # 线程相关 + self.heartbeat_thread: Optional[threading.Thread] = None + self.heartbeat_monitor_thread: Optional[threading.Thread] = None + + def _validate_token_format(self) -> bool: + """验证Token格式是否正确""" + return len(self.device_token.split('.')) == 3 + + def _test_token_validity(self) -> bool: + """测试Token有效性""" + try: + params = {"deviceId": self.device_id, "sourceVersion": Constants.DEVICE_VERSION.value} + headers = self._get_common_headers() + + response = requests.get( + Constants.UPGRADE_TEST_URL.value, + params=params, + headers=headers, + timeout=10, + verify=False + ) + + if response.status_code == 401: + logger.error("Token已失效或无效") + return False + return True + + except Exception as e: + logger.error(f"Token验证接口调用失败: {str(e)}") + return False + + def _get_common_headers(self) -> Dict[str, str]: + """获取通用请求头""" + return { + "deviceToken": self.device_token, + "deviceId": self.device_id, + "deviceIccid": self.device_iccid, + "deviceModel": Constants.DEVICE_MODEL.value, + "deviceVersion": Constants.DEVICE_VERSION.value, + "Content-Type": "application/json" + } + + def connect(self) -> bool: + """建立WebSocket连接，尝试多种参数风格""" + if not self._test_token_validity(): + logger.error("Token无效，终止连接尝试") + return False + + try: + # URL编码参数 + encoded_token = urllib.parse.quote(self.device_token) + encoded_device_id = urllib.parse.quote(self.device_id) + encoded_iccid = urllib.parse.quote(self.device_iccid) + + # 参数风格组合 + param_styles = [ + f"deviceId={encoded_device_id}&deviceToken={encoded_token}&deviceIccid={encoded_iccid}&reconnect={self.reconnect_count}", + f"deviceid={encoded_device_id}&devicetoken={encoded_token}&deviceiccid={encoded_iccid}&reconnect={self.reconnect_count}", + f"deviceId={encoded_device_id}&deviceToken={encoded_token}&deviceModel={Constants.DEVICE_MODEL.value}&deviceVersion={Constants.DEVICE_VERSION.value}&reconnect={self.reconnect_count}", + f"reconnect={self.reconnect_count}&deviceToken={encoded_token}&deviceId={encoded_device_id}" + ] + + for idx, style in enumerate(param_styles, 1): + url = f"{self.ws_url}?{style}" + logger.info(f"尝试连接（风格{idx}）: {url}") + + self.ws = websocket.WebSocketApp( + url, + on_open=self._on_open, + on_message=self._on_message, + on_error=lambda ws, err: self._on_error(ws, err, idx), + on_close=self._on_close, + header=[ + f"deviceId: {self.device_id}", + f"deviceIccid: {self.device_iccid}", + f"deviceModel: {Constants.DEVICE_MODEL.value}", + f"deviceVersion: {Constants.DEVICE_VERSION.value}" + ] + ) + + # 启动连接线程 + connect_thread = threading.Thread(target=self._run_forever, daemon=True) + connect_thread.start() + + # 等待连接成功 + for _ in range(Constants.CONNECT_TIMEOUT.value): + if self.connected: + return True + time.sleep(1) + + # 连接超时，关闭当前尝试 + self.connected = False + if self.ws: + self.ws.close() + + logger.warning("所有参数风格均尝试失败") + return False + + except Exception as e: + logger.error(f"连接建立失败: {str(e)}") + return False + + def _run_forever(self) -> None: + """运行WebSocket连接循环""" + try: + self.ws.run_forever( + sslopt={ + "cert_reqs": ssl.CERT_NONE, + "check_hostname": False + } + ) + except Exception as e: + logger.error(f"WebSocket运行错误: {str(e)}") + + def _on_open(self, ws): + print("✅ WebSocket连接已建立") + self.connected = True + self.reconnect_count = 0 + self._start_heartbeat() + + # 发送上线通知（修正缩进和变量名） + online_msg = json.dumps({"type": "online", "deviceId": self.device_id, "timestamp": int(time.time())}) + self.ws.send(online_msg) + print(f"📤 已发送上线消息: {online_msg}") + + def _on_message(self, ws, message: str) -> None: + """消息接收回调""" + try: + logger.info(f"收到消息: {message}") + + # 处理心跳响应 + if message == "pong": + self.last_heartbeat_response = time.time() + logger.info("收到心跳响应") + return + + # 处理业务消息 + data = json.loads(message) + command = data.get("command") + + if command: + handler = self.command_handlers.get(command) + if handler: + handler(data) + else: + logger.warning(f"未处理的命令: {command}") + else: + logger.warning("收到的消息不包含command字段") + + except json.JSONDecodeError: + logger.warning(f"无法解析消息: {message}") + except Exception as e: + logger.error(f"处理消息出错: {str(e)}") + + def _on_error(self, ws, error: Exception, style_idx: int) -> None: + """错误回调""" + logger.error(f"WebSocket错误（参数风格: {style_idx}）: {str(error)}") + if "400" in str(error): + logger.info("可能原因：") + logger.info(" 1. WebSocket需要额外的参数（如deviceIccid）") + logger.info(" 2. 参数顺序不符合服务器要求") + logger.info(" 3. 需要在请求头而非URL中传递某些参数") + + def _on_close(self, ws, close_status_code: int, close_msg: str) -> None: + """连接关闭回调""" + logger.info(f"连接关闭，状态码: {close_status_code}, 消息: {close_msg}") + self.connected = False + self._stop_heartbeat() + + if self.running: + self.reconnect_count += 1 + # 指数退避重连策略 + delay = min(2 ** self.reconnect_count, Constants.MAX_RECONNECT_DELAY.value) + logger.info(f"将在 {delay} 秒后尝试第 {self.reconnect_count} 次重连...") + threading.Timer(delay, self.connect).start() + + def _start_heartbeat(self) -> None: + """启动心跳发送和监控线程""" + self._stop_heartbeat() + + # 心跳发送线程 + self.heartbeat_thread = threading.Thread(target=self._heartbeat_loop, daemon=True) + self.heartbeat_thread.start() + + # 心跳监控线程 + self.heartbeat_monitor_thread = threading.Thread(target=self._heartbeat_monitor, daemon=True) + self.heartbeat_monitor_thread.start() + + def _stop_heartbeat(self) -> None: + """停止心跳相关线程""" + if self.heartbeat_thread and self.heartbeat_thread.is_alive(): + self.heartbeat_thread.join(timeout=1) + + if self.heartbeat_monitor_thread and self.heartbeat_monitor_thread.is_alive(): + self.heartbeat_monitor_thread.join(timeout=1) + + def _heartbeat_loop(self) -> None: + """心跳发送循环""" + while self.connected and self.running: + try: + self.ws.send("ping") + logger.info("发送心跳 ping") + time.sleep(Constants.HEARTBEAT_INTERVAL.value) + except Exception as e: + logger.error(f"发送心跳出错: {str(e)}") + break + + def _heartbeat_monitor(self) -> None: + """心跳监控，检测超时""" + while self.connected and self.running: + time.sleep(2) + if time.time() - self.last_heartbeat_response > Constants.HEARTBEAT_TIMEOUT.value + Constants.HEARTBEAT_INTERVAL.value: + logger.warning("心跳超时，触发重连") + self.connected = False + if self.ws: + self.ws.close() + break + + def send_message(self, message: str) -> bool: + """发送消息到服务器""" + if not self.connected or not self.ws: + logger.error("无法发送消息，连接未建立") + return False + + try: + self.ws.send(message) + logger.info(f"已发送消息: {message}") + return True + except Exception as e: + logger.error(f"发送消息失败: {str(e)}") + return False + + def close(self) -> None: + """关闭客户端连接""" + self.running = False + self._stop_heartbeat() + if self.ws: + self.ws.close() + self.connected = False + logger.info("WebSocket客户端已关闭") + + # 命令处理器实现 + def handle_character_change(self, data: Dict[str, Any]) -> None: + logger.info("收到角色切换命令") + command_data = data.get("data", {}) + logger.info(f"切换到角色: {command_data.get('characterName')} (ID: {command_data.get('characterId')})") + + def handle_character_config_change(self, data: Dict[str, Any]) -> None: + logger.info("收到角色配置更改命令") + + def handle_sleep_mode_start(self, data: Dict[str, Any]) -> None: + logger.info("收到哄睡模式开始命令") + + def handle_sleep_mode_end(self, data: Dict[str, Any]) -> None: + logger.info("收到哄睡模式结束命令") + + def handle_device_bind(self, data: Dict[str, Any]) -> None: + logger.info("收到设备绑定命令") + + def handle_device_unbind(self, data: Dict[str, Any]) -> None: + logger.info("收到设备解绑命令") + + def handle_schedule_reminder(self, data: Dict[str, Any]) -> None: + logger.info("收到日程提醒命令") + + +def get_valid_device_token(device_id: str) -> Tuple[Optional[str], str]: + """获取有效的设备Token和ICCID""" + try: + device_iccid = "86123456789012345678" + timestamp = str(int(time.time() * 1000)) + nonce = ''.join(random.choices(string.ascii_letters + string.digits, k=12)) + + # 生成签名 + sign_data = f"{device_id}:{device_iccid}:{timestamp}:{nonce}" + hmac_obj = hmac.new( + Constants.SIGN_SECRET.value.encode("utf-8"), + sign_data.encode("utf-8"), + hashlib.sha256 + ) + signature = hmac_obj.hexdigest().lower() + + logger.debug("签名调试信息：") + logger.debug(f" 签名数据: {sign_data}") + logger.debug(f" 生成的签名: {signature}") + + headers = { + "deviceId": device_id, + "deviceIccid": device_iccid, + "timestamp": timestamp, + "nonce": nonce, + "signature": signature, + "deviceModel": Constants.DEVICE_MODEL.value, + "deviceVersion": Constants.DEVICE_VERSION.value, + "Content-Type": "application/json" + } + + response = requests.get( + Constants.TOKEN_API_URL.value, + headers=headers, + timeout=10, + verify=False + ) + response_data = response.json() + logger.debug(f"服务器响应: {response_data}") + + if response_data.get("errorCode") is None and "data" in response_data: + token = response_data["data"].get("token") + logger.info("成功获取并验证新Token") + return token, device_iccid + else: + logger.error(f"获取Token失败: {response_data.get('message', '未知错误')}") + return None, device_iccid + + except Exception as e: + logger.error(f"获取Token过程出错: {str(e)}") + return None, "" + + +def main() -> NoReturn: + device_id = "DEVICE_PC_TEST_001".strip() + if not device_id: + logger.error("deviceId不能为空") + return + + # 获取Token和ICCID + device_token, device_iccid = get_valid_device_token(device_id) + if not device_token: + logger.error("无法获取有效Token，程序退出") + return + + # 创建并启动客户端 + ws_client = DeviceWebSocketClient(device_id, device_token, device_iccid) + ws_client.running = True + if not ws_client.connect(): + logger.error("WebSocket连接失败，程序退出") + return + + try: + while True: + time.sleep(1) + except KeyboardInterrupt: + logger.info("\n用户中断程序") + finally: + ws_client.close() + + +# 忽略不安全请求警告 +import urllib3 +urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/code/sdk/zhihui/leiniao/leiniao_python_test/get_device_token_1.py b/code/sdk/zhihui/leiniao/leiniao_python_test/get_device_token_1.py new file mode 100644 index 000000000..ab03e6c87 --- /dev/null +++ b/code/sdk/zhihui/leiniao/leiniao_python_test/get_device_token_1.py @@ -0,0 +1,80 @@ +import requests +import time +import hashlib +import hmac +import random +import string +import json # 移到顶部导入，避免重复导入 + +# -------------------------- 1. 接口基础配置 -------------------------- +BASE_URL = "https://ambybox-api.test.leiniao.com/ambybox-device-api" +INTERFACE_PATH = "/v1/auth/getDeviceToken" +FULL_URL = BASE_URL + INTERFACE_PATH +SIGN_SECRET = "vDdXVXgPUEUM6_piUjz5AEhMvYTSFzQdPaN-ogKnSus=" + +# -------------------------- 2. 生成合法请求参数 -------------------------- +def generate_request_params(): + device_id = "DEVICE_PC_TEST_001" + device_iccid = "86123456789012345678" + + # 1. 时间戳格式修正：确保是字符串类型的毫秒级时间戳 + timestamp = str(int(time.time() * 1000)) + + # 2. 随机数生成：确保长度在8-16位之间（当前12位符合要求） + nonce_length = 12 + nonce = ''.join(random.choices(string.ascii_letters + string.digits, k=nonce_length)) + + # 3. 签名生成（保持不变，符合协议要求） + sign_data = f"{device_id}:{device_iccid}:{timestamp}:{nonce}" + hmac_obj = hmac.new(SIGN_SECRET.encode("utf-8"), sign_data.encode("utf-8"), hashlib.sha256) + signature = hmac_obj.hexdigest().lower() + + # 4. 返回参数（键名严格匹配文档，保持不变） + return { + "deviceId": device_id, + "deviceIccid": device_iccid, + "timestamp": timestamp, + "nonce": nonce, + "signature": signature, + "deviceModel": "abmybox_1", + "deviceVersion": "1.0.0" + } + +# -------------------------- 3. 发送请求并解析响应 -------------------------- +def send_get_device_token_request(): + headers = generate_request_params() + headers["Content-Type"] = "application/json" + + try: + response = requests.get(url=FULL_URL, headers=headers, timeout=10) + response.raise_for_status() # 新增：抛出HTTP错误状态码异常（如404、500） + response_data = response.json() + + print("=" * 50) + print("【请求信息】") + print(f"请求 URL: {FULL_URL}") + print(f"请求头: {headers}") + print("=" * 50) + + print("【响应结果】") + print(json.dumps(response_data, ensure_ascii=False, indent=2)) + + # 响应判断逻辑优化：严格按文档判断errorCode是否为null + if response_data.get("errorCode") is None: + print("\n✅ 请求成功！") + # 增加数据存在性检查，避免KeyError + if "data" in response_data: + print(f"设备 Token: {response_data['data'].get('token')}") + print(f"Token 过期时间: {response_data['data'].get('expireTime')} 秒（{response_data['data'].get('expireTime', 0)//86400} 天）") + else: + print(f"\n❌ 请求失败！错误码: {response_data['errorCode']}, 原因: {response_data.get('message', '未知错误')}") + + except requests.exceptions.RequestException as e: + print(f"\n❌ 网络请求异常: {str(e)}") + except json.JSONDecodeError: + # 新增：处理非JSON格式响应 + print(f"\n❌ 响应格式错误，非JSON数据: {response.text}") + +# -------------------------- 4. 执行请求 -------------------------- +if __name__ == "__main__": + send_get_device_token_request() \ No newline at end of file diff --git a/code/sdk/zhihui/leiniao/leiniao_python_test/integrated_speech_chatstream_2.py b/code/sdk/zhihui/leiniao/leiniao_python_test/integrated_speech_chatstream_2.py new file mode 100644 index 000000000..bde84f590 --- /dev/null +++ b/code/sdk/zhihui/leiniao/leiniao_python_test/integrated_speech_chatstream_2.py @@ -0,0 +1,177 @@ +import os +import requests +import time +import hashlib +import hmac +import random +import string +import json +import re + +# -------------------------- 基础配置 -------------------------- +BASE_URL = "https://ambybox-api.test.leiniao.com/ambybox-device-api" +TOKEN_INTERFACE = "/v1/auth/getDeviceToken" +CHAT_INTERFACE = "/v1/speechChat/chatStream" +TOKEN_URL = BASE_URL + TOKEN_INTERFACE +CHAT_URL = BASE_URL + CHAT_INTERFACE + +SIGN_SECRET = "vDdXVXgPUEUM6_piUjz5AEhMvYTSFzQdPaN-ogKnSus=" +DEVICE_ID = "DEVICE_PC_TEST_001" +DEVICE_ICCID = "86123456789012345678" +AUDIO_FILE_PATH = "test_audio.mp3" + +# 新增：指定音频保存目录（可自定义路径） +AUDIO_SAVE_DIR = "audio_segments" +# 确保目录存在 +os.makedirs(AUDIO_SAVE_DIR, exist_ok=True) + +# -------------------------- 1. 获取设备Token -------------------------- +def get_device_token(): + # （原有代码不变） + try: + timestamp = str(int(time.time() * 1000)) + nonce = ''.join(random.choices(string.ascii_letters + string.digits, k=12)) + sign_data = f"{DEVICE_ID}:{DEVICE_ICCID}:{timestamp}:{nonce}" + hmac_obj = hmac.new(SIGN_SECRET.encode("utf-8"), sign_data.encode("utf-8"), hashlib.sha256) + signature = hmac_obj.hexdigest().lower() + headers = { + "deviceId": DEVICE_ID, + "deviceIccid": DEVICE_ICCID, + "timestamp": timestamp, + "nonce": nonce, + "signature": signature, + "deviceModel": "abmybox_1", + "deviceVersion": "1.0.0", + "Content-Type": "application/json" + } + response = requests.get(TOKEN_URL, headers=headers, timeout=10) + response_data = response.json() + if response_data.get("errorCode") is None and "data" in response_data: + device_token = response_data["data"]["token"] + print(f"设备 Token: {response_data['data']['token']}") + print(f"Token 过期时间: {response_data['data']['expireTime']} 秒（{response_data['data']['expireTime']//86400} 天）") + return device_token + else: + print(f"❌ 获取Token失败：{response_data.get('message', '未知错误')}") + return None + except Exception as e: + print(f"❌ 获取Token时发生异常：{str(e)}") + return None + +# -------------------------- 2. 发送流式语音对话请求 -------------------------- +def send_stream_speech_chat_request(token): + if not token: + print("❌ Token为空，无法调用接口") + return + + headers = { + "deviceId": DEVICE_ID, + "deviceToken": token, + "Accept": "text/event-stream", + "Cache-Control": "no-cache" + } + + try: + with open(AUDIO_FILE_PATH, 'rb') as audio_file: + files = { + 'file': ('test_audio.mp3', audio_file, 'audio/mpeg') + } + + print("正在发送流式语音对话请求...") + with requests.post( + url=CHAT_URL, + headers=headers, + files=files, + stream=True, + timeout=60 + ) as response: + if response.status_code != 200: + print(f"❌ 连接接口失败，状态码：{response.status_code}") + return + + print("=" * 50) + print("【请求信息】") + print(f"请求URL: {CHAT_URL}") + print(f"设备ID: {DEVICE_ID}") + print(f"上传文件: {AUDIO_FILE_PATH}") + print("=" * 50) + print("【流式响应结果】") + + # 新增：初始化自增计数器（确保按服务器返回顺序编号） + segment_counter = 0 + + for line in response.iter_lines(decode_unicode=True): + if line: + data_line = re.sub(r'^data:\s*', '', line.strip()) + if not data_line: + continue + + try: + segment = json.loads(data_line) + print(json.dumps(segment, ensure_ascii=False, indent=2)) + + if segment.get("errorCode"): + error_code = segment["errorCode"] + error_msg = segment.get("message", "未知错误") + print(f"\n❌ 发生错误：{error_code} - {error_msg}") + + if error_code == "AUTH_FAILURE": + print("🔄 尝试重新获取Token...") + new_token = get_device_token() + if new_token: + print("🔄 重新发送对话请求...") + send_stream_speech_chat_request(new_token) + elif error_code == "DEVICE_NOT_FOUND": + print("🔄 设备信息异常，尝试重新获取Token...") + new_token = get_device_token() + if new_token: + send_stream_speech_chat_request(new_token) + return + + data = segment.get("data", {}) + if data: + text = data.get("textContent", "") + if text: + print(f"📝 文本片段：{text}") + + audio_url = data.get("audioUrl") + if audio_url: + # 关键修改：使用自增计数器作为序号，而非len(text) + segment_counter += 1 + download_audio_segment(audio_url, segment_counter) + + intention = data.get("intention") + if intention == "StartSleepMode": + print("\n😴 检测到哄睡意图，准备进入哄睡模式...") + + except json.JSONDecodeError: + print(f"❌ 解析响应片段失败：{data_line}") + + print("\n✅ 流式对话已完成") + + except FileNotFoundError: + print(f"\n❌ 音频文件未找到：{AUDIO_FILE_PATH}") + except Exception as e: + print(f"\n❌ 流式对话请求异常：{str(e)}") + +# -------------------------- 3. 下载音频片段 -------------------------- +def download_audio_segment(audio_url, segment_idx): + """下载流式响应中的音频片段（按服务器返回顺序编号，保存到指定路径）""" + try: + print(f"🎵 正在下载音频片段 {segment_idx}...") + response = requests.get(audio_url, timeout=10) + # 关键修改：保存到指定目录，文件名仅包含序号（确保顺序性） + output_path = os.path.join(AUDIO_SAVE_DIR, f"response_audio_segment_{segment_idx}.mp3") + with open(output_path, 'wb') as f: + f.write(response.content) + print(f"✅ 音频片段已保存至：{output_path}") + except Exception as e: + print(f"❌ 音频片段下载失败：{str(e)}") + +# -------------------------- 主流程 -------------------------- +if __name__ == "__main__": + device_token = get_device_token() + if device_token: + send_stream_speech_chat_request(device_token) + else: + print("❌ 无法获取有效Token，终止流程") \ No newline at end of file diff --git a/code/sdk/zhihui/leiniao/leiniao_python_test/mock_ws_server.py b/code/sdk/zhihui/leiniao/leiniao_python_test/mock_ws_server.py new file mode 100644 index 000000000..ebe95fbdb --- /dev/null +++ b/code/sdk/zhihui/leiniao/leiniao_python_test/mock_ws_server.py @@ -0,0 +1,64 @@ + +# 本地起一个 Mock WebSocket 服务端，向你的脚本推送业务消息 +import asyncio, websockets, json, time, uuid + +clients = set() + +def mk_msg(cmd, data): + return { + "deviceId": "DEVICE_PC_TEST_001", + "timestamp": int(time.time()), + "messageId": str(uuid.uuid4()), + "command": cmd, + "data": data, + } + +async def handler(ws, path): + clients.add(ws) + print("client connected:", path) + try: + async for msg in ws: + print("recv:", msg) + if msg == "ping": + await ws.send("pong") + finally: + clients.remove(ws) + print("client disconnected") + +async def push_loop(): + # 等待客户端连上 + while not clients: + await asyncio.sleep(0.5) + + # 依次推送多条业务消息，验证各个 handler + samples = [ + mk_msg("CharacterChange", {"characterId": 101, "characterName": "小雷"}), + mk_msg("CharacterConfigChange", {"hairColor": "black", "eyesColor": "brown"}), + mk_msg("SleepModeStart", {"storyTheme": "晚安故事"}), + mk_msg("ScheduleReminder", {"scheduleName": "吃饭", "reminderContent": "记得吃午饭"}), + mk_msg("DeviceBind", {"userId": "u123", "userName": "Tom"}), + mk_msg("SleepModeEnd", {}), + mk_msg("DeviceUnbind", {"reason": "解绑测试"}), + ] + + for m in samples: + s = json.dumps(m, ensure_ascii=False) + for c in list(clients): + await c.send(s) + print("sent:", s) + await asyncio.sleep(2) + + # 循环心跳演示（可选） + while clients: + for c in list(clients): + await c.send("pong") # 服务端也可主动发一条文本 + await asyncio.sleep(30) + +async def main(): + server = await websockets.serve(handler, "0.0.0.0", 8765) + print("mock ws server on ws://127.0.0.1:8765") + asyncio.create_task(push_loop()) + await server.wait_closed() + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/code/sdk/zhihui/leiniao/leiniao_python_test/ota_upgrade_task_5.py b/code/sdk/zhihui/leiniao/leiniao_python_test/ota_upgrade_task_5.py new file mode 100644 index 000000000..7c3e9d575 --- /dev/null +++ b/code/sdk/zhihui/leiniao/leiniao_python_test/ota_upgrade_task_5.py @@ -0,0 +1,288 @@ +import requests +import time +import hashlib +import hmac +import random +import string +import json +import os +import shutil +from typing import Dict, Optional, Tuple +import subprocess # 用于模拟设备重启 + +# -------------------------- 基础配置 -------------------------- +BASE_URL = "https://ambybox-api.test.leiniao.com/ambybox-device-api" + +# 接口路径 +TOKEN_INTERFACE = "/v1/auth/getDeviceToken" +OTA_TASK_INTERFACE = "/v1/upgrade/getTask" + +# 完整URL +TOKEN_URL = BASE_URL + TOKEN_INTERFACE +OTA_TASK_URL = BASE_URL + OTA_TASK_INTERFACE + +# 设备与认证信息 +SIGN_SECRET = "vDdXVXgPUEUM6_piUjz5AEhMvYTSFzQdPaN-ogKnSus=" +DEVICE_ID = "DEVICE_001" +DEVICE_ICCID = "86123456789012345678" +CURRENT_VERSION = "1.0.0" # 当前固件版本 +UPGRADE_PACKAGE_DIR = "upgrade_packages" # 升级包存储目录 + +# -------------------------- 工具函数 -------------------------- +def get_device_token() -> Optional[str]: + """获取设备Token""" + try: + timestamp = str(int(time.time() * 1000)) + nonce = ''.join(random.choices(string.ascii_letters + string.digits, k=12)) + + # 生成签名 + sign_data = f"{DEVICE_ID}:{DEVICE_ICCID}:{timestamp}:{nonce}" + hmac_obj = hmac.new(SIGN_SECRET.encode("utf-8"), sign_data.encode("utf-8"), hashlib.sha256) + signature = hmac_obj.hexdigest().lower() + + headers = { + "deviceId": DEVICE_ID, + "deviceIccid": DEVICE_ICCID, + "timestamp": timestamp, + "nonce": nonce, + "signature": signature, + "deviceModel": "abmybox_1", + # 注意：升级后这里应该使用新的固件版本 + "deviceVersion": CURRENT_VERSION, + "Content-Type": "application/json" + } + + response = requests.get(TOKEN_URL, headers=headers, timeout=10) + response_data = response.json() + + if response_data.get("errorCode") is None and "data" in response_data: + print("✅ 成功获取设备Token") + return response_data["data"]["token"] + else: + print(f"❌ 获取Token失败：{response_data.get('message', '未知错误')}") + return None + + except Exception as e: + print(f"❌ 获取Token异常：{str(e)}") + return None + + +def check_ota_update() -> Optional[Dict]: + """检查是否有可用的OTA升级任务""" + try: + params = { + "deviceId": DEVICE_ID, + "sourceVersion": CURRENT_VERSION + } + + token = get_device_token() + if not token: + print("❌ 无法获取 Token") + return None + + headers = { + "deviceId": DEVICE_ID, + "deviceToken": token, + "deviceModel": "abmybox_1", + "deviceVersion": CURRENT_VERSION, + "Content-Type": "application/json", + } + + print(f"📱 检查OTA升级 (当前版本: {CURRENT_VERSION})...") + print("DEBUG URL:", OTA_TASK_URL, "params:", params) + print("DEBUG headers:", headers) + response = requests.get(OTA_TASK_URL, params=params, headers=headers, timeout=10) + response_data = response.json() + + print("=" * 50) + print("【OTA升级检查响应】") + print(json.dumps(response_data, ensure_ascii=False, indent=2)) + + error_code = response_data.get("errorCode") + if error_code is None: + # 有可用升级 + return response_data.get("data") + elif error_code == "UPGRADE_NOT_FOUND": + print("\nℹ️ 已是最新版本，无需升级") + return None + else: + print(f"\n❌ 检查升级失败：{response_data.get('message', '未知错误')}") + return None + + except Exception as e: + print(f"❌ 检查OTA升级异常：{str(e)}") + return None + + +def download_upgrade_package(package_url: str, target_path: str) -> bool: + """下载升级包""" + try: + print(f"\n📥 开始下载升级包: {package_url}") + + # 创建存储目录 + os.makedirs(os.path.dirname(target_path), exist_ok=True) + + # 流式下载大文件 + with requests.get(package_url, stream=True, timeout=300) as r: + r.raise_for_status() + with open(target_path, 'wb') as f: + for chunk in r.iter_content(chunk_size=8192): + f.write(chunk) + + print(f"✅ 升级包下载完成，保存至: {target_path}") + return True + + except Exception as e: + print(f"❌ 升级包下载失败：{str(e)}") + if os.path.exists(target_path): + os.remove(target_path) + return False + + +def verify_package_md5(file_path: str, expected_md5: str) -> bool: + """验证升级包的MD5值""" + try: + print(f"\n🔍 验证升级包MD5...") + hash_md5 = hashlib.md5() + + with open(file_path, "rb") as f: + for chunk in iter(lambda: f.read(4096), b""): + hash_md5.update(chunk) + + actual_md5 = hash_md5.hexdigest() + print(f"计算得到的MD5: {actual_md5}") + print(f"预期的MD5: {expected_md5}") + + result = actual_md5 == expected_md5 + if result: + print("✅ MD5验证通过") + else: + print("❌ MD5验证失败") + return result + + except Exception as e: + print(f"❌ MD5验证异常：{str(e)}") + return False + + +def simulate_device_reboot(new_version: str) -> Tuple[bool, str]: + """ + 模拟设备重启 + + 在实际设备中，这会是硬件级别的重启 + 这里通过修改版本号并模拟重启过程来实现 + """ + global CURRENT_VERSION + + try: + print("\n🔄 准备重启设备...") + + # 模拟关闭设备服务 + print("正在关闭设备服务...") + time.sleep(2) + + # 模拟重启过程 + print("设备重启中", end="") + for _ in range(3): + time.sleep(1) + print(".", end="", flush=True) + print() + + # 更新版本号（实际设备中这会由固件升级过程自动完成） + CURRENT_VERSION = new_version + print(f"✅ 设备重启完成，当前版本已更新为: {CURRENT_VERSION}") + + return True, CURRENT_VERSION + + except Exception as e: + print(f"❌ 设备重启失败：{str(e)}") + return False, CURRENT_VERSION + + +def perform_ota_upgrade() -> bool: + """执行完整的OTA升级流程""" + # 1. 检查是否有可用升级 + upgrade_info = check_ota_update() + if not upgrade_info: + return False + + # 2. 解析升级信息 + target_version = upgrade_info.get("targetVersion") + update_type = upgrade_info.get("updateType") + package_url = upgrade_info.get("packageUrl") + package_md5 = upgrade_info.get("packageMd5") + package_size = upgrade_info.get("packageSize") + + print(f"\n📋 升级信息:") + print(f"目标版本: {target_version}") + print(f"升级类型: {'强制升级' if update_type == 1 else '可选升级'}") + print(f"升级包大小: {package_size} bytes") + + # 3. 处理可选升级的用户确认 + if update_type == 0: + choice = input("\n是否执行升级? (y/n): ").strip().lower() + if choice != 'y': + print("❌ 用户取消升级") + return False + + # 4. 下载升级包 + package_filename = f"upgrade_{target_version}.bin" + package_path = os.path.join(UPGRADE_PACKAGE_DIR, package_filename) + + if not download_upgrade_package(package_url, package_path): + return False + + # 5. 验证升级包完整性 + if not verify_package_md5(package_path, package_md5): + # 尝试重新下载一次 + print("尝试重新下载升级包...") + if not download_upgrade_package(package_url, package_path) or \ + not verify_package_md5(package_path, package_md5): + return False + + # 6. 模拟固件升级过程 + print("\n🚀 开始固件升级...") + time.sleep(3) # 模拟升级过程 + print("✅ 固件升级完成") + + # 7. 模拟设备重启 + reboot_success, new_version = simulate_device_reboot(target_version) + if not reboot_success: + return False + + # 8. 重启后重新获取设备Token（关键步骤） + print("\n🔑 重启后重新获取设备Token...") + new_token = get_device_token() + + if new_token: + print("\n🎉 OTA升级流程已完成，设备正常运行") + # 可以在这里将新Token保存到安全存储 + with open("device_token.txt", "w") as f: + f.write(new_token) + return True + else: + print("\n❌ 升级后无法获取Token，设备可能无法正常工作") + return False + + +# -------------------------- 主流程 -------------------------- +def main(): + print("=" * 60) + print(" OTA升级流程演示 ") + print("=" * 60) + + # 执行OTA升级 + success = perform_ota_upgrade() + + # 升级后的操作 + if success: + print("\n升级后设备状态检查:") + print(f"当前固件版本: {CURRENT_VERSION}") + + # 检查Token是否有效（可以调用一个需要认证的接口） + token = get_device_token() + if token: + print("设备已准备就绪，可以正常使用") + +if __name__ == "__main__": + main() diff --git a/code/sdk/zhihui/leiniao/leiniao_python_test/story_apis_test_3.py b/code/sdk/zhihui/leiniao/leiniao_python_test/story_apis_test_3.py new file mode 100644 index 000000000..e0c0e8cc1 --- /dev/null +++ b/code/sdk/zhihui/leiniao/leiniao_python_test/story_apis_test_3.py @@ -0,0 +1,358 @@ +import requests +import time +import hashlib +import hmac +import random +import string +import json +import logging +from typing import Dict, Optional, Callable +from pathlib import Path + +# -------------------------- 配置与日志初始化 -------------------------- +class Config: + """配置常量类，集中管理所有配置参数""" + BASE_URL = "https://ambybox-api.test.leiniao.com/ambybox-device-api" + # 接口路径 + TOKEN_INTERFACE = "/v1/auth/getDeviceToken" + CHOOSE_STORY_INTERFACE = "/v1/story/chooseStory" + RANDOM_STORY_INTERFACE = "/v1/story/randomStory" + EXIT_INQUIRY_INTERFACE = "/v1/story/getExitSleepInquiry" + FINISH_SLEEP_INTERFACE = "/v1/story/finishSleepMode" + + # 完整URL + TOKEN_URL = f"{BASE_URL}{TOKEN_INTERFACE}" + CHOOSE_STORY_URL = f"{BASE_URL}{CHOOSE_STORY_INTERFACE}" + RANDOM_STORY_URL = f"{BASE_URL}{RANDOM_STORY_INTERFACE}" + EXIT_INQUIRY_URL = f"{BASE_URL}{EXIT_INQUIRY_INTERFACE}" + FINISH_SLEEP_URL = f"{BASE_URL}{FINISH_SLEEP_INTERFACE}" + + # 认证信息 + SIGN_SECRET = "vDdXVXgPUEUM6_piUjz5AEhMvYTSFzQdPaN-ogKnSus=" + DEVICE_ID = "DEVICE_PC_TEST_001" + DEVICE_ICCID = "86123456789012345678" + DEVICE_MODEL = "abmybox_1" + DEVICE_VERSION = "1.0.0" + + # 资源与超时配置 + AUDIO_FILE_PATH = Path("story_request.mp3") # 语音文件路径 + SSE_TIMEOUT = 120 # SSE流式响应超时时间(秒) + NORMAL_TIMEOUT = 10 # 普通接口超时时间(秒) + TOKEN_RETRY_COUNT = 3 # Token获取重试次数 + + +# 日志配置 +logging.basicConfig( + level=logging.INFO, + format="%(message)s", + handlers=[logging.StreamHandler()] +) +logger = logging.getLogger(__name__) + + +# -------------------------- 工具函数 -------------------------- +def exception_handler(func: Callable) -> Callable: + """通用异常处理装饰器""" + def wrapper(*args, **kwargs): + try: + return func(*args, **kwargs) + except requests.exceptions.RequestException as e: + logger.error(f"❌ 网络请求异常: {str(e)}") + except FileNotFoundError as e: + logger.error(f"❌ 文件未找到: {str(e)}") + except json.JSONDecodeError as e: + logger.error(f"❌ JSON解析失败: {str(e)}") + except Exception as e: + logger.error(f"❌ 发生未知异常: {str(e)}") + return None + return wrapper + + +def generate_signature(device_id: str, device_iccid: str, timestamp: str, nonce: str) -> str: + """生成签名""" + sign_data = f"{device_id}:{device_iccid}:{timestamp}:{nonce}" + hmac_obj = hmac.new( + Config.SIGN_SECRET.encode("utf-8"), + sign_data.encode("utf-8"), + hashlib.sha256 + ) + return hmac_obj.hexdigest().lower() + + +def create_auth_headers(token: str) -> Dict[str, str]: + """创建通用认证请求头""" + return { + "deviceId": Config.DEVICE_ID, + "deviceToken": token, + "Content-Type": "application/json" + } + + +def create_sse_headers(token: str) -> Dict[str, str]: + """创建SSE流式请求头""" + return { + "deviceId": Config.DEVICE_ID, + "deviceToken": token, + "Accept": "text/event-stream", + "Cache-Control": "no-cache" + } + + +def process_sse_stream(response: requests.Response) -> None: + """处理SSE流式响应的通用函数""" + if response.status_code != 200: + logger.error(f"❌ 请求失败，状态码: {response.status_code}") + logger.error(f"响应内容: {response.text}") + return + + logger.info("📥 开始接收流式故事内容:") + logger.info("=" * 50) + + for line in response.iter_lines(): + if line: + line_str = line.decode('utf-8').strip() + if line_str.startswith('data:'): + json_data = line_str[5:].strip() + try: + data = json.loads(json_data) + if error_code := data.get("errorCode"): + logger.error(f"❌ 错误 [{error_code}]: {data.get('message')}") + break + + story_data = data.get("data", {}) + logger.info(f"主题: {story_data.get('storyTheme', '无')}") + logger.info(f"内容: {story_data.get('storyContent', '无')}") + logger.info(f"音频URL: {story_data.get('storyAudioUrl', '无')}") + logger.info(f"时长（秒）：{story_data.get('storyDuration', '未知')}") # 新增行 + logger.info("-" * 50) + except json.JSONDecodeError: + logger.error(f"❌ 解析JSON失败: {json_data}") + + logger.info("✅ 故事流接收完成") + + +# -------------------------- 接口测试函数 -------------------------- +@exception_handler +def get_device_token() -> Optional[str]: + """ + 获取设备Token + + 重试机制: 最多重试Config.TOKEN_RETRY_COUNT次 + 返回: 成功返回token字符串，失败返回None + """ + for retry in range(Config.TOKEN_RETRY_COUNT): + try: + timestamp = str(int(time.time() * 1000)) + nonce = ''.join(random.choices(string.ascii_letters + string.digits, k=12)) + signature = generate_signature(Config.DEVICE_ID, Config.DEVICE_ICCID, timestamp, nonce) + + headers = { + "deviceId": Config.DEVICE_ID, + "deviceIccid": Config.DEVICE_ICCID, + "timestamp": timestamp, + "nonce": nonce, + "signature": signature, + "deviceModel": Config.DEVICE_MODEL, + "deviceVersion": Config.DEVICE_VERSION, + "Content-Type": "application/json" + } + + response = requests.get( + Config.TOKEN_URL, + headers=headers, + timeout=Config.NORMAL_TIMEOUT + ) + response_data = response.json() + + if response_data.get("errorCode") is None and "data" in response_data: + logger.info("✅ 成功获取设备Token") + return response_data["data"]["token"] + else: + error_msg = response_data.get("message", "未知错误") + if retry < Config.TOKEN_RETRY_COUNT - 1: + logger.warning(f"⚠️ 获取Token失败(将重试 {Config.TOKEN_RETRY_COUNT - retry - 1} 次): {error_msg}") + time.sleep(1) # 重试前等待1秒 + else: + logger.error(f"❌ 获取Token失败: {error_msg}") + return None + + except Exception as e: + if retry < Config.TOKEN_RETRY_COUNT - 1: + logger.warning(f"⚠️ 获取Token异常(将重试 {Config.TOKEN_RETRY_COUNT - retry - 1} 次): {str(e)}") + time.sleep(1) + else: + raise e # 由装饰器捕获处理 + + return None + + +@exception_handler +def test_choose_story(token: str): + """测试智能选择故事接口（SSE流式响应）""" + if not token: + logger.error("❌ Token为空，无法调用接口") + return + + # 检查音频文件是否存在 + if not Config.AUDIO_FILE_PATH.exists(): + logger.error(f"❌ 音频文件未找到: {Config.AUDIO_FILE_PATH.resolve()}") + return + + try: + headers = create_sse_headers(token) + + with open(Config.AUDIO_FILE_PATH, 'rb') as audio_file: + files = { + 'file': (Config.AUDIO_FILE_PATH.name, audio_file, 'audio/mpeg') + } + + logger.info("📤 发送智能选择故事请求（SSE）...") + with requests.post( + Config.CHOOSE_STORY_URL, + headers=headers, + files=files, + stream=True, + timeout=Config.SSE_TIMEOUT + ) as response: + process_sse_stream(response) + + except requests.exceptions.Timeout: + logger.error(f"❌ 请求超时（超过{Config.SSE_TIMEOUT}秒）") + + +@exception_handler +def test_random_story(token: str, story_theme: Optional[str] = None): + """ + 测试随机故事接口（SSE流式响应） + + 参数: + token: 设备令牌 + story_theme: 可选故事主题，None则随机 + """ + if not token: + logger.error("❌ Token为空，无法调用接口") + return + + try: + headers = create_sse_headers(token) + payload = {"storyTheme": story_theme} if story_theme else {} + + logger.info("📤 发送随机故事请求（SSE）...") + with requests.post( + Config.RANDOM_STORY_URL, + headers=headers, + json=payload, + stream=True, + timeout=Config.SSE_TIMEOUT + ) as response: + process_sse_stream(response) + + except requests.exceptions.Timeout: + logger.error(f"❌ 请求超时（超过{Config.SSE_TIMEOUT}秒）") + + +@exception_handler +def test_exit_sleep_inquiry(token: str): + """测试获取退出哄睡询问接口""" + if not token: + logger.error("❌ Token为空，无法调用接口") + return + + headers = create_auth_headers(token) + logger.info("📤 发送获取退出哄睡询问请求...") + + response = requests.post( + Config.EXIT_INQUIRY_URL, + headers=headers, + timeout=Config.NORMAL_TIMEOUT + ) + response_data = response.json() + + logger.info("=" * 50) + logger.info("【响应结果】") + logger.info(json.dumps(response_data, ensure_ascii=False, indent=2)) + + if response_data.get("errorCode") is None: + data = response_data.get("data") + if data and isinstance(data, dict): + logger.info("\n✅ 获取退出哄睡询问成功") + logger.info(f"询问文本: {data.get('textContent', '无')}") + logger.info(f"意图类型: {data.get('intention', '无')}") + logger.info(f"音频数据长度: {len(data.get('audioContent', ''))} 字符（base64）") + else: + logger.warning("\n⚠️ 获取退出哄睡询问成功，但返回数据为空") + logger.warning("提示: 可能需要先调用故事接口（选项1或2）进入哄睡模式后再尝试") + else: + logger.error(f"\n❌ 请求失败: {response_data.get('message', '未知错误')}") + + +@exception_handler +def test_finish_sleep_mode(token: str): + """测试确认结束哄睡模式接口""" + if not token: + logger.error("❌ Token为空，无法调用接口") + return + + headers = create_auth_headers(token) + logger.info("📤 发送确认结束哄睡模式请求...") + + response = requests.post( + Config.FINISH_SLEEP_URL, + headers=headers, + timeout=Config.NORMAL_TIMEOUT + ) + response_data = response.json() + + logger.info("=" * 50) + logger.info("【响应结果】") + logger.info(json.dumps(response_data, ensure_ascii=False, indent=2)) + + if response_data.get("errorCode") is None: + logger.info("\n✅ 确认结束哄睡模式成功") + else: + logger.error(f"\n❌ 请求失败: {response_data.get('message')}") + + +# -------------------------- 主流程 -------------------------- +def main(): + # 预处理检查 + if Config.AUDIO_FILE_PATH.name in str(Config.CHOOSE_STORY_URL) and not Config.AUDIO_FILE_PATH.exists(): + logger.warning(f"⚠️ 检测到智能选择故事接口依赖的音频文件不存在: {Config.AUDIO_FILE_PATH.resolve()}") + + # 获取Token + token = get_device_token() + if not token: + return + + logger.info("\n" + "=" * 60) + logger.info("请选择要测试的接口：") + logger.info("1 - 3.1 智能选择故事接口（需要音频文件）") + logger.info("2 - 3.2 随机故事接口") + logger.info("3 - 3.3 获取退出哄睡询问接口") + logger.info("4 - 3.4 确认结束哄睡模式接口") + logger.info("0 - 退出") + logger.info("=" * 60) + + # 输入验证 + while True: + choice = input("请输入选项（0-4）：").strip() + if choice in ["0", "1", "2", "3", "4"]: + break + logger.warning("❌ 无效选项，请输入0-4之间的数字") + + if choice == "1": + test_choose_story(token) + elif choice == "2": + theme = input("可选：输入故事主题（如'童话故事'，直接回车则随机）：").strip() + test_random_story(token, theme if theme else None) + elif choice == "3": + test_exit_sleep_inquiry(token) + elif choice == "4": + test_finish_sleep_mode(token) + elif choice == "0": + logger.info("退出程序") + return + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/code/sdk/zhihui/leiniao/leiniao_python_test/story_request.mp3 b/code/sdk/zhihui/leiniao/leiniao_python_test/story_request.mp3 new file mode 100644 index 000000000..128e8dc6b Binary files /dev/null and b/code/sdk/zhihui/leiniao/leiniao_python_test/story_request.mp3 differ diff --git a/code/sdk/zhihui/leiniao/leiniao_python_test/test_audio.mp3 b/code/sdk/zhihui/leiniao/leiniao_python_test/test_audio.mp3 new file mode 100644 index 000000000..779a03877 Binary files /dev/null and b/code/sdk/zhihui/leiniao/leiniao_python_test/test_audio.mp3 differ diff --git a/code/sdk/zhihui/leiniao/leiniao_python_test/upgrade_packages/upgrade_2.0.0.bin b/code/sdk/zhihui/leiniao/leiniao_python_test/upgrade_packages/upgrade_2.0.0.bin new file mode 100644 index 000000000..f8c0a7271 Binary files /dev/null and b/code/sdk/zhihui/leiniao/leiniao_python_test/upgrade_packages/upgrade_2.0.0.bin differ