使用 Python 和 Py2Neo 构建 Neo4j 管理脚本
Neo4j 是一个强大的图数据库，适合处理复杂的关系型数据。借助 Python 的 py2neo 库，我们可以快速实现对 Neo4j 数据库的管理和操作。本文介绍一个功能丰富的 Python 脚本，帮助用户轻松管理 Neo4j 数据库，包含启动/停止服务、清空数据、统计分析、图谱可视化等功能。
1. 脚本功能一览
用户通过数字选择对应的功能：
- 启动 Neo4j：通过命令行启动 Neo4j 服务。
- 停止 Neo4j：停止运行中的 Neo4j 服务。
- 清空 Neo4j：删除数据库中所有数据。
- 图谱可视化：展示知识图谱的结构，并生成图片。
- 统计分析：统计节点和关系的数量。
- 导出图谱：将图谱导出为 JSON 文件，支持选择保存路径。
- 数据健康检查：检查图谱中的孤立节点和重复节点。
- 退出程序。
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("退出程序。")
break
else:
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("图谱中没有可视化的数据。")
return
G = 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. 使用前的准备
-
环境依赖安装：
确保已安装以下库：pip install py2neo networkx matplotlib -
Neo4j 连接配置：
修改脚本中连接数据库的代码：graph = Graph("bolt://localhost:7687", auth=("neo4j", "password")) -
运行脚本：
运行脚本文件：python neo4j_manager.py