--- /dev/null
+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
--- /dev/null
+eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJXWVNaIiwiZGV2aWNlTW9kZWwiOiJhYm15Ym94XzEiLCJkZXZpY2VWZXJzaW9uIjoiMi4wLjAiLCJkZXZpY2VJZCI6IkRFVklDRV8wMDEiLCJleHAiOjE3NTg4NTE0NDV9.joDdP7owmZf51cwaY16g9VM-J0Egdi6lt-4X7ZqNid0
\ No newline at end of file
--- /dev/null
+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
--- /dev/null
+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
--- /dev/null
+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
--- /dev/null
+
+# 本地起一个 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())
--- /dev/null
+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()
--- /dev/null
+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
projects / mirrors / FD230.git / commitdiff