免翻墙！微信官方通道直连 AI 助手：iLink / ClawBot 扫码登录完整教程

#!/usr/bin/env python3
"""微信 iLink 桥接：扫码登录 + 长轮询收发。只用标准库 urllib（+ qrcode 画二维码）。"""

import argparse, base64, json, os, random, sys, time
import urllib.parse, urllib.request, urllib.error, uuid
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent
STATE_DIR = BASE_DIR / "state"          # 登录态等敏感文件，只放这里
INBOX_DIR = BASE_DIR / "inbox"          # 收到的消息
PENDING_DIR = BASE_DIR / "outbox" / "pending"  # 待发送的回复
SENT_DIR = BASE_DIR / "outbox" / "sent"         # 已发送的回复
SESSION_FILE = STATE_DIR / "session.json"
LOGIN_FILE = STATE_DIR / "login.json"
CURSOR_FILE = STATE_DIR / "cursor.txt"
PID_FILE = STATE_DIR / "serve.pid"
NEEDS_RELOGIN = STATE_DIR / "NEEDS_RELOGIN"
QR_PNG = BASE_DIR / "qr.png"

DEFAULT_BASE = "https://ilinkai.weixin.qq.com"
ILINK_APP_ID = "bot"
ILINK_APP_CLIENT_VERSION = str((2 << 16) | (4 << 8) | 9)  # 2.4.9 -> 132105
CHANNEL_VERSION = "2.4.9"
BOT_AGENT = "OpenClaw"
BOT_TYPE = "3"
QR_POLL_TIMEOUT = 40
UPDATES_POLL_TIMEOUT = 40
API_TIMEOUT = 15
STALE_TOKEN_ERRCODE = -14  # token 失效的唯一信号，没有刷新机制

def log(msg):
    print(f"[{time.strftime('%H:%M:%S')}] {msg}", flush=True)

def mask(s, keep=4):
    s = str(s or "")
    return s[:keep] + "..." if len(s) > keep else "****"

def atomic_write_json(path: Path, obj, mode=0o600):
    """原子写入：先写临时文件再 rename，不会留下写一半的文件。"""
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_name(path.name + ".tmp")
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(obj, f, ensure_ascii=False, indent=2)
    os.chmod(tmp, mode)
    os.replace(tmp, path)

def atomic_write_text(path: Path, text, mode=0o600):
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_name(path.name + ".tmp")
    with open(tmp, "w", encoding="utf-8") as f:
        f.write(text)
    os.chmod(tmp, mode)
    os.replace(tmp, path)

def rand_uin():
    n = random.getrandbits(32)
    return base64.b64encode(str(n).encode()).decode()

def common_headers():
    return {"iLink-App-Id": ILINK_APP_ID,
            "iLink-App-ClientVersion": ILINK_APP_CLIENT_VERSION}

def post_headers(token=None):
    h = {"Content-Type": "application/json",
         "AuthorizationType": "ilink_bot_token",
         "X-WECHAT-UIN": rand_uin()}
    h.update(common_headers())
    if token:
        h["Authorization"] = f"Bearer {token}"
    return h

def base_info():
    return {"channel_version": CHANNEL_VERSION, "bot_agent": BOT_AGENT}

def http_post(base, endpoint, body: dict, token=None, timeout=API_TIMEOUT,
              with_base_info=True):
    if with_base_info:
        body = dict(body); body["base_info"] = base_info()
    raw = json.dumps(body, ensure_ascii=False).encode("utf-8")
    url = base.rstrip("/") + "/" + endpoint.lstrip("/")
    req = urllib.request.Request(url, data=raw, headers=post_headers(token), method="POST")
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            text = r.read().decode("utf-8", "replace").strip()
    except urllib.error.HTTPError as e:
        raise RuntimeError(f"POST {endpoint} HTTP {e.code}")
    return json.loads(text) if text and text != "{}" else {"ret": 0}

def http_get(base, endpoint, timeout=QR_POLL_TIMEOUT):
    url = base.rstrip("/") + "/" + endpoint.lstrip("/")
    req = urllib.request.Request(url, headers=common_headers(), method="GET")
    with urllib.request.urlopen(req, timeout=timeout) as r:
        text = r.read().decode("utf-8", "replace").strip()
    return json.loads(text) if text else {}

def load_session():
    if not SESSION_FILE.exists():
        return None
    return json.loads(SESSION_FILE.read_text(encoding="utf-8"))

class TokenDead(Exception): pass

