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