vinaymodel / model /coding_ai.py
hackerbhai's picture
🎯 Add model/coding_ai.py - Complete AI capabilities
50203b3 verified
Raw
History Blame Contribute Delete
21.8 kB
"""
πŸ’» 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 # πŸ”΄ critical, 🟑 warning, πŸ”΅ info
message: str
suggestion: str
category: str # bug, security, performance, style
class CodingAI:
"""πŸ’» Advanced coding intelligence"""
def __init__(self):
self.codebase_cache = {}
self.analysis_history = []
# πŸ—οΈ LARGE PROJECT UNDERSTANDING
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": []
}
# Walk through project
for root, dirs, files in os.walk(project_path):
# Skip common non-code directories
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)
# Track languages
lang = code_file.language
codebase["languages"][lang] = codebase["languages"].get(lang, 0) + 1
# Track dependencies
for imp in code_file.imports:
codebase["dependencies"][imp] = codebase["dependencies"].get(imp, 0) + 1
# Analyze architecture
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()
# Detect language
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')
# Extract functions and classes
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:
# Regex-based extraction for other languages
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": []
}
# Check for common patterns
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")
# Check for microservices
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 = []
# Check for common 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
# πŸ”§ REFACTORING
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":
# PEP 8 improvements
if ' ' not in code and '\t' in code:
refactored = refactored.replace('\t', ' ')
improvements.append("πŸ“ Converted tabs to spaces (PEP 8)")
# Naming conventions
if re.search(r'def [A-Z]\w+\(', refactored):
improvements.append("πŸ“ Consider using snake_case for function names")
# Docstrings
if 'def ' in refactored and '"""' not in refactored:
improvements.append("πŸ“š Add docstrings to functions")
# Type hints
if 'def ' in refactored and '->' not in refactored:
improvements.append("πŸ’‘ Add type hints for better code clarity")
# Magic numbers
if re.search(r'[=<>]=?\s*\d{2,}', refactored):
improvements.append("πŸ”’ Replace magic numbers with named constants")
elif language in ["javascript", "typescript"]:
# Modern JS improvements
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")
# Common improvements
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!"
}
# πŸ› DEBUGGING
def debug_code(self, code: str, error_message: str = None) -> Dict:
"""πŸ› Debug code and find issues"""
print("πŸ› Debugging code...")
issues = []
# Syntax errors
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"
))
# Common bugs
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"
))
# Security issues
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"
))
# Performance issues
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"
))
# Style issues
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!"
}
# πŸ€– AUTONOMOUS AGENT
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": "πŸ€–"
}
# Parse task
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
# πŸ§ͺ CODE GENERATION
def generate_code(self, requirement: str, language: str = "python") -> Dict:
"""πŸ’» Generate code from requirements"""
print(f"πŸ’» Generating {language} code...")
# Template-based generation
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()
'''
# Export classes
__all__ = ['CodingAI', 'CodeFile', 'CodeIssue']