"""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")