QingLong/AI/Neo4j/K2_Neo4jExecutor.py

# K2_Neo4jExecutor.py
from py2neo import Graph
import re


class Neo4jExecutor:
    def __init__(self, uri, auth):
        self.graph = Graph(uri, auth=auth)

    def execute_cypher_file(self, file_path: str) -> dict:  # 确保方法名称正确
        """执行Cypher文件"""
        stats = {'total': 0, 'success': 0, 'failed': 0}
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                cypher_script = f.read()
                statements = re.split(r';\s*\n', cypher_script)
                statements = [s.strip() for s in statements if s.strip()]

                stats['total'] = len(statements)

                for stmt in statements:
                    try:
                        self.graph.run(stmt)
                        stats['success'] += 1
                    except Exception as e:
                        stats['failed'] += 1
                        print(f"执行失败: {stmt[:50]}... \n错误: {str(e)[:100]}")

                return stats

        except Exception as e:
            print(f"文件错误: {str(e)}")
            return stats


if __name__ == '__main__':
    executor = Neo4jExecutor(
        uri="neo4j://10.10.21.20:7687",
        auth=("neo4j", "DsideaL4r5t6y7u")
    )

    # 确保调用名称与方法定义一致
    result = executor.execute_cypher_file("knowledge_graph.cypher")

    print(f"\n执行结果：")
    print(f"- 总语句数: {result.get('total', 0)}")
    print(f"- 成功: {result.get('success', 0)}")
    print(f"- 失败: {result.get('failed', 0)}")