import ast import re from typing import Dict, List, Any, Optional from backend.indexer.manager import WorkspaceIndexer from backend.indexer.incremental import IncrementalIndexerCache class CodeIntelligenceEngine: """Provides navigation, symbol refactoring, and code diagnostics.""" def __init__(self, indexer: WorkspaceIndexer): self.indexer = indexer self.cache = IncrementalIndexerCache() self.file_contents: Dict[str, str] = {} def index_workspace_file(self, filepath: str, content: str): if self.cache.should_reindex(filepath, content): self.file_contents[filepath] = content self.indexer.index_file(filepath, content) def go_to_definition(self, symbol: str) -> Optional[Dict[str, Any]]: """Finds definition target (filepath and line number) for a given symbol.""" for filepath, content in self.file_contents.items(): try: tree = ast.parse(content) for node in ast.walk(tree): if isinstance(node, (ast.FunctionDef, ast.ClassDef)) and node.name == symbol: return { "filepath": filepath, "line": node.lineno, "col": node.col_offset, "type": "class" if isinstance(node, ast.ClassDef) else "function" } except SyntaxError: continue return None def find_references(self, symbol: str) -> List[Dict[str, Any]]: """Finds all occurrences/usages of a symbol across the workspace.""" refs = [] pattern = re.compile(r'\b' + re.escape(symbol) + r'\b') for filepath, content in self.file_contents.items(): for line_no, line in enumerate(content.splitlines(), start=1): for match in pattern.finditer(line): refs.append({ "filepath": filepath, "line": line_no, "col": match.start(), "text": line.strip() }) return refs def rename_symbol(self, old_name: str, new_name: str) -> Dict[str, str]: """Performs refactoring by renaming symbols across workspace files.""" pattern = re.compile(r'\b' + re.escape(old_name) + r'\b') modified_files = {} for filepath, content in self.file_contents.items(): if pattern.search(content): new_content = pattern.sub(new_name, content) modified_files[filepath] = new_content self.file_contents[filepath] = new_content self.cache.invalidate(filepath) self.indexer.index_file(filepath, new_content) return modified_files def get_diagnostics(self, filepath: str) -> List[Dict[str, Any]]: """Analyzes AST syntax errors and basic dead-code/unused imports.""" diagnostics = [] content = self.file_contents.get(filepath, "") if not content: return diagnostics try: tree = ast.parse(content) except SyntaxError as err: diagnostics.append({ "severity": "error", "message": f"SyntaxError: {err.msg}", "line": err.lineno, "col": err.offset }) return diagnostics # Check unused imports / unused names imported_names = set() for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: imported_names.add(alias.asname or alias.name) elif isinstance(node, ast.ImportFrom): for alias in node.names: imported_names.add(alias.asname or alias.name) full_text = content for name in imported_names: # Simple heuristic check if name occurs only once (the import statement itself) occurrences = len(re.findall(r'\b' + re.escape(name) + r'\b', full_text)) if occurrences <= 1: diagnostics.append({ "severity": "warning", "message": f"Potentially unused import: '{name}'", "line": 1 }) return diagnostics