Openclaw折腾实例(二)——优雅安全地远程和操控家庭设备(无需域名和固定公网IP)

 <debug>执行耗时 1m 33s · 查看 3 个步骤Skill Read (14 ms)ikuai ipv6Run command (129 ms)time python3 -c "import requests, json, hashlib, base64, os, sys, time, re; print('导入完成')" 2>&1Run command (235 ms)time python3 -c " import requests, json, hashlib, base64, time from datetime import datetime ROUTER_IP = 'xxx.xxx.xxx' USERNAME = 'xxx' PASSWORD = 'xxxxxx' BASE_URL = {ROUTER_IP}' start = time.time() passwd_md5 = hashlib.md5(PASSWORD.encode()).hexdigest() pass_base64 = base64.b64encode(f'salt_11{PASSWORD}'.encode()).decode() session = requests.Session() login_resp = session.post( login', json={'username': USERNAME, 'passwd': passwd_md5, 'pass': pass_base64, 'remember_password': None}, headers={'Content-Type': 'application/json;charset=UTF-8'}, timeout=10 ) print(f'登录响应: {time.time() - start:.2f}秒') result = login_resp.json() cookie = login_resp.cookies.get('sess_key') print(f'sess_key: {cookie[:10]}...') resp = session.post( call', json={'func_name': 'monitor_lanipv6', 'action': 'show', 'param': {'TYPE': 'data,total', 'limit': '0,50'}}, headers={'Content-Type': 'application/json;charset=UTF-8'}, cookies=[redacted]'username': USERNAME, 'sess_key': cookie, 'login': '1'}, timeout=10 ) print(f'获取设备: {time.time() - start:.2f}秒') result = resp.json() devices = result.get('Data', {}).get('data', []) print(f'设备数量: {len(devices)}') print(f'总耗时: {time.time() - start:.2f}秒') " 2>&1

我（飞书等APP）发消息 →Openclaw Gateway 组装 Prompt（系统提示词+上下文文件+历史对话+您的消息）→ 发送给大模型 API → 大模型推理生成回复 → 返回给我

ikuai_ipv6.py(这个脚本在Openclaw应用系列一中有初版)

