# 微信 Bot iLink 协议重构方案

> **版本**: v1.0 | **日期**: 2026-08-15 | **作者**: 七虎AI 技术团队
> **背景**: 参考微信官方 iLink (ClawBot) 协议完整解析，对平台现有个人微信 Bot 实现做差距分析并重构
> **关联**: `im_bot_user.py`（前台用户级）、`admin/im_bot_admin.py`（后台平台级）、`im_llm.py`（EngineCore 薄壳）

---

## 一、协议背景：微信官方 iLink (ClawBot)

### 1.1 定位

iLink（智联）是腾讯微信 2026 年官方开放的个人微信机器人通信协议，对外产品名 **ClawBot**。这是**唯一合规**的个人微信 Bot 通道，有《微信 ClawBot 功能使用条款》法律背书。

| 维度 | 逆向 iPad/PC Hook（旧灰产） | iLink Bot API（官方合规） |
|------|---------------------------|--------------------------|
| 合规性 | 违反微信用户协议 | 官方开放，签署使用条款 |
| 封号风险 | 极高，大号易永久封禁 | 正常使用无封号风险 |
| 协议稳定性 | 微信升级即失效 | 服务端 API，版本可控 |
| 登录方式 | 模拟设备/劫持会话 | 微信扫码授权，临时 token |
| 通信方式 | 模拟心跳，触发风控 | 长轮询，无需公网 IP |

### 1.2 核心架构（三层）

```
上层（逻辑）: 业务代码 —— 消息收发、AI 应答、工具调用
中层（文件）: CDN 媒体加解密 —— AES-128-ECB + PKCS7
底层（通道）: iLink 长轮询 —— 消息通道（ilinkai.weixin.qq.com）
```

两个核心域名：
| 域名 | 用途 |
|------|------|
| `https://ilinkai.weixin.qq.com` | 所有 API 请求 |
| `https://novac2c.cdn.weixin.qq.com/c2c` | 图片/语音/视频媒体文件 |

### 1.3 七个 API 端点

| 端点 | 方法 | 功能 | 关键参数 |
|------|------|------|---------|
| `/ilink/bot/get_bot_qrcode` | GET | 获取登录二维码 | `?bot_type=3` |
| `/ilink/bot/get_qrcode_status` | GET | 轮询扫码状态 | `?qrcode=xxx` |
| `/ilink/bot/getupdates` | POST | 长轮询收消息（35s hold） | `{ get_updates_buf }` |
| `/ilink/bot/sendmessage` | POST | 发送消息 | `{ msg: { context_token, item_list } }` |
| `/ilink/bot/getuploadurl` | POST | 获取 CDN 上传地址 | `{ filekey, md5, len }` |
| `/ilink/bot/getconfig` | POST | 获取配置 | `{ base_info }` |
| `/ilink/bot/sendtyping` | POST | 发送"正在输入"状态 | `{ to_user_id, typing_ticket }` |

### 1.4 鉴权机制

```
Authorization: Bearer {bot_token}       # 扫码登录获取
AuthorizationType: ilink_bot_token       # 固定值
X-WECHAT-UIN: {随机uint32 → base64}      # 每次请求换新，防重放攻击
```

### 1.5 消息机制

- **长轮询**：`get_updates_buf` 是游标/位置书签，每次请求必须用返回的新值替换旧值，否则重复收消息死循环
- **context_token**：每条消息的"身份证"，回复必须原样回带，否则不关联到正确对话窗口
- **不能主动推送**：必须由用户先发消息触发
- **消息类型**：`type 1`文本 / `2`图片 / `3`语音 / `4`文件 / `5`视频
- **语音自带 ASR**：VoiceItem 含 `text` 字段，微信已做好语音转文字

---

## 二、现状分析

### 2.1 已实现能力（im_bot_user.py）

| 能力 | 实现 | 状态 |
|------|------|------|
| 扫码登录 | `get_bot_qrcode` + `get_qrcode_status` 轮询 | ✅ 完整 |
| 长轮询收消息 | `getupdates` + `get_updates_buf` 游标推进 | ✅ 完整 |
| 发消息 | `sendmessage` + `context_token` 回带 | ✅ 完整 |
| 鉴权 | `ilink_bot_token` + `X-WECHAT-UIN` 随机防重放 | ✅ 完整 |
| 文本分段 | `WECHAT_TEXT_MAX_LEN=1500` 超长分段 | ✅ 完整 |
| 会话过期检测 | `ret=-14/-2` 区分过期/参数错误 | ✅ 完整 |
| 用户级隔离 | 前台用户各自绑定，`im_bot_wechat_account` 按 user_id 隔离 | ✅ 完整 |
| token 持久化 | `bot_token`/`ilink_bot_id` 存 PG | ✅ 完整 |
| 对话历史 | `_log_chat` + 镜像到 AI 聊天窗 | ✅ 完整 |
| 异常日志 | 接收器异常写 `/tmp/wechat_receiver_err.log` | ✅ 完整 |
| 双 Claw 单引擎 | `im_llm.py` 薄壳复用 EngineCore | ✅ 完整 |

### 2.2 差距分析（核心问题）