def check_stale(resp, where):
    if resp.get("errcode") == STALE_TOKEN_ERRCODE or resp.get("ret") == STALE_TOKEN_ERRCODE:
        raise TokenDead(f"{where}: token 失效 (errcode -14)")

def mark_needs_relogin(reason):
    atomic_write_text(NEEDS_RELOGIN, f"{time.strftime('%Y-%m-%d %H:%M:%S')} {reason}\n")

def cmd_login(_args):
    log("正在申请登录二维码 ...")
    resp = http_post(DEFAULT_BASE, f"ilink/bot/get_bot_qrcode?bot_type={BOT_TYPE}",
                     {"local_token_list": []}, token=None,
                     with_base_info=False, timeout=API_TIMEOUT)
    qrcode, img_content = resp.get("qrcode"), resp.get("qrcode_img_content")
    if not qrcode or not img_content:
        raise RuntimeError(f"二维码接口返回异常: {json.dumps(resp)[:200]}")
    atomic_write_json(LOGIN_FILE, {"qrcode": qrcode, "base": DEFAULT_BASE,
                                   "created_at": time.time()})
    import qrcode as qrmod
    qr = qrmod.QRCode(border=2); qr.add_data(img_content); qr.make(fit=True)
    qr.make_image(fill_color="black", back_color="white").save(str(QR_PNG))
    log(f"二维码已保存: {QR_PNG}（5 分钟有效，尽快扫码）")

def cmd_wait_login(args):
    if not LOGIN_FILE.exists():
        sys.exit("先运行 login 拿二维码")
    login = json.loads(LOGIN_FILE.read_text(encoding="utf-8"))
    qrcode, base = login["qrcode"], login.get("base", DEFAULT_BASE)
    deadline = time.time() + (args.timeout or 300)
    log("轮询扫码状态中（35 秒长轮询）...")
    while time.time() < deadline:
        try:
            st = http_get(base, f"ilink/bot/get_qrcode_status?qrcode={urllib.parse.quote(qrcode)}")
        except Exception as e:
            log(f"轮询出错，重试: {e}"); time.sleep(3); continue
        status = st.get("status")
        if status == "confirmed":
            sess = {"bot_token": st.get("bot_token"),
                    "baseurl": st.get("baseurl") or DEFAULT_BASE,
                    "ilink_bot_id": st.get("ilink_bot_id"),
                    "ilink_user_id": st.get("ilink_user_id"),
                    "context_token": "",
                    "login_at": time.strftime("%Y-%m-%d %H:%M:%S")}
            atomic_write_json(SESSION_FILE, sess, mode=0o600)  # 仅自己可读
            LOGIN_FILE.unlink(missing_ok=True)
            if NEEDS_RELOGIN.exists(): NEEDS_RELOGIN.unlink()
            log(f"登录成功，用户: {mask(sess['ilink_user_id'])}")
            print("LOGIN_OK"); return
        if status == "expired":
            LOGIN_FILE.unlink(missing_ok=True); sys.exit("二维码过期，重新运行 login")
        if status in ("need_verifycode", "verify_code_blocked"):
            sys.exit("微信要求输入配对码，请在手机上按提示操作后重试")
        if status == "scaned":
            log("已扫码，等待手机上点确认 ...")
        time.sleep(1)
    sys.exit("等待超时")

def extract_text(msg):
    return "\n".join((i.get("text_item") or {}).get("text", "")
                     for i in msg.get("item_list") or [] if i.get("type") == 1).strip()

