from typing import Dict, Any, Optional import logging import ast import re from modules.methods.methodBase import MethodBase, AuthSource, MethodResult logger = logging.getLogger(__name__) class MethodCoder(MethodBase): """Coder method implementation for code operations""" def __init__(self): super().__init__() self.name = "coder" self.description = "Handle code operations like analysis, generation, and refactoring" self.authSource = AuthSource.LOCAL # Code operations typically don't need auth @property def actions(self) -> Dict[str, Dict[str, Any]]: """Available actions and their parameters""" return { "analyze": { "description": "Analyze code structure and quality", "retryMax": 2, "timeout": 30, "parameters": { "code": {"type": "string", "required": True}, "language": {"type": "string", "required": False}, "metrics": {"type": "array", "items": "string", "required": False} } }, "generate": { "description": "Generate code based on requirements", "retryMax": 2, "timeout": 60, "parameters": { "requirements": {"type": "string", "required": True}, "language": {"type": "string", "required": False}, "style": {"type": "string", "required": False} } }, "refactor": { "description": "Refactor code for better quality", "retryMax": 2, "timeout": 60, "parameters": { "code": {"type": "string", "required": True}, "language": {"type": "string", "required": False}, "improvements": {"type": "array", "items": "string", "required": False} } } } async def execute(self, action: str, parameters: Dict[str, Any], authData: Optional[Dict[str, Any]] = None) -> MethodResult: """Execute coder method""" try: # Validate parameters if not await self.validateParameters(action, parameters): return self._createResult( success=False, data={"error": f"Invalid parameters for {action}"} ) # Execute action if action == "analyze": return await self._analyzeCode(parameters) elif action == "generate": return await self._generateCode(parameters) elif action == "refactor": return await self._refactorCode(parameters) else: return self._createResult( success=False, data={"error": f"Unknown action: {action}"} ) except Exception as e: logger.error(f"Error executing coder {action}: {e}") return self._createResult( success=False, data={"error": str(e)} ) async def _analyzeCode(self, parameters: Dict[str, Any]) -> MethodResult: """Analyze code structure and quality""" try: code = parameters["code"] language = parameters.get("language", "python") metrics = parameters.get("metrics", ["complexity", "style", "documentation"]) analysis = {} if language.lower() == "python": # Parse Python code try: tree = ast.parse(code) # Calculate basic metrics analysis["metrics"] = { "lines": len(code.splitlines()), "classes": len([node for node in ast.walk(tree) if isinstance(node, ast.ClassDef)]), "functions": len([node for node in ast.walk(tree) if isinstance(node, ast.FunctionDef)]), "imports": len([node for node in ast.walk(tree) if isinstance(node, ast.Import) or isinstance(node, ast.ImportFrom)]) } # Check for common issues analysis["issues"] = [] # Check for missing docstrings if "documentation" in metrics: for node in ast.walk(tree): if isinstance(node, (ast.ClassDef, ast.FunctionDef)) and not ast.get_docstring(node): analysis["issues"].append({ "type": "missing_docstring", "line": node.lineno, "name": node.name }) # Check for long functions if "complexity" in metrics: for node in ast.walk(tree): if isinstance(node, ast.FunctionDef): bodyLines = len(node.body) if bodyLines > 20: # Arbitrary threshold analysis["issues"].append({ "type": "long_function", "line": node.lineno, "name": node.name, "lines": bodyLines }) # Check for style issues if "style" in metrics: # Check line length for i, line in enumerate(code.splitlines(), 1): if len(line) > 100: # PEP 8 recommendation analysis["issues"].append({ "type": "line_too_long", "line": i, "length": len(line) }) # Check for mixed tabs and spaces if "\t" in code and " " in code: analysis["issues"].append({ "type": "mixed_tabs_spaces", "message": "Code mixes tabs and spaces" }) except SyntaxError as e: return self._createResult( success=False, data={"error": f"Syntax error: {str(e)}"} ) else: # TODO: Implement analysis for other languages return self._createResult( success=False, data={"error": f"Unsupported language: {language}"} ) return self._createResult( success=True, data={ "language": language, "analysis": analysis } ) except Exception as e: logger.error(f"Error analyzing code: {e}") return self._createResult( success=False, data={"error": f"Analysis failed: {str(e)}"} ) async def _generateCode(self, parameters: Dict[str, Any]) -> MethodResult: """Generate code based on requirements""" try: requirements = parameters["requirements"] language = parameters.get("language", "python") style = parameters.get("style", "standard") # TODO: Implement code generation using AI or templates # This is a placeholder implementation if language.lower() == "python": # Generate a simple Python class based on requirements className = re.sub(r'[^a-zA-Z0-9]', '', requirements.split()[0].title()) code = f"""class {className}: \"\"\" {requirements} \"\"\" def __init__(self): pass def process(self): pass """ else: return self._createResult( success=False, data={"error": f"Unsupported language: {language}"} ) return self._createResult( success=True, data={ "language": language, "code": code } ) except Exception as e: logger.error(f"Error generating code: {e}") return self._createResult( success=False, data={"error": f"Generation failed: {str(e)}"} ) async def _refactorCode(self, parameters: Dict[str, Any]) -> MethodResult: """Refactor code for better quality""" try: code = parameters["code"] language = parameters.get("language", "python") improvements = parameters.get("improvements", ["style", "complexity"]) if language.lower() == "python": # Parse Python code try: tree = ast.parse(code) # Apply improvements if "style" in improvements: # Format code (placeholder) code = code.strip() if "complexity" in improvements: # TODO: Implement complexity reduction pass if "documentation" in improvements: # Add missing docstrings for node in ast.walk(tree): if isinstance(node, (ast.ClassDef, ast.FunctionDef)) and not ast.get_docstring(node): # TODO: Generate docstring pass except SyntaxError as e: return self._createResult( success=False, data={"error": f"Syntax error: {str(e)}"} ) else: return self._createResult( success=False, data={"error": f"Unsupported language: {language}"} ) return self._createResult( success=True, data={ "language": language, "code": code, "improvements": improvements } ) except Exception as e: logger.error(f"Error refactoring code: {e}") return self._createResult( success=False, data={"error": f"Refactoring failed: {str(e)}"} )