#!/usr/bin/env python3 # ilink_bridge.py — 微信 ClawBot iLink 最小示例（Python） # 流程: 扫码登录（官方 POST get_bot_qrcode + local_token_list + 完整状态机）→ 长轮询收消息 → 回声回复 # 依赖: pip install requests（二维码展示可加装 qrcode，非必需） import json, os, time, uuid, base64, random, requests BASE_URL = "https://ilinkai.weixin.qq.com" STATE_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "bridge-state.json") # ---------- 状态持久化: bot_token / 游标 / context_token ---------- 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) # ---------- 通用请求头: 鉴权 + 防重放 X-WECHAT-UIN + iLink-App 标识 ---------- def headers(): # 注: get_bot_qrcode / get_qrcode_status（预绑定阶段）无 token，不加 Authorization； # 但 iLink-App-Id / iLink-App-ClientVersion / AuthorizationType / X-WECHAT-UIN 恒带。 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): return requests.post(BASE_URL + path, headers=headers(), json=body, timeout=timeout).json() # ---------- 接口1: POST /ilink/bot/get_bot_qrcode 获取登录二维码（官方姿势） ---------- def fetch_qrcode(): # 官方为 POST + body {"local_token_list":[...]}，非裸 GET（裸 GET 也能拿到码但非官方姿势）。 # local_token_list: 本实例已登录的 bot token 列表；首次绑定传空数组 []。若已持久化其它 bot， # 可收集 token 传入（最多 10 个），官方据此识别（较可靠地触发 binded_redirect）。 return requests.post(f"{BASE_URL}/ilink/bot/get_bot_qrcode?bot_type=3", headers=headers(), json={"local_token_list": []}, timeout=15).json() def show_qrcode(url): # qrcode_img_content 是 liteapp 跳转 URL（liteapp.weixin.qq.com/q/...），须作为"二维码内容" # 用二维码库编码展示；勿把响应里的 qrcode 字段当二维码（会扫出一串文本）。 try: import qrcode qr = qrcode.QRCode(border=1) qr.add_data(url) qr.make() qr.print_ascii() return except Exception: pass # 无 qrcode 库时降级：打印 URL（微信扫此 URL 同样可完成确认，只是非标准"二维码"展示）。 print("二维码内容(URL):", url) # ---------- 接口2: GET /ilink/bot/get_qrcode_status 轮询扫码状态（完整状态机） ---------- 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 # scaned_but_redirect 时切换轮询 host qr = fetch_qrcode() qrcode = qr["qrcode"] show_qrcode(qr["qrcode_img_content"]) 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 host = s.get("redirect_host") if host: base_url = f"https://{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"]; show_qrcode(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.get("ilink_bot_id"), owner_id=s.get("ilink_user_id"), baseurl=s.get("baseurl")) save_state() print("登录成功, bot:", s.get("ilink_bot_id")) return elif st == "expired": # 码过期（服务端作废）→ 换新码；仅在 expired 时换，scaned 后绝不换 verify_code = None print("二维码已过期，重新获取") qr = fetch_qrcode(); qrcode = qr["qrcode"]; show_qrcode(qr["qrcode_img_content"]) continue else: print("未知状态:", st) time.sleep(1) raise RuntimeError("扫码超时") # ---------- 你的 AI 后端: 替换成自己项目的调用 ---------- def ask_your_ai(peer_id: str, text: str) -> str: # 例: requests.post("http://127.0.0.1:8000/chat", # json={"session_id": peer_id, "text": text}).json()["reply"] return f"你说的是：{text}" # ---------- 接口4: POST /ilink/bot/sendmessage 发送消息 ---------- def send_text(to_user_id, text, context_token): for i in range(0, max(len(text), 1), 1500): # 长文本分片 api_post("/ilink/bot/sendmessage", { "msg": { "to_user_id": to_user_id, "client_id": str(uuid.uuid4()), # 去重 ID "message_type": 2, # 2 = bot 消息 "message_state": 2, # 2 = FINISH "context_token": context_token, # 必填! 来自入站消息 "item_list": [{"type": 1, "text_item": {"text": text[i:i+1500]}}], } }) # ---------- 接口3: POST /ilink/bot/getupdates 长轮询收消息 ---------- 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)，请删除 bridge-state.json 重新扫码") 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: send_text(msg["from_user_id"], ask_your_ai(msg["from_user_id"], text), msg["context_token"]) except Exception as e: print("处理消息失败:", e) failures = 0 except requests.exceptions.Timeout: continue # 长轮询正常超时 except Exception as e: failures += 1 wait = 2 if failures <= 3 else 30 # 退避: 2s ×3 → 30s print(f"轮询异常({failures}次)，{wait}s 后重试:", e) time.sleep(wait) if __name__ == "__main__": if not state["bot_token"]: login() poll_loop()