182 lines
6.7 KiB
Python
182 lines
6.7 KiB
Python
"""探测二:微信 ClawBot 的【对话】与【主动推送】能力(命门验证)。
|
|
|
|
流程(都在一次运行里,不落库):
|
|
1. 扫码绑定拿 bot_token(同探测一)
|
|
2. getupdates 长轮询,等你给「微信 ClawBot」联系人发一条消息
|
|
3. 收到后,依次测三种发送,逐一报 ret:
|
|
A. 带 context_token 回复 -> 验「被动回复」是否通
|
|
B. 等 25s 后,用【同一个】context_token 再发 -> 验「开口一次后能否延迟主动推」
|
|
C. context_token 置空再发 -> 验「冷推(无 token)」是否被拒
|
|
判读:
|
|
A 通 = 双向对话成立
|
|
B 通 = 用户开口一次后可后续推送(简报可走"先开口、后定时推"的弱化版)
|
|
C 通 = 可冷推(几乎不可能,但要验)
|
|
B/C 都不通 = ClawBot 纯被动回复,定时主动推送这条路不成立
|
|
|
|
ASCII-only 输出。bot_token 不打印。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import os
|
|
import random
|
|
import sys
|
|
import time
|
|
|
|
import httpx
|
|
import segno
|
|
|
|
BASE = "https://ilinkai.weixin.qq.com"
|
|
QR_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "clawbot_qr.png")
|
|
CHANNEL_VER = "1.0.2"
|
|
|
|
|
|
def _uin() -> str:
|
|
return base64.b64encode(str(random.randint(0, 2**32 - 1)).encode()).decode()
|
|
|
|
|
|
def _headers(token: str | None = None) -> dict:
|
|
h = {
|
|
"Content-Type": "application/json",
|
|
"AuthorizationType": "ilink_bot_token",
|
|
"X-WECHAT-UIN": _uin(),
|
|
}
|
|
if token:
|
|
h["Authorization"] = f"Bearer {token}"
|
|
return h
|
|
|
|
|
|
def bind() -> tuple[str, str] | None:
|
|
print("[bind] GET get_bot_qrcode ...")
|
|
with httpx.Client(timeout=20) as c:
|
|
r = c.get(f"{BASE}/ilink/bot/get_bot_qrcode",
|
|
params={"bot_type": "3"}, headers=_headers())
|
|
if r.status_code != 200:
|
|
print(f"[FAIL] get_bot_qrcode http {r.status_code}: {r.text[:300]}")
|
|
return None
|
|
d = r.json()
|
|
qid = d.get("qrcode", "")
|
|
link = d.get("qrcode_img_content", "")
|
|
segno.make(link, error="m").save(QR_PATH, scale=8, border=3)
|
|
try:
|
|
import webbrowser
|
|
webbrowser.open("file://" + QR_PATH.replace("\\", "/"))
|
|
except Exception:
|
|
pass
|
|
print(f"[bind] QR opened -> {QR_PATH} SCAN IT NOW with phone WeChat.")
|
|
deadline = time.time() + 180
|
|
with httpx.Client(timeout=40) as c:
|
|
last = ""
|
|
while time.time() < deadline:
|
|
try:
|
|
r = c.get(f"{BASE}/ilink/bot/get_qrcode_status",
|
|
params={"qrcode": qid}, headers=_headers())
|
|
j = r.json()
|
|
st = j.get("status", "")
|
|
if st != last:
|
|
print(f"[bind] status={st!r}")
|
|
last = st
|
|
if st == "confirmed":
|
|
print("[bind] confirmed.")
|
|
return j.get("bot_token", ""), (j.get("baseurl") or BASE)
|
|
if st == "expired":
|
|
print("[bind] QR expired before scan.")
|
|
return None
|
|
except Exception as e:
|
|
print(f"[bind] poll err: {type(e).__name__}: {e}")
|
|
time.sleep(2)
|
|
print("[bind] timeout waiting for scan.")
|
|
return None
|
|
|
|
|
|
def _send(client: httpx.Client, token: str, to_user: str, text: str,
|
|
context_token: str) -> dict:
|
|
body = {
|
|
"msg": {
|
|
"to_user_id": to_user,
|
|
"message_type": 2,
|
|
"message_state": 2,
|
|
"context_token": context_token,
|
|
"item_list": [{"type": 1, "text_item": {"text": text}}],
|
|
}
|
|
}
|
|
r = client.post(f"{BASE}/ilink/bot/sendmessage",
|
|
json=body, headers=_headers(token))
|
|
try:
|
|
return {"http": r.status_code, "json": r.json()}
|
|
except Exception:
|
|
return {"http": r.status_code, "text": r.text[:300]}
|
|
|
|
|
|
def main() -> int:
|
|
b = bind()
|
|
if not b:
|
|
return 2
|
|
token, base_url = b
|
|
global BASE
|
|
BASE = base_url or BASE
|
|
|
|
print("[chat] now SEND a message (e.g. 'hi') to the WeChat ClawBot contact on your phone.")
|
|
print("[chat] waiting via getupdates (up to ~150s)...")
|
|
buf = ""
|
|
deadline = time.time() + 150
|
|
got = None
|
|
with httpx.Client(timeout=40) as c:
|
|
while time.time() < deadline and got is None:
|
|
try:
|
|
r = c.post(f"{BASE}/ilink/bot/getupdates",
|
|
json={"get_updates_buf": buf,
|
|
"base_info": {"channel_version": CHANNEL_VER}},
|
|
headers=_headers(token))
|
|
j = r.json()
|
|
buf = j.get("get_updates_buf", buf)
|
|
for m in j.get("msgs", []) or []:
|
|
txt = ""
|
|
for it in m.get("item_list", []) or []:
|
|
txt += (it.get("text_item", {}) or {}).get("text", "")
|
|
print(f"[chat] <- from={m.get('from_user_id')} text={txt!r}")
|
|
got = m
|
|
break
|
|
except Exception as e:
|
|
print(f"[chat] getupdates err: {type(e).__name__}: {e}")
|
|
time.sleep(2)
|
|
if got is None:
|
|
print("[chat] no message received in window. Re-run and send promptly after scan.")
|
|
return 1
|
|
|
|
to_user = got.get("from_user_id", "")
|
|
ctx = got.get("context_token", "")
|
|
print(f"[chat] captured to_user={to_user} context_token_len={len(ctx)}")
|
|
|
|
with httpx.Client(timeout=30) as c:
|
|
print("\n[testA] reply WITH context_token ...")
|
|
ra = _send(c, token, to_user, "[zcbot 测试A] 收到你的消息,这是带 token 的回复。", ctx)
|
|
print(f"[testA] result={ra}")
|
|
|
|
print("\n[testB] wait 25s, then push again with the SAME context_token (delayed proactive)...")
|
|
time.sleep(25)
|
|
rb = _send(c, token, to_user, "[zcbot 测试B] 这是25秒后用同一token的延迟主动推送。", ctx)
|
|
print(f"[testB] result={rb}")
|
|
|
|
print("\n[testC] push with EMPTY context_token (cold push) ...")
|
|
rc = _send(c, token, to_user, "[zcbot 测试C] 这是空token的冷推送。", "")
|
|
print(f"[testC] result={rc}")
|
|
|
|
def ok(r):
|
|
j = r.get("json") or {}
|
|
return r.get("http") == 200 and j.get("ret", -1) == 0
|
|
|
|
print("\n========== VERDICT ==========")
|
|
print(f"A reply(with token) : {'OK' if ok(ra) else 'FAIL'}")
|
|
print(f"B delayed push(same token) : {'OK' if ok(rb) else 'FAIL'}")
|
|
print(f"C cold push(empty token) : {'OK' if ok(rc) else 'FAIL'}")
|
|
print("Interpretation:")
|
|
print(" - A only -> reply-only; scheduled PROACTIVE push NOT possible.")
|
|
print(" - A+B -> after user opens chat once, delayed push works (weak push OK).")
|
|
print(" - C -> true cold push works (unlikely).")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|