From 125e00a48504e5c708ee16e16fffb58882f87314 Mon Sep 17 00:00:00 2001 From: xiaoli-cli bot Date: Tue, 23 Jun 2026 09:25:25 +0000 Subject: [PATCH] =?UTF-8?q?chore:=20=E6=95=B4=E7=90=86=E6=8F=92=E4=BB=B6?= =?UTF-8?q?=E5=88=86=E7=B1=BB=20-=20=E5=88=A0=E9=99=A4=E5=9B=BE=E5=83=8F?= =?UTF-8?q?=E7=B1=BB=EF=BC=8C=E7=A7=BB=E5=8A=A8=E5=A8=B1=E4=B9=90=E7=B1=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 删除（图像类）: - ai_image.py (AI生图) - image_generator.py (ComfyUI生图) - send_image.py (远程发图) 移至插件仓库（娱乐/可选）: - color_picker.py - bookmark_manager.py - audio_player.py 保留24个必备生产力插件: 核心: code_editor/git_tools/file_manager/cmd_executor/terminal 工程: auto_engineer/frontend_tester/workflow/sub_agent 搜索: ai_search/tool_search/network_tools 自动化: browser_auto/gui_auto 数据: json_toolkit/regex_tester/markdown_tool/id_generator/snippet_manager 系统: scheduler/task_manager/memory_plugin/operit_bridge/speech_recognition 更新 README.md/plugin_registry.json/plugins_config.py --- README.md | 27 +- plugin_registry.json | 44 ++- plugins/ai_image.py | 589 ------------------------------------ plugins/audio_player.py | 169 ----------- plugins/bookmark_manager.py | 489 ------------------------------ plugins/color_picker.py | 459 ---------------------------- plugins/image_generator.py | 370 ---------------------- plugins/send_image.py | 144 --------- plugins_config.py | 18 +- tests/test_terminal.py | 310 ++++++++----------- 10 files changed, 194 insertions(+), 2425 deletions(-) delete mode 100644 plugins/ai_image.py delete mode 100644 plugins/audio_player.py delete mode 100644 plugins/bookmark_manager.py delete mode 100644 plugins/color_picker.py delete mode 100644 plugins/image_generator.py delete mode 100644 plugins/send_image.py diff --git a/README.md b/README.md index a2ca955..0902435 100644 --- a/README.md +++ b/README.md @@ -94,7 +94,7 @@ launcher.exe } ``` -## 插件列表 (19 个) +## 插件列表 (24 个) > 更多社区插件请访问 [xiaoli-cli-plugins](https://gitee.com/shuiyu1123/xiaoli-cli-plugins)，使用 `/plugin install <插件码>` 一键安装。 @@ -114,24 +114,28 @@ launcher.exe | 插件 | 操作数 | 说明 | |------|--------|------| | `sub_agent` | 8 | 子 Agent 系统。spawn/list/status/result/send/kill，多 Agent 并行协作 | -| `browser_auto` | 28 | 浏览器自动化。导航/点击/填写/截图/JS执行/PDF/cookies（基于 Playwright） | -| `gui_auto` | 20 | 桌面 GUI 自动化。基于 Windows UIA，屏幕元素识别/点击/输入/截图（仅 Windows） | -| `ai_image` | — | AI 生图。基于 ai.2x.nz 自然语言生图，异步生成 + WebSocket 进度推送 | -| `image_generator` | 7 | 图像生成。基于 ComfyUI 在线服务，自然语言/Tag 双模式，28 个角色工作流 | -| `memory_plugin` | 11 | 记忆系统。AI 写日记/搜索记忆/管理聊天记录/长期记忆 | -| `scheduler` | 4 | 定时任务。支持相对时间/绝对时间/重复任务，到期自动激活 AI | -| `tool_search` | — | 工具搜索。按关键词发现可用插件 | +| `terminal` | 8 | 终端模拟器。AI 的虚拟键盘和屏幕，打字/按键/看屏幕，支持 Python REPL | +| `operit_bridge` | — | Operit 双向工具桥接。解析 47 个 Operit 脚本，288 个工具，格式转换 | ### 辅助工具 | 插件 | 说明 | |------|------| +| `browser_auto` | 浏览器自动化。导航/点击/填写/截图/JS执行（基于 Playwright） | +| `gui_auto` | 桌面 GUI 自动化。Windows UIA 元素识别/点击/输入（仅 Windows） | +| `memory_plugin` | 记忆系统。AI 写日记/搜索记忆/管理聊天记录 | +| `scheduler` | 定时任务。支持相对/绝对/重复任务，到期自动激活 AI | +| `tool_search` | 工具搜索。按关键词发现可用插件 | +| `workflow` | 工作流引擎。YAML 多步骤自动化，变量传递，条件分支 | | `network_tools` | 网络工具。ping/get/status/headers/ip | | `ai_search` | AI 搜索。jina.ai 智能搜索 + 网页内容提取 | | `frontend_tester` | 前端测试。HTML/CSS/JS 语法检查、响应式验证 | | `speech_recognition` | 语音识别。录音 + 语音转文字（Whisper） | -| `audio_player` | 音频播放 | -| `send_image` | 远程模式下发图片到手机 | +| `json_toolkit` | JSON 工具箱。格式化/压缩/校验/路径查询/差异比较 | +| `regex_tester` | 正则表达式测试器。匹配/提取/替换/解释 | +| `snippet_manager` | 代码片段管理器。保存/搜索/复用常用代码 | +| `markdown_tool` | Markdown 工具。TOC/统计/表格生成/待办提取 | +| `id_generator` | ID 生成器。UUID/NanoID/短ID/雪花ID/ULID | ## 记忆系统 @@ -285,7 +289,8 @@ xiaoli-cli/ │ ├── ai_search.py │ ├── frontend_tester.py │ ├── speech_recognition.py -│ └── send_image.py +│ ├── terminal.py +│ └── operit_bridge.py ├── skills/ # 技能 ├── image_engine/ # 图像识别引擎 ├── tests/ # 测试套件 (150+ 用例) diff --git a/plugin_registry.json b/plugin_registry.json index f71623e..52b5d1c 100644 --- a/plugin_registry.json +++ b/plugin_registry.json @@ -10,9 +10,20 @@ "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", "脉冲"] + "deps": [ + "websocket-client" + ], + "optional_deps": [ + "qrcode" + ], + "keywords": [ + "dg-lab", + "dglab", + "郊狼", + "coyote", + "websocket", + "脉冲" + ] }, "operit_bridge": { "name": "Operit Bridge 双向工具桥接", @@ -24,7 +35,15 @@ "docs": "", "deps": [], "optional_deps": [], - "keywords": ["operit", "bridge", "桥接", "兼容", "格式转换", "METADATA", "插件互通"] + "keywords": [ + "operit", + "bridge", + "桥接", + "兼容", + "格式转换", + "METADATA", + "插件互通" + ] }, "onebot": { "name": "OneBot v11 QQ 机器人", @@ -34,9 +53,20 @@ "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"], + "deps": [ + "websocket-client" + ], "optional_deps": [], - "keywords": ["onebot", "qq", "机器人", "bot", "go-cqhttp", "lagrange", "群聊", "私聊"] + "keywords": [ + "onebot", + "qq", + "机器人", + "bot", + "go-cqhttp", + "lagrange", + "群聊", + "私聊" + ] } } -} +} \ No newline at end of file diff --git a/plugins/ai_image.py b/plugins/ai_image.py deleted file mode 100644 index 8602921..0000000 --- a/plugins/ai_image.py +++ /dev/null @@ -1,589 +0,0 @@ -""" -AI 生图插件 - 基于 ai.2x.nz 自然语言生图接口 -支持自然语言描述、直接 Tag、工作流切换、画风选择等 -""" - -import os -import json -import time -import threading -import hashlib -import re -from datetime import datetime -from urllib.parse import quote - -try: - import requests - REQUESTS_AVAILABLE = True -except ImportError: - REQUESTS_AVAILABLE = False - -try: - import websocket - WEBSOCKET_AVAILABLE = True -except ImportError: - WEBSOCKET_AVAILABLE = False - - -class Liugin: - """AI 生图插件 - 调用 ai.2x.nz 接口生成图片""" - - BASE_URL = "https://ai.2x.nz" - WS_URL = "wss://ai.2x.nz/ws/run" - API_URL = "https://ai.2x.nz/api" - - # 保存目录 - DEFAULT_SAVE_DIR = "generated_images" - - def __init__(self): - self.usage = """AI 生图工具 - 基于自然语言描述生成图片 - -基本用法: - generate <描述> - 用自然语言描述生成图片 - tag <直接Tag> - 用 Danbooru Tag 生成图片 - both <<<>>> <描述> - 同时使用 Tag 和自然语言 - -高级用法: - workflows - 列出所有可用工作流 - switch <工作流名> - 切换工作流 - width <数值> - 设置图片宽度 (最大1344) - height <数值> - 设置图片高度 (最大1344) - size <宽> <高> - 同时设置宽高 - rewrite on|off - 改写模式开关 - override on|off - 覆写模式开关 - status - 查看当前状态和配置 - -示例: - ai_image generate 一个可爱的猫耳少女，樱花背景 - ai_image tag 1girl, cat_ears, cherry_blossoms, smile - ai_image both 1girl, cat_ears <<<>>> 一个可爱的猫耳少女在樱花树下微笑 - ai_image workflows - ai_image switch WAI - 37.json - ai_image size 1024 1024 - ai_image rewrite on -""" - self.cli = None - self.save_dir = self.DEFAULT_SAVE_DIR - - # 配置状态 - self.current_workflow = None - self.direct_prompt = "" - self.nl_prompt = "" - self.rewrite = False - self.override = False - self.width = None - self.height = None - self.workflows_cache = None - self.is_generating = False - - def set_cli(self, cli): - """设置 CLI 实例引用""" - self.cli = cli - # 注册插件命令 - if hasattr(cli, 'register_liugin_command'): - cli.register_liugin_command('ai_image', self.command_handler) - - def get_tool_info(self): - return { - "name": "ai_image", - "description": "AI 生图工具 — 基于 ai.2x.nz 接口，支持自然语言描述和 Danbooru Tag 生成图片。当用户想要生成图片、画图、AI绘画、生图时使用此工具。", - "keywords": [ - "生图", "画图", "图片", "AI绘画", "生成图片", "image", "generate", - "draw", "绘画", "插画", "二次元", "动漫", "Danbooru", "tag", - "工作流", "workflow", "ComfyUI" - ], - "usage": self.usage - } - - def get_mcp_definition(self): - return { - "name": "ai_image", - "description": "AI 生图工具，基于 ai.2x.nz 接口生成图片", - "inputSchema": { - "type": "object", - "properties": { - "operation": { - "type": "string", - "enum": ["generate", "tag", "both", "workflows", "switch", "size", "status"], - "description": "操作类型" - }, - "prompt": { - "type": "string", - "description": "生成提示词" - } - }, - "required": ["operation"] - } - } - - def convert_mcp_args(self, arguments): - op = arguments.get("operation", "") - prompt = arguments.get("prompt", "") - return f"{op} {prompt}" - - def command_handler(self, args): - """处理 /ai_image 命令""" - return self.handle(args) - - def handle(self, args): - """处理 AI 生图请求""" - if not REQUESTS_AVAILABLE: - return "❌ 缺少 requests 库，请运行: pip install requests" - - args = args.strip() - if not args: - return self.usage - - # 解析操作 - parts = args.split(maxsplit=1) - operation = parts[0].lower() - remaining = parts[1] if len(parts) > 1 else "" - - try: - if operation in ["generate", "gen", "g"]: - return self._cmd_generate(remaining) - elif operation in ["tag", "t", "direct", "d"]: - return self._cmd_tag(remaining) - elif operation in ["both", "b"]: - return self._cmd_both(remaining) - elif operation in ["workflows", "wf", "list"]: - return self._cmd_workflows() - elif operation in ["switch", "sw"]: - return self._cmd_switch(remaining) - elif operation in ["width", "w"]: - return self._cmd_set_width(remaining) - elif operation in ["height", "h"]: - return self._cmd_set_height(remaining) - elif operation in ["size", "s"]: - return self._cmd_set_size(remaining) - elif operation in ["rewrite", "rw"]: - return self._cmd_set_rewrite(remaining) - elif operation in ["override", "ov"]: - return self._cmd_set_override(remaining) - elif operation in ["status", "st"]: - return self._cmd_status() - elif operation in ["help", "?"]: - return self.usage - else: - # 尝试当作自然语言描述处理 - return self._cmd_generate(args) - except Exception as e: - return f"❌ 操作失败: {str(e)}" - - # ── 命令实现 ── - - def _cmd_generate(self, prompt): - """自然语言生成图片""" - if not prompt.strip(): - return "❌ 请提供描述内容。示例: ai_image generate 一个可爱的猫耳少女" - - self.nl_prompt = prompt.strip() - self.direct_prompt = "" - self.override = False - return self._run_generation() - - def _cmd_tag(self, tags): - """使用直接 Tag 生成图片""" - if not tags.strip(): - return "❌ 请提供 Tag 内容。示例: ai_image tag 1girl, cat_ears, smile" - - self.direct_prompt = tags.strip() - self.nl_prompt = "" - self.override = True - return self._run_generation() - - def _cmd_both(self, args): - """同时使用 Tag 和自然语言""" - if "<<<>>>" in args: - parts = args.split("<<<>>>", 1) - self.direct_prompt = parts[0].strip() - self.nl_prompt = parts[1].strip() - else: - # 尝试用 || 分隔 - if "||" in args: - parts = args.split("||", 1) - self.direct_prompt = parts[0].strip() - self.nl_prompt = parts[1].strip() - else: - return "❌ 格式错误。用法: ai_image both <<<>>> <描述>" - - self.override = False - return self._run_generation() - - def _cmd_workflows(self): - """列出所有工作流""" - workflows = self._fetch_workflows() - if not workflows: - return "❌ 获取工作流列表失败" - - result = f"📂 共 {len(workflows)} 个工作流:\n\n" - for i, wf in enumerate(workflows, 1): - name = wf.get("path", "未知") - current = " ← [当前]" if name == self.current_workflow else "" - result += f" {i:2d}. {name}{current}\n" - - result += f"\n💡 使用 'ai_image switch <工作流名>' 切换工作流" - return result - - def _cmd_switch(self, name): - """切换工作流""" - if not name.strip(): - return "❌ 请提供工作流名称。先用 'ai_image workflows' 查看列表" - - # 模糊匹配 - workflows = self._fetch_workflows() - if not workflows: - return "❌ 获取工作流列表失败" - - matched = None - name_lower = name.strip().lower() - for wf in workflows: - wf_path = wf.get("path", "") - if name_lower == wf_path.lower(): - matched = wf_path - break - if name_lower in wf_path.lower(): - matched = wf_path - - if not matched: - # 尝试数字索引 - try: - idx = int(name.strip()) - 1 - if 0 <= idx < len(workflows): - matched = workflows[idx].get("path", "") - except ValueError: - pass - - if not matched: - return f"❌ 未找到匹配的工作流: {name}\n💡 先用 'ai_image workflows' 查看列表" - - self.current_workflow = matched - return f"✅ 已切换工作流: {matched}" - - def _cmd_set_width(self, val): - """设置宽度""" - try: - w = int(val.strip()) - if w < 64 or w > 1344: - return "❌ 宽度范围: 64 ~ 1344" - self.width = w - return f"✅ 宽度已设置为: {w}" - except ValueError: - return "❌ 请输入有效数字" - - def _cmd_set_height(self, val): - """设置高度""" - try: - h = int(val.strip()) - if h < 64 or h > 1344: - return "❌ 高度范围: 64 ~ 1344" - self.height = h - return f"✅ 高度已设置为: {h}" - except ValueError: - return "❌ 请输入有效数字" - - def _cmd_set_size(self, val): - """同时设置宽高""" - parts = val.strip().split() - if len(parts) < 2: - return "❌ 用法: ai_image size <宽> <高>" - try: - w, h = int(parts[0]), int(parts[1]) - if w < 64 or w > 1344 or h < 64 or h > 1344: - return "❌ 宽高范围: 64 ~ 1344" - self.width = w - self.height = h - return f"✅ 尺寸已设置为: {w} x {h}" - except ValueError: - return "❌ 请输入有效数字" - - def _cmd_set_rewrite(self, val): - """设置改写模式""" - val = val.strip().lower() - if val in ["on", "开", "1", "true", "yes"]: - self.rewrite = True - return "✅ 改写模式已开启 — LLM 将智能重写 prompt" - elif val in ["off", "关", "0", "false", "no"]: - self.rewrite = False - return "✅ 改写模式已关闭 — 普通翻译/追加模式" - else: - self.rewrite = not self.rewrite - status = "开启" if self.rewrite else "关闭" - return f"✅ 改写模式已{status}" - - def _cmd_set_override(self, val): - """设置覆写模式""" - val = val.strip().lower() - if val in ["on", "开", "1", "true", "yes"]: - self.override = True - return "✅ 覆写模式已开启 — 忽略工作流内置 prompt" - elif val in ["off", "关", "0", "false", "no"]: - self.override = False - return "✅ 覆写模式已关闭 — 保留工作流内置 prompt" - else: - self.override = not self.override - status = "开启" if self.override else "关闭" - return f"✅ 覆写模式已{status}" - - def _cmd_status(self): - """查看当前状态""" - result = " AI 生图状态:\n\n" - result += f" 工作流: {self.current_workflow or '(默认)'}\n" - result += f" 直接 Tag: {self.direct_prompt or '(未设置)'}\n" - result += f" 自然语言: {self.nl_prompt or '(未设置)'}\n" - result += f" 改写模式: {'✅ 开启' if self.rewrite else '❌ 关闭'}\n" - result += f" 覆写模式: {'✅ 开启' if self.override else '❌ 关闭'}\n" - result += f" 宽度: {self.width or '(默认)'}\n" - result += f" 高度: {self.height or '(默认)'}\n" - result += f" 保存目录: {self.save_dir}\n" - result += f" 生成中: {'是' if self.is_generating else '否'}\n" - return result - - # ── 核心生图逻辑 ── - - def _fetch_workflows(self): - """获取工作流列表""" - if self.workflows_cache: - return self.workflows_cache - - try: - resp = requests.get(f"{self.API_URL}/workflows", timeout=15) - if resp.status_code == 200: - data = resp.json() - self.workflows_cache = data.get("workflows", []) - return self.workflows_cache - except Exception as e: - pass - return [] - - def _run_generation(self): - """执行图片生成""" - if self.is_generating: - return "⚠️ 已有生成任务在进行中，请等待完成" - - if not self.direct_prompt and not self.nl_prompt: - return "❌ 请提供至少一种 prompt（直接 Tag 或自然语言描述）" - - self.is_generating = True - result_lines = [] - image_paths = [] - - try: - # 获取默认工作流（如果未指定则使用列表中第一个） - workflow = self.current_workflow - if not workflow: - workflows = self._fetch_workflows() - if workflows: - # 优先选 "无Lora" 工作流，否则用第一个 - for wf in workflows: - if "无lora" in wf.get("path", "").lower(): - workflow = wf["path"] - break - if not workflow: - workflow = workflows[0].get("path", "") - - # 构建 payload - payload = { - "workflow_path": workflow or "", - "inline_workflow": None, - "direct_prompt": self.direct_prompt, - "nl_prompt": self.nl_prompt, - "rewrite": self.rewrite, - "override": self.override, - "width": self.width, - "height": self.height, - } - - result_lines.append(" 正在连接 ai.2x.nz ...") - result_lines.append(f" 直接 Tag: {self.direct_prompt[:80] or '(无)'}") - result_lines.append(f" 自然语言: {self.nl_prompt[:80] or '(无)'}") - result_lines.append(f" 工作流: {workflow or '(默认)'}") - result_lines.append("") - - # 使用同步方式通过 REST + 轮询获取结果 - # 由于 websocket-client 可能未安装，提供降级方案 - if WEBSOCKET_AVAILABLE: - images = self._generate_via_ws(payload, result_lines) - else: - # 降级：尝试使用 requests + SSE 或提示安装 - result_lines.append("⚠️ websocket-client 未安装，尝试安装...") - images = self._generate_via_ws_fallback(payload, result_lines) - - if images: - image_paths = images - result_lines.append(f"\n✅ 生成完成！共 {len(images)} 张图片") - for i, path in enumerate(images, 1): - result_lines.append(f" {i}. {path}") - else: - result_lines.append("\n❌ 未生成任何图片") - - except Exception as e: - result_lines.append(f"\n❌ 生成失败: {str(e)}") - finally: - self.is_generating = False - - return "\n".join(result_lines) - - def _generate_via_ws(self, payload, result_lines): - """通过 WebSocket 生成图片""" - images = [] - final_prompt = "" - error_msg = None - - ws_result = {"images": [], "error": None, "final_prompt": "", "done": False} - - def on_message(ws, message): - try: - m = json.loads(message) - msg_type = m.get("type", "") - - if msg_type == "log": - result_lines.append(f" {m.get('message', '')}") - elif msg_type == "llm_start": - result_lines.append(" LLM 正在处理 prompt...") - elif msg_type == "llm_chunk": - pass # 流式输出，不逐块显示 - elif msg_type == "llm_done": - result_lines.append(" LLM 处理完成") - elif msg_type == "progress": - node = m.get("node", "") - value = m.get("value", 0) - max_val = m.get("max", 0) - if max_val and max_val > 1: - pct = int(value * 100 / max_val) - result_lines.append(f"⏳ 进度: {node} {value}/{max_val} ({pct}%)") - else: - result_lines.append(f"⏳ 执行: {node}") - elif msg_type == "prompt_id": - fp = m.get("final_prompt", "") - if fp: - ws_result["final_prompt"] = fp - result_lines.append(f"\n 最终 Prompt:\n{fp[:300]}{'...' if len(fp) > 300 else ''}\n") - elif msg_type == "image": - url = m.get("url", "") - filename = m.get("filename", "") - if url: - # 下载图片 - saved = self._download_image(url, filename) - if saved: - ws_result["images"].append(saved) - result_lines.append(f" 收到图片: {filename}") - elif msg_type == "done": - count = m.get("count", 0) - result_lines.append(f"✅ 服务端完成，共 {count} 张") - ws_result["done"] = True - ws.close() - elif msg_type == "error": - ws_result["error"] = m.get("message", "未知错误") - result_lines.append(f"❌ 错误: {ws_result['error']}") - ws.close() - except Exception as e: - result_lines.append(f"⚠️ 解析消息出错: {e}") - - def on_error(ws, error): - ws_result["error"] = str(error) - result_lines.append(f"❌ WebSocket 错误: {error}") - - def on_close(ws, close_status_code, close_msg): - ws_result["done"] = True - - def on_open(ws): - ws.send(json.dumps(payload)) - - try: - ws = websocket.WebSocketApp( - self.WS_URL, - on_message=on_message, - on_error=on_error, - on_close=on_close, - on_open=on_open - ) - - # 在独立线程中运行 WebSocket - ws_thread = threading.Thread(target=ws.run_forever, kwargs={"ping_timeout": 10}) - ws_thread.daemon = True - ws_thread.start() - - # 等待完成（最多 5 分钟） - timeout = 300 - start = time.time() - while not ws_result["done"] and (time.time() - start) < timeout: - time.sleep(1) - - if ws_result["error"]: - raise Exception(ws_result["error"]) - - if not ws_result["done"]: - ws.close() - raise Exception("生成超时（超过 5 分钟）") - - return ws_result["images"] - - except Exception as e: - raise Exception(f"WebSocket 连接失败: {e}") - - def _generate_via_ws_fallback(self, payload, result_lines): - """降级方案：尝试自动安装 websocket-client""" - try: - import subprocess - result_lines.append(" 正在安装 websocket-client ...") - subprocess.run( - ["pip", "install", "websocket-client"], - capture_output=True, timeout=60 - ) - import importlib - global websocket - websocket = importlib.import_module("websocket") - global WEBSOCKET_AVAILABLE - WEBSOCKET_AVAILABLE = True - result_lines.append("✅ websocket-client 安装成功") - return self._generate_via_ws(payload, result_lines) - except Exception as e: - result_lines.append(f"❌ 无法安装 websocket-client: {e}") - result_lines.append("💡 请手动运行: pip install websocket-client") - return [] - - def _download_image(self, url, filename): - """下载图片到本地""" - try: - # 确保 URL 是完整的 - if url.startswith("/"): - url = f"{self.BASE_URL}{url}" - elif not url.startswith("http"): - url = f"{self.BASE_URL}/api/output/file?path={quote(url)}" - - # 创建保存目录 - os.makedirs(self.save_dir, exist_ok=True) - - # 生成文件名 - if not filename: - filename = f"img_{datetime.now().strftime('%Y%m%d_%H%M%S')}.png" - - # 清理文件名 - filename = re.sub(r'[<>:"/\\|?*]', '_', filename) - save_path = os.path.join(self.save_dir, filename) - - # 如果文件已存在，添加序号 - if os.path.exists(save_path): - name, ext = os.path.splitext(filename) - for i in range(1, 100): - new_path = os.path.join(self.save_dir, f"{name}_{i}{ext}") - if not os.path.exists(new_path): - save_path = new_path - break - - # 下载 - resp = requests.get(url, timeout=60, stream=True) - if resp.status_code == 200: - with open(save_path, 'wb') as f: - for chunk in resp.iter_content(chunk_size=8192): - f.write(chunk) - return save_path - else: - return None - - except Exception as e: - return None - - -# 兼容 Liugin 类名 diff --git a/plugins/audio_player.py b/plugins/audio_player.py deleted file mode 100644 index 13fbebe..0000000 --- a/plugins/audio_player.py +++ /dev/null @@ -1,169 +0,0 @@ -import os -import threading -from colorama import Fore, Style - - -class Liugin: - """音频播放插件""" - - def __init__(self): - self.usage = """音频播放工具使用方法： -audio_player play - 播放音频文件 -例如: -- audio_player play - 播放Ciallo～(∠・ω- )⌒-.mp3音频文件 - -使用工具的JSON格式示例: -{"action": "use_tool", "tool": "audio_player", "args": "play"} - 播放音频文件 - -直接命令: -- /ciallo - 播放音频文件 -""" - self.cli = None # CLI实例引用 - self.audio_file = None # 音频文件路径将在set_cli中设置 - - def set_cli(self, cli): - """设置CLI实例引用""" - self.cli = cli - # 获取主程序目录路径，通过CLI实例 - import os - # 使用CLI实例中的项目目录信息 - self.audio_file = os.path.join(os.path.dirname(cli.chat_history_dir), "Ciallo～(∠・ω- )⌒-.mp3") - # 注册插件命令 - self.cli.register_liugin_command('ciallo', self.play_command_handler) - - def get_tool_info(self): - return { - "name": "audio_player", - "description": "音频播放工具，播放指定的音频文件", - "keywords": ["音频", "播放", "audio", "play", "音乐", "声音"], - "usage": """audio_player 工具使用说明： - JSON格式示例： - {"action": "use_tool", "tool": "audio_player", "args": "play"} - 播放音频文件 - 直接命令：/ciallo - 播放音频文件""" - } - - def get_mcp_definition(self): - return { - "name": "audio_player", - "description": "音频播放工具", - "inputSchema": { - "type": "object", - "properties": { - "file": { - "type": "string", - "description": "音频文件路径（可选）" - } - } - } - } - - def convert_mcp_args(self, arguments): - return arguments.get("file", "") - - def handle(self, args): - """处理音频播放请求""" - try: - # 解析参数 - parts = args.strip().split() - if len(parts) < 1: - return "错误：请提供操作类型。支持的操作：play" - - operation = parts[0].lower() - - if operation == "play": - return self._play_audio() - else: - return f"错误：不支持的操作 '{operation}'。支持的操作：play" - - except Exception as e: - return f"音频播放错误: {str(e)}" - - def play_command_handler(self, args): - """处理 /ciallo 命令""" - # 在新线程中播放音频，避免阻塞主程序 - thread = threading.Thread(target=self._play_audio_internal) - thread.daemon = True - thread.start() - return f"正在播放音频文件: {self.audio_file}" - - def _play_audio(self): - """播放音频文件""" - try: - # 检查音频文件是否存在 - if not os.path.exists(self.audio_file): - return f"错误：音频文件 '{self.audio_file}' 不存在。" - - # 使用多种方式尝试播放音频 - success = self._try_play_audio() - - if success: - return f"音频文件 '{self.audio_file}' 播放成功！" - else: - return f"音频播放失败。请确保已安装相应的音频播放库。" - - except Exception as e: - return f"播放音频时发生错误: {str(e)}" - - def _play_audio_internal(self): - """在新线程中播放音频""" - try: - # 检查音频文件是否存在 - if not os.path.exists(self.audio_file): - print(f"{Fore.RED}错误：音频文件 '{self.audio_file}' 不存在。{Style.RESET_ALL}") - return - - # 使用多种方式尝试播放音频 - self._try_play_audio() - - except Exception as e: - print(f"{Fore.RED}播放音频时发生错误: {str(e)}{Style.RESET_ALL}") - - def _try_play_audio(self): - """尝试多种方式播放音频""" - # 首先尝试使用pygame（如果已安装） - try: - import pygame - pygame.mixer.init() - pygame.mixer.music.load(self.audio_file) - pygame.mixer.music.play() - - # 等待音频播放完成 - while pygame.mixer.music.get_busy(): - pygame.time.Clock().tick(10) - - pygame.mixer.quit() - return True - except ImportError: - pass # pygame未安装，尝试其他方式 - except Exception: - pass # 其他pygame相关错误，尝试其他方式 - - # 尝试使用playsound库（如果已安装） - try: - from playsound import playsound - playsound(self.audio_file) - return True - except ImportError: - pass # playsound未安装，尝试其他方式 - except Exception: - pass # 其他playsound相关错误，尝试其他方式 - - # 尝试使用winsound（仅Windows） - try: - import winsound - # 对于MP3文件，先尝试使用高级API - import subprocess - subprocess.run(["start", "", self.audio_file], shell=True) - return True - except Exception: - pass # winsound相关错误，尝试其他方式 - - # 尝试使用os.startfile（仅Windows） - try: - os.startfile(self.audio_file) # 这会使用默认程序打开文件 - return True - except Exception: - pass # os.startfile相关错误 - - # 如果所有方式都失败，返回False - return False \ No newline at end of file diff --git a/plugins/bookmark_manager.py b/plugins/bookmark_manager.py deleted file mode 100644 index db29c21..0000000 --- a/plugins/bookmark_manager.py +++ /dev/null @@ -1,489 +0,0 @@ -""" -书签管理器插件 - 管理开发相关链接和资源 -支持分类、标签、搜索、导入导出，JSON 持久化 -""" - -import json -import os -import hashlib -import time -from datetime import datetime -from typing import List, Optional - - -class Liugin: - """书签管理器 - 开发资源和链接管理""" - - def __init__(self): - self.usage = """书签管理器 - -操作: - add -n <名称> [-c 分类] [-t 标签1,标签2] [-d 描述] - - 添加书签 - list [-c 分类] [-t 标签] - 列出书签 - open <名称或ID> - 获取书签 URL - search <关键词> - 搜索书签 - delete <名称或ID> - 删除书签 - edit <名称或ID> [-n 名称] [-d 描述] [-c 分类] [-t 标签] - - 编辑书签 - categories - 列出所有分类 - tags - 列出所有标签 - export [文件] - 导出书签 - import <文件> - 导入书签 - stats - 统计信息 - recent [-n 数量] - 最近添加的书签 - check - 检查书签是否可访问 - favicon - 获取网站 favicon URL - -示例: - bookmark_manager add https://github.com -n GitHub -c 开发 -t 代码,托管 -d 代码托管平台 - bookmark_manager list -c 开发 - bookmark_manager search github - bookmark_manager open GitHub - bookmark_manager categories -""" - self.cli = None - self.bookmarks_file = "bookmarks.json" - self.bookmarks = [] - self._load_bookmarks() - - def set_cli(self, cli): - self.cli = cli - - def get_tool_info(self): - return { - "name": "bookmark_manager", - "description": "书签管理器 - 管理开发相关链接和资源，支持分类、标签、搜索", - "keywords": ["书签", "链接", "收藏", "bookmark", "网址", "资源", "网站", "URL"], - "usage": self.usage - } - - def get_mcp_definition(self): - return { - "name": "bookmark_manager", - "description": "书签管理器，支持添加、搜索、获取、删除书签", - "inputSchema": { - "type": "object", - "properties": { - "operation": { - "type": "string", - "enum": ["add", "list", "open", "search", "delete", - "edit", "categories", "tags", "export", - "import", "stats", "recent", "check"], - "description": "操作类型" - }, - "url": {"type": "string", "description": "URL"}, - "name": {"type": "string", "description": "书签名称"}, - "category": {"type": "string", "description": "分类"}, - "tags": {"type": "string", "description": "标签（逗号分隔）"}, - "description": {"type": "string", "description": "描述"}, - "keyword": {"type": "string", "description": "搜索关键词"}, - "file": {"type": "string", "description": "文件路径"}, - "count": {"type": "integer", "description": "数量"} - }, - "required": ["operation"] - } - } - - def convert_mcp_args(self, arguments): - op = arguments.get("operation", "") - name = arguments.get("name", "") - return f"{op} {name}".strip() - - def _load_bookmarks(self): - try: - if os.path.exists(self.bookmarks_file): - with open(self.bookmarks_file, 'r', encoding='utf-8') as f: - self.bookmarks = json.load(f) - except Exception: - self.bookmarks = [] - - def _save_bookmarks(self): - try: - with open(self.bookmarks_file, 'w', encoding='utf-8') as f: - json.dump(self.bookmarks, f, ensure_ascii=False, indent=2) - return None - except Exception as e: - return f"保存失败: {e}" - - def _gen_id(self, url: str) -> str: - return hashlib.md5(f"{url}{time.time()}".encode()).hexdigest()[:8] - - def handle(self, args: str) -> str: - try: - parts = self._parse_args(args) - if not parts: - return self._op_list([]) - - operation = parts[0].lower() - handlers = { - "add": self._op_add, - "list": self._op_list, - "open": self._op_open, - "search": self._op_search, - "delete": self._op_delete, - "edit": self._op_edit, - "categories": self._op_categories, - "tags": self._op_tags, - "export": self._op_export, - "import": self._op_import, - "stats": self._op_stats, - "recent": self._op_recent, - "check": self._op_check, - "favicon": self._op_favicon, - } - - handler = handlers.get(operation) - if not handler: - return f"错误：不支持的操作 '{operation}'" - return handler(parts[1:]) - - except Exception as e: - return f"书签管理错误: {str(e)}" - - def _parse_args(self, args: str) -> List[str]: - parts, current, in_quotes, qc = [], "", False, None - for ch in args: - if ch in ('"', "'") and not in_quotes: - in_quotes, qc = True, ch - elif ch == qc and in_quotes: - in_quotes, qc = False, None - elif ch == ' ' and not in_quotes: - if current: - parts.append(current) - current = "" - else: - current += ch - if current: - parts.append(current) - return parts - - def _op_add(self, args: List[str]) -> str: - if not args: - return "错误：请提供 URL" - url = args[0] - name, category, tags, desc = "", "", [], "" - - i = 1 - while i < len(args): - if args[i] == "-n" and i + 1 < len(args): - name = args[i + 1]; i += 1 - elif args[i] == "-c" and i + 1 < len(args): - category = args[i + 1]; i += 1 - elif args[i] == "-t" and i + 1 < len(args): - tags = [t.strip() for t in args[i + 1].split(",") if t.strip()]; i += 1 - elif args[i] == "-d" and i + 1 < len(args): - desc = args[i + 1]; i += 1 - i += 1 - - if not name: - # 从 URL 推断名称 - from urllib.parse import urlparse - parsed = urlparse(url if '://' in url else 'https://' + url) - name = parsed.netloc.replace('www.', '').split('.')[0].capitalize() - url = parsed.geturl() - - # 检查重复 - for b in self.bookmarks: - if b["url"] == url: - return f"错误：URL 已存在 (书签: {b['name']})" - - bookmark = { - "id": self._gen_id(url), - "name": name, - "url": url, - "category": category, - "tags": tags, - "description": desc, - "created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), - "access_count": 0 - } - self.bookmarks.append(bookmark) - err = self._save_bookmarks() - if err: - return err - return f"✅ 已添加书签 '{name}'\n URL: {url}\n 分类: {category or '未分类'}" - - def _op_list(self, args: List[str]) -> str: - cat_filter, tag_filter = None, None - i = 0 - while i < len(args): - if args[i] == "-c" and i + 1 < len(args): - cat_filter = args[i + 1].lower(); i += 1 - elif args[i] == "-t" and i + 1 < len(args): - tag_filter = args[i + 1].lower(); i += 1 - i += 1 - - filtered = self.bookmarks - if cat_filter: - filtered = [b for b in filtered if b.get("category", "").lower() == cat_filter] - if tag_filter: - filtered = [b for b in filtered if any(tag_filter in t.lower() for t in b.get("tags", []))] - - if not filtered: - return "📭 没有找到书签" - - # 按分类分组 - grouped = {} - for b in filtered: - cat = b.get("category", "未分类") - grouped.setdefault(cat, []).append(b) - - result = f"\n 书签列表 (共 {len(filtered)} 个)\n" - result += "═" * 55 + "\n" - - for cat, items in sorted(grouped.items()): - result += f"\n {cat} ({len(items)} 个)\n" - result += "─" * 45 + "\n" - for b in items: - tags = ", ".join(b.get("tags", [])) if b.get("tags") else "" - result += f" [{b['id']}] {b['name']}\n" - result += f" {b['url']}\n" - if tags: - result += f" ️ {tags}\n" - if b.get("description"): - result += f" {b['description'][:50]}\n" - result += "\n" - - return result.rstrip() - - def _op_open(self, args: List[str]) -> str: - if not args: - return "错误：请提供书签名称或 ID" - key = args[0] - bookmark = self._find_bookmark(key) - if not bookmark: - return f"❌ 未找到书签: {key}" - - bookmark["access_count"] = bookmark.get("access_count", 0) + 1 - self._save_bookmarks() - - return (f" {bookmark['name']}\n" - f" URL: {bookmark['url']}\n" - f" 访问次数: {bookmark.get('access_count', 0)}") - - def _op_search(self, args: List[str]) -> str: - if not args: - return "错误：请提供搜索关键词" - keyword = " ".join(args).lower() - - results = [] - for b in self.bookmarks: - searchable = f"{b['name']} {b['url']} {b.get('description', '')} {' '.join(b.get('tags', []))} {b.get('category', '')}".lower() - if keyword in searchable: - results.append(b) - - if not results: - return f" 未找到匹配 '{keyword}' 的书签" - - output = f" 搜索结果 ({len(results)} 个):\n" - output += "─" * 50 + "\n" - for b in results: - output += f" [{b['id']}] {b['name']} - {b['url']}\n" - if b.get("description"): - output += f" {b['description'][:40]}\n" - return output.rstrip() - - def _op_delete(self, args: List[str]) -> str: - if not args: - return "错误：请提供书签名称或 ID" - key = args[0] - for i, b in enumerate(self.bookmarks): - if b["name"].lower() == key.lower() or b["id"] == key: - removed = self.bookmarks.pop(i) - self._save_bookmarks() - return f" ️ 已删除书签: {removed['name']} ({removed['url']})" - return f"❌ 未找到书签: {key}" - - def _op_edit(self, args: List[str]) -> str: - if not args: - return "错误：请提供书签名称或 ID" - key = args[0] - bookmark = self._find_bookmark(key) - if not bookmark: - return f"❌ 未找到书签: {key}" - - i = 1 - while i < len(args): - if args[i] == "-n" and i + 1 < len(args): - bookmark["name"] = args[i + 1]; i += 1 - elif args[i] == "-d" and i + 1 < len(args): - bookmark["description"] = args[i + 1]; i += 1 - elif args[i] == "-c" and i + 1 < len(args): - bookmark["category"] = args[i + 1]; i += 1 - elif args[i] == "-t" and i + 1 < len(args): - bookmark["tags"] = [t.strip() for t in args[i + 1].split(",") if t.strip()]; i += 1 - i += 1 - - bookmark["updated_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") - self._save_bookmarks() - return f"✅ 已更新书签: {bookmark['name']}" - - def _op_categories(self, args: List[str]) -> str: - cat_count = {} - for b in self.bookmarks: - cat = b.get("category", "未分类") - cat_count[cat] = cat_count.get(cat, 0) + 1 - - if not cat_count: - return " 暂无分类" - - result = " 分类列表:\n" - for cat, count in sorted(cat_count.items(), key=lambda x: -x[1]): - bar = "█" * min(count, 20) - result += f" {cat:15s} {bar} ({count})\n" - return result.rstrip() - - def _op_tags(self, args: List[str]) -> str: - tag_count = {} - for b in self.bookmarks: - for t in b.get("tags", []): - tag_count[t] = tag_count.get(t, 0) + 1 - - if not tag_count: - return " ️ 暂无标签" - - result = " ️ 标签列表:\n" - for tag, count in sorted(tag_count.items(), key=lambda x: -x[1]): - result += f" {tag} ({count})\n" - return result.rstrip() - - def _op_export(self, args: List[str]) -> str: - output_file = args[0] if args else "bookmarks_export.json" - try: - with open(output_file, 'w', encoding='utf-8') as f: - json.dump(self.bookmarks, f, ensure_ascii=False, indent=2) - return f"✅ 已导出 {len(self.bookmarks)} 个书签到: {output_file}" - except Exception as e: - return f"❌ 导出失败: {e}" - - def _op_import(self, args: List[str]) -> str: - if not args: - return "错误：请提供导入文件路径" - file_path = args[0] - if not os.path.isfile(file_path): - return f"❌ 文件不存在: {file_path}" - try: - with open(file_path, 'r', encoding='utf-8') as f: - imported = json.load(f) - if not isinstance(imported, list): - return "❌ 文件格式错误：需要 JSON 数组" - - existing_urls = {b["url"] for b in self.bookmarks} - added = 0 - for b in imported: - if isinstance(b, dict) and b.get("url") and b["url"] not in existing_urls: - b["id"] = self._gen_id(b["url"]) - self.bookmarks.append(b) - existing_urls.add(b["url"]) - added += 1 - - self._save_bookmarks() - return f"✅ 已导入 {added} 个书签 (跳过 {len(imported) - added} 个重复)" - except Exception as e: - return f"❌ 导入失败: {e}" - - def _op_stats(self, args: List[str]) -> str: - total = len(self.bookmarks) - if not total: - return "📊 暂无书签" - - cat_count = {} - total_access = 0 - for b in self.bookmarks: - cat = b.get("category", "未分类") - cat_count[cat] = cat_count.get(cat, 0) + 1 - total_access += b.get("access_count", 0) - - most_accessed = max(self.bookmarks, key=lambda b: b.get("access_count", 0)) - - result = f"📊 书签统计\n" - result += "─" * 35 + "\n" - result += f" 总数: {total}\n" - result += f" 总访问: {total_access} 次\n" - result += f" 分类数: {len(cat_count)}\n" - result += f" 最常访问: {most_accessed['name']} ({most_accessed.get('access_count', 0)} 次)\n" - result += f"\n 分类分布:\n" - for cat, count in sorted(cat_count.items(), key=lambda x: -x[1]): - result += f" {cat}: {count}\n" - return result.rstrip() - - def _op_recent(self, args: List[str]) -> str: - count = 5 - for i, a in enumerate(args): - if a == "-n" and i + 1 < len(args): - count = int(args[i + 1]) - - recent = sorted(self.bookmarks, key=lambda b: b.get("created_at", ""), reverse=True)[:count] - if not recent: - return "📭 暂无书签" - - result = f"⏰ 最近 {len(recent)} 个书签:\n" - for b in recent: - result += f" [{b['id']}] {b['name']} - {b.get('created_at', '')}\n" - return result.rstrip() - - def _op_check(self, args: List[str]) -> str: - import socket - from urllib.parse import urlparse - - bookmarks = self.bookmarks - if args: - # 检查指定书签 - bookmark = self._find_bookmark(args[0]) - if not bookmark: - return f"❌ 未找到书签: {args[0]}" - bookmarks = [bookmark] - - result = f" ️ 书签可访问性检查 ({len(bookmarks)} 个):\n" - result += "─" * 50 + "\n" - - for b in bookmarks[:20]: - url = b["url"] - try: - parsed = urlparse(url) - host = parsed.netloc or parsed.path.split('/')[0] - socket.create_connection((host, 80), timeout=3) - result += f" {b['name']} ({host})\n" - except (socket.timeout, socket.error, OSError): - result += f" {b['name']} ({host}) - 不可达\n" - - if len(bookmarks) > 20: - result += f" ... 仅检查前 20 个\n" - return result.rstrip() - - def _op_favicon(self, args: List[str]) -> str: - if not args: - return "错误：请提供 URL" - url = args[0] - if '://' not in url: - url = 'https://' + url - - from urllib.parse import urlparse - parsed = urlparse(url) - base = f"{parsed.scheme}://{parsed.netloc}" - favicon_url = f"{base}/favicon.ico" - - return (f" Favicon:\n" - f" 网站: {base}\n" - f" Favicon: {favicon_url}\n" - f" Google: https://www.google.com/s2/favicons?domain={parsed.netloc}&sz=64") - - def _find_bookmark(self, key: str) -> Optional[dict]: - for b in self.bookmarks: - if b["name"].lower() == key.lower() or b["id"] == key: - return b - return None - - -def test_plugin(): - plugin = Liugin() - print("书签管理器测试:") - print(plugin.handle('add https://github.com -n GitHub -c 开发 -t 代码,托管 -d 代码托管平台')) - print(plugin.handle('add https://stackoverflow.com -n StackOverflow -c 学习 -t 问答,编程')) - print(plugin.handle('list')) - print(plugin.handle('search github')) - - -if __name__ == "__main__": - test_plugin() diff --git a/plugins/color_picker.py b/plugins/color_picker.py deleted file mode 100644 index 34fea2c..0000000 --- a/plugins/color_picker.py +++ /dev/null @@ -1,459 +0,0 @@ -""" -颜色工具插件 - 颜色格式转换、调色板、对比度检查 -支持 HEX/RGB/HSL/HSV/CMYK 格式互转，WCAG 对比度检查 -""" - -import re -import colorsys -from typing import List, Tuple - - -class Liugin: - """颜色工具 - 格式转换、调色板、对比度检查""" - - def __init__(self): - self.usage = """颜色工具 - -操作: - convert <颜色> - 自动检测并转换为所有格式 - hex - 转换为 HEX - rgb - 转换为 RGB - hsl - 转换为 HSL - hsv - 转换为 HSV - cmyk - 转换为 CMYK - contrast <颜色1> <颜色2> - WCAG 对比度检查 - palette <颜色> [-n 数量] - 生成调色板（明暗变体） - complementary <颜色> - 互补色 - analogous <颜色> - 类似色 - triadic <颜色> - 三色组 - random [-n 数量] - 随机颜色 - mix <颜色1> <颜色2> [-w 权重] - 混合两种颜色 - blind <颜色> <类型> - 色盲模拟 - named - 列出 CSS 命名颜色 - -颜色格式: - HEX: #FF6600, #F60, FF6600 - RGB: rgb(255,102,0) 或 255,102,0 - HSL: hsl(24,100%,50%) - HSV: hsv(24,100%,100%) - CMYK: cmyk(0,60,100,0) - -示例: - color_picker convert "#3498db" - color_picker rgb "hsl(207,62%,54%)" - color_picker contrast "#FFFFFF" "#000000" - color_picker palette "#e74c3c" -n 9 - color_picker complementary "#3498db" -""" - self.cli = None - - def set_cli(self, cli): - self.cli = cli - - def get_tool_info(self): - return { - "name": "color_picker", - "description": "颜色工具 - HEX/RGB/HSL/HSV/CMYK 转换、对比度检查、调色板生成", - "keywords": ["颜色", "color", "HEX", "RGB", "HSL", "调色板", "对比度", "色盲", "转换"], - "usage": self.usage - } - - def get_mcp_definition(self): - return { - "name": "color_picker", - "description": "颜色工具，支持格式转换、对比度检查、调色板生成", - "inputSchema": { - "type": "object", - "properties": { - "operation": { - "type": "string", - "enum": ["convert", "hex", "rgb", "hsl", "hsv", "cmyk", - "contrast", "palette", "complementary", "analogous", - "triadic", "random", "mix", "blind"], - "description": "操作类型" - }, - "color": {"type": "string", "description": "颜色值"}, - "color2": {"type": "string", "description": "第二个颜色值"}, - "count": {"type": "integer", "description": "数量"}, - "weight": {"type": "number", "description": "混合权重 (0-1)"}, - "blind_type": {"type": "string", "description": "色盲类型"} - }, - "required": ["operation"] - } - } - - def convert_mcp_args(self, arguments): - op = arguments.get("operation", "") - color = arguments.get("color", "") - return f"{op} {color}".strip() - - def handle(self, args: str) -> str: - try: - parts = self._parse_args(args) - if not parts: - return self._op_random(["1"]) - - operation = parts[0].lower() - handlers = { - "convert": self._op_convert, - "hex": self._op_hex, - "rgb": self._op_rgb, - "hsl": self._op_hsl, - "hsv": self._op_hsv, - "cmyk": self._op_cmyk, - "contrast": self._op_contrast, - "palette": self._op_palette, - "complementary": self._op_complementary, - "analogous": self._op_analogous, - "triadic": self._op_triadic, - "random": self._op_random, - "mix": self._op_mix, - "blind": self._op_blind, - } - - handler = handlers.get(operation) - if not handler: - return f"错误：不支持的操作 '{operation}'" - return handler(parts[1:]) - - except Exception as e: - return f"颜色工具错误: {str(e)}" - - def _parse_args(self, args: str) -> List[str]: - parts, current, in_quotes, qc = [], "", False, None - for ch in args: - if ch in ('"', "'") and not in_quotes: - in_quotes, qc = True, ch - elif ch == qc and in_quotes: - in_quotes, qc = False, None - elif ch == ' ' and not in_quotes: - if current: - parts.append(current) - current = "" - else: - current += ch - if current: - parts.append(current) - return parts - - def _parse_opts(self, args: List[str]) -> dict: - opts = {} - i = 0 - while i < len(args): - if args[i].startswith("-") and i + 1 < len(args): - key = args[i].lstrip("-") - opts[key] = args[i + 1] - i += 1 - i += 1 - return opts - - def _parse_color(self, color_str: str) -> Tuple[int, int, int]: - """解析颜色字符串为 RGB (0-255)""" - s = color_str.strip().lower() - - # HEX - hex_match = re.match(r'^#?([0-9a-f]{3,8})$', s) - if hex_match: - h = hex_match.group(1) - if len(h) == 3: - h = h[0]*2 + h[1]*2 + h[2]*2 - elif len(h) == 8: - h = h[:6] # ignore alpha - if len(h) == 6: - return (int(h[0:2], 16), int(h[2:4], 16), int(h[4:6], 16)) - - # RGB - rgb_match = re.match(r'rgb\s*\(\s*(\d+)\s*,\s*(\d+)\s*,\s*(\d+)\s*\)', s) - if rgb_match: - return (int(rgb_match.group(1)), int(rgb_match.group(2)), int(rgb_match.group(3))) - - # plain r,g,b - plain_match = re.match(r'^(\d+)\s*,\s*(\d+)\s*,\s*(\d+)$', s) - if plain_match: - return (int(plain_match.group(1)), int(plain_match.group(2)), int(plain_match.group(3))) - - # HSL - hsl_match = re.match(r'hsl\s*\(\s*(\d+)\s*,\s*(\d+)%,\s*(\d+)%\s*\)', s) - if hsl_match: - h = int(hsl_match.group(1)) / 360 - sl = int(hsl_match.group(2)) / 100 - l = int(hsl_match.group(3)) / 100 - r, g, b = colorsys.hls_to_rgb(h, l, sl) - return (int(r * 255), int(g * 255), int(b * 255)) - - # HSV/HSB - hsv_match = re.match(r'hsv\s*\(\s*(\d+)\s*,\s*(\d+)%,\s*(\d+)%\s*\)', s) - if hsv_match: - h = int(hsv_match.group(1)) / 360 - sv = int(hsv_match.group(2)) / 100 - v = int(hsv_match.group(3)) / 100 - r, g, b = colorsys.hsv_to_rgb(h, sv, v) - return (int(r * 255), int(g * 255), int(b * 255)) - - # CMYK - cmyk_match = re.match(r'cmyk\s*\(\s*(\d+)%?\s*,\s*(\d+)%?\s*,\s*(\d+)%?\s*,\s*(\d+)%?\s*\)', s) - if cmyk_match: - c = int(cmyk_match.group(1)) / 100 - m = int(cmyk_match.group(2)) / 100 - y = int(cmyk_match.group(3)) / 100 - k = int(cmyk_match.group(4)) / 100 - r = int(255 * (1 - c) * (1 - k)) - g = int(255 * (1 - m) * (1 - k)) - b = int(255 * (1 - y) * (1 - k)) - return (r, g, b) - - raise ValueError(f"无法解析颜色: {color_str}") - - def _rgb_to_hex(self, r: int, g: int, b: int) -> str: - return f"#{r:02X}{g:02X}{b:02X}" - - def _rgb_to_hsl(self, r: int, g: int, b: int) -> Tuple[int, int, int]: - h, l, s = colorsys.rgb_to_hls(r / 255, g / 255, b / 255) - return (int(h * 360), int(s * 100), int(l * 100)) - - def _rgb_to_hsv(self, r: int, g: int, b: int) -> Tuple[int, int, int]: - h, s, v = colorsys.rgb_to_hsv(r / 255, g / 255, b / 255) - return (int(h * 360), int(s * 100), int(v * 100)) - - def _rgb_to_cmyk(self, r: int, g: int, b: int) -> Tuple[int, int, int, int]: - if r == g == b == 0: - return (0, 0, 0, 100) - c = 1 - r / 255 - m = 1 - g / 255 - y = 1 - b / 255 - k = min(c, m, y) - c = int((c - k) / (1 - k) * 100) - m = int((m - k) / (1 - k) * 100) - y = int((y - k) / (1 - k) * 100) - k = int(k * 100) - return (c, m, y, k) - - def _luminance(self, r: int, g: int, b: int) -> float: - """计算相对亮度 (WCAG 2.0)""" - def linearize(c): - c = c / 255 - return c / 12.92 if c <= 0.03928 else ((c + 0.055) / 1.055) ** 2.4 - return 0.2126 * linearize(r) + 0.7152 * linearize(g) + 0.0722 * linearize(b) - - def _format_color_block(self, r: int, g: int, b: int) -> str: - """生成颜色预览块""" - return f" ■ RGB({r}, {g}, {b})" - - def _op_convert(self, args: List[str]) -> str: - if not args: - return "错误：请提供颜色值" - r, g, b = self._parse_color(args[0]) - h, s, l = self._rgb_to_hsl(r, g, b) - hh, sv, vv = self._rgb_to_hsv(r, g, b) - c, m, y, k = self._rgb_to_cmyk(r, g, b) - - result = f" 颜色转换: {args[0]}\n" - result += self._format_color_block(r, g, b) + "\n" - result += "─" * 35 + "\n" - result += f" HEX: {self._rgb_to_hex(r, g, b)}\n" - result += f" RGB: rgb({r}, {g}, {b})\n" - result += f" HSL: hsl({h}, {s}%, {l}%)\n" - result += f" HSV: hsv({hh}, {sv}%, {vv}%)\n" - result += f" CMYK: cmyk({c}%, {m}%, {y}%, {k}%)\n" - return result.rstrip() - - def _op_hex(self, args: List[str]) -> str: - if not args: - return "错误：请提供颜色值" - r, g, b = self._parse_color(args[0]) - return f" HEX: {self._rgb_to_hex(r, g, b)}" - - def _op_rgb(self, args: List[str]) -> str: - if not args: - return "错误：请提供颜色值" - r, g, b = self._parse_color(args[0]) - return f" RGB: rgb({r}, {g}, {b})" - - def _op_hsl(self, args: List[str]) -> str: - if not args: - return "错误：请提供颜色值" - r, g, b = self._parse_color(args[0]) - h, s, l = self._rgb_to_hsl(r, g, b) - return f" HSL: hsl({h}, {s}%, {l}%)" - - def _op_hsv(self, args: List[str]) -> str: - if not args: - return "错误：请提供颜色值" - r, g, b = self._parse_color(args[0]) - h, s, v = self._rgb_to_hsv(r, g, b) - return f" HSV: hsv({h}, {s}%, {v}%)" - - def _op_cmyk(self, args: List[str]) -> str: - if not args: - return "错误：请提供颜色值" - r, g, b = self._parse_color(args[0]) - c, m, y, k = self._rgb_to_cmyk(r, g, b) - return f" CMYK: cmyk({c}%, {m}%, {y}%, {k}%)" - - def _op_contrast(self, args: List[str]) -> str: - if len(args) < 2: - return "错误：请提供两种颜色" - r1, g1, b1 = self._parse_color(args[0]) - r2, g2, b2 = self._parse_color(args[1]) - - l1 = self._luminance(r1, g1, b1) - l2 = self._luminance(r2, g2, b2) - lighter = max(l1, l2) - darker = min(l1, l2) - ratio = (lighter + 0.05) / (darker + 0.05) - - aa_large = ratio >= 3 - aa_normal = ratio >= 4.5 - aaa_large = ratio >= 4.5 - aaa_normal = ratio >= 7 - - result = f" WCAG 对比度检查\n" - result += "─" * 40 + "\n" - result += f" 颜色1: {self._rgb_to_hex(r1, g1, b1)} {self._format_color_block(r1, g1, b1)}\n" - result += f" 颜色2: {self._rgb_to_hex(r2, g2, b2)} {self._format_color_block(r2, g2, b1)}\n" - result += f" 对比度: {ratio:.2f}:1\n" - result += "─" * 40 + "\n" - result += f" {' ' if aa_normal else '❌'} AA 正文 (≥4.5:1)\n" - result += f" {' ' if aa_large else '❌'} AA 大字 (≥3:1)\n" - result += f" {' ' if aaa_normal else '❌'} AAA 正文 (≥7:1)\n" - result += f" {' ' if aaa_large else '❌'} AAA 大字 (≥4.5:1)\n" - return result.rstrip() - - def _op_palette(self, args: List[str]) -> str: - if not args: - return "错误：请提供基础颜色" - r, g, b = self._parse_color(args[0]) - opts = self._parse_opts(args[1:]) - count = int(opts.get("n", 9)) - count = max(3, min(count, 15)) - - h, l, s = colorsys.rgb_to_hls(r / 255, g / 255, b / 255) - - result = f" 调色板 (基于 {args[0]}):\n" - result += "─" * 50 + "\n" - - steps = count - for i in range(steps): - # 从暗到亮 - new_l = 0.1 + (0.8 * i / (steps - 1)) - nr, ng, nb = colorsys.hls_to_rgb(h, new_l, s) - nr, ng, nb = int(nr * 255), int(ng * 255), int(nb * 255) - hex_val = self._rgb_to_hex(nr, ng, nb) - bar = "█" * int(new_l * 20) - result += f" {hex_val} {bar} L:{int(new_l*100)}%\n" - - return result.rstrip() - - def _op_complementary(self, args: List[str]) -> str: - if not args: - return "错误：请提供颜色" - r, g, b = self._parse_color(args[0]) - h, l, s = colorsys.rgb_to_hls(r / 255, g / 255, b / 255) - # 互补色: 色相旋转180° - comp_h = (h + 0.5) % 1.0 - cr, cg, cb = colorsys.hls_to_rgb(comp_h, l, s) - cr, cg, cb = int(cr * 255), int(cg * 255), int(cb * 255) - - result = f" 互补色:\n" - result += f" 原色: {self._rgb_to_hex(r, g, b)}\n" - result += f" 互补: {self._rgb_to_hex(cr, cg, cb)}\n" - return result - - def _op_analogous(self, args: List[str]) -> str: - if not args: - return "错误：请提供颜色" - r, g, b = self._parse_color(args[0]) - h, l, s = colorsys.rgb_to_hls(r / 255, g / 255, b / 255) - - result = f" 类似色:\n" - for offset in [-30, -15, 0, 15, 30]: - ah = (h + offset / 360) % 1.0 - ar, ag, ab = colorsys.hls_to_rgb(ah, l, s) - ar, ag, ab = int(ar * 255), int(ag * 255), int(ab * 255) - marker = " ◀" if offset == 0 else "" - result += f" {self._rgb_to_hex(ar, ag, ab)} ({offset:+d}°){marker}\n" - return result.rstrip() - - def _op_triadic(self, args: List[str]) -> str: - if not args: - return "错误：请提供颜色" - r, g, b = self._parse_color(args[0]) - h, l, s = colorsys.rgb_to_hls(r / 255, g / 255, b / 255) - - result = f" 三色组:\n" - for i, offset in enumerate([0, 120, 240]): - th = (h + offset / 360) % 1.0 - tr, tg, tb = colorsys.hls_to_rgb(th, l, s) - tr, tg, tb = int(tr * 255), int(tg * 255), int(tb * 255) - result += f" {i+1}. {self._rgb_to_hex(tr, tg, tb)} ({offset}°)\n" - return result.rstrip() - - def _op_random(self, args: List[str]) -> str: - opts = self._parse_opts(args) - count = int(opts.get("n", 5)) - count = min(count, 20) - - result = f" 随机颜色 ({count} 个):\n" - for _ in range(count): - r, g, b = __import__('random').randint(0, 255), __import__('random').randint(0, 255), __import__('random').randint(0, 255) - result += f" {self._rgb_to_hex(r, g, b)} rgb({r}, {g}, {b})\n" - return result.rstrip() - - def _op_mix(self, args: List[str]) -> str: - if len(args) < 2: - return "错误：请提供两种颜色" - r1, g1, b1 = self._parse_color(args[0]) - r2, g2, b2 = self._parse_color(args[1]) - opts = self._parse_opts(args[2:]) - w = float(opts.get("w", 0.5)) - w = max(0, min(1, w)) - - mr = int(r1 * (1 - w) + r2 * w) - mg = int(g1 * (1 - w) + g2 * w) - mb = int(b1 * (1 - w) + b2 * w) - - result = f" 颜色混合 (权重 {w}):\n" - result += f" 颜色1: {self._rgb_to_hex(r1, g1, b1)}\n" - result += f" 颜色2: {self._rgb_to_hex(r2, g2, b2)}\n" - result += f" 混合: {self._rgb_to_hex(mr, mg, mb)}\n" - return result - - def _op_blind(self, args: List[str]) -> str: - if len(args) < 2: - return "错误：请提供颜色和色盲类型\n类型: protanopia, deuteranopia, tritanopia" - r, g, b = self._parse_color(args[0]) - blind_type = args[1].lower() - - # 简化的色盲模拟矩阵 - matrices = { - "protanopia": [[0.567, 0.433, 0.0], [0.558, 0.442, 0.0], [0.0, 0.242, 0.758]], - "deuteranopia": [[0.625, 0.375, 0.0], [0.7, 0.3, 0.0], [0.0, 0.3, 0.7]], - "tritanopia": [[0.95, 0.05, 0.0], [0.0, 0.433, 0.567], [0.0, 0.475, 0.525]], - } - - matrix = matrices.get(blind_type) - if not matrix: - return f"错误：未知类型 '{blind_type}'\n支持: protanopia, deuteranopia, tritanopia" - - nr = int(matrix[0][0] * r + matrix[0][1] * g + matrix[0][2] * b) - ng = int(matrix[1][0] * r + matrix[1][1] * g + matrix[1][2] * b) - nb = int(matrix[2][0] * r + matrix[2][1] * g + matrix[2][2] * b) - nr, ng, nb = max(0, min(255, nr)), max(0, min(255, ng)), max(0, min(255, nb)) - - result = f" 色盲模拟 ({blind_type}):\n" - result += f" 原色: {self._rgb_to_hex(r, g, b)}\n" - result += f" 模拟: {self._rgb_to_hex(nr, ng, nb)}\n" - return result - - -def test_plugin(): - plugin = Liugin() - print("颜色工具测试:") - print(plugin.handle('convert "#3498db"')) - print(plugin.handle('contrast "#FFFFFF" "#000000"')) - print(plugin.handle('palette "#e74c3c" -n 7')) - print(plugin.handle('complementary "#3498db"')) - - -if __name__ == "__main__": - test_plugin() diff --git a/plugins/image_generator.py b/plugins/image_generator.py deleted file mode 100644 index cb9c5ab..0000000 --- a/plugins/image_generator.py +++ /dev/null @@ -1,370 +0,0 @@ -""" -图像生成插件 - 基于 ai.2x.nz 的 ComfyUI 在线服务 -支持自然语言描述和英文 Tag 两种模式 -""" -import os -import json -import asyncio -import urllib.request -import time -from typing import Optional - - -class Liugin: - """图像生成工具 — 自然语言/Tag 生图，基于 ComfyUI 在线服务""" - - def __init__(self): - self.usage = """图像生成工具 — 基于 ai.2x.nz 的 ComfyUI 在线服务 - -操作: - generate <描述> - 自然语言描述生成图片（中文/英文均可） - tag <英文tags> - 直接 Tag 模式（逗号分隔的英文标签） - list - 列出可用的工作流/角色 - status - 查看 GPU 状态 - -示例: - image_generator generate 一个白发蓝眼的猫耳少女在写代码 - image_generator tag masterpiece, 1girl, white hair, blue eyes, cat ears, coding - image_generator generate 画一个赛博朋克风格的城市夜景 --width 1024 --height 768 - -说明: - - 自然语言模式会经过 LLM 翻译为 Tag，效果更好 - - 直接 Tag 模式适合熟悉 Danbooru 标签的用户 - - 默认分辨率 1024x1344（WAI 推荐） - - 最大分辨率 1344x1344 - - 生成的图片保存在 screenshots/ 目录 -""" - self.cli = None - self.api_base = "https://ai.2x.nz" - self.ws_url = "wss://ai.2x.nz/ws/run" - self.default_workflow = "WAI - 无Lora.json" - self.output_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "screenshots") - - def set_cli(self, cli): - self.cli = cli - - def get_tool_info(self): - return { - "name": "image_generator", - "description": "图像生成 — 自然语言/Tag 生图，基于 ComfyUI 在线服务（ai.2x.nz），支持中文描述", - "keywords": ["图像", "生成", "画图", "绘图", "AI画", "生图", "image", "generate", "draw", "paint", - "diffusion", "comfyui", "stable", "sd", "文生图", "text2img"], - "usage": self.usage - } - - def handle(self, args: str) -> str: - try: - parts = args.strip().split(maxsplit=1) - if not parts: - return "错误：请提供操作类型\n\n" + self.usage - - op = parts[0].lower() - rest = parts[1] if len(parts) > 1 else "" - - handlers = { - "generate": self._generate, - "tag": self._generate_tag, - "gen": self._generate, - "g": self._generate, - "list": self._list_workflows, - "ls": self._list_workflows, - "status": self._gpu_status, - } - - handler = handlers.get(op) - if not handler: - # 如果没有指定操作，把整个 args 当作自然语言描述 - return self._generate(args) - - return handler(rest) - - except Exception as e: - return f"图像生成错误: {e}" - - def _generate(self, args: str) -> str: - """自然语言模式生成图片""" - # 解析 --width 和 --height 参数 - width, height = 1024, 1344 - prompt = args.strip() - - if "--width" in prompt or "--height" in prompt or "--w" in prompt or "--h" in prompt: - import re - w_match = re.search(r'--w(?:idth)?\s+(\d+)', prompt) - h_match = re.search(r'--h(?:eight)?\s+(\d+)', prompt) - if w_match: - width = int(w_match.group(1)) - prompt = prompt[:w_match.start()] + prompt[w_match.end():] - if h_match: - height = int(h_match.group(1)) - prompt = prompt[:h_match.start()] + prompt[h_match.end():] - prompt = prompt.strip() - - if not prompt: - return "错误：请提供图片描述\n示例: image_generator generate 一个白发猫耳少女" - - # 限制分辨率 - width = min(width, 1344) - height = min(height, 1344) - - return self._run_generation(prompt, "", width, height, rewrite=True, override=False) - - def _generate_tag(self, args: str) -> str: - """直接 Tag 模式""" - width, height = 1024, 1344 - prompt = args.strip() - - if "--width" in prompt or "--height" in prompt or "--w" in prompt or "--h" in prompt: - import re - w_match = re.search(r'--w(?:idth)?\s+(\d+)', prompt) - h_match = re.search(r'--h(?:eight)?\s+(\d+)', prompt) - if w_match: - width = int(w_match.group(1)) - prompt = prompt[:w_match.start()] + prompt[w_match.end():] - if h_match: - height = int(h_match.group(1)) - prompt = prompt[:h_match.start()] + prompt[h_match.end():] - prompt = prompt.strip() - - if not prompt: - return "错误：请提供 Tag\n示例: image_generator tag masterpiece, 1girl, white hair, blue eyes" - - width = min(width, 1344) - height = min(height, 1344) - - return self._run_generation("", prompt, width, height, rewrite=False, override=False) - - def _run_generation(self, nl_prompt: str, direct_prompt: str, width: int, height: int, - rewrite: bool = False, override: bool = False) -> str: - """执行图像生成（同步包装）""" - try: - import websocket as ws_lib - except ImportError: - # 回退：用 urllib 的方式不行，WebSocket 必须用专用库 - # 尝试用 subprocess 调用 python 脚本 - return self._run_generation_subprocess(nl_prompt, direct_prompt, width, height, rewrite, override) - - try: - start_time = time.time() - ws = ws_lib.create_connection(self.ws_url, timeout=180) - - payload = { - "workflow_path": self.default_workflow, - "inline_workflow": None, - "direct_prompt": direct_prompt, - "nl_prompt": nl_prompt, - "rewrite": rewrite, - "override": override, - "width": width, - "height": height, - } - ws.send(json.dumps(payload)) - - image_path = None - final_prompt = "" - error_msg = "" - - while True: - try: - raw = ws.recv() - data = json.loads(raw) - msg_type = data.get("type", "") - - if msg_type == "image": - url = data.get("url", "") - filename = data.get("filename", "") - if url: - image_path = self._download_image(url, filename) - elif msg_type == "final_prompt": - final_prompt = data.get("final_prompt", "") if isinstance(data.get("final_prompt"), str) else data.get("prompt", "") - elif msg_type == "done": - break - elif msg_type == "error": - error_msg = data.get("message", "未知错误") - break - except Exception as e: - error_msg = str(e) - break - - ws.close() - elapsed = time.time() - start_time - - if error_msg: - return f" 生成失败: {error_msg}" - - if image_path: - size = os.path.getsize(image_path) - size_str = f"{size/1024:.0f}KB" if size < 1024*1024 else f"{size/1024/1024:.1f}MB" - result = f" 图片生成成功！\n" - result += f" 路径: {image_path}\n" - result += f" 大小: {size_str}\n" - result += f" 分辨率: {width}x{height}\n" - result += f"⏱ 耗时: {elapsed:.1f}s\n" - if final_prompt: - result += f" Prompt: {final_prompt[:120]}...\n" - result += f"\n💡 提示: 用 'send_image {image_path}' 可以推送到手机" - return result - else: - return " 生成完成但未收到图片数据" - - except Exception as e: - return f" 连接失败: {e}\n💡 请检查网络连接或稍后重试" - - def _run_generation_subprocess(self, nl_prompt: str, direct_prompt: str, width: int, height: int, - rewrite: bool, override: bool) -> str: - """用子进程执行 WebSocket 生成（无需 websocket 库）""" - import subprocess - import tempfile - - script = f''' -import json, asyncio, urllib.request, os, sys - -async def main(): - try: - import websockets - except ImportError: - print("ERROR:需要安装 websockets: pip install websockets") - sys.exit(1) - - uri = "wss://ai.2x.nz/ws/run" - async with websockets.connect(uri) as ws: - payload = {{ - "workflow_path": "{self.default_workflow}", - "inline_workflow": None, - "direct_prompt": {json.dumps(direct_prompt)}, - "nl_prompt": {json.dumps(nl_prompt)}, - "rewrite": {"true" if rewrite else "false"}, - "override": {"true" if override else "false"}, - "width": {width}, - "height": {height}, - }} - await ws.send(json.dumps(payload)) - - image_url = None - while True: - try: - msg = await asyncio.wait_for(ws.recv(), timeout=180) - data = json.loads(msg) - t = data.get("type", "") - if t == "image": - image_url = data.get("url", "") - elif t == "done": - break - elif t == "error": - print(f"ERROR:{{data.get('message','')}}") - break - except asyncio.TimeoutError: - print("ERROR:超时") - break - - if image_url: - full_url = f"https://ai.2x.nz{{image_url}}" - out_path = os.path.join("{self.output_dir}", f"generated_{{os.path.basename(image_url)}}") - os.makedirs(os.path.dirname(out_path), exist_ok=True) - urllib.request.urlretrieve(full_url, out_path) - print(f"OK:{{out_path}}") - -asyncio.run(main()) -''' - - script_path = os.path.join(self.output_dir, "_gen_temp.py") - os.makedirs(self.output_dir, exist_ok=True) - with open(script_path, 'w') as f: - f.write(script) - - try: - result = subprocess.run( - ["python3", script_path], - capture_output=True, text=True, timeout=200 - ) - output = result.stdout.strip() - - if output.startswith("OK:"): - image_path = output[3:] - size = os.path.getsize(image_path) - size_str = f"{size/1024:.0f}KB" if size < 1024*1024 else f"{size/1024/1024:.1f}MB" - res = f" 图片生成成功！\n" - res += f" 路径: {image_path}\n" - res += f" 大小: {size_str}\n" - res += f" 分辨率: {width}x{height}\n" - res += f"\n💡 提示: 用 'send_image {image_path}' 可以推送到手机" - return res - elif output.startswith("ERROR:"): - return f" {output[6:]}" - else: - return f" 生成异常:\n{result.stdout}\n{result.stderr}" - except subprocess.TimeoutExpired: - return " 生成超时（>200s），请稍后重试" - finally: - try: - os.remove(script_path) - except: - pass - - def _download_image(self, url: str, filename: str) -> str: - """下载生成的图片""" - full_url = f"{self.api_base}{url}" - os.makedirs(self.output_dir, exist_ok=True) - out_path = os.path.join(self.output_dir, f"gen_{filename}") - urllib.request.urlretrieve(full_url, out_path) - return out_path - - def _list_workflows(self, args: str) -> str: - """列出可用的工作流""" - try: - import urllib.request - req = urllib.request.Request(f"{self.api_base}/api/workflows") - with urllib.request.urlopen(req, timeout=10) as resp: - data = json.loads(resp.read()) - - workflows = data.get("workflows", []) - result = f" 可用工作流 ({len(workflows)} 个):\n\n" - - lora_workflows = [] - no_lora = [] - for w in workflows: - name = w.get("path", "") - if "无Lora" in name or "no" in name.lower(): - no_lora.append(name) - else: - lora_workflows.append(name) - - if no_lora: - result += " 通用（无角色绑定）:\n" - for w in no_lora: - result += f" • {w}\n" - - if lora_workflows: - result += "\n 角色工作流:\n" - for w in lora_workflows: - result += f" • {w}\n" - - result += "\n💡 使用 'image_generator generate <描述>' 自然语言生图" - result += "\n💡 使用 'image_generator tag ' 直接 Tag 生图" - return result - - except Exception as e: - return f" 获取工作流列表失败: {e}" - - def _gpu_status(self, args: str) -> str: - """查看 GPU 状态""" - try: - import urllib.request - req = urllib.request.Request(f"{self.api_base}/api/gpu") - with urllib.request.urlopen(req, timeout=10) as resp: - data = json.loads(resp.read()) - - result = " 服务器 GPU 状态:\n" - for gpu in data.get("gpus", []): - name = gpu.get("name", "未知") - util = gpu.get("utilization", 0) - mem_used = gpu.get("memory_used", 0) - mem_total = gpu.get("memory_total", 0) - temp = gpu.get("temperature", 0) - result += f" {name}\n" - result += f" GPU 负载: {util}%\n" - result += f" 显存: {mem_used}/{mem_total} MiB\n" - result += f" 温度: {temp}°C\n" - return result - - except Exception as e: - return f" 获取 GPU 状态失败: {e}" diff --git a/plugins/send_image.py b/plugins/send_image.py deleted file mode 100644 index fb045eb..0000000 --- a/plugins/send_image.py +++ /dev/null @@ -1,144 +0,0 @@ -""" -发送图片插件 - Clawli 模式下向手机端发送图片 -支持独立进程服务器模式 -""" -import os -import json -from typing import Dict, Any - - -def get_clawli_server(): - """获取 Clawli 服务器实例（优先使用独立进程服务器）""" - try: - import clawli_server - if clawli_server.is_running(): - return clawli_server - except ImportError: - pass - - # 回退到旧的 WebSocket 服务器 - try: - from websocket_server import websocket_server - return websocket_server - except ImportError: - return None - - -class Liugin: - """发送图片插件 - 在 Clawli 模式下发送图片到手机端""" - - def __init__(self): - self.name = "send_image" - self.cli = None - - def set_cli(self, cli_instance): - """设置 CLI 引用""" - self.cli = cli_instance - - def get_tool_info(self): - """返回工具信息""" - return { - "name": "send_image", - "description": "在 Clawli 模式下发送图片到手机端，支持 JPG、PNG、GIF、BMP、WEBP 格式", - "keywords": ["图片", "发送", "手机", "clawli", "image", "send"], - "usage": "send_image {\"image_path\": \"图片路径\", \"caption\": \"说明文字\"}" - } - - def handle(self, args): - """处理工具调用 - - Args: - args: 可以是 JSON 字符串或字典 - """ - # 解析参数 - if isinstance(args, str): - try: - params = json.loads(args) - except json.JSONDecodeError: - params = {"image_path": args.strip()} - elif isinstance(args, dict): - params = args - else: - return "错误: 参数格式无效" - - image_path = params.get("image_path", "") - caption = params.get("caption", "") - - if not image_path: - return "错误: 请提供图片路径" - - # 检查文件是否存在 - if not os.path.isfile(image_path): - return f"错误: 图片文件不存在: {image_path}" - - # 检查是否是图片文件 - ext = image_path.lower().split('.')[-1] if '.' in image_path else '' - supported_formats = ['jpg', 'jpeg', 'png', 'gif', 'bmp', 'webp'] - if ext not in supported_formats: - return f"错误: 不支持的图片格式: {ext}，支持: {', '.join(supported_formats)}" - - # 获取服务器 - server = get_clawli_server() - if not server: - return "错误: Clawli 服务器未初始化" - - # 尝试使用独立进程服务器 - try: - import clawli_server - if clawli_server.is_running(): - success, msg = clawli_server.send_image(image_path, caption) - if success: - file_size = os.path.getsize(image_path) - size_str = f"{file_size/1024:.1f}KB" if file_size < 1024*1024 else f"{file_size/1024/1024:.2f}MB" - return f"图片已发送到手机端: {os.path.basename(image_path)} ({size_str})" - else: - return f"错误: {msg}" - except ImportError: - pass - - # 回退到旧的 WebSocket 服务器 - if hasattr(server, 'running') and not server.running: - return "错误: Clawli 模式未启动，请先使用 /remote start 启动远程服务" - - if hasattr(server, 'clients') and not server.clients: - return "错误: 没有连接的手机端，请确保手机已连接" - - if hasattr(server, 'send_image_to_clients'): - success = server.send_image_to_clients(image_path, caption) - if success: - file_size = os.path.getsize(image_path) - size_str = f"{file_size/1024:.1f}KB" if file_size < 1024*1024 else f"{file_size/1024/1024:.2f}MB" - return f"图片已发送到手机端: {os.path.basename(image_path)} ({size_str})" - - return "错误: 发送图片失败" - - -# 工具定义 -TOOLS = [ - { - "type": "function", - "function": { - "name": "send_image", - "description": "在 Clawli 模式下发送图片到手机端。支持 JPG、PNG、GIF、BMP、WEBP 格式。", - "parameters": { - "type": "object", - "properties": { - "image_path": { - "type": "string", - "description": "图片文件的绝对路径" - }, - "caption": { - "type": "string", - "description": "图片说明文字（可选）" - } - }, - "required": ["image_path"] - } - } - } -] - - -def get_tools(): - """返回工具定义列表""" - return TOOLS \ No newline at end of file diff --git a/plugins_config.py b/plugins_config.py index 06a4e42..39aac69 100644 --- a/plugins_config.py +++ b/plugins_config.py @@ -1,11 +1,10 @@ # 小狸插件配置 +# ── 必备生产力 ── ENABLE_AI_SEARCH = True -ENABLE_AUDIO_PLAYER = True ENABLE_CMD_EXECUTOR = True ENABLE_FILE_MANAGER = True ENABLE_CODE_EDITOR = True -ENABLE_CODE_SEARCH = True ENABLE_GIT_TOOLS = True ENABLE_FRONTEND_TESTER = True ENABLE_NETWORK_TOOLS = True @@ -14,10 +13,15 @@ ENABLE_SPEECH_RECOGNITION = True ENABLE_TASK_MANAGER = True ENABLE_TOOL_SEARCH = True ENABLE_AUTO_ENGINEER = True -ENABLE_GIT_WORKFLOW = True -ENABLE_PROJECT_ANALYZER = True - -ENABLE_DGLAB_WS = False - ENABLE_TERMINAL = True ENABLE_OPERIT_BRIDGE = True +ENABLE_BROWSER_AUTO = True +ENABLE_GUI_AUTO = True +ENABLE_SUB_AGENT = True +ENABLE_WORKFLOW = True +ENABLE_MEMORY_PLUGIN = True +ENABLE_JSON_TOOLKIT = True +ENABLE_REGEX_TESTER = True +ENABLE_SNIPPET_MANAGER = True +ENABLE_MARKDOWN_TOOL = True +ENABLE_ID_GENERATOR = True diff --git a/tests/test_terminal.py b/tests/test_terminal.py index 77d05cc..38b1260 100644 --- a/tests/test_terminal.py +++ b/tests/test_terminal.py @@ -1,229 +1,179 @@ -"""终端模拟器插件测试""" +"""终端模拟器 v2 测试""" import os import sys import time sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from plugins.terminal import Liugin, TerminalSession +from plugins.terminal import Liugin, Terminal def test_basic_open_close(): - """打开和关闭终端""" p = Liugin() r = p.handle("open") - assert "✓" in r - assert "s1" in r or "main" in r - + assert "已打开" in r r = p.handle("list") - assert "终端" in r - + assert "main" in r r = p.handle("close") - assert "✓" in r - + assert "已关闭" in r r = p.handle("list") - assert "暂无" in r + assert "无终端" in r print("✓ test_basic_open_close") -def test_send_command(): - """发送命令并获取输出""" +def test_type_and_enter(): p = Liugin() p.handle("open") - - r = p.handle("send echo hello_world") - assert "hello_world" in r, f"期望 hello_world，实际: {r}" - - r = p.handle("send echo 42") - assert "42" in r - - p.handle("close") - print("✓ test_send_command") - - -def test_cwd(): - """切换目录""" - p = Liugin() - p.handle("open") - - r = p.handle("cwd /tmp") - assert "✓" in r - - r = p.handle("send pwd") - assert "/tmp" in r - - p.handle("close") - print("✓ test_cwd") - - -def test_env(): - """环境变量""" - p = Liugin() - p.handle("open") - - r = p.handle("env MY_TEST_VAR hello123") - assert "✓" in r - - r = p.handle("send echo $MY_TEST_VAR") - assert "hello123" in r - - p.handle("close") - print("✓ test_env") - - -def test_multiple_sessions(): - """多会话管理""" - p = Liugin() - - p.handle("open bash session1") - p.handle("open bash session2") - - r = p.handle("list") - assert "2 个" in r - - # 第一个会话是 main - p.handle("send echo AAA") # 当前是 s2，发给 s2 - p.handle("switch main") - p.handle("send echo BBB") # 发给 main - - r = p.handle("close main") - assert "✓" in r - - r = p.handle("list") - assert "1 个" in r - - p.handle("close s2") - print("✓ test_multiple_sessions") - - -def test_keys(): - """特殊按键""" - p = Liugin() - p.handle("open") - - r = p.handle("keys") - assert "ctrl+c" in r - - # 发送 Ctrl+C（不应该报错） - r = p.handle("keys ctrl+c") - assert "←" in r - - p.handle("close") - print("✓ test_keys") - - -def test_input(): - """纯文本输入""" - p = Liugin() - p.handle("open") - - p.handle("send cat << 'EOF'") - p.handle("input line1") - p.handle("input line2") - p.handle("input EOF") - - p.handle("close") - print("✓ test_input") - - -def test_status(): - """状态查询""" - p = Liugin() - p.handle("open") - - r = p.handle("status") - assert "终端状态" in r - assert "运行中" in r - - p.handle("close") - print("✓ test_status") - - -def test_read_output(): - """读取输出""" - p = Liugin() - p.handle("open") - - p.handle("send echo AAA") - p.handle("send echo BBB") - p.handle("send echo CCC") - + p.handle("type echo hello_v2") + p.handle("key enter") + time.sleep(0.5) r = p.handle("read") - # read 返回最近一次 send_cmd 的结果 - assert "CCC" in r, f"read 结果: {r[:100]}" - + assert "hello_v2" in r p.handle("close") - print("✓ test_read_output") + print("✓ test_type_and_enter") + + +def test_last_n(): + p = Liugin() + p.handle("open") + for i in range(10): + p.handle(f"type echo line_{i}") + p.handle("key enter") + time.sleep(0.15) + time.sleep(0.5) + r = p.handle("last 3") + assert "line_9" in r + # last 3 不应该包含 line_0 + assert "line_0" not in r + p.handle("close") + print("✓ test_last_n") + + +def test_read_shortcut(): + """read 5 等同 last 5""" + p = Liugin() + p.handle("open") + p.handle("type echo aaa") + p.handle("key enter") + time.sleep(0.3) + p.handle("type echo bbb") + p.handle("key enter") + time.sleep(0.3) + r = p.handle("read 5") + assert "bbb" in r + p.handle("close") + print("✓ test_read_shortcut") + + +def test_env_persist(): + p = Liugin() + p.handle("open") + p.handle("type export MYVAR=hello") + p.handle("key enter") + time.sleep(0.3) + p.handle("type echo val=$MYVAR") + p.handle("key enter") + time.sleep(0.5) + r = p.handle("read") + assert "val=hello" in r + p.handle("close") + print("✓ test_env_persist") def test_python_repl(): - """Python REPL 交互""" p = Liugin() - p.handle("open python3 Python-REPL") - import time; time.sleep(1) - - # REPL 用 input 发送（不触发哨兵），用 read 拿输出 - p.handle('input print(2 + 3)') - time.sleep(0.8) - r = p.handle('read') - assert "5" in r, f"Python REPL 输出: {r[:200]}" - + p.handle("open python3") + time.sleep(1) + p.handle("type print(2+3)") + p.handle("key enter") + time.sleep(0.5) + r = p.handle("read") + assert "5" in r p.handle("close") print("✓ test_python_repl") -def test_session_persistence(): - """会话持久性 — 环境变量在命令间保持""" +def test_ctrl_c(): p = Liugin() p.handle("open") - - p.handle("send export FOO=bar123") - r = p.handle("send echo $FOO") - assert "bar123" in r, f"环境变量未保持: {r}" - - p.handle("send cd /var") - r = p.handle("send pwd") - assert "/var" in r, f"目录未保持: {r}" - + p.handle("type python3 -c 'import time; time.sleep(99)'") + p.handle("key enter") + time.sleep(0.5) + p.handle("key ctrl+c") + time.sleep(0.3) + # 终端应该还活着 + r = p.handle("list") + assert "●" in r p.handle("close") - print("✓ test_session_persistence") + print("✓ test_ctrl_c") -def test_auto_create_session(): - """无会话时自动创建""" +def test_multi_session(): p = Liugin() - # 不 open，直接 send - r = p.handle("send echo auto_created") - assert "auto_created" in r + p.handle("open") + p.handle("open python3") + r = p.handle("list") + assert "main" in r + assert "s2" in r + p.handle("switch main") + p.handle("type echo from_bash") + p.handle("key enter") + time.sleep(0.3) + r = p.handle("read") + assert "from_bash" in r + p.handle("close s2") + p.handle("close main") + print("✓ test_multi_session") + +def test_cwd(): + p = Liugin() + p.handle("open") + p.handle("type cd /tmp") + p.handle("key enter") + time.sleep(0.3) + p.handle("type pwd") + p.handle("key enter") + time.sleep(0.5) + r = p.handle("read") + assert "/tmp" in r p.handle("close") - print("✓ test_auto_create_session") + print("✓ test_cwd") def test_long_output(): - """长输出处理""" p = Liugin() p.handle("open") - - # 生成长输出 - r = p.handle("send seq 1 200") - assert "200" in r or "1" in r - + p.handle("type seq 1 100") + p.handle("key enter") + time.sleep(1) + r = p.handle("last 5") + assert "100" in r p.handle("close") print("✓ test_long_output") +def test_auto_create(): + """不 open 直接 type，自动创建终端""" + p = Liugin() + p.handle("type echo auto") + p.handle("key enter") + time.sleep(0.5) + r = p.handle("read") + assert "auto" in r + p.handle("close") + print("✓ test_auto_create") + + if __name__ == "__main__": test_basic_open_close() - test_send_command() - test_cwd() - test_env() - test_multiple_sessions() - test_keys() - test_input() - test_status() - test_read_output() + test_type_and_enter() + test_last_n() + test_read_shortcut() + test_env_persist() test_python_repl() - test_session_persistence() - test_auto_create_session() + test_ctrl_c() + test_multi_session() + test_cwd() test_long_output() - print("\n🎉 全部 13 个测试通过!") + test_auto_create() + print("\n🎉 全部 11 个测试通过!")