feat(clawbot): add native WeChat ClawBot protocol engine (clawbot.py) and WechatClawModal UI

Browse Source
This commit is contained in:
Yin Kou committed 2026-11-07 00:51:38 +08:00
1 parent 07d2050c5f
commit 8fa46a2b8d
14 files changed
+1551 -201

No files matched your search

+599
View File
@@ -0,0 +1,599 @@
#!/usr/bin/env python3
# ==============================================================================
# HomeLab WeChat ClawBot Engine (Tencent iLink Bot Protocol v2.4.10)
# 100% 兼容 @tencent-weixin/openclaw-weixin 官方协议 (ilinkai.weixin.qq.com)
# 支持:
# 1. 生成微信 ClawBot 绑定二维码 & 轮询扫码状态 (支持 IDC 重定向与配对验证码)
# 2. 7x24h 长轮询接收微信消息 (getupdates) + "正在输入..." 状态推送 (sendtyping)
# 3. 微信内直接查询 Mac mini / 极空间状态、触发备份、执行终端指令、与 AGY AI 对话
# ==============================================================================
import base64
import json
import os
import random
import re
import subprocess
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
from pathlib import Path
FIXED_BASE_URL = "https://ilinkai.weixin.qq.com"
DEFAULT_BOT_TYPE = "3"
CHANNEL_VERSION = "2.4.10"
# (2 << 16) | (4 << 8) | 10 = 132106
ILINK_CLIENT_VERSION = str((2 << 16) | (4 << 8) | 10)
ILINK_APP_ID = "bot"
STATE_DIR = Path("/Users/ixvii0iv/homelab/stacks/wechat-bot")
SESSION_FILE = STATE_DIR / "clawbot_session.json"
SYNC_BUF_FILE = STATE_DIR / "clawbot_sync_buf.txt"
def _random_wechat_uin() -> str:
uint32 = random.randint(10000000, 0xFFFFFFFF)
return base64.b64encode(str(uint32).encode("utf-8")).decode("utf-8")
def _common_headers() -> dict:
return {
"iLink-App-Id": ILINK_APP_ID,
"iLink-App-ClientVersion": ILINK_CLIENT_VERSION,
}
def _post_headers(token: str = "") -> dict:
h = {
"Content-Type": "application/json",
"AuthorizationType": "ilink_bot_token",
"X-WECHAT-UIN": _random_wechat_uin(),
**_common_headers(),
}
if token.strip():
h["Authorization"] = f"Bearer {token.strip()}"
return h
def _base_info() -> dict:
return {
"channel_version": CHANNEL_VERSION,
"bot_agent": "OpenClaw",
}
def _parse_lossless_json(raw_text: str) -> dict:
# Quote uint64 message_id / msg_id / svr_id fields so JS/Python keeps exact strings
patched = re.sub(
r'("(?:message_id|msg_id|svr_id)"\s*:\s*)(-?\d+)',
r'\1"\2"',
raw_text,
)
return json.loads(patched)
def _api_get(base_url: str, endpoint: str, timeout: float = 36.0) -> dict:
url = base_url.rstrip("/") + "/" + endpoint.lstrip("/")
req = urllib.request.Request(url, headers=_common_headers(), method="GET")
with urllib.request.urlopen(req, timeout=timeout) as resp:
raw = resp.read().decode("utf-8")
return _parse_lossless_json(raw)
def _api_post(base_url: str, endpoint: str, payload: dict, token: str = "", timeout: float = 20.0) -> dict:
url = base_url.rstrip("/") + "/" + endpoint.lstrip("/")
data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
req = urllib.request.Request(url, data=data, headers=_post_headers(token), method="POST")
with urllib.request.urlopen(req, timeout=timeout) as resp:
raw = resp.read().decode("utf-8")
return _parse_lossless_json(raw)
class ClawBotManager:
def __init__(self, telemetry_fn=None, action_runner_fn=None):
STATE_DIR.mkdir(parents=True, exist_ok=True)
self.telemetry_fn = telemetry_fn
self.action_runner_fn = action_runner_fn
self._lock = threading.Lock()
# Saved session credentials
self.session = self._load_session()
# Active QR login state
self.login_state = {
"active": False,
"qrcode": "",
"qrcode_url": "",
"status": "idle", # idle | wait | scaned | need_verifycode | confirmed | expired | error
"message": "尚未发起扫码登录",
"api_base_url": FIXED_BASE_URL,
"pending_verify_code": "",
"started_at": 0.0,
}
# Runtime monitor stats
self.last_event_ts = 0.0
self.last_inbound_text = ""
self.last_reply_text = ""
self.msg_count = 0
self._stop_event = threading.Event()
self._monitor_thread = None
if self.is_connected():
self._start_monitor_thread()
def _load_session(self) -> dict:
if SESSION_FILE.exists():
try:
return json.loads(SESSION_FILE.read_text(encoding="utf-8"))
except Exception:
pass
return {}
def _save_session(self, data: dict):
self.session = data
SESSION_FILE.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
def is_connected(self) -> bool:
return bool(self.session.get("bot_token") and self.session.get("ilink_bot_id"))
def get_public_status(self) -> dict:
return {
"connected": self.is_connected(),
"bot_id": self.session.get("ilink_bot_id", ""),
"user_id": self.session.get("ilink_user_id", ""),
"base_url": self.session.get("baseurl", FIXED_BASE_URL),
"connected_at": self.session.get("connected_at", ""),
"msg_count": self.msg_count,
"last_inbound_text": self.last_inbound_text,
"last_reply_text": self.last_reply_text,
"login": {
"active": self.login_state["active"],
"qrcode_url": self.login_state["qrcode_url"],
"status": self.login_state["status"],
"message": self.login_state["message"],
},
}
# ------------------------------------------------------------------
# 1. QR Code Login Flow
# ------------------------------------------------------------------
def start_qr_login(self) -> dict:
try:
local_tokens = [self.session["bot_token"]] if self.session.get("bot_token") else []
resp = _api_post(
FIXED_BASE_URL,
f"ilink/bot/get_bot_qrcode?bot_type={DEFAULT_BOT_TYPE}",
{"local_token_list": local_tokens},
timeout=12.0,
)
qrcode = resp.get("qrcode", "")
qrcode_url = resp.get("qrcode_img_content", "")
if not qrcode or not qrcode_url:
raise RuntimeError(f"未获取到二维码: {resp}")
with self._lock:
self.login_state = {
"active": True,
"qrcode": qrcode,
"qrcode_url": qrcode_url,
"status": "wait",
"message": "请使用手机微信扫描二维码(或在微信 ClawBot 页面点击「开始扫一扫」)",
"api_base_url": FIXED_BASE_URL,
"pending_verify_code": "",
"started_at": time.time(),
}
t = threading.Thread(target=self._poll_qr_worker, args=(qrcode,), daemon=True)
t.start()
return {"ok": True, **self.get_public_status()}
except Exception as e:
with self._lock:
self.login_state["status"] = "error"
self.login_state["message"] = f"获取二维码失败: {e}"
return {"ok": False, "message": str(e)}
def submit_verify_code(self, code: str) -> dict:
clean = code.strip()
if not clean:
return {"ok": False, "message": "请输入有效的数字配对码"}
with self._lock:
self.login_state["pending_verify_code"] = clean
self.login_state["message"] = f"已提交配对码 {clean},正在向微信服务器验证..."
return {"ok": True, **self.get_public_status()}
def disconnect(self) -> dict:
self._stop_event.set()
if SESSION_FILE.exists():
SESSION_FILE.unlink()
if SYNC_BUF_FILE.exists():
SYNC_BUF_FILE.unlink()
self.session = {}
with self._lock:
self.login_state["active"] = False
self.login_state["status"] = "idle"
self.login_state["message"] = "已断开微信 ClawBot 连接"
return {"ok": True, **self.get_public_status()}
def _poll_qr_worker(self, initial_qrcode: str):
deadline = time.time() + 300.0 # 5 minutes
while time.time() < deadline:
with self._lock:
if not self.login_state["active"] or self.login_state["qrcode"] != initial_qrcode:
return
base_url = self.login_state["api_base_url"]
qrcode = self.login_state["qrcode"]
verify_code = self.login_state["pending_verify_code"]
endpoint = f"ilink/bot/get_qrcode_status?qrcode={urllib.parse.quote(qrcode)}"
if verify_code:
endpoint += f"&verify_code={urllib.parse.quote(verify_code)}"
try:
resp = _api_get(base_url, endpoint, timeout=35.0)
except Exception:
resp = {"status": "wait"}
st = resp.get("status", "wait")
with self._lock:
if self.login_state["qrcode"] != initial_qrcode:
return
self.login_state["status"] = st
if st == "wait":
pass
elif st == "scaned":
self.login_state["pending_verify_code"] = ""
self.login_state["message"] = "📱 已扫码!请在手机微信上点击确认授权..."
elif st == "scaned_but_redirect":
r_host = resp.get("redirect_host", "")
if r_host:
self.login_state["api_base_url"] = f"https://{r_host}"
self.login_state["message"] = "📱 已扫码,正在切换微信数据中心节点..."
elif st == "need_verifycode":
if self.login_state["pending_verify_code"]:
self.login_state["message"] = "❌ 配对码不匹配,请重新输入手机微信上显示的数字:"
self.login_state["pending_verify_code"] = ""
else:
self.login_state["message"] = "🔢 请输入手机微信屏幕上显示的配对数字:"
elif st == "binded_redirect":
self.login_state["active"] = False
self.login_state["message"] = "✅ 此微信已绑定过当前节点!"
return
elif st in ("expired", "verify_code_blocked"):
self.login_state["active"] = False
self.login_state["message"] = "⏳ 二维码已过期,请点击「刷新二维码」重新扫描"
return
elif st == "confirmed":
bot_token = resp.get("bot_token", "")
bot_id = resp.get("ilink_bot_id", "")
baseurl = resp.get("baseurl") or FIXED_BASE_URL
user_id = resp.get("ilink_user_id", "")
self._save_session(
{
"bot_token": bot_token,
"ilink_bot_id": bot_id,
"baseurl": baseurl,
"ilink_user_id": user_id,
"connected_at": time.strftime("%Y-%m-%d %H:%M:%S"),
}
)
self.login_state["active"] = False
self.login_state["message"] = "🎉 微信 ClawBot 绑定成功!现在可以直接在微信里收发消息了"
self._start_monitor_thread()
# Send immediate welcome message to the user on WeChat!
if user_id:
threading.Thread(
target=self.send_text_message,
args=(
user_id,
"✅ M6 Mac mini (HomeLab & AGY AI) 已成功连接微信 ClawBot!\n\n"
"你可以直接发送以下内容与我交互:\n"
"• 发送「状态」:查看 M6 Mac mini 与极空间 Z4 实时遥测\n"
"• 发送「备份」:立即打包备份核心数据至极空间 20TB\n"
"• 发送「屏幕」:查看 AGY 云端终端当前实时画面\n"
"• 发送「$ 命令」:在 Mac mini 上执行终端命令 (如 $ df -h)\n"
"• 直接发任何问题:直接转发给 Mac mini 上的 AI 助手回答",
),
daemon=True,
).start()
return
time.sleep(1.0)
# ------------------------------------------------------------------
# 2. Long-Poll Monitor & Outbound Messaging
# ------------------------------------------------------------------
def _start_monitor_thread(self):
if self._monitor_thread and self._monitor_thread.is_alive():
return
self._stop_event.clear()
self._monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True)
self._monitor_thread.start()
def send_text_message(self, to_user_id: str, text: str, context_token: str = "", run_id: str = ""):
if not self.is_connected() or not to_user_id or not text:
return
base_url = self.session.get("baseurl") or FIXED_BASE_URL
token = self.session.get("bot_token", "")
client_id = f"openclaw-weixin-{uuid.uuid4().hex[:12]}"
msg_obj = {
"from_user_id": "",
"to_user_id": to_user_id,
"client_id": client_id,
"message_type": 2, # BOT
"message_state": 2, # FINISH
"item_list": [{"type": 1, "text_item": {"text": text}}],
}
if context_token:
msg_obj["context_token"] = context_token
if run_id:
msg_obj["run_id"] = run_id
payload = {
"msg": msg_obj,
"base_info": _base_info(),
}
try:
_api_post(base_url, "ilink/bot/sendmessage", payload, token=token, timeout=15.0)
self.last_reply_text = text[:120]
except Exception as e:
print(f"[ClawBot] send_text_message error: {e}")
def _send_typing(self, to_user_id: str, context_token: str = ""):
if not self.is_connected() or not to_user_id:
return
base_url = self.session.get("baseurl") or FIXED_BASE_URL
token = self.session.get("bot_token", "")
try:
cfg = _api_post(
base_url,
"ilink/bot/getconfig",
{
"ilink_user_id": to_user_id,
"context_token": context_token,
"base_info": _base_info(),
},
token=token,
timeout=8.0,
)
ticket = cfg.get("typing_ticket", "")
if ticket:
_api_post(
base_url,
"ilink/bot/sendtyping",
{
"ilink_user_id": to_user_id,
"typing_ticket": ticket,
"status": 1,
"base_info": _base_info(),
},
token=token,
timeout=8.0,
)
except Exception:
pass
def _monitor_loop(self):
base_url = self.session.get("baseurl") or FIXED_BASE_URL
token = self.session.get("bot_token", "")
try:
_api_post(
base_url,
"ilink/bot/msg/notifystart",
{"base_info": _base_info()},
token=token,
timeout=10.0,
)
except Exception:
pass
sync_buf = ""
if SYNC_BUF_FILE.exists():
try:
sync_buf = SYNC_BUF_FILE.read_text(encoding="utf-8").strip()
except Exception:
pass
while not self._stop_event.is_set() and self.is_connected():
base_url = self.session.get("baseurl") or FIXED_BASE_URL
token = self.session.get("bot_token", "")
try:
resp = _api_post(
base_url,
"ilink/bot/getupdates",
{"get_updates_buf": sync_buf, "base_info": _base_info()},
token=token,
timeout=38.0,
)
new_buf = resp.get("get_updates_buf")
if new_buf:
sync_buf = new_buf
SYNC_BUF_FILE.write_text(sync_buf, encoding="utf-8")
msgs = resp.get("msgs") or []
for m in msgs:
# Only handle USER messages (message_type == 1)
if m.get("message_type") == 2:
continue
from_user = m.get("from_user_id", "")
ctx_token = m.get("context_token", "")
run_id = m.get("run_id", "")
items = m.get("item_list") or []
text_parts = []
for it in items:
t_type = it.get("type")
if t_type == 1 and it.get("text_item", {}).get("text"):
text_parts.append(it["text_item"]["text"])
elif t_type == 3 and it.get("voice_item", {}).get("text"):
text_parts.append(it["voice_item"]["text"])
user_text = "\n".join(text_parts).strip()
if from_user and user_text:
self.msg_count += 1
self.last_inbound_text = user_text
self.last_event_ts = time.time()
threading.Thread(
target=self._handle_inbound_message,
args=(from_user, user_text, ctx_token, run_id),
daemon=True,
).start()
except urllib.error.URLError as e:
# Normal long-poll timeout
if "timed out" in str(e).lower():
continue
time.sleep(2.0)
except TimeoutError:
continue
except Exception as e:
if "timed out" in str(e).lower():
continue
print(f"[ClawBot] getupdates error: {e}")
time.sleep(3.0)
# ------------------------------------------------------------------
# 3. Intelligent Command & AI Router for WeChat Messages
# ------------------------------------------------------------------
def _handle_inbound_message(self, from_user: str, text: str, ctx_token: str, run_id: str):
self._send_typing(from_user, ctx_token)
cmd = text.strip()
low = cmd.lower()
# 1. Help
if low in ("帮助", "help", "/help", "?"):
reply = (
"🍎 HomeLab & AGY 微信中控台指令清单:\n"
"━━━━━━━━━━━━━━\n"
"1️⃣ 发送「状态」或 /status\n"
" 查看 M6 Mac mini、极空间 20TB、联通公网 IP 与狮城专线实时指标\n"
"2️⃣ 发送「备份」或 /backup\n"
" 立即将 Vaultwarden 与核心配置打包灾备至极空间 Z4\n"
"3️⃣ 发送「刷新」或 /ddns\n"
" 立即刷新 Cloudflare 泛域名 DDNS 解析\n"
"4️⃣ 发送「屏幕」或 /screen\n"
" 抓取 Mac mini 上 tmux (agy) 终端的实时画面\n"
"5️⃣ 发送「$ <终端命令>」\n"
" 在 Mac mini 上直接运行 Shell 命令并返回结果 (如: $ df -h)\n"
"6️⃣ 发送「> <按键内容>」\n"
" 直接向 Mac mini 的 tmux (agy) 终端输入文字并回车\n"
"7️⃣ 直接发送任意问题\n"
" 自动调用 Mac mini AI 助手为你解答或执行任务"
)
self.send_text_message(from_user, reply, ctx_token, run_id)
return
# 2. Telemetry Status
if low in ("状态", "status", "/status", "监控"):
if self.telemetry_fn:
t = self.telemetry_fn()
mini = t["mini"]
nas = t["nas"]
net = t["network"]
reply = (
f"🍏 HomeLab 实时状态 ({t['timestamp']})\n"
f"━━━━━━━━━━━━━━\n"
f"🖥 M6 Mac mini ({mini['ip']})\n"
f"• CPU 负载: {mini['cpu_pct']}% (1m: {mini['load_1m']})\n"
f"• 内存占用: {mini['mem_used_gb']} / {mini['mem_total_gb']} GB ({mini['mem_pct']}%)\n"
f"• SSD 占用: {mini['disk_used_gb']} / {mini['disk_total_gb']} GB ({mini['disk_pct']}%)\n"
f"• 在线时长: {mini['uptime']}\n"
f"• AGY 终端: {'🟢 常驻运行中' if mini['agy_tmux'] else '⚪ 待机'}\n\n"
f"💾 极空间 Z4 ({nas['ip']})\n"
f"• 可用余量: {nas['free_tb']} TB / {nas['total_tb']} TB\n"
f"• 最近灾备: {nas['latest_backup']}\n\n"
f"🌐 网络与 AI 算力专线\n"
f"• 家庭公网 IP: {net['home_wan_ip']}\n"
f"• AI 分流策略: {net['ai_policy']}\n"
f"• 狮城节点延迟: {net['sg_delay_ms']} ms\n"
f"• 免端口中控台: https://nav.961504.xyz"
)
else:
reply = "状态采集接口未就绪"
self.send_text_message(from_user, reply, ctx_token, run_id)
return
# 3. Backup to ZSpace NAS
if low in ("备份", "backup", "/backup"):
out = self._exec_shell("/Users/ixvii0iv/homelab/scripts/backup-to-nas.sh", timeout=20.0)
self.send_text_message(from_user, f"✅ 极空间 Z4 灾备执行完毕:\n{out or '已完成备份'}", ctx_token, run_id)
return
# 4. Refresh DDNS
if low in ("刷新", "ddns", "/ddns"):
out = self._exec_shell("/Users/ixvii0iv/homelab/scripts/cloudflare-ddns.sh", timeout=15.0)
self.send_text_message(from_user, f"✅ Cloudflare DDNS 刷新结果:\n{out or '已同步最新公网 IP'}", ctx_token, run_id)
return
# 5. Capture tmux screen
if low in ("屏幕", "screen", "/screen", "tmux"):
out = self._exec_shell("/opt/homebrew/bin/tmux capture-pane -t agy -p | tail -n 25", timeout=5.0)
self.send_text_message(
from_user,
f"🖥 AGY 终端 (tmux: agy) 最新画面:\n━━━━━━━━━━━━━━\n{out or '(空)'}",
ctx_token,
run_id,
)
return
# 6. Direct Shell Command ($ ...)
if cmd.startswith("$"):
shell_cmd = cmd[1:].strip()
if not shell_cmd:
self.send_text_message(from_user, "请在 $ 后输入要执行的命令,例如:$ df -h", ctx_token, run_id)
return
out = self._exec_shell(shell_cmd, timeout=25.0)
if len(out) > 3500:
out = out[-3500:] + "\n...(已截断)"
self.send_text_message(from_user, f"💻 终端执行结果 ($ {shell_cmd}):\n━━━━━━━━━━━━━━\n{out or '(无输出)'}", ctx_token, run_id)
return
# 7. Inject keys into tmux agy session (> ...)
if cmd.startswith(">"):
keys = cmd[1:].strip()
safe_keys = keys.replace('"', '\\"')
self._exec_shell(f'/opt/homebrew/bin/tmux send-keys -t agy "{safe_keys}" C-m', timeout=5.0)
time.sleep(2.5)
out = self._exec_shell("/opt/homebrew/bin/tmux capture-pane -t agy -p | tail -n 25", timeout=5.0)
self.send_text_message(
from_user,
f"⌨️ 已向 AGY 终端发送指令并捕获最新画面:\n━━━━━━━━━━━━━━\n{out or '(已发送)'}",
ctx_token,
run_id,
)
return
# 8. Natural language -> Try agy -p on Mac mini, or inject into tmux agy session!
agy_out = self._exec_shell(
f'cd /Users/ixvii0iv/homelab && /Users/ixvii0iv/.local/bin/agy -c --dangerously-skip-permissions -p {json.dumps(cmd)}',
timeout=90.0,
)
if agy_out and "authentication required" not in agy_out.lower():
if len(agy_out) > 3800:
agy_out = agy_out[:3800] + "\n...(内容较长,完整输出请查看 https://agy.961504.xyz)"
self.send_text_message(from_user, agy_out, ctx_token, run_id)
else:
# Fallback: send into tmux agy session and return terminal capture + quick guide
self.send_text_message(
from_user,
f"🤖 收到你的消息:「{cmd}」\n\n"
"💡 提示:检测到 Mac mini 上的 `agy` 终端首次启动需要完成一次 Google 账号登录授权。你可以:\n"
"1. 打开 https://agy.961504.xyz 完成一次扫码/授权码粘贴,之后微信即可直接与 AGY 自由对话!\n"
"2. 现在你已可以直接在微信发送「状态」「备份」「屏幕」或「$ 命令」(如 `$ docker ps`)直接操控 Mac mini!",
ctx_token,
run_id,
)
def _exec_shell(self, cmd: str, timeout: float = 15.0) -> str:
try:
env = os.environ.copy()
env["PATH"] = "/Users/ixvii0iv/.local/bin:/Users/ixvii0iv/.orbstack/bin:/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin"
env["HOME"] = "/Users/ixvii0iv"
env["LANG"] = "en_US.UTF-8"
res = subprocess.run(
cmd, shell=True, capture_output=True, text=True, timeout=timeout, env=env
)
out = (res.stdout or "") + ("\n" + res.stderr if res.stderr else "")
return out.strip()
except Exception as e:
return f"执行超时或异常: {e}"