homelab

/

stacks

/

portal

/

or Cancel

Edit File Preview Preview Changes

Spaces Tabs

2 4 8

Wrap No wrap

#!/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" DEFAULT_AGY_CONV_ID = "6293ebb4-df60-4db6-bf5d-df33433ba106" 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 # 7b. One-step Google OAuth Code injection from WeChat (e.g. "4/0A..." or "授权 4/0A...") if cmd.startswith("4/") or low.startswith("授权 ") or low.startswith("auth "): auth_code = cmd.split(None, 1)[-1].strip() if (" " in cmd) else cmd.strip() safe_code = auth_code.replace('"', '\\"') self._exec_shell(f'/opt/homebrew/bin/tmux send-keys -t agy "{safe_code}" C-m', timeout=5.0) time.sleep(5.0) out = self._exec_shell("/opt/homebrew/bin/tmux capture-pane -t agy -p | tail -n 20", timeout=5.0) self.send_text_message( from_user, f"🔐 已将 Google 授权码注入 Mac mini 的 AGY 终端！\n" f"已挂载当前会话 ID: {DEFAULT_AGY_CONV_ID}\n" f"━━━━━━━━━━━━━━\n{out or '(授权处理中)'}\n\n" f"现在你可以直接在微信里和我继续刚才的对话了！", ctx_token, run_id, ) return # 8. Natural language -> Resume exact migrated session (6293ebb4-df60-4db6-bf5d-df33433ba106) on Mac mini! agy_out = self._exec_shell( f'cd /Users/ixvii0iv/homelab && /Users/ixvii0iv/.local/bin/agy --conversation {DEFAULT_AGY_CONV_ID} --dangerously-skip-permissions -p {json.dumps(cmd)}', timeout=120.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: oauth_url = ( "https://accounts.google.com/o/oauth2/auth?access_type=offline" "&client_id=1071006060591-tmhssin2h21lcre235vtolojh4g403ep.apps.googleusercontent.com" "&code_challenge=6R2ML1GiPTslGESIIdY7kBlWPGi3Z7J_SUKiEb-wVc0&code_challenge_method=S256" "&prompt=consent&redirect_uri=https%3A%2F%2Fantigravity.google%2Foauth-callback" "&response_type=code&scope=https%3A%2F%2Fwww.googleapis.com%2Fauth%2Fcloud-platform+" "https%3A%2F%2Fwww.googleapis.com%2Fauth%2Fuserinfo.email+" "https%3A%2F%2Fwww.googleapis.com%2Fauth%2Fuserinfo.profile+" "https%3A%2F%2Fwww.googleapis.com%2Fauth%2Fcclog+" "https%3A%2F%2Fwww.googleapis.com%2Fauth%2Fexperimentsandconfigs+" "https%3A%2F%2Fwww.googleapis.com%2Fauth%2Faicode+openid&state=EHQdZm-HmDWTKGqBFxqf1g" ) self.send_text_message( from_user, f"🤖 当前会话 ({DEFAULT_AGY_CONV_ID[:8]}...) 已同步至 Mac mini！\n\n" "🔐 只需最后一步激活 Mac mini 的 Google AI 登录态：\n" f"1️⃣ 点击打开此链接登录 Google：\n{oauth_url}\n\n" "2️⃣ 复制页面生成的授权码（以 4/ 开头），直接发到这个微信对话框里，即可在微信无缝接力我们当前的完整会话！", 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}"

Commit Changes

Add a Signed-off-by trailer by the committer at the end of the commit log message.

Commit directly to the main branch.

Create a new branch for this commit and start a pull request.

Commit Changes Cancel