weixin-ilink 0.3.4

微信 iLink Bot 协议的 Python SDK — 扫码登录、长轮询、收发文本/图片/语音/文件/视频、markdown 过滤、表情翻译、配对 ACL。

weixin-ilink

微信 iLink Bot 协议的 Python SDK —— 扫码登录、长轮询、收发文本/图片/语音/文件/视频、markdown 过滤、内置表情翻译、本地配对 ACL。

与腾讯官方 @tencent-weixin/openclaw-weixin 插件协议字段完全兼容。

---

安装

pip install weixin-ilink              # 核心（文本 + 图片 + 视频 + 文件）
pip install "weixin-ilink[qr]"        # + qrcode（终端显示 ASCII 二维码）
pip install "weixin-ilink[voice]"     # + pilk（入站语音 SILK → WAV 解码）
pip install "weixin-ilink[all]"       # 全部

Python ≥ 3.9。核心依赖：requests + cryptography（CDN 加密必需）。

---

快速上手

把下面这段存成 my_bot.py：

from weixin_ilink import WeixinBot

bot = WeixinBot.from_login(save_to="creds.json")

@bot.on_text
def handle(msg):
    msg.reply_text(f"收到 [OK]: {msg.text}")  # 用户看到 "收到 👌: hello"

@bot.on_image
def save_pic(msg):
    msg.save("inbox/pic.jpg")
    msg.reply_text("图片已收到")

bot.run()

跑：

pip install "weixin-ilink[qr]"   # 首次跑需要 qrcode 扩展画二维码
python my_bot.py

终端会打出 ASCII 二维码，微信扫码 → 看到 bot running — account=... 即成功。

登录成功后会在当前目录生成：

• creds.json — 扫码凭据（botToken / accountId / baseUrl）
• creds.json.sync — 长轮询 cursor，重启自动续传

更多示例见 examples/simple_bot.py / media_bot.py。

---

目录

• 安装
• 快速上手
• 架构
• 登录流程
• 构造 WeixinBot
• 处理入站消息
• IncomingMessage 参考
• 主动发送消息
• Markdown 过滤
• 表情翻译
• 配对 ACL
• 语音：入站解码，出站注意事项
• 错误处理与重连
• 底层 API (ILinkClient)
• 能力与限制

---

架构

mermaid 图在 GitHub 上能渲染，PyPI 上不渲染。PyPI 读者请跳转 GitHub README。

模块结构

flowchart TB
    subgraph HL [高层 API]
        Bot[WeixinBot]
        Msg[IncomingMessage]
        Login["login()"]
    end

    subgraph LL [底层 API]
        Client[ILinkClient]
        Auth[auth.login_with_qr]
    end

    subgraph Protocol [协议原语]
        Api[api.py<br/>5 个 CGI 端点]
        Types[types.py<br/>枚举]
        Cdn[cdn.py<br/>AES-128-ECB]
        Media[media.py<br/>上传 + 消息项构造]
    end

    subgraph Content [内容处理]
        Markdown[markdown.py<br/>微信兼容过滤器]
        Emoji[emoji.py<br/>标签 → Unicode]
        Voice[voice.py<br/>SILK ↔ WAV/PCM]
    end

    subgraph Ops [运维]
        Pairing[pairing.py<br/>allowFrom ACL]
    end

    Bot --> Client
    Bot --> Msg
    Bot --> Markdown
    Bot --> Emoji
    Bot --> Pairing
    Msg --> Client
    Login --> Auth
    Client --> Api
    Client --> Media
    Auth --> Api
    Api --> Types
    Media --> Cdn
    Media --> Api

    style HL fill:#e1f5ff
    style LL fill:#fff4e1
    style Protocol fill:#f0f0f0
    style Content fill:#f5e1ff
    style Ops fill:#e1ffe1

入站消息流程

