docs/examples: 扫码登录改为官方 POST+local_token_list+iLink-App-Id/ClientVersion 头 + 完整状态机 + 换码竞态警告
This commit is contained in:
AskaEth committed
1 parent
38b776705d
commit
8a36915377
6 files changed
+566
-96
No files matched your search
@@ -1,7 +1,7 @@
|
||||
#!/usr/bin/env python3
|
||||
# ilink_bridge.py — 微信 ClawBot iLink 最小示例（Python）
|
||||
# 流程: 扫码登录 → 长轮询收消息 → 回声回复
|
||||
# 依赖: pip install requests
|
||||
# 流程: 扫码登录（官方 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"
|
||||
@@ -17,12 +17,17 @@ def save_state():
|
||||
json.dump(state, f, indent=2)
|
||||
os.chmod(STATE_FILE, 0o600)
|
||||
|
||||
# ---------- 通用请求头: 鉴权 + 防重放 X-WECHAT-UIN ----------
|
||||
# ---------- 通用请求头: 鉴权 + 防重放 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']}"
|
||||
@@ -31,22 +36,87 @@ def headers():
|
||||
def api_post(path, body, timeout=45):
|
||||
return requests.post(BASE_URL + path, headers=headers(), json=body, timeout=timeout).json()
|
||||
|
||||
# ---------- 接口1: GET /ilink/bot/get_bot_qrcode 获取登录二维码 ----------
|
||||
# ---------- 接口2: GET /ilink/bot/get_qrcode_status 轮询扫码状态 ----------
|
||||
# ---------- 接口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():
|
||||
qr = requests.get(f"{BASE_URL}/ilink/bot/get_bot_qrcode?bot_type=3").json()
|
||||
print("请用微信扫码:", qr["qrcode_img_content"])
|
||||
# 状态机: 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:
|
||||
s = requests.get(f"{BASE_URL}/ilink/bot/get_qrcode_status",
|
||||
params={"qrcode": qr["qrcode"]},
|
||||
headers={"iLink-App-ClientVersion": "1"}).json()
|
||||
if s.get("status") == "confirmed": # 状态流转: wait → scaned → confirmed
|
||||
state.update(bot_token=s["bot_token"], bot_id=s["ilink_bot_id"],
|
||||
owner_id=s["ilink_user_id"])
|
||||
save_state()
|
||||
print("登录成功, bot:", s["ilink_bot_id"])
|
||||
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("扫码超时")
|
||||
|
||||
|
||||
Reference in new issue
Block a user