19 KiB
19 KiB
05 — 实现指南与代码骨架
给出两个可直接改造的最小实现骨架：Node.js（零依赖，官方 demo 的整理版）和 Python（requests 版）。 均已按本文档的可靠性建议补强（游标持久化、退避、-14 处理）。媒体收发先略，接入文本跑通后再按 03 文档加。
1. 最小流程回顾
1. 扫码拿 bot_token（一次性） ← auth
2. loop: ← poller
resp = POST getupdates(cursor)
保存新 cursor
for msg in resp.msgs:
text = msg.item_list[*].text_item.text
reply = 你的AI后端(text) ← backend
POST sendmessage(reply, msg.context_token)
2. Node.js 骨架（bridge.mjs）
零第三方依赖（Node 18+ 自带 fetch）。二维码展示可加装 qrcode-terminal。
// bridge.mjs — 微信 ClawBot iLink Bridge 最小骨架
import { readFileSync, writeFileSync, existsSync } from "node:fs";
import { randomBytes } from "node:crypto";
const BASE_URL = "https://ilinkai.weixin.qq.com";
const STATE_FILE = "./bridge-state.json";
// ---------- 状态持久化 ----------
const state = existsSync(STATE_FILE)
? JSON.parse(readFileSync(STATE_FILE, "utf8"))
: { bot_token: null, cursor: "", contexts: {} };
const saveState = () =>
writeFileSync(STATE_FILE, JSON.stringify(state, null, 2), { mode: 0o600 });
// ---------- iLink HTTP 封装 ----------
// 注: get_bot_qrcode / get_qrcode_status（预绑定阶段）无 token 时不带 Authorization，
// 但 iLink-App-Id / iLink-App-ClientVersion / AuthorizationType / X-WECHAT-UIN 恒带。
function headers() {
const uin = Buffer.from(String(randomBytes(4).readUInt32BE(0))).toString("base64");
return {
"Content-Type": "application/json",
AuthorizationType: "ilink_bot_token",
"X-WECHAT-UIN": uin,
"iLink-App-Id": "bot",
// 2.4.8 -> (2&0xff)<<16 | (4&0xff)<<8 | (8&0xff) = 132104（数字字符串，非 "1"）
"iLink-App-ClientVersion": "132104",
...(state.bot_token ? { Authorization: `Bearer ${state.bot_token}` } : {}),
};
}
async function apiPost(path, body, timeoutMs = 45_000) {
const res = await fetch(`${BASE_URL}${path}`, {
method: "POST",
headers: headers(),
body: JSON.stringify(body),
signal: AbortSignal.timeout(timeoutMs),
});
return res.json();
}
// ---------- 1. 扫码登录（一次性） ----------
// 接口1: POST /ilink/bot/get_bot_qrcode（官方姿势：body={"local_token_list":[]}，非裸 GET）
async function refreshQR() {
const qr = await fetch(`${BASE_URL}/ilink/bot/get_bot_qrcode?bot_type=3`, {
method: "POST",
headers: headers(),
body: JSON.stringify({ local_token_list: state.bot_token ? [state.bot_token] : [] }),
}).then((r) => r.json());
console.log("请用微信扫码:", qr.qrcode_img_content);
// qrcode_img_content 是 liteapp 跳转 URL，须作为二维码内容编码展示，勿把 qr.qrcode 当二维码
// 终端显示: npm i qrcode-terminal，然后 qrcode.generate(qr.qrcode_img_content, {small:true})
return qr.qrcode;
}
async function login() {
// 状态机: wait / scaned / scaned_but_redirect / need_verifycode /
// verify_code_blocked / binded_redirect / confirmed / expired
// 两大坑（务必读 docs/02 §4.5）：
// ① 码服务端 TTL≈60s，生成后须在窗口内扫（超窗服务端自返回 expired，此处换新码）。
// ② 「换码竞态」：一旦收到 scaned（或 scaned_but_redirect / need_verifycode），
// 绝不能再换码——换码会砍掉"扫码→确认"会话，微信端显示成功但后端永远收不到 confirmed。
let base = BASE_URL;
let qrcode = await refreshQR();
let verifyCode = null;
const deadline = Date.now() + 8 * 60_000;
while (Date.now() < deadline) {
let ep = `${base}/ilink/bot/get_qrcode_status?qrcode=${encodeURIComponent(qrcode)}`;
if (verifyCode) ep += `&verify_code=${encodeURIComponent(verifyCode)}`;
const s = await fetch(ep, { headers: headers() }).then((r) => r.json());
switch (s.status) {
case "wait":
process.stdout.write(`\r等待扫码... status=${s.status} `);
await new Promise((r) => setTimeout(r, 1000));
continue;
case "scaned":
verifyCode = null;
console.log("\n已扫码，请在手机上确认...（此后绝不换码，避免砍掉确认会话）");
await new Promise((r) => setTimeout(r, 1000));
continue;
case "scaned_but_redirect":
verifyCode = null;
if (s.redirect_host) {
base = `https://${s.redirect_host}`;
console.log("IDC 重定向，切换轮询至", base);
}
await new Promise((r) => setTimeout(r, 1000));
continue;
case "need_verifycode": {
const { createInterface } = await import("node:readline");
const rl = createInterface({ input: process.stdin, output: process.stdout });
verifyCode = await new Promise((r) => rl.question("请在手机微信上输入显示的数字后回车: ", r));
rl.close();
continue;
}
case "verify_code_blocked":
verifyCode = null;
console.log("\n配对码多次输入错误，重新获取二维码");
qrcode = await refreshQR();
continue;
case "binded_redirect":
console.log("该 bot 已绑定过本实例，视为连接成功");
return;
case "confirmed":
state.bot_token = s.bot_token;
state.bot_id = s.ilink_bot_id;
state.owner_id = s.ilink_user_id;
state.baseurl = s.baseurl;
saveState();
console.log("登录成功, bot:", s.ilink_bot_id);
return;
case "expired":
verifyCode = null;
console.log("\n二维码已过期，重新获取");
qrcode = await refreshQR();
continue;
default:
console.log("\n未知状态:", s.status);
await new Promise((r) => setTimeout(r, 1000));
continue;
}
}
throw new Error("扫码超时，请重试");
}
// ---------- 2. AI 后端调用（换成你的项目） ----------
async function askYourAI(peerId, text) {
// 例：调本地 HTTP 后端
// const r = await fetch("http://127.0.0.1:8000/chat", {
// method: "POST",
// headers: { "Content-Type": "application/json" },
// body: JSON.stringify({ session_id: peerId, text }),
// }).then(r => r.json());
// return r.reply;
return `你说的是：${text}`; // 回声占位
}
// ---------- 3. 发送回复 ----------
async function sendText(toUserId, text, contextToken) {
// 长文本简单分片
for (let i = 0; i < text.length; i += 1500) {
await apiPost("/ilink/bot/sendmessage", {
msg: {
to_user_id: toUserId,
client_id: crypto.randomUUID(),
message_type: 2, // bot 消息
message_state: 2, // FINISH
context_token: contextToken,
item_list: [{ type: 1, text_item: { text: text.slice(i, i + 1500) } }],
},
});
}
}
// ---------- 4. 长轮询主循环 ----------
async function pollLoop() {
let failures = 0;
for (;;) {
try {
const resp = await apiPost("/ilink/bot/getupdates", {
get_updates_buf: state.cursor,
base_info: { channel_version: "1.0.2" },
});
if (resp.errcode === -14) {
console.error("会话过期(errcode=-14)，请删除 state 文件重新扫码");
process.exit(2);
}
if (resp.ret !== 0 && resp.errcode !== undefined) {
throw new Error(`getupdates 错误: ${JSON.stringify(resp)}`);
}
if (resp.get_updates_buf) {
state.cursor = resp.get_updates_buf;
saveState(); // 游标立即落盘
}
for (const msg of resp.msgs ?? []) {
if (msg.message_type !== 1) continue; // 只处理用户消息
const text = (msg.item_list ?? [])
.filter((it) => it.type === 1)
.map((it) => it.text_item?.text ?? "")
.join("\n")
.trim();
if (!text) continue; // 媒体消息此处略，按 03 文档扩展
state.contexts[msg.from_user_id] = msg.context_token;
saveState();
try {
const reply = await askYourAI(msg.from_user_id, text);
await sendText(msg.from_user_id, reply, msg.context_token);
} catch (e) {
console.error("处理消息失败:", e);
await sendText(msg.from_user_id, "处理时出错了，请重试", msg.context_token)
.catch(() => {});
}
}
failures = 0;
} catch (e) {
// AbortError(长轮询超时) 视为正常，直接下一轮
if (e.name === "TimeoutError" || e.name === "AbortError") continue;
failures++;
const wait = failures <= 3 ? 2000 : 30_000;
console.error(`轮询异常(${failures}次)，${wait / 1000}s 后重试:`, e.message);
await new Promise((r) => setTimeout(r, wait));
}
}
}
// ---------- main ----------
if (!state.bot_token) await login();
await pollLoop();
运行：
node bridge.mjs # 首次会打印二维码链接，微信扫码确认
3. Python 骨架（bridge.py）
依赖只有 requests。二维码展示可加装 qrcode + 终端渲染。
# bridge.py — 微信 ClawBot iLink Bridge 最小骨架
import json, os, time, uuid, base64, random, requests
BASE_URL = "https://ilinkai.weixin.qq.com"
STATE_FILE = "./bridge-state.json"
# ---------- 状态持久化 ----------
state = {"bot_token": None, "cursor": "", "contexts": {}}
if os.path.exists(STATE_FILE):
state.update(json.load(open(STATE_FILE)))
def save_state():
with open(STATE_FILE, "w") as f:
json.dump(state, f, indent=2)
os.chmod(STATE_FILE, 0o600)
# ---------- iLink HTTP 封装 ----------
# 注: get_bot_qrcode / get_qrcode_status（预绑定阶段）无 token 时不带 Authorization，
# 但 iLink-App-Id / iLink-App-ClientVersion / AuthorizationType / X-WECHAT-UIN 恒带。
def headers():
h = {
"Content-Type": "application/json",
"AuthorizationType": "ilink_bot_token",
"X-WECHAT-UIN": base64.b64encode(str(random.getrandbits(32)).encode()).decode(),
"iLink-App-Id": "bot",
# 2.4.8 -> (2&0xff)<<16 | (4&0xff)<<8 | (8&0xff) = 132104（数字字符串，非 "1"）
"iLink-App-ClientVersion": "132104",
}
if state["bot_token"]:
h["Authorization"] = f"Bearer {state['bot_token']}"
return h
def api_post(path, body, timeout=45):
r = requests.post(BASE_URL + path, headers=headers(), json=body, timeout=timeout)
return r.json()
# ---------- 1. 扫码登录（一次性） ----------
# 接口1: POST /ilink/bot/get_bot_qrcode（官方姿势：body={"local_token_list":[]}，非裸 GET）
def fetch_qrcode():
return requests.post(
f"{BASE_URL}/ilink/bot/get_bot_qrcode?bot_type=3",
headers=headers(), json={"local_token_list": []}, timeout=15,
).json()
def login():
# 状态机: wait / scaned / scaned_but_redirect / need_verifycode /
# verify_code_blocked / binded_redirect / confirmed / expired
# 两大坑（务必读 docs/02 §4.5）：
# ① 码服务端 TTL≈60s，生成后须在窗口内扫（超窗服务端自返回 expired，此处换新码）。
# ② 「换码竞态」：一旦收到 scaned（或 scaned_but_redirect / need_verifycode），
# 绝不能再换码——换码会砍掉"扫码→确认"会话，微信端显示成功但后端永远收不到 confirmed。
base_url = BASE_URL
qr = fetch_qrcode()
qrcode = qr["qrcode"]
print("请用微信扫码:", qr["qrcode_img_content"])
# qrcode_img_content 是 liteapp 跳转 URL，须作为二维码内容编码展示，勿把 qr["qrcode"] 当二维码
deadline = time.time() + 8 * 60
verify_code = None
while time.time() < deadline:
params = {"qrcode": qrcode}
if verify_code:
params["verify_code"] = verify_code
s = requests.get(f"{base_url}/ilink/bot/get_qrcode_status",
params=params, headers=headers(), timeout=35).json()
st = s.get("status")
if st == "wait":
print("等待扫码...")
elif st == "scaned":
verify_code = None
print("已扫码，请在手机上确认...（此后绝不换码，避免砍掉确认会话）")
elif st == "scaned_but_redirect":
verify_code = None
if s.get("redirect_host"):
base_url = f"https://{s['redirect_host']}"
print("IDC 重定向，切换轮询至", base_url)
elif st == "need_verifycode":
verify_code = input("请在手机微信上输入显示的数字后回车: ").strip()
continue
elif st == "verify_code_blocked":
verify_code = None
print("配对码多次输入错误，重新获取二维码")
qr = fetch_qrcode(); qrcode = qr["qrcode"]; print("新二维码:", qr["qrcode_img_content"])
continue
elif st == "binded_redirect":
print("该 bot 已绑定过本实例，视为连接成功")
return
elif st == "confirmed":
state.update(bot_token=s["bot_token"], bot_id=s["ilink_bot_id"],
owner_id=s["ilink_user_id"], baseurl=s.get("baseurl"))
save_state()
print("登录成功, bot:", s["ilink_bot_id"])
return
elif st == "expired":
# 码过期（服务端作废）→ 换新码；仅在 expired 时换，scaned 后绝不换
verify_code = None
print("二维码已过期，重新获取")
qr = fetch_qrcode(); qrcode = qr["qrcode"]; print("新二维码:", qr["qrcode_img_content"])
continue
else:
print("未知状态:", st)
time.sleep(1)
raise RuntimeError("扫码超时")
# ---------- 2. AI 后端调用（换成你的项目） ----------
def ask_your_ai(peer_id: str, text: str) -> str:
# r = requests.post("http://127.0.0.1:8000/chat",
# json={"session_id": peer_id, "text": text}).json()
# return r["reply"]
return f"你说的是：{text}"
# ---------- 3. 发送回复 ----------
def send_text(to_user_id, text, context_token):
for i in range(0, len(text), 1500):
api_post("/ilink/bot/sendmessage", {
"msg": {
"to_user_id": to_user_id,
"client_id": str(uuid.uuid4()),
"message_type": 2,
"message_state": 2,
"context_token": context_token,
"item_list": [{"type": 1, "text_item": {"text": text[i:i+1500]}}],
}
})
# ---------- 4. 长轮询主循环 ----------
def poll_loop():
failures = 0
while True:
try:
resp = api_post("/ilink/bot/getupdates", {
"get_updates_buf": state["cursor"],
"base_info": {"channel_version": "1.0.2"},
})
if resp.get("errcode") == -14:
print("会话过期(errcode=-14)，请删除 state 文件重新扫码")
raise SystemExit(2)
if resp.get("ret", 0) != 0:
raise RuntimeError(f"getupdates 错误: {resp}")
if resp.get("get_updates_buf"):
state["cursor"] = resp["get_updates_buf"]
save_state()
for msg in resp.get("msgs") or []:
if msg.get("message_type") != 1:
continue
text = "\n".join(
it.get("text_item", {}).get("text", "")
for it in msg.get("item_list", []) if it.get("type") == 1
).strip()
if not text:
continue
state["contexts"][msg["from_user_id"]] = msg["context_token"]
save_state()
try:
reply = ask_your_ai(msg["from_user_id"], text)
send_text(msg["from_user_id"], reply, msg["context_token"])
except Exception as e:
print("处理消息失败:", e)
try:
send_text(msg["from_user_id"], "处理时出错了，请重试",
msg["context_token"])
except Exception:
pass
failures = 0
except requests.exceptions.Timeout:
continue # 长轮询正常超时
except Exception as e:
failures += 1
wait = 2 if failures <= 3 else 30
print(f"轮询异常({failures}次)，{wait}s 后重试:", e)
time.sleep(wait)
if __name__ == "__main__":
if not state["bot_token"]:
login()
poll_loop()
4. 接入你自己 AI 项目的改造点
只需替换 askYourAI / ask_your_ai 一个函数：
| 你的后端形态 | 改法 |
|---|---|
| HTTP API（FastAPI/Flask/Express…） | 函数里 POST 过去，session_id 传 peer_id |
| OpenAI 兼容接口 | 函数里组 messages 调 chat/completions，历史自己存（JSON/SQLite，按 peer_id 分键） |
| LangChain / LangGraph | 函数里调 chain.invoke，memory 用 peer_id 做会话键 |
| 本地 Agent 框架（Claude Agent SDK 等） | 参考 cc-weixin 的 askClaude() 包装 |
| 命令行工具 | subprocess 调一下取 stdout（注意超时与转义） |
5. 常见坑（按踩坑频率排序）
- 游标没存/存晚了 → 重启后消息重复轰炸。收到响应立刻落盘。
- 回复忘带
context_token→ 消息发不出或关联不上对话。 - HTTP 客户端超时 < 长轮询超时 → 每轮都假性超时。骨架里给了 45s。
- Markdown 直接发 → 微信不渲染，先转纯文本。
- 只处理
item_list[0]→ 图文混排丢内容。 - token 提交进 git → 泄漏=bot 被接管。state 文件加
.gitignore。 - 消息处理抛异常没兜住 → 整个轮询循环挂掉。骨架里已包 try。
- 多开实例 → 同 token 多处轮询互相抢游标，状态错乱。单实例跑。
6. 下一步扩展路线
- 文本收发跑通（本骨架）
- 接入你的 AI 后端 + 会话历史
- 入站图片/语音（CDN 下载 + AES 解密，见 03 文档）
sendtyping（处理慢时提升体验）- 出站图片/文件（getuploadurl + CDN 上传）
- 白名单、限流、多通道（按 04 文档模式 D 演进）
7. 参考项目清单
| 项目 | 值得借鉴的部分 |
|---|---|
| hao-ji-xing/openclaw-weixin | demo.mjs 裸调最小实现、协议逆向文档 |
| hao-ji-xing/cc-weixin | Agent SDK 的接入包装方式 |
| openclaw-java/openclaw-weixin-sdk | 完整工程化：状态存储、退避策略、-14 暂停窗口、typing、媒体服务 |
| openakita/openakita | Python 适配器、多通道网关抽象、媒体惰性下载策略 |
@tencent-weixin/openclaw-weixin（npm） |
官方协议实现，字段有疑问时的最终依据：npm pack @tencent-weixin/openclaw-weixin |