from typing import Dict, List, Any, Optional def convert_node_to_g6_v5(neo4j_node: Dict[str, Any]) -> Dict[str, Any]: """ 将 Neo4j 节点(含 id 和 props)转换为 G6 v5 节点格式。 :param neo4j_node: 来自 Neo4jUtil 的节点字典,必须包含 'id' 字段 :return: G6 v5 节点对象 """ node_id = neo4j_node.get("id") if node_id is None: raise ValueError("节点必须包含 'id' 字段") # 构建 data:排除 'id',保留其他属性 data = {k: v for k, v in neo4j_node.items() if k != "id"} # 确保有显示用的 name 或 label(G6 默认使用 data.name) if "name" not in data and "label" not in data: data["name"] = str(node_id) g6_node = { "id": node_id, "data": data, # 可选字段(前端可覆盖): # "type": "circle", # "style": {"size": 32, "fill": "violet"}, "states": [], "combo": None } return g6_node def build_g6_graph_data( neo4j_util: 'Neo4jUtil', node_label: str, rel_type: Optional[str] = None ) -> Dict[str, List[Dict[str, Any]]]: """ 构建 G6 v5 兼容的图数据结构(nodes + edges)。 :param neo4j_util: 已连接的 Neo4jUtil 实例 :param node_label: 中心节点标签(用于获取初始节点集) :param rel_type: 关系类型(可选,若为 None 则获取所有关系) :return: {'nodes': [...], 'edges': [...]} """ # 1. 获取中心节点(指定标签的所有节点) center_nodes = neo4j_util.find_all_nodes(node_label) # 2. 获取所有相关关系(可按类型过滤) relationships = neo4j_util.find_all_relationships(rel_type=rel_type) # 3. 使用字典去重节点(key: elementId) g6_node_map: Dict[str, Dict[str, Any]] = {} # 添加中心节点 for node in center_nodes: node_id = node.get("id") if node_id: g6_node_map[node_id] = convert_node_to_g6_v5(node) # 4. 处理关系,提取 source/target 节点并构建边 g6_edges: List[Dict[str, Any]] = [] for rel in relationships: source_node = rel.get("source") target_node = rel.get("target") if not source_node or not target_node: continue source_id = source_node.get("id") target_id = target_node.get("id") if not source_id or not target_id: continue # 自动加入 source 和 target 节点(去重) g6_node_map[source_id] = convert_node_to_g6_v5(source_node) g6_node_map[target_id] = convert_node_to_g6_v5(target_node) # 构建 G6 边对象 edge_data = {} # 关系类型 rel_type_str = rel.get("type") if rel_type_str: edge_data["relationship"] = rel_type_str # 合并关系属性 rel_props = rel.get("relProps") or {} edge_data.update(rel_props) g6_edge = { "source": source_id, "target": target_id, "type": "line", # G6 默认边类型 "data": edge_data, # "style": {}, # 可由前端统一配置 "states": [] } g6_edges.append(g6_edge) # 5. 组装结果 result = { "nodes": list(g6_node_map.values()), "edges": g6_edges } return result