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

refactor: 插件大优化 17→13 个，5689→4309 行

Browse Source 
合并:
- code_editor + code_search → code_editor (编辑+搜索统一)
- git_tools + git_workflow → git_tools (基础+工作流统一)
- auto_engineer + project_analyzer → auto_engineer (质量+分析统一)

修复:
- cmd_executor: 移除 Windows-only 限制，跨平台兼容
- cmd_executor: 移除坏掉的 from ai_cli import

精简:
- file_manager: 移除和 code_editor 重叠的行编辑操作

删除:
- scheduler.py (空文件)
- git_workflow.py (已合并)
- code_search.py (已合并)
- project_analyzer.py (已合并)
This commit is contained in:
xiaoli-cli committed 2026-04-26 05:36:18 +08:00
1 parent 119e939a23
commit 9aae8b2ca2
9 files changed
+1295 -2676
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

plugins/auto_engineer.py
+267 -163
View File
@@ -1,29 +1,38 @@
"""
工程化自动化插件 - 代码质量检查、格式化、测试运行
工程化工具插件 - 代码质量 + 项目分析
合并原 auto_engineer + project_analyzer，消除重复
"""
import os
import subprocess
import json
import ast
import re
from typing import Tuple
import json
import subprocess
from typing import Tuple, List
class Plugin:
"""工程化工具 - lint、format、test、build 自动化"""
"""工程化工具 - lint/format/test/build + 代码分析"""
def __init__(self):
self.usage = """工程化自动化工具
self.usage = """工程化与分析工具
操作类型:
lint [目录] [文件模式] - 代码质量检查 (ruff/flake8/pylint)
format [目录] [文件模式] - 代码格式化 (black/ruff)
质量检查:
lint [目录] [文件模式] - 代码质量检查 (ruff/flake8)
format [目录] [文件模式] - 代码格式化 (ruff/black)
test [目录] [参数] - 运行测试 (pytest)
build [命令] - 构建项目
deps - 检查依赖 (pip list --outdated)
deps - 检查过时依赖
deps-check - 安全漏洞检查
pre-commit - 运行 pre-commit 检查
项目分析:
complexity [目录] - 圈复杂度分析
duplicates [目录] - 重复代码检测
security [目录] - 安全风险扫描
metrics [目录] - 代码质量指标
suggest [目录] - 改进建议
info - 项目信息摘要
init [类型] - 初始化项目结构 (python/web/cli)
init [类型] - 初始化项目 (python/web)
"""
self.cli = None
@@ -33,31 +42,22 @@ class Plugin:
def get_tool_info(self):
return {
"name": "auto_engineer",
"description": "工程化自动化 - lint/format/test/build/deps 一键执行",
"keywords": ["lint", "format", "test", "build", "代码质量", "格式化", "测试", "构建"],
"description": "工程化工具 - lint/format/test/build + 代码复杂度/安全/质量分析",
"keywords": ["lint", "format", "test", "build", "代码质量", "格式化", "测试",
"复杂度", "安全", "分析", "依赖", "项目"],
"usage": self.usage
}
def get_mcp_definition(self):
return {
"name": "auto_engineer",
"description": "工程化自动化工具",
"description": "工程化与分析工具",
"inputSchema": {
"type": "object",
"properties": {
"operation": {
"type": "string",
"enum": ["lint", "format", "test", "build", "deps", "deps-check", "pre-commit", "info", "init"],
"description": "操作类型"
},
"target": {
"type": "string",
"description": "目标目录或命令"
},
"options": {
"type": "string",
"description": "额外选项"
}
"operation": {"type": "string"},
"target": {"type": "string"},
"options": {"type": "string"}
},
"required": ["operation"]
}
@@ -66,8 +66,45 @@ class Plugin:
def convert_mcp_args(self, arguments):
op = arguments.get("operation", "")
target = arguments.get("target", "")
options = arguments.get("options", "")
return f"{op} {target} {options}".strip()
return f"{op} {target}".strip()
# ── 工具方法 ──
def _run(self, cmd: str, cwd: str = None, timeout: int = 60) -> Tuple[bool, str]:
try:
result = subprocess.run(
cmd, shell=True, capture_output=True, text=True,
cwd=cwd or os.getcwd(), timeout=timeout
)
output = result.stdout + (("\n" + result.stderr) if result.stderr else "")
return result.returncode == 0, output.strip()
except subprocess.TimeoutExpired:
return False, f"命令超时 ({timeout}s)"
except Exception as e:
return False, str(e)
def _find_tool(self, *tools) -> str:
for tool in tools:
mod = tool.replace("-", "_")
ok, _ = self._run(f"python3 -c 'import {mod}'", timeout=5)
if ok:
return tool
ok, _ = self._run(f"which {tool}", timeout=5)
if ok:
return tool
return None
def _walk_py(self, target: str) -> List[str]:
files = []
skip = {'.git', '__pycache__', '.venv', 'venv', 'node_modules'}
for root, dirs, fnames in os.walk(target):
dirs[:] = [d for d in dirs if d not in skip]
for f in fnames:
if f.endswith('.py'):
files.append(os.path.join(root, f))
return files
# ── 路由 ──
def handle(self, args: str) -> str:
parts = args.strip().split(maxsplit=2)
@@ -86,205 +123,286 @@ class Plugin:
"deps": self._deps,
"deps-check": self._deps_check,
"pre-commit": self._pre_commit,
"complexity": self._complexity,
"duplicates": self._duplicates,
"security": self._security,
"metrics": self._metrics,
"suggest": self._suggest,
"info": self._info,
"init": self._init_project,
}
handler = handlers.get(op)
if not handler:
return f"错误: 不支持的操作 '{op}'"
return f"错误：不支持的操作 '{op}'"
try:
return handler(target, options)
except Exception as e:
return f"执行错误: {e}"
def _run(self, cmd: str, cwd: str = None, timeout: int = 60) -> Tuple[bool, str]:
"""执行命令"""
try:
result = subprocess.run(
cmd, shell=True, capture_output=True, text=True,
cwd=cwd or os.getcwd(), timeout=timeout
)
output = result.stdout
if result.stderr:
output += "\n" + result.stderr
return result.returncode == 0, output.strip()
except subprocess.TimeoutExpired:
return False, f"命令超时 ({timeout}s)"
except Exception as e:
return False, str(e)
def _find_tool(self, *tools) -> str:
"""查找可用的工具"""
for tool in tools:
# 检查 pip 包
mod_name = tool.replace("-", "_")
ok, _ = self._run(f"python3 -c 'import {mod_name}'", timeout=5)
if ok:
return tool
# 检查命令行
ok, _ = self._run(f"which {tool}", timeout=5)
if ok:
return tool
return None
# ══════════════════════════════════════
# 质量检查
# ══════════════════════════════════════
def _lint(self, target: str, options: str) -> str:
"""代码质量检查"""
# 尝试 ruff > flake8 > pylint
tool = self._find_tool("ruff", "flake8")
if not tool:
return "❌ 未安装 lint 工具。安装: pip install ruff"
if tool == "ruff":
cmd = f"ruff check {target} {options}"
else:
cmd = f"flake8 {target} {options}"
cmd = f"ruff check {target} {options}" if tool == "ruff" else f"flake8 {target} {options}"
ok, output = self._run(cmd, timeout=120)
if not output:
return f"✅ lint 检查通过 ({tool})"
# 统计问题数
lines = output.split('\n')
error_count = len([l for l in lines if l.strip() and not l.startswith(' ')])
status = "❌" if not ok else "⚠️"
return f"{status} lint ({tool}): {error_count} 个问题\n\n{output[:2000]}"
count = len([l for l in lines if l.strip() and not l.startswith(' ')])
return f"{'❌' if not ok else '⚠️'} lint ({tool}): {count} 个问题\n\n{output[:2000]}"
def _format(self, target: str, options: str) -> str:
"""代码格式化"""
tool = self._find_tool("ruff", "black")
if not tool:
return "❌ 未安装格式化工具。安装: pip install ruff"
if tool == "ruff":
cmd = f"ruff format {target} {options}"
else:
cmd = f"black {target} {options}"
cmd = f"ruff format {target} {options}" if tool == "ruff" else f"black {target} {options}"
ok, output = self._run(cmd, timeout=120)
return f"{'✅' if ok else '❌'} 格式化 ({tool})\n{output[:1000]}"
def _test(self, target: str, options: str) -> str:
"""运行测试"""
# 检查 pytest
tool = self._find_tool("pytest")
if not tool:
# 回退到 unittest
ok, output = self._run(f"python3 -m pytest {target} {options} -v", timeout=120)
else:
ok, output = self._run(f"pytest {target} {options} -v", timeout=120)
ok, output = self._run(f"pytest {target} {options} -v", timeout=120)
if not ok and "no tests ran" in output.lower():
return f"⚠️ 未找到测试文件\n{output[:1000]}"
# 提取摘要
for line in output.split('\n'):
if 'passed' in line or 'failed' in line or 'error' in line:
return f"{'✅' if ok else '❌'} 测试结果\n{line}\n\n{output[:2000]}"
if 'passed' in line or 'failed' in line:
return f"{'✅' if ok else '❌'} 测试\n{line}\n\n{output[:2000]}"
return f"{'✅' if ok else '❌'} 测试\n{output[:2000]}"
def _build(self, target: str, options: str) -> str:
"""构建项目"""
if not target or target == ".":
# 自动检测构建系统
if os.path.exists("Makefile"):
cmd = "make"
elif os.path.exists("setup.py"):
cmd = "python3 setup.py build"
elif os.path.exists("pyproject.toml"):
cmd = "python3 -m build"
elif os.path.exists("package.json"):
cmd = "npm run build"
else:
return "❌ 未找到构建配置 (Makefile/setup.py/pyproject.toml/package.json)"
return "❌ 未找到构建配置"
else:
cmd = target
ok, output = self._run(cmd, timeout=300)
return f"{'✅' if ok else '❌'} 构建: {cmd}\n{output[:2000]}"
def _deps(self, target: str, options: str) -> str:
"""检查依赖"""
ok, output = self._run("pip list --outdated --format=json", timeout=30)
if not ok:
return f"❌ 检查依赖失败\n{output}"
return f"❌ 检查失败\n{output}"
try:
outdated = json.loads(output)
if not outdated:
return "✅ 所有依赖都是最新版本"
lines = [f"📦 {len(outdated)} 个依赖可更新:\n"]
for pkg in outdated[:20]:
name = pkg.get('name', '?')
current = pkg.get('version', '?')
latest = pkg.get('latest_version', '?')
lines.append(f" {name:30} {current:12} → {latest}")
if len(outdated) > 20:
lines.append(f"\n ... 还有 {len(outdated) - 20} 个")
lines.append(f" {pkg['name']:30} {pkg['version']:12} → {pkg['latest_version']}")
return '\n'.join(lines)
except json.JSONDecodeError:
return f"📦 依赖状态:\n{output[:2000]}"
def _deps_check(self, target: str, options: str) -> str:
"""安全漏洞检查"""
tool = self._find_tool("pip-audit", "safety")
if not tool:
return "❌ 未安装安全检查工具。安装: pip install pip-audit"
if tool == "pip-audit":
cmd = "pip-audit"
else:
cmd = "safety check"
cmd = "pip-audit" if tool == "pip-audit" else "safety check"
ok, output = self._run(cmd, timeout=60)
return f"{'✅' if ok else '⚠️'} 安全检查 ({tool})\n{output[:2000]}"
def _pre_commit(self, target: str, options: str) -> str:
"""运行 pre-commit"""
ok, output = self._run("pre-commit run --all-files", timeout=120)
return f"{'✅' if ok else '❌'} pre-commit\n{output[:2000]}"
# ══════════════════════════════════════
# 项目分析
# ══════════════════════════════════════
def _complexity(self, target: str, options: str) -> str:
files = self._walk_py(target)
results = []
for fp in files:
try:
with open(fp, 'r', encoding='utf-8') as f:
tree = ast.parse(f.read())
except Exception:
continue
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
complexity = 1
for child in ast.walk(node):
if isinstance(child, (ast.If, ast.While, ast.For, ast.ExceptHandler)):
complexity += 1
elif isinstance(child, ast.BoolOp):
complexity += len(child.values) - 1
if complexity > 5:
rel = os.path.relpath(fp, target)
results.append((rel, node.lineno, node.name, complexity))
if not results:
return "✅ 所有函数复杂度正常 (≤5)"
results.sort(key=lambda x: x[3], reverse=True)
lines = [f"⚠️ 高复杂度函数 ({len(results)} 个):\n"]
for file, line, func, c in results[:20]:
icon = "🔴" if c > 10 else "🟡"
lines.append(f" {icon} {file}:{line} {func}() 复杂度={c}")
return '\n'.join(lines)
def _duplicates(self, target: str, options: str) -> str:
files = self._walk_py(target)
func_defs = {}
for fp in files:
try:
with open(fp, 'r', encoding='utf-8') as f:
tree = ast.parse(f.read())
except Exception:
continue
rel = os.path.relpath(fp, target)
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
func_defs.setdefault(node.name, []).append(f"{rel}:{node.lineno}")
dupes = {k: v for k, v in func_defs.items() if len(v) > 1}
if not dupes:
return "✅ 没有重复函数名"
lines = [f"⚠️ 重复函数名 ({len(dupes)} 个):\n"]
for name, locs in sorted(dupes.items(), key=lambda x: -len(x[1])):
lines.append(f" {name} ({len(locs)} 处):")
for loc in locs[:5]:
lines.append(f" • {loc}")
return '\n'.join(lines)
def _security(self, target: str, options: str) -> str:
files = self._walk_py(target)
patterns = [
(r'\beval\s*\(', 'eval() - 代码注入', 'HIGH'),
(r'\bexec\s*\(', 'exec() - 代码注入', 'HIGH'),
(r'subprocess\.call\s*\(\s*shell\s*=\s*True', 'shell=True - 命令注入', 'HIGH'),
(r'pickle\.loads?\s*\(', 'pickle 反序列化', 'HIGH'),
(r'password\s*=\s*["\']', '硬编码密码', 'HIGH'),
(r'api_key\s*=\s*["\']', '硬编码 API Key', 'HIGH'),
]
risks = []
for fp in files:
try:
with open(fp, 'r', encoding='utf-8') as f:
for i, line in enumerate(f, 1):
for pat, desc, level in patterns:
if re.search(pat, line, re.IGNORECASE):
rel = os.path.relpath(fp, target)
risks.append((rel, i, level, desc, line.strip()))
except Exception:
continue
if not risks:
return "✅ 没有发现安全风险"
lines = [f"🔒 安全风险 ({len(risks)} 个):\n"]
for file, line, level, desc, code in risks[:20]:
lines.append(f" {'🔴' if level == 'HIGH' else '🟡'} {file}:{line} [{level}] {desc}")
lines.append(f" {code[:60]}")
return '\n'.join(lines)
def _metrics(self, target: str, options: str) -> str:
files = self._walk_py(target)
total = blank = comment = code = funcs = classes = max_lines = 0
long_files = []
for fp in files:
try:
with open(fp, 'r', encoding='utf-8') as f:
lines = f.readlines()
except Exception:
continue
fl = len(lines)
total += fl
max_lines = max(max_lines, fl)
if fl > 300:
long_files.append((os.path.relpath(fp, target), fl))
for line in lines:
s = line.strip()
if not s:
blank += 1
elif s.startswith('#'):
comment += 1
else:
code += 1
try:
with open(fp, 'r', encoding='utf-8') as f:
tree = ast.parse(f.read())
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
funcs += 1
elif isinstance(node, ast.ClassDef):
classes += 1
except Exception:
pass
rate = comment / max(code, 1) * 100
lines = [
f"📊 代码指标 ({target}):\n",
f" 文件: {len(files)} | 总行: {total} | 代码: {code}",
f" 注释: {comment} ({rate:.1f}%) | 空行: {blank}",
f" 函数: {funcs} | 类: {classes} | 最大文件: {max_lines} 行",
]
if long_files:
lines.append(f"\n ⚠️ 超过 300 行的文件 ({len(long_files)}):")
for f, l in sorted(long_files, key=lambda x: -x[1])[:5]:
lines.append(f" • {f}: {l} 行")
return '\n'.join(lines)
def _suggest(self, target: str, options: str) -> str:
files = os.listdir(target) if os.path.isdir(target) else []
py_files = self._walk_py(target)
suggestions = []
if 'README.md' not in files:
suggestions.append("📝 添加 README.md")
if 'pyproject.toml' not in files:
suggestions.append("📦 使用 pyproject.toml")
if '.gitignore' not in files:
suggestions.append("🚫 添加 .gitignore")
test_files = [f for f in py_files if 'test' in os.path.basename(f).lower()]
if not test_files and len(py_files) > 5:
suggestions.append("🧪 添加单元测试")
uses_print = sum(1 for f in py_files if 'print(' in open(f, errors='ignore').read())
uses_log = sum(1 for f in py_files if 'logging' in open(f, errors='ignore').read())
if uses_print > uses_log and uses_print > 3:
suggestions.append("📊 考虑用 logging 替代 print")
return "💡 改进建议:\n" + '\n'.join(f" {s}" for s in suggestions) if suggestions else "✅ 项目结构良好"
# ══════════════════════════════════════
# 项目信息与初始化
# ══════════════════════════════════════
def _info(self, target: str, options: str) -> str:
"""项目信息摘要"""
info = []
# 检测项目类型
files = os.listdir(target or ".")
project_type = "未知"
if "pyproject.toml" in files:
project_type = "Python (pyproject.toml)"
info.append("📁 项目类型: Python (pyproject.toml)")
elif "setup.py" in files:
project_type = "Python (setup.py)"
info.append("📁 项目类型: Python (setup.py)")
elif "package.json" in files:
project_type = "Node.js"
elif "Cargo.toml" in files:
project_type = "Rust"
elif "go.mod" in files:
project_type = "Go"
info.append("📁 项目类型: Node.js")
else:
info.append("📁 项目类型: 未知")
info.append(f"📁 项目类型: {project_type}")
# Python 版本
ok, ver = self._run("python3 --version", timeout=5)
if ok:
info.append(f"🐍 Python: {ver}")
# 依赖数
ok, req = self._run("pip list --format=json", timeout=10)
if ok:
try:
deps = json.loads(req)
info.append(f"📦 依赖: {len(deps)} 个")
except:
info.append(f"📦 依赖: {len(json.loads(req))} 个")
except Exception:
pass
# Git 状态
ok, git = self._run("git log --oneline -1", timeout=5)
if ok:
info.append(f"🔀 最新提交: {git}")
@@ -293,47 +411,33 @@ class Plugin:
if ok:
info.append(f"🌿 分支: {branch}")
# 代码统计
ok, stats = self._run("find . -name '*.py' -not -path './.git/*' | wc -l", timeout=10)
if ok:
info.append(f"📄 Python 文件: {stats.strip()} 个")
# 测试文件
ok, tests = self._run("find . -name 'test_*.py' -o -name '*_test.py' | wc -l", timeout=10)
if ok:
info.append(f"🧪 测试文件: {tests.strip()} 个")
py_count = len([f for f in self._walk_py(target or ".")])
info.append(f"📄 Python 文件: {py_count}")
return '\n'.join(info)
def _init_project(self, target: str, options: str) -> str:
"""初始化项目结构"""
project_type = target if target != "." else "python"
if project_type == "python":
ptype = target if target != "." else "python"
if ptype == "python":
dirs = ["src", "tests", "docs"]
files = {
"pyproject.toml": '[project]\nname = "my-project"\nversion = "0.1.0"\n',
"README.md": "# My Project\n",
".gitignore": "__pycache__/\n*.pyc\n.venv/\n",
"tests/__init__.py": "",
"src/__init__.py": "",
}
elif project_type == "web":
elif ptype == "web":
dirs = ["src", "public", "tests"]
files = {
"package.json": '{\n "name": "my-project",\n "version": "1.0.0"\n}\n',
"README.md": "# My Project\n",
".gitignore": "node_modules/\ndist/\n",
}
else:
return f"❌ 不支持的项目类型: {project_type}。支持: python, web"
return f"❌ 不支持: {ptype}。支持: python, web"
created = []
for d in dirs:
if not os.path.exists(d):
os.makedirs(d)
created.append(f"📁 {d}/")
os.makedirs(d, exist_ok=True)
created.append(f"📁 {d}/")
for f, content in files.items():
if not os.path.exists(f):
os.makedirs(os.path.dirname(f) or ".", exist_ok=True)
@@ -341,4 +445,4 @@ class Plugin:
fp.write(content)
created.append(f"📄 {f}")
return f"✅ 已初始化 {project_type} 项目\n" + '\n'.join(created)
return f"✅ 已初始化 {ptype} 项目\n" + '\n'.join(created)
plugins/cmd_executor.py
+97 -168
View File
@@ -1,215 +1,144 @@
import subprocess
"""
命令执行器插件 - 跨平台 Shell 命令执行
"""
import os
import sys
import time
from colorama import Fore, Style
import re
import subprocess
import platform
class Liugin:
"""CMD命令执行插件 - 仅限Windows环境"""
class Plugin:
"""Shell 命令执行器 - 跨平台命令行工具"""
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相关进程
self.usage = """Shell 命令执行器
详细说明:
- run操作: cmd_executor run <命令> - 执行指定的CMD命令
操作:
run <命令> [--timeout 秒] - 执行 Shell 命令 (默认超时 30s)
使用工具的JSON格式示例:
{"action": "use_tool", "tool": "cmd_executor", "args": "run dir"} - 执行dir命令
{"action": "use_tool", "tool": "cmd_executor", "args": "run ipconfig /all"} - 执行ipconfig命令
示例:
run ls -la - 列出当前目录
run python3 --version - 查看 Python 版本
run git status - 查看 Git 状态
run make build --timeout 60 - 执行构建 (60秒超时)
注意:
- 跨平台兼容 (Linux/macOS/Windows)
- 输出限制 5000 字符
- 某些危险命令会被拦截
"""
self.cli = None
重要说明:
- 此工具仅在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环境使用"""
}
return {
"name": "cmd_executor",
"description": "Shell 命令执行器 - 跨平台执行系统命令，支持超时控制",
"keywords": ["命令", "执行", "shell", "cmd", "终端", "命令行", "subprocess"],
"usage": self.usage
}
def get_mcp_definition(self):
return {
"name": "cmd_executor",
"description": "CMD命令执行工具，用于执行系统命令行指令",
"description": "Shell 命令执行器",
"inputSchema": {
"type": "object",
"properties": {
"operation": {
"type": "string",
"enum": ["run"],
"description": "操作类型，固定为 run"
},
"command": {
"type": "string",
"description": "要执行的命令（如 dir, ipconfig, ls -la）"
}
"command": {"type": "string", "description": "要执行的命令"},
"timeout": {"type": "integer", "description": "超时秒数", "default": 30}
},
"required": ["operation", "command"]
"required": ["command"]
}
}
def convert_mcp_args(self, arguments):
op = arguments.get("operation", "run")
cmd = arguments.get("command", "")
return f"{op} {cmd}"
timeout = arguments.get("timeout", "")
result = f"run {cmd}"
if timeout:
result += f" --timeout {timeout}"
return result
def handle(self, args):
"""处理CMD命令执行请求"""
# 危险命令黑名单
BLOCKED = {
'rm -rf /', 'rm -rf ~', 'mkfs', 'dd if=', 'wipefs', 'shred',
':(){ :|:& };:', 'chmod -R 777 /', 'mv / ', 'wget http',
}
def handle(self, args: str) -> str:
try:
# 解析参数
parts = args.strip().split(maxsplit=1)
if len(parts) < 1:
return "错误：参数不足。请提供操作类型。\n可用操作: run"
if not parts:
return "错误：请提供操作。可用: 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 operation != "run":
return f"错误：不支持的操作 '{operation}'。可用: run"
if len(parts) < 2:
return "错误：请提供要执行的命令"
rest = parts[1]
# 解析 --timeout
timeout = 30
timeout_match = re.search(r'--timeout\s+(\d+)', rest)
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
# 执行命令
rest = re.sub(r'--timeout\s+\d+', '', rest).strip()
if not rest:
return "错误：请提供要执行的命令"
# 安全检查
for blocked in self.BLOCKED:
if blocked in rest:
return f"🚫 安全拦截: 检测到危险命令模式 '{blocked}'"
return self._execute(rest, timeout)
except Exception as e:
return f"命令执行错误: {str(e)}"
def _execute(self, command: str, timeout: int) -> str:
try:
# Windows 下用 shell=True，Unix 下直接执行
use_shell = platform.system() == 'Windows'
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
timeout=timeout, # 使用自定义超时时间
encoding='utf-8', # 指定编码
errors='replace' # 处理编码错误
timeout=timeout,
encoding='utf-8',
errors='replace'
)
# 获取输出
stdout = result.stdout.strip()
stderr = result.stderr.strip()
return_code = result.returncode
rc = result.returncode
output = ""
if stdout:
output += f"命令输出:\n{stdout}\n"
output += stdout + "\n"
if stderr:
output += f"错误信息:\n{stderr}\n"
if return_code != 0:
output += f"命令退出码: {return_code}\n"
output += f"[stderr]\n{stderr}\n"
if rc != 0:
output += f"[exit code: {rc}]\n"
if not output:
output = f"命令执行完成，无输出。(超时设置: {timeout}秒)"
output = f"命令执行完成，无输出 (超时: {timeout}s)"
# 限制输出长度为5000字符
if len(output) > 5000:
output = output[:5000] + f"\n... (输出已截断，超时设置: {timeout}秒)"
output = output[:5000] + f"\n... (截断，超时: {timeout}s)"
return output.strip()
return output
except subprocess.TimeoutExpired:
return f"错误：命令执行超时（超过{timeout}秒），已取消执行。"
except ValueError:
return f"错误：超时值无效，已使用默认30秒超时。命令: {command}"
return f"⏱️ 命令超时 ({timeout}s): {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()
return f"执行失败: {str(e)}"
plugins/code_editor.py
+482 -336
View File
File diff suppressed because it is too large. Load diff
plugins/code_search.py
-472
View File
@@ -1,472 +0,0 @@
"""
代码搜索与理解插件 - 代码库级别的智能搜索
对标 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}%"""
plugins/file_manager.py
+194 -638
View File
@@ -1,161 +1,82 @@
"""
文件管理器插件 - 小狸自有协议版本
支持文件和目录的创建、读取、写入、删除、移动、复制、搜索等操作
文件管理器插件 - 文件系统操作
聚焦文件系统操作，代码编辑交给 code_editor
"""
import os
import json
import shutil
from datetime import datetime
from pathlib import Path
from typing import Optional, Dict, Any, List
from typing import List
class Liugin:
"""文件管理器插件"""
class Plugin:
"""文件管理器 - 文件和目录的增删改查"""
def __init__(self):
self.usage = """文件管理器使用方法：
file_manager <操作> <参数>
self.usage = """文件管理器
操作类型:
- 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"}
操作:
list <路径> [-p "模式"] - 列出目录内容
read <文件> [-n 行数] - 读取文件内容
write <文件> -c "内容" - 写入文件
append <文件> -c "内容" - 追加内容
copy <源> <目标> - 复制文件/目录
move <源> <目标> - 移动/重命名
delete <路径> [-r] - 删除文件/目录
search <目录> [-p 模式] [-c 内容] - 搜索文件
info <路径> - 文件详细信息
mkdir <路径> - 创建目录
"""
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"],
"description": "文件管理器 - 文件和目录的创建、读写、复制、移动、删除、搜索",
"keywords": ["文件", "目录", "管理", "file", "复制", "删除", "搜索", "mkdir",
"读取", "写入", "移动", "重命名"],
"usage": self.usage
}
def get_mcp_definition(self) -> Dict[str, Any]:
"""返回符合 MCP 标准的工具定义"""
def get_mcp_definition(self):
return {
"name": "file_manager",
"description": "文件和目录管理工具，支持创建、读取、写入、删除、移动、复制、搜索文件等操作。操作类型: list(列出目录), read(读取文件), write(写入文件), append(追加内容), copy(复制), move(移动/重命名), delete(删除), search(搜索), info(文件信息), mkdir(创建目录), modify(修改行), insert(插入行), delete-line(删除行), replace(替换文本)",
"description": "文件管理器",
"inputSchema": {
"type": "object",
"properties": {
"operation": {
"type": "string",
"enum": ["list", "read", "write", "append", "copy", "move", "delete", "search", "info", "mkdir", "modify", "insert", "delete-line", "replace"],
"enum": ["list", "read", "write", "append", "copy",
"move", "delete", "search", "info", "mkdir"],
"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
}
"path": {"type": "string"},
"content": {"type": "string"},
"destination": {"type": "string"},
"pattern": {"type": "string"},
"recursive": {"type": "boolean", "default": False}
},
"required": ["operation"]
}
}
def convert_mcp_args(self, arguments: Dict[str, Any]) -> str:
"""将 MCP 参数转换为命令行格式"""
def convert_mcp_args(self, arguments):
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)
path = arguments.get("path", "")
return f"{op} {path}".strip()
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"
return "错误：请提供操作类型"
operation = parts[0].lower()
handlers = {
"list": self._op_list,
"read": self._op_read,
@@ -167,627 +88,262 @@ JSON调用格式:
"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 f"错误：不支持的操作 '{operation}'"
return handler(parts[1:])
except Exception as e:
return f"文件操作错误: {str(e)}"
def _parse_args(self, args: str) -> List[str]:
"""解析命令行参数，支持引号"""
parts = []
current = ""
in_quotes = 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:
parts, current, in_quotes, qc = [], "", False, None
for ch in args:
if ch in ('"', "'") and not in_quotes:
in_quotes, qc = True, ch
elif ch == qc and in_quotes:
in_quotes, qc = False, None
elif ch == ' ' and not in_quotes:
if current:
parts.append(current)
current = ""
else:
current += char
current += ch
if current:
parts.append(current)
return parts
def _op_list(self, args: List[str]) -> str:
"""列出目录内容 - 默认只列出当前目录，不递归"""
path = "."
pattern = None
path, pattern = ".", None
i = 0
while i < len(args):
if args[i] == "-p" and i + 1 < len(args):
pattern = args[i + 1]
i += 1
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)
full = os.path.join(path, item)
if pattern and not Path(item).match(pattern):
continue
if os.path.isdir(full_path):
if os.path.isdir(full):
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"
size = os.path.getsize(full)
size_str = f"{size/1024:.1f}KB" if size > 1024 else f"{size}B"
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})")
dirs = sum(1 for x in items if x[0] == 'dir')
files = len(items) - dirs
result = [f"📁 {path}", "─" * 50]
for t, name, size in items[:100]:
result.append(f" {'📁' if t == 'dir' else '📄'} {name}{'/' if t == 'dir' else f' ({size})'}")
if len(items) > 100:
result.append(f" ... 还有 {len(items) - 100} 项")
result.append("─" * 50)
result.append(f"共 {dir_count} 个文件夹, {file_count} 个文件")
result.append(f"共 {dirs} 个文件夹, {files} 个文件")
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
file_path = os.path.abspath(args[0])
max_lines = None
i = 1
while i < len(args):
if args[i] == "-n" and i + 1 < len(args):
try:
lines = int(args[i + 1])
max_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 参数限制行数"
return f"错误：文件不存在: {file_path}"
size = os.path.getsize(file_path)
if size > 10 * 1024 * 1024:
return f"文件过大 ({size/1024/1024:.1f}MB)，请用 -n 限制行数"
try:
with open(file_path, 'r', encoding='utf-8') as f:
if lines:
content_lines = []
line_count = 0
if max_lines:
lines = []
total = 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
total = i + 1
if i < max_lines:
lines.append(f"{i+1:4d} | {line.rstrip()}")
content = '\n'.join(lines)
header = f"📄 {file_path} ({size}B, 前 {min(max_lines, total)}/{total} 行)"
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)
content = '\n'.join(f"{i+1:4d} | {l.rstrip()}" for i, l in enumerate(all_lines))
header = f"📄 {file_path} ({size}B, {len(all_lines)} 行)"
return f"{header}\n{'─' * 50}\n{content}"
except UnicodeDecodeError:
return f"错误：无法解码文件（可能是二进制文件或编码不正确）"
except Exception as e:
return f"读取文件失败: {e}"
return "错误：无法解码（可能是二进制文件）"
def _op_write(self, args: List[str]) -> str:
"""写入文件"""
if len(args) < 2:
return "错误：请提供文件路径和内容"
file_path = args[0]
return "错误：格式: write <文件> -c <内容>"
fp = os.path.abspath(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
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}"
return "错误：请用 -c 提供内容"
os.makedirs(os.path.dirname(fp) or ".", exist_ok=True)
with open(fp, 'w', encoding='utf-8') as f:
f.write(content)
return f"✅ 已写入: {fp} ({len(content)} 字符)"
def _op_append(self, args: List[str]) -> str:
"""追加内容到文件"""
if len(args) < 2:
return "错误：请提供文件路径和内容"
file_path = args[0]
return "错误：格式: append <文件> -c <内容>"
fp = os.path.abspath(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
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}"
return "错误：请用 -c 提供内容"
if not os.path.isfile(fp):
return f"错误：文件不存在: {fp}"
with open(fp, 'a', encoding='utf-8') as f:
f.write(content)
return f"✅ 已追加: {fp}"
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}"
return "错误：请提供源和目标路径"
src, dst = os.path.abspath(args[0]), os.path.abspath(args[1])
if not os.path.exists(src):
return f"错误：源不存在: {src}"
os.makedirs(os.path.dirname(dst) or ".", exist_ok=True)
if os.path.isfile(src):
shutil.copy2(src, dst)
else:
shutil.copytree(src, dst, dirs_exist_ok=True)
return f"✅ 已复制:\n {src}\n → {dst}"
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}"
return "错误：请提供源和目标路径"
src, dst = os.path.abspath(args[0]), os.path.abspath(args[1])
if not os.path.exists(src):
return f"错误：源不存在: {src}"
os.makedirs(os.path.dirname(dst) or ".", exist_ok=True)
shutil.move(src, dst)
return f"✅ 已移动:\n {src}\n → {dst}"
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}"
return f"错误：不存在: {path}"
if os.path.isfile(path):
os.remove(path)
return f"✅ 已删除文件: {path}"
elif os.path.isdir(path):
if recursive:
shutil.rmtree(path)
return f"✅ 已删除目录: {path}"
elif not os.listdir(path):
os.rmdir(path)
return f"✅ 已删除空目录: {path}"
else:
return "错误：目录不为空，用 -r 递归删除"
def _op_search(self, args: List[str]) -> str:
"""搜索文件"""
if not args:
return "错误：请提供搜索目录"
directory = args[0]
pattern = None
content_search = None
directory = os.path.abspath(args[0])
pattern, content_search = None, None
i = 1
while i < len(args):
if args[i] == "-p" and i + 1 < len(args):
pattern = args[i + 1]
i += 1
pattern = args[i + 1]; i += 1
elif args[i] == "-c" and i + 1 < len(args):
content_search = args[i + 1]
i += 1
content_search = args[i + 1]; i += 1
i += 1
directory = os.path.abspath(directory)
if not os.path.exists(directory):
if not os.path.isdir(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):
for root, dirs, files in os.walk(directory):
for fn in files:
if pattern and not Path(fn).match(pattern):
continue
fp = os.path.join(root, fn)
rel = os.path.relpath(fp, directory)
if content_search:
try:
with open(fp, 'r', encoding='utf-8') as f:
for i, line in enumerate(f, 1):
if content_search in line:
results.append(f"{rel}:{i}: {line.strip()[:80]}")
if len(results) >= 50:
break
except (UnicodeDecodeError, IOError):
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:
else:
results.append(rel)
if len(results) >= 50:
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}"
if len(results) >= 50:
break
return f"搜索结果 ({len(results)}):\n" + '\n'.join(results) if results else "未找到匹配"
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}"
return f"错误：不存在: {path}"
stat = os.stat(path)
result = [
f"路径: {path}",
f"类型: {'目录' if os.path.isdir(path) else '文件'}",
f"大小: {stat.st_size:,} 字节 ({stat.st_size/1024:.1f} KB)",
f"修改: {datetime.fromtimestamp(stat.st_mtime):%Y-%m-%d %H:%M:%S}",
f"访问: {datetime.fromtimestamp(stat.st_atime):%Y-%m-%d %H:%M:%S}",
]
if os.path.isfile(path):
ext = os.path.splitext(path)[1]
result.append(f"扩展名: {ext or '无'}")
return '\n'.join(result)
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"))
return f"错误：已存在: {path}"
os.makedirs(path, exist_ok=True)
return f"✅ 已创建: {path}"
plugins/git_tools.py
+255 -171
View File
@@ -1,38 +1,46 @@
"""
Git 集成插件 - 版本控制能力
对标 Claude Code 的 Git 工作流支持
Git 工具插件 - 统一的版本控制能力
合并原 git_tools + git_workflow，消除重复
"""
import os
import re
import subprocess
from typing import Optional, Tuple
from typing import Tuple, Optional
class Plugin:
"""Git 集成工具 - 提供完整的 Git 操作支持"""
"""Git 版本控制 - 基础操作 + 工作流自动化"""
def __init__(self):
self.usage = """Git 集成工具
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> [目录] - 克隆仓库
基础操作:
status - 查看仓库状态
diff [文件路径] - 查看未暂存变更
diff --cached - 查看已暂存变更
log [数量] - 提交历史 (默认10条)
log --oneline [数量] - 简洁提交历史
add <文件路径> - 暂存文件
add . - 暂存所有变更
commit <提交信息> - 提交变更
branch - 查看分支列表
branch <名称> - 创建新分支
checkout <分支名> - 切换分支
checkout -b <分支名> - 创建并切换分支
stash - 暂存当前变更
stash pop - 恢复暂存的变更
show <commit> - 查看指定提交
blame <文件路径> - 逐行修改记录
remote - 查看远程仓库
init - 初始化仓库
clone <URL> [目录] - 克隆仓库
工作流:
smart-commit [消息] - 智能提交 (自动 add + 规范化消息)
changelog [版本] - 生成 CHANGELOG
summary - 变更摘要 (未提交改动)
contributors - 贡献者统计
stale [天数] - 查找长期未更新文件
"""
self.cli = None
@@ -42,8 +50,9 @@ class Plugin:
def get_tool_info(self):
return {
"name": "git_tools",
"description": "Git 版本控制工具 - 查看状态、提交、分支管理、查看历史等",
"keywords": ["git", "版本控制", "提交", "分支", "commit", "branch", "status"],
"description": "Git 版本控制 - 状态查看、提交、分支管理、diff、历史、blame、工作流自动化",
"keywords": ["git", "版本控制", "提交", "分支", "commit", "branch", "status", "diff",
"changelog", "log", "blame", "stash", "checkout", "merge"],
"usage": self.usage
}
@@ -55,8 +64,7 @@ class Plugin:
"type": "object",
"properties": {
"operation": {"type": "string", "description": "Git 操作"},
"args": {"type": "string", "description": "操作参数"},
"cwd": {"type": "string", "description": "工作目录"}
"args": {"type": "string", "description": "操作参数"}
},
"required": ["operation"]
}
@@ -67,95 +75,114 @@ class Plugin:
args = arguments.get("args", "")
return f"{op} {args}".strip()
# ── 执行 Git 命令 ──
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
capture_output=True, text=True,
cwd=cwd, timeout=30
)
output = result.stdout
if result.stderr:
output += result.stderr
return result.returncode == 0, output.strip()
output = (result.stdout + result.stderr).strip()
return result.returncode == 0, output
except FileNotFoundError:
return False, "错误：git 未安装。请先安装 Git。"
return False, "git 未安装，请先安装 Git"
except subprocess.TimeoutExpired:
return False, "错误：Git 命令执行超时"
return False, "Git 命令执行超时"
except Exception as e:
return False, f"错误：{str(e)}"
return False, str(e)
def _run_shell(self, cmd: str, cwd: Optional[str] = None) -> Tuple[bool, str]:
try:
result = subprocess.run(
cmd, shell=True, capture_output=True, text=True,
cwd=cwd or os.getcwd(), timeout=30
)
return result.returncode == 0, (result.stdout + result.stderr).strip()
except Exception as e:
return False, str(e)
def _get_cwd(self):
if self.cli and hasattr(self.cli, 'current_dir'):
return self.cli.current_dir
return None
# ── 路由 ──
def handle(self, args: str) -> str:
try:
parts = args.strip().split(maxsplit=1)
if not parts:
return self._git_status("")
return self._status()
operation = parts[0].lower()
op = 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
cwd = self._get_cwd()
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,
# 基础
"status": lambda r: self._status(cwd),
"diff": lambda r: self._diff(r, cwd),
"log": lambda r: self._log(r, cwd),
"add": lambda r: self._add(r, cwd),
"commit": lambda r: self._commit(r, cwd),
"branch": lambda r: self._branch(r, cwd),
"checkout": lambda r: self._checkout(r, cwd),
"stash": lambda r: self._stash(r, cwd),
"show": lambda r: self._show(r, cwd),
"blame": lambda r: self._blame(r, cwd),
"remote": lambda r: self._remote(cwd),
"init": lambda r: self._init(r, cwd),
"clone": lambda r: self._clone(r, cwd),
# 工作流
"smart-commit": lambda r: self._smart_commit(r, cwd),
"changelog": lambda r: self._changelog(r, cwd),
"summary": lambda r: self._summary(cwd),
"contributors": lambda r: self._contributors(cwd),
"stale": lambda r: self._stale(r, cwd),
}
handler = handlers.get(operation)
if not handler:
# 尝试直接执行 git 命令
return self._git_raw(args, cwd)
handler = handlers.get(op)
if handler:
return handler(rest)
return handler(rest, cwd)
# 回退：直接执行 git 命令
ok, output = self._run_git(args.split(), cwd)
return output if ok else f"Git 错误: {output}"
except Exception as e:
return f"Git 工具错误: {str(e)}"
def _git_status(self, args: str, cwd=None) -> str:
# ── 基础操作 ──
def _status(self, cwd=None) -> str:
ok, output = self._run_git(["status", "--porcelain"], cwd)
if not ok:
return f"Git 错误: {output}"
if not output:
return "✅ 工作区干净，没有未提交的变更"
# 获取分支
_, branch = self._run_git(["branch", "--show-current"], cwd)
return f"✅ 工作区干净 (分支: {branch})"
# 解析状态
lines = output.split('\n')
staged = []
unstaged = []
untracked = []
staged, unstaged, untracked = [], [], []
for line in lines:
if len(line) < 4:
continue
status = line[:2]
filename = line[3:]
s, filename = line[:2], line[3:]
if s[0] in 'MADRC':
staged.append(f" ✓ {s[0]} {filename}")
if s[1] in 'MD':
unstaged.append(f" ! {s[1]} {filename}")
if s == '??':
untracked.append(f" ? {filename}")
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}")
_, branch = self._run_git(["branch", "--show-current"], cwd)
result = [f"📊 Git 状态 (分支: {branch})"]
result = ["📊 Git 状态:"]
if staged:
result.append(f"\n已暂存 ({len(staged)}):")
result.extend(staged)
@@ -166,37 +193,30 @@ class Plugin:
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)
def _diff(self, args: str, cwd=None) -> str:
args = args.strip()
if args == "--cached":
git_args = ["diff", "--cached"]
elif args:
git_args = ["diff", args]
else:
ok, output = self._run_git(["diff"], cwd)
git_args = ["diff"]
ok, output = self._run_git(git_args, cwd)
if not ok:
return f"Git 错误: {output}"
if not output:
return "没有变更"
# 限制输出长度
if len(output) > 5000:
output = output[:5000] + f"\n... (输出截断，共 {len(output)} 字符)"
output = output[:5000] + f"\n... (截断，共 {len(output)} 字符)"
return output
return f"📝 变更:\n{output}"
def _git_log(self, args: str, cwd=None) -> str:
def _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)
@@ -210,137 +230,201 @@ class Plugin:
if not ok:
return f"Git 错误: {output}"
if not output:
return "暂无提交历史"
return output or "暂无提交历史"
return f"📜 提交历史:\n{output}"
def _git_add(self, args: str, cwd=None) -> str:
def _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}" if ok else f"Git add 失败: {output}"
return f"✅ 已暂存: {target}"
def _git_commit(self, args: str, cwd=None) -> str:
def _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}" if ok else f"Git commit 失败: {output}"
return f"✅ 提交成功:\n{output}"
def _git_branch(self, args: str, cwd=None) -> str:
def _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}"
return f"🌿 分支列表:\n{output}" if ok else f"Git 错误: {output}"
else:
ok, output = self._run_git(["branch", args.strip()], cwd)
if not ok:
return f"创建分支失败: {output}"
return f"✅ 已创建分支: {args.strip()}"
return f"✅ 已创建分支: {args.strip()}" if ok else f"创建分支失败: {output}"
def _git_checkout(self, args: str, cwd=None) -> str:
def _checkout(self, args: str, cwd=None) -> str:
args = args.strip()
if args.startswith("-b "):
branch = args[3:].strip()
ok, output = self._run_git(["checkout", "-b", branch], cwd)
else:
branch = args.strip()
branch = args
ok, output = self._run_git(["checkout", branch], cwd)
return f"✅ 已切换到: {branch}" if ok else f"切换失败: {output}"
if not ok:
return f"切换分支失败: {output}"
return f"✅ 已切换到分支: {branch}"
def _git_stash(self, args: str, cwd=None) -> str:
def _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)
return f"✅ {output}" if ok else f"Git stash 失败: {output}"
if not ok:
return f"Git stash 失败: {output}"
return f"✅ {output}"
def _git_show(self, args: str, cwd=None) -> str:
def _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... (输出截断)"
output = output[:3000] + "\n... (截断)"
return output
def _git_blame(self, args: str, cwd=None) -> str:
def _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 = []
result = []
author = ""
for line in lines:
if line.startswith('author '):
author = line[7:]
elif line.startswith('\t'):
simplified.append(f" {author}: {line[1:]}")
result.append(f" {author}: {line[1:]}")
if len(simplified) > 100:
simplified = simplified[:100]
simplified.append(f" ... (共 {len(simplified)} 行)")
if len(result) > 100:
result = result[:100]
result.append(f" ... (共 {len(result)} 行)")
return f"📝 {path} 修改记录:\n" + '\n'.join(simplified)
return f"📝 {path} 修改记录:\n" + '\n'.join(result)
def _git_remote(self, args: str, cwd=None) -> str:
def _remote(self, 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}"
return output or "没有配置远程仓库"
def _git_init(self, args: str, cwd=None) -> str:
def _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}"
return f"✅ {output}" if ok else f"Git init 失败: {output}"
def _git_clone(self, args: str, cwd=None) -> str:
def _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)
cmd = ["clone", parts[0]]
if len(parts) > 1:
cmd.append(parts[1])
ok, output = self._run_git(cmd, cwd)
if not ok:
return f"Git clone 失败: {output}"
return f"✅ 克隆成功:\n{output}"
return f"✅ 克隆成功:\n{output}" if ok else f"Git clone 失败: {output}"
def _git_raw(self, args: str, cwd=None) -> str:
"""直接执行 git 命令"""
parts = args.split()
ok, output = self._run_git(parts, cwd)
# ── 工作流 ──
def _smart_commit(self, message: str, cwd=None) -> str:
"""智能提交：自动 add + 规范化消息"""
ok, status = self._run_git(["status", "--porcelain"], cwd)
if not status:
return "✅ 工作区干净，没有需要提交的变更"
self._run_git(["add", "-A"], cwd)
if not message:
message = self._auto_commit_message(status)
# conventional commits 检查
if not re.match(r'^(feat|fix|docs|style|refactor|perf|test|build|ci|chore|revert)(\(.+\))?: .+', message):
if 'test' in message.lower():
message = f"test: {message}"
elif 'doc' in message.lower() or 'readme' in message.lower():
message = f"docs: {message}"
else:
message = f"feat: {message}"
ok, output = self._run_git(["commit", "-m", message], cwd)
if ok:
_, hash_out = self._run_git(["log", "--oneline", "-1"], cwd)
return f"✅ 已提交: {message}\n {hash_out}"
return f"❌ 提交失败: {output}"
def _auto_commit_message(self, status: str) -> str:
lines = status.strip().split('\n')
added = len([l for l in lines if l.startswith('A ') or l.startswith('A') and l[1] == ' '])
modified = len([l for l in lines if l.startswith('M ') or (len(l) > 1 and l[1] == 'M')])
deleted = len([l for l in lines if l.startswith('D ') or (len(l) > 1 and l[1] == 'D')])
parts = []
if added:
parts.append(f"添加 {added} 个文件")
if modified:
parts.append(f"修改 {modified} 个文件")
if deleted:
parts.append(f"删除 {deleted} 个文件")
return ", ".join(parts) if parts else "更新代码"
def _changelog(self, version: str, cwd=None) -> str:
ok, output = self._run_shell(
'git log --oneline --format="%s" | head -50', cwd)
if not ok:
return f"Git 错误: {output}"
return output
return f"❌ {output}"
commits = output.strip().split('\n')
categories = {
'feat': [], 'fix': [], 'docs': [], 'refactor': [],
'test': [], 'chore': [], 'other': []
}
for commit in commits:
matched = False
for prefix in categories:
if commit.startswith(prefix + ':') or commit.startswith(prefix + '('):
categories[prefix].append(commit)
matched = True
break
if not matched:
categories['other'].append(commit)
labels = {
'feat': '🚀 新增', 'fix': '🐛 修复', 'docs': '📝 文档',
'refactor': '♻️ 重构', 'test': '🧪 测试', 'chore': '🔧 杂项',
'other': '📦 其他'
}
ver = version or "未发布"
result = [f"## {ver}\n"]
for prefix, label in labels.items():
items = categories[prefix]
if items:
result.append(f"\n### {label}")
for item in items[:10]:
result.append(f"- {item}")
return '\n'.join(result)
def _summary(self, cwd=None) -> str:
ok, output = self._run_git(["diff", "--stat"], cwd)
if not output:
return "✅ 没有未提交的变更"
_, short = self._run_git(["diff", "--shortstat"], cwd)
return f"📊 变更摘要:\n\n{short}\n\n{output[:2000]}"
def _contributors(self, cwd=None) -> str:
ok, output = self._run_git(["shortlog", "-sn", "--all", "--no-merges"], cwd)
if not ok:
return f"❌ {output}"
lines = output.strip().split('\n')
result = [f"👥 贡献者 ({len(lines)} 人):\n"]
for line in lines[:20]:
result.append(f" {line.strip()}")
return '\n'.join(result)
def _stale(self, args: str, cwd=None) -> str:
days = int(args) if args.strip().isdigit() else 90
ok, output = self._run_shell(
f"git log --diff-filter=A --format='%aI %n' --since='{days} days ago' | head -20", cwd)
return f"📁 最近 {days} 天新增的文件:\n{output[:2000]}" if ok else f"❌ {output}"
plugins/git_workflow.py
-284
View File
@@ -1,284 +0,0 @@
"""
Git 工作流自动化 - commit 规范、分支管理、PR 生成
"""
import os
import subprocess
import re
from typing import Tuple
class Plugin:
"""Git 工作流自动化 - 规范提交、分支管理、变更摘要"""
def __init__(self):
self.usage = """Git 工作流自动化
操作类型:
commit [消息] - 智能提交 (自动 add + 规范化消息)
log [数量] - 格式化提交历史
changelog [版本] - 生成 CHANGELOG
summary - 变更摘要 (本次未提交的改动)
blame [文件] - 逐行追溯
contributors - 贡献者统计
stale [天数] - 查找长期未更新的文件
hooks - 管理 git hooks
"""
self.cli = None
def set_cli(self, cli):
self.cli = cli
def get_tool_info(self):
return {
"name": "git_workflow",
"description": "Git 工作流自动化 - 规范提交、变更摘要、贡献者统计",
"keywords": ["git", "commit", "changelog", "提交", "变更", "工作流"],
"usage": self.usage
}
def get_mcp_definition(self):
return {
"name": "git_workflow",
"description": "Git 工作流自动化工具",
"inputSchema": {
"type": "object",
"properties": {
"operation": {
"type": "string",
"enum": ["commit", "log", "changelog", "summary", "blame", "contributors", "stale", "hooks"],
"description": "操作类型"
},
"args": {
"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(self, cmd: str, cwd: str = None) -> Tuple[bool, str]:
try:
result = subprocess.run(
cmd, shell=True, capture_output=True, text=True,
cwd=cwd or os.getcwd(), timeout=30
)
return result.returncode == 0, (result.stdout + result.stderr).strip()
except Exception as e:
return False, str(e)
def handle(self, args: str) -> str:
parts = args.strip().split(maxsplit=1)
if not parts:
return self.usage
op = parts[0].lower()
rest = parts[1] if len(parts) > 1 else ""
handlers = {
"commit": self._commit,
"log": self._log,
"changelog": self._changelog,
"summary": self._summary,
"blame": self._blame,
"contributors": self._contributors,
"stale": self._stale,
"hooks": self._hooks,
}
handler = handlers.get(op)
if not handler:
return f"错误: 不支持的操作 '{op}'"
try:
return handler(rest)
except Exception as e:
return f"执行错误: {e}"
def _commit(self, message: str) -> str:
"""智能提交"""
# 检查是否有变更
ok, status = self._run("git status --porcelain")
if not status:
return "✅ 工作区干净，没有需要提交的变更"
# 自动 add
self._run("git add -A")
# 规范化提交消息
if not message:
# 根据变更自动生成消息
message = self._auto_commit_message(status)
# 检查是否符合 conventional commits
if not re.match(r'^(feat|fix|docs|style|refactor|perf|test|build|ci|chore|revert)(\(.+\))?: .+', message):
# 自动添加前缀
if any('.py' in l for l in status.split('\n')):
if 'test' in message.lower():
message = f"test: {message}"
elif 'doc' in message.lower() or 'readme' in message.lower():
message = f"docs: {message}"
else:
message = f"feat: {message}"
# 提交
ok, output = self._run(f'git commit -m "{message}"')
if ok:
# 获取 commit hash
_, hash_out = self._run("git log --oneline -1")
return f"✅ 已提交: {message}\n {hash_out}"
return f"❌ 提交失败: {output}"
def _auto_commit_message(self, status: str) -> str:
"""根据变更自动生成提交消息"""
lines = status.strip().split('\n')
added = [l[3:] for l in lines if l.startswith('A')]
modified = [l[3:] for l in lines if l.startswith('M')]
deleted = [l[3:] for l in lines if l.startswith('D')]
parts = []
if added:
parts.append(f"添加 {len(added)} 个文件")
if modified:
parts.append(f"修改 {len(modified)} 个文件")
if deleted:
parts.append(f"删除 {len(deleted)} 个文件")
return ", ".join(parts) if parts else "更新代码"
def _log(self, args: str) -> str:
"""格式化提交历史"""
count = int(args) if args.strip().isdigit() else 10
ok, output = self._run(f'git log --oneline -{count} --format="%h %s (%ar, %an)"')
if not ok:
return f"❌ {output}"
lines = output.strip().split('\n')
result = [f"📜 最近 {len(lines)} 次提交:\n"]
for line in lines:
result.append(f" {line}")
return '\n'.join(result)
def _changelog(self, version: str) -> str:
"""生成 CHANGELOG"""
# 获取所有提交
ok, output = self._run('git log --oneline --format="%s" | head -50')
if not ok:
return f"❌ {output}"
commits = output.strip().split('\n')
# 分类
categories = {
'feat': [], 'fix': [], 'docs': [], 'refactor': [],
'test': [], 'chore': [], 'other': []
}
for commit in commits:
matched = False
for prefix in categories:
if commit.startswith(prefix + ':') or commit.startswith(prefix + '('):
categories[prefix].append(commit)
matched = True
break
if not matched:
categories['other'].append(commit)
# 生成
ver = version or "未发布"
result = [f"## {ver}\n"]
labels = {
'feat': '🚀 新增', 'fix': '🐛 修复', 'docs': '📝 文档',
'refactor': '♻️ 重构', 'test': '🧪 测试', 'chore': '🔧 杂项',
'other': '📦 其他'
}
for prefix, label in labels.items():
items = categories[prefix]
if items:
result.append(f"\n### {label}")
for item in items[:10]:
result.append(f"- {item}")
return '\n'.join(result)
def _summary(self, args: str) -> str:
"""变更摘要"""
ok, output = self._run("git diff --stat")
if not output:
return "✅ 没有未提交的变更"
ok2, short = self._run("git diff --shortstat")
return f"📊 变更摘要:\n\n{short}\n\n{output[:2000]}"
def _blame(self, args: str) -> str:
"""逐行追溯"""
if not args.strip():
return "错误: 请提供文件路径"
ok, output = self._run(f"git blame --line-porcelain {args} | grep -E '^(author | )'")
if not ok:
return f"❌ {output}"
# 简化输出
lines = output.strip().split('\n')
result = [f"📝 {args} 修改记录:\n"]
author = ""
for line in lines:
if line.startswith('author '):
author = line[7:]
elif line.startswith('\t'):
result.append(f" {author:20} {line[1:]}")
return '\n'.join(result[:50])
def _contributors(self, args: str) -> str:
"""贡献者统计"""
ok, output = self._run("git shortlog -sn --all --no-merges")
if not ok:
return f"❌ {output}"
lines = output.strip().split('\n')
result = [f"👥 贡献者 ({len(lines)} 人):\n"]
for line in lines[:20]:
result.append(f" {line.strip()}")
return '\n'.join(result)
def _stale(self, args: str) -> str:
"""查找长期未更新的文件"""
days = int(args) if args.strip().isdigit() else 90
ok, output = self._run(f"git log --diff-filter=A --format='%aI %n' --since='{days} days ago' | head -20")
if not ok:
return f"❌ {output}"
return f"📁 最近 {days} 天新增的文件:\n{output[:2000]}"
def _hooks(self, args: str) -> str:
"""管理 git hooks"""
hooks_dir = ".git/hooks"
if not os.path.exists(hooks_dir):
return "❌ 不是 git 仓库"
hooks = os.listdir(hooks_dir)
active = [h for h in hooks if not h.endswith('.sample') and os.path.isfile(os.path.join(hooks_dir, h))]
result = ["🪝 Git Hooks:\n"]
for h in hooks:
path = os.path.join(hooks_dir, h)
if h.endswith('.sample'):
result.append(f" ⭕ {h.replace('.sample', '')} (未激活)")
elif os.path.isfile(path):
result.append(f" ✅ {h} (活跃)")
if not active:
result.append("\n 没有活跃的 hooks。可以添加 pre-commit 等钩子。")
return '\n'.join(result)
plugins/project_analyzer.py
-443
View File
@@ -1,443 +0,0 @@
"""
项目分析插件 - 代码质量分析、依赖分析、架构建议
"""
import os
import ast
import re
import json
from typing import Dict, List, Tuple
class Plugin:
"""项目分析工具 - 深度分析代码质量、依赖关系、架构问题"""
def __init__(self):
self.usage = """项目分析工具
操作类型:
complexity [目录] - 圈复杂度分析
imports [文件] - 依赖关系图
duplicates [目录] - 重复代码检测
todo [目录] - TODO/FIXME 汇总
security [目录] - 安全风险扫描
metrics [目录] - 代码质量指标
suggest [目录] - 改进建议
"""
self.cli = None
def set_cli(self, cli):
self.cli = cli
def get_tool_info(self):
return {
"name": "project_analyzer",
"description": "项目分析工具 - 代码质量、依赖关系、安全扫描、改进建议",
"keywords": ["分析", "质量", "复杂度", "重复", "安全", "建议", "analyze"],
"usage": self.usage
}
def get_mcp_definition(self):
return {
"name": "project_analyzer",
"description": "项目深度分析工具",
"inputSchema": {
"type": "object",
"properties": {
"operation": {
"type": "string",
"enum": ["complexity", "imports", "duplicates", "todo", "security", "metrics", "suggest"],
"description": "分析类型"
},
"target": {
"type": "string",
"description": "目标目录或文件"
}
},
"required": ["operation"]
}
}
def convert_mcp_args(self, arguments):
op = arguments.get("operation", "")
target = arguments.get("target", ".")
return f"{op} {target}"
def handle(self, args: str) -> str:
parts = args.strip().split(maxsplit=1)
if not parts:
return self.usage
op = parts[0].lower()
target = parts[1] if len(parts) > 1 else "."
handlers = {
"complexity": self._complexity,
"imports": self._imports,
"duplicates": self._duplicates,
"todo": self._todo,
"security": self._security,
"metrics": self._metrics,
"suggest": self._suggest,
}
handler = handlers.get(op)
if not handler:
return f"错误: 不支持的操作 '{op}'"
try:
return handler(target)
except Exception as e:
return f"分析错误: {e}"
def _walk_py(self, target: str) -> List[str]:
"""遍历 Python 文件"""
files = []
skip = {'.git', '__pycache__', '.venv', 'venv', 'node_modules'}
for root, dirs, fnames in os.walk(target):
dirs[:] = [d for d in dirs if d not in skip]
for f in fnames:
if f.endswith('.py'):
files.append(os.path.join(root, f))
return files
def _complexity(self, target: str) -> str:
"""圈复杂度分析"""
files = self._walk_py(target)
results = []
for filepath in files:
try:
with open(filepath, 'r', encoding='utf-8') as f:
source = f.read()
tree = ast.parse(source)
except:
continue
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
complexity = 1
for child in ast.walk(node):
if isinstance(child, (ast.If, ast.While, ast.For, ast.ExceptHandler)):
complexity += 1
elif isinstance(child, ast.BoolOp):
complexity += len(child.values) - 1
if complexity > 5:
rel = os.path.relpath(filepath, target)
results.append({
'file': rel,
'func': node.name,
'line': node.lineno,
'complexity': complexity
})
if not results:
return "✅ 所有函数复杂度正常 (≤5)"
results.sort(key=lambda x: x['complexity'], reverse=True)
lines = [f"⚠️ 高复杂度函数 ({len(results)} 个):\n"]
for r in results[:20]:
level = "🔴" if r['complexity'] > 10 else "🟡"
lines.append(f" {level} {r['file']}:{r['line']} {r['func']}() 复杂度={r['complexity']}")
return '\n'.join(lines)
def _imports(self, target: str) -> str:
"""依赖关系分析"""
if os.path.isfile(target):
files = [target]
else:
files = self._walk_py(target)
all_imports = {}
internal_deps = {}
for filepath in files:
try:
with open(filepath, 'r', encoding='utf-8') as f:
tree = ast.parse(f.read())
except:
continue
rel = os.path.relpath(filepath, target if os.path.isdir(target) else os.path.dirname(target))
imports = []
for node in ast.iter_child_nodes(tree):
if isinstance(node, ast.Import):
for alias in node.names:
imports.append(alias.name)
elif isinstance(node, ast.ImportFrom):
if node.module:
imports.append(node.module)
all_imports[rel] = imports
# 统计
stdlib = {'os', 'sys', 'json', 're', 'time', 'datetime', 'logging',
'threading', 'subprocess', 'pathlib', 'typing', 'collections',
'functools', 'itertools', 'abc', 'dataclasses', 'enum',
'importlib', 'ast', 'io', 'traceback', 'uuid', 'shutil'}
third_party = set()
for imports in all_imports.values():
for imp in imports:
root = imp.split('.')[0]
if root not in stdlib and root not in {os.path.splitext(os.path.basename(f))[0] for f in files}:
third_party.add(root)
lines = [f"📦 依赖分析 ({len(files)} 个文件):\n"]
lines.append(f" 第三方依赖: {len(third_party)} 个")
for dep in sorted(third_party):
lines.append(f" • {dep}")
return '\n'.join(lines)
def _duplicates(self, target: str) -> str:
"""重复代码检测（简单版：检测相同函数名）"""
files = self._walk_py(target)
func_defs = {}
for filepath in files:
try:
with open(filepath, 'r', encoding='utf-8') as f:
tree = ast.parse(f.read())
except:
continue
rel = os.path.relpath(filepath, target)
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
key = node.name
if key not in func_defs:
func_defs[key] = []
func_defs[key].append(f"{rel}:{node.lineno}")
# 找重复
duplicates = {k: v for k, v in func_defs.items() if len(v) > 1}
if not duplicates:
return "✅ 没有发现重复函数名"
lines = [f"⚠️ 重复函数名 ({len(duplicates)} 个):\n"]
for name, locations in sorted(duplicates.items(), key=lambda x: -len(x[1])):
lines.append(f" {name} ({len(locations)} 处):")
for loc in locations[:5]:
lines.append(f" • {loc}")
return '\n'.join(lines)
def _todo(self, target: str) -> str:
"""TODO/FIXME 汇总"""
files = self._walk_py(target)
pattern = re.compile(r'#\s*(TODO|FIXME|HACK|XXX|BUG|NOTE)\b[:\s]*(.*)', re.IGNORECASE)
results = []
for filepath in files:
try:
with open(filepath, 'r', encoding='utf-8') as f:
for i, line in enumerate(f, 1):
match = pattern.search(line)
if match:
rel = os.path.relpath(filepath, target)
tag = match.group(1).upper()
desc = match.group(2).strip()
results.append((rel, i, tag, desc))
except:
continue
if not results:
return "✅ 没有 TODO/FIXME"
lines = [f"📝 TODO/FIXME ({len(results)} 个):\n"]
for file, line, tag, desc in results[:30]:
icon = {'TODO': '📌', 'FIXME': '🔧', 'HACK': '⚡', 'XXX': '⚠️', 'BUG': '🐛'}.get(tag, '📝')
lines.append(f" {icon} {file}:{line} [{tag}] {desc}")
return '\n'.join(lines)
def _security(self, target: str) -> str:
"""安全风险扫描"""
files = self._walk_py(target)
risks = []
patterns = [
(r'\beval\s*\(', '使用 eval() - 代码注入风险', 'HIGH'),
(r'\bexec\s*\(', '使用 exec() - 代码注入风险', 'HIGH'),
(r'\b__import__\s*\(', '动态导入 - 潜在风险', 'MEDIUM'),
(r'subprocess\.call\s*\(\s*shell\s*=\s*True', 'shell=True - 命令注入风险', 'HIGH'),
(r'pickle\.loads?\s*\(', 'pickle 反序列化 - 代码执行风险', 'HIGH'),
(r'password\s*=\s*["\']', '硬编码密码', 'HIGH'),
(r'api_key\s*=\s*["\']', '硬编码 API Key', 'HIGH'),
(r'secret\s*=\s*["\']', '硬编码 Secret', 'HIGH'),
]
for filepath in files:
try:
with open(filepath, 'r', encoding='utf-8') as f:
for i, line in enumerate(f, 1):
for pattern, desc, level in patterns:
if re.search(pattern, line, re.IGNORECASE):
rel = os.path.relpath(filepath, target)
risks.append((rel, i, level, desc, line.strip()))
except:
continue
if not risks:
return "✅ 没有发现安全风险"
lines = [f"🔒 安全风险 ({len(risks)} 个):\n"]
for file, line, level, desc, code in risks[:20]:
icon = '🔴' if level == 'HIGH' else '🟡'
lines.append(f" {icon} {file}:{line} [{level}] {desc}")
lines.append(f" {code[:60]}")
return '\n'.join(lines)
def _metrics(self, target: str) -> str:
"""代码质量指标"""
files = self._walk_py(target)
total_lines = 0
total_code = 0
total_comment = 0
total_blank = 0
total_funcs = 0
total_classes = 0
max_file_lines = 0
long_files = []
for filepath in files:
try:
with open(filepath, 'r', encoding='utf-8') as f:
lines = f.readlines()
except:
continue
file_lines = len(lines)
total_lines += file_lines
if file_lines > max_file_lines:
max_file_lines = file_lines
if file_lines > 300:
rel = os.path.relpath(filepath, target)
long_files.append((rel, file_lines))
for line in lines:
stripped = line.strip()
if not stripped:
total_blank += 1
elif stripped.startswith('#'):
total_comment += 1
else:
total_code += 1
try:
with open(filepath, 'r', encoding='utf-8') as f:
tree = ast.parse(f.read())
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
total_funcs += 1
elif isinstance(node, ast.ClassDef):
total_classes += 1
except:
pass
comment_rate = (total_comment / max(total_code, 1)) * 100
avg_lines = total_lines / max(len(files), 1)
lines = [
f"📊 代码质量指标 ({target}):\n",
f" 文件数: {len(files)}",
f" 总行数: {total_lines}",
f" 代码行: {total_code}",
f" 注释行: {total_comment} ({comment_rate:.1f}%)",
f" 空行: {total_blank}",
f" 函数数: {total_funcs}",
f" 类数: {total_classes}",
f" 平均文件行: {avg_lines:.0f}",
f" 最大文件行: {max_file_lines}",
]
if long_files:
lines.append(f"\n ⚠️ 超过 300 行的文件 ({len(long_files)} 个):")
for f, l in sorted(long_files, key=lambda x: -x[1])[:5]:
lines.append(f" • {f}: {l} 行")
return '\n'.join(lines)
def _suggest(self, target: str) -> str:
"""改进建议"""
suggestions = []
# 检查项目结构
files = os.listdir(target) if os.path.isdir(target) else []
py_files = self._walk_py(target)
if not any(f == 'README.md' for f in files):
suggestions.append("📝 添加 README.md 项目说明")
if not any(f == 'pyproject.toml' for f in files):
suggestions.append("📦 使用 pyproject.toml 替代 setup.py")
if not any(f == '.gitignore' for f in files):
suggestions.append("🚫 添加 .gitignore 文件")
if not any(f == 'requirements.txt' or f == 'pyproject.toml' for f in files):
suggestions.append("📋 添加依赖声明文件")
# 检查测试
test_files = [f for f in py_files if 'test' in os.path.basename(f).lower()]
if not test_files and len(py_files) > 5:
suggestions.append("🧪 添加单元测试")
# 检查类型提示
files_without_typehints = 0
for filepath in py_files[:20]:
try:
with open(filepath, 'r', encoding='utf-8') as f:
source = f.read()
if '-> ' not in source and ': str' not in source and ': int' not in source:
files_without_typehints += 1
except:
pass
if files_without_typehints > len(py_files) * 0.5:
suggestions.append("🏷️ 添加类型提示 (typing)")
# 检查日志
uses_print = 0
uses_logging = 0
for filepath in py_files:
try:
with open(filepath, 'r', encoding='utf-8') as f:
source = f.read()
if 'print(' in source:
uses_print += 1
if 'logging' in source:
uses_logging += 1
except:
pass
if uses_print > uses_logging and uses_print > 3:
suggestions.append("📊 考虑使用 logging 替代 print")
# 检查文档
has_docstrings = 0
for filepath in py_files:
try:
with open(filepath, 'r', encoding='utf-8') as f:
if '"""' in f.read():
has_docstrings += 1
except:
pass
if has_docstrings < len(py_files) * 0.3:
suggestions.append("📚 添加函数/类文档字符串 (docstrings)")
if not suggestions:
return "✅ 项目结构良好，暂无改进建议"
lines = [f"💡 改进建议 ({len(suggestions)} 条):\n"]
for s in suggestions:
lines.append(f" {s}")
return '\n'.join(lines)
plugins/scheduler.py
-1
View File
@@ -1 +0,0 @@
"hello"