feat: 增强编码能力 - 新增代码编辑器、代码搜索、Git集成

- 新增 plugins/code_editor.py: 精准代码编辑(搜索替换、批量编辑、diff对比)
- 新增 plugins/code_search.py: 代码搜索与理解(跨文件搜索、符号提取、依赖分析)
- 新增 plugins/git_tools.py: Git 版本控制(status/diff/log/commit/branch)
- 新增 core/ 模块: 配置管理、引擎管理、对话处理、显示输出、代码沙箱
- 修复 requirements.txt: 使用正确的 pip 包名(Pillow、opencv-python 等)
- 更新 unified_tool_manager.py: 兼容 Plugin 和 Liugin 两种类名
- 更新系统提示词: 增强编程工作流指导

本次更新由 mimo 自动化执行,与开发者无关。
This commit is contained in:
Xiaoli Dev committed 2026-04-25 10:55:27 +08:00
commit e22652fbbb
53 files changed
+20067

No files matched your search

+174
View File
@@ -0,0 +1,174 @@
import subprocess
import json
from urllib.parse import quote
import requests
class Liugin:
"""AI搜索插件 - 使用jina.ai进行智能搜索"""
def __init__(self):
self.usage = """AI搜索工具使用方法:/r
ai_search <操作> <参数>/r
例如:/r
- ai_search url https://www.example.com - 搜索网址内容/r
- ai_search search Python教程 - 普通搜索/r
- ai_search web 什么是人工智能 - 搜索网络内容/r
/r
详细说明:/r
- url操作: ai_search url <网址> - 提取并搜索指定网址的内容/r
- search操作: ai_search search <搜索词> - 使用必应搜索指定内容/r
- web操作: ai_search web <搜索词> - 使用必应搜索指定内容(与search相同)/r
/r
使用工具的JSON格式示例:/r
{"action": "use_tool", "tool": "ai_search", "args": "url https://www.example.com"} - 搜索网址内容/r
{"action": "use_tool", "tool": "ai_search", "args": "search Python教程"} - 普通搜索/r
{"action": "use_tool", "tool": "ai_search", "args": "web 什么是人工智能"} - 搜索网络内容/r
"""
self.cli = None # CLI实例引用
self.timeout = 30 # 请求超时时间(秒)
def set_cli(self, cli):
"""设置CLI实例引用"""
self.cli = cli
# 注册插件命令
self.cli.register_liugin_command('ai_search', self.command_handler)
self.cli.register_liugin_command('search', self.search_command_handler)
self.cli.register_liugin_command('web', self.web_command_handler)
def command_handler(self, args):
"""处理 /ai_search 命令"""
# 解析参数
parts = args.strip().split(maxsplit=1)
if len(parts) < 1:
return "请提供操作类型: url, search, web"
operation = parts[0].lower()
remaining_args = parts[1] if len(parts) > 1 else ""
# 调用实际的AI搜索功能
return self.handle(f"{operation} {remaining_args}")
def search_command_handler(self, args):
"""处理 /search 命令"""
return self.handle(f"search {args}")
def web_command_handler(self, args):
"""处理 /web 命令"""
return self.handle(f"search {args}")
def get_tool_info(self):
return {
"name": "ai_search",
"description": "AI搜索工具,使用jina.ai服务进行智能搜索,支持网址内容提取和网络搜索。当用户询问天气、新闻、汇率、股票、航班、实时信息、网络内容搜索等问题时,应该直接使用此工具。使用格式:ai_search search <搜索词> 或 ai_search url <网址>",
"keywords": ["搜索", "search", "网址", "url", "web", "AI搜索", "网络搜索", "内容提取", "天气", "新闻", "实时信息"],
"usage": """ai_search 工具使用说明:
JSON格式示例:
{\"action\": \"use_tool\", \"tool\": \"ai_search\", \"args\": \"url https://www.example.com\"} - 搜索网址内容
{\"action\": \"use_tool\", \"tool\": \"ai_search\", \"args\": \"search Python教程\"} - 普通搜索
{\"action\": \"use_tool\", \"tool\": \"ai_search\", \"args\": \"web 什么是人工智能\"} - 搜索网络内容"""
}
def handle(self, args):
"""处理AI搜索请求"""
try:
# 解析参数
parts = args.strip().split(maxsplit=1)
if len(parts) < 1:
return "错误:参数不足。请提供操作类型。\n可用操作: url, search, web"
operation = parts[0].lower()
if operation == "url":
# 搜索网址内容
if len(parts) < 2:
return "错误:请提供URL。格式: url <网址>"
url = parts[1].strip()
return self._search_url(url)
elif operation in ["search", "web"]:
# 普通搜索
if len(parts) < 2:
return "错误:请提供搜索词。格式: search <搜索词>"
query = parts[1].strip()
return self._search_web(query)
else:
return f"错误:不支持的操作 '{operation}'。支持的操作有: url, search, web。"
except Exception as e:
return f"AI搜索操作错误: {str(e)}"
def _search_url(self, url):
"""搜索网址内容(使用jina.ai提取)"""
try:
# 构建jina.ai的URL - 直接使用URL
jina_url = f"https://r.jina.ai/{url}"
result = self._execute_curl(jina_url)
if result.startswith("错误:"):
return result
return f"📄 网址内容 (来自 {url}):\n\n{result}"
except Exception as e:
return f"网址搜索错误: {str(e)}"
def _search_web(self, query):
"""普通搜索(使用jina.ai + 必应搜索)"""
try:
# 对搜索词进行URL编码,确保中文等特殊字符正确编码
# 使用safe=''确保所有非字母数字字符都被编码
encoded_query = quote(query, safe='')
# 构建必应搜索URL
bing_url = f"https://cn.bing.com/search?q={encoded_query}"
# 构建jina.ai的URL - 直接使用必应URL,不需要二次编码
jina_url = f"https://r.jina.ai/{bing_url}"
result = self._execute_curl(jina_url)
if result.startswith("错误:"):
return result
return f"🔍 搜索结果 (搜索词: {query}):\n\n{result}"
except Exception as e:
return f"网络搜索错误: {str(e)}"
def _execute_curl(self, url):
"""使用 requests 库获取内容(替代 curl 命令)"""
try:
# 使用 requests 库发送请求
response = requests.get(
url,
timeout=self.timeout,
headers={'User-Agent': 'Mozilla/5.0 (compatible; AI-Search-Bot)'}
)
# 检查状态码
if response.status_code != 200:
return f"错误:HTTP请求失败,状态码: {response.status_code}"
# 获取内容
content = response.text.strip()
if not content:
return "错误:未获取到任何内容"
# 限制输出长度以避免过长输出
if len(content) > 5000:
content = content[:5000] + f"\n\n... (内容已截断,完整内容共{len(content)}字符)"
return content
except requests.Timeout:
return f"错误:请求超时 (>{self.timeout}秒)"
except requests.ConnectionError:
return "错误:网络连接失败,请检查网络"
except Exception as e:
return f"错误:请求时发生错误: {str(e)}"
+150
View File
@@ -0,0 +1,150 @@
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 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
+188
View File
@@ -0,0 +1,188 @@
import subprocess
import os
import sys
import time
from colorama import Fore, Style
class Liugin:
"""CMD命令执行插件 - 仅限Windows环境"""
def __init__(self):
self.usage = """CMD命令执行工具使用方法:
cmd_executor <操作> <参数>
例如:
- cmd_executor run dir - 执行dir命令查看当前目录内容
- cmd_executor run "ipconfig /all" - 执行ipconfig命令查看网络配置
- cmd_executor run "tasklist | findstr python" - 查找Python相关进程
详细说明:
- run操作: cmd_executor run <命令> - 执行指定的CMD命令
使用工具的JSON格式示例:
{"action": "use_tool", "tool": "cmd_executor", "args": "run dir"} - 执行dir命令
{"action": "use_tool", "tool": "cmd_executor", "args": "run ipconfig /all"} - 执行ipconfig命令
重要说明:
- 此工具仅在Windows环境下可用
- 为安全起见,某些系统命令可能被限制执行
- 命令执行结果会返回给AI进行分析"""
self.cli = None # CLI实例引用
def set_cli(self, cli):
"""设置CLI实例引用"""
self.cli = cli
# 注册插件命令
self.cli.register_liugin_command('cmd', self.command_handler)
def command_handler(self, args):
"""处理 /cmd 命令"""
# 解析参数
parts = args.strip().split(maxsplit=1)
if len(parts) < 1:
return "请提供操作类型: run <命令>"
operation = parts[0].lower()
remaining_args = parts[1] if len(parts) > 1 else ""
# 调用实际的CMD执行功能
return self.handle(f"{operation} {remaining_args}")
def get_tool_info(self):
return {
"name": "cmd_executor",
"description": "CMD命令执行工具,用于执行Windows命令行指令",
"keywords": ["命令", "执行", "CMD", "cmd", "executor", "Windows", "指令", "操作"],
"usage": """cmd_executor 工具使用说明:
JSON格式示例:
{"action": "use_tool", "tool": "cmd_executor", "args": "run dir"} - 执行dir命令
{"action": "use_tool", "tool": "cmd_executor", "args": "run ipconfig /all"} - 执行ipconfig命令
{"action": "use_tool", "tool": "cmd_executor", "args": "run tasklist"} - 列出所有进程
注意:仅限Windows环境使用"""
}
def handle(self, args):
"""处理CMD命令执行请求"""
try:
# 解析参数
parts = args.strip().split(maxsplit=1)
if len(parts) < 1:
return "错误:参数不足。请提供操作类型。\n可用操作: run"
operation = parts[0].lower()
if operation == "run":
# 检查是否为Windows系统
if os.name != 'nt':
return "错误:CMD命令执行工具仅支持Windows系统。"
if len(parts) < 2:
return "错误:请提供要执行的命令。格式: run <命令>"
command = parts[1]
return self._execute_command(command)
else:
return f"错误:不支持的操作 '{operation}'。支持的操作有: run。"
except Exception as e:
return f"CMD命令执行错误: {str(e)}"
def _execute_command(self, command):
"""执行CMD命令"""
try:
# 解析命令和可能的超时参数
# 允许格式如 "command --timeout 10" 或 "command -t 10"
timeout = 30 # 默认超时时间
import re
# 检查是否在命令中指定超时
timeout_pattern = r'(?:--timeout|-t)\s+(\d+)'
timeout_match = re.search(timeout_pattern, command, re.IGNORECASE)
if timeout_match:
timeout = int(timeout_match.group(1))
# 移除超时参数,只保留实际命令
command = re.sub(timeout_pattern, '', command).strip()
# 检查是否有CLI实例可用,请求用户确认
if self.cli is not None:
# 检查是否是 Clawli 远程模式
if not getattr(self.cli, 'is_clawli_mode', False):
# 非远程模式,请求用户确认
try:
from ai_cli import ask_user_confirmation
# 生成唯一的请求ID
import uuid
request_id = str(uuid.uuid4())
# 存储请求信息
self.cli.user_input_queue[request_id] = {
'type': 'cmd_confirm',
'command': command,
'timeout': timeout,
'status': 'pending'
}
# 发送确认请求给用户
confirmed = ask_user_confirmation(
f"即将执行CMD命令: {command}\n超时时间: {timeout}秒\n是否确认执行?",
request_id
)
if not confirmed:
return f"命令执行已取消: {command}"
except Exception as e:
# 如果无法请求确认,直接执行
pass
# 执行命令
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
timeout=timeout, # 使用自定义超时时间
encoding='utf-8', # 指定编码
errors='replace' # 处理编码错误
)
# 获取输出
stdout = result.stdout.strip()
stderr = result.stderr.strip()
return_code = result.returncode
output = ""
if stdout:
output += f"命令输出:\n{stdout}\n"
if stderr:
output += f"错误信息:\n{stderr}\n"
if return_code != 0:
output += f"命令退出码: {return_code}\n"
if not output:
output = f"命令执行完成,无输出。(超时设置: {timeout}秒)"
# 限制输出长度为5000字符
if len(output) > 5000:
output = output[:5000] + f"\n... (输出已截断,超时设置: {timeout}秒)"
return output
except subprocess.TimeoutExpired:
return f"错误:命令执行超时(超过{timeout}秒),已取消执行。"
except ValueError:
return f"错误:超时值无效,已使用默认30秒超时。命令: {command}"
except Exception as e:
return f"命令执行失败: {str(e)}"
# 测试函数
def test_plugin():
"""测试插件功能"""
plugin = Plugin()
print("CMD执行器插件测试:")
print("工具信息:", plugin.get_tool_info())
print("使用说明:", plugin.usage)
if __name__ == "__main__":
test_plugin()
+542
View File
@@ -0,0 +1,542 @@
"""
代码编辑器插件 - 小狸的精准代码编辑能力
对标 Claude Code 的多文件编辑、上下文感知修改
"""
import os
import re
import difflib
from typing import Optional, Tuple, List, Dict
class Plugin:
"""精准代码编辑器 - 支持搜索替换、多文件编辑、diff 预览"""
def __init__(self):
self.usage = """代码编辑器 - 精准修改代码文件
操作类型:
edit <文件路径> <old_text> <new_text> - 精准替换代码片段
multi <文件路径> <JSON批量操作> - 批量修改(JSON数组)
diff <文件路径1> <文件路径2> - 比较两个文件差异
diff_text <old_text> <new_text> - 比较两段文本差异
insert <文件路径> <行号> <内容> - 在指定行插入内容
delete_lines <文件路径> <起始行> <结束行> - 删除指定行范围
read_range <文件路径> <起始行> <结束行> - 读取指定行范围
search <目录> <正则表达式> [文件模式] - 在代码中搜索
ast_info <文件路径> - 获取 Python 文件的 AST 摘要
create <文件路径> <内容> - 创建新文件
write <文件路径> <内容> - 写入/覆盖文件内容
append <文件路径> <内容> - 追加内容到文件
"""
self.cli = None
def set_cli(self, cli):
self.cli = cli
def get_tool_info(self):
return {
"name": "code_editor",
"description": "精准代码编辑器 - 支持搜索替换、多文件批量编辑、代码搜索、diff对比。用于修改代码文件的核心工具。",
"keywords": ["编辑", "修改", "代码", "文件", "edit", "replace", "search", "diff", "code"],
"usage": self.usage
}
def get_mcp_definition(self):
return {
"name": "code_editor",
"description": "精准代码编辑器",
"inputSchema": {
"type": "object",
"properties": {
"operation": {
"type": "string",
"enum": ["edit", "multi", "diff", "diff_text", "insert",
"delete_lines", "read_range", "search", "ast_info",
"create", "write", "append"],
"description": "操作类型"
},
"path": {"type": "string", "description": "文件路径"},
"old_text": {"type": "string", "description": "要替换的原文"},
"new_text": {"type": "string", "description": "替换后的新文本"},
"content": {"type": "string", "description": "文件内容"},
"line_number": {"type": "integer", "description": "行号"},
"start_line": {"type": "integer", "description": "起始行"},
"end_line": {"type": "integer", "description": "结束行"},
"pattern": {"type": "string", "description": "搜索正则表达式"},
"file_pattern": {"type": "string", "description": "文件匹配模式"},
"directory": {"type": "string", "description": "搜索目录"},
"operations": {"type": "array", "description": "批量操作列表"}
},
"required": ["operation"]
}
}
def convert_mcp_args(self, arguments):
op = arguments.get("operation", "")
path = arguments.get("path", "")
return f"{op} {path}"
def handle(self, args: str) -> str:
try:
parts = args.strip().split(maxsplit=2)
if not parts:
return "错误:请提供操作类型"
operation = parts[0].lower()
handlers = {
"edit": self._op_edit,
"multi": self._op_multi_edit,
"diff": self._op_diff,
"diff_text": self._op_diff_text,
"insert": self._op_insert,
"delete_lines": self._op_delete_lines,
"read_range": self._op_read_range,
"search": self._op_search,
"ast_info": self._op_ast_info,
"create": self._op_create,
"write": self._op_write,
"append": self._op_append,
}
handler = handlers.get(operation)
if not handler:
return f"错误:不支持的操作 '{operation}'"
return handler(parts[1] if len(parts) > 1 else "",
parts[2] if len(parts) > 2 else "")
except Exception as e:
return f"代码编辑器错误: {str(e)}"
def _normalize_path(self, path: str) -> str:
"""规范化路径"""
path = path.strip().strip('"').strip("'")
path = os.path.expanduser(path)
path = os.path.normpath(path)
return path
def _op_edit(self, path: str, rest: str) -> str:
"""精准替换代码片段 - 核心编辑能力"""
if not path:
return "错误:请提供文件路径"
file_path = self._normalize_path(path)
if not os.path.exists(file_path):
return f"错误:文件不存在: {file_path}"
# 从 rest 中解析 old_text 和 new_text
# 格式: old_text <<<SEPARATOR>>> new_text
separator = " <<<>>> "
if separator not in rest:
return "错误:格式应为: edit <文件> <old_text> <<<>>> <new_text>"
old_text, new_text = rest.split(separator, 1)
if not old_text.strip():
return "错误:old_text 不能为空"
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
# 精准匹配(去除首尾空白但保留内部格式)
old_text = old_text.strip()
new_text = new_text.strip()
if old_text not in content:
# 尝试模糊匹配
lines = content.split('\n')
old_lines = old_text.split('\n')
match_info = self._fuzzy_match(lines, old_lines)
if match_info:
return f"错误:未找到精确匹配。最接近的位置在第 {match_info[0]+1}-{match_info[1]} 行,但内容不完全一致。请检查缩进和空白字符。"
return f"错误:在文件中未找到要替换的内容。请确保文本完全匹配(包括缩进)。"
count = content.count(old_text)
if count > 1:
return f"警告:找到 {count} 处匹配,请提供更多上下文以确保唯一匹配。"
new_content = content.replace(old_text, new_text, 1)
with open(file_path, 'w', encoding='utf-8') as f:
f.write(new_content)
# 生成 diff 预览
diff = self._make_diff(old_text, new_text, file_path)
return f"✅ 已编辑: {file_path}\n\n{diff}"
def _op_multi_edit(self, path: str, rest: str) -> str:
"""批量编辑 - 一次修改多处"""
import json
if not path:
return "错误:请提供文件路径"
file_path = self._normalize_path(path)
if not os.path.exists(file_path):
return f"错误:文件不存在: {file_path}"
try:
operations = json.loads(rest)
except json.JSONDecodeError:
return "错误:operations 必须是有效的 JSON 数组。格式: [{\"old\": \"...\", \"new\": \"...\"}]"
if not isinstance(operations, list):
return "错误:operations 必须是数组"
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
results = []
for i, op in enumerate(operations):
old_text = op.get("old", "").strip()
new_text = op.get("new", "").strip()
if not old_text:
results.append(f" ❌ 操作 {i+1}: old 为空")
continue
if old_text not in content:
results.append(f" ❌ 操作 {i+1}: 未找到匹配")
continue
content = content.replace(old_text, new_text, 1)
results.append(f" ✅ 操作 {i+1}: 已替换")
with open(file_path, 'w', encoding='utf-8') as f:
f.write(content)
return f"✅ 批量编辑完成: {file_path}\n" + "\n".join(results)
def _op_diff(self, path: str, rest: str) -> str:
"""比较两个文件差异"""
if not path:
return "错误:请提供第一个文件路径"
file1 = self._normalize_path(path)
file2 = self._normalize_path(rest.split()[0] if rest.strip() else "")
if not os.path.exists(file1):
return f"错误:文件不存在: {file1}"
if not os.path.exists(file2):
return f"错误:文件不存在: {file2}"
with open(file1, 'r', encoding='utf-8') as f:
lines1 = f.readlines()
with open(file2, 'r', encoding='utf-8') as f:
lines2 = f.readlines()
diff = difflib.unified_diff(lines1, lines2,
fromfile=file1, tofile=file2, lineterm='')
diff_text = '\n'.join(diff)
if not diff_text:
return "两个文件完全相同"
return f"文件差异:\n{diff_text}"
def _op_diff_text(self, path: str, rest: str) -> str:
"""比较两段文本差异"""
separator = " <<<>>> "
if separator not in rest:
return "错误:格式: diff_text <old_text> <<<>>> <new_text>"
old_text, new_text = rest.split(separator, 1)
diff = self._make_diff(old_text.strip(), new_text.strip(), "修改")
return diff
def _op_insert(self, path: str, rest: str) -> str:
"""在指定行插入内容"""
if not path:
return "错误:请提供文件路径"
file_path = self._normalize_path(path)
if not os.path.exists(file_path):
return f"错误:文件不存在: {file_path}"
parts = rest.split(maxsplit=1)
if len(parts) < 2:
return "错误:格式: insert <文件> <行号> <内容>"
try:
line_num = int(parts[0])
except ValueError:
return "错误:行号必须是整数"
content_to_insert = parts[1].strip()
with open(file_path, 'r', encoding='utf-8') as f:
lines = f.readlines()
if line_num < 1 or line_num > len(lines) + 1:
return f"错误:行号超出范围 (1-{len(lines)+1})"
insert_lines = [l + '\n' for l in content_to_insert.split('\n')]
lines[line_num-1:line_num-1] = insert_lines
with open(file_path, 'w', encoding='utf-8') as f:
f.writelines(lines)
return f"✅ 已在第 {line_num} 行插入 {len(insert_lines)} 行内容"
def _op_delete_lines(self, path: str, rest: str) -> str:
"""删除指定行范围"""
if not path:
return "错误:请提供文件路径"
file_path = self._normalize_path(path)
if not os.path.exists(file_path):
return f"错误:文件不存在: {file_path}"
parts = rest.split()
if len(parts) < 2:
return "错误:格式: delete_lines <文件> <起始行> <结束行>"
try:
start = int(parts[0])
end = int(parts[1])
except ValueError:
return "错误:行号必须是整数"
with open(file_path, 'r', encoding='utf-8') as f:
lines = f.readlines()
if start < 1 or end > len(lines) or start > end:
return f"错误:行号范围无效 (1-{len(lines)})"
deleted = lines[start-1:end]
lines[start-1:end] = []
with open(file_path, 'w', encoding='utf-8') as f:
f.writelines(lines)
return f"✅ 已删除第 {start}-{end} 行 ({len(deleted)} 行)"
def _op_read_range(self, path: str, rest: str) -> str:
"""读取指定行范围"""
if not path:
return "错误:请提供文件路径"
file_path = self._normalize_path(path)
if not os.path.exists(file_path):
return f"错误:文件不存在: {file_path}"
parts = rest.split()
start = int(parts[0]) if len(parts) > 0 else 1
end = int(parts[1]) if len(parts) > 1 else None
with open(file_path, 'r', encoding='utf-8') as f:
lines = f.readlines()
total = len(lines)
if end is None:
end = total
start = max(1, min(start, total))
end = max(start, min(end, total))
selected = lines[start-1:end]
numbered = [f"{start+i:4d} | {line.rstrip()}" for i, line in enumerate(selected)]
return f"📄 {file_path} (第 {start}-{end} 行,共 {total} 行)\n{'─'*60}\n" + '\n'.join(numbered)
def _op_search(self, path: str, rest: str) -> str:
"""在代码中搜索(类似 grep)"""
if not path:
return "错误:请提供搜索目录"
search_dir = self._normalize_path(path)
if not os.path.isdir(search_dir):
return f"错误:目录不存在: {search_dir}"
parts = rest.split(maxsplit=1)
if not parts:
return "错误:请提供搜索正则表达式"
pattern = parts[0]
file_pattern = parts[1] if len(parts) > 1 else "*.py"
try:
regex = re.compile(pattern)
except re.error as e:
return f"错误:无效的正则表达式: {e}"
# 构建文件匹配模式
if file_pattern.startswith("*."):
ext = file_pattern[1:] # .py
file_filter = lambda f: f.endswith(ext)
elif file_pattern == "*":
file_filter = lambda f: True
else:
file_filter = lambda f: file_pattern in f
results = []
max_results = 50
for root, dirs, files in os.walk(search_dir):
# 跳过隐藏目录和常见不需要搜索的目录
dirs[:] = [d for d in dirs if not d.startswith('.') and d not in
{'node_modules', '__pycache__', '.git', 'venv', '.venv', 'env'}]
for filename in files:
if not file_filter(filename):
continue
filepath = os.path.join(root, filename)
try:
with open(filepath, 'r', encoding='utf-8') as f:
for line_num, line in enumerate(f, 1):
if regex.search(line):
rel_path = os.path.relpath(filepath, search_dir)
results.append(f" {rel_path}:{line_num}: {line.rstrip()}")
if len(results) >= max_results:
break
except (UnicodeDecodeError, PermissionError):
continue
if len(results) >= max_results:
break
if len(results) >= max_results:
break
if not results:
return f"未找到匹配 '{pattern}' 的结果"
header = f"🔍 搜索结果 ({len(results)}{'+' if len(results) >= max_results else ''} 匹配):\n"
return header + '\n'.join(results)
def _op_ast_info(self, path: str, rest: str) -> str:
"""获取 Python 文件的 AST 摘要"""
import ast
if not path:
return "错误:请提供文件路径"
file_path = self._normalize_path(path)
if not os.path.exists(file_path):
return f"错误:文件不存在: {file_path}"
with open(file_path, 'r', encoding='utf-8') as f:
source = f.read()
try:
tree = ast.parse(source)
except SyntaxError as e:
return f"错误:语法错误 - {e}"
info = [f"📄 {file_path} AST 摘要:"]
for node in ast.iter_child_nodes(tree):
if isinstance(node, ast.ClassDef):
methods = [n.name for n in ast.iter_child_nodes(node)
if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef))]
info.append(f" class {node.name} (行 {node.lineno})")
for m in methods:
info.append(f" └─ {m}()")
elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
info.append(f" def {node.name}() (行 {node.lineno})")
elif isinstance(node, ast.Import):
names = [a.name for a in node.names]
info.append(f" import {', '.join(names)}")
elif isinstance(node, ast.ImportFrom):
names = [a.name for a in node.names]
info.append(f" from {node.module} import {', '.join(names)}")
if len(info) == 1:
return f"📄 {file_path}: 文件为空或无顶层定义"
return '\n'.join(info)
def _op_create(self, path: str, rest: str) -> str:
"""创建新文件"""
if not path:
return "错误:请提供文件路径"
file_path = self._normalize_path(path)
if os.path.exists(file_path):
return f"错误:文件已存在: {file_path}"
# 确保目录存在
dir_path = os.path.dirname(file_path)
if dir_path and not os.path.exists(dir_path):
os.makedirs(dir_path, exist_ok=True)
content = rest if rest else ""
with open(file_path, 'w', encoding='utf-8') as f:
f.write(content)
return f"✅ 已创建文件: {file_path} ({len(content)} 字符)"
def _op_write(self, path: str, rest: str) -> str:
"""写入/覆盖文件内容"""
if not path:
return "错误:请提供文件路径"
file_path = self._normalize_path(path)
# 确保目录存在
dir_path = os.path.dirname(file_path)
if dir_path and not os.path.exists(dir_path):
os.makedirs(dir_path, exist_ok=True)
with open(file_path, 'w', encoding='utf-8') as f:
f.write(rest)
return f"✅ 已写入文件: {file_path} ({len(rest)} 字符)"
def _op_append(self, path: str, rest: str) -> str:
"""追加内容到文件"""
if not path:
return "错误:请提供文件路径"
file_path = self._normalize_path(path)
if not os.path.exists(file_path):
return f"错误:文件不存在: {file_path}"
with open(file_path, 'a', encoding='utf-8') as f:
f.write(rest)
return f"✅ 已追加内容到: {file_path} ({len(rest)} 字符)"
def _make_diff(self, old_text: str, new_text: str, label: str) -> str:
"""生成 diff 预览"""
old_lines = old_text.splitlines(keepends=True)
new_lines = new_text.splitlines(keepends=True)
diff = difflib.unified_diff(old_lines, new_lines,
fromfile=f"原始", tofile=f"修改后", lineterm='')
diff_text = '\n'.join(diff)
if not diff_text:
return "内容无变化"
return f"📝 变更预览:\n{diff_text}"
def _fuzzy_match(self, lines: list, old_lines: list) -> Optional[Tuple[int, int]]:
"""模糊匹配,尝试找到最接近的位置"""
if not old_lines:
return None
best_ratio = 0
best_pos = None
for i in range(len(lines) - len(old_lines) + 1):
candidate = lines[i:i+len(old_lines)]
ratio = difflib.SequenceMatcher(None,
''.join(candidate), ''.join(old_lines)).ratio()
if ratio > best_ratio:
best_ratio = ratio
best_pos = (i, i + len(old_lines) - 1)
if best_ratio > 0.6:
return best_pos
return None
+472
View File
@@ -0,0 +1,472 @@
"""
代码搜索与理解插件 - 代码库级别的智能搜索
对标 Claude Code 的代码理解能力
"""
import os
import re
import ast
from typing import List, Dict, Optional, Tuple
class Plugin:
"""代码搜索与理解工具 - 跨文件搜索、依赖分析、代码结构理解"""
def __init__(self):
self.usage = """代码搜索与理解工具
操作类型:
find <目录> <关键词> [文件模式] - 搜索代码中的关键词
regex <目录> <正则表达式> [文件模式] - 正则表达式搜索
symbols <目录> [文件模式] - 提取所有符号(类、函数、变量)
imports <文件路径> - 分析文件的导入依赖
callers <目录> <函数名> - 查找调用指定函数的位置
todo <目录> [文件模式] - 搜索 TODO/FIXME/HACK 注释
structure <目录> - 显示项目目录结构
context <文件路径> [行号] [范围] - 获取某行的上下文
stats <目录> - 代码统计(行数、文件数等)
"""
self.cli = None
def set_cli(self, cli):
self.cli = cli
def get_tool_info(self):
return {
"name": "code_search",
"description": "代码搜索与理解工具 - 跨文件搜索、符号提取、依赖分析、项目结构查看",
"keywords": ["搜索", "查找", "grep", "search", "find", "代码", "符号", "结构"],
"usage": self.usage
}
def get_mcp_definition(self):
return {
"name": "code_search",
"description": "代码搜索与理解工具",
"inputSchema": {
"type": "object",
"properties": {
"operation": {"type": "string"},
"directory": {"type": "string"},
"pattern": {"type": "string"},
"file_pattern": {"type": "string"},
"file_path": {"type": "string"},
"line_number": {"type": "integer"},
"range": {"type": "integer"}
},
"required": ["operation"]
}
}
def convert_mcp_args(self, arguments):
op = arguments.get("operation", "")
directory = arguments.get("directory", "")
pattern = arguments.get("pattern", "")
return f"{op} {directory} {pattern}".strip()
def handle(self, args: str) -> str:
try:
parts = args.strip().split(maxsplit=2)
if not parts:
return "错误:请提供操作类型"
operation = parts[0].lower()
arg1 = parts[1] if len(parts) > 1 else ""
arg2 = parts[2] if len(parts) > 2 else ""
handlers = {
"find": self._op_find,
"regex": self._op_regex,
"symbols": self._op_symbols,
"imports": self._op_imports,
"callers": self._op_callers,
"todo": self._op_todo,
"structure": self._op_structure,
"context": self._op_context,
"stats": self._op_stats,
}
handler = handlers.get(operation)
if not handler:
return f"错误:不支持的操作 '{operation}'"
return handler(arg1, arg2)
except Exception as e:
return f"代码搜索错误: {str(e)}"
def _normalize_path(self, path: str) -> str:
path = path.strip().strip('"').strip("'")
path = os.path.expanduser(path)
return os.path.normpath(path)
def _walk_files(self, directory: str, file_pattern: str = "*.py",
max_files: int = 200) -> List[str]:
"""遍历目录中的文件"""
if file_pattern.startswith("*."):
ext = file_pattern[1:]
file_filter = lambda f: f.endswith(ext)
elif file_pattern == "*":
file_filter = lambda f: True
else:
file_filter = lambda f: file_pattern in f
files = []
skip_dirs = {'.git', 'node_modules', '__pycache__', '.venv', 'venv',
'env', '.env', '.tox', '.mypy_cache', '.pytest_cache'}
for root, dirs, filenames in os.walk(directory):
dirs[:] = [d for d in dirs if not d.startswith('.') and d not in skip_dirs]
for f in filenames:
if file_filter(f):
files.append(os.path.join(root, f))
if len(files) >= max_files:
return files
return files
def _op_find(self, directory: str, rest: str) -> str:
"""关键词搜索"""
if not directory:
return "错误:请提供搜索目录"
search_dir = self._normalize_path(directory)
parts = rest.split(maxsplit=1)
keyword = parts[0] if parts else ""
file_pattern = parts[1] if len(parts) > 1 else "*.py"
if not keyword:
return "错误:请提供搜索关键词"
if not os.path.isdir(search_dir):
return f"错误:目录不存在: {search_dir}"
files = self._walk_files(search_dir, file_pattern)
results = []
max_results = 80
for filepath in files:
try:
with open(filepath, 'r', encoding='utf-8') as f:
for line_num, line in enumerate(f, 1):
if keyword.lower() in line.lower():
rel_path = os.path.relpath(filepath, search_dir)
results.append(f" {rel_path}:{line_num}: {line.rstrip()}")
if len(results) >= max_results:
break
except (UnicodeDecodeError, PermissionError):
continue
if len(results) >= max_results:
break
if not results:
return f"未找到 '{keyword}' 的匹配"
return f"🔍 搜索 '{keyword}' ({len(results)}{'+' if len(results) >= max_results else ''} 匹配):\n" + '\n'.join(results)
def _op_regex(self, directory: str, rest: str) -> str:
"""正则表达式搜索"""
if not directory:
return "错误:请提供搜索目录"
search_dir = self._normalize_path(directory)
parts = rest.split(maxsplit=1)
pattern = parts[0] if parts else ""
file_pattern = parts[1] if len(parts) > 1 else "*.py"
if not pattern:
return "错误:请提供正则表达式"
try:
regex = re.compile(pattern)
except re.error as e:
return f"错误:无效的正则表达式: {e}"
files = self._walk_files(search_dir, file_pattern)
results = []
max_results = 50
for filepath in files:
try:
with open(filepath, 'r', encoding='utf-8') as f:
for line_num, line in enumerate(f, 1):
if regex.search(line):
rel_path = os.path.relpath(filepath, search_dir)
results.append(f" {rel_path}:{line_num}: {line.rstrip()}")
if len(results) >= max_results:
break
except (UnicodeDecodeError, PermissionError):
continue
if len(results) >= max_results:
break
if not results:
return f"未找到匹配 '{pattern}' 的结果"
return f"🔍 正则搜索 ({len(results)}{'+' if len(results) >= max_results else ''} 匹配):\n" + '\n'.join(results)
def _op_symbols(self, directory: str, rest: str) -> str:
"""提取目录中所有 Python 符号"""
if not directory:
return "错误:请提供目录路径"
search_dir = self._normalize_path(directory)
file_pattern = rest.strip() or "*.py"
files = self._walk_files(search_dir, file_pattern)
all_symbols = []
for filepath in files:
if not filepath.endswith('.py'):
continue
try:
with open(filepath, 'r', encoding='utf-8') as f:
source = f.read()
tree = ast.parse(source)
rel_path = os.path.relpath(filepath, search_dir)
symbols = []
for node in ast.iter_child_nodes(tree):
if isinstance(node, ast.ClassDef):
methods = [n.name for n in ast.iter_child_nodes(node)
if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef))]
symbols.append(f" class {node.name} (行 {node.lineno})")
for m in methods[:10]: # 限制每个类最多显示10个方法
symbols.append(f" └─ {m}()")
elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
symbols.append(f" def {node.name}() (行 {node.lineno})")
if symbols:
all_symbols.append(f"\n📄 {rel_path}:")
all_symbols.extend(symbols)
except (SyntaxError, UnicodeDecodeError):
continue
if not all_symbols:
return "未找到 Python 符号"
return f"📋 代码符号:\n" + '\n'.join(all_symbols)
def _op_imports(self, file_path: str, rest: str) -> str:
"""分析文件的导入依赖"""
if not file_path:
return "错误:请提供文件路径"
path = self._normalize_path(file_path)
if not os.path.exists(path):
return f"错误:文件不存在: {path}"
with open(path, 'r', encoding='utf-8') as f:
source = f.read()
try:
tree = ast.parse(source)
except SyntaxError as e:
return f"错误:语法错误 - {e}"
imports = []
for node in ast.iter_child_nodes(tree):
if isinstance(node, ast.Import):
for alias in node.names:
imports.append(f" import {alias.name} (行 {node.lineno})")
elif isinstance(node, ast.ImportFrom):
names = [a.name for a in node.names]
imports.append(f" from {node.module} import {', '.join(names)} (行 {node.lineno})")
if not imports:
return f"📄 {file_path}: 没有导入语句"
return f"📦 {file_path} 导入依赖:\n" + '\n'.join(imports)
def _op_callers(self, directory: str, rest: str) -> str:
"""查找调用指定函数的位置"""
if not directory:
return "错误:请提供搜索目录"
search_dir = self._normalize_path(directory)
parts = rest.split(maxsplit=1)
func_name = parts[0] if parts else ""
file_pattern = parts[1] if len(parts) > 1 else "*.py"
if not func_name:
return "错误:请提供函数名"
# 搜索函数调用模式
patterns = [
re.compile(rf'\b{re.escape(func_name)}\s*\('),
re.compile(rf'\.{re.escape(func_name)}\s*\('),
]
files = self._walk_files(search_dir, file_pattern)
results = []
for filepath in files:
try:
with open(filepath, 'r', encoding='utf-8') as f:
for line_num, line in enumerate(f, 1):
for pat in patterns:
if pat.search(line):
rel_path = os.path.relpath(filepath, search_dir)
results.append(f" {rel_path}:{line_num}: {line.rstrip()}")
break
except (UnicodeDecodeError, PermissionError):
continue
if not results:
return f"未找到调用 '{func_name}' 的位置"
return f"🔎 '{func_name}' 的调用位置 ({len(results)}):\n" + '\n'.join(results)
def _op_todo(self, directory: str, rest: str) -> str:
"""搜索 TODO/FIXME/HACK 注释"""
if not directory:
return "错误:请提供目录路径"
search_dir = self._normalize_path(directory)
file_pattern = rest.strip() or "*.py"
pattern = re.compile(r'#\s*(TODO|FIXME|HACK|XXX|BUG|NOTE)\b[:\s]*(.*)', re.IGNORECASE)
files = self._walk_files(search_dir, file_pattern)
results = []
for filepath in files:
try:
with open(filepath, 'r', encoding='utf-8') as f:
for line_num, line in enumerate(f, 1):
match = pattern.search(line)
if match:
rel_path = os.path.relpath(filepath, search_dir)
tag = match.group(1).upper()
desc = match.group(2).strip()
results.append(f" {rel_path}:{line_num} [{tag}] {desc}")
except (UnicodeDecodeError, PermissionError):
continue
if not results:
return "没有找到 TODO/FIXME 注释"
return f"📝 TODO/FIXME ({len(results)}):\n" + '\n'.join(results)
def _op_structure(self, directory: str, rest: str) -> str:
"""显示项目目录结构"""
if not directory:
return "错误:请提供目录路径"
search_dir = self._normalize_path(directory)
if not os.path.isdir(search_dir):
return f"错误:目录不存在: {search_dir}"
max_depth = int(rest) if rest.strip().isdigit() else 3
lines = [f"📁 {os.path.basename(search_dir)}/"]
skip_dirs = {'.git', 'node_modules', '__pycache__', '.venv', 'venv',
'env', '.tox', '.mypy_cache'}
def _tree(path, prefix="", depth=0):
if depth >= max_depth:
return
try:
entries = sorted(os.listdir(path))
except PermissionError:
return
dirs = [e for e in entries if os.path.isdir(os.path.join(path, e))
and e not in skip_dirs and not e.startswith('.')]
files = [e for e in entries if os.path.isfile(os.path.join(path, e))]
# 限制每个目录显示的文件数
if len(files) > 20:
files = files[:20] + [f"... ({len(files)-20} more)"]
for i, d in enumerate(dirs):
is_last_dir = (i == len(dirs) - 1) and not files
connector = "└── " if is_last_dir else "├── "
lines.append(f"{prefix}{connector}{d}/")
extension = " " if is_last_dir else "│ "
_tree(os.path.join(path, d), prefix + extension, depth + 1)
for i, f in enumerate(files):
is_last = i == len(files) - 1
connector = "└── " if is_last else "├── "
# 文件大小
fpath = os.path.join(path, f)
if os.path.isfile(fpath):
size = os.path.getsize(fpath)
if size > 1024 * 1024:
size_str = f" ({size // (1024*1024)}MB)"
elif size > 1024:
size_str = f" ({size // 1024}KB)"
else:
size_str = f" ({size}B)"
else:
size_str = ""
lines.append(f"{prefix}{connector}{f}{size_str}")
_tree(search_dir)
return '\n'.join(lines)
def _op_context(self, file_path: str, rest: str) -> str:
"""获取某行的上下文"""
if not file_path:
return "错误:请提供文件路径"
path = self._normalize_path(file_path)
if not os.path.exists(path):
return f"错误:文件不存在: {path}"
parts = rest.split()
line_num = int(parts[0]) if parts else 1
context_range = int(parts[1]) if len(parts) > 1 else 5
with open(path, 'r', encoding='utf-8') as f:
lines = f.readlines()
total = len(lines)
start = max(1, line_num - context_range)
end = min(total, line_num + context_range)
result = [f"📄 {path} 第 {start}-{end} 行 (目标: 第 {line_num} 行):"]
result.append("─" * 60)
for i in range(start - 1, end):
marker = ">>>" if i + 1 == line_num else " "
result.append(f"{marker} {i+1:4d} | {lines[i].rstrip()}")
return '\n'.join(result)
def _op_stats(self, directory: str, rest: str) -> str:
"""代码统计"""
if not directory:
return "错误:请提供目录路径"
search_dir = self._normalize_path(directory)
files = self._walk_files(search_dir, "*.py")
total_lines = 0
total_blank = 0
total_comment = 0
total_code = 0
file_count = len(files)
for filepath in files:
try:
with open(filepath, 'r', encoding='utf-8') as f:
for line in f:
total_lines += 1
stripped = line.strip()
if not stripped:
total_blank += 1
elif stripped.startswith('#'):
total_comment += 1
else:
total_code += 1
except (UnicodeDecodeError, PermissionError):
continue
return f"""📊 代码统计 ({search_dir}):
文件数: {file_count}
总行数: {total_lines}
代码行: {total_code}
注释行: {total_comment}
空行: {total_blank}
注释率: {total_comment/max(total_code,1)*100:.1f}%"""
+793
View File
@@ -0,0 +1,793 @@
"""
文件管理器插件 - 小狸自有协议版本
支持文件和目录的创建、读取、写入、删除、移动、复制、搜索等操作
"""
import os
import json
import shutil
from datetime import datetime
from pathlib import Path
from typing import Optional, Dict, Any, List
class Liugin:
"""文件管理器插件"""
def __init__(self):
self.usage = """文件管理器使用方法:
file_manager <操作> <参数>
操作类型:
- list <路径> [-p "模式"] - 列出目录内容(只列当前目录)
- read <文件路径> [-n 行数] - 读取文件内容
- write <文件路径> -c "内容" - 写入文件
- append <文件路径> -c "内容" - 追加内容到文件
- copy <源路径> <目标路径> - 复制文件或目录
- move <源路径> <目标路径> - 移动或重命名
- delete <路径> [-r] - 删除文件或目录
- search <目录> [-p "模式"] [-c "内容"] - 搜索文件(递归)
- info <路径> - 获取文件信息
- mkdir <路径> - 创建目录
- modify <文件路径> <行号> "新内容" - 修改指定行
- insert <文件路径> <行号> "内容" - 在指定行插入内容
- delete-line <文件路径> <行号> - 删除指定行
- replace <文件路径> "旧文本" "新文本" [--start 行号] [--end 行号] - 替换文本
示例:
- file_manager list . - 列出当前目录
- file_manager list . -p "*.py" - 列出所有Python文件
- file_manager read config.json - 读取文件
- file_manager write test.txt -c "Hello World" - 写入文件
- file_manager copy file1.txt file2.txt - 复制文件
- file_manager move old.txt new.txt - 重命名文件
- file_manager delete temp -r - 递归删除目录
- file_manager search . -c "TODO" - 搜索包含TODO的文件
JSON调用格式:
{"action": "use_tool", "tool": "file_manager", "args": "list ."}
{"action": "use_tool", "tool": "file_manager", "args": "read config.json -n 50"}
{"action": "use_tool", "tool": "file_manager", "args": "write test.txt -c Hello World"}
"""
self.cli = None
def set_cli(self, cli):
"""设置CLI实例引用"""
self.cli = cli
def get_tool_info(self):
return {
"name": "file_manager",
"description": "文件和目录管理工具,支持创建、读取、写入、删除、移动、复制、搜索文件等操作",
"keywords": ["文件", "目录", "管理", "file", "manager", "读写", "复制", "删除", "搜索", "mkdir"],
"usage": self.usage
}
def get_mcp_definition(self) -> Dict[str, Any]:
"""返回符合 MCP 标准的工具定义"""
return {
"name": "file_manager",
"description": "文件和目录管理工具,支持创建、读取、写入、删除、移动、复制、搜索文件等操作。操作类型: list(列出目录), read(读取文件), write(写入文件), append(追加内容), copy(复制), move(移动/重命名), delete(删除), search(搜索), info(文件信息), mkdir(创建目录), modify(修改行), insert(插入行), delete-line(删除行), replace(替换文本)",
"inputSchema": {
"type": "object",
"properties": {
"operation": {
"type": "string",
"enum": ["list", "read", "write", "append", "copy", "move", "delete", "search", "info", "mkdir", "modify", "insert", "delete-line", "replace"],
"description": "操作类型"
},
"path": {
"type": "string",
"description": "文件或目录路径"
},
"content": {
"type": "string",
"description": "文件内容(用于 write/append 操作)"
},
"destination": {
"type": "string",
"description": "目标路径(用于 copy/move 操作)"
},
"pattern": {
"type": "string",
"description": "文件匹配模式(用于 list/search 操作)"
},
"line_number": {
"type": "integer",
"description": "行号(用于 modify/insert/delete-line 操作)"
},
"old_text": {
"type": "string",
"description": "旧文本(用于 replace 操作)"
},
"new_text": {
"type": "string",
"description": "新文本(用于 replace/modify 操作)"
},
"recursive": {
"type": "boolean",
"description": "是否递归(用于 delete/search 操作)",
"default": False
}
},
"required": ["operation"]
}
}
def convert_mcp_args(self, arguments: Dict[str, Any]) -> str:
"""将 MCP 参数转换为命令行格式"""
op = arguments.get("operation", "")
parts = [op]
if "path" in arguments:
parts.append(arguments["path"])
if "destination" in arguments:
parts.append(arguments["destination"])
if "content" in arguments:
parts.extend(["-c", f'"{arguments["content"]}"'])
if "pattern" in arguments:
parts.extend(["-p", f'"{arguments["pattern"]}"'])
if "line_number" in arguments:
parts.append(str(arguments["line_number"]))
if "old_text" in arguments:
parts.append(f'"{arguments["old_text"]}"')
if "new_text" in arguments:
parts.append(f'"{arguments["new_text"]}"')
if arguments.get("recursive"):
parts.append("-r")
return " ".join(parts)
def handle(self, args: str) -> str:
"""处理文件管理请求"""
try:
# 解析参数
parts = self._parse_args(args)
if not parts:
return "错误:请提供操作类型。\n可用操作: list, read, write, append, copy, move, delete, search, info, mkdir, modify, insert, delete-line, replace"
operation = parts[0].lower()
handlers = {
"list": self._op_list,
"read": self._op_read,
"write": self._op_write,
"append": self._op_append,
"copy": self._op_copy,
"move": self._op_move,
"delete": self._op_delete,
"search": self._op_search,
"info": self._op_info,
"mkdir": self._op_mkdir,
"modify": self._op_modify_line,
"insert": self._op_insert_line,
"delete-line": self._op_delete_line,
"replace": self._op_replace_text,
}
handler = handlers.get(operation)
if not handler:
return f"错误:不支持的操作 '{operation}'。\n支持的操作: {', '.join(handlers.keys())}"
return handler(parts[1:])
except Exception as e:
return f"文件操作错误: {str(e)}"
def _parse_args(self, args: str) -> List[str]:
"""解析命令行参数,支持引号"""
parts = []
current = ""
in_quotes = False
quote_char = None
for char in args:
if char in ('"', "'") and not in_quotes:
in_quotes = True
quote_char = char
elif char == quote_char and in_quotes:
in_quotes = False
quote_char = None
elif char == ' ' and not in_quotes:
if current:
parts.append(current)
current = ""
else:
current += char
if current:
parts.append(current)
return parts
def _op_list(self, args: List[str]) -> str:
"""列出目录内容 - 默认只列出当前目录,不递归"""
path = "."
pattern = None
i = 0
while i < len(args):
if args[i] == "-p" and i + 1 < len(args):
pattern = args[i + 1]
i += 1
elif not args[i].startswith("-"):
path = args[i]
i += 1
path = os.path.abspath(path)
if not os.path.exists(path):
return f"错误:路径不存在: {path}"
if not os.path.isdir(path):
return f"错误:不是目录: {path}"
items = []
try:
for item in os.listdir(path):
full_path = os.path.join(path, item)
if pattern and not Path(item).match(pattern):
continue
if os.path.isdir(full_path):
items.append(('dir', item, ''))
else:
size = os.path.getsize(full_path)
if size < 1024:
size_str = f"{size}B"
elif size < 1024 * 1024:
size_str = f"{size/1024:.1f}KB"
else:
size_str = f"{size/1024/1024:.1f}MB"
items.append(('file', item, size_str))
# 排序:文件夹优先
items.sort(key=lambda x: (0 if x[0] == 'dir' else 1, x[1].lower()))
result = [f"📁 {path}"]
result.append("─" * 50)
dir_count = sum(1 for x in items if x[0] == 'dir')
file_count = len(items) - dir_count
for item_type, name, size in items[:100]:
if item_type == 'dir':
result.append(f" 📁 {name}/")
else:
result.append(f" 📄 {name} ({size})")
if len(items) > 100:
result.append(f" ... 还有 {len(items) - 100} 项")
result.append("─" * 50)
result.append(f"共 {dir_count} 个文件夹, {file_count} 个文件")
return '\n'.join(result)
except Exception as e:
return f"列出目录失败: {e}"
def _op_read(self, args: List[str]) -> str:
"""读取文件内容"""
if not args:
return "错误:请提供文件路径"
file_path = args[0]
lines = None
i = 1
while i < len(args):
if args[i] == "-n" and i + 1 < len(args):
try:
lines = int(args[i + 1])
except ValueError:
return "错误:行数必须是整数"
i += 1
i += 1
file_path = os.path.abspath(file_path)
if not os.path.exists(file_path):
return f"错误:文件不存在: {file_path}"
if not os.path.isfile(file_path):
return f"错误:不是文件: {file_path}"
file_size = os.path.getsize(file_path)
if file_size > 10 * 1024 * 1024: # 10MB
return f"警告:文件过大 ({file_size / 1024 / 1024:.1f}MB),请使用 -n 参数限制行数"
try:
with open(file_path, 'r', encoding='utf-8') as f:
if lines:
content_lines = []
line_count = 0
for i, line in enumerate(f):
line_count = i + 1
if i < lines:
content_lines.append(f"{i+1:4d} | {line.rstrip()}")
content = '\n'.join(content_lines)
total_lines = line_count
else:
all_lines = f.readlines()
content = '\n'.join(f"{i+1:4d} | {line.rstrip()}" for i, line in enumerate(all_lines))
total_lines = len(all_lines)
result = [f"文件: {file_path}"]
result.append(f"大小: {file_size} 字节")
result.append(f"行数: {total_lines}")
if lines:
result.append(f"显示: 前 {min(lines, total_lines)} 行")
result.append(f"\n{'─' * 50}")
result.append(content)
return '\n'.join(result)
except UnicodeDecodeError:
return f"错误:无法解码文件(可能是二进制文件或编码不正确)"
except Exception as e:
return f"读取文件失败: {e}"
def _op_write(self, args: List[str]) -> str:
"""写入文件"""
if len(args) < 2:
return "错误:请提供文件路径和内容"
file_path = args[0]
content = None
i = 1
while i < len(args):
if args[i] == "-c" and i + 1 < len(args):
content = args[i + 1]
i += 1
i += 1
if content is None:
return "错误:请使用 -c 参数提供内容"
file_path = os.path.abspath(file_path)
try:
# 创建父目录
parent = os.path.dirname(file_path)
if parent and not os.path.exists(parent):
os.makedirs(parent, exist_ok=True)
with open(file_path, 'w', encoding='utf-8') as f:
f.write(content)
return f"✅ 文件已写入: {file_path}\n大小: {len(content.encode('utf-8'))} 字节"
except Exception as e:
return f"写入文件失败: {e}"
def _op_append(self, args: List[str]) -> str:
"""追加内容到文件"""
if len(args) < 2:
return "错误:请提供文件路径和内容"
file_path = args[0]
content = None
i = 1
while i < len(args):
if args[i] == "-c" and i + 1 < len(args):
content = args[i + 1]
i += 1
i += 1
if content is None:
return "错误:请使用 -c 参数提供内容"
file_path = os.path.abspath(file_path)
try:
with open(file_path, 'a', encoding='utf-8') as f:
f.write(content)
return f"✅ 内容已追加: {file_path}"
except Exception as e:
return f"追加内容失败: {e}"
def _op_copy(self, args: List[str]) -> str:
"""复制文件或目录"""
if len(args) < 2:
return "错误:请提供源路径和目标路径"
source = os.path.abspath(args[0])
destination = os.path.abspath(args[1])
if not os.path.exists(source):
return f"错误:源路径不存在: {source}"
try:
parent = os.path.dirname(destination)
if parent and not os.path.exists(parent):
os.makedirs(parent, exist_ok=True)
if os.path.isfile(source):
shutil.copy2(source, destination)
item_type = "文件"
else:
shutil.copytree(source, destination, dirs_exist_ok=True)
item_type = "目录"
return f"✅ {item_type}已复制:\n 源: {source}\n 目标: {destination}"
except Exception as e:
return f"复制失败: {e}"
def _op_move(self, args: List[str]) -> str:
"""移动或重命名文件或目录"""
if len(args) < 2:
return "错误:请提供源路径和目标路径"
source = os.path.abspath(args[0])
destination = os.path.abspath(args[1])
if not os.path.exists(source):
return f"错误:源路径不存在: {source}"
try:
parent = os.path.dirname(destination)
if parent and not os.path.exists(parent):
os.makedirs(parent, exist_ok=True)
shutil.move(source, destination)
return f"✅ 已移动/重命名:\n 源: {source}\n 目标: {destination}"
except Exception as e:
return f"移动失败: {e}"
def _op_delete(self, args: List[str]) -> str:
"""删除文件或目录"""
if not args:
return "错误:请提供路径"
path = os.path.abspath(args[0])
recursive = "-r" in args
if not os.path.exists(path):
return f"错误:路径不存在: {path}"
try:
if os.path.isfile(path):
os.remove(path)
item_type = "文件"
elif os.path.isdir(path):
if recursive:
shutil.rmtree(path)
item_type = "目录"
else:
# 检查是否为空
if os.listdir(path):
return f"错误:目录不为空。使用 -r 参数递归删除。"
os.rmdir(path)
item_type = "空目录"
return f"✅ 已删除 {item_type}: {path}"
except Exception as e:
return f"删除失败: {e}"
def _op_search(self, args: List[str]) -> str:
"""搜索文件"""
if not args:
return "错误:请提供搜索目录"
directory = args[0]
pattern = None
content_search = None
i = 1
while i < len(args):
if args[i] == "-p" and i + 1 < len(args):
pattern = args[i + 1]
i += 1
elif args[i] == "-c" and i + 1 < len(args):
content_search = args[i + 1]
i += 1
i += 1
directory = os.path.abspath(directory)
if not os.path.exists(directory):
return f"错误:目录不存在: {directory}"
results = []
try:
for root, dirs, files in os.walk(directory):
for filename in files:
file_path = os.path.join(root, filename)
rel_path = os.path.relpath(file_path, directory)
# 模式匹配
if pattern and not Path(filename).match(pattern):
continue
# 内容搜索
if content_search:
try:
with open(file_path, 'r', encoding='utf-8') as f:
file_content = f.read()
if content_search not in file_content:
continue
# 找到匹配行
for i, line in enumerate(file_content.split('\n'), 1):
if content_search in line:
results.append(f"{rel_path}:{i}: {line.strip()[:80]}")
if len(results) >= 50:
break
except (UnicodeDecodeError, IOError):
continue
else:
results.append(rel_path)
if len(results) >= 100:
break
if len(results) >= 100:
break
if not results:
return "未找到匹配的文件"
result_text = [f"搜索结果 (共 {len(results)} 个):"]
result_text.append(f"目录: {directory}")
if pattern:
result_text.append(f"模式: {pattern}")
if content_search:
result_text.append(f"内容: {content_search}")
result_text.append("")
result_text.extend(results[:100])
if len(results) > 100:
result_text.append(f"... 还有 {len(results) - 100} 个结果")
return '\n'.join(result_text)
except Exception as e:
return f"搜索失败: {e}"
def _op_info(self, args: List[str]) -> str:
"""获取文件信息"""
if not args:
return "错误:请提供路径"
path = os.path.abspath(args[0])
if not os.path.exists(path):
return f"错误:路径不存在: {path}"
try:
stat_info = os.stat(path)
result = [f"路径: {path}"]
result.append(f"名称: {os.path.basename(path)}")
result.append(f"类型: {'目录' if os.path.isdir(path) else '文件'}")
result.append(f"大小: {stat_info.st_size:,} 字节 ({stat_info.st_size / 1024:.1f} KB)")
result.append(f"创建时间: {datetime.fromtimestamp(stat_info.st_ctime).strftime('%Y-%m-%d %H:%M:%S')}")
result.append(f"修改时间: {datetime.fromtimestamp(stat_info.st_mtime).strftime('%Y-%m-%d %H:%M:%S')}")
result.append(f"访问时间: {datetime.fromtimestamp(stat_info.st_atime).strftime('%Y-%m-%d %H:%M:%S')}")
if os.path.isfile(path):
ext = os.path.splitext(path)[1]
result.append(f"扩展名: {ext if ext else '无'}")
# 尝试检测编码
try:
with open(path, 'r', encoding='utf-8') as f:
f.read(1024)
result.append("编码: UTF-8 (文本文件)")
except UnicodeDecodeError:
result.append("编码: 二进制文件")
return '\n'.join(result)
except Exception as e:
return f"获取信息失败: {e}"
def _op_mkdir(self, args: List[str]) -> str:
"""创建目录"""
if not args:
return "错误:请提供目录路径"
path = os.path.abspath(args[0])
if os.path.exists(path):
return f"错误:路径已存在: {path}"
try:
os.makedirs(path, exist_ok=True)
return f"✅ 目录已创建: {path}"
except Exception as e:
return f"创建目录失败: {e}"
def _op_modify_line(self, args: List[str]) -> str:
"""修改指定行"""
if len(args) < 3:
return "错误:请提供文件路径、行号和新内容"
file_path = os.path.abspath(args[0])
try:
line_number = int(args[1])
except ValueError:
return "错误:行号必须是整数"
new_content = args[2]
if not os.path.exists(file_path):
return f"错误:文件不存在: {file_path}"
if line_number < 1:
return "错误:行号必须 >= 1"
try:
with open(file_path, 'r', encoding='utf-8') as f:
lines = f.readlines()
if line_number > len(lines):
return f"错误:文件只有 {len(lines)} 行,无法修改第 {line_number} 行"
old_content = lines[line_number - 1].rstrip('\n\r')
lines[line_number - 1] = new_content + '\n'
with open(file_path, 'w', encoding='utf-8') as f:
f.writelines(lines)
return f"✅ 第 {line_number} 行已修改:\n 旧: {old_content}\n 新: {new_content}"
except Exception as e:
return f"修改行失败: {e}"
def _op_insert_line(self, args: List[str]) -> str:
"""插入行"""
if len(args) < 3:
return "错误:请提供文件路径、行号和内容"
file_path = os.path.abspath(args[0])
try:
line_number = int(args[1])
except ValueError:
return "错误:行号必须是整数"
content = args[2]
if not os.path.exists(file_path):
return f"错误:文件不存在: {file_path}"
if line_number < 1:
return "错误:行号必须 >= 1"
try:
with open(file_path, 'r', encoding='utf-8') as f:
lines = f.readlines()
insert_pos = min(line_number - 1, len(lines))
lines.insert(insert_pos, content + '\n')
with open(file_path, 'w', encoding='utf-8') as f:
f.writelines(lines)
return f"✅ 已在第 {line_number} 行插入内容: {content}"
except Exception as e:
return f"插入行失败: {e}"
def _op_delete_line(self, args: List[str]) -> str:
"""删除指定行"""
if len(args) < 2:
return "错误:请提供文件路径和行号"
file_path = os.path.abspath(args[0])
try:
line_number = int(args[1])
except ValueError:
return "错误:行号必须是整数"
if not os.path.exists(file_path):
return f"错误:文件不存在: {file_path}"
if line_number < 1:
return "错误:行号必须 >= 1"
try:
with open(file_path, 'r', encoding='utf-8') as f:
lines = f.readlines()
if line_number > len(lines):
return f"错误:文件只有 {len(lines)} 行,无法删除第 {line_number} 行"
deleted = lines[line_number - 1].rstrip('\n\r')
del lines[line_number - 1]
with open(file_path, 'w', encoding='utf-8') as f:
f.writelines(lines)
return f"✅ 已删除第 {line_number} 行: {deleted}"
except Exception as e:
return f"删除行失败: {e}"
def _op_replace_text(self, args: List[str]) -> str:
"""替换文本"""
if len(args) < 3:
return "错误:请提供文件路径、旧文本和新文本"
file_path = os.path.abspath(args[0])
old_text = args[1]
new_text = args[2]
start_line = None
end_line = None
i = 3
while i < len(args):
if args[i] == "--start" and i + 1 < len(args):
start_line = int(args[i + 1])
i += 1
elif args[i] == "--end" and i + 1 < len(args):
end_line = int(args[i + 1])
i += 1
i += 1
if not os.path.exists(file_path):
return f"错误:文件不存在: {file_path}"
try:
with open(file_path, 'r', encoding='utf-8') as f:
lines = f.readlines()
total_lines = len(lines)
replacements = []
start = max(0, (start_line or 1) - 1)
end = min(total_lines, end_line or total_lines)
for i in range(start, end):
if old_text in lines[i]:
count = lines[i].count(old_text)
lines[i] = lines[i].replace(old_text, new_text)
replacements.append(f"第 {i+1} 行: 替换了 {count} 处")
if not replacements:
return f"未找到匹配文本: {old_text}"
with open(file_path, 'w', encoding='utf-8') as f:
f.writelines(lines)
result = [f"✅ 文本替换完成"]
result.append(f"文件: {file_path}")
result.append(f"'{old_text}' -> '{new_text}'")
if start_line or end_line:
result.append(f"范围: 第 {start+1} 行到第 {end} 行")
result.append(f"修改了 {len(replacements)} 行:")
result.extend(replacements[:20])
if len(replacements) > 20:
result.append(f"... 还有 {len(replacements) - 20} 行")
return '\n'.join(result)
except Exception as e:
return f"文本替换失败: {e}"
# 测试
if __name__ == "__main__":
plugin = Plugin()
print("文件管理器插件测试:")
print(plugin.get_tool_info())
print("\n测试 list 操作:")
print(plugin.handle("list . -p *.py"))
+452
View File
@@ -0,0 +1,452 @@
"""
前端测试验证插件 - 类似 iFlow CLI 的 frontend-tester
支持HTML/CSS/JS自动化测试和UI验证
"""
import os
import subprocess
from colorama import Fore, Style
class Liugin:
"""前端测试验证插件 - 自动化测试Web界面"""
def __init__(self):
self.usage = """前端测试工具使用方法:
JSON格式示例:
{"action": "use_tool", "tool": "frontend_tester", "args": "validate index.html"} - 验证HTML文件
{"action": "use_tool", "tool": "frontend_tester", "args": "test style.css"} - 测试CSS文件
{"action": "use_tool", "tool": "frontend_tester", "args": "check app.js"} - 检查JS文件
{"action": "use_tool", "tool": "frontend_tester", "args": "preview index.html"} - 在浏览器中预览
{"action": "use_tool", "tool": "frontend_tester", "args": "analyze project"} - 分析整个项目
功能说明:
- validate <HTML文件路径> - 验证HTML文件的语法和结构
- test <CSS文件路径> - 测试CSS文件的样式和兼容性
- check <JS文件路径> - 检查JS文件的语法和潜在问题
- preview <HTML文件路径> - 在默认浏览器中打开预览
- analyze <项目路径> - 分析整个前端项目的结构和问题
支持的检查项:
✓ HTML语法验证
✓ CSS语法和兼容性检查
✓ JavaScript语法检查
✓ 响应式设计检查
✓ 可访问性检查
✓ 性能优化建议"""
self.cli = None
def set_cli(self, cli):
"""设置CLI实例引用"""
self.cli = cli
# 注册插件命令
self.cli.register_liugin_command('ftest', self.command_handler)
self.cli.register_liugin_command('frontend', self.command_handler)
def command_handler(self, args):
"""处理 /ftest 或 /frontend 命令"""
return self.handle(args)
def get_tool_info(self):
return {
"name": "frontend_tester",
"description": "前端测试验证工具,支持HTML/CSS/JS自动化测试和UI验证,包括语法检查、响应式设计、可访问性检查等",
"keywords": ["前端", "测试", "HTML", "CSS", "JavaScript", "UI", "验证", "preview", "validate", "frontend", "test"],
"usage": self.usage
}
def handle(self, args):
"""处理前端测试请求"""
try:
parts = args.strip().split(maxsplit=1)
if not parts:
return "错误:请提供操作类型和参数。使用 'frontend_tester help' 查看帮助"
operation = parts[0].lower()
if operation == "help":
return self.usage
elif operation == "validate":
# 验证HTML文件
if len(parts) < 2:
return "错误:请提供HTML文件路径。格式: validate <HTML文件路径>"
file_path = parts[1]
return self._validate_html(file_path)
elif operation == "test":
# 测试CSS文件
if len(parts) < 2:
return "错误:请提供CSS文件路径。格式: test <CSS文件路径>"
file_path = parts[1]
return self._test_css(file_path)
elif operation == "check":
# 检查JS文件
if len(parts) < 2:
return "错误:请提供JS文件路径。格式: check <JS文件路径>"
file_path = parts[1]
return self._check_js(file_path)
elif operation == "preview":
# 在浏览器中预览
if len(parts) < 2:
return "错误:请提供HTML文件路径。格式: preview <HTML文件路径>"
file_path = parts[1]
return self._preview_in_browser(file_path)
elif operation == "analyze":
# 分析整个项目
if len(parts) < 2:
return "错误:请提供项目路径。格式: analyze <项目路径>"
project_path = parts[1]
return self._analyze_project(project_path)
else:
return f"错误:不支持的操作 '{operation}'。支持的操作有: validate, test, check, preview, analyze, help"
except Exception as e:
return f"前端测试操作错误: {str(e)}"
def _validate_html(self, file_path):
"""验证HTML文件"""
if not os.path.exists(file_path):
return f"❌ 文件不存在: {file_path}"
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
result = f"\n🧪 HTML验证结果: {file_path}\n"
result += "=" * 60 + "\n\n"
# 基本检查
issues = []
# 检查DOCTYPE
if not content.strip().startswith('<!DOCTYPE'):
issues.append({
"level": "warning",
"message": "缺少DOCTYPE声明,建议添加 <!DOCTYPE html>"
})
# 检查lang属性
if '<html' in content and 'lang=' not in content:
issues.append({
"level": "warning",
"message": "建议在<html>标签中添加lang属性,如 lang=\"zh-CN\""
})
# 检查meta标签
if '<meta charset=' not in content:
issues.append({
"level": "error",
"message": "缺少字符集meta标签,建议添加 <meta charset=\"UTF-8\">"
})
if '<meta name="viewport"' not in content:
issues.append({
"level": "warning",
"message": "缺少viewport meta标签,建议添加以支持响应式设计"
})
# 检查title标签
if '<title>' not in content:
issues.append({
"level": "error",
"message": "缺少<title>标签"
})
# 统计元素
div_count = content.count('<div')
img_count = content.count('<img')
img_alt_count = content.count('alt=')
link_count = content.count('<link')
script_count = content.count('<script')
result += f"📊 文件统计:\n"
result += f" - div元素: {div_count}\n"
result += f" - img元素: {img_count}\n"
result += f" - img alt属性: {img_alt_count}\n"
result += f" - link标签: {link_count}\n"
result += f" - script标签: {script_count}\n\n"
# 检查img alt属性
if img_count > 0 and img_alt_count < img_count:
issues.append({
"level": "warning",
"message": f"有 {img_count - img_alt_count} 个img元素缺少alt属性"
})
# 显示问题
if issues:
result += f"⚠️ 发现 {len(issues)} 个问题:\n\n"
for i, issue in enumerate(issues, 1):
level = issue['level']
if level == 'error':
icon = f"{Fore.RED}❌{Style.RESET_ALL}"
elif level == 'warning':
icon = f"{Fore.YELLOW}⚠️{Style.RESET_ALL}"
else:
icon = f"{Fore.GREEN}ℹ️{Style.RESET_ALL}"
result += f"{icon} {i}. {issue['message']}\n"
else:
result += f"{Fore.GREEN}✅ 未发现问题{Style.RESET_ALL}\n"
result += "\n" + "=" * 60 + "\n"
return result
except UnicodeDecodeError:
return "❌ 文件编码问题,无法读取"
except Exception as e:
return f"❌ 验证失败: {str(e)}"
def _test_css(self, file_path):
"""测试CSS文件"""
if not os.path.exists(file_path):
return f"❌ 文件不存在: {file_path}"
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
result = f"\n🧪 CSS测试结果: {file_path}\n"
result += "=" * 60 + "\n\n"
issues = []
# 检查常见的CSS问题
if content.count('{') != content.count('}'):
issues.append({
"level": "error",
"message": "大括号不匹配"
})
# 检查颜色使用
if '#000000' in content or '#ffffff' in content:
issues.append({
"level": "info",
"message": "使用了纯黑或纯白,建议使用更柔和的颜色"
})
# 检查单位使用
if content.count('px') > 10 and 'rem' not in content and 'em' not in content:
issues.append({
"level": "warning",
"message": "大量使用px单位,建议考虑使用rem或em以支持更好的缩放"
})
# 统计
rule_count = content.count('{')
selector_count = len([line for line in content.split('\n') if '{' in line])
result += f"📊 文件统计:\n"
result += f" - CSS规则: {rule_count}\n"
result += f" - 选择器: {selector_count}\n\n"
# 显示问题
if issues:
result += f"⚠️ 发现 {len(issues)} 个问题:\n\n"
for i, issue in enumerate(issues, 1):
level = issue['level']
if level == 'error':
icon = f"{Fore.RED}❌{Style.RESET_ALL}"
elif level == 'warning':
icon = f"{Fore.YELLOW}⚠️{Style.RESET_ALL}"
else:
icon = f"{Fore.GREEN}ℹ️{Style.RESET_ALL}"
result += f"{icon} {i}. {issue['message']}\n"
else:
result += f"{Fore.GREEN}✅ 未发现问题{Style.RESET_ALL}\n"
result += "\n" + "=" * 60 + "\n"
return result
except UnicodeDecodeError:
return "❌ 文件编码问题,无法读取"
except Exception as e:
return f"❌ 测试失败: {str(e)}"
def _check_js(self, file_path):
"""检查JS文件"""
if not os.path.exists(file_path):
return f"❌ 文件不存在: {file_path}"
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
result = f"\n🧪 JavaScript检查结果: {file_path}\n"
result += "=" * 60 + "\n\n"
issues = []
# 检查常见问题
if 'var ' in content:
issues.append({
"level": "warning",
"message": "使用了var声明,建议使用let或const"
})
if '==' in content and '!==' not in content:
issues.append({
"level": "warning",
"message": "使用了==比较,建议使用===严格相等"
})
if content.count('console.log') > 5:
issues.append({
"level": "info",
"message": f"有{content.count('console.log')}个console.log,生产环境建议移除"
})
if 'eval(' in content:
issues.append({
"level": "error",
"message": "使用了eval(),存在安全风险"
})
# 统计
line_count = len(content.split('\n'))
function_count = content.count('function ')
result += f"📊 文件统计:\n"
result += f" - 代码行数: {line_count}\n"
result += f" - 函数数量: {function_count}\n\n"
# 显示问题
if issues:
result += f"⚠️ 发现 {len(issues)} 个问题:\n\n"
for i, issue in enumerate(issues, 1):
level = issue['level']
if level == 'error':
icon = f"{Fore.RED}❌{Style.RESET_ALL}"
elif level == 'warning':
icon = f"{Fore.YELLOW}⚠️{Style.RESET_ALL}"
else:
icon = f"{Fore.GREEN}ℹ️{Style.RESET_ALL}"
result += f"{icon} {i}. {issue['message']}\n"
else:
result += f"{Fore.GREEN}✅ 未发现问题{Style.RESET_ALL}\n"
result += "\n" + "=" * 60 + "\n"
return result
except UnicodeDecodeError:
return "❌ 文件编码问题,无法读取"
except Exception as e:
return f"❌ 检查失败: {str(e)}"
def _preview_in_browser(self, file_path):
"""在浏览器中预览"""
if not os.path.exists(file_path):
return f"❌ 文件不存在: {file_path}"
try:
# Windows系统
os.startfile(file_path)
return f"✅ 已在默认浏览器中打开: {file_path}"
except Exception as e:
return f"❌ 无法打开浏览器: {str(e)}"
def _analyze_project(self, project_path):
"""分析整个前端项目"""
if not os.path.exists(project_path):
return f"❌ 项目路径不存在: {project_path}"
result = f"\n🧪 前端项目分析: {project_path}\n"
result += "=" * 60 + "\n\n"
try:
# 统计文件
html_files = []
css_files = []
js_files = []
for root, dirs, files in os.walk(project_path):
for file in files:
if file.endswith('.html'):
html_files.append(os.path.join(root, file))
elif file.endswith('.css'):
css_files.append(os.path.join(root, file))
elif file.endswith('.js'):
js_files.append(os.path.join(root, file))
result += f"📊 项目文件统计:\n"
result += f" - HTML文件: {len(html_files)}\n"
result += f" - CSS文件: {len(css_files)}\n"
result += f" - JavaScript文件: {len(js_files)}\n\n"
# 建议检查
result += f"💡 建议检查:\n"
if html_files:
result += f" - HTML文件:\n"
for html_file in html_files[:3]: # 只显示前3个
result += f" • {os.path.basename(html_file)}\n"
if len(html_files) > 3:
result += f" • ... 还有 {len(html_files) - 3} 个文件\n"
if css_files:
result += f" - CSS文件:\n"
for css_file in css_files[:3]:
result += f" • {os.path.basename(css_file)}\n"
if len(css_files) > 3:
result += f" • ... 还有 {len(css_files) - 3} 个文件\n"
if js_files:
result += f" - JavaScript文件:\n"
for js_file in js_files[:3]:
result += f" • {os.path.basename(js_file)}\n"
if len(js_files) > 3:
result += f" • ... 还有 {len(js_files) - 3} 个文件\n"
result += f"\n{Fore.GREEN}✅ 分析完成{Style.RESET_ALL}\n"
result += "\n" + "=" * 60 + "\n"
return result
except Exception as e:
return f"❌ 分析失败: {str(e)}"
# 测试函数
def test_plugin():
"""测试插件功能"""
plugin = Plugin()
print("前端测试验证插件测试:")
print("\n1. 创建测试HTML文件:")
test_html = """<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>测试页面</title>
</head>
<body>
<h1>测试</h1>
</body>
</html>"""
with open('test.html', 'w', encoding='utf-8') as f:
f.write(test_html)
print("\n2. 验证HTML:")
print(plugin.handle("validate test.html"))
print("\n3. 在浏览器中预览:")
print(plugin.handle("preview test.html"))
print("\n4. 分析项目:")
print(plugin.handle("analyze ."))
# 清理测试文件
if os.path.exists('test.html'):
os.remove('test.html')
if __name__ == "__main__":
test_plugin()
+346
View File
@@ -0,0 +1,346 @@
"""
Git 集成插件 - 版本控制能力
对标 Claude Code 的 Git 工作流支持
"""
import os
import subprocess
from typing import Optional, Tuple
class Plugin:
"""Git 集成工具 - 提供完整的 Git 操作支持"""
def __init__(self):
self.usage = """Git 集成工具
操作类型:
status - 查看仓库状态
diff [文件路径] - 查看未暂存的变更
diff --cached - 查看已暂存的变更
log [数量] - 查看提交历史
log --oneline [数量] - 查看简洁提交历史
add <文件路径> - 暂存文件
add . - 暂存所有变更
commit <提交信息> - 提交变更
branch - 查看分支列表
branch <名称> - 创建新分支
checkout <分支名> - 切换分支
checkout -b <分支名> - 创建并切换分支
stash - 暂存当前变更
stash pop - 恢复暂存的变更
show <commit> - 查看指定提交
blame <文件路径> - 查看文件逐行修改记录
remote - 查看远程仓库
init - 初始化仓库
clone <URL> [目录] - 克隆仓库
"""
self.cli = None
def set_cli(self, cli):
self.cli = cli
def get_tool_info(self):
return {
"name": "git_tools",
"description": "Git 版本控制工具 - 查看状态、提交、分支管理、查看历史等",
"keywords": ["git", "版本控制", "提交", "分支", "commit", "branch", "status"],
"usage": self.usage
}
def get_mcp_definition(self):
return {
"name": "git_tools",
"description": "Git 版本控制工具",
"inputSchema": {
"type": "object",
"properties": {
"operation": {"type": "string", "description": "Git 操作"},
"args": {"type": "string", "description": "操作参数"},
"cwd": {"type": "string", "description": "工作目录"}
},
"required": ["operation"]
}
}
def convert_mcp_args(self, arguments):
op = arguments.get("operation", "")
args = arguments.get("args", "")
return f"{op} {args}".strip()
def _run_git(self, args: list, cwd: Optional[str] = None) -> Tuple[bool, str]:
"""执行 git 命令"""
try:
result = subprocess.run(
["git"] + args,
capture_output=True,
text=True,
cwd=cwd,
timeout=30
)
output = result.stdout
if result.stderr:
output += result.stderr
return result.returncode == 0, output.strip()
except FileNotFoundError:
return False, "错误:git 未安装。请先安装 Git。"
except subprocess.TimeoutExpired:
return False, "错误:Git 命令执行超时"
except Exception as e:
return False, f"错误:{str(e)}"
def handle(self, args: str) -> str:
try:
parts = args.strip().split(maxsplit=1)
if not parts:
return self._git_status("")
operation = parts[0].lower()
rest = parts[1] if len(parts) > 1 else ""
# 获取工作目录
cwd = None
if self.cli and hasattr(self.cli, 'current_dir'):
cwd = self.cli.current_dir
handlers = {
"status": self._git_status,
"diff": self._git_diff,
"log": self._git_log,
"add": self._git_add,
"commit": self._git_commit,
"branch": self._git_branch,
"checkout": self._git_checkout,
"stash": self._git_stash,
"show": self._git_show,
"blame": self._git_blame,
"remote": self._git_remote,
"init": self._git_init,
"clone": self._git_clone,
}
handler = handlers.get(operation)
if not handler:
# 尝试直接执行 git 命令
return self._git_raw(args, cwd)
return handler(rest, cwd)
except Exception as e:
return f"Git 工具错误: {str(e)}"
def _git_status(self, args: str, cwd=None) -> str:
ok, output = self._run_git(["status", "--porcelain"], cwd)
if not ok:
return f"Git 错误: {output}"
if not output:
return "✅ 工作区干净,没有未提交的变更"
# 解析状态
lines = output.split('\n')
staged = []
unstaged = []
untracked = []
for line in lines:
if len(line) < 4:
continue
status = line[:2]
filename = line[3:]
if status[0] in ('M', 'A', 'D', 'R', 'C'):
staged.append(f" ✅ {status[0]} {filename}")
if status[1] in ('M', 'D'):
unstaged.append(f" ⚠️ {status[1]} {filename}")
if status == '??':
untracked.append(f" 🆕 {filename}")
result = ["📊 Git 状态:"]
if staged:
result.append(f"\n已暂存 ({len(staged)}):")
result.extend(staged)
if unstaged:
result.append(f"\n未暂存 ({len(unstaged)}):")
result.extend(unstaged)
if untracked:
result.append(f"\n未跟踪 ({len(untracked)}):")
result.extend(untracked)
# 获取分支信息
ok2, branch = self._run_git(["branch", "--show-current"], cwd)
if ok2 and branch:
result.insert(1, f"🌿 分支: {branch}")
return '\n'.join(result)
def _git_diff(self, args: str, cwd=None) -> str:
if args.strip() == "--cached":
ok, output = self._run_git(["diff", "--cached"], cwd)
elif args.strip():
ok, output = self._run_git(["diff", args.strip()], cwd)
else:
ok, output = self._run_git(["diff"], cwd)
if not ok:
return f"Git 错误: {output}"
if not output:
return "没有变更"
# 限制输出长度
if len(output) > 5000:
output = output[:5000] + f"\n... (输出截断,共 {len(output)} 字符)"
return f"📝 变更:\n{output}"
def _git_log(self, args: str, cwd=None) -> str:
parts = args.split()
oneline = "--oneline" in parts
count = 10
for p in parts:
if p.isdigit():
count = int(p)
if oneline:
ok, output = self._run_git(
["log", "--oneline", f"-{count}", "--graph", "--decorate"], cwd)
else:
ok, output = self._run_git(
["log", f"-{count}", "--format=%h %an %ad %s", "--date=short"], cwd)
if not ok:
return f"Git 错误: {output}"
if not output:
return "暂无提交历史"
return f"📜 提交历史:\n{output}"
def _git_add(self, args: str, cwd=None) -> str:
target = args.strip()
if not target:
return "错误:请提供文件路径(使用 . 暂存所有)"
ok, output = self._run_git(["add", target], cwd)
if not ok:
return f"Git add 失败: {output}"
return f"✅ 已暂存: {target}"
def _git_commit(self, args: str, cwd=None) -> str:
message = args.strip()
if not message:
return "错误:请提供提交信息"
ok, output = self._run_git(["commit", "-m", message], cwd)
if not ok:
return f"Git commit 失败: {output}"
return f"✅ 提交成功:\n{output}"
def _git_branch(self, args: str, cwd=None) -> str:
if not args.strip():
ok, output = self._run_git(["branch", "-a"], cwd)
if not ok:
return f"Git 错误: {output}"
return f"🌿 分支列表:\n{output}"
else:
ok, output = self._run_git(["branch", args.strip()], cwd)
if not ok:
return f"创建分支失败: {output}"
return f"✅ 已创建分支: {args.strip()}"
def _git_checkout(self, args: str, cwd=None) -> str:
if args.startswith("-b "):
branch = args[3:].strip()
ok, output = self._run_git(["checkout", "-b", branch], cwd)
else:
branch = args.strip()
ok, output = self._run_git(["checkout", branch], cwd)
if not ok:
return f"切换分支失败: {output}"
return f"✅ 已切换到分支: {branch}"
def _git_stash(self, args: str, cwd=None) -> str:
if args.strip() == "pop":
ok, output = self._run_git(["stash", "pop"], cwd)
else:
ok, output = self._run_git(["stash"], cwd)
if not ok:
return f"Git stash 失败: {output}"
return f"✅ {output}"
def _git_show(self, args: str, cwd=None) -> str:
commit = args.strip() or "HEAD"
ok, output = self._run_git(["show", "--stat", commit], cwd)
if not ok:
return f"Git show 失败: {output}"
if len(output) > 3000:
output = output[:3000] + "\n... (输出截断)"
return output
def _git_blame(self, args: str, cwd=None) -> str:
path = args.strip()
if not path:
return "错误:请提供文件路径"
ok, output = self._run_git(["blame", "--line-porcelain", path], cwd)
if not ok:
return f"Git blame 失败: {output}"
# 简化输出
lines = output.split('\n')
simplified = []
for line in lines:
if line.startswith('author '):
author = line[7:]
elif line.startswith('\t'):
simplified.append(f" {author}: {line[1:]}")
if len(simplified) > 100:
simplified = simplified[:100]
simplified.append(f" ... (共 {len(simplified)} 行)")
return f"📝 {path} 修改记录:\n" + '\n'.join(simplified)
def _git_remote(self, args: str, cwd=None) -> str:
ok, output = self._run_git(["remote", "-v"], cwd)
if not ok:
return f"Git 错误: {output}"
if not output:
return "没有配置远程仓库"
return f"🌐 远程仓库:\n{output}"
def _git_init(self, args: str, cwd=None) -> str:
path = args.strip() or "."
ok, output = self._run_git(["init", path], cwd)
if not ok:
return f"Git init 失败: {output}"
return f"✅ {output}"
def _git_clone(self, args: str, cwd=None) -> str:
parts = args.split()
if not parts:
return "错误:请提供仓库 URL"
url = parts[0]
target = parts[1] if len(parts) > 1 else ""
cmd = ["clone", url]
if target:
cmd.append(target)
ok, output = self._run_git(cmd, cwd)
if not ok:
return f"Git clone 失败: {output}"
return f"✅ 克隆成功:\n{output}"
def _git_raw(self, args: str, cwd=None) -> str:
"""直接执行 git 命令"""
parts = args.split()
ok, output = self._run_git(parts, cwd)
if not ok:
return f"Git 错误: {output}"
return output
+1216
View File
File diff suppressed because it is too large. Load diff
+280
View File
@@ -0,0 +1,280 @@
import requests
import time
import socket
from urllib.parse import urlparse
import json
from colorama import Fore, Style
class Liugin:
"""网络工具插件"""
def __init__(self):
self.usage = """网络工具使用方法:/r
network_tools <操作> <参数>/r
例如:/r
- network_tools ping www.baidu.com - 测试网站连通性和延迟/r
- network_tools get https://www.example.com - 获取网页内容/r
- network_tools status https://www.example.com - 检查网站状态/r
- network_tools headers https://www.example.com - 获取网站响应头/r
- network_tools ip www.example.com - 获取网站IP地址/r
/r
详细说明:/r
- ping操作: network_tools ping <域名或IP> - 测试连通性和延迟/r
- get操作: network_tools get <URL> - 获取网页内容/r
- status操作: network_tools status <URL> - 检查网站HTTP状态码/r
- headers操作: network_tools headers <URL> - 获取网站响应头信息/r
- ip操作: network_tools ip <域名> - 获取域名对应的IP地址/r
/r
使用工具的JSON格式示例:/r
{"action": "use_tool", "tool": "network_tools", "args": "ping www.baidu.com"} - 测试百度连通性/r
{"action": "use_tool", "tool": "network_tools", "args": "get https://www.example.com"} - 获取网页内容/r
{"action": "use_tool", "tool": "network_tools", "args": "status https://www.example.com"} - 检查网站状态/r
{"action": "use_tool", "tool": "network_tools", "args": "headers https://www.example.com"} - 获取响应头/r
{"action": "use_tool", "tool": "network_tools", "args": "ip www.example.com"} - 获取IP地址/r
"""
self.cli = None # CLI实例引用
self.timeout = 10 # 请求超时时间(秒)
def set_cli(self, cli):
"""设置CLI实例引用"""
self.cli = cli
# 注册插件命令
self.cli.register_liugin_command('net', self.command_handler)
self.cli.register_liugin_command('ping', self.ping_command_handler)
self.cli.register_liugin_command('website', self.website_command_handler)
def command_handler(self, args):
"""处理 /net 命令"""
# 解析参数
parts = args.strip().split(maxsplit=1)
if len(parts) < 1:
return "请提供操作类型: ping, get, status, headers, ip 等"
operation = parts[0].lower()
remaining_args = parts[1] if len(parts) > 1 else ""
# 调用实际的网络工具功能
return self.handle(f"{operation} {remaining_args}")
def ping_command_handler(self, args):
"""处理 /ping 命令"""
return self.handle(f"ping {args}")
def website_command_handler(self, args):
"""处理 /website 命令"""
return self.handle(f"status {args}")
def get_tool_info(self):
return {
"name": "network_tools",
"description": "网络工具,用于测试网站连通性、获取网页内容、检查网站状态、获取响应头、解析IP地址等",
"keywords": ["网络", "ping", "网站", "状态", "网页", "IP", "headers", "get", "网络工具", "连通性", "延迟"],
"usage": """network_tools 工具使用说明:
JSON格式示例:
{"action": "use_tool", "tool": "network_tools", "args": "ping www.baidu.com"} - 测试百度连通性
{"action": "use_tool", "tool": "network_tools", "args": "get https://www.example.com"} - 获取网页内容
{"action": "use_tool", "tool": "network_tools", "args": "status https://www.example.com"} - 检查网站状态
{"action": "use_tool", "tool": "network_tools", "args": "headers https://www.example.com"} - 获取网站响应头
{"action": "use_tool", "tool": "network_tools", "args": "ip www.example.com"} - 获取网站IP地址"""
}
def handle(self, args):
"""处理网络工具请求"""
try:
# 解析参数
parts = args.strip().split(maxsplit=1)
if len(parts) < 1:
return "错误:参数不足。请提供操作类型。\n可用操作: ping, get, status, headers, ip"
operation = parts[0].lower()
if operation == "ping":
# 测试网站连通性和延迟
if len(parts) < 2:
return "错误:请提供要测试的域名或IP地址。格式: ping <域名或IP>"
target = parts[1].strip()
return self._ping(target)
elif operation == "get":
# 获取网页内容
if len(parts) < 2:
return "错误:请提供URL。格式: get <URL>"
url = parts[1].strip()
return self._get_content(url)
elif operation == "status":
# 检查网站状态码
if len(parts) < 2:
return "错误:请提供URL。格式: status <URL>"
url = parts[1].strip()
return self._check_status(url)
elif operation == "headers":
# 获取网站响应头
if len(parts) < 2:
return "错误:请提供URL。格式: headers <URL>"
url = parts[1].strip()
return self._get_headers(url)
elif operation == "ip":
# 获取域名IP地址
if len(parts) < 2:
return "错误:请提供域名。格式: ip <域名>"
domain = parts[1].strip()
return self._get_ip(domain)
else:
return f"错误:不支持的操作 '{operation}'。支持的操作有: ping, get, status, headers, ip。"
except Exception as e:
return f"网络工具操作错误: {str(e)}"
def _ping(self, target):
"""测试网站连通性和延迟"""
try:
# 确保URL格式正确
if not target.startswith(('http://', 'https://')):
test_url = 'http://' + target
else:
test_url = target
# 添加协议如果缺失
if not test_url.startswith(('http://', 'https://')):
test_url = 'http://' + test_url
start_time = time.time()
response = requests.get(test_url, timeout=self.timeout)
end_time = time.time()
response_time = (end_time - start_time) * 1000 # 转换为毫秒
result = f".ping 测试结果:\n"
result += f"目标: {target}\n"
result += f"HTTP状态码: {response.status_code}\n"
result += f"响应时间: {response_time:.2f} ms\n"
result += f"连接状态: {'成功' if response.status_code < 400 else '失败'}\n"
if response.status_code >= 400:
result += f"错误信息: {response.reason if hasattr(response, 'reason') else '请求失败'}\n"
return result
except requests.exceptions.Timeout:
return f".ping 测试结果:\n目标: {target}\n错误: 请求超时 (>{self.timeout}秒)\n连接状态: 失败"
except requests.exceptions.ConnectionError:
return f".ping 测试结果:\n目标: {target}\n错误: 连接失败 (无法连接到服务器)\n连接状态: 失败"
except requests.exceptions.RequestException as e:
return f".ping 测试结果:\n目标: {target}\n错误: {str(e)}\n连接状态: 失败"
except Exception as e:
return f".ping 测试结果:\n目标: {target}\n错误: {str(e)}\n连接状态: 失败"
def _get_content(self, url):
"""获取网页内容"""
try:
# 确保URL格式正确
if not url.startswith(('http://', 'https://')):
url = 'https://' + url
response = requests.get(url, timeout=self.timeout)
if response.status_code == 200:
content = response.text
# 限制内容长度以避免过长输出
if len(content) > 2000:
content = content[:2000] + f"\n... (内容已截断,完整内容共{len(content)}字符)"
result = f"网页内容 (来自 {url}):\n"
result += content
return result
else:
return f"获取网页内容失败,状态码: {response.status_code}, 原因: {response.reason if hasattr(response, 'reason') else ''}"
except requests.exceptions.Timeout:
return f"获取网页内容超时 (>{self.timeout}秒)"
except requests.exceptions.ConnectionError:
return f"无法连接到 {url},请检查URL是否正确"
except requests.exceptions.RequestException as e:
return f"请求失败: {str(e)}"
except Exception as e:
return f"获取网页内容时发生错误: {str(e)}"
def _check_status(self, url):
"""检查网站状态码"""
try:
# 确保URL格式正确
if not url.startswith(('http://', 'https://')):
url = 'https://' + url
response = requests.get(url, timeout=self.timeout)
result = f"网站状态 (来自 {url}):\n"
result += f"HTTP状态码: {response.status_code}\n"
result += f"状态描述: {response.reason if hasattr(response, 'reason') else 'Unknown'}\n"
result += f"响应时间: {response.elapsed.total_seconds():.2f} 秒\n"
result += f"内容长度: {len(response.content)} 字节\n"
result += f"服务器: {response.headers.get('Server', '未知')}\n"
result += f"内容类型: {response.headers.get('Content-Type', '未知')}\n"
return result
except requests.exceptions.Timeout:
return f"检查网站状态超时 (>{self.timeout}秒)"
except requests.exceptions.ConnectionError:
return f"无法连接到 {url},请检查URL是否正确"
except requests.exceptions.RequestException as e:
return f"请求失败: {str(e)}"
except Exception as e:
return f"检查网站状态时发生错误: {str(e)}"
def _get_headers(self, url):
"""获取网站响应头"""
try:
# 确保URL格式正确
if not url.startswith(('http://', 'https://')):
url = 'https://' + url
response = requests.get(url, timeout=self.timeout)
headers = dict(response.headers)
result = f"响应头 (来自 {url}):\n"
for key, value in headers.items():
result += f"{key}: {value}\n"
return result
except requests.exceptions.Timeout:
return f"获取响应头超时 (>{self.timeout}秒)"
except requests.exceptions.ConnectionError:
return f"无法连接到 {url},请检查URL是否正确"
except requests.exceptions.RequestException as e:
return f"请求失败: {str(e)}"
except Exception as e:
return f"获取响应头时发生错误: {str(e)}"
def _get_ip(self, domain):
"""获取域名对应的IP地址"""
try:
# 如果域名包含协议,去掉协议部分
if domain.startswith(('http://', 'https://')):
parsed = urlparse(domain)
domain = parsed.netloc
ip_address = socket.gethostbyname(domain)
result = f"域名解析结果:\n"
result += f"域名: {domain}\n"
result += f"IP地址: {ip_address}\n"
return result
except socket.gaierror:
return f"无法解析域名: {domain},请检查域名是否正确"
except Exception as e:
return f"域名解析失败: {str(e)}"
+198
View File
@@ -0,0 +1,198 @@
"""
Ollama图像识别插件
支持多种图像识别引擎的切换
"""
import os
import sys
from colorama import Fore, Style
# 添加项目根目录到sys.path
current_dir = os.path.dirname(os.path.abspath(__file__))
project_root = os.path.dirname(current_dir)
sys.path.insert(0, project_root)
from image_engine import get_engine_manager
class Liugin:
"""图像识别插件 - 支持多引擎切换"""
def __init__(self):
self.usage = """图像识别插件使用方法:
/image <操作> <参数>
操作命令:
- analyze <图片路径> - 详细分析图片
- ocr <图片路径> - 快速OCR文字识别
- quick <图片路径> - 快速图片描述
- describe <图片路径> - 完整描述图片
- model [模型名] - 查看/切换模型
- models - 列出可用的视觉模型
- engine [引擎名] - 查看/切换识别引擎
- engines - 列出所有识别引擎
示例:
- image analyze C:\\photo.png
- image ocr C:\\screenshot.png
- image model llava:13b
- image engine openai
JSON调用示例:
{"action": "use_tool", "tool": "image", "args": "analyze c:\\\\test.png"}
"""
self.cli = None
self.engine_manager = None
# 支持的图片格式
self.supported_formats = ['png', 'jpg', 'jpeg', 'gif', 'webp', 'bmp']
def set_cli(self, cli):
"""设置CLI实例引用"""
self.cli = cli
# 注册插件命令
self.cli.register_liugin_command('image', self.command_handler)
self.cli.register_liugin_command('img', self.command_handler)
def command_handler(self, args):
"""处理命令"""
parts = args.strip().split(maxsplit=1)
if len(parts) < 1:
return self._get_help()
operation = parts[0].lower()
remaining_args = parts[1] if len(parts) > 1 else ""
return self.handle(f"{operation} {remaining_args}")
def get_tool_info(self):
return {
"name": "image",
"description": "图像识别工具,支持多引擎切换,使用本地或云端模型进行图像分析",
"keywords": ["图像", "图片", "image", "analyze", "分析", "视觉", "识别", "OCR"],
"usage": """image 工具使用说明:
- analyze <图片路径> - 详细分析
- ocr <图片路径> - OCR文字识别
- quick <图片路径> - 快速描述
- engine <引擎名> - 切换引擎
- engines - 列出所有引擎
支持格式:PNG, JPG, JPEG, GIF, WEBP, BMP"""
}
def handle(self, args):
"""处理请求"""
# 延迟初始化引擎管理器
if self.engine_manager is None:
self.engine_manager = get_engine_manager()
try:
parts = args.strip().split(maxsplit=1)
if len(parts) < 1:
return self._get_help()
operation = parts[0].lower()
image_path = parts[1].strip() if len(parts) > 1 else ""
# 引擎管理命令
if operation == "engines":
return self.engine_manager.list_engines()
elif operation == "engine":
if image_path:
success, msg = self.engine_manager.switch_engine(image_path)
return msg
return f"当前引擎: {self.engine_manager.get_current_engine_name()}"
elif operation == "models":
engine = self.engine_manager.get_engine()
if engine and hasattr(engine, 'get_model_list'):
return engine.get_model_list()
return "当前引擎不支持模型列表"
elif operation == "model":
engine = self.engine_manager.get_engine()
if not image_path:
if engine and hasattr(engine, 'model'):
return f"当前模型: {engine.model}"
return "当前引擎不支持模型切换"
if engine and hasattr(engine, 'set_model'):
success, msg = engine.set_model(image_path)
return msg
return "当前引擎不支持模型切换"
# 图像分析命令
elif operation in ["analyze", "ocr", "quick", "describe"]:
if not image_path:
return f"错误:请提供图片路径。用法: {operation} <图片路径>"
return self._analyze_image(image_path, operation)
else:
# 兼容旧用法,直接当作图片路径处理
return self._analyze_image(operation, "analyze")
except Exception as e:
return f"错误: {str(e)}"
def _get_help(self):
"""获取帮助信息"""
current = self.engine_manager.get_current_engine_name() if self.engine_manager else "未初始化"
return f"""图像识别插件命令:
- analyze <图片路径> 详细分析图片
- ocr <图片路径> OCR文字识别
- quick <图片路径> 快速描述
- describe <图片路径> 完整描述
- model [模型名] 查看/切换模型
- models 列出视觉模型
- engine [引擎名] 查看/切换引擎
- engines 列出所有引擎
当前引擎: {current}"""
def _analyze_image(self, image_path, mode="analyze"):
"""使用引擎分析图片"""
# 检查文件
if not os.path.exists(image_path):
return f"文件不存在: {image_path}"
ext = os.path.splitext(image_path)[1].lower().lstrip('.')
if ext not in self.supported_formats:
return f"不支持的图片格式: {ext}。支持: {', '.join(self.supported_formats)}"
# 检查引擎
engine = self.engine_manager.get_engine()
if not engine:
return "没有可用的图像引擎"
file_size = os.path.getsize(image_path)
engine_name = self.engine_manager.get_current_engine_name()
# 获取模型名(如果支持)
model_info = ""
if hasattr(engine, 'model'):
model_info = f" | 模型: {engine.model}"
# 文件大小警告
size_warning = ""
if file_size > 1024 * 1024: # 大于1MB
size_warning = f" {Fore.YELLOW}(较大图片,处理可能需要较长时间){Style.RESET_ALL}"
print(f"{Fore.CYAN}📷 图像识别: {os.path.basename(image_path)}{Style.RESET_ALL}{size_warning}")
print(f"{Fore.YELLOW}📦 大小: {file_size/1024:.1f}KB | 引擎: {engine_name}{model_info}{Style.RESET_ALL}")
print(f"{Fore.YELLOW}⏳ 正在识别,请耐心等待...{Style.RESET_ALL}")
try:
# 使用引擎分析
full_response = ""
for chunk in self.engine_manager.analyze(image_path, mode, stream=True):
if chunk:
print(chunk, end='', flush=True)
full_response += chunk
print() # 换行
print(f"{Fore.GREEN}✓ 识别完成{Style.RESET_ALL}")
return full_response if full_response else "识别结果为空"
except Exception as e:
error_msg = str(e)
print(f"{Fore.RED}✗ {error_msg}{Style.RESET_ALL}")
return f"识别失败: {error_msg}"
+1
View File
@@ -0,0 +1 @@
"hello"
+144
View File
@@ -0,0 +1,144 @@
"""
发送图片插件 - 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
+549
View File
@@ -0,0 +1,549 @@
"""
语音识别插件 - 弹窗录音界面
AI 触发后弹出独立录音窗口
"""
import os
import json
import subprocess
import sys
import time
from typing import Optional, Dict, Any, List
# 结果保存路径
RESULT_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "speech_temp", "last_result.txt")
TEMP_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "speech_temp")
class Liugin:
"""语音识别插件"""
def __init__(self):
self.plugin_dir = os.path.dirname(os.path.abspath(__file__))
self.temp_dir = TEMP_DIR
os.makedirs(self.temp_dir, exist_ok=True)
self.usage = """语音识别插件使用方法:
speech_recognition <操作> [参数]
录音识别:
- record - 弹出录音窗口,录音并识别
文件识别:
- file <音频文件路径> - 识别音频文件内容
其他:
- status - 查看状态
- result - 获取上次识别结果
支持的音频格式: wav, mp3, m4a, flac, ogg
"""
self.cli = None
def set_cli(self, cli):
self.cli = cli
def get_tool_info(self):
return {
"name": "speech_recognition",
"description": "语音识别插件,弹出录音窗口进行语音识别,将语音转换为文字",
"keywords": ["语音", "识别", "录音", "转文字", "speech", "voice", "recognition"],
"usage": self.usage
}
def get_mcp_definition(self):
return {
"name": "speech_recognition",
"description": "语音识别插件,弹出录音窗口进行语音识别",
"inputSchema": {
"type": "object",
"properties": {
"operation": {
"type": "string",
"enum": ["record", "file", "status", "result"],
"description": "操作类型"
},
"file_path": {"type": "string", "description": "音频文件路径"}
},
"required": ["operation"]
}
}
def convert_mcp_args(self, arguments: Dict[str, Any]) -> str:
op = arguments.get("operation", "")
parts = [op]
if op == "file":
parts.append(arguments.get("file_path", ""))
return " ".join(filter(None, parts))
def handle(self, args: str) -> str:
try:
parts = self._parse_args(args)
if not parts:
return "错误:请提供操作类型。\n" + self.usage
operation = parts[0].lower()
handlers = {
"record": self._op_record,
"file": self._op_file,
"status": self._op_status,
"result": self._op_result,
}
if operation in handlers:
return handlers[operation](parts[1:])
else:
return f"错误:未知操作 '{operation}'\n可用操作: " + ", ".join(handlers.keys())
except Exception as e:
return f"处理请求失败: {e}"
def _parse_args(self, args: str) -> List[str]:
parts = []
current = ""
in_quotes = False
quote_char = None
for char in args:
if char in ('"', "'") and not in_quotes:
in_quotes = True
quote_char = char
elif char == quote_char and in_quotes:
in_quotes = False
quote_char = None
elif char == ' ' and not in_quotes:
if current:
parts.append(current)
current = ""
else:
current += char
if current:
parts.append(current)
return parts
# ==================== 弹窗录音 ====================
def _op_record(self, args: List[str]) -> str:
"""弹出录音窗口"""
# 创建录音窗口脚本
recorder_script = os.path.join(self.temp_dir, "recorder_gui.py")
gui_code = f'''import tkinter as tk
from tkinter import messagebox
import os
import sys
import threading
import time
RESULT_FILE = r"{RESULT_FILE}"
TEMP_DIR = r"{TEMP_DIR}"
class VoiceRecorder:
def __init__(self):
self.root = tk.Tk()
self.root.title("语音识别录音")
self.root.geometry("450x350")
self.root.resizable(False, False)
# 居中显示
self.root.update_idletasks()
x = (self.root.winfo_screenwidth() - 450) // 2
y = (self.root.winfo_screenheight() - 350) // 2
self.root.geometry(f"450x350+{{x}}+{{y}}")
# 录音状态
self.is_recording = False
self.record_file = None
self.frames = []
self.start_time = None
self.p = None
self.stream = None
# 创建界面
self._create_ui()
# 检查 pyaudio
self.has_pyaudio = self._check_pyaudio()
if not self.has_pyaudio:
self.status_label.config(text="请安装 pyaudio: pip install pyaudio", fg="red")
def _check_pyaudio(self):
try:
import pyaudio
return True
except:
return False
def _create_ui(self):
# 标题
title_label = tk.Label(self.root, text="🎙️ 语音识别录音", font=("Arial", 18, "bold"))
title_label.pack(pady=15)
# 状态标签
self.status_label = tk.Label(self.root, text="点击开始录音", font=("Arial", 11))
self.status_label.pack(pady=5)
# 时间标签
self.time_label = tk.Label(self.root, text="00:00", font=("Arial", 28, "bold"), fg="#2196F3")
self.time_label.pack(pady=10)
# 按钮框架
btn_frame = tk.Frame(self.root)
btn_frame.pack(pady=15)
# 开始/停止按钮
self.record_btn = tk.Button(
btn_frame, text="🎤 开始录音", font=("Arial", 12),
width=12, height=2, command=self.toggle_record,
bg="#4CAF50", fg="white", cursor="hand2"
)
self.record_btn.pack(side=tk.LEFT, padx=10)
# 识别按钮
self.recognize_btn = tk.Button(
btn_frame, text="📝 识别", font=("Arial", 12),
width=12, height=2, command=self.recognize,
bg="#2196F3", fg="white", cursor="hand2", state=tk.DISABLED
)
self.recognize_btn.pack(side=tk.LEFT, padx=10)
# 结果显示
result_frame = tk.LabelFrame(self.root, text="识别结果", font=("Arial", 10))
result_frame.pack(pady=10, padx=15, fill=tk.BOTH, expand=True)
self.result_text = tk.Text(result_frame, height=5, width=50, font=("Arial", 10))
self.result_text.pack(pady=5, padx=5, fill=tk.BOTH, expand=True)
def toggle_record(self):
if not self.has_pyaudio:
messagebox.showerror("错误", "请先安装 pyaudio:\npip install pyaudio")
return
if self.is_recording:
self.stop_record()
else:
self.start_record()
def start_record(self):
import pyaudio
self.is_recording = True
self.frames = []
self.start_time = time.time()
self.record_btn.config(text="⏹️ 停止录音", bg="#f44336")
self.status_label.config(text="🔴 正在录音...", fg="red")
self.recognize_btn.config(state=tk.DISABLED)
CHUNK = 1024
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 16000
self.p = pyaudio.PyAudio()
self.stream = self.p.open(
format=FORMAT, channels=CHANNELS, rate=RATE,
input=True, frames_per_buffer=CHUNK
)
# 更新时间
self._update_time()
# 录音线程
def record_loop():
while self.is_recording:
try:
data = self.stream.read(CHUNK, exception_on_overflow=False)
self.frames.append(data)
except:
break
self.record_thread = threading.Thread(target=record_loop, daemon=True)
self.record_thread.start()
def stop_record(self):
self.is_recording = False
self.record_btn.config(text="🎤 开始录音", bg="#4CAF50")
self.status_label.config(text="录音完成,可以识别", fg="green")
self.recognize_btn.config(state=tk.NORMAL)
# 停止录音
if self.stream:
self.stream.stop_stream()
self.stream.close()
if self.p:
self.p.terminate()
# 保存文件
import wave
import pyaudio
self.record_file = os.path.join(TEMP_DIR, f"record_{{int(time.time())}}.wav")
wf = wave.open(self.record_file, "wb")
wf.setnchannels(1)
wf.setsampwidth(pyaudio.get_sample_size(pyaudio.paInt16))
wf.setframerate(16000)
wf.writeframes(b"".join(self.frames))
wf.close()
self.result_text.delete(1.0, tk.END)
self.result_text.insert(tk.END, "✅ 录音已保存\n点击 [识别] 按钮进行语音识别")
def _update_time(self):
if self.is_recording:
elapsed = int(time.time() - self.start_time)
mins = elapsed // 60
secs = elapsed % 60
self.time_label.config(text=f"{{mins:02d}}:{{secs:02d}}")
self.root.after(100, self._update_time)
def recognize(self):
if not self.record_file or not os.path.exists(self.record_file):
messagebox.showerror("错误", "没有录音文件,请先录音")
return
self.status_label.config(text="正在识别...", fg="blue")
self.result_text.delete(1.0, tk.END)
self.result_text.insert(tk.END, "🔍 识别中,请稍候...")
self.root.update()
# 尝试识别
result = self._do_recognize()
self.result_text.delete(1.0, tk.END)
if result:
self.result_text.insert(tk.END, result)
self.status_label.config(text="识别完成", fg="green")
# 保存结果
with open(RESULT_FILE, "w", encoding="utf-8") as f:
f.write(result)
else:
self.result_text.insert(tk.END, "❌ 识别失败\n\n请安装识别引擎:\n• pip install openai-whisper\n• pip install SpeechRecognition")
self.status_label.config(text="识别失败", fg="red")
def _do_recognize(self):
# 尝试 Whisper
try:
import whisper
self.status_label.config(text="使用 Whisper 识别...", fg="blue")
self.root.update()
model = whisper.load_model("base")
result = model.transcribe(self.record_file, language="zh")
text = result.get("text", "").strip()
if text:
return f"✅ {{text}}"
except Exception as e:
print(f"Whisper error: {{e}}")
# 尝试 SpeechRecognition
try:
import speech_recognition as sr
self.status_label.config(text="使用 Google 识别...", fg="blue")
self.root.update()
r = sr.Recognizer()
with sr.AudioFile(self.record_file) as source:
audio = r.record(source)
text = r.recognize_google(audio, language="zh-CN")
if text:
return f"✅ {{text}}"
except Exception as e:
print(f"SpeechRecognition error: {{e}}")
return None
def run(self):
self.root.mainloop()
# 清理
if self.is_recording:
self.is_recording = False
if self.stream:
self.stream.close()
if self.p:
self.p.terminate()
if __name__ == "__main__":
app = VoiceRecorder()
app.run()
'''
# 写入脚本
with open(recorder_script, "w", encoding="utf-8") as f:
f.write(gui_code)
# 启动录音窗口
try:
subprocess.Popen([sys.executable, recorder_script],
creationflags=subprocess.CREATE_NEW_CONSOLE)
return "🎙️ 已弹出录音窗口,请在窗口中进行录音和识别\n\n使用方法:\n1. 点击「开始录音」\n2. 对着麦克风说话\n3. 点击「停止录音」\n4. 点击「识别」获取文字结果"
except Exception as e:
return f"❌ 启动录音窗口失败: {e}"
# ==================== 文件识别 ====================
def _op_file(self, args: List[str]) -> str:
"""识别音频文件"""
if not args:
return "错误:请提供音频文件路径"
file_path = args[0]
if not os.path.exists(file_path):
return f"❌ 文件不存在: {file_path}"
return self._recognize_file(file_path)
def _recognize_file(self, file_path: str) -> str:
"""识别音频文件内容"""
# 尝试使用 whisper(本地)
result = self._try_whisper(file_path)
if result:
self._save_result(result)
return result
# 尝试使用 speech_recognition 库
result = self._try_speech_recognition(file_path)
if result:
self._save_result(result)
return result
return "❌ 语音识别失败:未安装识别引擎\n\n安装方法:\n1. Whisper (推荐): pip install openai-whisper\n2. SpeechRecognition: pip install SpeechRecognition\n3. PyAudio (录音): pip install pyaudio"
def _save_result(self, result: str):
"""保存识别结果"""
try:
with open(RESULT_FILE, "w", encoding="utf-8") as f:
f.write(result)
except:
pass
def _try_whisper(self, file_path: str) -> Optional[str]:
"""使用 Whisper 识别"""
try:
import whisper
print(f"[语音识别] 使用 Whisper 识别中...")
model = whisper.load_model("base")
result = model.transcribe(file_path, language="zh")
text = result.get("text", "").strip()
if text:
return f"✅ 识别结果:\n{text}"
return None
except ImportError:
return None
except Exception as e:
print(f"[语音识别] Whisper 错误: {e}")
return None
def _try_speech_recognition(self, file_path: str) -> Optional[str]:
"""使用 SpeechRecognition 库识别"""
wav_file = file_path
try:
import speech_recognition as sr
print(f"[语音识别] 使用 SpeechRecognition 识别中...")
r = sr.Recognizer()
# 转换为 wav 格式
wav_file = self._convert_to_wav(file_path)
with sr.AudioFile(wav_file) as source:
audio = r.record(source)
# 尝试 Google 语音识别
try:
text = r.recognize_google(audio, language="zh-CN")
if text:
return f"✅ 识别结果:\n{text}"
except:
pass
return None
except ImportError:
return None
except Exception as e:
print(f"[语音识别] SpeechRecognition 错误: {e}")
return None
finally:
# 清理转换的临时文件
if wav_file != file_path and os.path.exists(wav_file):
try:
os.remove(wav_file)
except:
pass
def _convert_to_wav(self, file_path: str) -> str:
"""转换音频为 wav 格式"""
if file_path.lower().endswith('.wav'):
return file_path
try:
wav_file = os.path.join(self.temp_dir, f"convert_{int(time.time())}.wav")
# 使用 ffmpeg 转换
subprocess.run([
'ffmpeg', '-y', '-i', file_path,
'-acodec', 'pcm_s16le',
'-ar', '16000',
'-ac', '1',
wav_file
], capture_output=True, check=True)
return wav_file
except Exception as e:
print(f"[语音识别] 转换错误: {e}")
return file_path
# ==================== 其他 ====================
def _op_status(self, args: List[str]) -> str:
"""查看状态"""
lines = ["📊 语音识别插件状态"]
# 检查可用引擎
engines = []
try:
import whisper
engines.append("whisper ✅")
except:
engines.append("whisper ❌ (pip install openai-whisper)")
try:
import speech_recognition
engines.append("speech_recognition ✅")
except:
engines.append("speech_recognition ❌ (pip install SpeechRecognition)")
try:
import pyaudio
engines.append("pyaudio ✅ (录音支持)")
except:
engines.append("pyaudio ❌ (pip install pyaudio)")
lines.append("\n可用引擎:")
for e in engines:
lines.append(f" - {e}")
return "\n".join(lines)
def _op_result(self, args: List[str]) -> str:
"""获取上次识别结果"""
if os.path.exists(RESULT_FILE):
try:
with open(RESULT_FILE, "r", encoding="utf-8") as f:
return f.read()
except:
pass
return "暂无识别结果"
if __name__ == "__main__":
plugin = Plugin()
print("语音识别插件")
print(f"临时目录: {plugin.temp_dir}")
print("\n" + plugin.usage)
+263
View File
@@ -0,0 +1,263 @@
"""
任务管理插件 - 提供任务追踪和进度管理功能
"""
import json
import os
from datetime import datetime
from colorama import Fore, Style
class Liugin:
"""任务管理插件 - 类似 iFlow CLI 的 todo 功能"""
def __init__(self):
self.usage = """任务管理工具使用方法:
JSON格式示例:
{"action": "use_tool", "tool": "task_manager", "args": "add 创建登录页面"} - 添加新任务
{"action": "use_tool", "tool": "task_manager", "args": "list"} - 列出所有任务
{"action": "use_tool", "tool": "task_manager", "args": "complete 1"} - 完成任务ID为1的任务
{"action": "use_tool", "tool": "task_manager", "args": "in_progress 2"} - 将任务ID为2标记为进行中
{"action": "use_tool", "tool": "task_manager", "args": "delete 3"} - 删除任务ID为3的任务
{"action": "use_tool", "tool": "task_manager", "args": "clear"} - 清除所有已完成的任务
功能说明:
- add <任务描述> - 添加新任务(自动分配ID)
- list - 列出所有任务及其状态
- complete <任务ID> - 标记任务为已完成
- in_progress <任务ID> - 标记任务为进行中
- delete <任务ID> - 删除指定任务
- clear - 清除所有已完成的任务
状态说明:
- pending: 待处理
- in_progress: 进行中
- completed: 已完成
- failed: 失败"""
self.cli = None
self.tasks_file = "tasks.json"
self.tasks = []
self._load_tasks()
def set_cli(self, cli):
"""设置CLI实例引用"""
self.cli = cli
# 注册插件命令
self.cli.register_liugin_command('task', self.command_handler)
self.cli.register_liugin_command('todo', self.command_handler)
def command_handler(self, args):
"""处理 /task 或 /todo 命令"""
return self.handle(args)
def get_tool_info(self):
return {
"name": "task_manager",
"description": "任务管理工具,提供任务追踪和进度管理功能,支持添加、完成、删除任务",
"keywords": ["任务", "task", "todo", "管理", "进度", "追踪", "list", "complete"],
"usage": self.usage
}
def _load_tasks(self):
"""从文件加载任务"""
try:
if os.path.exists(self.tasks_file):
with open(self.tasks_file, 'r', encoding='utf-8') as f:
self.tasks = json.load(f)
except Exception as e:
self.tasks = []
def _save_tasks(self):
"""保存任务到文件"""
try:
with open(self.tasks_file, 'w', encoding='utf-8') as f:
json.dump(self.tasks, f, ensure_ascii=False, indent=2)
except Exception as e:
return f"保存任务失败: {e}"
return None
def handle(self, args):
"""处理任务管理请求"""
try:
parts = args.strip().split()
if not parts:
return self._list_tasks()
operation = parts[0].lower()
if operation == "add":
# 添加任务
if len(parts) < 2:
return "错误:请提供任务描述。格式: add <任务描述>"
task_desc = " ".join(parts[1:])
return self._add_task(task_desc)
elif operation == "list":
# 列出任务
return self._list_tasks()
elif operation == "complete":
# 完成任务
if len(parts) < 2:
return "错误:请提供任务ID。格式: complete <任务ID>"
task_id = parts[1]
return self._update_task_status(task_id, "completed")
elif operation == "in_progress":
# 标记为进行中
if len(parts) < 2:
return "错误:请提供任务ID。格式: in_progress <任务ID>"
task_id = parts[1]
return self._update_task_status(task_id, "in_progress")
elif operation == "delete":
# 删除任务
if len(parts) < 2:
return "错误:请提供任务ID。格式: delete <任务ID>"
task_id = parts[1]
return self._delete_task(task_id)
elif operation == "clear":
# 清除已完成任务
return self._clear_completed()
else:
return f"错误:不支持的操作 '{operation}'。支持的操作有: add, list, complete, in_progress, delete, clear"
except Exception as e:
return f"任务管理操作错误: {str(e)}"
def _add_task(self, description):
"""添加新任务"""
# 使用最大ID+1的方式分配ID,避免删除后ID重复
existing_ids = [int(t.get("id", 0)) for t in self.tasks if t.get("id", "").isdigit()]
task_id = str(max(existing_ids, default=0) + 1)
new_task = {
"id": task_id,
"task": description,
"status": "pending",
"created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"priority": "medium"
}
self.tasks.append(new_task)
error = self._save_tasks()
if error:
return error
return f"✅ 已添加任务 #{task_id}: {description}"
def _list_tasks(self):
"""列出所有任务"""
if not self.tasks:
return "📝 当前没有任务"
result = f"\n📋 任务列表 (共 {len(self.tasks)} 个任务)\n"
result += "=" * 50 + "\n"
pending_count = 0
in_progress_count = 0
completed_count = 0
for task in self.tasks:
status = task.get('status', 'pending')
task_id = task.get('id', '?')
description = task.get('task', '')
created_at = task.get('created_at', '')
# 统计
if status == 'pending':
pending_count += 1
status_icon = "⏳"
status_color = Fore.YELLOW
elif status == 'in_progress':
in_progress_count += 1
status_icon = "🔄"
status_color = Fore.CYAN
elif status == 'completed':
completed_count += 1
status_icon = "✅"
status_color = Fore.GREEN
else:
status_icon = "❌"
status_color = Fore.RED
result += f"{status_icon} [{task_id}] {status_color}{status}{Style.RESET_ALL} - {description}\n"
result += f" 创建时间: {created_at}\n\n"
result += "=" * 50 + "\n"
result += f"📊 统计: {Fore.YELLOW}待处理 {pending_count}{Style.RESET_ALL} | {Fore.CYAN}进行中 {in_progress_count}{Style.RESET_ALL} | {Fore.GREEN}已完成 {completed_count}{Style.RESET_ALL}\n"
return result
def _update_task_status(self, task_id, new_status):
"""更新任务状态"""
for task in self.tasks:
if task.get('id') == task_id:
old_status = task.get('status')
task['status'] = new_status
if new_status == 'completed':
task['completed_at'] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
error = self._save_tasks()
if error:
return error
return f"✅ 任务 #{task_id} 状态已更新: {old_status} → {new_status}"
return f"❌ 未找到任务ID: {task_id}"
def _delete_task(self, task_id):
"""删除任务"""
for i, task in enumerate(self.tasks):
if task.get('id') == task_id:
deleted_task = self.tasks.pop(i)
error = self._save_tasks()
if error:
return error
return f"🗑️ 已删除任务 #{task_id}: {deleted_task.get('task', '')}"
return f"❌ 未找到任务ID: {task_id}"
def _clear_completed(self):
"""清除所有已完成的任务"""
completed_count = len([t for t in self.tasks if t.get('status') == 'completed'])
if completed_count == 0:
return "📝 没有已完成的任务需要清除"
self.tasks = [t for t in self.tasks if t.get('status') != 'completed']
error = self._save_tasks()
if error:
return error
return f"✅ 已清除 {completed_count} 个已完成的任务"
# 测试函数
def test_plugin():
"""测试插件功能"""
plugin = Plugin()
print("任务管理插件测试:")
print("\n1. 添加任务:")
print(plugin.handle("add 创建登录页面"))
print(plugin.handle("add 实现用户认证"))
print(plugin.handle("add 添加数据库连接"))
print("\n2. 列出任务:")
print(plugin.handle("list"))
print("\n3. 标记任务为进行中:")
print(plugin.handle("in_progress 1"))
print("\n4. 完成任务:")
print(plugin.handle("complete 1"))
print("\n5. 再次列出任务:")
print(plugin.handle("list"))
print("\n6. 清除已完成任务:")
print(plugin.handle("clear"))
print("\n7. 最终任务列表:")
print(plugin.handle("list"))
if __name__ == "__main__":
test_plugin()
+107
View File
@@ -0,0 +1,107 @@
"""
工具搜索插件 - 支持按需发现工具
"""
import json
class Liugin:
"""工具搜索插件 - 允许 AI 按需搜索和发现可用工具"""
def __init__(self):
self.usage = """工具搜索使用方法:
tool_search <关键词> - 根据关键词搜索可用工具
例如:
- tool_search 文件 - 搜索与文件相关的工具
- tool_search 网络 - 搜索与网络相关的工具
- tool_search 图片 - 搜索与图片相关的工具
此插件帮助 AI 在需要特定功能时快速找到合适的工具,
而不是预加载所有工具到上下文中。"""
self.cli = None
def set_cli(self, cli):
"""设置 CLI 实例引用"""
self.cli = cli
def get_tool_info(self):
return {
"name": "tool_search",
"description": "工具搜索插件,支持按需发现和搜索可用工具。当你不确定有哪些工具可用时,可以使用此工具搜索相关功能的工具。",
"keywords": ["搜索", "工具", "发现", "search", "tool", "find", "可用工具", "工具列表"],
"usage": """tool_search 工具使用说明:
JSON格式示例:
{"action": "use_tool", "tool": "tool_search", "args": "文件"} - 搜索与文件相关的工具
{"action": "use_tool", "tool": "tool_search", "args": "网络"} - 搜索与网络相关的工具
{"action": "use_tool", "tool": "tool_search", "args": "图片"} - 搜索与图片相关的工具
功能说明:
- 支持关键词匹配(工具名称、描述、关键词)
- 返回匹配度最高的前 5 个工具
- 包含工具的详细使用说明
- 帮助 AI 快速找到合适的工具而不需要预加载所有工具"""
}
def handle(self, args):
"""处理工具搜索请求"""
try:
# 解析参数
query = args.strip()
if not query:
return "错误:请提供搜索关键词。格式: tool_search <关键词>"
# 搜索工具
if self.cli and hasattr(self.cli, 'liugin_manager'):
matches = self.cli.liugin_manager.search_tools(query, limit=5)
if not matches:
return f"未找到与 '{query}' 相关的工具。"
# 格式化搜索结果
result_lines = [f"找到 {len(matches)} 个与 '{query}' 相关的工具:\n"]
for i, match in enumerate(matches, 1):
result_lines.append(f"{i}. {match['name']}")
result_lines.append(f" 描述: {match['description']}")
# 显示协议类型
protocol = match.get('protocol', 'liugin')
result_lines.append(f" 协议: {protocol}")
# 关键词
keywords = match.get('keywords', [])
if keywords:
result_lines.append(f" 关键词: {', '.join(keywords)}")
# 根据协议类型显示不同的使用说明
if protocol == 'skill':
# Skill 协议:显示完整指令
instance = match.get('instance')
if instance:
instructions = getattr(instance, 'instructions', '')
if instructions:
# 限制长度
if len(instructions) > 1500:
instructions = instructions[:1500] + "\n...(内容已截断)"
result_lines.append(f" 指令:\n{instructions}")
else:
usage = match.get('usage', '')
if usage:
result_lines.append(f" 用法: {usage}")
else:
# Plugin 协议:显示 usage
if match.get('usage'):
usage = match['usage']
if len(usage) > 500:
usage = usage[:500] + "..."
result_lines.append(f" 用法: {usage}")
result_lines.append("")
return "\n".join(result_lines)
else:
return "错误:无法访问插件管理器。"
except Exception as e:
return f"工具搜索失败: {str(e)}"