Skip to content

U
ubains-module-test
提交 ff749e33 authored 作者: 陈泽健's avatar 陈泽健
浏览文件
操作

feat(device-sim): 门口屏模拟器启动时调用 getTokenInfoByToken 获取 token 

新增 DoorTokenClient（固定 X-SIGN + regUdid 派生 + 响应解析），
TokenPool 登记池保证批量场景 appToken 不重复（来自导入授权码，不可自动生成）；
DoorSimulator.start() 接入 token 获取并回填 topic_params（company_id/room_id/conference）；
base_simulator / create_simulator / start_simulator 透传 token_api_host（来源 EnvConfig）
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 0e4ead5b
显示空白字符变更
内嵌 并排
正在显示 包含 1297 行增加 和 67 行删除
Docs/PRD/需求文档/设备模拟/_PRD_需求优化_门口屏模拟器Token获取.md 0 → 100644
浏览文件 @ ff749e33
# PRD 需求优化文档 — 门口屏模拟器 Token 获取
> **文档类型**: PRD 需求优化文档
> **创建日期**: 2026-08-19
> **作者**: czj
> **优先级**: P0
> **状态**: 待评审
> **关联文档**: `_PRD_需求文档_设备模拟模块.md`、`_执行计划_门口屏模拟器Token获取.md`
---
## 一、需求背景
### 1.1 当前问题
门口屏（DoorSimulator）模拟器当前启动流程为：连接 MQTT Broker → 发布注册消息 → 订阅主题 → 发布初始状态 → 启动周期上报。**整个过程不涉及任何 HTTP 接口调用**。
但在真实 Android 门口屏设备（UbGview 应用）中，应用启动后的**第一个动作**是调用 HTTP 接口 `POST /exapi/api-android/token/getTokenInfoByToken` 获取 token，成功后才会进行后续的 MQTT 连接、主题订阅和数据上报。
当前模拟器与真实设备存在以下差距：
| 对比项 | 真实设备 | 当前模拟器 |
|--------|---------|-----------|
| 启动第一步 | HTTP 调用获取 token | 直接连接 MQTT |
| token 管理 | 从接口响应获取 Authorization + token 对象 | 无 token 概念 |
| company_id / room_id | 从接口响应动态获取（companyNumber / cnum） | 手动在 topic_params 中配置 |
| 签名机制 | X-SIGN / X-TIMESTAMP / X-RANDOM | 无 |
### 1.2 目标用户
| 用户角色 | 使用场景 | 核心需求 |
|---------|---------|---------|
| 测试工程师 | 模拟门口屏设备启动 | 模拟真实设备启动流程，先调用 token 接口 |
| 测试工程师 | 多设备并发模拟 | 每个设备独立获取 token，regUdid 不重复 |
| 开发工程师 | 调试 token 获取逻辑 | 查看接口调用日志，验证 token 获取是否成功 |
### 1.3 需求目标
1. **启动流程补全**：门口屏模拟器启动时，第一步调用 `getTokenInfoByToken` 接口获取 token
2. **Token 持久化**：成功获取 token 后保存 Authorization 及响应中的必要字段
3. **regUdid 唯一性**：多设备场景下 regUdid 不能重复，由 device_id 派生
4. **参数回填**：将接口返回的 companyNumber → company_id、cnum → room_id 存入 topic_params
### 1.4 本期范围与排除
**本期范围：**
| 模块 | 说明 |
|------|------|
| Token 接口调用 | 启动时调用 `getTokenInfoByToken`，使用日志中的固定 X-SIGN |
| 响应解析与保存 | 解析 Authorization、token、conference、company 信息并保存 |
| regUdid 唯一性 | 多设备场景下 regUdid 由 device_id 派生，保证不重复 |
| 参数回填 | 将接口返回的 companyNumber / cnum 回填到 topic_params |
**本期不涉及（后续迭代）：**
| 项目 | 说明 |
|------|------|
| MQTT 主题解析 | 根据回填后的 company_id + room_id 解析 MQTT 主题 |
| 会议服务主题订阅 | 订阅 `/iot/v1/conference/service/request/{company_id}/{room_id}/` 等主题 |
| MQTT 周期上报 | 启动 MQTT 周期上报任务 |
| 动态 X-SIGN 算法 | 日志中固定 X-SIGN 的生成算法（需逆向 APK 确认） |
---
## 二、功能需求
### 2.1 接口规格
#### 2.1.1 基本信息
| 项目 | 值 |
|------|-----|
| 接口 URL | `POST https://{host}/exapi/api-android/token/getTokenInfoByToken` |
| 请求头 | Content-Type: application/json; charset=utf-8 |
| 请求头 | X-SIGN: 固定值（从日志中获取） |
| 请求头 | X-TIMESTAMP: 当前 Unix 秒级时间戳 |
| 请求头 | X-RANDOM: 16 位随机字符串 |
#### 2.1.2 请求体
```json
{
"appToken": "AND-34H-0101",
"regUdid": "{device_id 派生值}"
}
```
| 字段 | 说明 | 来源 |
|------|------|------|
| appToken | 应用令牌 | 固定值 `AND-34H-0101`（从日志中获取） |
| regUdid | 设备注册唯一标识 | 由 device_id 派生，多设备不重复 |
#### 2.1.3 响应体（成功）
```json
{
"code": 200,
"message": "OPERATION_SUCCESSFUL",
"success": true,
"result": {
"Authorization": "Bearer eyJhbGciOiJIUzI1NiJ9...",
"token": {
"tokenId": 7186,
"companyNumber": "CN-34H-UBAINS",
"appToken": "AND-34H-0101",
"regUdid": "79f18c1a9a3bcfb1",
"cnum": "upjp7gzkjmgsu15y06aoahff55xh6c53",
"startTime": "2026-04-28 17:34:38",
"endTime": "2099-12-31 23:59:59",
"state": 1,
"annotation": "彭甘宇在2026-05-08 02:02:17绑定了 设备"
},
"conference": {
"conferenceId": 50,
"conferenceName": "PGY测试会议室1",
"conferenceNumber": "upjp7gzkjmgsu15y06aoahff55xh6c53",
"qrCode": "https://192.168.5.48/group1/M00/00/00/wKgFMGn9rkqAb4dYAAAlXncPlGo276.png",
"generalField": "{...}"
},
"company": {
"companyId": 466,
"companyName": "马来西亚项目测试",
"companyNumber": "CN-34H-UBAINS",
"ubainsKey": "ubains76zu81x6vu",
"ubainsSecret": "gKv2nq2O-nPA9S3yp-SPJeRxVH-N9HBSF4y"
}
}
}
```
#### 2.1.4 字段映射规则
| 响应字段 | 映射目标 | 说明 |
|---------|---------|------|
| `result.Authorization` | token_authorization | Bearer JWT，MQTT 和后续接口使用 |
| `result.token.tokenId` | token_id | Token 标识 |
| `result.token.companyNumber` | topic_params.company_id | 公司编号（如 `CN-34H-UBAINS`） |
| `result.token.cnum` | topic_params.room_id | 会议室编号（32 位字符串，非旧格式短编号） |
| `result.token.appToken` | topic_params.app_token | 应用令牌 |
| `result.token.regUdid` | token_reg_udid | 设备注册标识 |
| `result.conference` | token_conference | 会议信息（conferenceId, conferenceName 等） |
| `result.company` | token_company | 公司信息（companyId, companyName, ubainsKey, ubainsSecret） |
### 2.2 签名机制（本期）
#### 2.2.1 当前方案
由于动态 X-SIGN 生成算法未知（需要逆向 APK 确认），**本期使用日志中已验证通过的固定值**：
| 请求头 | 值 |
|--------|-----|
| X-SIGN | `uEkHrug...`（从日志中获取的固定值，已验证可调用成功） |
| X-TIMESTAMP | 当前 Unix 秒级时间戳（动态生成，跟随当前时间） |
| X-RANDOM | 16 位随机字符串 |
**注意**：由于 X-SIGN 是固定值，与 X-TIMESTAMP / X-RANDOM 的校验关系未知。当前方案的前提是服务端**不校验** X-SIGN 与时间戳/随机数的绑定关系，或该校验已关闭（测试环境特性）。
#### 2.2.2 后续计划
| 阶段 | 内容 | 优先级 |
|------|------|--------|
| 本期 | 使用日志固定 X-SIGN | P0 |
| 后续 | 逆向 APK 确认签名算法，实现动态生成 | P2 |
### 2.3 regUdid 唯一性
#### 2.3.1 规则定义
**约束**：多设备场景下，regUdid 不能重复（目标环境会校验唯一性）。
**策略**：regUdid 由 device_id 派生，确保每个设备有独立的 regUdid。
```
regUdid = f"{device_id}_{random_suffix}"
```
| 场景 | device_id | regUdid 示例 |
|------|-----------|-------------|
| 单设备（默认） | DS001 | DS001_a3b7k9m2 |
| 多设备 | DS001 | DS001_a3b7k9m2 |
| 多设备 | DS002 | DS002_x5p8q1w4 |
| 多设备 | DS003 | DS003_j2n6r4t8 |
#### 2.3.2 生成逻辑
1. 取 `device_id` 作为前缀（如 `DS001`）
2. 拼接 `_` + 8 位随机字母数字
3. 最终格式：`{device_id}_{8位随机}`
4. 每次启动时重新生成（固定后缀可能导致服务端缓存冲突）
### 2.4 Token 存储与持久化
#### 2.4.1 存储字段
| 字段 | 类型 | 说明 |
|------|------|------|
| token_authorization | str | Bearer JWT |
| token_id | int | token 记录 ID |
| token_reg_udid | str | 设备注册标识 |
| token_conference | dict | 会议信息（JSON 对象） |
| token_company | dict | 公司信息（JSON 对象） |
#### 2.4.2 存储位置
| 存储方式 | 用途 | 说明 |
|---------|------|------|
| 模拟器实例属性 | 运行时使用 | 直接保存在 DoorSimulator 实例中 |
| topic_params（部分字段） | 参数回填 | company_id / room_id / app_token 回填到 topic_params |
| 数据库（可选） | 持久化 | 可选保存到 DeviceSimulator 记录的 extra_attrs 字段 |
### 2.5 启动流程变更
#### 2.5.1 当前流程
```
DoorSimulator.start()
├── 连接 MQTT Broker
├── 发布注册消息（reboot_response）
├── 订阅全部 publish topics
├── 发布初始状态
└── 启动周期上报（_auto_report_loop）
```
#### 2.5.2 改造后流程
```
DoorSimulator.start()
├── 第一步：HTTP 调用 getTokenInfoByToken
│ ├── 构建请求头（X-SIGN 固定值，X-TIMESTAMP 当前秒，X-RANDOM 16位随机）
│ ├── 构建请求体（appToken 固定值，regUdid 由 device_id 派生）
│ ├── 发送 POST 请求
│ ├── 解析响应（code=200, success=true）
│ ├── 保存 Authorization / token / conference / company
│ ├── 回填 topic_params（company_id, room_id, app_token）
│ └── 失败则抛出异常，启动终止
├── 连接 MQTT Broker（使用 MQTT 配置）
├── 发布注册消息（reboot_response）
├── 订阅全部 publish topics
├── 发布初始状态
└── 启动周期上报（_auto_report_loop） ← 本期不实现
```
**本期改造范围**：仅实现 `第一步` 的 HTTP 调用 + token 保存 + 参数回填。
**MQTT 相关步骤**（连接/注册/订阅/状态发布/周期上报）：保持现有逻辑不变，但需注意启动顺序（token 获取成功后，下一期才接入 MQTT 步骤）。
---
## 三、技术方案
### 3.1 后端改动
#### 3.1.1 新增 HTTP 客户端
在 `backend/app/simulators/` 目录下新增 `door_token_client.py`：
| 方法 | 说明 |
|------|------|
| `DoorTokenClient.__init__(host, device_id, app_token)` | 初始化客户端 |
| `DoorTokenClient.get_token()` | 调用 getTokenInfoByToken 接口 |
| `DoorTokenClient._build_headers()` | 构造请求头（X-SIGN 固定值, X-TIMESTAMP, X-RANDOM） |
| `DoorTokenClient._build_body()` | 构造请求体（appToken, regUdid） |
| `DoorTokenClient._generate_reg_udid()` | 由 device_id 生成唯一 regUdid |
| `DoorTokenClient._parse_response()` | 解析响应，提取 Authorization / token / conference / company |
#### 3.1.2 修改 DoorSimulator
在 `DoorSimulator.start()` 方法头部增加 token 获取逻辑：
| 改动 | 说明 |
|------|------|
| `start()` 方法开头 | 调用 `DoorTokenClient.get_token()` |
| 新增属性 `_token_info` | 保存 token 响应信息 |
| `topic_params` 回填 | 将 companyNumber → company_id, cnum → room_id, appToken → app_token |
#### 3.1.3 配置项
在 `EnvConfig` 或 `DeviceSimulator` 中增加：
| 配置项 | 说明 | 默认值 |
|--------|------|--------|
| `token_api_host` | Token 接口主机地址 | 从环境配置中获取（目标系统 IP） |
| `app_token` | 应用令牌 | `AND-34H-0101` |
| `x_sign_fixed` | 固定 X-SIGN 值 | 从日志中获取 |
| `token_enabled` | 是否启用 token 获取 | `true` |
### 3.2 异常处理
#### 3.2.1 失败场景
| 场景 | 处理方式 |
|------|---------|
| 网络不可达（连接拒绝/超时） | 记录错误日志，抛出异常，启动终止 |
| HTTP 非 200 状态码 | 记录错误日志，抛出异常，启动终止 |
| 响应 `success: false` | 记录错误日志（含 message），抛出异常，启动终止 |
| 响应缺少必需字段 | 记录错误日志，抛出异常，启动终止 |
#### 3.2.2 重试策略
| 策略 | 值 |
|------|-----|
| 最大重试次数 | 3 次 |
| 重试间隔 | 2 秒（指数退避） |
| 重试条件 | 仅网络错误可重试，业务错误不重试 |
---
## 四、非功能需求
### 4.1 性能要求
- Token 接口调用耗时 < 5s（含网络延迟）
- 启动总耗时增加 < 10s（含 3 次重试）
- 不影响已启动设备的运行性能
### 4.2 兼容性要求
- 向后兼容：已创建的门禁设备升级后，启动时自动获取 token
- 单设备兼容：单设备场景下 regUdid 唯一性不影响
- 多设备兼容：多设备场景下 regUdid 自动派生，无需手动配置
### 4.3 可扩展性
- 签名算法扩展：后续支持动态 X-SIGN 生成时，替换 `_build_headers()` 方法即可
- 设备类型扩展：其他设备类型（如无纸化屏）需要 token 时，可复用 `DoorTokenClient`
---
## 五、验收标准
### 5.1 功能验收
| 编号 | 验收项 | 预期结果 |
|------|--------|---------|
| F01 | 门口屏启动时调用 token 接口 | 日志显示 `POST /exapi/api-android/token/getTokenInfoByToken` 调用记录 |
| F02 | 接口调用成功 | 响应 `code=200, success=true`，Authorization 已保存 |
| F03 | Token 信息保存 | 模拟器实例中可访问 `_token_info`，包含 Authorization / token / conference / company |
| F04 | topic_params 回填 | `company_id` 和 `room_id` 自动回填为接口返回的值 |
| F05 | regUdid 唯一性 | 多设备启动时，各设备的 regUdid 不同（前缀不同） |
| F06 | 接口调用失败 | 启动终止，日志显示错误原因，设备状态为 `error` |
### 5.2 边界验收
| 编号 | 边界场景 | 预期结果 |
|------|---------|---------|
| B01 | 网络不可达 | 3 次重试后失败，启动终止 |
| B02 | 接口返回 `success: false` | 立即失败，不重试，启动终止 |
| B03 | 响应缺少 Authorization 字段 | 解析异常，启动终止 |
| B04 | 单设备启动 | regUdid 正常生成，token 获取成功 |
| B05 | 多设备（10 台）同时启动 | 各设备独立获取 token，regUdid 均不重复 |
---
## 六、附录
### 6.1 参考文档
- 真实门口屏设备日志（UbGview 应用，2026-08-19）
- `_PRD_需求文档_设备模拟模块.md`
- `backend/app/simulators/base_simulator.py` — BaseSimulator 启动流程
- `backend/app/simulators/door_simulator.py` — DoorSimulator 实现
- `backend/app/executors/http_client.py` — 现有签名算法（与日志不匹配，仅供参考）
### 6.2 术语说明
| 术语 | 说明 |
|------|------|
| X-SIGN | 请求签名，本期使用日志固定值 |
| X-TIMESTAMP | 请求时间戳，秒级 Unix 时间戳 |
| X-RANDOM | 请求随机数，16 位随机字符串 |
| regUdid | 设备注册唯一标识，需保证多设备不重复 |
| topic_params | 设备模拟器的主题动态参数，存储在 DeviceSimulator 记录中 |
| companyNumber | 接口返回的公司编号，映射为 topic_params.company_id |
| cnum | 接口返回的会议室编号，映射为 topic_params.room_id |
### 6.3 已确认的决策记录
| 决策 | 内容 | 依据 |
|------|------|------|
| X-SIGN 使用固定值 | 本期复用日志中已验证通过的 X-SIGN | 动态算法未知，日志固定值已可调用成功 |
| regUdid 由 device_id 派生 | 格式 `{device_id}_{8位随机}` | 多设备场景下不能重复 |
| 只保存 token，不做 MQTT 操作 | 本期仅获取并保存 token | 用户明确要求 |
| 使用日志中的 appToken | `AND-34H-0101` | 与 X-SIGN 配套 |
---
*文档结束*
\ No newline at end of file
Docs/PRD/需求文档/设备模拟/_执行计划_门口屏模拟器Token获取.md 0 → 100644
浏览文件 @ ff749e33
# 执行计划 — 门口屏模拟器 Token 获取
> **文档类型**: 执行计划文档
> **创建日期**: 2026-08-19
> **作者**: czj
> **关联文档**: `_PRD_需求优化_门口屏模拟器Token获取.md`、`_PRD_需求文档_设备模拟模块.md`
> **预计工期**: 2.5 天
---
## 一、执行概述
### 1.1 项目背景
真实门口屏设备（UbGview 应用）启动后第一个动作是调用 HTTP 接口 `POST /exapi/api-android/token/getTokenInfoByToken` 获取 token。本计划为门口屏模拟器补充该接口调用链，使模拟器启动流程与真实设备一致——先获取 token，为后续 MQTT 主题解析和上报做准备。
### 1.2 执行目标
1. 门口屏模拟器启动时第一步调用 `getTokenInfoByToken` 接口
2. 使用日志中的固定 X-SIGN（已验证可调用成功）
3. 多设备场景下 regUdid 由 device_id 派生，保证唯一
4. 成功获取并保存 Authorization / token / conference / company 信息
5. 将 companyNumber / cnum 回填到 topic_params
### 1.3 范围界定
| 范围 | 内容 |
|------|------|
| ✅ 本期执行 | HTTP 接口调用、签名头构造、regUdid 生成、响应解析、token 保存、参数回填 |
| ❌ 本期排除 | MQTT 主题解析、会议服务主题订阅、MQTT 周期上报、动态 X-SIGN 算法 |
---
## 二、任务分解与实施计划
### 阶段 1：接口调研与数据确认（0.5 天）
| # | 任务 | 说明 | 产出 |
|---|------|------|------|
| 1.1 | 确认接口调用细节 | 复核日志中的 URL / Header / Body / Response 结构 | 接口调用手顺（已确认：HTTP 200 + OPERATION_SUCCESSFUL） |
| 1.2 | 确认目标环境地址 | 从 EnvConfig 中获取目标系统 host（如 192.168.5.48） | host 配置项 |
| 1.3 | 确认 X-SIGN 固定值 | 从日志中提取固定 X-SIGN 作为常量 | 常量定义 |
| 1.4 | 确认 appToken 固定值 | 从日志 body 中提取 `AND-34H-0101` | 常量定义 |
**阶段 1 完成标准**：接口调用所需的最小常量集合（host / X-SIGN / appToken）已确认并记录。
### 阶段 2：Token 客户端实现（1 天）
| # | 任务 | 说明 | 产出 |
|---|------|------|------|
| 2.1 | 新增 `DoorTokenClient` 类 | 文件：`backend/app/simulators/door_token_client.py` | 客户端类 |
| 2.2 | 实现 `_build_headers()` | X-SIGN 固定值 + X-TIMESTAMP（秒级）+ X-RANDOM（16 位） | 请求头构造 |
| 2.3 | 实现 `_generate_reg_udid()` | `{device_id}_{8位随机}`，保证多设备唯一 | regUdid 生成 |
| 2.4 | 实现 `get_token()` | POST 调用 + 超时控制（5s）+ 重试（3 次, 指数退避） | 接口调用 |
| 2.5 | 实现 `_parse_response()` | 解析 Authorization / token / conference / company | 响应解析 |
| 2.6 | 实现 `_get_mqtt_config()`（预留） | 解析日志中的 mqttSavedConfig（本期仅预留，不做订阅） | 预留方法 |
**阶段 2 完成标准**：`DoorTokenClient` 可独立调用接口成功，单元验证返回完整 token 数据。
### 阶段 3：模拟器启动流程集成（0.5 天）
| # | 任务 | 说明 | 产出 |
|---|------|------|------|
| 3.1 | `DoorSimulator.start()` 接入 token 获取 | 在 start() 最前面调用 `DoorTokenClient.get_token()` | 启动流程改造 |
| 3.2 | 保存 token 信息 | 新增 `self._token_info` 属性保存完整响应 | token 保存 |
| 3.3 | topic_params 回填 | company_id ← companyNumber, room_id ← cnum, app_token ← appToken | 参数回填 |
| 3.4 | 失败处理 | token 获取失败 → 记录错误日志 → 启动终止 → 设备状态 error | 异常处理 |
**阶段 3 完成标准**：启动门口屏设备，日志显示接口调用成功，token 已保存，topic_params 已回填。
### 阶段 4：验证与部署（0.5 天）
| # | 任务 | 说明 | 产出 |
|---|------|------|------|
| 4.1 | 单设备验证 | 创建 1 台门口屏设备，启动，确认 token 获取成功 | 验证记录 |
| 4.2 | 多设备验证 | 创建 3-5 台门口屏设备，启动，确认 regUdid 均不重复 | 验证记录 |
| 4.3 | 失败场景验证 | 模拟网络不可达，确认重试 3 次后启动终止 | 验证记录 |
| 4.4 | 回归验证 | 其他设备类型（paperless/central）启动不受影响 | 回归记录 |
| 4.5 | 部署 | 提交代码，部署到测试服务器（192.168.5.60 双容器） | 部署完成 |
**阶段 4 完成标准**：单设备 + 多设备 + 失败场景全部验证通过，线上部署完成。
---
## 三、验收标准
### 3.1 功能验收
| 编号 | 验收项 | 预期结果 |
|------|--------|---------|
| F01 | 门口屏启动时调用 token 接口 | 日志显示 `POST /exapi/api-android/token/getTokenInfoByToken` 记录 |
| F02 | 接口调用成功 | `code=200, success=true`，Authorization 已保存 |
| F03 | token 信息保存 | `_token_info` 包含 Authorization / token / conference / company |
| F04 | topic_params 回填 | company_id / room_id 自动回填为接口返回值 |
| F05 | regUdid 唯一性 | 多设备启动时 regUdid 各不相同 |
| F06 | 接口调用失败 | 启动终止，日志显示错误原因，设备状态 error |
### 3.2 边界验收
| 编号 | 边界场景 | 预期结果 |
|------|---------|---------|
| B01 | 网络不可达 | 3 次重试后失败，启动终止 |
| B02 | 接口返回 success: false | 立即失败，不重试 |
| B03 | 响应缺少 Authorization | 解析异常，启动终止 |
| B04 | 单设备启动 | regUdid 正常生成，token 获取成功 |
| B05 | 多设备（10 台）同时启动 | 各设备独立获取 token，regUdid 均不重复 |
---
## 四、测试计划
| 测试类型 | 范围 | 方式 |
|---------|------|------|
| 单元测试 | `DoorTokenClient` 的 header/body/regUdid/解析逻辑 | pytest，mock 网络 |
| 集成测试 | 真实调用目标环境接口 | 手工验证（已验证 HTTP 200 可调用成功） |
| 手动验证 | 单设备 / 多设备 / 失败场景 | 前端界面操作 + 后端日志 |
---
## 五、风险评估
| 风险 | 等级 | 影响 | 应对措施 |
|------|------|------|---------|
| 固定 X-SIGN 后续失效 | 中 | 接口调用 401 | 短期限固定值，后续逆向 APK 生成动态签名 |
| X-SIGN 与时间戳校验 | 中 | 当前时间戳可能导致签名失效 | 已实际调用验证成功，若失效回退用日志时间戳 |
| 服务端 regUdid 唯一校验严格 | 低 | 多设备启动部分失败 | regUdid 由 device_id 派生 + 随机后缀，冲突概率极低 |
| 目标环境网络不可达 | 中 | 启动失败 | 重试 3 次 + 明确错误日志 |
| 多设备并发调用目标环境 | 中 | 目标环境压力 | 每设备独立请求，接口轻量，风险可控 |
---
## 六、实施记录
| 日期 | 任务 | 状态 | 备注 |
|------|------|------|------|
| 2026-08-19 | 梳理需求，确认决策 | ✅ 完成 | 与用户确认 X-SIGN 固定值、regUdid 派生、只存 token |
| 2026-08-19 | 接口实际调用验证 | ✅ 完成 | HTTP 200 + OPERATION_SUCCESSFUL，token 完整返回 |
| 2026-08-19 | 撰写 PRD 需求文档 | ✅ 完成 | `_PRD_需求优化_门口屏模拟器Token获取.md` |
| 2026-08-19 | 撰写执行计划 | ⏳ 进行中 | 本文档 |
| - | 阶段 1 接口调研与数据确认 | ⬜ 待执行 | |
| - | 阶段 2 Token 客户端实现 | ⬜ 待执行 | |
| - | 阶段 3 模拟器启动流程集成 | ⬜ 待执行 | |
| - | 阶段 4 验证与部署 | ⬜ 待执行 | |
---
## 七、后续工作（不在本期）
| 后续项 | 说明 | 优先级 |
|--------|------|--------|
| MQTT 主题解析 | 根据回填后的 company_id + room_id 解析 MQTT 主题 | P1 |
| 会议服务主题订阅 | 订阅 `/iot/v1/conference/service/request/{company_id}/{room_id}/` | P1 |
| MQTT 周期上报 | 启动周期上报任务 | P1 |
| 动态 X-SIGN 算法 | 逆向 APK 确认签名算法，替换固定值 | P2 |
| 其他设备类型接入 | 无纸化屏等设备复用 TokenClient | P2 |
---
## 八、附录
### 8.1 涉及代码文件
| 文件 | 操作 | 说明 |
|------|------|------|
| `backend/app/simulators/door_token_client.py` | 新增 | Token 获取客户端 |
| `backend/app/simulators/door_simulator.py` | 修改 | start() 接入 token 获取 |
| `backend/app/config.py` | 修改（可选） | token 接口相关配置项 |
### 8.2 接口调用手顺（已验证）
```
POST https://192.168.5.48/exapi/api-android/token/getTokenInfoByToken
Headers:
Content-Type: application/json; charset=utf-8
X-SIGN: {日志固定值}
X-TIMESTAMP: {当前秒级时间戳}
X-RANDOM: {16 位随机}
Body:
{"appToken": "AND-34H-0101", "regUdid": "{device_id}_{8位随机}"}
Response:
code=200, success=true
result.Authorization → 保存
result.token.companyNumber → topic_params.company_id
result.token.cnum → topic_params.room_id
```
---
*文档结束*
\ No newline at end of file
HANDOFF_设备模拟.md
浏览文件 @ ff749e33
...@@ -9,6 +9,71 @@ ...@@ -9,6 +9,71 @@
## 📊 会话进度记录 ## 📊 会话进度记录
### 2026-08-19 会话 E：门口屏模拟器 Token 获取（已实现 + 真实调用验证通过，未提交未部署）
**会话目标**：为门口屏（door）模拟器补充启动时的真实 HTTP 接口调用链 —— 模拟真实 Android 门口屏设备（UbGview 应用）启动第一步调用的 `POST /exapi/api-android/token/getTokenInfoByToken` 接口获取 token。
**状态**：✅ 代码实现完成 + 真实接口调用验证通过（HTTP 200 + 完整 token）+ 登记池 5 场景验证通过；⚠️ 未提交 git、未部署
**背景**：真实门口屏设备启动第一步是调 token 接口，成功后才有 Authorization 继续进行后续交互；而模拟器之前是直接连 MQTT，且 company_id / room_id 依赖手动配置。
**本期只做**：启动时调用 token 接口 → 保存 token → 回填 topic_params。
**明确不做**（后续迭代）：MQTT 主题解析、会议服务主题订阅、MQTT 周期上报、动态 X-SIGN 算法。
---
#### ① 新增 Token 客户端（`backend/app/simulators/door_token_client.py`）
- **接口**：`POST {scheme}://{host}/exapi/api-android/token/getTokenInfoByToken`（默认 https；host 自动识别协议，兼容 `https://` / `http://` / `host:port` 等写法）
- **请求头**：X-SIGN 用日志固定值 `X_SIGN_FIXED`（动态算法未复现，实测可调用成功）；X-TIMESTAMP 秒级；X-RANDOM 16 位随机
- **请求体**：`{"appToken": ..., "regUdid": ...}`
- **regUdid**：由 device_id 确定性派生 `reg_{md5(device_id)[:16]}` —— 批量不重复、同设备多次启动结果一致（格式与真实日志 16 位 hex 一致）
- **appToken 登记池**（用户明确的关键决策）：appToken **不自动生成**，来自设备导入时提供的授权码。`TokenPool.allocate()` 只做登记与去重：
- 未提供 → 报错"门口屏 token 获取必须使用导入设备时提供的服务端授权码，不可自动生成"
- 重复使用 → 报错"appToken 已被其他设备使用，批量场景授权码不能重复"
- **响应解析（勘误要点）**：`code` 是**字符串 "200"**；`conference` / `company` **嵌套在 `result.token` 对象内部**（不在 result 顶层）
- **重试**：网络错误（ConnectionError/Timeout）指数退避重试 3 次；业务错误（ValueError）不重试
- **host 来源**：`EnvConfig.default_topic_params['token_api_host']`，回退 `broker_host`
#### ② 启动流程集成（4 个后端文件）
- `DoorSimulator.start()` 改造为三步：
1. 调 token 接口（登记 appToken → `DoorTokenClient.get_token()` → 保存 `self._token_info`）
2. 回填 `topic_params`：companyNumber→company_id、cnum→room_id、conferenceId→conference_id、conferenceName→conference_name、appToken→app_token
3. 重新解析主题（`_resolve_topics()`）后走父类 MQTT 启动流程
- token 获取失败 → 记录错误日志 → `_notify_report` 上报 `token_get` failed 消息 → 启动终止返回 False
- **未配置 `token_api_host` → 仅 warning 跳过 token 获取，不影响启动（向后兼容，非门口屏不受影响）**
#### 验证结果
| 验证项 | 结果 |
|--------|------|
| 真实调用（appToken=AND-34H-0101 + regUdid） | ✅ HTTP 200 + OPERATION_SUCCESSFUL，token 完整返回 |
| 字段回填 | ✅ companyNumber=CN-34H-UBAINS、cnum=upjp7gzk..., conferenceName=PGY测试会议室1 |
| 未提供 appToken | ✅ 明确报错拒绝自动生成 |
| 重复使用同一授权码 | ✅ 拒绝 |
| 不同授权码 | ✅ 正常登记 |
| regUdid 派生唯一性 | ✅ `reg_{md5(device_id)[:16]}` 确定性且唯一 |
**修改文件清单（本次会话）**：
| 文件 | 变更类型 | 说明 |
|------|----------|------|
| `backend/app/simulators/door_token_client.py` | 新增 | DoorTokenClient + TokenPool 登记池 + regUdid 派生 + host 规范化 + 默认 host 选择 |
| `backend/app/simulators/door_simulator.py` | 修改 | start() 接入 token 获取 + `_token_info` + topic_params 回填 |
| `backend/app/simulators/base_simulator.py` | 修改 | `__init__` 增加 `token_api_host` 参数 |
| `backend/app/simulators/__init__.py` | 修改 | `create_simulator()` 透传 `token_api_host` |
| `backend/app/services/device_sim_service.py` | 修改 | `start_simulator()` 从环境配置读取 token_api_host |
| `Docs/PRD/需求文档/设备模拟/_PRD_需求优化_门口屏模拟器Token获取.md` | 新增 | PRD 需求文档 |
| `Docs/PRD/需求文档/设备模拟/_执行计划_门口屏模拟器Token获取.md` | 新增 | 执行计划文档 |
**待办**：
- ⚠️ 本次改动未提交 git（连同此前会话 B/C/D 未提交的设备模拟改动一起），下次会话 `/GitCommit` 提交
- ⚠️ 未部署。需要门口屏环境配置在 `EnvConfig.default_topic_params` 里配 `token_api_host`（如 `https://192.168.5.48`）才会触发 token 获取；批量创建门口屏时 Excel 必须提供"授权码(app_token)"列（每台唯一）
- 后续迭代：动态 X-SIGN 算法（P2）、MQTT 主题解析/订阅/上报（P1）见 `_执行计划_门口屏模拟器Token获取.md` 七、后续工作
---
### 2026-08-19 会话 D：无纸化授权码 + 环境配置弹窗简化 + 批量删除性能修复（已全部完成并部署） ### 2026-08-19 会话 D：无纸化授权码 + 环境配置弹窗简化 + 批量删除性能修复（已全部完成并部署）
**会话目标**：三项设备模拟模块优化 — ①无纸化导入模板加授权码列 ②环境配置弹窗简化 ③100 条设备批量删除卡"处理中"修复 **会话目标**：三项设备模拟模块优化 — ①无纸化导入模板加授权码列 ②环境配置弹窗简化 ③100 条设备批量删除卡"处理中"修复
......
backend/app/services/device_sim_service.py
浏览文件 @ ff749e33
...@@ -797,11 +797,12 @@ class DeviceSimService: ...@@ -797,11 +797,12 @@ class DeviceSimService:
if not config: if not config:
raise ValueError(f"关联环境配置不存在: {simulator.env_config_id}") raise ValueError(f"关联环境配置不存在: {simulator.env_config_id}")
# 先检查是否已在运行（用短锁快速判读，避免 IO 操作占用锁）
with _running_simulators_lock: with _running_simulators_lock:
if device_id in _running_simulators and _running_simulators[device_id].is_running(): if device_id in _running_simulators and _running_simulators[device_id].is_running():
return True return True
# 确保 MQTT 已连接 # 确保 MQTT 已连接（IO 操作，放在锁外，避免批量启动时长锁竞争）
if not mqtt_manager.is_connected(config.id): if not mqtt_manager.is_connected(config.id):
password = "" password = ""
if config.password_encrypted: if config.password_encrypted:
...@@ -824,7 +825,7 @@ class DeviceSimService: ...@@ -824,7 +825,7 @@ class DeviceSimService:
config.status = "connected" config.status = "connected"
config.last_connected_at = datetime.now() config.last_connected_at = datetime.now()
# 创建并启动模拟器，传递多主题配置 # 创建并启动模拟器（IO 操作，放在锁外）
# 兼容旧数据：topics 为空时，从 topic_prefix 自动生成默认映射 # 兼容旧数据：topics 为空时，从 topic_prefix 自动生成默认映射
topics = config.topics topics = config.topics
if not topics: if not topics:
...@@ -836,6 +837,11 @@ class DeviceSimService: ...@@ -836,6 +837,11 @@ class DeviceSimService:
"client": f"{prefix}/client", "client": f"{prefix}/client",
} }
# 获取 token_api_host（从环境配置的 default_topic_params 中）
token_api_host = None
if config.default_topic_params:
token_api_host = config.default_topic_params.get("token_api_host") or config.broker_host
sim = create_simulator( sim = create_simulator(
device_type=simulator.device_type, device_type=simulator.device_type,
device_id=simulator.device_id, device_id=simulator.device_id,
...@@ -844,6 +850,7 @@ class DeviceSimService: ...@@ -844,6 +850,7 @@ class DeviceSimService:
report_config=simulator.report_config, report_config=simulator.report_config,
topics=topics, topics=topics,
topic_params=simulator.topic_params, topic_params=simulator.topic_params,
token_api_host=token_api_host,
) )
if not sim: if not sim:
raise ValueError(f"不支持的设备类型: {simulator.device_type}") raise ValueError(f"不支持的设备类型: {simulator.device_type}")
...@@ -861,6 +868,8 @@ class DeviceSimService: ...@@ -861,6 +868,8 @@ class DeviceSimService:
success = sim.start() success = sim.start()
if success: if success:
# 启动成功后，用短锁写入内存表
with _running_simulators_lock:
_running_simulators[device_id] = sim _running_simulators[device_id] = sim
simulator.status = "running" simulator.status = "running"
simulator.updated_at = datetime.now() simulator.updated_at = datetime.now()
......
backend/app/simulators/__init__.py
浏览文件 @ ff749e33
...@@ -39,7 +39,8 @@ def create_simulator(device_type: str, device_id: str, ...@@ -39,7 +39,8 @@ def create_simulator(device_type: str, device_id: str,
env_config_id: str, mqtt_manager: MqttManager, env_config_id: str, mqtt_manager: MqttManager,
report_config: Optional[dict] = None, report_config: Optional[dict] = None,
topics: Optional[dict] = None, topics: Optional[dict] = None,
topic_params: Optional[dict] = None) -> Optional[BaseSimulator]: topic_params: Optional[dict] = None,
token_api_host: Optional[str] = None) -> Optional[BaseSimulator]:
""" """
模拟器工厂方法 模拟器工厂方法
...@@ -53,6 +54,7 @@ def create_simulator(device_type: str, device_id: str, ...@@ -53,6 +54,7 @@ def create_simulator(device_type: str, device_id: str,
report_config: 上报配置，如 {'interval': 30, 'enabled': True} report_config: 上报配置，如 {'interval': 30, 'enabled': True}
topics: 多主题前缀映射，key 为设备类型，value 为主题前缀 topics: 多主题前缀映射，key 为设备类型，value 为主题前缀
topic_params: 主题动态参数，如 {'room_id': 'A101', 'company_id': '001'} topic_params: 主题动态参数，如 {'room_id': 'A101', 'company_id': '001'}
token_api_host: Token 接口主机地址（门口屏设备启动时调用 getTokenInfoByToken 用）
Returns: Returns:
BaseSimulator: 模拟器实例，如果设备类型不支持则返回 None BaseSimulator: 模拟器实例，如果设备类型不支持则返回 None
...@@ -73,6 +75,7 @@ def create_simulator(device_type: str, device_id: str, ...@@ -73,6 +75,7 @@ def create_simulator(device_type: str, device_id: str,
report_config=report_config, report_config=report_config,
topics=topics or {}, topics=topics or {},
topic_params=topic_params, topic_params=topic_params,
token_api_host=token_api_host,
) )
......
backend/app/simulators/base_simulator.py
浏览文件 @ ff749e33
...@@ -44,7 +44,8 @@ class BaseSimulator(ABC): ...@@ -44,7 +44,8 @@ class BaseSimulator(ABC):
device_type: str, mqtt_manager: MqttManager, device_type: str, mqtt_manager: MqttManager,
report_config: Optional[dict] = None, report_config: Optional[dict] = None,
topics: Optional[dict] = None, topics: Optional[dict] = None,
topic_params: Optional[dict] = None): topic_params: Optional[dict] = None,
token_api_host: Optional[str] = None):
""" """
初始化模拟设备 初始化模拟设备
...@@ -56,6 +57,7 @@ class BaseSimulator(ABC): ...@@ -56,6 +57,7 @@ class BaseSimulator(ABC):
report_config: 上报配置，如 {'interval': 30, 'enabled': True} report_config: 上报配置，如 {'interval': 30, 'enabled': True}
topics: 多主题前缀映射，key 为设备类型，value 为主题前缀 topics: 多主题前缀映射，key 为设备类型，value 为主题前缀
topic_params: 主题动态参数，如 {'room_id': 'A101', 'company_id': '001'} topic_params: 主题动态参数，如 {'room_id': 'A101', 'company_id': '001'}
token_api_host: Token 接口主机地址（设备启动时调用 getTokenInfoByToken 用）
""" """
self.device_id = device_id self.device_id = device_id
self.env_config_id = env_config_id self.env_config_id = env_config_id
...@@ -70,6 +72,7 @@ class BaseSimulator(ABC): ...@@ -70,6 +72,7 @@ class BaseSimulator(ABC):
self._lock = threading.Lock() self._lock = threading.Lock()
self.topics = topics or {} self.topics = topics or {}
self.topic_params = topic_params or {} self.topic_params = topic_params or {}
self.token_api_host = token_api_host
# 解析真实主题模板 # 解析真实主题模板
self._resolved_topics: dict = {} self._resolved_topics: dict = {}
......
backend/app/simulators/door_simulator.py
浏览文件 @ ff749e33
...@@ -40,7 +40,8 @@ class DoorSimulator(BaseSimulator): ...@@ -40,7 +40,8 @@ class DoorSimulator(BaseSimulator):
def __init__(self, device_id: str, env_config_id: str, def __init__(self, device_id: str, env_config_id: str,
mqtt_manager: MqttManager, report_config: Optional[dict] = None, mqtt_manager: MqttManager, report_config: Optional[dict] = None,
topics: Optional[dict] = None, topics: Optional[dict] = None,
topic_params: Optional[dict] = None): topic_params: Optional[dict] = None,
token_api_host: Optional[str] = None):
""" """
初始化门口屏模拟器 初始化门口屏模拟器
...@@ -51,12 +52,14 @@ class DoorSimulator(BaseSimulator): ...@@ -51,12 +52,14 @@ class DoorSimulator(BaseSimulator):
report_config: 上报配置，如 {'interval': 30, 'enabled': True} report_config: 上报配置，如 {'interval': 30, 'enabled': True}
topics: 多主题前缀映射 topics: 多主题前缀映射
topic_params: 主题动态参数 topic_params: 主题动态参数
token_api_host: Token 接口主机地址（启动时调用 getTokenInfoByToken）
""" """
super().__init__(device_id, env_config_id, "door", mqtt_manager, super().__init__(device_id, env_config_id, "door", mqtt_manager,
report_config, topics, topic_params) report_config, topics, topic_params, token_api_host)
self._door_status = "closed" self._door_status = "closed"
self._battery_level = random.randint(60, 100) self._battery_level = random.randint(60, 100)
self._call_active = False self._call_active = False
self._token_info: Optional[dict] = None
def build_register_payload(self) -> dict: def build_register_payload(self) -> dict:
"""构建设备注册消息""" """构建设备注册消息"""
...@@ -88,6 +91,81 @@ class DoorSimulator(BaseSimulator): ...@@ -88,6 +91,81 @@ class DoorSimulator(BaseSimulator):
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S") "timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
} }
def start(self) -> bool:
"""
启动门口屏模拟器
流程：
1. 调用 getTokenInfoByToken 接口获取 token
2. 保存 token 信息并回填 topic_params
3. 调用父类 start() 执行 MQTT 注册/订阅/上报
Returns:
bool: 是否启动成功
"""
# 第一步：获取 token
token_host = self.token_api_host
if not token_host:
logger.warning(f"门口屏未配置 token_api_host，跳过 token 获取: {self.device_id}")
else:
try:
from app.simulators.door_token_client import DoorTokenClient, allocate_app_token
# 登记用户提供的 appToken（Excel 导入或手动创建时传入，保证不重复）
app_token = allocate_app_token(self.device_id, self.topic_params.get("app_token"))
client = DoorTokenClient(
host=token_host,
device_id=self.device_id,
app_token=app_token,
)
token_info = client.get_token()
# 保存 token 完整信息
self._token_info = token_info
# 回填 topic_params
token_data = token_info.get("token", {})
company_number = token_data.get("companyNumber", "")
cnum = token_data.get("cnum", "")
conference = token_info.get("conference", {})
conference_id = str(conference.get("conferenceId", ""))
conference_name = conference.get("conferenceName", "")
if company_number:
self.topic_params["company_id"] = company_number
if cnum:
self.topic_params["room_id"] = cnum
if conference_id:
self.topic_params["conference_id"] = conference_id
if conference_name:
self.topic_params["conference_name"] = conference_name
self.topic_params["app_token"] = app_token
logger.info(
f"门口屏 token 获取成功: device_id={self.device_id}, "
f"company_id={company_number}, room_id={cnum}"
)
except Exception as e:
logger.error(f"门口屏 token 获取失败: device_id={self.device_id}, error={e}")
self._running = False
# 将错误信息通知回调
self._notify_report(
topic="token_get",
payload={"device_id": self.device_id, "error": str(e)},
direction="publish",
status="failed",
error=str(e),
)
return False
# 第二步：重新解析主题（topic_params 已回填，解析出新的 MQTT 主题）
self._resolve_topics()
# 第三步：调用父类 MQTT 启动流程
return super().start()
def _on_command(self, topic: str, payload: dict) -> None: def _on_command(self, topic: str, payload: dict) -> None:
""" """
处理平台下发的指令 处理平台下发的指令
......
backend/app/simulators/door_token_client.py 0 → 100644
浏览文件 @ ff749e33
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称：door_token_client.py
模块描述：门口屏模拟器 Token 获取客户端
真实 Android 门口屏设备（UbGview 应用）启动后的第一个动作是调用
HTTP 接口 POST /exapi/api-android/token/getTokenInfoByToken 获取 token。
本客户端负责：
- 构造请求头（X-SIGN 固定值 / X-TIMESTAMP 秒级时间戳 / X-RANDOM 16位随机）
- 构造请求体（appToken / regUdid）
- appToken、regUdid 唯一性分配（批量场景不重复）
- 调用接口、解析响应、保存 Authorization / token / conference / company
签名说明（重要）：
- 真实设备日志中的 X-SIGN 为 108 字符（AES-CBC 加密，秒级时间戳）
- 现有 http_client.py 的算法（毫秒时间戳）与日志不匹配，无法本地复现
- 本期按已确认方案：复用日志中已验证可通过的固定 X-SIGN 值
作者：czj
创建日期：2026-08-19
最后修改：2026-08-19
"""
import hashlib
import logging
import random
import re
import string
import threading
import time
from typing import Dict, List, Optional
import requests
import urllib3
logger = logging.getLogger(__name__)
# 关闭测试环境自签名证书告警
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# ==================== 常量（从真实设备日志确认） ====================
# 接口调用路径（真实设备日志），如 https://192.168.5.48/exapi/...
TOKEN_API_PATH = "/exapi/api-android/token/getTokenInfoByToken"
# 固定 X-SIGN（真实设备日志样本，已实测可调用成功 HTTP 200）
# 真实设备日志：X-TIMESTAMP=1787131139（秒级）, X-RANDOM=13位
X_SIGN_FIXED = "mjDLL13nJnDM9sigjWxkLksYK4EpWOkaxv2QbIDhK6BCuqiN40J1D14arFdDYXXSG+Lv37btG/eKym7WoZ+5wZsKX7aDS3QPVIa05J2Al8g="
# 协议默认值：https（默认端口 443），可配置 http 走 80
DEFAULT_SCHEME = "https"
# 默认 appToken 前缀（AND = Android 门口屏；真实日志为 AND-34H-0101）
DEFAULT_APP_TOKEN_PREFIX = "AND"
# appToken 公司分片编号（默认 34H，来自日志 AND-34H-0101）
DEFAULT_APP_TOKEN_SEGMENT = "34H"
def normalize_host(host: str) -> str:
"""
规范化主机地址，确保形如 host:port（默认 HTTPS 443）
Args:
host: 主机地址，如 "192.168.5.48" 或 "192.168.5.48:8080"
或完整 URL "https://192.168.5.48/exapi/"
Returns:
str: 规范化后的主机地址（不含协议和路径）
"""
host = (host or "").strip()
if not host:
raise ValueError("token_api_host 不能为空")
# 1. 去掉协议前缀
if "://" in host:
host = host.split("://", 1)[1]
# 2. 去掉路径（取第一个 / 之前）
host = host.split("/", 1)[0]
# 3. 已经是 host:port 形式则保持
if re.match(r"^[\w.\-]+:\d+$", host):
return host
return host
def pick_default_host() -> str:
"""
从 MQTT 环境配置中选中默认 token 接口主机
优先级：
1. EnvConfig.default_topic_params['token_api_host']
2. EnvConfig.broker_host（MQTT broker 与 token 接口同机时）
Returns:
str: token 接口主机地址
"""
from app.models.device_sim import EnvConfig
from app.database import async_session_factory
try:
# 同步阻塞轮询所有环境配置，选择第一个配置了 token_api_host 的
import asyncio
loop = asyncio.new_event_loop()
try:
asyncio.set_event_loop(loop)
result = loop.run_until_complete(_pick_default_host_async())
finally:
loop.close()
return result
except Exception as e:
logger.warning(f"获取默认 token 接口主机失败，使用空值: {e}")
return ""
async def _pick_default_host_async() -> str:
"""异步轮询环境配置获取默认 token 接口主机"""
from app.models.device_sim import EnvConfig
from app.database import async_session_maker
async with async_session_maker() as db:
from sqlalchemy import select
result = await db.execute(select(EnvConfig))
configs = list(result.scalars().all())
if not configs:
return ""
# 1. 优先 default_topic_params.token_api_host
for cfg in configs:
params = cfg.default_topic_params or {}
if params.get("token_api_host"):
return str(params["token_api_host"]).strip()
# 2. 回退 broker_host
for cfg in configs:
if cfg.broker_host:
return cfg.broker_host.strip()
return ""
def _normalize_segment(raw: str) -> str:
"""规范化公司分片编号：去除非字母数字，限长 4"""
cleaned = re.sub(r"[^A-Za-z0-9]", "", str(raw or ""))
return (cleaned or DEFAULT_APP_TOKEN_SEGMENT)[:4]
def _device_seed(device_id: str) -> int:
"""由 device_id 生成稳定的正整数种子（同设备多次启动结果一致）"""
digest = hashlib.md5(str(device_id).encode("utf-8")).hexdigest()
return int(digest[:8], 16) # md5 前 8 hex → 32 位无符号整数
def generate_reg_udid(device_id: str) -> str:
"""
由 device_id 派生唯一的 regUdid（确定性，同设备多次启动一致）
格式：reg_{md5(device_id)前16位}
regUdid 为 16 位十六进制，仅含小写字母和数字，避免注册字符集冲突。
Args:
device_id: 设备 ID
Returns:
str: 16 位 regUdid
"""
digest = hashlib.md5(str(device_id).encode("utf-8")).hexdigest()[:16]
return f"reg_{digest}"
class DoorTokenClient:
"""
门口屏模拟器 token 获取客户端
负责 getTokenInfoByToken 接口的调用，支持：
- 固定 X-SIGN 签名头
- appToken / regUdid 唯一性分配
- 超时重试、响应解析
Attributes:
host (str): token 接口主机（如 "192.168.5.48"）
device_id (str): 设备 ID
app_token (str): 分配的 appToken
reg_udid (str): 分配的 regUdid
timeout (float): 请求超时（秒）
max_retries (int): 网络错误最大重试次数
_session (requests.Session): 复用的会话
"""
def __init__(
self,
host: str,
device_id: str,
app_token: Optional[str] = None,
x_sign: Optional[str] = None,
timeout: float = 5.0,
max_retries: int = 3,
scheme: str = DEFAULT_SCHEME,
):
"""
初始化客户端
Args:
host: token 接口主机（如 "192.168.5.48" 或 "192.168.5.48:8080"）
device_id: 设备 ID
app_token: 分配的 appToken；为 None 时自动分配
x_sign: 固定 X-SIGN；为 None 时使用内置默认值
timeout: 请求超时（秒）
max_retries: 网络错误最大重试次数（默认 3）
scheme: 协议（https 默认 / http）
"""
if not host:
raise ValueError("host 不能为空，需提供 token 接口主机地址")
# 从原始输入自动识别协议（如 "https://192.168.5.48" 或 "192.168.5.48:8080"）
raw_host = str(host).strip()
raw_lower = raw_host.lower()
if raw_lower.startswith("https://"):
scheme = "https"
elif raw_lower.startswith("http://"):
scheme = "http"
elif raw_lower.startswith("wss://"):
scheme = "https" # wss 视为带 TLS 的 https
elif raw_lower.startswith("ws://"):
scheme = "http"
self.host = normalize_host(host)
self.device_id = device_id
self.timeout = timeout
self.max_retries = max_retries
self.scheme = (scheme or DEFAULT_SCHEME).lower().rstrip(":")
if self.scheme not in ("http", "https"):
raise ValueError(f"不支持的协议: {self.scheme}")
# 唯一性分配
self.app_token = app_token or tokenizer.allocate_app_token(device_id)
self.reg_udid = generate_reg_udid(device_id)
# 日志中的固定 X-SIGN
self.x_sign = x_sign or X_SIGN_FIXED
logger.info(
f"TokenClient 初始化: device_id={device_id}, app_token={self.app_token}, "
f"reg_udid={self.reg_udid}"
)
self._session = self._create_session()
@property
def base_url(self) -> str:
"""接口基础 URL：{scheme}://{host}/"""
return f"{self.scheme}://{self.host}"
def _create_session(self) -> requests.Session:
"""创建带连接级重试的会话（网络错误可重试）"""
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
session = requests.Session()
retry = Retry(
total=self.max_retries,
backoff_factor=0.5,
status_forcelist=[500, 502, 503, 504],
allowed_methods=["POST"],
raise_on_status=False,
)
adapter = HTTPAdapter(max_retries=retry)
session.mount("http://", adapter)
session.mount("https://", adapter)
return session
def _build_headers(self, x_timestamp: Optional[str] = None, x_random: Optional[str] = None) -> Dict[str, str]:
"""
构造请求头
Args:
x_timestamp: 秒级时间戳；为 None 自动生成
x_random: 随机串；为 None 自动生成 16 位
Returns:
dict: 请求头
"""
headers = {
"Content-Type": "application/json; charset=utf-8",
"Accept": "application/json, text/plain, */*",
"X-SIGN": self.x_sign,
"X-TIMESTAMP": x_timestamp or str(int(time.time())),
"X-RANDOM": x_random or "".join(
random.choices(string.ascii_letters + string.digits, k=16)
),
}
return headers
def _build_body(self) -> Dict[str, str]:
"""构造请求体"""
return {
"appToken": self.app_token,
"regUdid": self.reg_udid,
}
@staticmethod
def _parse_response(resp: requests.Response) -> dict:
"""
解析接口响应
注意：实际响应中 code 为字符串 "200"，
conference 和 company 嵌套在 token 对象内部。
Args:
resp: HTTP 响应对象
Returns:
dict: {authorization, token, conference, company}
Raises:
ValueError: 响应解析失败或业务失败
"""
try:
data = resp.json()
except Exception as e:
raise ValueError(f"响应 JSON 解析失败: {e}, body={resp.text[:500]}")
if not isinstance(data, dict):
raise ValueError(f"响应格式异常: {str(data)[:200]}")
code = str(data.get("code", ""))
success = data.get("success", False)
if code not in ("200", 200) or not success:
message = data.get("message", "")
raise ValueError(f"接口业务失败: code={code}, message={message}")
result = data.get("result") or {}
authorization = result.get("Authorization") or ""
if not authorization:
raise ValueError("响应缺少 Authorization 字段")
# 注意：实际接口响应中 conference 和 company 嵌套在 token 内部
token_info = result.get("token") or {}
conference_info = token_info.get("conference") or {}
company_info = token_info.get("company") or {}
# 去掉 token 中嵌套的 conference/company，保持顶层干净
token_clean = dict(token_info)
token_clean.pop("conference", None)
token_clean.pop("company", None)
return {
"authorization": authorization,
"token": token_clean,
"conference": conference_info,
"company": company_info,
"code": code,
"message": data.get("message", ""),
}
def get_token(self) -> dict:
"""
调用 getTokenInfoByToken 接口获取 token
Returns:
dict: {authorization, token, conference, company, ...}
Raises:
ConnectionError: 网络可达但业务失败（重试）后仍失败
ValueError: 参数/解析错误
"""
url = f"{self.base_url}{TOKEN_API_PATH}"
headers = self._build_headers()
body = self._build_body()
logger.info(f"[Token] 调用接口: POST {url}")
logger.info(f"[Token] 请求头: X-TIMESTAMP={headers['X-TIMESTAMP']}, X-RANDOM={headers['X-RANDOM']}")
logger.info(f"[Token] 请求体: {body}")
last_error: Optional[Exception] = None
for attempt in range(1, self.max_retries + 1):
try:
resp = self._session.post(
url,
json=body,
headers=headers,
verify=False,
timeout=self.timeout,
)
logger.info(f"[Token] 响应: HTTP {resp.status_code}, body={resp.text[:500]}")
if resp.status_code == 200:
parsed = self._parse_response(resp)
logger.info(f"[Token] 获取成功: device_id={self.device_id}, appToken={self.app_token}")
return parsed
# 非 200 状态码：业务错误，不重试
raise ValueError(f"HTTP {resp.status_code}: {resp.text[:300]}")
except ValueError as e:
# 业务错误：不重试
raise e
except (ConnectionError, requests.ConnectionError, requests.Timeout) as e:
last_error = e
logger.warning(f"[Token] 网络错误（第 {attempt}/{self.max_retries} 次）: {e}")
if attempt < self.max_retries:
import time as _time
_time.sleep(2 ** (attempt - 1)) # 指数退避 1s, 2s, 4s
raise ConnectionError(
f"token 接口调用失败（重试 {self.max_retries} 次）: {last_error}"
)
class TokenPool:
"""
进程级 appToken 登记池（保证批量场景不重复）
注意：appToken 不是自动生成的，而是导入设备时由用户提供的
服务端已注册授权码（如 AND-34H-0101）。本池只做登记与去重：
- 同一 appToken 在同一进程内只能被一台设备使用
- 重复使用抛 ValueError，避免批量导入/启动时静默冲突
"""
def __init__(
self,
prefix: str = DEFAULT_APP_TOKEN_PREFIX,
segment: str = DEFAULT_APP_TOKEN_SEGMENT,
):
"""
初始化登记池
Args:
prefix: appToken 固定前缀（AND，仅供参考不做生成）
segment: 公司分片编号（仅供参考不做生成）
"""
self._prefix = prefix.upper()
self._segment = _normalize_segment(segment)
self._used: set = set()
self._lock = threading.Lock()
def allocate(self, device_id: str, app_token: Optional[str] = None) -> str:
"""
登记设备使用的 appToken（必须由导入提供，不自动生成）
Args:
device_id: 设备 ID
app_token: 用户提供的 appToken（Excel 导入/手动创建）
Returns:
str: 登记后的 appToken
Raises:
ValueError: appToken 为空或已被其他设备使用
"""
with self._lock:
if not app_token:
raise ValueError(
f"设备 {device_id} 未配置 appToken：门口屏 token 获取必须使用"
f"导入设备时提供的服务端授权码，不可自动生成"
)
if app_token in self._used:
raise ValueError(f"appToken 已被其他设备使用，批量场景授权码不能重复: {app_token}")
self._used.add(app_token)
return app_token
# 全局单例（进程内共享，保证跨设备分配唯一）
tokenizer = TokenPool()
def allocate_app_token(device_id: str, app_token: Optional[str] = None) -> str:
"""
登记设备使用的 appToken（由导入提供，不自动生成）
Args:
device_id: 设备 ID
app_token: 用户提供的 appToken（Excel 导入/手动创建）
Returns:
str: 登记后的 appToken
Raises:
ValueError: 未提供 appToken 或重复使用
"""
return tokenizer.allocate(device_id, app_token)
__all__ = [
"DoorTokenClient",
"TokenPool",
"tokenizer",
"allocate_app_token",
"generate_reg_udid",
"normalize_host",
"pick_default_host",
"TOKEN_API_PATH",
"X_SIGN_FIXED",
"DEFAULT_APP_TOKEN_PREFIX",
"DEFAULT_APP_TOKEN_SEGMENT",
]
\ No newline at end of file
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论