Navegador

navegador / navegador / graph / export.py
Source Blame History 174 lines
57ba118… lmata 1 """
57ba118… lmata 2 Text-based graph export and import for navegador.
57ba118… lmata 3
57ba118… lmata 4 Exports the full graph to a deterministic JSON Lines format (.jsonl)
57ba118… lmata 5 suitable for committing to version control. Each line is a self-contained
57ba118… lmata 6 JSON object — either a node or an edge.
57ba118… lmata 7
57ba118… lmata 8 Format:
57ba118… lmata 9 {"kind": "node", "label": "Function", "props": {"name": "foo", ...}}
57ba118… lmata 10 {"kind": "edge", "type": "CALLS", "from": {...}, "to": {...}}
57ba118… lmata 11 """
57ba118… lmata 12
57ba118… lmata 13 import json
57ba118… lmata 14 import logging
57ba118… lmata 15 from pathlib import Path
57ba118… lmata 16
57ba118… lmata 17 from navegador.graph.store import GraphStore
57ba118… lmata 18
57ba118… lmata 19 logger = logging.getLogger(__name__)
57ba118… lmata 20
57ba118… lmata 21
57ba118… lmata 22 def export_graph(store: GraphStore, output_path: str | Path) -> dict[str, int]:
57ba118… lmata 23 """
57ba118… lmata 24 Export the full graph to a JSONL file.
57ba118… lmata 25
57ba118… lmata 26 Returns:
57ba118… lmata 27 Dict with counts: nodes, edges.
57ba118… lmata 28 """
57ba118… lmata 29 output_path = Path(output_path)
57ba118… lmata 30 output_path.parent.mkdir(parents=True, exist_ok=True)
57ba118… lmata 31
57ba118… lmata 32 nodes = _export_nodes(store)
57ba118… lmata 33 edges = _export_edges(store)
57ba118… lmata 34
57ba118… lmata 35 # Sort for deterministic output
57ba118… lmata 36 nodes.sort(key=lambda n: (n["label"], json.dumps(n["props"], sort_keys=True)))
e5e1a3b… lmata 37 edges.sort(
e5e1a3b… lmata 38 key=lambda e: (
e5e1a3b… lmata 39 e["type"],
e5e1a3b… lmata 40 json.dumps(e["from"], sort_keys=True),
e5e1a3b… lmata 41 json.dumps(e["to"], sort_keys=True),
e5e1a3b… lmata 42 )
e5e1a3b… lmata 43 )
57ba118… lmata 44
57ba118… lmata 45 with output_path.open("w", encoding="utf-8") as f:
57ba118… lmata 46 for node in nodes:
57ba118… lmata 47 f.write(json.dumps(node, sort_keys=True) + "\n")
57ba118… lmata 48 for edge in edges:
57ba118… lmata 49 f.write(json.dumps(edge, sort_keys=True) + "\n")
57ba118… lmata 50
57ba118… lmata 51 logger.info("Exported %d nodes, %d edges to %s", len(nodes), len(edges), output_path)
57ba118… lmata 52 return {"nodes": len(nodes), "edges": len(edges)}
57ba118… lmata 53
57ba118… lmata 54
57ba118… lmata 55 def import_graph(store: GraphStore, input_path: str | Path, clear: bool = True) -> dict[str, int]:
57ba118… lmata 56 """
57ba118… lmata 57 Import a graph from a JSONL file.
57ba118… lmata 58
57ba118… lmata 59 Args:
57ba118… lmata 60 store: Target GraphStore.
57ba118… lmata 61 input_path: Path to the JSONL file.
57ba118… lmata 62 clear: If True (default), wipe the graph before importing.
57ba118… lmata 63
57ba118… lmata 64 Returns:
57ba118… lmata 65 Dict with counts: nodes, edges.
57ba118… lmata 66 """
57ba118… lmata 67 input_path = Path(input_path)
57ba118… lmata 68 if not input_path.exists():
57ba118… lmata 69 raise FileNotFoundError(f"Export file not found: {input_path}")
57ba118… lmata 70
57ba118… lmata 71 if clear:
57ba118… lmata 72 store.clear()
57ba118… lmata 73
57ba118… lmata 74 node_count = 0
57ba118… lmata 75 edge_count = 0
57ba118… lmata 76
57ba118… lmata 77 with input_path.open("r", encoding="utf-8") as f:
57ba118… lmata 78 for line in f:
57ba118… lmata 79 line = line.strip()
57ba118… lmata 80 if not line:
57ba118… lmata 81 continue
57ba118… lmata 82 record = json.loads(line)
57ba118… lmata 83
57ba118… lmata 84 if record["kind"] == "node":
57ba118… lmata 85 _import_node(store, record)
57ba118… lmata 86 node_count += 1
57ba118… lmata 87 elif record["kind"] == "edge":
57ba118… lmata 88 _import_edge(store, record)
57ba118… lmata 89 edge_count += 1
57ba118… lmata 90
57ba118… lmata 91 logger.info("Imported %d nodes, %d edges from %s", node_count, edge_count, input_path)
57ba118… lmata 92 return {"nodes": node_count, "edges": edge_count}
57ba118… lmata 93
57ba118… lmata 94
57ba118… lmata 95 def _export_nodes(store: GraphStore) -> list[dict]:
57ba118… lmata 96 """Export all nodes with their labels and properties."""
e5e1a3b… lmata 97 result = store.query("MATCH (n) RETURN labels(n)[0] AS label, properties(n) AS props")
57ba118… lmata 98 nodes = []
57ba118… lmata 99 for row in result.result_set or []:
57ba118… lmata 100 label = row[0]
57ba118… lmata 101 props = row[1] if isinstance(row[1], dict) else {}
57ba118… lmata 102 nodes.append({"kind": "node", "label": label, "props": props})
57ba118… lmata 103 return nodes
57ba118… lmata 104
57ba118… lmata 105
57ba118… lmata 106 def _export_edges(store: GraphStore) -> list[dict]:
57ba118… lmata 107 """Export all edges with type and endpoint identifiers."""
57ba118… lmata 108 result = store.query(
57ba118… lmata 109 "MATCH (a)-[r]->(b) "
57ba118… lmata 110 "RETURN type(r) AS type, labels(a)[0] AS from_label, "
57ba118… lmata 111 "a.name AS from_name, coalesce(a.file_path, a.path, '') AS from_path, "
57ba118… lmata 112 "labels(b)[0] AS to_label, b.name AS to_name, "
57ba118… lmata 113 "coalesce(b.file_path, b.path, '') AS to_path"
57ba118… lmata 114 )
57ba118… lmata 115 edges = []
57ba118… lmata 116 for row in result.result_set or []:
e5e1a3b… lmata 117 edges.append(
e5e1a3b… lmata 118 {
e5e1a3b… lmata 119 "kind": "edge",
e5e1a3b… lmata 120 "type": row[0],
e5e1a3b… lmata 121 "from": {"label": row[1], "name": row[2], "path": row[3]},
e5e1a3b… lmata 122 "to": {"label": row[4], "name": row[5], "path": row[6]},
e5e1a3b… lmata 123 }
e5e1a3b… lmata 124 )
57ba118… lmata 125 return edges
57ba118… lmata 126
57ba118… lmata 127
57ba118… lmata 128 def _import_node(store: GraphStore, record: dict) -> None:
57ba118… lmata 129 """Create a node from an export record."""
57ba118… lmata 130 label = record["label"]
57ba118… lmata 131 props = record["props"]
57ba118… lmata 132 # Ensure required merge keys exist
57ba118… lmata 133 if "name" not in props:
57ba118… lmata 134 props["name"] = ""
57ba118… lmata 135 if "file_path" not in props and "path" not in props:
57ba118… lmata 136 props["file_path"] = ""
57ba118… lmata 137
57ba118… lmata 138 prop_str = ", ".join(f"n.{k} = ${k}" for k in props)
57ba118… lmata 139 # Use name + file_path or path for merge key
57ba118… lmata 140 if "file_path" in props:
57ba118… lmata 141 cypher = f"MERGE (n:{label} {{name: $name, file_path: $file_path}}) SET {prop_str}"
57ba118… lmata 142 else:
57ba118… lmata 143 cypher = f"MERGE (n:{label} {{name: $name, path: $path}}) SET {prop_str}"
57ba118… lmata 144 store.query(cypher, props)
57ba118… lmata 145
57ba118… lmata 146
57ba118… lmata 147 def _import_edge(store: GraphStore, record: dict) -> None:
57ba118… lmata 148 """Create an edge from an export record."""
57ba118… lmata 149 edge_type = record["type"]
57ba118… lmata 150 from_info = record["from"]
57ba118… lmata 151 to_info = record["to"]
57ba118… lmata 152
e5e1a3b… lmata 153 from_key = "name: $from_name"
e5e1a3b… lmata 154 to_key = "name: $to_name"
57ba118… lmata 155
57ba118… lmata 156 params = {
57ba118… lmata 157 "from_name": from_info["name"],
57ba118… lmata 158 "to_name": to_info["name"],
57ba118… lmata 159 }
57ba118… lmata 160
57ba118… lmata 161 if from_info.get("path"):
57ba118… lmata 162 from_key += ", file_path: $from_path"
57ba118… lmata 163 params["from_path"] = from_info["path"]
57ba118… lmata 164
57ba118… lmata 165 if to_info.get("path"):
57ba118… lmata 166 to_key += ", path: $to_path"
57ba118… lmata 167 params["to_path"] = to_info["path"]
57ba118… lmata 168
57ba118… lmata 169 cypher = (
57ba118… lmata 170 f"MATCH (a:{from_info['label']} {{{from_key}}}), "
57ba118… lmata 171 f"(b:{to_info['label']} {{{to_key}}}) "
57ba118… lmata 172 f"MERGE (a)-[r:{edge_type}]->(b)"
57ba118… lmata 173 )
57ba118… lmata 174 store.query(cypher, params)

Keyboard Shortcuts

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