Spaces:
Paused
Paused
| """Code analyzer tool implementation.""" | |
| from __future__ import annotations | |
| import ast | |
| import logging | |
| import re | |
| from pathlib import Path | |
| from typing import Any | |
| from hermes.tools.base.tool import BaseTool, ToolSchema | |
| logger = logging.getLogger(__name__) | |
| class CodeAnalyzerTool(BaseTool): | |
| """Tool for analyzing code quality, structure, and patterns.""" | |
| BASE_DIR: str = "." | |
| def __init__(self, base_dir: str | None = None) -> None: | |
| super().__init__() | |
| if base_dir: | |
| self.BASE_DIR = base_dir | |
| def _resolve_analysis_path(self, path: str) -> str: | |
| """Resolve and validate analysis path within base directory.""" | |
| base = Path(self.BASE_DIR).resolve() | |
| target = (base / path).resolve() | |
| try: | |
| target.relative_to(base) | |
| except ValueError: | |
| raise ValueError("Path escapes allowed analysis directory") from None | |
| if not target.exists(): | |
| raise FileNotFoundError(f"Path not found: {path}") | |
| return str(target) | |
| def _define_schema(self) -> ToolSchema: | |
| return ToolSchema( | |
| name="code_analyzer", | |
| description="Analyze code for quality, structure, dependencies, and patterns", | |
| parameters={ | |
| "action": { | |
| "type": "string", | |
| "description": "Action: analyze_file, analyze_project, detect_patterns, analyze_complexity", | |
| }, | |
| "path": { | |
| "type": "string", | |
| "description": "File or project path", | |
| }, | |
| "code": { | |
| "type": "string", | |
| "description": "Code string to analyze", | |
| }, | |
| "language": { | |
| "type": "string", | |
| "description": "Programming language", | |
| "default": "python", | |
| }, | |
| }, | |
| required=["action"], | |
| category="analysis", | |
| tags=["code", "analysis", "quality"], | |
| ) | |
| async def execute(self, **kwargs: Any) -> dict[str, Any]: | |
| """Execute code analysis.""" | |
| action = kwargs["action"] | |
| try: | |
| if action == "analyze_file": | |
| return await self._analyze_file(kwargs["path"]) | |
| elif action == "analyze_project": | |
| return await self._analyze_project(kwargs["path"]) | |
| elif action == "detect_patterns": | |
| return await self._detect_patterns(kwargs.get("code", ""), kwargs.get("language", "python")) | |
| elif action == "analyze_complexity": | |
| return await self._analyze_complexity(kwargs.get("code", "")) | |
| else: | |
| return {"error": f"Unknown action: {action}"} | |
| except Exception as e: | |
| logger.error(f"Code analysis error: {e}") | |
| return {"error": str(e)} | |
| async def _analyze_file(self, path: str) -> dict[str, Any]: | |
| """Analyze a single file with path traversal protection.""" | |
| resolved = self._resolve_analysis_path(path) | |
| file_path = Path(resolved) | |
| code = file_path.read_text(encoding="utf-8", errors="replace") | |
| language = self._detect_language(file_path.suffix) | |
| result = { | |
| "file": str(file_path), | |
| "language": language, | |
| "size_bytes": file_path.stat().st_size, | |
| "lines": len(code.splitlines()), | |
| } | |
| if language == "python": | |
| result.update(self._analyze_python(code)) | |
| else: | |
| result.update(self._analyze_generic(code)) | |
| return result | |
| async def _analyze_project(self, path: str) -> dict[str, Any]: | |
| """Analyze a project directory with path traversal protection.""" | |
| resolved = self._resolve_analysis_path(path) | |
| dir_path = Path(resolved) | |
| files = [] | |
| languages: dict[str, int] = {} | |
| total_lines = 0 | |
| total_size = 0 | |
| for file_path in dir_path.rglob("*"): | |
| if file_path.is_file() and not any( | |
| part.startswith(".") or part == "node_modules" or part == "__pycache__" | |
| for part in file_path.parts | |
| ): | |
| try: | |
| suffix = file_path.suffix.lower() | |
| lang = self._detect_language(suffix) | |
| size = file_path.stat().st_size | |
| lines = len(file_path.read_text(encoding="utf-8", errors="replace").splitlines()) | |
| files.append( | |
| { | |
| "path": str(file_path.relative_to(dir_path)), | |
| "language": lang, | |
| "lines": lines, | |
| "size": size, | |
| } | |
| ) | |
| languages[lang] = languages.get(lang, 0) + 1 | |
| total_lines += lines | |
| total_size += size | |
| except Exception: | |
| continue | |
| return { | |
| "project": str(dir_path), | |
| "total_files": len(files), | |
| "total_lines": total_lines, | |
| "total_size_bytes": total_size, | |
| "languages": languages, | |
| "files": files[:100], | |
| } | |
| async def _detect_patterns(self, code: str, language: str) -> dict[str, Any]: | |
| """Detect code patterns.""" | |
| patterns = { | |
| "design_patterns": [], | |
| "anti_patterns": [], | |
| "code_smells": [], | |
| } | |
| if language == "python": | |
| patterns.update(self._detect_python_patterns(code)) | |
| return patterns | |
| async def _analyze_complexity(self, code: str) -> dict[str, Any]: | |
| """Analyze code complexity.""" | |
| try: | |
| tree = ast.parse(code) | |
| complexity = { | |
| "cyclomatic": self._calculate_cyclomatic_complexity(tree), | |
| "functions": len([node for node in ast.walk(tree) if isinstance(node, ast.FunctionDef)]), | |
| "classes": len([node for node in ast.walk(tree) if isinstance(node, ast.ClassDef)]), | |
| "max_depth": self._calculate_max_depth(tree), | |
| "lines": len(code.splitlines()), | |
| } | |
| return complexity | |
| except SyntaxError: | |
| return {"error": "Invalid Python syntax"} | |
| def _analyze_python(self, code: str) -> dict[str, Any]: | |
| """Analyze Python code.""" | |
| try: | |
| tree = ast.parse(code) | |
| functions = [] | |
| classes = [] | |
| imports = [] | |
| for node in ast.walk(tree): | |
| if isinstance(node, ast.FunctionDef): | |
| functions.append( | |
| { | |
| "name": node.name, | |
| "args": len(node.args.args), | |
| "decorators": len(node.decorator_list), | |
| "line": node.lineno, | |
| } | |
| ) | |
| elif isinstance(node, ast.ClassDef): | |
| methods = [n for n in node.body if isinstance(n, ast.FunctionDef)] | |
| classes.append( | |
| { | |
| "name": node.name, | |
| "methods": len(methods), | |
| "bases": len(node.bases), | |
| "line": node.lineno, | |
| } | |
| ) | |
| elif isinstance(node, (ast.Import, ast.ImportFrom)): | |
| if isinstance(node, ast.Import): | |
| for alias in node.names: | |
| imports.append(alias.name) | |
| else: | |
| module = node.module or "" | |
| for alias in node.names: | |
| imports.append(f"{module}.{alias.name}") | |
| return { | |
| "functions": functions, | |
| "classes": classes, | |
| "imports": imports, | |
| "complexity": self._calculate_cyclomatic_complexity(tree), | |
| } | |
| except SyntaxError: | |
| return {"error": "Invalid Python syntax", "functions": [], "classes": [], "imports": []} | |
| def _analyze_generic(self, code: str) -> dict[str, Any]: | |
| """Generic code analysis.""" | |
| lines = code.splitlines() | |
| functions = len(re.findall(r"(?:function|def|func)\s+\w+", code)) | |
| classes = len(re.findall(r"(?:class|struct)\s+\w+", code)) | |
| comments = len(re.findall(r"(?:#|//|/\*)\s*", code)) | |
| return { | |
| "functions": functions, | |
| "classes": classes, | |
| "comments": comments, | |
| "blank_lines": sum(1 for line in lines if not line.strip()), | |
| } | |
| def _detect_python_patterns(self, code: str) -> dict[str, Any]: | |
| """Detect Python-specific patterns.""" | |
| patterns = { | |
| "design_patterns": [], | |
| "anti_patterns": [], | |
| "code_smells": [], | |
| } | |
| if re.search(r"class\s+\w+.*Singleton", code): | |
| patterns["design_patterns"].append("Singleton") | |
| if re.search(r"def\s+get_\w+\s*\(self\)", code) and re.search(r"def\s+set_\w+\s*\(self\)", code): | |
| patterns["design_patterns"].append("Getter/Setter") | |
| if re.search(r"except\s*:", code): | |
| patterns["anti_patterns"].append("Bare except clause") | |
| if re.search(r"except\s+Exception\s*:", code): | |
| patterns["code_smells"].append("Broad exception handling") | |
| if re.search(r"print\s*\(", code): | |
| patterns["code_smells"].append("Print statements in production code") | |
| if len(code.splitlines()) > 500: | |
| patterns["code_smells"].append("Large file (>500 lines)") | |
| return patterns | |
| def _calculate_cyclomatic_complexity(self, tree: ast.AST) -> int: | |
| """Calculate cyclomatic complexity.""" | |
| complexity = 1 | |
| for node in ast.walk(tree): | |
| if isinstance(node, (ast.If, ast.While, ast.For, ast.ExceptHandler)): | |
| complexity += 1 | |
| elif isinstance(node, ast.BoolOp): | |
| complexity += len(node.values) - 1 | |
| return complexity | |
| def _calculate_max_depth(self, tree: ast.AST) -> int: | |
| """Calculate maximum nesting depth.""" | |
| max_depth = 0 | |
| def _walk_depth(node: ast.AST, depth: int) -> None: | |
| nonlocal max_depth | |
| max_depth = max(max_depth, depth) | |
| for child in ast.iter_child_nodes(node): | |
| if isinstance(node, (ast.If, ast.For, ast.While, ast.With, ast.Try)): | |
| _walk_depth(child, depth + 1) | |
| else: | |
| _walk_depth(child, depth) | |
| _walk_depth(tree, 0) | |
| return max_depth | |
| def _detect_language(self, suffix: str) -> str: | |
| """Detect language from file extension.""" | |
| lang_map = { | |
| ".py": "python", | |
| ".js": "javascript", | |
| ".ts": "typescript", | |
| ".jsx": "javascript", | |
| ".tsx": "typescript", | |
| ".go": "go", | |
| ".rs": "rust", | |
| ".java": "java", | |
| ".rb": "ruby", | |
| ".yaml": "yaml", | |
| ".yml": "yaml", | |
| ".json": "json", | |
| ".toml": "toml", | |
| ".md": "markdown", | |
| } | |
| return lang_map.get(suffix, "unknown") | |