diff --git a/.gitignore b/.gitignore deleted file mode 100644 index d497e87..0000000 --- a/.gitignore +++ /dev/null @@ -1,8 +0,0 @@ -__pycache__/ -*.pyc -*.pyo -.env -.venv/ -*.egg-info/ -dist/ -build/ diff --git a/LICENSE b/LICENSE deleted file mode 100644 index 7203af6..0000000 --- a/LICENSE +++ /dev/null @@ -1,21 +0,0 @@ -MIT License - -Copyright (c) 2026 shuiyu - -Permission is hereby granted, free of charge, to any person obtaining a copy -of this software and associated documentation files (the "Software"), to deal -in the Software without restriction, including without limitation the rights -to use, copy, modify, merge, publish, distribute, sublicense, and/or sell -copies of the Software, and to permit persons to whom the Software is -furnished to do so, subject to the following conditions: - -The above copyright notice and this permission notice shall be included in all -copies or substantial portions of the Software. - -THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE -SOFTWARE. diff --git a/README.md b/README.md index dbbcd39..bebec60 100644 --- a/README.md +++ b/README.md @@ -2,8 +2,7 @@ 你好呀～这个仓库已经搬到更好用的地方啦。 -为了让代码管理更清爽、多平台同步更省心，我们把原先分散的镜像统一收拢到 -**Gitee** 与 **GitHub** 主仓维护，这里就只保留这张"迁移小纸条"咯。 +为了让代码管理更清爽、多平台同步更省心，我们把原先分散的镜像统一收拢到主仓维护，这里就只保留这张"迁移小纸条"咯。 📦 新家在这里： - Gitee：https://gitee.com/shuiyu1123/xiaoli-cli-plugins diff --git a/dglab_ws/README.md b/dglab_ws/README.md deleted file mode 100644 index 6280455..0000000 --- a/dglab_ws/README.md +++ /dev/null @@ -1,190 +0,0 @@ -# DG-Lab 郊狼 WebSocket 控制插件 - -> 基于 [DG-LAB-OPENSOURCE](https://github.com/DG-LAB-OPENSOURCE/DG-LAB-OPENSOURCE) 官方 Socket V2 协议 -> 支持郊狼 3.0 脉冲主机的远程强度控制与波形发送 - -## 前置准备 - -### 1. 安装依赖 - -```bash -pip install websocket-client # 必须 -pip install qrcode # 可选，终端显示ASCII二维码 -``` - -### 2. 安装插件 - -将 `dglab_ws.py` 复制到 xiaoli-cli 项目的 `plugins/` 目录下： - -```bash -cp dglab_ws.py /path/to/xiaoli-cli/plugins/ -``` - -在 `plugins_config.py` 中添加： - -```python -ENABLE_DGLAB_WS = True -``` - -### 3. 启动 WebSocket 服务端 - -你需要一个 DG-LAB Socket V2 后端服务。官方参考实现位于 [socket/v2/backend](https://github.com/DG-LAB-OPENSOURCE/DG-LAB-OPENSOURCE/tree/main/socket/v2/backend)： - -```bash -cd socket/v2/backend -npm install -npm start -``` - -默认监听 `ws://0.0.0.0:9999`。 - -## 使用教程 - -### 连接与配对 - -``` -# 第一步：连接到 WebSocket 服务端 -/dglab connect ws://192.168.1.100:9999 - -# 第二步：生成二维码，用 DG-LAB APP 扫码 -/dglab qr - -# 查看连接状态 -/dglab status -``` - -**配对流程：** -1. 本插件连接 WebSocket 服务端，获得 `clientId` -2. 生成二维码（含 `clientId`） -3. DG-LAB APP 扫码连接服务端，获得 `targetId` -4. 服务端自动完成绑定，双方即可通信 - -> 💡 APP 端操作：打开 DG-LAB APP → SOCKET 功能 → 扫描二维码 - -### 强度控制 - -控制 A/B 两个通道的脉冲强度（范围 0-200）： - -``` -# 通道A 增加 10 -/dglab strength a + 10 - -# 通道B 减少 5 -/dglab strength b - 5 - -# 通道A 设为 50 -/dglab strength a = 50 -``` - -| 操作符 | 说明 | 示例 | -|--------|------|------| -| `+` | 增加 | `strength a + 10` | -| `-` | 减少 | `strength b - 5` | -| `=` | 设为指定值 | `strength a = 50` | - -### 波形发送 - -发送参数化波形脉冲： - -``` -# 通道A: 频率30, 强度50, 持续5秒 -/dglab pulse a 30 50 5 - -# 通道B: 频率100, 强度80, 持续10秒 -/dglab pulse b 100 80 10 -``` - -**参数说明：** - -| 参数 | 范围 | 说明 | -|------|------|------| -| 频率 | 10-1000 | 脉冲频率，内部自动映射到 10-240 | -| 强度 | 0-100 | 波形强度 | -| 持续时间 | 1-60 | 秒数，默认5秒 | - -### 原始波形数据 - -如果你有现成的波形 HEX 数据，可以直接发送： - -``` -# 每条波形 = 8字节 = 16个HEX字符 -# 格式: 频率x4 + 强度x4 -/dglab wave a 0A0A0A0A64646464 -``` - -### 清空波形队列 - -``` -/dglab clear a # 清空通道A -/dglab clear b # 清空通道B -/dglab clear both # 清空双通道 -``` - -### 断开连接 - -``` -/dglab disconnect -``` - -## 命令速查 - -| 命令 | 说明 | -|------|------| -| `/dglab connect ` | 连接 WebSocket 服务器 | -| `/dglab qr` | 生成配对二维码 | -| `/dglab status` | 查看状态 | -| `/dglab strength <+/−/=> <值>` | 控制强度 | -| `/dglab pulse <频率> <强度> [秒数]` | 发送波形 | -| `/dglab wave ` | 发送原始波形 | -| `/dglab clear ` | 清空波形队列 | -| `/dglab disconnect` | 断开连接 | -| `/dglab help` | 显示帮助 | - -## 常见问题 - -**Q: 连接超时怎么办？** -A: 检查 WebSocket 服务端是否启动，防火墙是否放行端口，地址格式是否为 `ws://ip:port`。 - -**Q: 配对失败 (400)?** -A: 该 clientId 已被其他 APP 绑定。断开重连会生成新的 clientId。 - -**Q: 强度设为软上限以上会怎样？** -A: 实际强度不会超过软上限。比如软上限 100，设为 150 实际只会到 100。 - -**Q: 波形数据格式是什么？** -A: 每条 8 字节 HEX，前 4 字节是频率，后 4 字节是强度。详见 [官方波形说明](https://github.com/DG-LAB-OPENSOURCE/DG-LAB-OPENSOURCE/blob/main/coyote/extra/README.md)。 - -## 测试（无需真机） - -本插件附带一个 **APP 模拟器**，可以模拟 DG-LAB APP 端行为，无需真实设备即可测试。 - -### 测试步骤 - -``` -终端1: 启动 WebSocket 服务端 - cd socket/v2/backend && npm start - -终端2: 启动 APP 模拟器 - python dglab_app_simulator.py ws://127.0.0.1:9999 - -终端3: 在 xiaoli-cli 中测试 - /dglab connect ws://127.0.0.1:9999 - /dglab qr - /dglab strength a + 10 - /dglab pulse a 30 50 5 - /dglab status -``` - -模拟器会： -- 模拟 APP 连接 WebSocket 服务端 -- 自动接受配对请求 -- 实时显示收到的强度/波形指令 -- 回传强度状态（和真机行为一致） - -## 参考资料 - -- [DG-LAB 官方开源仓库](https://github.com/DG-LAB-OPENSOURCE/DG-LAB-OPENSOURCE) -- [Socket V2 协议文档](https://github.com/DG-LAB-OPENSOURCE/DG-LAB-OPENSOURCE/blob/main/socket/v2/README.md) -- [V3 蓝牙协议](https://github.com/DG-LAB-OPENSOURCE/DG-LAB-OPENSOURCE/blob/main/coyote/v3/README_V3.md) -- [PyDG-Lab-WS (Python库)](https://pypi.org/project/pydglab-ws/) -- [XToys.app 集成](https://xtoys.app) diff --git a/dglab_ws/dglab_app_simulator.py b/dglab_ws/dglab_app_simulator.py deleted file mode 100644 index e88b781..0000000 --- a/dglab_ws/dglab_app_simulator.py +++ /dev/null @@ -1,266 +0,0 @@ -""" -DG-Lab 郊狼 APP 模拟器 -模拟 DG-LAB APP 端连接 WebSocket 服务端，用于测试 dglab_ws 插件 - -用法: - python dglab_app_simulator.py [ws_url] - - ws_url: WebSocket 服务端地址，默认 ws://127.0.0.1:9999 - -流程: - 1. 先启动 WebSocket 服务端 (socket/v2/backend) - 2. 运行本模拟器，它会自动连接并等待配对 - 3. 在 xiaoli-cli 中用 /dglab connect 连接同一服务端 - 4. 用 /dglab qr 生成二维码，本模拟器会自动响应绑定请求 - 5. 开始测试各种命令 -""" - -import json -import sys -import time -import uuid -import threading - -try: - import websocket -except ImportError: - print("❌ 需要安装 websocket-client: pip install websocket-client") - sys.exit(1) - - -class DGLabAppSimulator: - """模拟 DG-LAB APP""" - - def __init__(self, ws_url="ws://127.0.0.1:9999"): - self.ws_url = ws_url - self.ws = None - self.target_id = "" - self.client_id = "" # 绑定的第三方终端 ID - self.connected = False - self.paired = False - - # 模拟设备状态 - self.strength_a = 0 - self.strength_b = 0 - self.limit_a = 100 - self.limit_b = 100 - - def start(self): - """启动模拟器""" - self.target_id = str(uuid.uuid4()) - connect_url = f"{self.ws_url}/{self.target_id}" - - print(f"🐺 DG-Lab APP 模拟器") - print(f"━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━") - print(f" 服务端: {self.ws_url}") - print(f" targetId: {self.target_id}") - print(f"━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━") - print(f"⏳ 正在连接...") - - self.ws = websocket.WebSocketApp( - connect_url, - on_message=self._on_message, - on_error=self._on_error, - on_close=self._on_close, - on_open=self._on_open - ) - - # 心跳线程 - threading.Thread(target=self._heartbeat, daemon=True).start() - - try: - self.ws.run_forever() - except KeyboardInterrupt: - print("\n👋 已退出") - - def _on_open(self, ws): - self.connected = True - print(f"✅ 已连接到 WebSocket 服务端") - print(f"📱 等待第三方终端发起配对...") - print(f" (在 xiaoli-cli 中执行: /dglab connect {self.ws_url})") - print() - - def _on_close(self, ws, code, reason): - self.connected = False - self.paired = False - print(f"🔌 连接已断开 (code={code})") - - def _on_error(self, ws, error): - print(f"❌ 错误: {error}") - - def _on_message(self, ws, raw): - try: - data = json.loads(raw) - except json.JSONDecodeError: - return - - msg_type = data.get("type", "") - msg = data.get("message", "") - cid = data.get("clientId", "") - tid = data.get("targetId", "") - - if msg_type == "bind": - self._handle_bind(data) - - elif msg_type == "msg": - self._handle_msg(data) - - elif msg_type == "heartbeat": - # 回复心跳 - pass - - elif msg_type == "error": - print(f" ❌ 服务端错误: {msg}") - - elif msg_type == "break": - self.paired = False - self.client_id = "" - print(f" ⚠️ 第三方终端已断开") - - def _handle_bind(self, data): - """处理绑定请求""" - msg = data.get("message", "") - cid = data.get("clientId", "") - tid = data.get("targetId", "") - - if msg == "200": - self.paired = True - self.client_id = cid - print(f" ✅ 配对成功! 与终端 {cid[:8]}... 绑定") - print(f" 现在可以在 xiaoli-cli 中使用 /dglab 命令了") - print() - self._print_status() - elif msg == "400": - print(f" ❌ 配对失败: ID 已被绑定") - else: - # 新的绑定请求，APP 端自动接受 - print(f" 📨 收到绑定请求: clientId={cid[:8]}...") - # 回复绑定成功 - bind_ok = { - "type": "bind", - "clientId": cid, - "targetId": self.target_id, - "message": "200" - } - ws.send(json.dumps(bind_ok)) - self.paired = True - self.client_id = cid - print(f" ✅ 已自动接受配对") - - def _handle_msg(self, data): - """处理来自第三方终端的指令""" - msg = data.get("message", "") - - if msg.startswith("strength-"): - self._handle_strength(msg) - elif msg.startswith("pulse-"): - self._handle_pulse(msg) - elif msg.startswith("clear-"): - self._handle_clear(msg) - else: - print(f" 📨 收到指令: {msg}") - - def _handle_strength(self, msg): - """处理强度指令: strength-通道+模式+数值""" - try: - parts = msg.replace("strength-", "").split("+") - channel = int(parts[0]) # 1=A, 2=B - mode = int(parts[1]) # 0=减少, 1=增加, 2=设为 - value = int(parts[2]) - - ch_name = "A" if channel == 1 else "B" - - if mode == 0: # 减少 - if channel == 1: - self.strength_a = max(0, self.strength_a - value) - else: - self.strength_b = max(0, self.strength_b - value) - print(f" 📉 通道{ch_name} 强度 -{value}") - - elif mode == 1: # 增加 - if channel == 1: - self.strength_a = min(self.limit_a, self.strength_a + value) - else: - self.strength_b = min(self.limit_b, self.strength_b + value) - print(f" 📈 通道{ch_name} 强度 +{value}") - - elif mode == 2: # 设为 - if channel == 1: - self.strength_a = min(self.limit_a, value) - else: - self.strength_b = min(self.limit_b, value) - print(f" 🎯 通道{ch_name} 强度设为 {value}") - - # 回传当前强度 - self._send_strength_feedback() - - except (ValueError, IndexError) as e: - print(f" ❌ 强度指令解析失败: {msg} ({e})") - - def _handle_pulse(self, msg): - """处理波形指令: pulse-A:[hex_data,...]""" - try: - content = msg.replace("pulse-", "") - ch = content[0] # A or B - json_str = content[2:] # 跳过 "A:" - data = json.loads(json_str) - count = len(data) - print(f" 🌊 通道{ch} 收到 {count} 条波形数据") - if count > 0: - sample = data[0] - print(f" 示例: {sample}") - except Exception as e: - print(f" ❌ 波形指令解析失败: {msg} ({e})") - - def _handle_clear(self, msg): - """处理清空指令: clear-1 / clear-2""" - channel = msg.replace("clear-", "") - if channel == "1": - print(f" 🗑️ 通道A 波形队列已清空") - elif channel == "2": - print(f" 🗑️ 通道B 波形队列已清空") - - def _send_strength_feedback(self): - """回传当前强度状态""" - if not self.paired or not self.ws: - return - feedback = { - "type": "msg", - "clientId": self.client_id, - "targetId": self.target_id, - "message": f"strength-{self.strength_a}+{self.strength_b}+{self.limit_a}+{self.limit_b}" - } - try: - self.ws.send(json.dumps(feedback)) - self._print_status() - except Exception: - pass - - def _print_status(self): - """打印当前状态""" - print(f" ┌─────────────────────────┐") - print(f" │ 通道A: {self.strength_a:>3}/{self.limit_a:<3} │") - print(f" │ 通道B: {self.strength_b:>3}/{self.limit_b:<3} │") - print(f" └─────────────────────────┘") - - def _heartbeat(self): - """心跳""" - while self.connected: - time.sleep(60) - if self.ws and self.connected: - try: - hb = { - "type": "heartbeat", - "clientId": self.client_id, - "targetId": self.target_id, - "message": "ping" - } - self.ws.send(json.dumps(hb)) - except Exception: - break - - -if __name__ == "__main__": - url = sys.argv[1] if len(sys.argv) > 1 else "ws://127.0.0.1:9999" - sim = DGLabAppSimulator(url) - sim.start() diff --git a/dglab_ws/dglab_ws.py b/dglab_ws/dglab_ws.py deleted file mode 100644 index 08ce3c5..0000000 --- a/dglab_ws/dglab_ws.py +++ /dev/null @@ -1,567 +0,0 @@ -""" -DG-Lab Coyote WebSocket 控制插件 -基于 DG-LAB-OPENSOURCE 官方 Socket V2 协议 -支持郊狼 3.0 脉冲主机的 WebSocket 远程控制 - -参考文档: -- https://github.com/DG-LAB-OPENSOURCE/DG-LAB-OPENSOURCE -- Socket V2 协议: socket/v2/README.md -""" - -import json -import time -import threading -import uuid -import os - -try: - import websocket - HAS_WS = True -except ImportError: - HAS_WS = False - -try: - import qrcode - HAS_QRCODE = True -except ImportError: - HAS_QRCODE = False - - -class Liugin: - """DG-Lab 郊狼 WebSocket 控制插件""" - - def __init__(self): - self.ws = None - self.client_id = "" - self.target_id = "" - self.connected = False - self.paired = False - self.ws_url = "" - self.strength_a = 0 - self.strength_b = 0 - self.limit_a = 100 - self.limit_b = 100 - self._heartbeat_thread = None - self._running = False - - self.usage = """DG-Lab 郊狼控制插件 使用说明： - -/dglab connect - 连接到 WebSocket 服务器 -/dglab qr - 生成配对二维码 (需APP扫码) -/dglab status - 查看连接状态和通道强度 -/dglab strength <+|-|=|set> - 控制通道强度 - ch: a/b (通道A/B) - 操作: +增加, -减少, =设置为指定值 - 示例: /dglab strength a + 10 - /dglab strength b = 50 -/dglab pulse - 发送波形 - ch: a/b - freq: 频率 (10-240) - intensity: 强度 (0-100) - duration: 持续秒数 - 示例: /dglab pulse a 30 50 5 -/dglab wave - 发送原始波形HEX数据 - 示例: /dglab wave a 0A0A0A0A64646464 -/dglab clear - 清空通道波形队列 (a/b/both) -/dglab limit - 设置通道强度软上限 (0-200) -/dglab disconnect - 断开连接 -""" - - def set_cli(self, cli): - """注册 CLI 命令""" - self.cli = cli - if hasattr(cli, 'register_liugin_command'): - cli.register_liugin_command('dglab', self._cmd_handler) - - def _cmd_handler(self, args): - """处理 /dglab 命令""" - parts = args.strip().split(maxsplit=1) - if not parts: - return self.handle("help") - return self.handle(args) - - def get_tool_info(self): - return { - "name": "dglab_ws", - "description": "DG-Lab 郊狼 3.0 WebSocket 控制插件，支持连接、配对、强度控制、波形发送", - "keywords": [ - "dg-lab", "dglab", "郊狼", "coyote", "websocket", - "脉冲", "强度", "波形", "蓝牙", "控制", "情趣" - ], - "usage": self.usage - } - - def get_mcp_definition(self): - return { - "name": "dglab_ws", - "description": "DG-Lab Coyote 3.0 WebSocket controller plugin", - "inputSchema": { - "type": "object", - "properties": { - "action": { - "type": "string", - "enum": [ - "connect", "disconnect", "status", - "strength", "pulse", "wave", - "clear", "limit", "qr" - ] - }, - "args": {"type": "string", "description": "命令参数"} - }, - "required": ["action"] - } - } - - def convert_mcp_args(self, arguments): - action = arguments.get("action", "") - args = arguments.get("args", "") - return f"{action} {args}" - - # ── 核心协议方法 ────────────────────────────────────── - - def _on_message(self, ws, raw): - """处理服务端消息""" - try: - data = json.loads(raw) - except json.JSONDecodeError: - return - - msg_type = data.get("type", "") - msg = data.get("message", "") - cid = data.get("clientId", "") - tid = data.get("targetId", "") - - if msg_type == "bind": - if cid and not self.client_id: - self.client_id = cid - if msg == "200": - self.paired = True - self.target_id = tid - print(f"\n[DG-Lab] ✅ 配对成功! targetId={tid}") - elif msg == "400": - print("\n[DG-Lab] ❌ 配对失败: 该ID已被绑定") - else: - print(f"\n[DG-Lab] 绑定状态: {msg}") - - elif msg_type == "msg": - if msg.startswith("strength-"): - self._parse_strength(msg) - elif msg.startswith("feedback-"): - btn = msg.split("-")[1] - print(f"\n[DG-Lab] APP反馈按钮: {btn}") - - elif msg_type == "heartbeat": - pass - - elif msg_type == "break": - self.paired = False - print("\n[DG-Lab] ⚠️ APP端已断开连接") - - elif msg_type == "error": - print(f"\n[DG-Lab] ❌ 错误: {msg}") - - def _on_error(self, ws, error): - print(f"\n[DG-Lab] WebSocket错误: {error}") - - def _on_close(self, ws, code, reason): - self.connected = False - self.paired = False - self._running = False - print(f"\n[DG-Lab] 连接已关闭 (code={code}, reason={reason})") - - def _on_open(self, ws): - self.connected = True - print("\n[DG-Lab] ✅ 已连接到WebSocket服务器") - self._running = True - self._heartbeat_thread = threading.Thread( - target=self._heartbeat_loop, daemon=True - ) - self._heartbeat_thread.start() - - def _heartbeat_loop(self): - """每60秒发送心跳""" - while self._running and self.connected: - time.sleep(60) - if self.ws and self.connected: - try: - hb = { - "type": "heartbeat", - "clientId": self.client_id, - "targetId": self.target_id, - "message": "ping" - } - self.ws.send(json.dumps(hb)) - except Exception: - break - - def _parse_strength(self, msg): - """解析APP回传的强度: strength-A+B+A上限+B上限""" - try: - vals = msg.replace("strength-", "").split("+") - if len(vals) >= 4: - self.strength_a = int(vals[0]) - self.strength_b = int(vals[1]) - self.limit_a = int(vals[2]) - self.limit_b = int(vals[3]) - print( - f"\n[DG-Lab] 📊 强度状态: " - f"A={self.strength_a}/{self.limit_a} " - f"B={self.strength_b}/{self.limit_b}" - ) - except (ValueError, IndexError): - pass - - def _send_to_app(self, msg_type, message): - """发送消息到服务端 (转发给APP)""" - if not self.connected or not self.ws: - return False, "未连接到服务器" - if not self.paired: - return False, "尚未与APP配对，请先用APP扫码" - - payload = { - "type": msg_type, - "clientId": self.client_id, - "targetId": self.target_id, - "message": message - } - try: - self.ws.send(json.dumps(payload)) - return True, "ok" - except Exception as e: - return False, str(e) - - def _send_strength(self, channel, mode, value): - """ - 发送强度控制指令 - channel: 1=A, 2=B - mode: 0=减少, 1=增加, 2=设为指定值 - value: 0-200 - """ - msg = f"strength-{channel}+{mode}+{value}" - return self._send_to_app("msg", msg) - - def _send_pulse(self, channel, freq, intensity, duration=5): - """生成并发送波形数据""" - ch = "A" if channel.lower() == "a" else "B" - - def map_freq(f): - f = max(10, min(1000, f)) - if f <= 100: - return f - elif f <= 600: - return (f - 100) // 5 + 100 - else: - return (f - 600) // 10 + 200 - - freq_byte = map_freq(freq) - intensity_byte = max(0, min(100, intensity)) - - wave_hex = "" - for _ in range(25 * max(1, duration)): - wave_hex += f"{freq_byte:02X}{intensity_byte:02X}" * 4 - - pulse_data = [] - for i in range(0, len(wave_hex), 16): - chunk = wave_hex[i:i + 16] - if len(chunk) == 16: - pulse_data.append(chunk) - - if not pulse_data: - pulse_data = [f"{freq_byte:02X}" * 4 + f"{intensity_byte:02X}" * 4] - - batch_size = min(len(pulse_data), 100) - message = f"{ch}:{json.dumps(pulse_data[:batch_size])}" - - payload = { - "type": "clientMsg", - "channel": ch, - "time": duration, - "message": message, - "clientId": self.client_id, - "targetId": self.target_id - } - - if not self.connected or not self.ws: - return False, "未连接到服务器" - if not self.paired: - return False, "尚未与APP配对" - - try: - self.ws.send(json.dumps(payload)) - return True, f"已发送 {batch_size} 条波形数据到通道{ch}，持续{duration}秒" - except Exception as e: - return False, str(e) - - def _send_raw_wave(self, channel, hex_data): - """发送原始HEX波形数据""" - ch = "A" if channel.lower() == "a" else "B" - pulse_data = [] - hex_clean = hex_data.replace(" ", "").replace(",", "") - for i in range(0, len(hex_clean), 16): - chunk = hex_clean[i:i + 16] - if len(chunk) == 16: - pulse_data.append(chunk) - - if not pulse_data: - return False, "波形数据不足8字节 (需要16个HEX字符)" - - message = f"{ch}:{json.dumps(pulse_data[:100])}" - payload = { - "type": "clientMsg", - "channel": ch, - "time": len(pulse_data) // 25 + 1, - "message": message, - "clientId": self.client_id, - "targetId": self.target_id - } - - if not self.connected or not self.ws: - return False, "未连接到服务器" - if not self.paired: - return False, "尚未与APP配对" - - try: - self.ws.send(json.dumps(payload)) - return True, f"已发送 {len(pulse_data)} 条波形数据到通道{ch}" - except Exception as e: - return False, str(e) - - # ── 命令入口 ────────────────────────────────────────── - - def handle(self, args): - """处理工具请求""" - parts = args.strip().split(maxsplit=1) - if not parts: - return self.usage - - action = parts[0].lower() - rest = parts[1] if len(parts) > 1 else "" - - if action == "help": - return self.usage - - elif action == "connect": - return self._do_connect(rest.strip()) - - elif action == "disconnect": - return self._do_disconnect() - - elif action == "qr": - return self._do_qr() - - elif action == "status": - return self._do_status() - - elif action == "strength": - return self._do_strength(rest) - - elif action == "pulse": - return self._do_pulse(rest) - - elif action == "wave": - return self._do_wave(rest) - - elif action == "clear": - return self._do_clear(rest) - - elif action == "limit": - return self._do_limit(rest) - - else: - return f"未知操作: {action}\n" + self.usage - - def _do_connect(self, url): - """连接WebSocket服务器""" - if not HAS_WS: - return "❌ 缺少 websocket-client 库，请运行: pip install websocket-client" - - if self.connected: - return f"已连接到 {self.ws_url}，请先 disconnect" - - if not url: - return "请提供WebSocket地址，如: ws://192.168.1.100:9999" - - self.ws_url = url - self.client_id = str(uuid.uuid4()) - connect_url = f"{url}/{self.client_id}" - - try: - self.ws = websocket.WebSocketApp( - connect_url, - on_message=self._on_message, - on_error=self._on_error, - on_close=self._on_close, - on_open=self._on_open - ) - t = threading.Thread(target=self.ws.run_forever, daemon=True) - t.start() - - for _ in range(30): - if self.connected: - break - time.sleep(0.1) - - if self.connected: - return ( - f"✅ 已连接到 {url}\n" - f"clientId: {self.client_id}\n" - f"请使用 /dglab qr 生成二维码供APP扫码配对" - ) - else: - return f"❌ 连接超时，请检查地址是否正确: {url}" - except Exception as e: - return f"❌ 连接失败: {e}" - - def _do_disconnect(self): - """断开连接""" - if self.ws: - self._running = False - try: - self.ws.close() - except Exception: - pass - self.connected = False - self.paired = False - return "已断开连接" - return "当前没有连接" - - def _do_qr(self): - """生成配对二维码""" - if not self.connected or not self.client_id: - return "❌ 请先连接到WebSocket服务器" - - qr_content = ( - f"https://www.dungeon-lab.com/app-download.php" - f"#DGLAB-SOCKET#{self.ws_url}/{self.client_id}" - ) - - result = f"📱 配对二维码内容:\n{qr_content}\n\n" - result += "请用 DG-LAB APP 扫描此二维码进行配对\n" - - if HAS_QRCODE: - try: - qr = qrcode.QRCode(version=1, box_size=10, border=4) - qr.add_data(qr_content) - qr.make(fit=True) - qr.print_ascii(invert=True) - result += "\n(已在终端显示ASCII二维码)" - except Exception as e: - result += f"\n(二维码生成失败: {e})" - else: - result += "\n💡 安装 qrcode 库可在终端显示ASCII二维码: pip install qrcode" - - return result - - def _do_status(self): - """查看状态""" - lines = ["📊 DG-Lab 连接状态:"] - lines.append(f" 连接: {'✅ 已连接' if self.connected else '❌ 未连接'}") - lines.append(f" 配对: {'✅ 已配对' if self.paired else '❌ 未配对'}") - if self.connected: - lines.append(f" 服务器: {self.ws_url}") - lines.append(f" clientId: {self.client_id}") - if self.paired: - lines.append(f" targetId: {self.target_id}") - lines.append(f" 通道A强度: {self.strength_a}/{self.limit_a}") - lines.append(f" 通道B强度: {self.strength_b}/{self.limit_b}") - return "\n".join(lines) - - def _do_strength(self, args): - """控制强度: /dglab strength """ - parts = args.strip().split() - if len(parts) < 3: - return "格式: strength <+|-|=> <值>\n示例: strength a + 10" - - ch = parts[0].lower() - op = parts[1] - try: - val = int(parts[2]) - except ValueError: - return "强度值必须是整数" - - channel = 1 if ch == "a" else 2 - - if op == "+": - ok, msg = self._send_strength(channel, 1, val) - return f"{'✅' if ok else '❌'} 通道{ch.upper()} 强度+{val}" if ok else f"❌ {msg}" - elif op == "-": - ok, msg = self._send_strength(channel, 0, val) - return f"{'✅' if ok else '❌'} 通道{ch.upper()} 强度-{val}" if ok else f"❌ {msg}" - elif op in ("=", "set"): - val = max(0, min(200, val)) - ok, msg = self._send_strength(channel, 2, val) - return f"{'✅' if ok else '❌'} 通道{ch.upper()} 强度设为 {val}" if ok else f"❌ {msg}" - else: - return f"未知操作: {op}，支持 +, -, =" - - def _do_pulse(self, args): - """发送波形: /dglab pulse """ - parts = args.strip().split() - if len(parts) < 3: - return "格式: pulse <频率10-240> <强度0-100> [持续秒数]" - - ch = parts[0].lower() - try: - freq = int(parts[1]) - intensity = int(parts[2]) - duration = int(parts[3]) if len(parts) > 3 else 5 - except ValueError: - return "参数必须是整数" - - ok, msg = self._send_pulse(ch, freq, intensity, duration) - return f"✅ {msg}" if ok else f"❌ {msg}" - - def _do_wave(self, args): - """发送原始波形HEX: /dglab wave """ - parts = args.strip().split(maxsplit=1) - if len(parts) < 2: - return "格式: wave \n示例: wave a 0A0A0A0A64646464" - - ch = parts[0].lower() - hex_data = parts[1].strip() - - ok, msg = self._send_raw_wave(ch, hex_data) - return f"✅ {msg}" if ok else f"❌ {msg}" - - def _do_clear(self, args): - """清空波形队列""" - ch = args.strip().lower() - if ch in ("a", "1"): - ok, msg = self._send_to_app("msg", "clear-1") - return "✅ 已清空通道A波形队列" if ok else f"❌ {msg}" - elif ch in ("b", "2"): - ok, msg = self._send_to_app("msg", "clear-2") - return "✅ 已清空通道B波形队列" if ok else f"❌ {msg}" - elif ch == "both": - self._send_to_app("msg", "clear-1") - self._send_to_app("msg", "clear-2") - return "✅ 已清空AB双通道波形队列" - else: - return "格式: clear " - - def _do_limit(self, args): - """设置软上限""" - parts = args.strip().split() - if len(parts) < 2: - return "格式: limit <0-200>" - - ch = parts[0].lower() - try: - val = max(0, min(200, int(parts[1]))) - except ValueError: - return "上限值必须是0-200的整数" - - if ch in ("a", "1"): - return ( - f"⚠️ 软上限需要通过蓝牙BF指令设置\n" - f" 当前通道A上限: {self.limit_a}\n" - f" 请在DG-LAB APP中调整软上限" - ) - elif ch in ("b", "2"): - return ( - f"⚠️ 软上限需要通过蓝牙BF指令设置\n" - f" 当前通道B上限: {self.limit_b}\n" - f" 请在DG-LAB APP中调整软上限" - ) - else: - return "格式: limit <0-200>" diff --git a/onebot/README.md b/onebot/README.md deleted file mode 100644 index 77e408f..0000000 --- a/onebot/README.md +++ /dev/null @@ -1,159 +0,0 @@ -# OneBot v11 QQ 机器人插件 - -> 让 xiaoli-cli 成为 QQ 机器人，通过 OneBot 协议连接 QQ 消息平台 - -## 什么是 OneBot - -[OneBot](https://github.com/botuniverse/onebot) 是一个聊天机器人应用标准，支持 QQ、微信等平台。常见的 OneBot 实现： - -| 实现 | 语言 | 说明 | -|------|------|------| -| [go-cqhttp](https://github.com/Mrs4s/go-cqhttp) | Go | 最流行，功能全 | -| [Lagrange.Core](https://github.com/LagrangeDev/Lagrange.Core) | C# | 轻量，NTQQ 协议 | -| [LLOneBot](https://github.com/LLOneBot/LLOneBot) | TypeScript | QQNT 插件形式 | -| [NapCat](https://github.com/NapNeko/NapCatQQ) | TypeScript | 另一个 QQNT 方案 | - -## 前置准备 - -### 1. 安装依赖 - -```bash -pip install websocket-client -``` - -### 2. 安装 OneBot 实现 - -以 Lagrange 为例： - -```bash -# 下载 Lagrange -# https://github.com/LagrangeDev/Lagrange.Core/releases - -# 编辑配置 appsettings.json，设置正向 WebSocket -{ - "Implementations": [{ - "Type": "ForwardWebSocket", - "Host": "0.0.0.0", - "Port": 8080 - }] -} - -# 启动，扫码登录 -./Lagrange.Core -``` - -### 3. 安装插件 - -```bash -# 方式1: 通过插件码安装 -/plugin install onebot - -# 方式2: 手动复制 -cp onebot/onebot.py /path/to/xiaoli-cli/plugins/ -``` - -## 使用教程 - -### 方式一：正向 WebSocket（推荐） - -xiaoli-cli 主动连接 OneBot 服务： - -``` -# 连接 -/onebot connect ws://127.0.0.1:8080 - -# 查看状态 -/onebot status -``` - -### 方式二：反向 WebSocket - -xiaoli-cli 监听端口，等 OneBot 主动连接： - -``` -# 启动监听（默认 8765 端口） -/onebot listen 8765 -``` - -然后在 OneBot 配置中设置反向 WebSocket 地址为 `ws://你的IP:8765`。 - -### 配置触发方式 - -``` -# 只响应 @机器人 的消息（默认） -/onebot config trigger_mode at - -# 响应所有消息 -/onebot config trigger_mode all - -# 响应以特定前缀开头的消息 -/onebot config trigger_mode prefix -/onebot config trigger_prefix ! -``` - -### 群白名单 - -``` -# 只在指定群响应 -/onebot config group_whitelist 123456789 -/onebot config group_whitelist 987654321 -``` - -### 手动发消息 - -``` -# 发私聊 -/onebot send private 123456 你好 - -# 发群聊 -/onebot send group 123456789 大家好 -``` - -## 命令速查 - -| 命令 | 说明 | -|------|------| -| `/onebot connect ` | 正向 WebSocket 连接 | -| `/onebot listen [port]` | 反向 WebSocket 监听 | -| `/onebot disconnect` | 断开连接 | -| `/onebot status` | 查看状态 | -| `/onebot config ` | 配置参数 | -| `/onebot send ` | 手动发消息 | -| `/onebot help` | 显示帮助 | - -## 工作原理 - -``` -QQ 用户发消息 - ↓ -OneBot 实现 (Lagrange/go-cqhttp) - ↓ WebSocket 推送 -onebot 插件 (接收消息) - ↓ 注入上下文 -xiaoli-cli AI 引擎 (处理消息) - ↓ 生成回复 -onebot 插件 (发送回复) - ↓ OneBot API -QQ 用户收到回复 -``` - -## 常见问题 - -**Q: 连接不上怎么办？** -A: 检查 OneBot 实现是否启动，端口是否正确，防火墙是否放行。 - -**Q: 群里不回复？** -A: 检查 trigger_mode 设置，以及是否在群白名单中。 - -**Q: 支持图片/语音吗？** -A: 当前版本只处理文本消息。图片消息会提取文本部分处理。 - -**Q: 安全性如何？** -A: 建议设置群白名单，不要在公开群暴露。管理员功能可后续扩展。 - -## 参考 - -- [OneBot 标准文档](https://github.com/botuniverse/onebot) -- [OneBot v11 接口文档](https://github.com/botuniverse/onebot-11) -- [Lagrange.Core](https://github.com/LagrangeDev/Lagrange.Core) -- [go-cqhttp](https://github.com/Mrs4s/go-cqhttp) diff --git a/onebot/onebot.py b/onebot/onebot.py deleted file mode 100644 index a1b553e..0000000 --- a/onebot/onebot.py +++ /dev/null @@ -1,541 +0,0 @@ -""" -OneBot v11 插件 — 让 xiaoli-cli 成为 QQ 机器人 -通过 WebSocket 连接 OneBot 实现（go-cqhttp / Lagrange / LLOneBot 等） - -功能: -- WebSocket 正向/反向连接 -- 接收私聊和群消息，转发给 AI 处理 -- AI 回复自动发回 QQ -- 支持 @机器人 触发 / 全部消息 / 前缀触发 -- 群消息可配置是否响应 -""" - -import json -import time -import threading -import re -import os - -try: - import websocket - HAS_WS = True -except ImportError: - HAS_WS = False - - -class Liugin: - """OneBot v11 QQ 机器人插件""" - - def __init__(self): - self.ws = None - self.connected = False - self.running = False - self.ws_url = "" - self.mode = "" # "forward" 正向 / "reverse" 反向 - - # 配置 - self.trigger_mode = "at" # at: 只响应@ / prefix: 前缀触发 / all: 全部消息 - self.trigger_prefix = "/" - self.group_whitelist = [] # 群白名单，空=全部响应 - self.admin_users = [] # 管理员 QQ 号 - self.listen_port = 8765 # 反向 WebSocket 监听端口 - self.self_id = "" # 机器人自身 QQ 号 - - # 消息队列 (用于 AI 回复) - self._pending_replies = {} # {message_id: {"context": ..., "reply": ...}} - self._msg_counter = 0 - - self.cli = None - self._listener_thread = None - - self.usage = """OneBot v11 QQ 机器人插件 使用说明： - -/onebot connect 连接 OneBot (正向WebSocket) -/onebot listen [port] 启动反向WebSocket监听 (默认8765) -/onebot disconnect 断开连接 -/onebot status 查看连接状态 -/onebot config 配置: - trigger_mode at|prefix|all 触发模式 - trigger_prefix <前缀> 前缀触发的前缀 - group_whitelist <群号> 添加群白名单 - admin 添加管理员 - self_id 设置机器人QQ号 -/onebot send <消息> 手动发消息 - type: private/group - id: QQ号/群号 -/onebot help 显示帮助 -""" - - def set_cli(self, cli): - self.cli = cli - if hasattr(cli, 'register_liugin_command'): - cli.register_liugin_command('onebot', self._cmd_handler) - - def _cmd_handler(self, args): - parts = args.strip().split(maxsplit=1) - if not parts: - return self.handle("help") - return self.handle(args) - - def get_tool_info(self): - return { - "name": "onebot", - "description": "OneBot v11 QQ 机器人插件，连接 go-cqhttp/Lagrange/LLOneBot 等，让 AI 接管 QQ 消息", - "keywords": [ - "onebot", "qq", "机器人", "bot", "go-cqhttp", - "lagrange", "消息", "群聊", "私聊", "cqhttp" - ], - "usage": self.usage - } - - def get_mcp_definition(self): - return { - "name": "onebot", - "description": "OneBot v11 QQ bot plugin", - "inputSchema": { - "type": "object", - "properties": { - "action": { - "type": "string", - "enum": ["connect", "listen", "disconnect", "status", "config", "send"] - }, - "args": {"type": "string"} - }, - "required": ["action"] - } - } - - def convert_mcp_args(self, arguments): - return f"{arguments.get('action', '')} {arguments.get('args', '')}" - - # ── WebSocket 连接 ── - - def _on_message(self, ws, raw): - try: - data = json.loads(raw) - except json.JSONDecodeError: - return - self._dispatch(data) - - def _on_error(self, ws, error): - print(f"\n[OneBot] ❌ WebSocket 错误: {error}") - - def _on_close(self, ws, code, reason): - self.connected = False - self.running = False - print(f"\n[OneBot] 🔌 连接断开 (code={code})") - - def _on_open(self, ws): - self.connected = True - self.running = True - print(f"\n[OneBot] ✅ 已连接到 OneBot 服务") - - def _connect_forward(self, url): - """正向 WebSocket 连接""" - if not HAS_WS: - return "❌ 需要 websocket-client: pip install websocket-client" - - self.ws_url = url - self.mode = "forward" - - try: - self.ws = websocket.WebSocketApp( - url, - on_message=self._on_message, - on_error=self._on_error, - on_close=self._on_close, - on_open=self._on_open - ) - self._listener_thread = threading.Thread(target=self.ws.run_forever, daemon=True) - self._listener_thread.start() - - for _ in range(30): - if self.connected: - break - time.sleep(0.1) - - if self.connected: - return f"✅ 已连接到 {url}" - return f"❌ 连接超时: {url}" - except Exception as e: - return f"❌ 连接失败: {e}" - - def _start_reverse_listener(self, port): - """反向 WebSocket 监听""" - try: - from http.server import HTTPServer, BaseHTTPRequestHandler - import hashlib - import struct - except ImportError: - return "❌ 缺少标准库组件" - - self.listen_port = port - self.mode = "reverse" - plugin_ref = self - - class ReverseWSHandler(BaseHTTPRequestHandler): - def do_GET(self): - # WebSocket 握手 - if self.headers.get('Upgrade', '').lower() != 'websocket': - self.send_error(400) - return - - key = self.headers.get('Sec-WebSocket-Key', '') - accept = hashlib.sha1( - (key + "258EAFA5-E914-47DA-95CA-5AB5DC65C740").encode() - ).digest() - - self.send_response(101, 'Switching Protocols') - self.send_header('Upgrade', 'websocket') - self.send_header('Connection', 'Upgrade') - self.send_header('Sec-WebSocket-Accept', - __import__('base64').b64encode(accept).decode()) - self.end_headers() - - # 简化的 WebSocket 帧处理 - plugin_ref._handle_reverse_ws(self.rfile, self.wfile, self) - - def log_message(self, format, *args): - pass # 静默日志 - - class ReverseHTTPServer(HTTPServer): - allow_reuse_address = True - - def _run_server(): - try: - server = ReverseHTTPServer(('0.0.0.0', port), ReverseWSHandler) - self.connected = True - self.running = True - print(f"\n[OneBot] ✅ 反向 WebSocket 监听中: ws://0.0.0.0:{port}") - server.serve_forever() - except Exception as e: - print(f"\n[OneBot] ❌ 监听失败: {e}") - - self._listener_thread = threading.Thread(target=_run_server, daemon=True) - self._listener_thread.start() - return f"✅ 反向 WebSocket 已启动: ws://0.0.0.0:{port}" - - def _handle_reverse_ws(self, rfile, wfile, handler): - """处理反向 WebSocket 连接""" - self.connected = True - self.running = True - print(f"\n[OneBot] ✅ OneBot 已连接") - - try: - while self.running: - # 读取 WebSocket 帧 - header = rfile.read(2) - if len(header) < 2: - break - - opcode = header[0] & 0x0F - masked = bool(header[1] & 0x80) - length = header[1] & 0x7F - - if length == 126: - length = struct.unpack('>H', rfile.read(2))[0] - elif length == 127: - length = struct.unpack('>Q', rfile.read(8))[0] - - mask_key = rfile.read(4) if masked else b'' - payload = rfile.read(length) - - if masked: - payload = bytes( - b ^ mask_key[i % 4] - for i, b in enumerate(payload) - ) - - if opcode == 1: # Text - try: - data = json.loads(payload.decode('utf-8')) - self._dispatch(data) - except json.JSONDecodeError: - pass - elif opcode == 8: # Close - break - elif opcode == 9: # Ping - # 发 Pong - self._ws_send_pong(wfile, payload) - - except Exception as e: - print(f"\n[OneBot] 连接断开: {e}") - finally: - self.connected = False - - def _ws_send_pong(self, wfile, payload): - """发送 Pong 帧""" - try: - frame = bytes([0x8A, len(payload)]) + payload - wfile.write(frame) - wfile.flush() - except Exception: - pass - - # ── 消息分发 ── - - def _dispatch(self, data): - """分发 OneBot 事件""" - post_type = data.get("post_type", "") - - if post_type == "message": - self._handle_message(data) - elif post_type == "meta_event": - sub_type = data.get("meta_event_type", "") - if sub_type == "heartbeat": - pass - elif sub_type == "lifecycle": - if data.get("sub_type") == "connect": - self.self_id = str(data.get("self_id", "")) - print(f"\n[OneBot] 🤖 机器人已上线: {self.self_id}") - elif post_type == "request": - self._handle_request(data) - - def _handle_message(self, data): - """处理收到的消息""" - message_type = data.get("message_type", "") - raw_msg = data.get("raw_message", data.get("message", "")) - user_id = str(data.get("user_id", "")) - group_id = str(data.get("group_id", "")) - message_id = data.get("message_id", 0) - sender = data.get("sender", {}) - nickname = sender.get("nickname", user_id) - - # 群白名单检查 - if message_type == "group" and self.group_whitelist: - if group_id not in self.group_whitelist: - return - - # 解析纯文本 - text = self._extract_text(raw_msg) - - # 判断是否需要响应 - should_respond = False - - if message_type == "private": - should_respond = True - elif message_type == "group": - if self.trigger_mode == "all": - should_respond = True - elif self.trigger_mode == "at": - if f"[CQ:at,qq={self.self_id}]" in raw_msg: - should_respond = True - text = text.replace(f"@{self.self_id}", "").strip() - elif self.trigger_mode == "prefix": - if text.startswith(self.trigger_prefix): - should_respond = True - text = text[len(self.trigger_prefix):].strip() - - if not should_respond or not text: - return - - # 打印收到的消息 - if message_type == "group": - print(f"\n[OneBot] 💬 [{group_id}] {nickname}: {text[:80]}") - else: - print(f"\n[OneBot] 💬 {nickname}: {text[:80]}") - - # 交给 AI 处理 - self._process_with_ai(text, message_type, user_id, group_id, message_id, nickname) - - def _extract_text(self, raw_msg): - """从 CQ 码消息中提取纯文本""" - # 移除 CQ 码 - text = re.sub(r'\[CQ:[^\]]+\]', '', raw_msg) - return text.strip() - - def _handle_request(self, data): - """处理请求事件（好友申请、群邀请等）""" - request_type = data.get("request_type", "") - if request_type == "friend": - # 自动同意好友请求 - self._send_api("set_friend_add_request", { - "flag": data.get("flag", ""), - "approve": True - }) - print(f"\n[OneBot] ✅ 已自动同意好友请求: {data.get('user_id')}") - - def _process_with_ai(self, text, msg_type, user_id, group_id, msg_id, nickname): - """将消息交给 AI 处理并发回回复""" - if not self.cli: - return - - # 构建上下文 - if msg_type == "group": - context = f"[QQ群消息] 群{group_id} 用户{nickname}({user_id}): {text}" - else: - context = f"[QQ私聊] 用户{nickname}({user_id}): {text}" - - def _ai_worker(): - try: - # 调用 AI 处理 - response = self.cli._generate_response_with_animation(context) - if response: - # 清理回复 - reply = self._clean_reply(response) - if reply: - if msg_type == "group": - self._send_group_msg(group_id, reply, at_user=user_id) - else: - self._send_private_msg(user_id, reply) - except Exception as e: - print(f"\n[OneBot] ❌ AI 处理失败: {e}") - - threading.Thread(target=_ai_worker, daemon=True).start() - - def _clean_reply(self, response): - """清理 AI 回复，去掉 JSON 指令等""" - # 去掉工具调用 JSON - response = re.sub(r'\{"action":\s*"use_tool"[^}]*\}', '', response) - response = re.sub(r'\{"action":\s*"continue"[^}]*\}', '', response) - response = re.sub(r'[^<]*', '', response) - response = response.strip() - - # 限制长度 - if len(response) > 2000: - response = response[:1997] + "..." - - return response - - # ── OneBot API 调用 ── - - def _send_api(self, action, params=None): - """调用 OneBot API""" - if not self.connected or not self.ws: - return None - - self._msg_counter += 1 - payload = { - "action": action, - "params": params or {}, - "echo": self._msg_counter - } - - try: - if self.mode == "forward": - self.ws.send(json.dumps(payload)) - return self._msg_counter - except Exception as e: - print(f"\n[OneBot] ❌ API 调用失败: {e}") - return None - - def _send_private_msg(self, user_id, message): - """发送私聊消息""" - return self._send_api("send_private_msg", { - "user_id": int(user_id), - "message": message - }) - - def _send_group_msg(self, group_id, message, at_user=None): - """发送群消息""" - if at_user: - message = f"[CQ:at,qq={at_user}] {message}" - return self._send_api("send_group_msg", { - "group_id": int(group_id), - "message": message - }) - - # ── 命令入口 ── - - def handle(self, args): - parts = args.strip().split(maxsplit=1) - if not parts: - return self.usage - - action = parts[0].lower() - rest = parts[1] if len(parts) > 1 else "" - - if action == "help": - return self.usage - elif action == "connect": - return self._do_connect(rest.strip()) - elif action == "listen": - port = int(rest.strip()) if rest.strip().isdigit() else 8765 - return self._start_reverse_listener(port) - elif action == "disconnect": - return self._do_disconnect() - elif action == "status": - return self._do_status() - elif action == "config": - return self._do_config(rest) - elif action == "send": - return self._do_send(rest) - else: - return f"未知命令: {action}\n{self.usage}" - - def _do_connect(self, url): - if not url: - return "请提供 WebSocket 地址: /onebot connect ws://127.0.0.1:6700" - return self._connect_forward(url) - - def _do_disconnect(self): - self.running = False - if self.ws: - try: - self.ws.close() - except Exception: - pass - self.connected = False - return "已断开连接" - - def _do_status(self): - lines = ["🤖 OneBot 状态:"] - lines.append(f" 连接: {'✅ 已连接' if self.connected else '❌ 未连接'}") - lines.append(f" 模式: {self.mode or '未设置'}") - if self.ws_url: - lines.append(f" 地址: {self.ws_url}") - lines.append(f" 端口: {self.listen_port}") - lines.append(f" 机器人QQ: {self.self_id or '未知'}") - lines.append(f" 触发模式: {self.trigger_mode}") - if self.trigger_mode == "prefix": - lines.append(f" 触发前缀: {self.trigger_prefix}") - if self.group_whitelist: - lines.append(f" 群白名单: {', '.join(self.group_whitelist)}") - if self.admin_users: - lines.append(f" 管理员: {', '.join(self.admin_users)}") - return "\n".join(lines) - - def _do_config(self, args): - parts = args.split(maxsplit=1) - if len(parts) < 2: - return "用法: /onebot config \n可用: trigger_mode, trigger_prefix, group_whitelist, admin, self_id" - - key, val = parts[0].lower(), parts[1].strip() - - if key == "trigger_mode": - if val not in ("at", "prefix", "all"): - return "触发模式: at / prefix / all" - self.trigger_mode = val - return f"✅ 触发模式: {val}" - elif key == "trigger_prefix": - self.trigger_prefix = val - return f"✅ 触发前缀: {val}" - elif key == "group_whitelist": - if val not in self.group_whitelist: - self.group_whitelist.append(val) - return f"✅ 群白名单: {self.group_whitelist}" - elif key == "admin": - if val not in self.admin_users: - self.admin_users.append(val) - return f"✅ 管理员: {self.admin_users}" - elif key == "self_id": - self.self_id = val - return f"✅ 机器人QQ: {val}" - else: - return f"未知配置: {key}" - - def _do_send(self, args): - parts = args.split(maxsplit=2) - if len(parts) < 3: - return "用法: /onebot send <消息>" - - msg_type, target_id, message = parts[0].lower(), parts[1], parts[2] - - if msg_type in ("private", "p", "私聊"): - self._send_private_msg(target_id, message) - return f"✅ 已发送私聊消息给 {target_id}" - elif msg_type in ("group", "g", "群"): - self._send_group_msg(target_id, message) - return f"✅ 已发送群消息到 {target_id}" - else: - return "类型: private / group" diff --git a/plugin_registry.json b/plugin_registry.json deleted file mode 100644 index 72606bb..0000000 --- a/plugin_registry.json +++ /dev/null @@ -1,30 +0,0 @@ -{ - "version": 1, - "description": "小狸 CLI 插件码注册表", - "plugins": { - "dglab": { - "name": "DG-Lab 郊狼 WebSocket 控制", - "description": "基于 DG-LAB-OPENSOURCE 官方 Socket V2 协议，支持郊狼 3.0 脉冲主机远程控制", - "author": "xiaoli-cli-plugins", - "version": "1.0.0", - "file": "dglab_ws.py", - "url": "https://gitee.com/shuiyu1123/xiaoli-cli-plugins/raw/master/dglab_ws/dglab_ws.py", - "docs": "https://gitee.com/shuiyu1123/xiaoli-cli-plugins/raw/master/dglab_ws/README.md", - "deps": ["websocket-client"], - "optional_deps": ["qrcode"], - "keywords": ["dg-lab", "dglab", "郊狼", "coyote", "websocket", "脉冲"] - }, - "onebot": { - "name": "OneBot v11 QQ 机器人", - "description": "让 xiaoli-cli 成为 QQ 机器人，连接 go-cqhttp/Lagrange/LLOneBot 等 OneBot 实现", - "author": "xiaoli-cli-plugins", - "version": "1.0.0", - "file": "onebot.py", - "url": "https://gitee.com/shuiyu1123/xiaoli-cli-plugins/raw/master/onebot/onebot.py", - "docs": "https://gitee.com/shuiyu1123/xiaoli-cli-plugins/raw/master/onebot/README.md", - "deps": ["websocket-client"], - "optional_deps": [], - "keywords": ["onebot", "qq", "机器人", "bot", "go-cqhttp", "lagrange", "群聊", "私聊"] - } - } -} diff --git a/启动小狸就启动原神防止用户没原神玩会无聊死插件/pluginsgenshin_launcher_plugin.py b/启动小狸就启动原神防止用户没原神玩会无聊死插件/pluginsgenshin_launcher_plugin.py deleted file mode 100644 index d6c3545..0000000 --- a/启动小狸就启动原神防止用户没原神玩会无聊死插件/pluginsgenshin_launcher_plugin.py +++ /dev/null @@ -1,163 +0,0 @@ -""" -启动小狸就启动原神防止用户没原神玩会无聊死插件 (全自动版) -====================================================== -加载即运行，无需任何命令。 -启动小狸 → 检测原神 → 有游戏就启动 / 有启动器就打开 / 什么都没有就安装启动器。 -真正的防无聊守护神。 -""" - -import os -import sys -import subprocess -import tempfile -import time -import threading -import requests -from pathlib import Path -from colorama import Fore, Style - - -class Plugin: - """原神自动守护插件""" - - # 官方启动器下载链接（定期更新） - LAUNCHER_DOWNLOAD_URL = ( - "https://download.cg.qq.com/launcher/hoyoplay/HoYoPlay_installer_202503121746.exe" - ) - BACKUP_LAUNCHER_URL = ( - "https://autopatchcn.yuanshen.com/client_app/download/launcher/202503121746_1.0.0.0.exe" - ) - - # 常见游戏路径 - GENSHIN_PATHS = [ - r"C:\Program Files\Genshin Impact\Genshin Impact Game\YuanShen.exe", - r"C:\Program Files\Genshin Impact\Genshin Impact Game\GenshinImpact.exe", - r"C:\Program Files (x86)\Genshin Impact\Genshin Impact Game\YuanShen.exe", - r"D:\Genshin Impact\Genshin Impact Game\YuanShen.exe", - r"E:\Genshin Impact\Genshin Impact Game\YuanShen.exe", - ] - - LAUNCHER_DIR = r"C:\Program Files\miHoYo Launcher" - LAUNCHER_EXE = os.path.join(LAUNCHER_DIR, "launcher.exe") - - def __init__(self): - # 插件加载时自动触发，为避免阻塞主线程，新开线程执行 - print(f"{Fore.MAGENTA}🧸 原神守护插件已激活！正在尝试启动原神...{Style.RESET_ALL}") - t = threading.Thread(target=self._auto_launch, daemon=True) - t.start() - - def set_cli(self, cli): - """保存CLI引用（本插件不需要命令注册）""" - self.cli = cli - - def get_tool_info(self): - """提供工具信息（虽然不会主动调用，但会让AI知道这个插件存在）""" - return { - "name": "auto_genshin_launcher", - "description": "启动小狸即自动检测并启动原神/安装米哈游启动器，无需手动操作。", - "keywords": ["原神", "启动", "自动", "米哈游"], - "usage": "本插件无需手动调用，加载即生效。" - } - - def handle(self, args): - """防呆接口，如果有笨笨的AI非要调用，就返回状态""" - return "✨ 原神守护已就绪，它在后台默默守护着你～" - - # ---------- 核心自动逻辑 ---------- - def _auto_launch(self): - """自动检测并启动原神/启动器""" - if os.name != 'nt': - print(f"{Fore.RED}🛑 原神守护仅支持Windows系统～{Style.RESET_ALL}") - return - - # 1. 已安装游戏 → 直接启动 - game_path = self._find_genshin_game() - if game_path: - print(f"{Fore.GREEN}🎮 检测到原神！启动游戏...{Style.RESET_ALL}") - self._launch_genshin(game_path) - return - - # 2. 已安装启动器 → 打开启动器 - if self._launcher_exists(): - print(f"{Fore.GREEN}✅ 米哈游启动器已安装，正在打开...{Style.RESET_ALL}") - self._launch_launcher() - return - - # 3. 什么都没有 → 下载安装启动器 - print(f"{Fore.YELLOW}📦 未检测到原神和启动器，正在自动安装米哈游启动器（需要管理员权限）...{Style.RESET_ALL}") - if self._download_and_install_launcher(): - print(f"{Fore.GREEN}🎉 启动器安装完成！正在打开...{Style.RESET_ALL}") - self._launch_launcher() - else: - print(f"{Fore.RED}😭 安装失败，请手动从官网下载：https://ys.mihoyo.com{Style.RESET_ALL}") - - def _find_genshin_game(self): - for path in self.GENSHIN_PATHS: - if os.path.exists(path): - return path - return None - - def _launcher_exists(self): - return os.path.exists(self.LAUNCHER_EXE) - - def _launch_genshin(self, exe_path): - try: - subprocess.Popen([exe_path], cwd=os.path.dirname(exe_path)) - print(f"{Fore.CYAN}✨ 原神正在启动，旅行者，提瓦特见！{Style.RESET_ALL}") - except Exception as e: - print(f"{Fore.RED}启动原神失败: {e}{Style.RESET_ALL}") - - def _launch_launcher(self): - try: - subprocess.Popen([self.LAUNCHER_EXE], cwd=self.LAUNCHER_DIR) - print(f"{Fore.CYAN}✅ 启动器已打开，请手动下载原神～{Style.RESET_ALL}") - except Exception as e: - print(f"{Fore.RED}打开启动器失败: {e}{Style.RESET_ALL}") - - def _download_and_install_launcher(self): - tmp_dir = tempfile.gettempdir() - installer_path = os.path.join(tmp_dir, "HoYoPlay_installer.exe") - if os.path.exists(installer_path): - os.remove(installer_path) - - url = self.LAUNCHER_DOWNLOAD_URL - if not self._download_file(url, installer_path): - print(f"{Fore.YELLOW}⚠️ 主链接失败，尝试备用链接...{Style.RESET_ALL}") - if not self._download_file(self.BACKUP_LAUNCHER_URL, installer_path): - return False - - print(f"{Fore.YELLOW}⚙️ 正在静默安装（等待约30秒）...{Style.RESET_ALL}") - try: - # 静默安装，/S 适用于NSIS安装包 - subprocess.run([installer_path, "/S"], shell=True, capture_output=True) - time.sleep(15) # 等待安装完成 - # 可选尝试其他参数 - if not self._launcher_exists(): - subprocess.run([installer_path, "/verysilent", "/norestart"], shell=True) - time.sleep(10) - if os.path.exists(installer_path): - os.remove(installer_path) - return self._launcher_exists() - except Exception as e: - print(f"{Fore.RED}安装异常: {e}{Style.RESET_ALL}") - return False - - def _download_file(self, url, save_path): - try: - headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"} - response = requests.get(url, stream=True, headers=headers, timeout=60) - response.raise_for_status() - total = int(response.headers.get("content-length", 0)) - downloaded = 0 - with open(save_path, "wb") as f: - for chunk in response.iter_content(chunk_size=8192): - f.write(chunk) - downloaded += len(chunk) - if total > 0: - pct = int(downloaded / total * 100) - print(f"\r📥 下载启动器: {pct}% ({downloaded//1024}KB/{total//1024}KB)", end="", flush=True) - print() - return True - except Exception as e: - print(f"下载失败: {e}") - return False \ No newline at end of file