#!/usr/bin/env python3"""飞书快速通道机器人 - 使用官方 lark-oapi SDK 正确用法官方文档: https://open.feishu.cn/document/uAjLw4CM/ukTMukTMukTM/server-side-sdk/python--sdk/handle-events功能：1. 通过 lark-oapi SDK 的 WebSocket 长连接接收飞书消息2. 匹配关键词"电脑地址"/"所有地址"/"唤醒电脑"3. 直接执行本地脚本，秒级返回4. 通过飞书 API 回复消息运行：    python3 feishu_fast_mstsc.py依赖：    pip3 install lark-oapi"""import jsonimport timeimport subprocessimport osimport sysimport signalfrom datetime import datetimeimport lark_oapi as larkfrom lark_oapi.api.im.v1 import ReplyMessageRequest, ReplyMessageRequestBody# ========== 配置区域 ==========APP_ID = "你飞书机器人的APPID"APP_SECRET = "你飞书机器人的密钥"# 授权用户（只响应指定用户，留空表示响应所有人）# 多个用户用逗号分隔，例如: ["user1", "user2"]AUTHORIZED_USERS = ["你的飞书OPENID"]# 快捷指令映射QUICK_COMMANDS = {    "电脑地址": {        "script": "python3 ~/.openclaw/workspace/skills/ikuai_ipv6.py DESKTOP",        "description": "获取 Windows 电脑 IPv6 地址"    },    "所有地址": {        "script": "python3 ~/.openclaw/workspace/skills/ikuai_ipv6.py",        "description": "获取所有设备 IPv6 地址"    },    "唤醒电脑": {        "script": "python3 ~/.openclaw/workspace/skills/ikuai_ipv6.py -w 你的电脑MAC地址",        "description": "唤醒 Windows 电脑"    }}running = True# ========== 工具函数 ==========def log(msg):    """打印日志"""    print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {msg}")def execute_script(script_cmd):    """执行本地脚本，返回输出"""    try:        expanded_cmd = script_cmd.replace("~", os.path.expanduser("~"))        result = subprocess.run(            expanded_cmd,            shell=True,            capture_output=True,            text=True,            timeout=300        )        output = result.stdout if result.stdout else result.stderr        return output.strip() or "执行完成，无输出"    except subprocess.TimeoutExpired:        return "⏱ 执行超时（超过5分钟）"    except Exception as e:        return f"❌ 执行错误: {str(e)}"def format_reply(raw_output, elapsed, command_name):    """格式化脚本输出"""    import re    clean = re.sub(r'\x1b\[[0-9;]*m', '', raw_output)    header = f"⚡ 快速通道 | {command_name}⏱ 耗时: {elapsed:.1f}s{'='*40}"    return header + cleandef reply_to_message(message_id, content):    """回复指定消息"""    try:        client = lark.Client.builder() \            .app_id(APP_ID) \            .app_secret(APP_SECRET) \            .log_level(lark.LogLevel.INFO) \            .build()        request = ReplyMessageRequest.builder() \            .message_id(message_id) \            .request_body(ReplyMessageRequestBody.builder()                .content(json.dumps({"text": content}))                .msg_type("text")                .build()) \            .build()        response = client.im.v1.message.reply(request)        if response.success():            log("回复成功")        else:            log(f"回复失败: {response.code} - {response.msg}")    except Exception as e:        log(f"回复异常: {e}")# ========== 事件处理（官方 SDK 正确用法）==========def do_p2_im_message_receive_v1(data: lark.im.v1.P2ImMessageReceiveV1) -> None:    """处理消息事件 (v2.0 事件)"""    try:        log(f"收到消息事件!")        # 获取消息信息        message = data.event.message        sender = data.event.sender.sender_id.open_id        # 授权检查        if AUTHORIZED_USERS and sender not in AUTHORIZED_USERS:            log(f"未授权用户: {sender}")            return        # 只处理文本消息        if message.message_type != "text":            log(f"非文本消息: {message.message_type}")            return        # 解析消息内容        try:            content = json.loads(message.content)            text = content.get("text", "").strip()        except:            text = ""        message_id = message.message_id        log(f"收到消息: {text[:50]}")        # ========== 快速通道检查 ==========        for keyword, config in QUICK_COMMANDS.items():            if keyword in text:                log(f"触发快捷指令: {keyword}")                start_time = time.time()                # 执行脚本                output = execute_script(config["script"])                elapsed = time.time() - start_time                # 格式化回复                reply = format_reply(output, elapsed, config["description"])                log(f"执行完成: {elapsed:.1f}s")                # 回复消息                reply_to_message(message_id, reply)                return        # 未匹配指令        log(f"未匹配的指令: {text[:50]}")    except Exception as e:        log(f"处理消息异常: {e}")        import traceback        traceback.print_exc()# ========== 主入口 ==========def signal_handler(sig, frame):    """处理退出信号"""    global running    log("收到退出信号，正在关闭...")    running = False    sys.exit(0)if __name__ == "__main__":    # 注册信号处理    signal.signal(signal.SIGINT, signal_handler)    signal.signal(signal.SIGTERM, signal_handler)    print("=" * 60)    print("飞书快速通道机器人")    print("=" * 60)    print("模式: WebSocket 长连接 (官方 SDK)")    print("优势: 无需公网IP，无需端口暴露")    print("=" * 60)    print("支持的快捷指令:")    for k, v in QUICK_COMMANDS.items():        print(f"  • {k} → {v['description']}")    print("=" * 60)    # 检查配置    if not APP_ID or not APP_SECRET:        print("❌ 错误: APP_ID 和 APP_SECRET 未配置")        sys.exit(1)    # 创建事件处理器 - 官方 SDK 正确用法    log("正在创建事件处理器...")    event_handler = lark.EventDispatcherHandler.builder("", "") \        .register_p2_im_message_receive_v1(do_p2_im_message_receive_v1) \        .build()    # 创建 WebSocket 客户端 - 官方 SDK 正确用法    log("正在启动 WebSocket 客户端...")    cli = lark.ws.Client(        APP_ID,        APP_SECRET,        event_handler=event_handler,        log_level=lark.LogLevel.DEBUG    )    try:        # 启动连接（阻塞主线程，直到进程结束）        cli.start()    except KeyboardInterrupt:        log("用户中断")    except Exception as e:        log(f"启动异常: {e}")        import traceback        traceback.print_exc()

我（飞书等APP）发消息 →飞书快捷指令机器人 → 调用部署了Websocket脚本的内网设备（和我的Openclaw机器同一台设备） →通过Websocket脚本调用运行IPv6信息获取脚本登录路由器查询IPv6信息→通过Websocket将信息通过飞书机器人呈现在飞书聊天框