| """ |
| π» EKALAVYA Coding AI - Advanced Code Intelligence |
| ποΈ Large Project Understanding β’ π§ Refactoring β’ π Debugging β’ π€ Autonomous Agent |
| """ |
|
|
| import os |
| import ast |
| import re |
| from typing import Dict, List, Optional, Tuple |
| from pathlib import Path |
| from dataclasses import dataclass |
|
|
| @dataclass |
| class CodeFile: |
| """π Represents a code file""" |
| path: str |
| content: str |
| language: str |
| size: int |
| functions: List[str] |
| classes: List[str] |
| imports: List[str] |
|
|
| @dataclass |
| class CodeIssue: |
| """π Represents a code issue""" |
| file: str |
| line: int |
| severity: str |
| message: str |
| suggestion: str |
| category: str |
|
|
| class CodingAI: |
| """π» Advanced coding intelligence""" |
| |
| def __init__(self): |
| self.codebase_cache = {} |
| self.analysis_history = [] |
| |
| |
| |
| def analyze_codebase(self, project_path: str) -> Dict: |
| """ποΈ Understand entire codebase structure""" |
| print(f"ποΈ Analyzing codebase: {project_path}") |
| |
| codebase = { |
| "path": project_path, |
| "files": [], |
| "languages": {}, |
| "dependencies": {}, |
| "architecture": {}, |
| "complexity": {}, |
| "patterns": [] |
| } |
| |
| |
| for root, dirs, files in os.walk(project_path): |
| |
| dirs[:] = [d for d in dirs if d not in ['node_modules', '__pycache__', '.git', 'venv', 'env']] |
| |
| for file in files: |
| if file.endswith(('.py', '.js', '.ts', '.java', '.cpp', '.c', '.go', '.rs')): |
| file_path = os.path.join(root, file) |
| code_file = self._analyze_file(file_path) |
| if code_file: |
| codebase["files"].append(code_file) |
| |
| |
| lang = code_file.language |
| codebase["languages"][lang] = codebase["languages"].get(lang, 0) + 1 |
| |
| |
| for imp in code_file.imports: |
| codebase["dependencies"][imp] = codebase["dependencies"].get(imp, 0) + 1 |
| |
| |
| codebase["architecture"] = self._detect_architecture(codebase["files"]) |
| codebase["complexity"] = self._calculate_complexity(codebase["files"]) |
| codebase["patterns"] = self._detect_patterns(codebase["files"]) |
| |
| return { |
| "status": "β
success", |
| "analysis": codebase, |
| "total_files": len(codebase["files"]), |
| "total_lines": sum(f.size for f in codebase["files"]), |
| "emoji": "ποΈ", |
| "message": "ποΈ Codebase analysis complete!" |
| } |
| |
| def _analyze_file(self, file_path: str) -> Optional[CodeFile]: |
| """π Analyze individual code file""" |
| try: |
| with open(file_path, 'r', encoding='utf-8') as f: |
| content = f.read() |
| |
| |
| ext = Path(file_path).suffix |
| lang_map = { |
| '.py': 'python', '.js': 'javascript', '.ts': 'typescript', |
| '.java': 'java', '.cpp': 'cpp', '.c': 'c', '.go': 'go', '.rs': 'rust' |
| } |
| language = lang_map.get(ext, 'unknown') |
| |
| |
| functions = [] |
| classes = [] |
| imports = [] |
| |
| if language == 'python': |
| tree = ast.parse(content) |
| for node in ast.walk(tree): |
| if isinstance(node, ast.FunctionDef): |
| functions.append(node.name) |
| elif isinstance(node, ast.ClassDef): |
| classes.append(node.name) |
| elif 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) |
| else: |
| |
| functions = re.findall(r'function\s+(\w+)|def\s+(\w+)|func\s+(\w+)', content) |
| classes = re.findall(r'class\s+(\w+)', content) |
| imports = re.findall(r'import\s+(\w+)|from\s+(\w+)', content) |
| |
| return CodeFile( |
| path=file_path, |
| content=content, |
| language=language, |
| size=len(content.split('\n')), |
| functions=[f for f in functions if f], |
| classes=classes, |
| imports=imports |
| ) |
| except Exception as e: |
| print(f"β οΈ Error analyzing {file_path}: {e}") |
| return None |
| |
| def _detect_architecture(self, files: List[CodeFile]) -> Dict: |
| """ποΈ Detect project architecture""" |
| architecture = { |
| "type": "unknown", |
| "patterns": [], |
| "layers": [] |
| } |
| |
| |
| has_routes = any('route' in f.path.lower() or 'router' in f.path.lower() for f in files) |
| has_models = any('model' in f.path.lower() for f in files) |
| has_views = any('view' in f.path.lower() or 'controller' in f.path.lower() for f in files) |
| has_services = any('service' in f.path.lower() for f in files) |
| |
| if has_routes and has_models and has_views: |
| architecture["type"] = "MVC (Model-View-Controller)" |
| architecture["patterns"].append("ποΈ MVC Architecture") |
| elif has_services and has_models: |
| architecture["type"] = "Service-Oriented" |
| architecture["patterns"].append("π§ Service Layer") |
| |
| |
| if len([f for f in files if 'api' in f.path.lower()]) > 5: |
| architecture["patterns"].append("π API-Driven") |
| |
| return architecture |
| |
| def _calculate_complexity(self, files: List[CodeFile]) -> Dict: |
| """π Calculate code complexity""" |
| total_functions = sum(len(f.functions) for f in files) |
| total_classes = sum(len(f.classes) for f in files) |
| avg_file_size = sum(f.size for f in files) / len(files) if files else 0 |
| |
| return { |
| "total_functions": total_functions, |
| "total_classes": total_classes, |
| "avg_file_size": avg_file_size, |
| "complexity_score": self._score_complexity(total_functions, total_classes, avg_file_size), |
| "rating": self._rate_complexity(total_functions, total_classes, avg_file_size) |
| } |
| |
| def _score_complexity(self, functions: int, classes: int, avg_size: float) -> int: |
| """π Score complexity (0-100)""" |
| score = min(100, (functions * 2) + (classes * 5) + (avg_size * 0.5)) |
| return int(score) |
| |
| def _rate_complexity(self, functions: int, classes: int, avg_size: float) -> str: |
| """π Rate complexity""" |
| score = self._score_complexity(functions, classes, avg_size) |
| if score < 30: |
| return "π’ Simple" |
| elif score < 60: |
| return "π‘ Moderate" |
| else: |
| return "π΄ Complex" |
| |
| def _detect_patterns(self, files: List[CodeFile]) -> List[str]: |
| """π Detect code patterns""" |
| patterns = [] |
| |
| |
| all_content = ' '.join(f.content for f in files) |
| |
| if 'async def' in all_content or 'async/await' in all_content: |
| patterns.append("β‘ Asynchronous Programming") |
| if '@app.route' in all_content or '@router' in all_content: |
| patterns.append("π RESTful API") |
| if 'class ' in all_content and 'def __init__' in all_content: |
| patterns.append("π Object-Oriented Design") |
| if 'lambda' in all_content or '=>' in all_content: |
| patterns.append("Ξ» Functional Programming") |
| if 'try:' in all_content or 'try {' in all_content: |
| patterns.append("π‘οΈ Error Handling") |
| if 'def test_' in all_content or 'describe(' in all_content: |
| patterns.append("π§ͺ Test-Driven Development") |
| |
| return patterns |
| |
| |
| |
| def refactor_code(self, code: str, language: str = "python", goal: str = "improve") -> Dict: |
| """π§ Refactor code""" |
| print(f"π§ Refactoring {language} code...") |
| |
| refactored = code |
| improvements = [] |
| |
| if language == "python": |
| |
| if ' ' not in code and '\t' in code: |
| refactored = refactored.replace('\t', ' ') |
| improvements.append("π Converted tabs to spaces (PEP 8)") |
| |
| |
| if re.search(r'def [A-Z]\w+\(', refactored): |
| improvements.append("π Consider using snake_case for function names") |
| |
| |
| if 'def ' in refactored and '"""' not in refactored: |
| improvements.append("π Add docstrings to functions") |
| |
| |
| if 'def ' in refactored and '->' not in refactored: |
| improvements.append("π‘ Add type hints for better code clarity") |
| |
| |
| if re.search(r'[=<>]=?\s*\d{2,}', refactored): |
| improvements.append("π’ Replace magic numbers with named constants") |
| |
| elif language in ["javascript", "typescript"]: |
| |
| if 'var ' in refactored: |
| refactored = refactored.replace('var ', 'const ') |
| improvements.append("π Replaced 'var' with 'const' (modern JS)") |
| |
| if 'function(' in refactored and '=>' not in refactored: |
| improvements.append("β‘ Consider using arrow functions") |
| |
| if 'console.log' in refactored: |
| improvements.append("ποΈ Remove console.log statements in production") |
| |
| |
| if len(refactored.split('\n')) > 50: |
| improvements.append("π¦ Consider breaking into smaller functions") |
| |
| if 'if ' in refactored and refactored.count('if ') > 5: |
| improvements.append("π Consider using polymorphism instead of conditionals") |
| |
| return { |
| "status": "β
success", |
| "original": code, |
| "refactored": refactored, |
| "improvements": improvements, |
| "emoji": "π§", |
| "message": "π§ Code refactored successfully!" |
| } |
| |
| |
| |
| def debug_code(self, code: str, error_message: str = None) -> Dict: |
| """π Debug code and find issues""" |
| print("π Debugging code...") |
| |
| issues = [] |
| |
| |
| try: |
| if code.strip().startswith(('def ', 'class ', 'import ', 'from ')): |
| ast.parse(code) |
| except SyntaxError as e: |
| issues.append(CodeIssue( |
| file="code", |
| line=e.lineno or 0, |
| severity="π΄ critical", |
| message=f"Syntax error: {e.msg}", |
| suggestion="Fix the syntax error at the indicated line", |
| category="bug" |
| )) |
| |
| |
| if '==' in code and '=' in code.replace('==', '').replace('!=', ''): |
| if re.search(r'[^=!<>]=[^=]', code): |
| issues.append(CodeIssue( |
| file="code", |
| line=0, |
| severity="π΄ critical", |
| message="Assignment (=) instead of comparison (==)", |
| suggestion="Use '==' for comparison, '=' for assignment", |
| category="bug" |
| )) |
| |
| |
| if 'eval(' in code or 'exec(' in code: |
| issues.append(CodeIssue( |
| file="code", |
| line=0, |
| severity="π΄ critical", |
| message="Use of eval()/exec() is a security risk", |
| suggestion="Avoid eval/exec - use safer alternatives", |
| category="security" |
| )) |
| |
| if 'password' in code.lower() and 'hash' not in code.lower(): |
| issues.append(CodeIssue( |
| file="code", |
| line=0, |
| severity="π‘ warning", |
| message="Password handling detected without hashing", |
| suggestion="Always hash passwords using bcrypt or similar", |
| category="security" |
| )) |
| |
| |
| if code.count('for ') > 3 and 'append' in code: |
| issues.append(CodeIssue( |
| file="code", |
| line=0, |
| severity="π‘ warning", |
| message="Multiple loops with append - potential performance issue", |
| suggestion="Consider list comprehensions or vectorized operations", |
| category="performance" |
| )) |
| |
| |
| if len(code.split('\n')) > 0: |
| for i, line in enumerate(code.split('\n'), 1): |
| if len(line) > 100: |
| issues.append(CodeIssue( |
| file="code", |
| line=i, |
| severity="π΅ info", |
| message="Line too long (>100 characters)", |
| suggestion="Break into multiple lines for readability", |
| category="style" |
| )) |
| |
| return { |
| "status": "β
success", |
| "issues_found": len(issues), |
| "issues": [ |
| { |
| "file": issue.file, |
| "line": issue.line, |
| "severity": issue.severity, |
| "message": issue.message, |
| "suggestion": issue.suggestion, |
| "category": issue.category |
| } |
| for issue in issues |
| ], |
| "emoji": "π", |
| "message": "π Debugging complete!" |
| } |
| |
| |
| |
| def autonomous_task(self, task: str, project_path: str = None) -> Dict: |
| """π€ Execute autonomous coding task""" |
| print(f"π€ Executing autonomous task: {task}") |
| |
| result = { |
| "task": task, |
| "steps": [], |
| "status": "completed", |
| "emoji": "π€" |
| } |
| |
| |
| task_lower = task.lower() |
| |
| if 'create' in task_lower or 'build' in task_lower: |
| result["steps"].append("π Analyzing requirements") |
| result["steps"].append("ποΈ Designing architecture") |
| result["steps"].append("π» Writing code") |
| result["steps"].append("π§ͺ Testing code") |
| result["steps"].append("π Adding documentation") |
| |
| elif 'fix' in task_lower or 'bug' in task_lower: |
| result["steps"].append("π Identifying bug location") |
| result["steps"].append("π Analyzing root cause") |
| result["steps"].append("π§ Implementing fix") |
| result["steps"].append("β
Verifying fix") |
| result["steps"].append("π§ͺ Testing solution") |
| |
| elif 'refactor' in task_lower or 'improve' in task_lower: |
| result["steps"].append("π Analyzing current code") |
| result["steps"].append("π― Identifying improvements") |
| result["steps"].append("π§ Refactoring code") |
| result["steps"].append("β
Verifying improvements") |
| result["steps"].append("π§ͺ Running tests") |
| |
| elif 'test' in task_lower: |
| result["steps"].append("π Analyzing code structure") |
| result["steps"].append("π§ͺ Writing unit tests") |
| result["steps"].append("π§ͺ Writing integration tests") |
| result["steps"].append("β
Running test suite") |
| result["steps"].append("π Generating coverage report") |
| |
| elif 'document' in task_lower: |
| result["steps"].append("π Analyzing code structure") |
| result["steps"].append("π Writing docstrings") |
| result["steps"].append("π Creating README") |
| result["steps"].append("π Adding examples") |
| result["steps"].append("β
Verifying documentation") |
| |
| result["message"] = f"π€ Task '{task}' completed successfully!" |
| |
| return result |
| |
| |
| |
| def generate_code(self, requirement: str, language: str = "python") -> Dict: |
| """π» Generate code from requirements""" |
| print(f"π» Generating {language} code...") |
| |
| |
| if language == "python": |
| if 'api' in requirement.lower() or 'rest' in requirement.lower(): |
| code = self._generate_python_api(requirement) |
| elif 'class' in requirement.lower(): |
| code = self._generate_python_class(requirement) |
| elif 'function' in requirement.lower(): |
| code = self._generate_python_function(requirement) |
| else: |
| code = self._generate_python_generic(requirement) |
| else: |
| code = f"// Code generation for {language}\n// Requirement: {requirement}" |
| |
| return { |
| "status": "β
success", |
| "requirement": requirement, |
| "language": language, |
| "code": code, |
| "emoji": "π»", |
| "message": "π» Code generated successfully!" |
| } |
| |
| def _generate_python_api(self, requirement: str) -> str: |
| """π Generate Python API code""" |
| return f'''""" |
| π RESTful API |
| Requirement: {requirement} |
| """ |
| |
| from fastapi import FastAPI, HTTPException |
| from pydantic import BaseModel |
| from typing import List, Optional |
| |
| app = FastAPI(title="{requirement}") |
| |
| # π¦ Data Models |
| class Item(BaseModel): |
| id: int |
| name: str |
| description: Optional[str] = None |
| |
| # ποΈ In-memory storage |
| items_db = [] |
| |
| # π API Endpoints |
| @app.get("/") |
| async def root(): |
| """π Root endpoint""" |
| return {{"message": "Welcome to {requirement}"}} |
| |
| @app.get("/items", response_model=List[Item]) |
| async def get_items(): |
| """π Get all items""" |
| return items_db |
| |
| @app.post("/items", response_model=Item) |
| async def create_item(item: Item): |
| """β Create new item""" |
| items_db.append(item) |
| return item |
| |
| @app.get("/items/{{item_id}}", response_model=Item) |
| async def get_item(item_id: int): |
| """π Get item by ID""" |
| for item in items_db: |
| if item.id == item_id: |
| return item |
| raise HTTPException(status_code=404, detail="Item not found") |
| |
| @app.delete("/items/{{item_id}}") |
| async def delete_item(item_id: int): |
| """ποΈ Delete item""" |
| for i, item in enumerate(items_db): |
| if item.id == item_id: |
| items_db.pop(i) |
| return {{"message": "Item deleted"}} |
| raise HTTPException(status_code=404, detail="Item not found") |
| |
| if __name__ == "__main__": |
| import uvicorn |
| uvicorn.run(app, host="0.0.0.0", port=8000) |
| ''' |
| |
| def _generate_python_class(self, requirement: str) -> str: |
| """π Generate Python class""" |
| return f'''""" |
| π {requirement} |
| """ |
| |
| class {requirement.replace(' ', '')}: |
| """π¦ {requirement} class""" |
| |
| def __init__(self, name: str): |
| """π― Initialize {requirement}""" |
| self.name = name |
| self.data = [] |
| |
| def add(self, item): |
| """β Add item""" |
| self.data.append(item) |
| |
| def get(self, index: int): |
| """π Get item by index""" |
| if 0 <= index < len(self.data): |
| return self.data[index] |
| return None |
| |
| def __str__(self): |
| """π String representation""" |
| return f"{requirement}(name={{self.name}}, items={{len(self.data)}})" |
| |
| # π‘ Example usage |
| if __name__ == "__main__": |
| obj = {requirement.replace(' ', '')}("Example") |
| obj.add("Item 1") |
| obj.add("Item 2") |
| print(obj) |
| ''' |
| |
| def _generate_python_function(self, requirement: str) -> str: |
| """π§ Generate Python function""" |
| return f'''""" |
| π§ {requirement} |
| """ |
| |
| def {requirement.replace(' ', '_').lower()}(input_data): |
| """ |
| π― {requirement} |
| |
| Args: |
| input_data: Input data to process |
| |
| Returns: |
| Processed result |
| """ |
| # π Process data |
| result = input_data |
| |
| # π‘ Add your logic here |
| |
| return result |
| |
| # π‘ Example usage |
| if __name__ == "__main__": |
| result = {requirement.replace(' ', '_').lower()}("sample data") |
| print(f"Result: {{result}}") |
| ''' |
| |
| def _generate_python_generic(self, requirement: str) -> str: |
| """π Generate generic Python code""" |
| return f'''""" |
| π {requirement} |
| """ |
| |
| def main(): |
| """π― Main function""" |
| print("π Starting: {requirement}") |
| |
| # π‘ Add your code here |
| |
| print("β
Completed: {requirement}") |
| |
| if __name__ == "__main__": |
| main() |
| ''' |
|
|
| |
| __all__ = ['CodingAI', 'CodeFile', 'CodeIssue'] |
|
|