sequenceDiagram
    autonumber
    participant User as 微信用户
    participant WX as WeChat / iLink 服务器
    participant Bot as WeixinBot
    participant H as 你的 handler

    Note over Bot,WX: 长轮询循环，cursor 自动落盘
    Bot->>+WX: POST ilink/bot/getupdates (cursor)
    User->>WX: 发消息 (文本/图片/语音/文件)
    WX-->>-Bot: msgs[] (+ 新 cursor)

    loop 每条 USER 消息的每个 item
        Bot->>Bot: 按 from_user 缓存 context_token
        Bot->>Bot: 包装成 IncomingMessage
        Bot->>+H: 分发到 @on_text / @on_image / ...
        alt 是 image/voice/file/video
            H->>Client: msg.download() → AES-ECB 解密
            Client->>WX: GET CDN 下载链接
            WX-->>Client: 密文
            Client-->>H: 明文字节
        end
        H->>Bot: msg.reply_text("收到 [OK]")
        Bot->>Bot: auto_emoji: [OK] → 👌
        Bot->>+WX: POST ilink/bot/sendmessage
        WX-->>-User: 推送回复
    end

出站媒体上传流程

sequenceDiagram
    autonumber
    participant H as 你的 handler
    participant Bot as WeixinBot
    participant WX as iLink 服务器
    participant CDN as 微信 CDN

    H->>Bot: bot.send_image(uid, "pic.jpg", caption="图片 ☀️")
    Bot->>Bot: 读文件、算 md5、生成随机 aeskey + filekey

    Bot->>+WX: POST ilink/bot/getuploadurl<br/>(media_type=IMAGE, rawsize, md5, filesize, aeskey_hex)
    WX-->>-Bot: upload_full_url (或 upload_param + cdn_base)

    Bot->>Bot: AES-128-ECB + PKCS7 加密
    Bot->>+CDN: POST 密文 (application/octet-stream)
    CDN-->>-Bot: x-encrypted-param (= 下载令牌)

    Bot->>Bot: 构造 ImageItem：<br/>media.encrypt_query_param<br/>media.aes_key = base64(hex)<br/>encrypt_type=1, mid_size

    opt 有 caption
        Bot->>WX: POST sendmessage (TEXT item)
    end
    Bot->>WX: POST sendmessage (IMAGE item)

---

登录流程

三种方式都会拿到同样的 info_json（扁平 dict，类似 2FA 种子）：

{
  "botToken": "ilb_xxxxx...",
  "accountId": "<ilink_bot_id>",
  "baseUrl":   "https://<region>.weixin.qq.com",
  "userId":    "<扫码人的 ilink_user_id>"
}

方式 A — 函数调用，返回 info_json 由调用方保存

from weixin_ilink import login
info = login()                        # 终端打 ASCII 二维码
info = login(save_to="creds.json")    # 同时写入指定路径

方式 B — 登录 + 构造 bot 一步完成

from weixin_ilink import WeixinBot
bot = WeixinBot.from_login(save_to="creds.json")

方式 C — 已有凭据，直接复用

bot = WeixinBot(credentials=info_dict)          # 从内存
bot = WeixinBot(credentials_file="creds.json")  # 从文件

命名参考 google-auth / pyotp：credentials（dict）、credentials_file（路径）、save_to（落盘路径）。

自定义二维码渲染

def my_qr(url: str):
    print(f"用浏览器打开扫码: {url}")

info = login(on_qrcode=my_qr, on_status_change=lambda s: print(f"[{s}]"))

状态回调会传入 waiting / scanned / refreshing / redirected。

---

构造 WeixinBot

bot = WeixinBot(
    credentials=info,                  # dict
    # 或 credentials_file="creds.json",
    cdn_base_url=None,                 # CDN 备用域名（一般不用）
    cursor_file=None,                  # None 时默认 "<credentials_file>.sync"
    auto_save_cursor=True,             # 每次 poll 后自动写 cursor
    auto_emoji=True,                   # 所有出站文本自动翻译 [微笑] 为 🙂
)

字段	作用
credentials	info_json dict
credentials_file	JSON 文件路径；同时是 cursor 文件的锚点
cdn_base_url	仅在 getUploadUrl 不返回 upload_full_url 时需要
cursor_file	长轮询 cursor 的持久化路径（重启续传）
auto_save_cursor	False 时由调用方自行维护 cursor
auto_emoji	True（默认）时出站文本/markdown/caption 都会走 emoji.translate

读取 / 保存凭据

bot.info             # dict，等同 info_json
bot.account_id       # str
bot.save("backup.json")

---

处理入站消息

四种姿势任选。都在主线程跑，handler 抛异常会被捕获并打 traceback，不影响轮询主循环。