def cmd_serve(_args):
    sess = load_session()
    if not sess: sys.exit("还没登录，先跑 login + wait-login")
    token, base, my_id = sess["bot_token"], sess.get("baseurl") or DEFAULT_BASE, sess.get("ilink_user_id")
    atomic_write_text(PID_FILE, str(os.getpid()), mode=0o644)
    cursor = CURSOR_FILE.read_text(encoding="utf-8").strip() if CURSOR_FILE.exists() else ""
    log(f"常驻服务启动，只处理用户 {mask(my_id)} 的私聊消息")
    backoff = 5
    while True:
        try:
            resp = http_post(base, "ilink/bot/getupdates", {"get_updates_buf": cursor},
                             token=token, timeout=UPDATES_POLL_TIMEOUT)
            check_stale(resp, "getupdates")
            if resp.get("get_updates_buf"):
                cursor = resp["get_updates_buf"]; atomic_write_text(CURSOR_FILE, cursor)
            for msg in resp.get("msgs") or []:
                from_id = msg.get("from_user_id") or ""
                if msg.get("group_id") or from_id != my_id:  # 只收自己的私聊，群消息忽略
                    continue
                text = extract_text(msg)
                if not text: continue
                mid = str(msg.get("message_id") or int(time.time() * 1000))
                INBOX_DIR.mkdir(parents=True, exist_ok=True)
                atomic_write_json(INBOX_DIR / f"{mid}.json",
                    {"message_id": mid, "text": text, "from_user_id": from_id,
                     "context_token": msg.get("context_token") or "",
                     "ts": time.strftime("%Y-%m-%d %H:%M:%S")}, mode=0o600)
                if msg.get("context_token"):  # 缓存最新的 context_token，供回复用
                    sess["context_token"] = msg["context_token"]
                    atomic_write_json(SESSION_FILE, sess, mode=0o600)
                log(f"收到: {text[:60]}")
            # 发掉待发送队列里的回复
            PENDING_DIR.mkdir(parents=True, exist_ok=True); SENT_DIR.mkdir(parents=True, exist_ok=True)
            for p in sorted(PENDING_DIR.glob("*.json")):
                job = json.loads(p.read_text(encoding="utf-8"))
                ctx = job.get("context_token") or sess.get("context_token") or ""
                if not ctx:
                    log(f"{p.name}: 还没有 context_token（等用户先发消息），跳过"); continue
                send_text(base, token, job.get("to_user_id") or my_id, job.get("text", ""), ctx)
                p.rename(SENT_DIR / p.name); log(f"已发送: {job.get('text','')[:60]}")
            backoff = 5
        except TokenDead as e:
            mark_needs_relogin(str(e)); log("token 失效，退出等待重新扫码"); sys.exit(10)
        except Exception as e:
            log(f"出错 {type(e).__name__}，{backoff} 秒后重试"); time.sleep(backoff)
            backoff = min(backoff * 2, 120)

def send_text(base, token, to_user_id, text, context_token):
    body = {"msg": {"from_user_id": "",  # 必须是空字符串！
                    "to_user_id": to_user_id,
                    "client_id": f"bridge-{uuid.uuid4().hex}",  # 每次唯一
                    "message_type": 2, "message_state": 2,
                    "context_token": context_token,  # 原样回传用户消息里的值
                    "item_list": [{"type": 1, "text_item": {"text": text}}]}}
    resp = http_post(base, "ilink/bot/sendmessage", body, token=token, timeout=API_TIMEOUT)
    check_stale(resp, "sendmessage")
    return resp

def cmd_send(args):
    sess = load_session()
    if not sess: sys.exit("还没登录")
    ctx = sess.get("context_token") or ""
    if not ctx: sys.exit("还没有 context_token：先在微信里给 Bot 发一条消息")
    try:
        send_text(sess.get("baseurl") or DEFAULT_BASE, sess["bot_token"],
                  sess["ilink_user_id"], args.text, ctx)
    except TokenDead as e:
        mark_needs_relogin(str(e)); sys.exit("token 失效，请重新扫码")
    log("测试消息已发送"); print("SEND_OK")

def cmd_status(_args):
    sess = load_session()
    out = {"logged_in": bool(sess), "needs_relogin": NEEDS_RELOGIN.exists()}
    if sess:
        out.update({"login_at": sess.get("login_at"), "user": mask(sess.get("ilink_user_id")),
                    "has_context_token": bool(sess.get("context_token"))})
    alive = False
    if PID_FILE.exists():
        try: os.kill(int(PID_FILE.read_text().strip()), 0); alive = True
        except Exception: pass
    out["serve_alive"] = alive
    out["inbox_count"] = len(list(INBOX_DIR.glob("*.json"))) if INBOX_DIR.exists() else 0
    out["pending_count"] = len(list(PENDING_DIR.glob("*.json"))) if PENDING_DIR.exists() else 0
    print(json.dumps(out, ensure_ascii=False, indent=2))

def main():
    ap = argparse.ArgumentParser(prog="bridge.py"); sub = ap.add_subparsers(dest="cmd", required=True)
    sub.add_parser("login")
    w = sub.add_parser("wait-login"); w.add_argument("--timeout", type=int, default=300)
    sub.add_parser("serve")
    s = sub.add_parser("send"); s.add_argument("--text", required=True)
    sub.add_parser("status")
    args = ap.parse_args()
    {"login": cmd_login, "wait-login": cmd_wait_login, "serve": cmd_serve,
     "send": cmd_send, "status": cmd_status}[args.cmd](args)

if __name__ == "__main__":
    main()

python3 bridge.py login

python3 bridge.py wait-login

nohup python3 bridge.py serve > state/serve.log 2>&1 &

python3 bridge.py send --text "链路测试：收发正常"