#!/usr/bin/env python3 """ Codebase Ingestion System ========================= This module implements MIRA's codebase ingestion system that enables deep understanding of entire project source code. It provides permanent persistent storage of project knowledge that each steward can install into their codebase. Key Features: - Intelligent file discovery and filtering - Multi-language code parsing and understanding - Semantic code analysis and pattern recognition - Neural network integration for deep comprehension - Incremental updates for efficiency - Cross-project knowledge transfer - Code-aware memory storage The system goes beyond simple text storage - it understands: - Code structure and architecture - Dependencies and relationships - Design patterns and conventions - API surfaces and contracts - Technical debt and quality metrics - Evolution and change history Author: MIRA Codebase Intelligence System Version: 1.0.0 """ import os import sys import json import hashlib import datetime import ast import re from pathlib import Path from typing import Dict, List, Set, Optional, Tuple, Any, Union from dataclasses import dataclass, asdict from enum import Enum import logging import subprocess # Import MIRA's core systems sys.path.insert(0, str(Path(__file__).parent.parent)) from config import MEMORY_DIR from core.memory.memory_manager import get_memory_manager from intelligence.neural_consciousness_system import get_neural_consciousness_system from intelligence.neural_domain_classifier import get_neural_domain_classifier from conversations.conversation_insights import ConversationInsights logger = logging.getLogger(__name__) class CodeFileType(Enum): """Types of code files MIRA can understand.""" PYTHON = "python" JAVASCRIPT = "javascript" TYPESCRIPT = "typescript" JAVA = "java" CPP = "cpp" C = "c" RUST = "rust" GO = "go" RUBY = "ruby" PHP = "php" SWIFT = "swift" KOTLIN = "kotlin" HTML = "html" CSS = "css" JSON = "json" YAML = "yaml" MARKDOWN = "markdown" SHELL = "shell" SQL = "sql" OTHER = "other" class CodeElementType(Enum): """Types of code elements MIRA extracts.""" CLASS = "class" FUNCTION = "function" METHOD = "method" VARIABLE = "variable" CONSTANT = "constant" IMPORT = "import" EXPORT = "export" INTERFACE = "interface" TYPE = "type" ENUM = "enum" DECORATOR = "decorator" COMMENT = "comment" DOCSTRING = "docstring" TEST = "test" CONFIGURATION = "configuration" @dataclass class CodeElement: """Represents a single code element extracted from source.""" element_type: CodeElementType name: str file_path: str line_number: int signature: Optional[str] = None docstring: Optional[str] = None dependencies: List[str] = None complexity: int = 0 test_coverage: Optional[float] = None last_modified: Optional[str] = None def __post_init__(self): if self.dependencies is None: self.dependencies = [] if not self.last_modified: self.last_modified = datetime.datetime.now().isoformat() @dataclass class CodeFile: """Represents a single code file with its analysis.""" file_path: str file_type: CodeFileType language: str size_bytes: int line_count: int hash: str elements: List[CodeElement] imports: List[str] exports: List[str] dependencies: Set[str] complexity_score: float quality_score: float test_coverage: Optional[float] last_analyzed: str neural_embedding: Optional[List[float]] = None def __post_init__(self): if not self.last_analyzed: self.last_analyzed = datetime.datetime.now().isoformat() @dataclass class CodebaseKnowledge: """Represents comprehensive knowledge about a codebase.""" project_root: str project_name: str primary_language: str framework: Optional[str] total_files: int total_lines: int architecture_style: str design_patterns: List[str] api_endpoints: List[Dict[str, Any]] database_schemas: List[Dict[str, Any]] configuration: Dict[str, Any] dependencies: Dict[str, str] quality_metrics: Dict[str, float] technical_debt: List[Dict[str, Any]] ingestion_timestamp: str def __post_init__(self): if not self.ingestion_timestamp: self.ingestion_timestamp = datetime.datetime.now().isoformat() class CodebaseIngestionSystem: """ MIRA's codebase ingestion system for deep code understanding. """ def __init__(self, memory_dir: str = None): self.memory_dir = Path(memory_dir or MEMORY_DIR) self.codebase_dir = self.memory_dir / "codebase_knowledge" self.codebase_dir.mkdir(exist_ok=True) # Storage files self.knowledge_file = self.codebase_dir / "codebase_knowledge.json" self.files_db = self.codebase_dir / "analyzed_files.json" self.elements_db = self.codebase_dir / "code_elements.json" self.index_file = self.codebase_dir / "search_index.json" # Core systems self.memory_manager = get_memory_manager(str(self.memory_dir)) self.neural_system = None # Will use enhanced vidmem for neural encoding self.domain_classifier = get_neural_domain_classifier(str(self.memory_dir)) # Try to get enhanced vidmem for neural encoding try: from core.engine.enhanced_lightning_vidmem import get_enhanced_lightning_vidmem self.enhanced_vidmem = get_enhanced_lightning_vidmem(str(self.memory_dir)) except: self.enhanced_vidmem = None # Language configurations self.file_extensions = { '.py': CodeFileType.PYTHON, '.js': CodeFileType.JAVASCRIPT, '.jsx': CodeFileType.JAVASCRIPT, '.ts': CodeFileType.TYPESCRIPT, '.tsx': CodeFileType.TYPESCRIPT, '.java': CodeFileType.JAVA, '.cpp': CodeFileType.CPP, '.cc': CodeFileType.CPP, '.c': CodeFileType.C, '.h': CodeFileType.C, '.hpp': CodeFileType.CPP, '.rs': CodeFileType.RUST, '.go': CodeFileType.GO, '.rb': CodeFileType.RUBY, '.php': CodeFileType.PHP, '.swift': CodeFileType.SWIFT, '.kt': CodeFileType.KOTLIN, '.html': CodeFileType.HTML, '.css': CodeFileType.CSS, '.scss': CodeFileType.CSS, '.json': CodeFileType.JSON, '.yaml': CodeFileType.YAML, '.yml': CodeFileType.YAML, '.md': CodeFileType.MARKDOWN, '.sh': CodeFileType.SHELL, '.bash': CodeFileType.SHELL, '.sql': CodeFileType.SQL, } # Patterns to ignore self.ignore_patterns = { 'node_modules', '__pycache__', '.git', '.svn', '.hg', 'dist', 'build', 'target', 'out', 'bin', 'obj', '.idea', '.vscode', '.DS_Store', '*.pyc', '*.pyo', '*.class', '*.o', '*.so', '*.dll', '*.exe', '*.jar' } # Load existing knowledge self.codebase_knowledge: Optional[CodebaseKnowledge] = None self.analyzed_files: Dict[str, CodeFile] = {} self.code_elements: Dict[str, List[CodeElement]] = {} self._load_existing_knowledge() def _load_existing_knowledge(self): """Load previously analyzed codebase knowledge.""" if self.knowledge_file.exists(): try: with open(self.knowledge_file, 'r') as f: data = json.load(f) self.codebase_knowledge = CodebaseKnowledge(**data) logger.info(f"📚 Loaded codebase knowledge for {self.codebase_knowledge.project_name}") except Exception as e: logger.error(f"Failed to load codebase knowledge: {e}") if self.files_db.exists(): try: with open(self.files_db, 'r') as f: files_data = json.load(f) for path, file_data in files_data.items(): file_data['file_type'] = CodeFileType(file_data['file_type']) file_data['dependencies'] = set(file_data.get('dependencies', [])) file_data['elements'] = [] # Will load separately self.analyzed_files[path] = CodeFile(**file_data) logger.info(f"📄 Loaded {len(self.analyzed_files)} analyzed files") except Exception as e: logger.error(f"Failed to load analyzed files: {e}") def ingest_codebase(self, project_root: str, incremental: bool = True) -> CodebaseKnowledge: """ Ingest an entire codebase for deep understanding. Args: project_root: Root directory of the project incremental: Only analyze changed files if True Returns: Comprehensive codebase knowledge """ project_root = Path(project_root).resolve() logger.info(f"🧠 CODEBASE INGESTION: Starting analysis of {project_root}") # Phase 1: Project discovery project_info = self._discover_project_info(project_root) # Phase 2: File discovery and filtering code_files = self._discover_code_files(project_root) logger.info(f"📂 Discovered {len(code_files)} code files") # Phase 3: Incremental analysis check files_to_analyze = code_files if incremental and self.codebase_knowledge: files_to_analyze = self._filter_changed_files(code_files) logger.info(f"🔄 Incremental mode: {len(files_to_analyze)} files need analysis") # Phase 4: Deep code analysis analyzed_count = 0 total_elements = 0 for file_path in files_to_analyze: try: code_file = self._analyze_code_file(file_path) if code_file: self.analyzed_files[str(file_path)] = code_file total_elements += len(code_file.elements) analyzed_count += 1 # Store in neural memory self._store_code_knowledge(code_file) if analyzed_count % 100 == 0: logger.info(f"⚡ Analyzed {analyzed_count} files, extracted {total_elements} elements") except Exception as e: logger.error(f"Failed to analyze {file_path}: {e}") # Phase 5: Architecture and pattern analysis architecture_insights = self._analyze_architecture(self.analyzed_files) # Phase 6: Generate comprehensive knowledge self.codebase_knowledge = self._generate_codebase_knowledge( project_root, project_info, architecture_insights ) # Phase 7: Build search index self._build_search_index() # Phase 8: Store knowledge self._save_knowledge() logger.info(f"✅ CODEBASE INGESTION COMPLETE!") logger.info(f" 📊 Analyzed {analyzed_count} files") logger.info(f" 🧩 Extracted {total_elements} code elements") logger.info(f" 🏗️ Architecture: {self.codebase_knowledge.architecture_style}") logger.info(f" 📈 Quality Score: {self.codebase_knowledge.quality_metrics.get('overall', 0):.2f}/100") return self.codebase_knowledge def _discover_project_info(self, project_root: Path) -> Dict[str, Any]: """Discover basic project information.""" info = { 'name': project_root.name, 'type': 'unknown', 'language': None, 'framework': None, 'build_system': None, 'vcs': None, 'dependencies': {} } # Check for VCS if (project_root / '.git').exists(): info['vcs'] = 'git' # Check for package files if (project_root / 'package.json').exists(): info['type'] = 'node' try: with open(project_root / 'package.json') as f: pkg = json.load(f) info['name'] = pkg.get('name', info['name']) info['dependencies'] = pkg.get('dependencies', {}) info['language'] = 'javascript' # Detect framework deps = list(info['dependencies'].keys()) if 'react' in deps: info['framework'] = 'react' elif 'vue' in deps: info['framework'] = 'vue' elif 'express' in deps: info['framework'] = 'express' elif '@angular/core' in deps: info['framework'] = 'angular' except: pass elif (project_root / 'requirements.txt').exists() or (project_root / 'setup.py').exists(): info['type'] = 'python' info['language'] = 'python' # Check for frameworks req_file = project_root / 'requirements.txt' if req_file.exists(): content = req_file.read_text() if 'django' in content.lower(): info['framework'] = 'django' elif 'flask' in content.lower(): info['framework'] = 'flask' elif 'fastapi' in content.lower(): info['framework'] = 'fastapi' elif (project_root / 'pom.xml').exists(): info['type'] = 'maven' info['language'] = 'java' info['build_system'] = 'maven' elif (project_root / 'build.gradle').exists(): info['type'] = 'gradle' info['language'] = 'java' info['build_system'] = 'gradle' elif (project_root / 'Cargo.toml').exists(): info['type'] = 'rust' info['language'] = 'rust' info['build_system'] = 'cargo' elif (project_root / 'go.mod').exists(): info['type'] = 'go' info['language'] = 'go' info['build_system'] = 'go' return info def _discover_code_files(self, project_root: Path) -> List[Path]: """Discover all code files in the project.""" code_files = [] for root, dirs, files in os.walk(project_root): # Filter out ignored directories dirs[:] = [d for d in dirs if not any( pattern in d for pattern in self.ignore_patterns )] for file in files: file_path = Path(root) / file # Skip ignored patterns if any(pattern in str(file_path) for pattern in self.ignore_patterns): continue # Check if it's a code file ext = file_path.suffix.lower() if ext in self.file_extensions: code_files.append(file_path) return code_files def _filter_changed_files(self, files: List[Path]) -> List[Path]: """Filter files that have changed since last analysis.""" changed_files = [] for file_path in files: str_path = str(file_path) # New file if str_path not in self.analyzed_files: changed_files.append(file_path) continue # Check if file has changed try: current_hash = self._calculate_file_hash(file_path) if current_hash != self.analyzed_files[str_path].hash: changed_files.append(file_path) except: changed_files.append(file_path) return changed_files def _calculate_file_hash(self, file_path: Path) -> str: """Calculate hash of file contents.""" hasher = hashlib.sha256() with open(file_path, 'rb') as f: for chunk in iter(lambda: f.read(4096), b''): hasher.update(chunk) return hasher.hexdigest() def _analyze_code_file(self, file_path: Path) -> Optional[CodeFile]: """Analyze a single code file in depth.""" try: # Basic file info stat = file_path.stat() content = file_path.read_text(encoding='utf-8', errors='ignore') lines = content.split('\n') # Determine file type ext = file_path.suffix.lower() file_type = self.file_extensions.get(ext, CodeFileType.OTHER) # Extract code elements based on language elements = [] imports = [] exports = [] dependencies = set() if file_type == CodeFileType.PYTHON: elements, imports, exports = self._analyze_python_file(file_path, content) elif file_type in [CodeFileType.JAVASCRIPT, CodeFileType.TYPESCRIPT]: elements, imports, exports = self._analyze_javascript_file(file_path, content) # Add more language analyzers as needed # Calculate metrics complexity_score = self._calculate_complexity(content, file_type) quality_score = self._calculate_quality_score(content, elements) # Create CodeFile object code_file = CodeFile( file_path=str(file_path), file_type=file_type, language=file_type.value, size_bytes=stat.st_size, line_count=len(lines), hash=self._calculate_file_hash(file_path), elements=elements, imports=imports, exports=exports, dependencies=dependencies, complexity_score=complexity_score, quality_score=quality_score, test_coverage=None, # Would need external tool last_analyzed=datetime.datetime.now().isoformat() ) return code_file except Exception as e: logger.error(f"Error analyzing {file_path}: {e}") return None def _analyze_python_file(self, file_path: Path, content: str) -> Tuple[List[CodeElement], List[str], List[str]]: """Analyze Python file using AST.""" elements = [] imports = [] exports = [] try: tree = ast.parse(content) for node in ast.walk(tree): if isinstance(node, ast.ClassDef): # Extract class docstring = ast.get_docstring(node) elements.append(CodeElement( element_type=CodeElementType.CLASS, name=node.name, file_path=str(file_path), line_number=node.lineno, docstring=docstring )) exports.append(node.name) elif isinstance(node, ast.FunctionDef): # Extract function docstring = ast.get_docstring(node) signature = f"{node.name}({', '.join(arg.arg for arg in node.args.args)})" element_type = CodeElementType.FUNCTION if any(dec.id == 'test' if hasattr(dec, 'id') else False for dec in node.decorator_list): element_type = CodeElementType.TEST elements.append(CodeElement( element_type=element_type, name=node.name, file_path=str(file_path), line_number=node.lineno, signature=signature, docstring=docstring, complexity=self._calculate_function_complexity(node) )) exports.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) except SyntaxError: # Fallback to regex-based parsing pass return elements, imports, exports def _analyze_javascript_file(self, file_path: Path, content: str) -> Tuple[List[CodeElement], List[str], List[str]]: """Analyze JavaScript/TypeScript file using regex patterns.""" elements = [] imports = [] exports = [] # Extract imports import_pattern = r'import\s+(?:{[^}]+}|[\w\s,]+)\s+from\s+[\'"]([^\'\"]+)[\'"]' for match in re.finditer(import_pattern, content): imports.append(match.group(1)) # Extract classes class_pattern = r'(?:export\s+)?class\s+(\w+)' for match in re.finditer(class_pattern, content): elements.append(CodeElement( element_type=CodeElementType.CLASS, name=match.group(1), file_path=str(file_path), line_number=content[:match.start()].count('\n') + 1 )) if 'export' in match.group(0): exports.append(match.group(1)) # Extract functions func_pattern = r'(?:export\s+)?(?:async\s+)?function\s+(\w+)\s*\([^)]*\)' for match in re.finditer(func_pattern, content): elements.append(CodeElement( element_type=CodeElementType.FUNCTION, name=match.group(1), file_path=str(file_path), line_number=content[:match.start()].count('\n') + 1, signature=match.group(0) )) if 'export' in match.group(0): exports.append(match.group(1)) # Extract arrow functions arrow_pattern = r'(?:export\s+)?const\s+(\w+)\s*=\s*(?:async\s+)?\([^)]*\)\s*=>' for match in re.finditer(arrow_pattern, content): elements.append(CodeElement( element_type=CodeElementType.FUNCTION, name=match.group(1), file_path=str(file_path), line_number=content[:match.start()].count('\n') + 1 )) if 'export' in match.group(0): exports.append(match.group(1)) return elements, imports, exports def _calculate_function_complexity(self, node: ast.FunctionDef) -> int: """Calculate cyclomatic complexity of a Python function.""" complexity = 1 # Base complexity for child in ast.walk(node): if isinstance(child, (ast.If, ast.For, ast.While, ast.ExceptHandler)): complexity += 1 elif isinstance(child, ast.BoolOp): complexity += len(child.values) - 1 return complexity def _calculate_complexity(self, content: str, file_type: CodeFileType) -> float: """Calculate overall file complexity.""" lines = content.split('\n') loc = len([l for l in lines if l.strip() and not l.strip().startswith(('#', '//'))]) # Simple complexity heuristics complexity_factors = { 'nested_loops': len(re.findall(r'for.*:\s*\n.*for.*:', content)), 'conditionals': len(re.findall(r'if\s+.*:', content)), 'try_except': len(re.findall(r'try\s*:', content)), 'functions': len(re.findall(r'def\s+\w+|function\s+\w+', content)), 'classes': len(re.findall(r'class\s+\w+', content)) } # Weighted complexity score complexity = ( complexity_factors['nested_loops'] * 3 + complexity_factors['conditionals'] * 1 + complexity_factors['try_except'] * 2 + complexity_factors['functions'] * 0.5 + complexity_factors['classes'] * 1 ) / max(1, loc) * 100 return min(100, complexity) def _calculate_quality_score(self, content: str, elements: List[CodeElement]) -> float: """Calculate code quality score.""" score = 100.0 # Check for documentation documented_elements = len([e for e in elements if e.docstring]) if elements: doc_ratio = documented_elements / len(elements) score -= (1 - doc_ratio) * 20 # -20 points for no docs # Check for code smells lines = content.split('\n') long_lines = len([l for l in lines if len(l) > 120]) if lines: score -= (long_lines / len(lines)) * 10 # -10 points for long lines # Check for TODO/FIXME comments todos = len(re.findall(r'TODO|FIXME|HACK|XXX', content)) score -= min(10, todos * 2) # -2 points per TODO, max -10 # Check for consistent naming snake_case = len(re.findall(r'def\s+[a-z_]+\(', content)) camel_case = len(re.findall(r'def\s+[a-zA-Z]+\(', content)) if snake_case > 0 and camel_case > 0: score -= 5 # Mixed naming conventions return max(0, score) def _analyze_architecture(self, files: Dict[str, CodeFile]) -> Dict[str, Any]: """Analyze overall codebase architecture.""" insights = { 'style': 'unknown', 'patterns': [], 'layers': [], 'modules': {}, 'api_style': None } # Detect architecture patterns file_paths = list(files.keys()) # MVC pattern if any('controller' in p.lower() for p in file_paths): if any('model' in p.lower() for p in file_paths): if any('view' in p.lower() for p in file_paths): insights['patterns'].append('MVC') insights['style'] = 'mvc' # Layered architecture if any('service' in p.lower() for p in file_paths): if any('repository' in p.lower() or 'dao' in p.lower() for p in file_paths): insights['patterns'].append('Layered') insights['style'] = 'layered' # Microservices if any('api' in p.lower() for p in file_paths) and len(files) > 100: service_dirs = set() for path in file_paths: parts = path.split('/') if len(parts) > 2 and 'service' in parts[1].lower(): service_dirs.add(parts[1]) if len(service_dirs) > 3: insights['patterns'].append('Microservices') insights['style'] = 'microservices' # Detect API style rest_indicators = sum(1 for p in file_paths if any( method in p.lower() for method in ['get', 'post', 'put', 'delete', 'rest'] )) graphql_indicators = sum(1 for p in file_paths if 'graphql' in p.lower()) if graphql_indicators > 0: insights['api_style'] = 'GraphQL' elif rest_indicators > 5: insights['api_style'] = 'REST' return insights def _generate_codebase_knowledge(self, project_root: Path, project_info: Dict[str, Any], architecture: Dict[str, Any]) -> CodebaseKnowledge: """Generate comprehensive codebase knowledge.""" # Calculate statistics total_lines = sum(f.line_count for f in self.analyzed_files.values()) total_elements = sum(len(self.code_elements.get(path, [])) for path in self.analyzed_files) # Language distribution language_counts = {} for file in self.analyzed_files.values(): language_counts[file.language] = language_counts.get(file.language, 0) + 1 primary_language = max(language_counts.items(), key=lambda x: x[1])[0] if language_counts else 'unknown' # Calculate quality metrics quality_scores = [f.quality_score for f in self.analyzed_files.values()] complexity_scores = [f.complexity_score for f in self.analyzed_files.values()] quality_metrics = { 'overall': sum(quality_scores) / len(quality_scores) if quality_scores else 0, 'complexity': sum(complexity_scores) / len(complexity_scores) if complexity_scores else 0, 'documentation': self._calculate_documentation_coverage(), 'test_coverage': self._estimate_test_coverage(), 'maintainability': self._calculate_maintainability_index() } # Identify technical debt technical_debt = self._identify_technical_debt() # Extract design patterns design_patterns = architecture.get('patterns', []) design_patterns.extend(self._detect_design_patterns()) return CodebaseKnowledge( project_root=str(project_root), project_name=project_info['name'], primary_language=primary_language, framework=project_info.get('framework'), total_files=len(self.analyzed_files), total_lines=total_lines, architecture_style=architecture.get('style', 'monolithic'), design_patterns=list(set(design_patterns)), api_endpoints=self._extract_api_endpoints(), database_schemas=self._extract_database_schemas(), configuration=self._extract_configuration(), dependencies=project_info.get('dependencies', {}), quality_metrics=quality_metrics, technical_debt=technical_debt, ingestion_timestamp=datetime.datetime.now().isoformat() ) def _calculate_documentation_coverage(self) -> float: """Calculate percentage of documented code elements.""" total_elements = 0 documented_elements = 0 for elements_list in self.code_elements.values(): for element in elements_list: total_elements += 1 if element.docstring: documented_elements += 1 return (documented_elements / total_elements * 100) if total_elements > 0 else 0 def _estimate_test_coverage(self) -> float: """Estimate test coverage based on test file analysis.""" test_files = [f for f in self.analyzed_files.values() if 'test' in f.file_path.lower()] source_files = [f for f in self.analyzed_files.values() if 'test' not in f.file_path.lower()] if not source_files: return 0 # Simple heuristic: ratio of test files to source files test_ratio = len(test_files) / len(source_files) * 100 # Bonus for actual test functions test_functions = sum( len([e for e in self.code_elements.get(f.file_path, []) if e.element_type == CodeElementType.TEST]) for f in test_files ) source_functions = sum( len([e for e in self.code_elements.get(f.file_path, []) if e.element_type in [CodeElementType.FUNCTION, CodeElementType.METHOD]]) for f in source_files ) if source_functions > 0: function_coverage = test_functions / source_functions * 100 return min(100, (test_ratio + function_coverage) / 2) return min(100, test_ratio) def _calculate_maintainability_index(self) -> float: """Calculate maintainability index.""" factors = { 'quality': sum(f.quality_score for f in self.analyzed_files.values()) / len(self.analyzed_files), 'complexity': 100 - sum(f.complexity_score for f in self.analyzed_files.values()) / len(self.analyzed_files), 'documentation': self._calculate_documentation_coverage(), 'test_coverage': self._estimate_test_coverage() } # Weighted average weights = {'quality': 0.3, 'complexity': 0.3, 'documentation': 0.2, 'test_coverage': 0.2} maintainability = sum(factors[k] * weights[k] for k in factors) return maintainability def _identify_technical_debt(self) -> List[Dict[str, Any]]: """Identify technical debt items.""" debt_items = [] # Check for TODO/FIXME comments for file_path, file_obj in self.analyzed_files.items(): try: content = Path(file_path).read_text() for i, line in enumerate(content.split('\n'), 1): if any(marker in line for marker in ['TODO', 'FIXME', 'HACK', 'XXX']): debt_items.append({ 'type': 'todo', 'severity': 'medium', 'file': file_path, 'line': i, 'description': line.strip() }) except: pass # Check for high complexity functions for file_path, elements in self.code_elements.items(): for element in elements: if element.complexity > 10: debt_items.append({ 'type': 'complexity', 'severity': 'high' if element.complexity > 15 else 'medium', 'file': file_path, 'line': element.line_number, 'description': f"High complexity function: {element.name} (complexity: {element.complexity})" }) # Check for missing documentation undocumented_classes = [] for elements in self.code_elements.values(): for element in elements: if element.element_type == CodeElementType.CLASS and not element.docstring: undocumented_classes.append(element) if len(undocumented_classes) > 5: debt_items.append({ 'type': 'documentation', 'severity': 'low', 'description': f"{len(undocumented_classes)} classes lack documentation" }) return debt_items def _detect_design_patterns(self) -> List[str]: """Detect common design patterns in the codebase.""" patterns = [] # Singleton pattern for elements in self.code_elements.values(): for element in elements: if element.element_type == CodeElementType.CLASS: # Check for singleton indicators if any(keyword in element.name.lower() for keyword in ['singleton', 'instance']): patterns.append('Singleton') # Factory pattern factory_methods = [e for elements in self.code_elements.values() for e in elements if 'factory' in e.name.lower() or 'create' in e.name.lower()] if len(factory_methods) > 3: patterns.append('Factory') # Observer pattern observer_elements = [e for elements in self.code_elements.values() for e in elements if any(word in e.name.lower() for word in ['observer', 'listener', 'subscriber'])] if len(observer_elements) > 2: patterns.append('Observer') # Strategy pattern strategy_interfaces = [e for elements in self.code_elements.values() for e in elements if 'strategy' in e.name.lower() or 'policy' in e.name.lower()] if len(strategy_interfaces) > 2: patterns.append('Strategy') return list(set(patterns)) def _extract_api_endpoints(self) -> List[Dict[str, Any]]: """Extract API endpoints from the codebase.""" endpoints = [] # Common API patterns api_patterns = [ r'@app\.route\([\'"]([^\'\"]+)[\'"].*?\)', # Flask r'@router\.(get|post|put|delete|patch)\([\'"]([^\'\"]+)[\'"]', # FastAPI r'router\.(get|post|put|delete|patch)\([\'"]([^\'\"]+)[\'"]', # Express r'@(Get|Post|Put|Delete|Patch)Mapping\([\'"]([^\'\"]+)[\'"]', # Spring ] for file_path, file_obj in self.analyzed_files.items(): if file_obj.file_type in [CodeFileType.PYTHON, CodeFileType.JAVASCRIPT, CodeFileType.TYPESCRIPT, CodeFileType.JAVA]: try: content = Path(file_path).read_text() for pattern in api_patterns: for match in re.finditer(pattern, content): if len(match.groups()) >= 2: method = match.group(1) path = match.group(2) else: method = 'GET' path = match.group(1) endpoints.append({ 'method': method.upper(), 'path': path, 'file': file_path, 'line': content[:match.start()].count('\n') + 1 }) except: pass return endpoints def _extract_database_schemas(self) -> List[Dict[str, Any]]: """Extract database schema information.""" schemas = [] # Look for ORM models for file_path, elements in self.code_elements.items(): for element in elements: if element.element_type == CodeElementType.CLASS: # Common ORM base classes orm_indicators = ['model', 'entity', 'document', 'schema', 'table'] if any(indicator in element.name.lower() for indicator in orm_indicators): schemas.append({ 'name': element.name, 'type': 'orm_model', 'file': file_path, 'line': element.line_number }) # Look for SQL files sql_files = [f for f in self.analyzed_files.values() if f.file_type == CodeFileType.SQL] for sql_file in sql_files: schemas.append({ 'name': Path(sql_file.file_path).stem, 'type': 'sql_schema', 'file': sql_file.file_path }) return schemas def _extract_configuration(self) -> Dict[str, Any]: """Extract configuration information.""" config = {} # Look for config files config_patterns = ['config', 'settings', 'env', 'properties'] for file_path, file_obj in self.analyzed_files.items(): file_name = Path(file_path).name.lower() if any(pattern in file_name for pattern in config_patterns): if file_obj.file_type == CodeFileType.JSON: try: content = json.loads(Path(file_path).read_text()) config[file_name] = content except: pass elif file_obj.file_type == CodeFileType.YAML: config[file_name] = {'type': 'yaml', 'path': file_path} return config def _store_code_knowledge(self, code_file: CodeFile): """Store code knowledge in neural memory.""" # Create memory content memory_content = f""" Code File Analysis: {code_file.file_path} Language: {code_file.language} Type: {code_file.file_type.value} Lines: {code_file.line_count} Quality Score: {code_file.quality_score:.1f}/100 Complexity: {code_file.complexity_score:.1f}/100 Elements: """ for element in code_file.elements[:10]: # First 10 elements memory_content += f"- {element.element_type.value}: {element.name}\n" if len(code_file.elements) > 10: memory_content += f"... and {len(code_file.elements) - 10} more elements\n" # Store in memory with neural processing memory_id = self.memory_manager.store_memory(memory_content, "codebase_analysis") # Add to search index with semantic embedding if self.enhanced_vidmem and hasattr(self.enhanced_vidmem, 'encode_memory_with_consciousness'): frame = self.enhanced_vidmem.encode_memory_with_consciousness(memory_content, context="codebase file") # For now, use the frame_id as a simple identifier # TODO: Implement proper neural embeddings for semantic search code_file.neural_embedding = None def _build_search_index(self): """Build search index for code elements.""" index = { 'files': {}, 'elements': {}, 'symbols': {} } # Index files for file_path, file_obj in self.analyzed_files.items(): index['files'][file_path] = { 'type': file_obj.file_type.value, 'language': file_obj.language, 'quality': file_obj.quality_score, 'elements': len(file_obj.elements) } # Index elements for file_path, elements in self.code_elements.items(): for element in elements: if element.name not in index['symbols']: index['symbols'][element.name] = [] index['symbols'][element.name].append({ 'type': element.element_type.value, 'file': file_path, 'line': element.line_number }) # Save index with open(self.index_file, 'w') as f: json.dump(index, f, indent=2) def _save_knowledge(self): """Save all analyzed knowledge to disk.""" # Save codebase knowledge if self.codebase_knowledge: with open(self.knowledge_file, 'w') as f: json.dump(asdict(self.codebase_knowledge), f, indent=2) # Save analyzed files files_data = {} for path, file_obj in self.analyzed_files.items(): file_data = asdict(file_obj) file_data['file_type'] = file_obj.file_type.value file_data['dependencies'] = list(file_obj.dependencies) del file_data['elements'] # Save separately files_data[path] = file_data with open(self.files_db, 'w') as f: json.dump(files_data, f, indent=2) # Save code elements elements_data = {} for path, elements in self.code_elements.items(): elements_data[path] = [asdict(e) for e in elements] for e_data, e_obj in zip(elements_data[path], elements): e_data['element_type'] = e_obj.element_type.value with open(self.elements_db, 'w') as f: json.dump(elements_data, f, indent=2) def search_code(self, query: str, limit: int = 10) -> List[Dict[str, Any]]: """Search code using natural language.""" results = [] # Load analyzed files from disk if self.files_db.exists(): with open(self.files_db, 'r') as f: files_data = json.load(f) self.analyzed_files = {} for file_path, file_data in files_data.items(): # Reconstruct CodeFile objects elements = [] for elem_data in file_data.get('elements', []): elements.append(CodeElement( element_type=CodeElementType(elem_data['element_type']), name=elem_data['name'], line_number=elem_data['line_number'], end_line=elem_data.get('end_line'), docstring=elem_data.get('docstring'), signature=elem_data.get('signature'), parent=elem_data.get('parent'), modifiers=elem_data.get('modifiers', []) )) self.analyzed_files[file_path] = CodeFile( file_path=file_data['file_path'], file_type=CodeFileType(file_data['file_type']), language=file_data['language'], size_bytes=file_data['size_bytes'], line_count=file_data.get('line_count', file_data.get('lines_of_code', 0)), elements=elements, imports=file_data.get('imports', []), exports=file_data.get('exports', []), dependencies=set(file_data.get('dependencies', [])), complexity_score=file_data.get('complexity_score', 0), quality_score=file_data.get('quality_score', 0), test_coverage=file_data.get('test_coverage', 0), last_analyzed=file_data.get('last_analyzed', datetime.datetime.now().isoformat()), hash=file_data.get('hash', ''), neural_embedding=file_data.get('neural_embedding') ) # Keyword search with scoring query_lower = query.lower() query_words = query_lower.split() for file_path, file_obj in self.analyzed_files.items(): score = 0.0 file_path_lower = file_path.lower() # Score based on filename match for word in query_words: if word in file_path_lower: score += 0.5 # Score based on element names for element in file_obj.elements: element_name_lower = element.name.lower() for word in query_words: if word in element_name_lower: score += 0.3 # Quality bonus score += 0.1 * file_obj.quality_score / 100 if score > 0: results.append({ 'file': file_path, 'type': file_obj.file_type.value, 'language': file_obj.language, 'score': min(1.0, score), 'quality': file_obj.quality_score, 'elements': len(file_obj.elements), 'preview': self._get_element_preview(file_obj) }) # Sort by score results.sort(key=lambda x: x['score'], reverse=True) return results[:limit] def _get_element_preview(self, file_obj: CodeFile) -> str: """Get a preview of the file's main elements.""" preview_parts = [] # Get main classes classes = [e for e in file_obj.elements if e.element_type == CodeElementType.CLASS] if classes: preview_parts.append(f"classes: {', '.join(c.name for c in classes[:3])}") # Get main functions functions = [e for e in file_obj.elements if e.element_type in [CodeElementType.FUNCTION, CodeElementType.METHOD]] if functions: preview_parts.append(f"functions: {', '.join(f.name for f in functions[:3])}") return " | ".join(preview_parts) if preview_parts else f"{len(file_obj.elements)} elements" def _cosine_similarity(self, vec1: List[float], vec2: List[float]) -> float: """Calculate cosine similarity between two vectors.""" dot_product = sum(a * b for a, b in zip(vec1, vec2)) magnitude1 = sum(a * a for a in vec1) ** 0.5 magnitude2 = sum(b * b for b in vec2) ** 0.5 if magnitude1 == 0 or magnitude2 == 0: return 0.0 return dot_product / (magnitude1 * magnitude2) def get_code_insights(self) -> Dict[str, Any]: """Get high-level insights about the codebase.""" if not self.codebase_knowledge: return {"error": "No codebase analyzed yet"} insights = { "overview": { "project": self.codebase_knowledge.project_name, "language": self.codebase_knowledge.primary_language, "framework": self.codebase_knowledge.framework, "architecture": self.codebase_knowledge.architecture_style, "size": { "files": self.codebase_knowledge.total_files, "lines": self.codebase_knowledge.total_lines } }, "quality": self.codebase_knowledge.quality_metrics, "patterns": self.codebase_knowledge.design_patterns, "technical_debt": { "total_items": len(self.codebase_knowledge.technical_debt), "by_severity": {} }, "recommendations": [] } # Analyze technical debt by severity for item in self.codebase_knowledge.technical_debt: severity = item.get('severity', 'unknown') insights['technical_debt']['by_severity'][severity] = \ insights['technical_debt']['by_severity'].get(severity, 0) + 1 # Generate recommendations quality = self.codebase_knowledge.quality_metrics['overall'] if quality < 60: insights['recommendations'].append("Consider improving code documentation") if self.codebase_knowledge.quality_metrics['complexity'] > 40: insights['recommendations'].append("Refactor complex functions to improve maintainability") if self.codebase_knowledge.quality_metrics['test_coverage'] < 50: insights['recommendations'].append("Increase test coverage for better reliability") if len(self.codebase_knowledge.technical_debt) > 20: insights['recommendations'].append("Address technical debt to improve code health") return insights def get_codebase_ingestion_system(memory_dir: str = None) -> CodebaseIngestionSystem: """Get the codebase ingestion system instance.""" return CodebaseIngestionSystem(memory_dir) # CLI interface if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="MIRA Codebase Ingestion System") parser.add_argument("command", choices=["ingest", "search", "insights"], help="Command to execute") parser.add_argument("--path", default=".", help="Project path to analyze") parser.add_argument("--query", help="Search query") parser.add_argument("--incremental", action="store_true", help="Only analyze changed files") args = parser.parse_args() system = get_codebase_ingestion_system() if args.command == "ingest": print(f"🧠 Ingesting codebase at {args.path}...") knowledge = system.ingest_codebase(args.path, incremental=args.incremental) print(f"✅ Ingestion complete!") elif args.command == "search": if not args.query: print("❌ Please provide a search query with --query") else: results = system.search_code(args.query) print(f"🔍 Search results for '{args.query}':") for i, result in enumerate(results, 1): print(f"{i}. {result['file']} (score: {result['score']:.2f})") elif args.command == "insights": insights = system.get_code_insights() print("📊 Codebase Insights:") print(json.dumps(insights, indent=2))