docs: weixin ClawBot bridge protocol and examples

examples/python/ilink_bridge.py

@@ -0,0 +1,121 @@
+#!/usr/bin/env python3
+# ilink_bridge.py — 微信 ClawBot iLink 最小示例（Python）
+# 流程: 扫码登录 → 长轮询收消息 → 回声回复
+# 依赖: pip install requests
+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 ----------
+def headers():
+    h = {
+        "Content-Type": "application/json",
+        "AuthorizationType": "ilink_bot_token",
+        "X-WECHAT-UIN": base64.b64encode(str(random.getrandbits(32)).encode()).decode(),
+    }
+    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: GET /ilink/bot/get_bot_qrcode 获取登录二维码 ----------
+# ---------- 接口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"])
+    deadline = time.time() + 8 * 60
+    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"])
+            return
+        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()