ezviz-open-capture

Original source / Raw SKILL.md

---
name: phone-detection-alert
description: 玩手机检测告警技能。通过摄像头抓图→AI 分析→TTS 语音→设备播放的完整流程。Use when: 需要监控玩手机行为并自动告警、课堂/考场/会议室纪律监控、通过萤石摄像头进行 AI 行为检测。
metadata:
  {
    "openclaw": {
      "emoji": "📱",
      "requires": {
        "env": ["EZVIZ_APP_KEY", "EZVIZ_APP_SECRET", "EZVIZ_DEVICE_SERIAL"],
        "pip": ["requests", "edge-tts"]
      },
      "primaryEnv": "EZVIZ_APP_KEY"
    }
  }
---

# Phone Detection Alert (玩手机检测告警)

萤石开放平台 AI 算法 + 语音告警，检测到玩手机自动播放提醒。

## 快速开始

### 安装依赖

```bash
pip install requests edge-tts
```

### 设置环境变量（必需）

**必需的环境变量**:
```bash
export EZVIZ_APP_KEY="your_app_key"
export EZVIZ_APP_SECRET="your_app_secret"
export EZVIZ_DEVICE_SERIAL="dev1,dev2,dev3"
```

**可选的环境变量**:
```bash
export EZVIZ_CHANNEL_NO="1"  # 通道号，默认 1
```

⚠️ **重要**: 
- `EZVIZ_APP_KEY`, `EZVIZ_APP_SECRET`, `EZVIZ_DEVICE_SERIAL` 是**必需**的环境变量
- 技能运行前必须设置这些环境变量
- 不需要设置 `EZVIZ_ACCESS_TOKEN`！技能会自动用 `appKey + appSecret` 获取 Token（有效期 7 天，内存缓存）

运行：
```bash
python3 {baseDir}/scripts/phone_detection_alert.py
```

命令行参数：
```bash
# 单个设备
python3 {baseDir}/scripts/phone_detection_alert.py appKey appSecret dev1 1

# 多个设备（逗号分隔）
python3 {baseDir}/scripts/phone_detection_alert.py appKey appSecret "dev1,dev2,dev3" 1

# 指定通道号
python3 {baseDir}/scripts/phone_detection_alert.py appKey appSecret "dev1:1,dev2:2" 1

# 测试模式（跳过检测，直接播放告警）
python3 {baseDir}/scripts/phone_detection_alert.py appKey appSecret "dev1,dev2" 1 --test
```

## 工作流程

```
1. 获取 Token (appKey + appSecret → accessToken, 有效期 7 天)
       ↓
2. 设备抓图 (accessToken + deviceSerial → picUrl, 有效期 2 小时)
       ↓
3. AI 分析 (accessToken + picUrl → 是否玩手机)
       ↓ [检测到]
4. TTS 语音 ("检测到有人玩手机" → audio.mp3)
       ↓
5. 上传语音 (accessToken + audio.mp3 → fileUrl)
       ↓
6. 下发播放 (accessToken + deviceSerial + fileUrl → 设备播放)
```

## Token 自动获取说明

**你不需要手动获取或配置 `EZVIZ_ACCESS_TOKEN`！**

技能会自动处理 Token 的获取和缓存：

```
首次运行:
  appKey + appSecret → 调用萤石 API → 获取 accessToken
  ↓
  缓存到内存（有效期 7 天）
  ↓
后续运行 (7 天内):
  使用缓存的 accessToken（无需重复获取）
  ↓
7 天后:
  自动重新获取新的 accessToken
```

**Token 管理特性**:
- ✅ **自动获取**: 首次运行时自动调用萤石 API 获取
- ✅ **自动缓存**: Token 保存在内存中，7 天内重复使用
- ✅ **自动刷新**: Token 过期后自动重新获取
- ✅ **无需配置**: 不需要手动设置 `EZVIZ_ACCESS_TOKEN` 环境变量
- ✅ **安全**: Token 不写入日志，不保存到磁盘

**为什么这样设计**:
- 萤石 Token 有效期为 7 天，无需每次获取
- 自动管理减少用户配置负担
- 避免 Token 泄露风险（不暴露在环境变量中）

## 输出示例

```
============================================================
Phone Detection Alert System
============================================================
[INFO] Detected 2 device(s): [('dev1', 1), ('dev2', 1)]
[SUCCESS] Token obtained, expires: 2026-03-20 15:00:00

[Device] dev1 (Channel: 1)
[SUCCESS] Image captured: https://opencapture.ys7.com/...
[ALERT] Phone usage detected! (confidence: 0.95)
[SUCCESS] Voice uploaded: http://custom-voice-reminder-hn...
[SUCCESS] Alert sent to device dev1!

[Device] dev2 (Channel: 1)
[SUCCESS] Image captured: https://opencapture.ys7.com/...
[INFO] No phone usage detected

============================================================
DETECTION SUMMARY
============================================================
  Total devices:     2
  Phone detected:    1
  Not detected:      1
  Failed:            0
  Alerts sent:       1
============================================================
```

## API 接口

| 接口 | URL | 文档 |
|------|-----|------|
| 获取 Token | `POST /api/lapp/token/get` | https://open.ys7.com/help/81 |
| 设备抓图 | `POST /api/lapp/device/capture` | https://open.ys7.com/help/687 |
| 玩手机检测 | `POST /api/service/intelligence/algo/analysis/play_phone_detection` | https://open.ys7.com/help/3956 |
| 语音上传 | `POST /api/lapp/voice/upload` | https://open.ys7.com/help/1241 |
| 语音下发 | `POST /api/lapp/voice/send` | https://open.ys7.com/help/1253 |

## 网络端点

| 域名 | 用途 |
|------|------|
| `open.ys7.com` | 萤石开放平台 API |
| `aidialoggw.ys7.com` | 萤石 AI 智能体 |
| `aliyuncs.com` | 萤石语音存储（阿里云 OSS） |

## 格式代码

**检测返回**:
- `images: null` - 图片中没有人
- `label: "play_phone"` - 检测到玩手机行为
- `labelWeight: 0.95` - 置信度 95%

**错误码**:
- `200` - 操作成功
- `10002` - accessToken 过期
- `10028` - 抓图次数超限
- `20007` - 设备不在线
- `20008` - 设备响应超时

## Tips

- **多设备**: 逗号分隔 `dev1,dev2,dev3`
- **指定通道**: 冒号分隔 `dev1:1,dev2:2`
- **Token 有效期**: 7 天，自动缓存
- **图片有效期**: 2 小时
- **频率限制**: 建议间隔 ≥4 秒
- **定时任务**: 建议 ≥5 分钟

## 注意事项

⚠️ **频率限制**: 萤石抓图接口建议间隔 4 秒以上，频繁调用可能触发限流（错误码 10028）