| # | 能力 | 当前状态 | 影响 | 优先级 |
|---|------|---------|------|--------|
| 1 | **媒体内联发送** | ❌ 降级为纯文本链接（`🖼️ 图片已生成，点击查看：URL`） | 图片/视频不能内联显示，用户体验差 | 🔴 P0 |
| 2 | **sendtyping** | ❌ 未实现 | AI 回复等待期无"正在输入"提示 | 🟡 P1 |
| 3 | **语音消息** | ❌ 未提取 VoiceItem ASR text | 语音消息无法应答 | 🟡 P1 |
| 4 | **Session Guard 增强** | ⚠️ `-14` 仅标记 expired，无主动通知 | 会话过期用户无感知，需手动发现重扫 | 🟡 P1 |
| 5 | **CDN 下载解密** | ❌ 未实现（收图片/语音/文件无法解密） | 无法处理用户发来的图片/文件 | 🟢 P2 |

### 2.3 核心痛点详述：媒体内联发送

**当前实现**（`_wechat_send_media`）：
```python
def _wechat_send_media(...):
    # iLink 对 type=2/5 媒体 item 返回 ret=-2 invalid arguments
    # 降级为纯文本链接
    caption = "🖼️ 图片已生成，点击查看：" + relay_url
    return _wechat_send_message(...)  # 发文本
```

**根本原因**：iLink 媒体消息不是直接传 URL，而是需要走 **CDN 上传加密链路**：
1. 下载 AGNES 生成的图片字节
2. 用 `getuploadurl` 获取微信 CDN 上传地址
3. **AES-128-ECB 加密**图片字节
4. 上传到 `novac2c.cdn.weixin.qq.com`
5. 发消息时带 `download_param` + `aes_key`（type=2 图片 / type=5 视频）

当前代码缺少整个 CDN 加密链路，所以只能降级为文本链接。

---

## 三、重构方案

### 3.1 总体设计

```
用户微信 ──(文本/图片/语音)──> iLink getupdates
                                    │
                                    ▼
                        ┌── im_bot_user.py 接收循环 ──┐
                        │  提取 text / VoiceItem ASR  │
                        │  (P1) 语音转文字            │
                        └──────────┬─────────────────┘
                                   ▼
                        EngineCore._send_message_core
                        (双 Claw 单引擎，复用 LLM/MCP)
                                   │
                    ┌──────────────┼──────────────┐
                    ▼                             ▼
              文本回复                        媒体回复(__media__)
                    │                             │
                    │              ┌──────────────┴──────────────┐
                    │              │  (P0) 媒体内联发送链路      │
                    │              │  1. 下载 AGNES 图片字节     │
                    │              │  2. getuploadurl 拿地址     │
                    │              │  3. AES-128-ECB 加密       │
                    │              │  4. 上传微信 CDN           │
                    │              │  5. sendmessage type=2/5   │
                    │              └──────────────┬──────────────┘
                    ▼                             ▼
              sendmessage (文本)          sendmessage (媒体 item)
              + (P1) sendtyping 前置
```

### 3.2 分阶段实施

#### Phase 1（P0）：媒体内联发送 —— 核心

**目标**：图片/视频真正内联显示在微信，替代当前文本链接降级。

**新增模块** `app/api/im_ilink_media.py`：
```python
"""
iLink CDN 媒体加解密 —— 参考微信官方 iLink 协议
核心：AES-128-ECB + PKCS7 加密上传，下载解密
"""
import hashlib, os
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives import padding

def _aes_ecb_encrypt(plaintext: bytes, key: bytes) -> bytes:
    """AES-128-ECB + PKCS7 加密"""
    padder = padding.PKCS7(128).padder()
    padded = padder.update(plaintext) + padder.finalize()
    cipher = Cipher(algorithms.AES(key), modes.ECB())
    enc = cipher.encryptor()
    return enc.update(padded) + enc.finalize()

def _aes_ecb_decrypt(ciphertext: bytes, key: bytes) -> bytes:
    """AES-128-ECB 解密"""
    cipher = Cipher(algorithms.AES(key), modes.ECB())
    dec = cipher.decryptor()
    padded = dec.update(ciphertext) + dec.finalize()
    unpadder = padding.PKCS7(128).unpadder()
    return unpadder.update(padded) + unpadder.finalize()

def upload_media_to_wechat_cdn(bot_token: str, baseurl: str, media_bytes: bytes, filekey: str) -> dict:
    """六步上传流程：
    1. 计算 MD5
    2. 生成随机 AES-128 key (16字节)
    3. POST /ilink/bot/getuploadurl → upload_param
    4. AES-128-ECB 加密文件
    5. POST novac2c.cdn.weixin.qq.com/c2c/upload
    6. 返回 download_param + aes_key
    """
    md5 = hashlib.md5(media_bytes).hexdigest()
    aes_key = os.urandom(16)
    # ... getuploadurl 调用
    # ... 加密 + 上传
    return {"download_param": "...", "aes_key": aes_key.hex()}
```

