Spaces:
Sleeping
Sleeping
| 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 | |