⚠️ **隐私合规**: 使用摄像头监控可能涉及隐私问题，确保符合当地法律法规

⚠️ **设备要求**: 设备必须在线且支持语音对讲功能（`support_talk=1` 或 `3`）

⚠️ **Token 安全**: Token 仅在内存中使用，不写入日志，不发送到非萤石端点

## 数据流出说明

**⚠️ 重要：本技能会向以下第三方服务发送数据**

| 数据类型 | 发送到 | 用途 | 是否必需 | 隐私影响 |
|----------|--------|------|----------|----------|
| 摄像头抓拍图片 | `open.ys7.com` (萤石) | AI 玩手机行为分析 | ✅ 必需 | 🔴 包含监控画面 |
| TTS 文本 | `edge-tts.microsoft.com` (Microsoft Azure) | 生成语音文件 | ✅ 必需 | 🟢 仅文本 |
| 语音文件 | `aliyuncs.com` (阿里云 OSS) | 临时存储，供设备下载 | ✅ 必需 | 🟡 临时存储 2 小时 |
| appKey/appSecret | `open.ys7.com` (萤石) | 获取访问 Token | ✅ 必需 | 🔴 凭据 |
| 设备序列号 | `open.ys7.com` (萤石) | 设备控制 | ✅ 必需 | 🟡 设备标识 |

**数据流出详细说明**:

1. **萤石开放平台** (`open.ys7.com`, `aidialoggw.ys7.com`):
   - 发送：摄像头抓拍图片、appKey/appSecret、设备序列号
   - 用途：Token 获取、AI 玩手机行为分析、设备控制
   - 隐私：🔴 包含监控画面和凭据

2. **Microsoft Azure TTS** (`edge-tts.microsoft.com`):
   - 发送：TTS 文本（"检测到有人玩手机，请立即停止使用手机！"）
   - 用途：生成语音文件
   - 隐私：🟢 仅固定文本，不包含个人信息

3. **阿里云 OSS** (`aliyuncs.com`):
   - 发送：生成的语音文件（.mp3）
   - 用途：临时存储，供萤石设备下载播放
   - 隐私：🟡 临时存储 2 小时，自动过期
   - 存储位置：萤石合作的阿里云 OSS 存储桶

**数据不流出**:
- ❌ 不会发送数据到技能作者或任何个人
- ❌ 不会发送到 ClawHub 或 OpenClaw
- ❌ 不会永久存储任何数据

**凭证安全建议**:
- 使用**最小权限**的 appKey/appSecret（仅需设备控制和 AI 分析权限）
- 定期轮换凭据
- 不要使用主账号凭据
- 限制凭据的 IP 访问范围（如果萤石支持）

**凭证权限建议**:
- 使用**最小权限**的 appKey/appSecret
- 仅开通必要的 API 权限（设备抓图、AI 分析、语音）
- 定期轮换凭证
- 不要使用主账号凭证

**本地处理**:
- ✅ Token 缓存于内存，不写入磁盘
- ✅ 不记录完整 API 响应
- ✅ 不保存抓拍图片到本地（除非手动指定下载路径）


---

## Skill Companion Files

> Additional files collected from the skill directory layout.

### _meta.json

```json
{
  "owner": "ezviz-open",
  "slug": "ezviz-open-capture",
  "displayName": "Ezviz Open Capture Phone Detect",
  "latest": {
    "version": "1.0.10",
    "publishedAt": 1773407187544,
    "commit": "https://github.com/openclaw/skills/commit/d7ed7df0fdc6d21b87f01dd86d2916ef6effedf7"
  },
  "history": []
}

```

### references/ezviz-api-docs.md

```markdown
# 萤石开放平台 API 文档

本文档包含玩手机检测告警流程所需的**五个**核心 API 接口详细说明。

**完整工作流程**：
```
获取 Token → 设备抓图 → 玩手机检测 → 上传语音 → 下发语音
```

---

## 0. 获取 AccessToken 接口

**文档 URL**: https://open.ys7.com/help/81  
**更新时间**: 2025.02.21

### 接口说明

AccessToken 是访问令牌，接口调用必备的公共参数之一，用于校验接口访问/调用是否有权限。

**重要特性**：
- 有效期为 **7 天**，有效期内不需要重复申请，可以重复使用
- 有效期 7 天无法变更，请在业务端使用 AccessToken 的场景中，校验老 Token 的有效性和失效时重新获取 Token 的机制
- 新获取 Token 不会使老 Token 失效，每个 Token 独立拥有 7 天生命周期
- 由于 Token 属于用户身份认证令牌，在用户变更身份信息（用户注销、密码修改）后会将旧的 Token 进行失效处理

**最佳实践**：
> 获取到的 accessToken 有效期是 7 天，请在**即将过期**或者**接口报错 10002**时重新获取，**每个 token 具备独立的 7 天生命周期，请勿频繁调用避免占用过多接口调用次数**。最佳实践是在本地进行缓存，给对应有权限的用户使用，而不是在每次使用业务接口前获取一次。

### 请求地址

```
POST https://open.ys7.com/api/lapp/token/get
```

### 请求方式

POST (application/x-www-form-urlencoded)

### 请求参数

| 参数名 | 类型 | 描述 | 是否必选 |
|--------|------|------|----------|
| appKey | String | appKey（在官网 - 开发者服务 - 我的应用中获取） | Y |
| appSecret | String | appSecret | Y |

### HTTP 请求报文

```http
POST /api/lapp/token/get HTTP/1.1
Host: open.ys7.com
Content-Type: application/x-www-form-urlencoded

