QingLong/AI/Neo4j/Neo4jExecutor.py

import re
from Util import *
from Config import *


class 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"执行失败: {stmt[:100]}... \n完整错误: {str(e)}")  # 原为str(e)[:100]
            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


def init():
    executor = Neo4jExecutor(
        uri=NEO4J_URI,
        auth=NEO4J_AUTH
    )
    # 清库
    clear(executor.graph)