Module for finding relevant code snippets and analyzing relationships.
| 39 | return s.lower().replace('_', '').replace(' ', '') |
| 40 | |
| 41 | class CodeFinder: |
| 42 | """Module for finding relevant code snippets and analyzing relationships.""" |
| 43 | |
| 44 | def __init__(self, db_manager: DatabaseManager): |
| 45 | self.db_manager = db_manager |
| 46 | self.driver = self.db_manager.get_driver() |
| 47 | self._lacks_native_fulltext = getattr(db_manager, 'get_backend_type', lambda: 'neo4j')() != 'neo4j' |
| 48 | |
| 49 | def format_query(self, find_by: Literal["Class", "Function"], fuzzy_search:bool, repo_path: Optional[str] = None) -> str: |
| 50 | """Format the search query based on the search type and fuzzy search settings.""" |
| 51 | repo_filter = "AND node.path STARTS WITH $repo_path" if repo_path else "" |
| 52 | if self._lacks_native_fulltext: |
| 53 | # FalkorDB does not support CALL db.idx.fulltext.queryNodes. |
| 54 | # Fall back to a pure Cypher CONTAINS/toLower match on node name. |
| 55 | name_filter = "toLower(node.name) CONTAINS toLower($search_term)" |
| 56 | return f""" |
| 57 | MATCH (node:{find_by}) |
| 58 | WHERE {name_filter} {repo_filter} |
| 59 | RETURN node.name as name, node.path as path, node.line_number as line_number, |
| 60 | node.source as source, node.docstring as docstring, node.is_dependency as is_dependency |
| 61 | ORDER BY node.is_dependency ASC, node.name |
| 62 | LIMIT 20 |
| 63 | """ |
| 64 | return f""" |
| 65 | CALL db.index.fulltext.queryNodes("code_search_index", $search_term) YIELD node, score |
| 66 | WITH node, score |
| 67 | WHERE node:{find_by} {'AND node.name CONTAINS $search_term' if not fuzzy_search else ''} {repo_filter} |
| 68 | RETURN node.name as name, node.path as path, node.line_number as line_number, |
| 69 | node.source as source, node.docstring as docstring, node.is_dependency as is_dependency |
| 70 | ORDER BY score DESC |
| 71 | LIMIT 20 |
| 72 | """ |
| 73 | |
| 74 | def _find_by_name_fuzzy_portable( |
| 75 | self, |
| 76 | label: Literal["Function", "Class"], |
| 77 | search_term: str, |
| 78 | edit_distance: int, |
| 79 | repo_path: Optional[str], |
| 80 | ) -> List[Dict]: |
| 81 | """Fuzzy name match for backends without Lucene fuzzy syntax (Kùzu, FalkorDB, …). |
| 82 | |
| 83 | Compares both the raw query and its identifier-normalised form against each |
| 84 | candidate name, taking the minimum distance. This lets camelCase queries |
| 85 | match snake_case stored names and vice-versa without inflating the distance. |
| 86 | """ |
| 87 | if not search_term.strip(): |
| 88 | return [] |
| 89 | where_clause = "WHERE node.path STARTS WITH $repo_path" if repo_path else "" |
| 90 | # Without a repo filter we must cap the candidate scan. 20 000 is enough to |
| 91 | # cover any realistic single-repo codebase while keeping latency acceptable. |
| 92 | limit_tail = "" if repo_path else " LIMIT 20000" |
| 93 | params: Dict[str, Any] = {} |
| 94 | if repo_path: |
| 95 | params["repo_path"] = repo_path |
| 96 | query = f""" |
| 97 | MATCH (node:{label}) |
| 98 | {where_clause} |
no outgoing calls