1. 按类型装饰器

@bot.on_text
def handle_text(msg): msg.reply_text(f"echo: {msg.text}")

@bot.on_image
def handle_image(msg): msg.reply_text("收到图片")

@bot.on_voice
def handle_voice(msg): msg.reply_text(f"语音转写: {msg.text}")

@bot.on_file
def handle_file(msg): msg.reply_text(f"收到文件: {msg.file_name}")

@bot.on_video
def handle_video(msg): pass

bot.run()

2. 万能 handler

@bot.on_message
def handle(msg):
    if msg.is_text:
        ...

3. 手动迭代

for msg in bot.messages():
    if msg.is_text:
        msg.reply_text(...)

4. 停止

• Ctrl-C / SIGTERM —— 优雅退出，cursor 已落盘
• 代码内：任何线程调 bot.stop()（标志位控制，下次循环退出）

---

IncomingMessage 参考

一个 item 对应一个 IncomingMessage。一条微信消息可能带多个 item（比如 caption + image），拆开逐个 yield，handler 每次只处理一个载荷。

身份

属性	类型	说明
from_user	str	入站用户 ID
context_token	str	路由令牌，跨消息自动缓存
message_id	Optional[int]	每条入站消息唯一
session_id	Optional[str]	服务端对话线程 ID（主要是个参考，可不管）

类型判断

msg.is_text    # TEXT
msg.is_image   # IMAGE
msg.is_voice   # VOICE
msg.is_file    # FILE
msg.is_video   # VIDEO
msg.item_type  # int, MessageItemType

内容

属性	返回	来源
msg.text	str | None	TEXT 正文，或 VOICE 的 ASR 转写
msg.quoted_title	str | None	引用消息的摘要（如果是回复别的消息）
msg.file_name	str | None	FILE 消息的文件名
msg.voice_duration_ms	int	语音毫秒时长

下载媒体

# AES-ECB 解密，返回明文 bytes
data = msg.download()                          # bytes | None

# 或直接落盘（自动建父目录）
path = msg.save("./inbox/pic.jpg")             # Path

支持 IMAGE / VOICE / FILE / VIDEO，其他类型会抛错。

回复快捷方法

自动填 to_user_id + context_token：

msg.reply_text("hi")
msg.reply_markdown("**粗体** 和 `code`")
msg.reply_image("pic.jpg", caption="好看 ☀️")
msg.reply_video("clip.mp4")
msg.reply_file("doc.pdf", file_name="报告.pdf", caption="请查阅 [OK]")
msg.reply_voice("clip.silk", playtime_ms=3200)   # ⚠️ 通道通常会吞，见下文
msg.reply_typing()                                # "正在输入..."

---

主动发送消息

不在 handler 里主动发消息（比如定时推送），用 bot.send_*。这些方法需要 context_token —— 从该用户最近一条入站消息自动缓存：

# 任何入站消息后自动缓存，生命周期 = bot 进程
bot.send_text(user_id, "主动推送 [微笑]")

如果用户从未在当前进程发过消息给 bot，会抛 ValueError: 没有 context_token for <uid>。缓存仅存在于进程生命周期内，进程重启后需要收到一条新的入站消息才能重新获得该用户的 context_token。

完整 send API

bot.send_text(to, text, *, context_token=None, auto_emoji=None)
bot.send_markdown(to, md, *, context_token=None, auto_emoji=None)
bot.send_image(to, path, *, caption=None, context_token=None, auto_emoji=None)
bot.send_video(to, path, *, caption=None, context_token=None, auto_emoji=None)
bot.send_file(to, path, *, file_name=None, caption=None, context_token=None, auto_emoji=None)
bot.send_voice(to, silk_path, *, playtime_ms, sample_rate=24000, encode_type=SILK,
               context_token=None)   # 实验性
bot.send_typing(to, *, context_token=None)

auto_emoji=None 表示"用 bot 全局默认"。传 True / False 可逐调用覆盖。

---

Markdown 过滤

微信协议没有 "markdown" 消息类型 —— 所有文本都是纯文本。但客户端会就地渲染一部分 markdown 语法（粗体、代码块、表格），忽略其他（CJK 斜体、H5/H6、图片）。