appKey=9mqitppidgce4y8n54ranvyqc9fjtsrl&appSecret=096e76501644989b63ba0016ec5776
```

### 返回数据

```json
{
  "data": {
    "accessToken": "at.7jrcjmna8qnqg8d3dgnzs87m4v2dme3l-32enpqgusd-1jvdfe4-uxo15ik0s",
    "expireTime": 1470810222045
  },
  "code": "200",
  "msg": "操作成功!"
}
```

### 返回字段

| 字段名 | 类型 | 描述 |
|--------|------|------|
| accessToken | String | 获取的 accessToken |
| expireTime | long | 具体过期时间，精确到毫秒 |

### 返回码

| 返回码 | 返回消息 | 描述 |
|--------|----------|------|
| 200 | 操作成功 | 请求成功 |
| 10001 | 参数错误 | 参数为空或格式不正确 |
| 10005 | appKey 异常 | appKey 被冻结 |
| 10017 | appKey 不存在 | 确认 appKey 是否正确 |
| 10030 | appkey 和 appSecret 不匹配 | - |
| 49999 | 数据异常 | 接口调用异常 |

---

## 1. 设备抓拍图片接口

**文档 URL**: https://open.ys7.com/help/687  
**更新时间**: 2025.11.25

### 接口功能

抓拍设备当前画面，**该接口仅适用于 IPC 或者关联 IPC 的 DVR 设备**，该接口并非预览时的截图功能。

**海康型号设备**可能不支持萤石协议抓拍功能，使用该接口可能返回不支持或者超时。

**该接口需要设备支持能力集**：`support_capture=1`

> ⚠️ **注意**：设备抓图能力有限，请勿频繁调用，建议每个摄像头调用的间隔**4s 以上**。

### 请求地址

```
POST https://open.ys7.com/api/lapp/device/capture
```

### 请求方式

POST (application/x-www-form-urlencoded)

### 子账户 token 请求所需最小权限

```
"Permission":"Capture"
"Resource":"Cam:序列号：通道号"
```

### 请求参数

| 参数名 | 类型 | 描述 | 是否必选 |
|--------|------|------|----------|
| accessToken | String | 授权过程获取的 access_token | Y |
| deviceSerial | String | 设备序列号，存在英文字母的设备序列号，字母需为大写 | Y |
| channelNo | int | 通道号，IPC 设备填写 1 | Y |
| quality | int | 视频清晰度，0-流畅，1-高清 (720P),2-4CIF,3-1080P,4-400w **注：此参数不生效** | N |

### HTTP 请求报文

```http
POST /api/lapp/device/capture HTTP/1.1
Host: open.ys7.com
Content-Type: application/x-www-form-urlencoded

accessToken=at.12xp95k63bboast3aq0g5hg22q468929&deviceSerial=427734888&channelNo=1
```

### 返回数据

```json
{
  "data": {
    "picUrl": "https://ezviz-fastdfs-gateway.oss-cn-hangzhou.aliyuncs.com/1/capture/003eyM73IFbVHUM6Ktz7K6JXXLeUbFU.jpg?Expires=1654756106&OSSAccessKeyId=LTAIzI38nEHqg64n&Signature=SEGCPK0ExrKYZBEK3hc6ZZ%252FcPSY%3D"
  },
  "code": "200",
  "msg": "操作成功!"
}
```

### 返回字段

| 字段名 | 类型 | 描述 |
|--------|------|------|
| picUrl | String | 抓拍后的图片路径，**图片保存有效期为 2 小时** |

### 返回码

| 返回码 | 返回消息 | 描述 |
|--------|----------|------|
| 200 | 操作成功 | 请求成功 |
| 10001 | 参数错误 | 参数为空或格式不正确 |
| 10002 | accessToken 异常或过期 | 重新获取 accessToken |
| 10005 | appKey 异常 | appKey 被冻结 |
| 10028 | 抓图接口调用次数超限 | 抓图接口调用次数超限 |
| 10031 | 子账户或萤石用户没有权限 | 子账户或萤石用户没有权限 |
| 10051 | 无权限进行抓图 | 设备不属于当前用户或者未分享给当前用户 |
| 20002 | 设备不存在 | - |
| 20006 | 网络异常 | 检查设备网络状况，稍后再试 |
| 20007 | 设备不在线 | 检查设备是否在线 |
| 20008 | 设备响应超时 | 操作过于频繁或者设备不支持萤石协议抓拍 |
| 20014 | deviceSerial 不合法 | - |
| 20032 | 该用户下该通道不存在 | 检查设备是否包含该通道 |
| 49999 | 数据异常 | 接口调用异常 |
| 60012 | 设备抓图未知错误 | 可联系设备研发定位问题 |
| 60017 | 设备抓图失败，2030 等 | 可联系设备研发定位问题 |
| 60020 | 不支持该命令 | 确认设备是否支持抓图 |
| 60058 | 设备存在高风险，需求确权 | - |

---

## 2. 玩手机算法分析接口

**文档 URL**: https://open.ys7.com/help/3956  
**更新时间**: 2024.12.23

### 前置条件

- 开通 AI 算法服务，如未开通可联系客服支持
- 需联系客服手动开通该接口调用权限

### 接口 URL

```
POST https://open.ys7.com/api/service/intelligence/algo/analysis/play_phone_detection
```

### 请求参数

#### Header

| 名称 | 类型 | 必填 | 描述 |
|------|------|------|------|
| accessToken | string | Y | 萤石开放 API 访问令牌 |

#### Body (JSON)

| 名称 | 类型 | 必填 | 描述 |
|------|------|------|------|
| requestId | string | N | 请求 Id，请使用 uuid |
| stream | boolean | Y | 必传 false |
| dataInfo | array\<object\> | Y | 请求的输入数据内容 |
| └─ modal | string | Y | 数据模态：image（图片） |
| └─ type | string | Y | 数据类型：url |
| └─ data | object | Y | url 地址 |
| dataParams | array\<object\> | N | 请求的输入图片信息 |
| └─ modal | string | Y | 数据模态：image（图片） |
| └─ img_width | int | Y | 图片宽度 |
| └─ img_height | int | Y | 图片高度 |

### 请求示例

```bash
curl --location --request POST 'https://open.ys7.com/api/service/intelligence/algo/analysis/play_phone_detection' \
  --header 'accessToken: at.3xwsj8em6p28dw3t92nf4itq4mote8qr-6t75j5aq2m-1i7rkyf-pwz8z7rfi' \
  --header 'Content-Type: application/json' \
  --data-raw '{
    "taskType": "",
    "dataInfo": [
      {
        "data": "https://qna.smzdm.com/202403/20/65fa423e718c83216.jpg_e680.jpg",
        "type": "url",
        "modal": "image"
      }
    ],
    "dataParams": [
      {
        "modal": "image",
        "img_width": 1280,
        "img_height": 720
      }
    ]
  }'
