使用 Python 和 Py2Neo 构建 Neo4j 管理脚本

Neo4j 是一个强大的图数据库，适合处理复杂的关系型数据。借助 Python 的 py2neo 库，我们可以快速实现对 Neo4j 数据库的管理和操作。本文介绍一个功能丰富的 Python 脚本，帮助用户轻松管理 Neo4j 数据库，包含启动/停止服务、清空数据、统计分析、图谱可视化等功能。

---

1. 脚本功能一览

用户通过数字选择对应的功能：

1. 启动 Neo4j：通过命令行启动 Neo4j 服务。
2. 停止 Neo4j：停止运行中的 Neo4j 服务。
3. 清空 Neo4j：删除数据库中所有数据。
4. 图谱可视化：展示知识图谱的结构，并生成图片。
5. 统计分析：统计节点和关系的数量。
6. 导出图谱：将图谱导出为 JSON 文件，支持选择保存路径。
7. 数据健康检查：检查图谱中的孤立节点和重复节点。
8. 退出程序。

---

2. 脚本核心实现

2.1 脚本入口

以下是功能菜单的实现，每个功能都有对应的函数，用户输入数字即可调用：

def main():while True:print("\n请选择一个功能：")print("1. 启动 Neo4j")print("2. 停止 Neo4j")print("3. 清空 Neo4j")print("4. 图谱可视化")print("5. 统计分析")print("6. 导出图谱")print("7. 数据健康检查")print("8. 退出")choice = input("输入功能编号：")if choice == "1":start_neo4j()elif choice == "2":stop_neo4j()elif choice == "3":clear_database()elif choice == "4":visualize_graph()elif choice == "5":statistical_analysis()elif choice == "6":export_graph()elif choice == "7":health_check()elif choice == "8":print("退出程序。")breakelse:print("无效输入，请重试！")

2.2 功能实现

启动/停止 Neo4j 服务

借助系统命令行操作，我们可以在另一个终端中启动或停止 Neo4j：

def start_neo4j():"""启动 Neo4j 服务"""os.system("neo4j console &")print("Neo4j 服务已启动...")def stop_neo4j():"""停止 Neo4j 服务"""os.system("neo4j stop")print("Neo4j 服务已停止...")

清空 Neo4j 数据

利用 Cypher 查询可以快速清空图数据库中的所有节点和关系：

def clear_database():"""清空当前知识图谱中的所有数据"""graph.run("MATCH (n) DETACH DELETE n")print("图谱已清空！")

图谱可视化

使用 py2neo 查询数据，结合 networkx 和 matplotlib，我们可以绘制出知识图谱的可视化图像：

def visualize_graph():"""图谱可视化：生成图片展示"""query = "MATCH (n)-[r]->(m) RETURN n.name AS node1, type(r) AS relationship, m.name AS node2 LIMIT 50"results = graph.run(query).data()if not results:print("图谱中没有可视化的数据。")returnG = nx.DiGraph()  # 使用有向图for record in results:node1 = record["node1"] or "Unnamed Node"node2 = record["node2"] or "Unnamed Node"relationship = record["relationship"] or "RELATED_TO"G.add_edge(node1, node2, relationship=relationship)plt.figure(figsize=(12, 8))pos = nx.spring_layout(G)  # 布局nx.draw(G, pos, with_labels=True, node_color="skyblue", node_size=2000, font_size=12, font_weight="bold", arrowsize=15)edge_labels = nx.get_edge_attributes(G, "relationship")nx.draw_networkx_edge_labels(G, pos, edge_labels=edge_labels, font_size=10)plt.title("Neo4j Graph Visualization")plt.show()

数据统计分析

通过 Cypher 查询统计图谱中的节点和关系数量：

def statistical_analysis():"""统计分析：节点和关系数量"""node_count = graph.run("MATCH (n) RETURN COUNT(n) AS count").data()[0]["count"]relationship_count = graph.run("MATCH ()-[r]->() RETURN COUNT(r) AS count").data()[0]["count"]print(f"节点数量：{node_count}")print(f"关系数量：{relationship_count}")

导出图谱为 JSON 文件

允许用户选择保存路径，将查询结果保存为 JSON 格式：

def export_graph():"""导出图谱为 JSON 文件"""query = "MATCH (n)-[r]->(m) RETURN n, r, m"results = graph.run(query).data()data = [{"node1": dict(record["n"]), "relationship": dict(record["r"]), "node2": dict(record["m"])} for record in results]root = Tk()root.withdraw()  # 隐藏主窗口save_path = filedialog.asksaveasfilename(defaultextension=".json", filetypes=[("JSON files", "*.json")])if save_path:with open(save_path, "w", encoding="utf-8") as f:json.dump(data, f, indent=4)print(f"图谱已导出至：{save_path}")

数据健康检查

检查图谱中的孤立节点（没有关系）和重复节点：

def health_check():"""数据健康检查"""orphan_nodes = graph.run("MATCH (n) WHERE NOT (n)--() RETURN COUNT(n) AS count").data()[0]["count"]duplicate_nodes = graph.run("MATCH (n) WITH n.name AS name, COUNT(*) AS count WHERE count > 1 RETURN COUNT(name) AS count").data()[0]["count"]print(f"孤立节点数量：{orphan_nodes}")print(f"重复节点数量：{duplicate_nodes}")

---

3. 图谱可视化示例

执行“图谱可视化”功能后，脚本会绘制图谱的结构图

---

4. 使用前的准备

1. 环境依赖安装：
   确保已安装以下库：

   pip install py2neo networkx matplotlib
   

2. Neo4j 连接配置：
   修改脚本中连接数据库的代码：

   graph = Graph("bolt://localhost:7687", auth=("neo4j", "password"))
   

3. 运行脚本：
   运行脚本文件：

   python neo4j_manager.py
   

---