Navegador

navegador / navegador / ingestion / go.py
Source Blame History 233 lines
e52aae0… lmata 1 """
e52aae0… lmata 2 Go AST parser — extracts functions, methods, struct/interface types,
e52aae0… lmata 3 imports, and call edges from .go files using tree-sitter.
e52aae0… lmata 4 """
e52aae0… lmata 5
e52aae0… lmata 6 import logging
e52aae0… lmata 7 from pathlib import Path
e52aae0… lmata 8
e52aae0… lmata 9 from navegador.graph.schema import EdgeType, NodeLabel
e52aae0… lmata 10 from navegador.graph.store import GraphStore
e52aae0… lmata 11 from navegador.ingestion.parser import LanguageParser
e52aae0… lmata 12
e52aae0… lmata 13 logger = logging.getLogger(__name__)
e52aae0… lmata 14
e52aae0… lmata 15
e52aae0… lmata 16 def _get_go_language():
e52aae0… lmata 17 try:
e52aae0… lmata 18 import tree_sitter_go as tsgo # type: ignore[import]
e52aae0… lmata 19 from tree_sitter import Language
758c6b7… lmata 20
e52aae0… lmata 21 return Language(tsgo.language())
e52aae0… lmata 22 except ImportError as e:
e52aae0… lmata 23 raise ImportError("Install tree-sitter-go: pip install tree-sitter-go") from e
e52aae0… lmata 24
e52aae0… lmata 25
e52aae0… lmata 26 def _node_text(node, source: bytes) -> str:
758c6b7… lmata 27 return source[node.start_byte : node.end_byte].decode("utf-8", errors="replace")
e52aae0… lmata 28
e52aae0… lmata 29
e52aae0… lmata 30 class GoParser(LanguageParser):
e52aae0… lmata 31 """Parses Go source files into the navegador graph."""
e52aae0… lmata 32
e52aae0… lmata 33 def __init__(self) -> None:
e52aae0… lmata 34 from tree_sitter import Parser # type: ignore[import]
758c6b7… lmata 35
e52aae0… lmata 36 self._parser = Parser(_get_go_language())
e52aae0… lmata 37
e52aae0… lmata 38 def parse_file(self, path: Path, repo_root: Path, store: GraphStore) -> dict[str, int]:
e52aae0… lmata 39 source = path.read_bytes()
e52aae0… lmata 40 tree = self._parser.parse(source)
e52aae0… lmata 41 rel_path = str(path.relative_to(repo_root))
e52aae0… lmata 42
758c6b7… lmata 43 store.create_node(
758c6b7… lmata 44 NodeLabel.File,
758c6b7… lmata 45 {
758c6b7… lmata 46 "name": path.name,
758c6b7… lmata 47 "path": rel_path,
758c6b7… lmata 48 "language": "go",
758c6b7… lmata 49 "line_count": source.count(b"\n"),
758c6b7… lmata 50 },
758c6b7… lmata 51 )
e52aae0… lmata 52
e52aae0… lmata 53 stats = {"functions": 0, "classes": 0, "edges": 0}
e52aae0… lmata 54 self._walk(tree.root_node, source, rel_path, store, stats)
e52aae0… lmata 55 return stats
e52aae0… lmata 56
e52aae0… lmata 57 # ── AST walker ────────────────────────────────────────────────────────────
e52aae0… lmata 58
758c6b7… lmata 59 def _walk(self, node, source: bytes, file_path: str, store: GraphStore, stats: dict) -> None:
e52aae0… lmata 60 if node.type == "function_declaration":
e52aae0… lmata 61 self._handle_function(node, source, file_path, store, stats, receiver=None)
e52aae0… lmata 62 return
e52aae0… lmata 63 if node.type == "method_declaration":
e52aae0… lmata 64 self._handle_method(node, source, file_path, store, stats)
e52aae0… lmata 65 return
e52aae0… lmata 66 if node.type == "type_declaration":
e52aae0… lmata 67 self._handle_type(node, source, file_path, store, stats)
e52aae0… lmata 68 return
e52aae0… lmata 69 if node.type == "import_declaration":
e52aae0… lmata 70 self._handle_import(node, source, file_path, store, stats)
e52aae0… lmata 71 return
e52aae0… lmata 72 for child in node.children:
e52aae0… lmata 73 self._walk(child, source, file_path, store, stats)
e52aae0… lmata 74
e52aae0… lmata 75 # ── Handlers ──────────────────────────────────────────────────────────────
e52aae0… lmata 76
758c6b7… lmata 77 def _handle_function(
758c6b7… lmata 78 self,
758c6b7… lmata 79 node,
758c6b7… lmata 80 source: bytes,
758c6b7… lmata 81 file_path: str,
758c6b7… lmata 82 store: GraphStore,
758c6b7… lmata 83 stats: dict,
758c6b7… lmata 84 receiver: str | None,
758c6b7… lmata 85 ) -> None:
e52aae0… lmata 86 name_node = node.child_by_field_name("name")
e52aae0… lmata 87 if not name_node:
e52aae0… lmata 88 return
e52aae0… lmata 89 name = _node_text(name_node, source)
e52aae0… lmata 90 label = NodeLabel.Method if receiver else NodeLabel.Function
e52aae0… lmata 91
758c6b7… lmata 92 store.create_node(
758c6b7… lmata 93 label,
758c6b7… lmata 94 {
758c6b7… lmata 95 "name": name,
758c6b7… lmata 96 "file_path": file_path,
758c6b7… lmata 97 "line_start": node.start_point[0] + 1,
758c6b7… lmata 98 "line_end": node.end_point[0] + 1,
758c6b7… lmata 99 "docstring": "",
758c6b7… lmata 100 "class_name": receiver or "",
758c6b7… lmata 101 },
758c6b7… lmata 102 )
e52aae0… lmata 103
e52aae0… lmata 104 container_label = NodeLabel.Class if receiver else NodeLabel.File
e52aae0… lmata 105 container_key = (
758c6b7… lmata 106 {"name": receiver, "file_path": file_path} if receiver else {"path": file_path}
e52aae0… lmata 107 )
e52aae0… lmata 108 store.create_edge(
758c6b7… lmata 109 container_label,
758c6b7… lmata 110 container_key,
e52aae0… lmata 111 EdgeType.CONTAINS,
758c6b7… lmata 112 label,
758c6b7… lmata 113 {"name": name, "file_path": file_path},
e52aae0… lmata 114 )
e52aae0… lmata 115 stats["functions"] += 1
e52aae0… lmata 116 stats["edges"] += 1
e52aae0… lmata 117
e52aae0… lmata 118 self._extract_calls(node, source, file_path, name, label, store, stats)
e52aae0… lmata 119
758c6b7… lmata 120 def _handle_method(
758c6b7… lmata 121 self, node, source: bytes, file_path: str, store: GraphStore, stats: dict
758c6b7… lmata 122 ) -> None:
e52aae0… lmata 123 receiver_type = ""
e52aae0… lmata 124 recv_node = node.child_by_field_name("receiver")
e52aae0… lmata 125 if recv_node:
e52aae0… lmata 126 for child in recv_node.children:
e52aae0… lmata 127 if child.type == "parameter_declaration":
e52aae0… lmata 128 for c in child.children:
e52aae0… lmata 129 if c.type in ("type_identifier", "pointer_type"):
e52aae0… lmata 130 receiver_type = _node_text(c, source).lstrip("*").strip()
e52aae0… lmata 131 break
758c6b7… lmata 132 self._handle_function(node, source, file_path, store, stats, receiver=receiver_type or None)
e52aae0… lmata 133
758c6b7… lmata 134 def _handle_type(
758c6b7… lmata 135 self, node, source: bytes, file_path: str, store: GraphStore, stats: dict
758c6b7… lmata 136 ) -> None:
e52aae0… lmata 137 for child in node.children:
e52aae0… lmata 138 if child.type != "type_spec":
e52aae0… lmata 139 continue
e52aae0… lmata 140 name_node = child.child_by_field_name("name")
e52aae0… lmata 141 type_node = child.child_by_field_name("type")
e52aae0… lmata 142 if not name_node or not type_node:
e52aae0… lmata 143 continue
e52aae0… lmata 144 if type_node.type not in ("struct_type", "interface_type"):
e52aae0… lmata 145 continue
e52aae0… lmata 146 name = _node_text(name_node, source)
e52aae0… lmata 147 kind = "struct" if type_node.type == "struct_type" else "interface"
758c6b7… lmata 148 store.create_node(
758c6b7… lmata 149 NodeLabel.Class,
758c6b7… lmata 150 {
758c6b7… lmata 151 "name": name,
758c6b7… lmata 152 "file_path": file_path,
758c6b7… lmata 153 "line_start": node.start_point[0] + 1,
758c6b7… lmata 154 "line_end": node.end_point[0] + 1,
758c6b7… lmata 155 "docstring": kind,
758c6b7… lmata 156 },
758c6b7… lmata 157 )
e52aae0… lmata 158 store.create_edge(
758c6b7… lmata 159 NodeLabel.File,
758c6b7… lmata 160 {"path": file_path},
e52aae0… lmata 161 EdgeType.CONTAINS,
758c6b7… lmata 162 NodeLabel.Class,
758c6b7… lmata 163 {"name": name, "file_path": file_path},
e52aae0… lmata 164 )
e52aae0… lmata 165 stats["classes"] += 1
e52aae0… lmata 166 stats["edges"] += 1
e52aae0… lmata 167
758c6b7… lmata 168 def _handle_import(
758c6b7… lmata 169 self, node, source: bytes, file_path: str, store: GraphStore, stats: dict
758c6b7… lmata 170 ) -> None:
e52aae0… lmata 171 line_start = node.start_point[0] + 1
e52aae0… lmata 172 for child in node.children:
e52aae0… lmata 173 if child.type == "import_spec":
e52aae0… lmata 174 self._ingest_import_spec(child, source, file_path, line_start, store, stats)
e52aae0… lmata 175 elif child.type == "import_spec_list":
e52aae0… lmata 176 for spec in child.children:
e52aae0… lmata 177 if spec.type == "import_spec":
758c6b7… lmata 178 self._ingest_import_spec(spec, source, file_path, line_start, store, stats)
e52aae0… lmata 179
758c6b7… lmata 180 def _ingest_import_spec(
758c6b7… lmata 181 self, spec, source: bytes, file_path: str, line_start: int, store: GraphStore, stats: dict
758c6b7… lmata 182 ) -> None:
e52aae0… lmata 183 path_node = spec.child_by_field_name("path")
e52aae0… lmata 184 if not path_node:
e52aae0… lmata 185 return
e52aae0… lmata 186 module = _node_text(path_node, source).strip('"')
758c6b7… lmata 187 store.create_node(
758c6b7… lmata 188 NodeLabel.Import,
758c6b7… lmata 189 {
758c6b7… lmata 190 "name": module,
758c6b7… lmata 191 "file_path": file_path,
758c6b7… lmata 192 "line_start": line_start,
758c6b7… lmata 193 "module": module,
758c6b7… lmata 194 },
758c6b7… lmata 195 )
e52aae0… lmata 196 store.create_edge(
758c6b7… lmata 197 NodeLabel.File,
758c6b7… lmata 198 {"path": file_path},
e52aae0… lmata 199 EdgeType.IMPORTS,
758c6b7… lmata 200 NodeLabel.Import,
758c6b7… lmata 201 {"name": module, "file_path": file_path},
e52aae0… lmata 202 )
e52aae0… lmata 203 stats["edges"] += 1
e52aae0… lmata 204
758c6b7… lmata 205 def _extract_calls(
758c6b7… lmata 206 self,
758c6b7… lmata 207 fn_node,
758c6b7… lmata 208 source: bytes,
758c6b7… lmata 209 file_path: str,
758c6b7… lmata 210 fn_name: str,
758c6b7… lmata 211 fn_label: str,
758c6b7… lmata 212 store: GraphStore,
758c6b7… lmata 213 stats: dict,
758c6b7… lmata 214 ) -> None:
e52aae0… lmata 215 def walk(node):
e52aae0… lmata 216 if node.type == "call_expression":
e52aae0… lmata 217 func = node.child_by_field_name("function")
e52aae0… lmata 218 if func:
e52aae0… lmata 219 callee = _node_text(func, source).split(".")[-1]
e52aae0… lmata 220 store.create_edge(
758c6b7… lmata 221 fn_label,
758c6b7… lmata 222 {"name": fn_name, "file_path": file_path},
e52aae0… lmata 223 EdgeType.CALLS,
758c6b7… lmata 224 NodeLabel.Function,
758c6b7… lmata 225 {"name": callee, "file_path": file_path},
e52aae0… lmata 226 )
e52aae0… lmata 227 stats["edges"] += 1
e52aae0… lmata 228 for child in node.children:
e52aae0… lmata 229 walk(child)
e52aae0… lmata 230
e52aae0… lmata 231 body = fn_node.child_by_field_name("body")
e52aae0… lmata 232 if body:
e52aae0… lmata 233 walk(body)

Keyboard Shortcuts

Open search /
Next entry (timeline) j
Previous entry (timeline) k
Open focused entry Enter
Show this help ?
Toggle theme Top nav button