192 lines
8.9 KiB
Python
192 lines
8.9 KiB
Python
#!/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()
|