```

### 返回数据

| 名称 | 类型 | 描述 |
|------|------|------|
| meta | object | 响应信息 |
| └─ code | int | 服务响应状态码 |
| └─ message | string | 服务响应状态描述 |
| └─ moreInfo | object | 服务响应描述 |
| data | object | 算法分析结果 |
| └─ taskType | string | 算法类型：play_phone_detection |
| └─ requestId | string | 请求唯一 ID |
| └─ images | list[Object] | 图片结果 |
| └─ └─ contentAnn | Object | 结构化分析结果 |
| └─ └─ ─ bboxes | List[object] | 检测数据结果 |
| └─ └─ └─ └─ points | list[List[float]] | 检测框（归一化值） |
| └─ └─ └─ └─ weight | float | 置信度 |
| └─ └─ └─ └─ tagInfo | Object | 标签信息 |
| └─ └─ └─ └─ └─ tag | String | "person" |
| └─ └─ └─ └─ └─ labels | List[object] | 行为标签 |
| └─ └─ └─ └─ └─ └─ key | String | "behavior" |
| └─ └─ └─ └─ └─ └─ label | String | "play_phone"（玩手机） |
| └─ └─ └─ └─ └─ └─ labelWeight | float | 置信度 |

### 返回示例

```json
{
  "meta": {
    "code": 200,
    "message": "success",
    "moreInfo": {},
    "success": true
  },
  "data": {
    "requestId": "56442d3243cc45a2bec4e9b2fe596ad1",
    "taskType": "play_phone_detection",
    "images": [
      {
        "frameIdx": 0,
        "imageHeight": 510,
        "imageWidth": 680,
        "contentAnn": {
          "bboxes": [
            {
              "points": [
                {"x": 0.2623883, "y": 0.08050475},
                {"x": 0.8080127, "y": 0.9986414}
              ],
              "weight": 0.993,
              "tagInfo": {
                "tag": "person",
                "labels": [
                  {
                    "key": "behavior",
                    "label": "play_phone",
                    "labelWeight": 1.0
                  }
                ]
              }
            }
          ]
        }
      }
    ]
  }
}
```

### 错误码

| 错误码 | 错误信息 | 解决方案 |
|--------|----------|----------|
| 10002 | accessToken 过期或异常 | 重新获取 accessToken |
| 60202 | 参数解析错误 | 检查请求参数格式 |
| 60203 | 未开通相关服务 | 联系客服开通服务 |
| 60206 | 并发数超限 | 降低请求频率 |
| 60205 | 服务内部错误 | 联系技术支持 |

---

## 3. 语音文件上传接口

**文档 URL**: https://open.ys7.com/help/1241  
**更新时间**: 2026.03.09

### 接口功能

上传本地语音文件到萤石云

### 请求地址

```
POST https://open.ys7.com/api/lapp/voice/upload
```

### 请求方式

POST (multipart/form-data)

### 请求参数

| 参数名 | 类型 | ParameterType | 描述 | 是否必填 |
|--------|------|-------------|------|----------|
| accessToken | String | body | 访问令牌 | Y |
| voiceFile | MultipartFile | form-data | 语音文件，支持 wav、mp3、aac 格式，最大 5M，最长 60s | Y |
| voiceName | String | body | 语音名称，最长 50 个字符 | N |
| force | boolean | body | 如果已存在相同 voiceName 的语音文件，则替换原语音文件，true 表示强制替换，false 表示如果存在不替换，默认为 false | N |

### 请求报文

```http
POST /api/lapp/voice/upload HTTP/1.1
Host: <host>:<port>
Content-Type: multipart/form-data
```

### 返回结果

| 参数名 | 类型 | 描述 |
|--------|------|------|
| msg | String | 操作信息 |
| code | String | 操作码，200 表示操作成功 |
| data | Object | 语音文件信息 |
| └─ name | String | 语音名称 |
| └─ url | String | 语音文件下载地址，有效期 1 天 |

### 返回示例

```json
{
  "msg": "Operation succeeded",
  "code": "200",
  "data": [
    {
      "name": "babble.wav",
      "url": "http://oss-cn-shenzhen.aliyuncs.com/voice/bbe5cc634be34a7484947d63f6361c22.aac?Expires=1583059824&OSSAccessKeyId=testId&&Signature=kM91%2By"
    }
  ]
}
```

---

## 4. 语音文件下发接口

**文档 URL**: https://open.ys7.com/help/1253  
**更新时间**: 2026.03.09

### 接口功能

语音文件下发，指定设备播放，需要设备支持能力集 support_talk=1 或 3

### 请求地址

```
POST https://open.ys7.com/api/lapp/voice/send
```

### 请求方式

POST

### 请求参数

| 参数名 | 类型 | Parameter Type | 描述 | 是否必填 |
|--------|------|---------------|------|----------|
| accessToken | String | body | 访问令牌 | Y |
| deviceSerial | String | body | 设备序列号 | Y |
| channelNo | int | body | 通道号，默认通道号为 1 | N |
| fileUrl | String | body | 下载音频文件的 url（上传接口返回的 url） | Y |

### 请求报文

```http
POST /api/lapp/voice/send HTTP/1.1
Host: <host>:<port>
Content-Type: application/x-www-form-urlencoded
```

### 返回数据

| 参数名 | 类型 | 描述 |
|--------|------|------|
| msg | String | 操作成功! |
| code | String | 200 |

### 返回示例

```json
{
  "msg": "操作成功!",
  "code": "200"
}
```

### 错误码

| 返回值 | 返回信息 | 描述 |
|--------|----------|------|
| 10001 | fileUrl 参数不合法 | 检查 fileUrl 格式 |
| 10001 | channels 参数不合法 | 检查通道号 |
| 20002 | 设备不存在 | 检查设备序列号 |
| 10031 | 子账户或萤石用户没有权限 | 检查权限 |
| 20018 | 该用户不拥有该设备 | 检查设备归属 |
| 49999 | Data error | 未知错误 |
| 20007 | 设备不在线 | 设备离线 |
| 20008 | 设备响应超时 | 网络问题 |
| 20001 | 通道不存在 | 检查通道号 |
| 20035 | 该通道被隐藏 | 通道被隐藏 |
| 20015 | 设备不支持对讲 | 设备不支持此功能 |
| 111000 | 用户资源包余量不足 | 前往控制台购买 |
| 111012 | 文件下发失败，内部错误 | 参考下方内部错误码 |

### 内部错误码（111012 详情）

| 内部错误码 | 错误描述 | 说明 |
|-----------|----------|------|
| 1 | 排队超时 | 请求排队超时 |
| 2 | 处理超时 | 处理过程超时 |
| 3 | 设备链接失败 | 设备网络状况不佳 |
| 4 | 服务器内部错误 | 服务器问题 |
| 5 | 消息错误 | 消息格式问题 |
| 6 | 请求重定向 | 请求被重定向 |
| 7 | 无效 URL | URL 格式错误 |
| 8 | 认证 token 失败 | token 无效 |
| 9 | 验证码或者秘钥不匹配 | 验证失败 |
| 10 | 设备正在对讲 | 设备忙 |
| 11 | 设备通信超时 | 通信超时 |
| 12 | 设备不在线 | 设备离线 |
| 13 | 设备开启隐私保护 | 隐私模式 |
| 14 | token 无权限 | 权限不足 |
| 15 | session 不存在 | session 失效 |
| 16 | 验证 token 其他问题 | token 问题 |
| 17 | 设备监听超时 | 监听超时 |
| 18 | 设备链路断开 | 连接断开 |
| 19 | 隐私遮蔽 | 隐私保护 |
| 20 | 声源定位 | 声源功能 |
| 21 | ticket 认证失败 | ticket 问题 |
| 22 | ticket 未开启 | ticket 未启用 |
| 23 | ticket bizcode 错误 | bizcode 错误 |
| 1003 | 设备网络异常断开 | 网络问题 |
| 10001 | 创建 ticket 失败 | ticket 创建失败 |
| 10002 | 连接 tts 失败 | TTS 连接问题 |
| 10003 | 下载文件失败 | 文件下载失败 |
| 10004 | kafka 消息格式错误 | 消息格式问题 |
| 10005 | tts 断开连接 | TTS 断开 |
| 10006 | tts url 异常 | TTS URL 问题 |
| 10007 | 超过并发限制 | 并发超限 |
| 10008 | 连接已经存在，重复请求 | 重复请求 |
| 10009 | 发送 kafka 失败 | kafka 发送失败 |
| 10010 | 不支持的音频格式 | 格式不支持 |

---

## 完整工作流程

```
1. 抓图 → 2. 玩手机检测 → 3. 生成 TTS → 4. 上传语音 → 5. 下发到设备
```

### 步骤说明

1. **抓图**: 使用摄像头抓拍当前画面，获取图片 URL
2. **玩手机检测**: 调用接口 1，传入图片 URL，分析是否有人玩手机
3. **生成 TTS**: 如检测到玩手机，生成"检测到有人玩手机"语音文件
4. **上传语音**: 调用接口 2，上传语音文件，获取 fileUrl
5. **下发到设备**: 调用接口 3，传入 deviceSerial 和 fileUrl，设备播放告警

### 判断逻辑

从接口 1 的返回结果中判断是否玩手机：

```javascript
const isPhoneDetected = data.images?.[0]?.contentAnn?.bboxes?.some(bbox => 
  bbox.tagInfo?.labels?.some(label => 
    label.key === "behavior" && label.label === "play_phone" && label.labelWeight > 0.5
  )
);
```

```