本 SDK 带一个流式过滤器，把客户端不渲染的语法剥掉，让"带 markdown 的文本"到对方那里视觉干净：

语法	微信是否渲染	过滤器动作
``` 代码块	✅	保留
` 行内代码	✅	保留
| 表格 |	✅	保留
**粗体**	✅	保留
*italic*（英文）	✅	保留
*中文*（中文）	❌	脱掉星号
##### H5 / ###### H6	❌	脱掉 #####
 图片	❌	整段删除
--- 水平线	✅	保留

用法

msg.reply_markdown("## 标题\n**粗体** *中文* ![bad](x.png)")
# 实际发送: "## 标题\n**粗体** 中文 "

bot.send_markdown(user_id, "```python\nprint('hi')\n```")

流式（接 LLM 增量用）

from weixin_ilink.markdown import StreamingMarkdownFilter

f = StreamingMarkdownFilter()
for delta in llm.stream():
    piece = f.feed(delta)
    if piece:
        msg.reply_text(piece)
msg.reply_text(f.flush())

一次性函数

from weixin_ilink import markdown
clean = markdown.filter_markdown(raw_md)

---

表情翻译

微信内置了 ~108 个表情，在聊天里以 [微笑] / [奸笑] / [666] 这种文本标签表示。用户输入的标签会被客户端就地替换成黄脸图，但 bot 方向的文本不会再被客户端解析 —— 标签会原样显示成字面 [微笑]。

本 SDK 把每个标签映射到最接近的 Unicode emoji，这个会按系统字体正常渲染：

from weixin_ilink import emoji

emoji.translate("收到 [奸笑] [OK] [666]")
# → "收到 😏 👌 🔥"

默认在所有发送上启用

WeixinBot(auto_emoji=True)（默认）会让所有出站文本 / markdown / caption 都过 emoji.translate，免手动处理：

msg.reply_text("搞定 [OK]")           # 用户看到 "搞定 👌"
bot.send_image(uid, "p.jpg", caption="[666]")   # caption 也翻

关掉

bot = WeixinBot(credentials_file=..., auto_emoji=False)     # 全局
bot.send_text(uid, "字面标签 [微笑]", auto_emoji=False)     # 单条

其他工具

emoji.detag("嗨 [微笑] [某新表情]")   # → "嗨 🙂 某新表情"  (未识别标签也脱方括号)
emoji.known_tags()                    # → list[str]，已收录的所有标签
emoji.WECHAT_TO_UNICODE               # 完整映射 dict，可自由修改

注意

Unicode emoji 用各端系统字体渲染，不是微信自家那套黄脸。语义对得上，美术上略有差异。想要像素级还原微信黄脸，就得发 IMAGE —— 不在这个翻译器的范围内。

---

配对 ACL

对齐 OpenClaw 的账号级 allowFrom JSON（路径 ~/.openclaw/credentials/openclaw-weixin-<account>-allowFrom.json）。如果同一个账号同时用本 SDK 和官方 openclaw CLI，它们共用同一份 ACL 文件。

bot.allow(user_id)              # 加入白名单
bot.revoke(user_id)             # 移除
bot.is_allowed(user_id)         # bool
bot.allowed_users()             # list[str]

在 handler 里拦截未授权用户：

@bot.on_message
def gate(msg):
    if not bot.is_allowed(msg.from_user):
        msg.reply_text("未授权，请先配对")
        return
    # … 正常业务 …

可用环境变量 OPENCLAW_STATE_DIR / OPENCLAW_OAUTH_DIR 覆盖路径。

---

语音：入站解码，出站注意事项

入站：用户发语音给 bot

@bot.on_voice
def handle(msg):
    asr_text = msg.text                        # 微信自带 ASR 的转写文本
    duration = msg.voice_duration_ms

    # 原始 SILK 落盘
    silk_path = msg.save("inbox/voice.silk")

    # 可选：SILK → WAV（需要 `pip install "weixin-ilink[voice]"` 装 pilk）
    from weixin_ilink import voice
    wav_path = silk_path.with_suffix(".wav")
    voice.silk_to_wav(silk_path, wav_path)

出站：bot 发语音

⚠️ iLink 通道会静默丢弃 bot 方向的 VOICE 消息。HTTP API 返回 200，但客户端不会出现语音气泡。官方 openclaw-weixin 插件从不发 VOICE，统一用文件附件代替。