**修改 `_wechat_send_media`**：
```python
def _wechat_send_media(baseurl, bot_token, to_user_id, context_token, media_type, url):
    # 1. 下载 AGNES 图片字节（走 HK 代理，trust_env=False）
    media_bytes = _download_media(url)
    # 2. 上传微信 CDN + AES 加密
    cdn = upload_media_to_wechat_cdn(bot_token, baseurl, media_bytes, filekey)
    # 3. 发媒体消息（type=2 图片 / type=5 视频）
    item = {
        "type": 2 if media_type == "image" else 5,
        "image_item": {  # 或 video_item
            "media": cdn["download_param"],
            "aes_key": cdn["aes_key"],
        }
    }
    return _wechat_post_msg(baseurl, bot_token, to_user_id, context_token, [item])
```

**依赖**：`cryptography` 库（需加入 requirements.txt）

#### Phase 2（P1）：sendtyping 正在输入

```python
def _wechat_send_typing(baseurl, bot_token, to_user_id, typing_ticket):
    """发消息前先发'正在输入'状态，改善等待体验"""
    return _post(baseurl + "/ilink/bot/sendtyping",
                 {"to_user_id": to_user_id, "typing_ticket": typing_ticket})

# 接收循环中，调用 _qihua_reply 前：
_wechat_send_typing(baseurl, bot_token, sender, typing_ticket)
```
- `typing_ticket` 从 `getconfig` 获取，按用户缓存 24h

#### Phase 3（P1）：语音消息 ASR

```python
def _wechat_extract_text(msg):
    """扩展：支持语音消息自带 ASR 结果"""
    for it in (msg.get("item_list") or []):
        if it.get("type") == 1:  # 文本
            ...
        elif it.get("type") == 3:  # 语音
            voice = it.get("voice_item", {})
            if voice.get("text"):  # 微信已做好语音转文字
                parts.append(voice["text"])
    ...
```

#### Phase 4（P1）：Session Guard 增强

```python
def _mark_wechat_expired(account_id):
    """增强：标记 expired 的同时，主动给用户发站内通知/邮件提醒重扫"""
    # 1. 标记 expired + 清 token（现有逻辑）
    # 2. 停掉接收线程
    # 3. 触发站内通知：'您的微信 Bot 会话已过期，请重新扫码绑定'
    # 4. 记录 audit log
```

#### Phase 5（P2）：接收侧媒体解密

```python
def _download_wechat_media(msg):
    """接收用户发来的图片/语音/文件：
    1. GET novac2c.cdn.weixin.qq.com/c2c/download?encrypted_query_param=xxx
    2. 用消息里的 aes_key 解密
    """
```

### 3.3 技术要点

| 要点 | 说明 |
|------|------|
| AES 模式 | AES-128-ECB（微信指定，非标准 CBC；一次性随机 key + HTTPS 保证安全） |
| 密钥长度 | 16 字节（128-bit） |
| 填充 | PKCS7 |
| 密文大小 | `ceil((plaintext_size+1)/16)*16` |
| 媒体域名 | `novac2c.cdn.weixin.qq.com/c2c/upload|download` |
| 错误码 | `0`成功 / `-14`会话过期 / `-2`参数错误 |

### 3.4 风险与规避

| 风险 | 规避 |
|------|------|
| ECB 模式理论不安全 | 每文件一次性随机 key + HTTPS，风险可控 |
| 媒体上传超时 | 大图/视频走异步，超时重试 + 降级文本链接兜底 |
| `cryptography` 依赖 | 需在容器 venv 安装，纳入 requirements.txt |
| iLink 媒体协议细节不明确 | 保留文本链接降级 fallback，逐步灰度 |

---

## 四、实施计划

| Phase | 内容 | 优先级 | 预估工作量 | 依赖 |
|-------|------|--------|-----------|------|
| P0 | 媒体内联发送（CDN AES 加密） | 🔴 高 | 2-3 天 | `cryptography` 库 |
| P1 | sendtyping 正在输入 | 🟡 中 | 0.5 天 | 无 |
| P1 | 语音 ASR | 🟡 中 | 0.5 天 | 无 |
| P1 | Session Guard 增强 | 🟡 中 | 1 天 | 站内通知 |
| P2 | 接收侧媒体解密 | 🟢 低 | 2 天 | P0 经验复用 |

---

## 五、验收标准

- [ ] P0：微信里生成的图片/视频**内联显示**，不再是文本链接
- [ ] P1：AI 回复前微信显示"正在输入"
- [ ] P1：发语音消息能正确应答（ASR 文本进入 LLM）
- [ ] P1：会话过期时用户收到明确重扫通知
- [ ] 双 Claw 单引擎不回归（`tests/test_dual_claw.py` 绿）
- [ ] pytest + vitest 全绿

---

## 六、参考资料

- 微信官方 iLink (ClawBot) 协议（CSDN: 2401_88147261 / HiWangWenBing / bytenote）
- OpenClaw 框架 `@tencent-weixin/openclaw-weixin` 增强：Session Guard / Config Cache / SILK 转码 / CDN 加解密
- 平台现状代码：`app/api/im_bot_user.py`、`app/api/admin/im_bot_admin.py`、`app/api/im_llm.py`