### scripts/phone_detection_alert.py

```python
#!/usr/bin/env python3
"""
Phone Detection Alert Script

Detects phone usage from camera snapshot and sends voice alert to Ezviz device.

Workflow:
1. Capture snapshot from camera
2. Call phone detection API
3. If phone detected: generate TTS audio, upload to Ezviz, send to device
"""

import sys
import os
import json
import time
import hashlib
import hmac
import requests
from datetime import datetime

# Configuration
TOKEN_GET_API_URL = "https://open.ys7.com/api/lapp/token/get"  # 获取 AccessToken 接口
DEVICE_CAPTURE_API_URL = "https://open.ys7.com/api/lapp/device/capture"  # 设备抓拍图片接口
PHONE_DETECTION_API_URL = "https://open.ys7.com/api/service/intelligence/algo/analysis/play_phone_detection"  # 玩手机算法分析接口
VOICE_UPLOAD_API_URL = "https://open.ys7.com/api/lapp/voice/upload"  # 语音文件上传接口
VOICE_SEND_API_URL = "https://open.ys7.com/api/lapp/voice/send"  # 语音文件下发接口

# These should be provided by the user or environment
APP_KEY = os.environ.get("EZVIZ_APP_KEY", "")
APP_SECRET = os.environ.get("EZVIZ_APP_SECRET", "")
DEVICE_SERIAL = os.environ.get("EZVIZ_DEVICE_SERIAL", "")
CHANNEL_NO = os.environ.get("EZVIZ_CHANNEL_NO", "1")


def get_current_timestamp():
    """Get current Unix timestamp in milliseconds."""
    return int(time.time() * 1000)


def generate_sign(accessToken, timestamp):
    """Generate API signature for Ezviz API."""
    # sign = md5(accessToken + timestamp)
    sign_str = f"{accessToken}{timestamp}"
    return hashlib.md5(sign_str.encode()).hexdigest()


def get_access_token(app_key, app_secret):
    """
    Get access token using appKey and appSecret.
    
    API: POST /api/lapp/token/get
    Content-Type: application/x-www-form-urlencoded
    
    Args:
        app_key: Your application key
        app_secret: Your application secret
    
    Returns:
        dict: {success: bool, access_token: str, expire_time: int, error: str}
    """
    print(f"[INFO] Getting access token...")
    
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    
    data = {
        "appKey": app_key,
        "appSecret": app_secret
    }
    
    try:
        response = requests.post(
            TOKEN_GET_API_URL,
            headers=headers,
            data=data,
            timeout=30
        )
        
        result = response.json()
        # Don't log full response to avoid leaking credentials
        
        if result.get("code") == "200":
            data = result.get("data", {})
            expire_time = data.get("expireTime", 0)
            expire_str = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(expire_time / 1000))
            print(f"[SUCCESS] Token obtained, expires: {expire_str}")
            return {
                "success": True,
                "access_token": data.get("accessToken", ""),
                "expire_time": expire_time
            }
        else:
            error_msg = result.get("msg", "Unknown error")
            print(f"[ERROR] Get token failed: {error_msg}")
            return {
                "success": False,
                "error": error_msg,
                "code": result.get("code")
            }
    
    except Exception as e:
        print(f"[ERROR] Get token failed: {type(e).__name__}")
        return {
            "success": False,
            "error": str(e)
        }


def capture_device_image(access_token, device_serial, channel_no=1):
    """
    Capture image from device.
    
    API: POST /api/lapp/device/capture
    Content-Type: application/x-www-form-urlencoded
    
    Args:
        access_token: Ezviz access token
        device_serial: Device serial number (uppercase letters)
        channel_no: Channel number (default 1 for IPC)
    
    Returns:
        dict: {success: bool, pic_url: str, error: str}
    """
    print(f"[INFO] Capturing image from device: {device_serial}")
    
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    
    data = {
        "accessToken": access_token,
        "deviceSerial": device_serial.upper(),
        "channelNo": str(channel_no)
    }
    
    try:
        response = requests.post(
            DEVICE_CAPTURE_API_URL,
            headers=headers,
            data=data,
            timeout=30
        )
        
        result = response.json()
        # Don't log full response
        
        if result.get("code") == "200":
            data = result.get("data", {})
            pic_url = data.get("picUrl", "")
            print(f"[SUCCESS] Image captured: {pic_url[:50]}...")
            return {
                "success": True,
                "pic_url": pic_url,
                "expire_hours": 2
            }
        else:
            error_msg = result.get("msg", "Capture failed")
            print(f"[ERROR] Capture failed: {error_msg}")
            return {
                "success": False,
                "error": error_msg,
                "code": result.get("code")
            }
    
    except Exception as e:
        print(f"[ERROR] Device capture failed: {type(e).__name__}")
        return {
            "success": False,
            "error": str(e)
        }


def capture_snapshot(camera_id=None, output_path="/tmp/camera_snapshot.jpg"):
    """
    Capture a snapshot from the camera and upload to get URL.
    In OpenClaw environment, this would use the nodes tool.
    For standalone use, implement camera capture logic here.
    
    Returns:
        tuple: (local_path, image_url) - The URL is needed for the detection API
    """
    print(f"[INFO] Capturing snapshot from camera: {camera_id or 'default'}")
    # This is a placeholder - in OpenClaw, use nodes tool with camera_snap action
    # The API requires a URL, not local file, so you need to upload the image somewhere
    # For now, return the output path - user needs to implement image hosting
    return output_path, None  # URL needs to be provided by user


def detect_phone_usage(image_url, access_token, img_width=1280, img_height=720):
    """
    Call phone detection API to analyze if someone is using a phone.
    
    API: POST /api/service/intelligence/algo/analysis/play_phone_detection
    Content-Type: application/json
    
    Args:
        image_url: URL of the image to analyze
        access_token: Ezviz access token
        img_width: Image width (default 1280)
        img_height: Image height (default 720)
    
    Returns:
        dict: Detection result with 'is_phone_detected' boolean and details
    """
    print(f"[INFO] Analyzing image for phone usage: {image_url}")
    
    headers = {
        "accessToken": access_token,
        "Content-Type": "application/json"
    }
    
    payload = {
        "taskType": "",
        "dataInfo": [
            {
                "data": image_url,
                "type": "url",
                "modal": "image"
            }
        ],
        "dataParams": [
            {
                "modal": "image",
                "img_width": img_width,
                "img_height": img_height
            }
        ]
    }
    
    try:
        response = requests.post(
            PHONE_DETECTION_API_URL,
            headers=headers,
            json=payload,
            timeout=30
        )
        
        result = response.json()
        # Don't log full response to avoid leaking data
        
        # Parse response based on actual API structure
        # meta.code == 200 means success
        if result.get("meta", {}).get("code") == 200:
            data = result.get("data", {})
            images = data.get("images")
            
            # Handle case where images is None (no people detected in image)
            if images is None:
                print("[INFO] No people detected in image")
                return {
                    "success": True,
                    "is_phone_detected": False,
                    "details": data,
                    "confidence": 0.0
                }
            
            # Check if any bbox has play_phone behavior
            is_detected = False
            confidence = 0.0
            
            for img in images:
                content_ann = img.get("contentAnn", {})
                bboxes = content_ann.get("bboxes", [])
                for bbox in bboxes:
                    tag_info = bbox.get("tagInfo", {})
                    labels = tag_info.get("labels", [])
                    for label in labels:
                        if label.get("key") == "behavior" and label.get("label") == "play_phone":
                            is_detected = True
                            confidence = max(confidence, label.get("labelWeight", 0))
            
            if is_detected:
                print(f"[ALERT] Phone usage detected! (confidence: {confidence:.2f})")
            else:
                print("[INFO] No phone usage detected")
            
            return {
                "success": True,
                "is_phone_detected": is_detected,
                "details": data,
                "confidence": confidence
            }
        else:
            meta = result.get("meta", {})
            error_msg = meta.get("message", "Unknown error")
            print(f"[ERROR] Detection failed: {error_msg}")
            return {
                "success": False,
                "is_phone_detected": False,
                "error": error_msg,
                "code": meta.get("code")
            }
    
    except Exception as e:
        print(f"[ERROR] Phone detection failed: {type(e).__name__}")
        return {
            "success": False,
            "is_phone_detected": False,
            "error": str(e)
        }


def generate_tts_audio(text, output_path="/tmp/alert_audio.mp3"):
    """
    Generate TTS audio file using edge-tts (Microsoft Azure TTS).
    
    Args:
        text: Text to convert to speech
        output_path: Output audio file path
    
    Returns:
        str: Path to generated audio file
    """
    print(f"[INFO] Generating TTS audio: '{text}'")
    
    try:
        # Try to use edge-tts (free Microsoft Azure TTS)
        import asyncio
        import edge_tts
        
        async def generate_audio():
            communicate = edge_tts.Communicate(text, "zh-CN-XiaoxiaoNeural")
            await communicate.save(output_path)
        
        asyncio.run(generate_audio())
        print(f"[INFO] TTS audio saved to: {output_path}")
        
        # Verify file size
        file_size = os.path.getsize(output_path)
        print(f"[INFO] Audio file size: {file_size} bytes")
        
        if file_size < 1000:
            print("[WARNING] Audio file seems too small, may be invalid")
        
        return output_path
        
    except ImportError:
        print("[INFO] edge-tts not installed, trying gTTS...")
        try:
            from gtts import gTTS
            
            tts = gTTS(text=text, lang='zh-cn')
            tts.save(output_path)
            print(f"[INFO] TTS audio saved to: {output_path}")
            return output_path
            
        except ImportError:
            print("[WARNING] No TTS library available. Creating placeholder...")
            print("[INFO] Install edge-tts: pip install edge-tts")
            print("[INFO] Or install gTTS: pip install gTTS")
            
            # Download a sample beep sound as placeholder
            import urllib.request
            try:
                # Use a minimal valid MP3
                sample_url = "https://www.soundjay.com/button/beep-07.wav"
                urllib.request.urlretrieve(sample_url, output_path.replace(".mp3", ".wav"))
                output_path = output_path.replace(".mp3", ".wav")
                print(f"[INFO] Downloaded sample sound: {output_path}")
                return output_path
            except:
                print("[ERROR] Could not download sample sound")
                return None
    
    except Exception as e:
        print(f"[ERROR] TTS generation failed: {e}")
        return None


def upload_voice_file(audio_path, access_token, voice_name=None):
    """
    Upload voice file to Ezviz cloud.
    
    Returns:
        dict: Upload result with voice URL if successful
    """
    print(f"[INFO] Uploading voice file: {audio_path}")
    
    if voice_name is None:
        voice_name = f"alert_{datetime.now().strftime('%Y%m%d_%H%M%S')}.mp3"
    
    timestamp = get_current_timestamp()
    sign = generate_sign(access_token, timestamp)
    
    params = {
        "accessToken": access_token,
        "time": timestamp,
        "sign": sign
    }
    
    try:
        with open(audio_path, 'rb') as f:
            files = {
                'voiceFile': (voice_name, f, 'audio/mpeg')
            }
            data = {
                'voiceName': voice_name
            }
            response = requests.post(
                VOICE_UPLOAD_API_URL,
                params=params,
                files=files,
                data=data,
                timeout=30
            )
        
        result = response.json()
        # Don't log full response
        
        if result.get("code") == "200":
            data = result.get("data", {})
            voice_url = data.get("url", "")
            print(f"[SUCCESS] Voice uploaded: {voice_url[:50]}...")
            return {
                "success": True,
                "voice_url": voice_url,
                "voice_name": data.get("name", voice_name)
            }
        else:
            error_msg = result.get("msg", "Upload failed")
            print(f"[ERROR] Voice upload failed: {error_msg}")
            return {
                "success": False,
                "error": error_msg
            }
    
    except Exception as e:
        print(f"[ERROR] Voice upload failed: {type(e).__name__}")
        return {
            "success": False,
            "error": str(e)
        }


def send_voice_to_device(device_serial, voice_url, access_token, channel_no=1):
    """
    Send voice file to Ezviz device for playback.
    
    API: POST /api/lapp/voice/send
    Content-Type: application/x-www-form-urlencoded
    
    Args:
        device_serial: Device serial number
        voice_url: Voice file URL (from upload API response)
        access_token: Ezviz access token
        channel_no: Channel number (default 1)
    """
    print(f"[INFO] Sending voice to device: {device_serial}")
    
    headers = {
        "accessToken": access_token,
        "Content-Type": "application/x-www-form-urlencoded"
    }
    
    data = {
        "deviceSerial": device_serial,
        "channelNo": str(channel_no),
        "fileUrl": voice_url
    }
    
    try:
        response = requests.post(
            VOICE_SEND_API_URL,
            headers=headers,
            data=data,
            timeout=30
        )
        
        result = response.json()
        # Don't log full response
        
        if result.get("code") == "200":
            print(f"[SUCCESS] Alert sent to device {device_serial}!")
            return {
                "success": True,
                "message": "Voice sent successfully"
            }
        else:
            error_msg = result.get("msg", "Send failed")
            print(f"[ERROR] Voice send failed: {error_msg}")
            return {
                "success": False,
                "error": error_msg,
                "code": result.get("code")
            }
    
    except Exception as e:
        print(f"[ERROR] Voice send failed: {type(e).__name__}")
        return {
            "success": False,
            "error": str(e)
        }


def parse_device_list(device_str, channel_str="1"):
    """
    Parse device list from string.
    
    Supports formats:
    - Single device: "BA5918431" or "BA5918431,1"
    - Multiple devices: "BA5918431,BA5918432,BA5918433"
    - Multiple devices with channels: "BA5918431:1,BA5918432:1,BA5918433:2"
    - Mixed: "BA5918431,BA5918432:2,BA5918433"
    
    Args:
        device_str: Device serial(s), comma-separated
        channel_str: Default channel number (used if not specified per device)
    
    Returns:
        list: [(device_serial, channel_no), ...]
    """
    devices = []
    
    if not device_str:
        return devices
    
    for item in device_str.split(","):
        item = item.strip()
        if not item:
            continue
        
        if ":" in item:
            # Format: deviceSerial:channelNo
            parts = item.split(":")
            serial = parts[0].strip().upper()
            channel = int(parts[1].strip()) if len(parts) > 1 else int(channel_str)
        else:
            # Format: deviceSerial only
            serial = item.upper()
            channel = int(channel_str)
        
        devices.append((serial, channel))
    
    return devices


def main():
    """
    Main workflow: get token -> capture images from multiple devices -> detect phones -> alert.
    
    Complete flow:
    1. Get access token (appKey + appSecret)
    2. For each device:
       - Capture device image
       - Detect phone usage
       - If detected: generate TTS -> upload voice -> send to device
    """
    print("=" * 60)
    print("Phone Detection Alert System")
    print("=" * 60)
    
    # Configuration from environment or arguments
    app_key = APP_KEY or sys.argv[1] if len(sys.argv) > 1 else ""
    app_secret = APP_SECRET or sys.argv[2] if len(sys.argv) > 2 else ""
    device_str = DEVICE_SERIAL or sys.argv[3] if len(sys.argv) > 3 else ""
    channel_str = CHANNEL_NO or (sys.argv[4] if len(sys.argv) > 4 else "1")
    
    # Parse device list (supports multiple devices)
    devices = parse_device_list(device_str, channel_str)
    
    if not devices:
        print("[ERROR] At least one device serial required.")
        print("[INFO] Format: \"device1,device2,device3\" or \"device1:1,device2:2\"")
        print("[INFO] Set EZVIZ_DEVICE_SERIAL env var or pass as argument.")
        sys.exit(1)
    
    print(f"[INFO] Detected {len(devices)} device(s): {devices}")
    
    # Step 0: Get access token
    if not app_key or not app_secret:
        print("[ERROR] APP_KEY and APP_SECRET required.")
        print("[INFO] Set EZVIZ_APP_KEY and EZVIZ_APP_SECRET env vars.")
        sys.exit(1)
    
    token_result = get_access_token(app_key, app_secret)
    if not token_result["success"]:
        print(f"[ERROR] Failed to get token: {token_result.get('error', 'Unknown error')}")
        sys.exit(1)
    
    access_token = token_result["access_token"]
    expire_time = token_result.get("expire_time", 0)
    expire_str = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(expire_time / 1000))
    print(f"[INFO] Token obtained, expires: {expire_str}")
    
    # Track detection results
    detection_summary = {
        "total": len(devices),
        "detected": 0,
        "not_detected": 0,
        "failed": 0,
        "alerts_sent": 0
    }
    
    # Process each device
    for device_serial, channel_no in devices:
        print(f"\n{'='*60}")
        print(f"[Device] {device_serial} (Channel: {channel_no})")
        print(f"{'='*60}")
        
        # Step 1: Capture device image
        capture_result = capture_device_image(access_token, device_serial, channel_no)
        
        if not capture_result["success"]:
            print(f"[ERROR] Capture failed: {capture_result.get('error', 'Unknown error')}")
            detection_summary["failed"] += 1
            continue
        
        image_url = capture_result["pic_url"]
        print(f"[INFO] Image captured: {image_url[:50]}...")
        
        # Step 2: Detect phone usage
        detection_result = detect_phone_usage(image_url, access_token)
        
        if not detection_result["success"]:
            print(f"[ERROR] Detection failed: {detection_result.get('error', 'Unknown error')}")
            detection_summary["failed"] += 1
            continue
        
        if detection_result["is_phone_detected"]:
            print(f"[ALERT] Phone usage detected! (confidence: {detection_result['confidence']:.2f})")
            detection_summary["detected"] += 1
            
            # Step 3: Generate TTS audio
            alert_text = "检测到有人玩手机"
            audio_path = generate_tts_audio(alert_text)
            
            if not audio_path:
                print(f"[ERROR] TTS generation failed")
                continue
            
            # Step 4: Upload voice file
            upload_result = upload_voice_file(audio_path, access_token)
            
            if not upload_result["success"]:
                print(f"[ERROR] Voice upload failed: {upload_result.get('error', 'Unknown error')}")
                continue
            
            voice_url = upload_result["voice_url"]
            print(f"[INFO] Voice uploaded: {voice_url[:60]}...")
            
            # Step 5: Send voice to device
            send_result = send_voice_to_device(device_serial, voice_url, access_token, channel_no)
            
            if send_result["success"]:
                print(f"[SUCCESS] Alert sent to device {device_serial}!")
                detection_summary["alerts_sent"] += 1
            else:
                print(f"[ERROR] Failed to send alert: {send_result.get('error', 'Unknown error')}")
        else:
            print("[INFO] No phone usage detected.")
            detection_summary["not_detected"] += 1
    
    if not detection_result["success"]:
        print(f"[ERROR] Detection failed: {detection_result.get('error', 'Unknown error')}")
        sys.exit(1)
    
    if detection_result["is_phone_detected"]:
        print(f"[ALERT] Phone usage detected! (confidence: {detection_result['confidence']:.2f})")
        
        # Step 3: Generate TTS audio
        alert_text = "检测到有人玩手机"
        audio_path = generate_tts_audio(alert_text)
        
        # Step 4: Upload voice file
        upload_result = upload_voice_file(audio_path, access_token)
        
        if not upload_result["success"]:
            print(f"[ERROR] Voice upload failed: {upload_result.get('error', 'Unknown error')}")
            sys.exit(1)
        
        voice_url = upload_result["voice_url"]
        print(f"[INFO] Voice uploaded: {voice_url}")
        
        # Step 5: Send voice to device
        if device_serial:
            send_result = send_voice_to_device(
                device_serial, voice_url, access_token, channel_no
            )
            
            if send_result["success"]:
                print("[SUCCESS] Alert sent to device successfully!")
            else:
                print(f"[ERROR] Failed to send alert: {send_result.get('error', 'Unknown error')}")
                sys.exit(1)
        else:
            print("[INFO] Device serial not provided. Voice uploaded but not sent.")
        
    else:
        print("[INFO] No phone usage detected. No alert needed.")
        detection_summary["not_detected"] += 1
    
    # Print summary
    print(f"\n{'='*60}")
    print("DETECTION SUMMARY")
    print(f"{'='*60}")
    print(f"  Total devices:     {detection_summary['total']}")
    print(f"  Phone detected:    {detection_summary['detected']}")
    print(f"  Not detected:      {detection_summary['not_detected']}")
    print(f"  Failed:            {detection_summary['failed']}")
    print(f"  Alerts sent:       {detection_summary['alerts_sent']}")
    print(f"{'='*60}")
    
    if detection_summary["detected"] > 0:
        print(f"[RESULT] {detection_summary['detected']}/{detection_summary['total']} device(s) detected phone usage")
        print(f"[ACTION] {detection_summary['alerts_sent']} alert(s) sent successfully")
    else:
        print("[RESULT] No phone usage detected on any device")
    
    print("=" * 60)
    print("Workflow completed")
    print("=" * 60)


def test_alert_workflow(access_token, device_str, channel_str="1"):
    """
    Test the complete alert workflow without detection.
    This is for testing purposes only.
    
    Supports multiple devices.
    """
    print("=" * 60)
    print("Phone Detection Alert System - TEST MODE")
    print("=" * 60)
    
    devices = parse_device_list(device_str, channel_str)
    
    if not devices:
        print("[ERROR] No devices specified")
        return False
    
    print(f"[TEST] Testing alert workflow on {len(devices)} device(s): {devices}")
    
    results = {"success": 0, "failed": 0}
    
    for device_serial, channel_no in devices:
        print(f"\n{'='*60}")
        print(f"[Device] {device_serial} (Channel: {channel_no})")
        print(f"{'='*60}")
        
        # Step 1: Generate TTS audio
        alert_text = "检测到有人玩手机"
        print(f"\n[Step 1] Generating TTS: '{alert_text}'")
        audio_path = generate_tts_audio(alert_text)
        
        if not audio_path or not os.path.exists(audio_path):
            print(f"[ERROR] TTS audio file not created: {audio_path}")
            results["failed"] += 1
            continue
        
        # Step 2: Upload voice file
        print(f"\n[Step 2] Uploading voice file: {audio_path}")
        upload_result = upload_voice_file(audio_path, access_token)
        
        if not upload_result["success"]:
            print(f"[ERROR] Voice upload failed: {upload_result.get('error', 'Unknown error')}")
            results["failed"] += 1
            continue
        
        voice_url = upload_result["voice_url"]
        print(f"[SUCCESS] Voice uploaded: {voice_url[:60]}...")
        
        # Step 3: Send voice to device
        print(f"\n[Step 3] Sending voice to device: {device_serial}")
        send_result = send_voice_to_device(device_serial, voice_url, access_token, channel_no)
        
        if send_result["success"]:
            print(f"[SUCCESS] Alert sent to device {device_serial}!")
            results["success"] += 1
        else:
            print(f"[ERROR] Failed to send alert: {send_result.get('error', 'Unknown error')}")
            results["failed"] += 1
    
    # Print summary
    print(f"\n{'='*60}")
    print("TEST SUMMARY")
    print(f"{'='*60}")
    print(f"  Total devices:  {len(devices)}")
    print(f"  Success:        {results['success']}")
    print(f"  Failed:         {results['failed']}")
    print(f"{'='*60}")
    
    if results["failed"] == 0:
        print("TEST COMPLETED - All steps passed!")
        return True
    else:
        print("TEST COMPLETED - Some devices failed")
        return False


if __name__ == "__main__":
    # Check if test mode is requested
    if len(sys.argv) > 5 and sys.argv[5] == "--test":
        # Test mode: skip detection, test alert workflow
        app_key = APP_KEY or sys.argv[1] if len(sys.argv) > 1 else ""
        app_secret = APP_SECRET or sys.argv[2] if len(sys.argv) > 2 else ""
        device_str = DEVICE_SERIAL or sys.argv[3] if len(sys.argv) > 3 else ""
        channel_str = CHANNEL_NO or (sys.argv[4] if len(sys.argv) > 4 else "1")
        
        if not app_key or not app_secret:
            print("[ERROR] APP_KEY and APP_SECRET required")
            sys.exit(1)
        
        # Get token
        token_result = get_access_token(app_key, app_secret)
        if not token_result["success"]:
            print(f"[ERROR] Failed to get token: {token_result.get('error')}")
            sys.exit(1)
        
        access_token = token_result["access_token"]
        
        if not device_str:
            print("[ERROR] Device serial required")
            sys.exit(1)
        
        success = test_alert_workflow(access_token, device_str, channel_str)
        sys.exit(0 if success else 1)
    else:
        # Normal mode: full detection workflow
        main()

```