From b1b2097311a3ea2a40ebe99c291cbeed685e0a74 Mon Sep 17 00:00:00 2001 From: Xiaoli Dev Date: Sat, 25 Apr 2026 11:28:53 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=203=20=E4=B8=AA?= =?UTF-8?q?=E5=B7=A5=E7=A8=8B=E5=8C=96=E8=87=AA=E5=8A=A8=E5=8C=96=E6=8F=92?= =?UTF-8?q?=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - auto_engineer: lint/format/test/build/deps/info/init - git_workflow: 规范提交/changelog/变更摘要/贡献者统计 - project_analyzer: 复杂度/依赖/重复/安全/质量指标/改进建议 工具总数: 17 个 本次更新由 mimo 自动化执行，与开发者无关。 --- plugins/auto_engineer.py | 344 ++++++++++++++++++++++++++++ plugins/git_workflow.py | 284 +++++++++++++++++++++++ plugins/project_analyzer.py | 443 ++++++++++++++++++++++++++++++++++++ plugins_config.py | 4 +- 4 files changed, 1074 insertions(+), 1 deletion(-) create mode 100644 plugins/auto_engineer.py create mode 100644 plugins/git_workflow.py create mode 100644 plugins/project_analyzer.py diff --git a/plugins/auto_engineer.py b/plugins/auto_engineer.py new file mode 100644 index 0000000..7b0877a --- /dev/null +++ b/plugins/auto_engineer.py @@ -0,0 +1,344 @@ +""" +工程化自动化插件 - 代码质量检查、格式化、测试运行 +""" +import os +import subprocess +import json +import re +from typing import Tuple + + +class Plugin: + """工程化工具 - lint、format、test、build 自动化""" + + def __init__(self): + self.usage = """工程化自动化工具 + +操作类型: + lint [目录] [文件模式] - 代码质量检查 (ruff/flake8/pylint) + format [目录] [文件模式] - 代码格式化 (black/ruff) + test [目录] [参数] - 运行测试 (pytest) + build [命令] - 构建项目 + deps - 检查依赖 (pip list --outdated) + deps-check - 安全漏洞检查 + pre-commit - 运行 pre-commit 检查 + info - 项目信息摘要 + init [类型] - 初始化项目结构 (python/web/cli) +""" + self.cli = None + + def set_cli(self, cli): + self.cli = cli + + def get_tool_info(self): + return { + "name": "auto_engineer", + "description": "工程化自动化 - lint/format/test/build/deps 一键执行", + "keywords": ["lint", "format", "test", "build", "代码质量", "格式化", "测试", "构建"], + "usage": self.usage + } + + def get_mcp_definition(self): + return { + "name": "auto_engineer", + "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": "额外选项" + } + }, + "required": ["operation"] + } + } + + def convert_mcp_args(self, arguments): + op = arguments.get("operation", "") + target = arguments.get("target", "") + options = arguments.get("options", "") + return f"{op} {target} {options}".strip() + + def handle(self, args: str) -> str: + parts = args.strip().split(maxsplit=2) + if not parts: + return self.usage + + op = parts[0].lower() + target = parts[1] if len(parts) > 1 else "." + options = parts[2] if len(parts) > 2 else "" + + handlers = { + "lint": self._lint, + "format": self._format, + "test": self._test, + "build": self._build, + "deps": self._deps, + "deps-check": self._deps_check, + "pre-commit": self._pre_commit, + "info": self._info, + "init": self._init_project, + } + + handler = handlers.get(op) + if not handler: + 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}" + + 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]}" + + 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}" + + 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) + + 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]}" + + 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)" + 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}" + + 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} 个") + + 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" + + 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 _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)" + elif "setup.py" in files: + project_type = "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(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: + pass + + # Git 状态 + ok, git = self._run("git log --oneline -1", timeout=5) + if ok: + info.append(f"🔀 最新提交: {git}") + + ok, branch = self._run("git branch --show-current", timeout=5) + 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()} 个") + + return '\n'.join(info) + + def _init_project(self, target: str, options: str) -> str: + """初始化项目结构""" + project_type = target if target != "." else "python" + + if project_type == "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": + 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" + + created = [] + for d in dirs: + if not os.path.exists(d): + os.makedirs(d) + 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) + with open(f, 'w') as fp: + fp.write(content) + created.append(f"📄 {f}") + + return f"✅ 已初始化 {project_type} 项目\n" + '\n'.join(created) diff --git a/plugins/git_workflow.py b/plugins/git_workflow.py new file mode 100644 index 0000000..d7cda5b --- /dev/null +++ b/plugins/git_workflow.py @@ -0,0 +1,284 @@ +""" +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) diff --git a/plugins/project_analyzer.py b/plugins/project_analyzer.py new file mode 100644 index 0000000..8a94d80 --- /dev/null +++ b/plugins/project_analyzer.py @@ -0,0 +1,443 @@ +""" +项目分析插件 - 代码质量分析、依赖分析、架构建议 +""" +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) diff --git a/plugins_config.py b/plugins_config.py index 692577d..ab56c0b 100644 --- a/plugins_config.py +++ b/plugins_config.py @@ -1,5 +1,4 @@ # 小狸插件配置 -# 自动生成于: 2026-04-25 ENABLE_AI_SEARCH = True ENABLE_AUDIO_PLAYER = True @@ -14,3 +13,6 @@ ENABLE_SCHEDULER = True ENABLE_SPEECH_RECOGNITION = True ENABLE_TASK_MANAGER = True ENABLE_TOOL_SEARCH = True +ENABLE_AUTO_ENGINEER = True +ENABLE_GIT_WORKFLOW = True +ENABLE_PROJECT_ANALYZER = True