shuiyuPyLab/xiaoli-cli
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

chore: 整理插件分类 - 删除图像类，移动娱乐类

Browse Source 
删除（图像类）:
- 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
This commit is contained in:
xiaoli-cli bot committed 2026-06-23 09:25:25 +00:00
1 parent 5ea19b75ae
commit 125e00a485
10 files changed
+194 -2425
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

plugins/ai_image.py
-589
View File
@@ -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> <<<>>> <描述> - 同时使用 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 <Tag> <<<>>> <描述>"
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 类名
plugins/audio_player.py
-169
View File
@@ -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
plugins/bookmark_manager.py
-489
View File
@@ -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 <URL> -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 <URL> - 获取网站 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()
plugins/color_picker.py
-459
View File
@@ -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 <RGB或HSL> - 转换为 HEX
rgb <HEX或HSL> - 转换为 RGB
hsl <HEX或RGB> - 转换为 HSL
hsv <HEX或RGB> - 转换为 HSV
cmyk <HEX或RGB> - 转换为 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()
plugins/image_generator.py
-370
View File
@@ -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 <tags>' 直接 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}"
plugins/send_image.py
-144
View File
@@ -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