#!/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}"