QingLong/AI/Neo4j/K2_Neo4jExecutor.py

# K2_Neo4jExecutor.py
from py2neo import Graph
import re
from Util import *
from Config import *

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

    # 新增文本执行方法
    def execute_cypher_text(self, cypher_text: str) -> dict:
        """直接执行Cypher文本"""
        stats = {'total': 0, 'success': 0, 'failed': 0}
        try:
            statements = re.split(r';\s*\n', cypher_text)
            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

    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 = K2_Neo4jExecutor(
        uri=NEO4J_URI,
        auth=NEO4J_AUTH
    )
    # 清库
    clear(executor.graph)

    # 确保调用名称与方法定义一致
    # f="knowledge_graph.cypher"
    f = "小学数学一至三年级.cypher"
    result = executor.execute_cypher_file(f)
    print(f"\n执行结果：")
    print(f"- 总语句数: {result.get('total', 0)}")
    print(f"- 成功: {result.get('success', 0)}")
    print(f"- 失败: {result.get('failed', 0)}")

    f = "小学数学四至六年级.cypher"
    result = executor.execute_cypher_file(f)
    print(f"\n执行结果：")
    print(f"- 总语句数: {result.get('total', 0)}")
    print(f"- 成功: {result.get('success', 0)}")
    print(f"- 失败: {result.get('failed', 0)}")

    # 执行优化
    f = "优化.cypher"
    executor.execute_cypher_file(f)