# ❌ 实测不出现语音气泡：
msg.reply_voice("clip.silk", playtime_ms=3000)

# ✅ 推荐这样做 —— 用户会看到一个音频文件卡片，点击即可播放：
msg.reply_file("clip.wav", caption="🎵 点击播放")

reply_voice / send_voice 保留是为了对称，将来服务端放开限制就能用。

---

错误处理与重连

poll 循环内置防御：

情况	行为
长轮询超时（35s）	静默继续
errcode=-14 session 过期	睡 1 小时并打印重登提示
瞬时网络错误 / 5xx	2s 退避；连续 3 次后 30s
handler 抛异常	打 traceback，不影响循环
cursor 持久化失败	记日志，循环继续（少量消息可能重放）

Ctrl-C 不会泄漏状态或留死 socket。

---

底层 API (ILinkClient)

给需要绕过 WeixinBot 的场景（或者只想发消息、不需要收）。完整文档见 weixin_ilink.client 的 docstring。

from weixin_ilink import ILinkClient, MessageType, MessageItemType

c = ILinkClient(base_url=..., token=..., cdn_base_url=...)
c.cursor = "restored_from_disk"

while True:
    resp = c.poll()
    for msg in resp.get("msgs") or []:
        if msg.get("message_type") != int(MessageType.USER):
            continue
        text = ILinkClient.extract_text(msg)
        ctx = msg["context_token"]
        c.send_text_chunked(msg["from_user_id"], f"echo: {text}", ctx)

其他底层方法

c.send_image(to, path, ctx, caption=None)
c.send_video(to, path, ctx, caption=None)
c.send_file(to, path, ctx, file_name=None, caption=None)
c.send_voice(to, silk_path, ctx, playtime_ms=..., sample_rate=24000)
c.send_typing(uid, ctx)
c.download_media_item(item)               # → bytes
c.get_config(uid, ctx)                    # 原始 config（含 typing_ticket 等）
c.get_upload_url(params)                  # 预签名 CDN 上传 URL

CDN 原语

from weixin_ilink import cdn

cdn.encrypt_aes_ecb(plaintext, key)           # bytes
cdn.decrypt_aes_ecb(ciphertext, key)          # bytes
cdn.aes_ecb_padded_size(n)                    # int  — PKCS7 填充后的大小
cdn.upload_buffer_to_cdn(buf=..., aeskey=..., filekey=..., upload_full_url=...)
cdn.download_and_decrypt(aes_key_base64=..., encrypted_query_param=..., cdn_base_url=...)

---

能力与限制

支持

能力	对应 API
扫码登录	login() / WeixinBot.from_login()
收/发文本	send_text / on_text / reply_text
超长文本自动分片	send_text（默认 4000 字一片）
收/发图片（AES-ECB CDN）	send_image / on_image / msg.save()
收/发视频	send_video / on_video
收/发文件附件	send_file / on_file
接收语音 + 微信 ASR 转写	on_voice / msg.text
SILK → WAV 解码	voice.silk_to_wav()（需 [voice] extras）
Markdown 过滤（微信兼容）	send_markdown / StreamingMarkdownFilter
内置表情 → Unicode	emoji.translate()（默认开启 auto_emoji=True）
"正在输入…" 指示	reply_typing / send_typing
本地配对 ACL（allowFrom）	bot.allow / is_allowed / ...
长轮询 cursor 持久化	自动；通过 cursor_file 可定制
优雅重连	bot.run() / bot.messages() 内置
登录时 IDC 跳转	透明处理 scaned_but_redirect
iLink-App-Id / ClientVersion 头	对齐官方 2.1.8

不支持 / 不可靠

能力	状态
出站语音消息	通道静默丢弃 —— 用 send_file(wav) 代替
群聊（group_id）	字段暴露了，但 iLink bot 通道实际上只做 1v1
自定义表情包 / 气泡	协议没对应 item type；可以当 IMAGE 发
@ 提醒	协议不支持
位置分享 / 卡片链接	协议不支持
出站引用回复（ref_msg）	ref_msg 实测只在入站出现

---

更新日志

版本变更详见 CHANGELOG.md。

许可证

MIT