241 lines
9.8 KiB
Python
241 lines
9.8 KiB
Python
"""
|
||
Cypher 错误修复模块
|
||
|
||
功能：修复常见的 Cypher 语法错误和格式问题
|
||
"""
|
||
|
||
import re
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def parse_cypher_error(error_message: str, cypher: str) -> dict:
|
||
"""
|
||
解析Cypher错误信息，提取错误类型、位置等详细信息
|
||
"""
|
||
error_info = {
|
||
"error_type": "unknown",
|
||
"error_message": error_message,
|
||
"error_position": None,
|
||
"suggested_fix": None
|
||
}
|
||
|
||
error_lower = error_message.lower()
|
||
|
||
# ========== 识别错误类型 ==========
|
||
if "unexpected" in error_lower or "expected" in error_lower:
|
||
error_info["error_type"] = "syntax_unexpected"
|
||
elif "invalid" in error_lower:
|
||
error_info["error_type"] = "syntax_invalid"
|
||
elif "missing" in error_lower:
|
||
error_info["error_type"] = "syntax_missing"
|
||
elif "cannot" in error_lower or "can't" in error_lower:
|
||
error_info["error_type"] = "semantic_error"
|
||
elif "unknown" in error_lower or "not found" in error_lower:
|
||
error_info["error_type"] = "schema_error"
|
||
elif "syntax" in error_lower:
|
||
error_info["error_type"] = "syntax_error"
|
||
|
||
# ========== 提取错误位置（如果错误信息中包含位置信息） ==========
|
||
# Neo4j错误信息可能包含位置，如 "line 1, column 10"
|
||
position_match = re.search(r'line\s+(\d+)[,\s]+column\s+(\d+)', error_message, re.IGNORECASE)
|
||
if position_match:
|
||
line_num = int(position_match.group(1))
|
||
col_num = int(position_match.group(2))
|
||
error_info["error_position"] = {"line": line_num, "column": col_num}
|
||
|
||
# 提取错误位置的上下文
|
||
lines = cypher.split('\n')
|
||
if line_num <= len(lines):
|
||
error_line = lines[line_num - 1]
|
||
error_info["error_context"] = error_line
|
||
if col_num <= len(error_line):
|
||
# 标记错误位置
|
||
marker = ' ' * (col_num - 1) + '^'
|
||
error_info["error_marker"] = marker
|
||
|
||
# ========== 提取错误相关的关键词 ==========
|
||
# 提取错误信息中提到的标签、关系、关键字等
|
||
keywords = []
|
||
# 匹配可能的标签名（大写字母开头的单词）
|
||
label_matches = re.findall(r'\b([A-Z][a-zA-Z0-9_]+)\b', error_message)
|
||
keywords.extend(label_matches)
|
||
# 匹配引号中的内容
|
||
quoted_matches = re.findall(r'["\']([^"\']+)["\']', error_message)
|
||
keywords.extend(quoted_matches)
|
||
error_info["keywords"] = list(set(keywords))
|
||
|
||
return error_info
|
||
|
||
|
||
def force_undirected_relationships(cypher: str) -> str:
|
||
"""
|
||
强制将所有有方向的关系转换为无方向关系
|
||
"""
|
||
if not cypher:
|
||
return cypher
|
||
|
||
# 替换所有有方向的关系为无方向关系
|
||
# 匹配模式：-[:RELATIONSHIP]-> 或 <-[:RELATIONSHIP]-
|
||
# 替换为：-[:RELATIONSHIP]-
|
||
|
||
# 替换 -> 为 -（在关系类型之后）
|
||
# 需要小心处理，只替换关系箭头，不替换其他地方的 ->
|
||
# 使用正则表达式匹配关系模式
|
||
fixed = cypher
|
||
|
||
# 替换 -[:RELATIONSHIP]-> 为 -[:RELATIONSHIP]-
|
||
fixed = re.sub(r'(\[:[^\]]+\])->', r'\1-', fixed)
|
||
|
||
# 替换 <-[:RELATIONSHIP]- 为 -[:RELATIONSHIP]-
|
||
fixed = re.sub(r'<-(\[:[^\]]+\])-', r'-\1-', fixed)
|
||
|
||
# 替换 -> 在关系模式中（处理可变长度关系，如 [*1..3]->）
|
||
fixed = re.sub(r'(\[[^\]]+\])->', r'\1-', fixed)
|
||
|
||
# 替换 <- 在关系模式中（处理可变长度关系，如 <-[*1..3]-）
|
||
fixed = re.sub(r'<-(\[[^\]]+\])-', r'-\1-', fixed)
|
||
|
||
# 如果进行了替换，记录日志
|
||
if fixed != cypher:
|
||
logger.info(f"[关系方向转换] 检测到有方向关系，已自动转换为无方向关系")
|
||
logger.debug(f"[关系方向转换] 原始: {cypher[:200]}...")
|
||
logger.debug(f"[关系方向转换] 转换后: {fixed[:200]}...")
|
||
|
||
return fixed
|
||
|
||
|
||
def fix_cypher_based_on_error(cypher: str, error_info: dict) -> str:
|
||
"""
|
||
基于实际错误信息智能修复Cypher
|
||
"""
|
||
if not error_info or not error_info.get("error_type"):
|
||
return None
|
||
|
||
error_type = error_info["error_type"]
|
||
error_message = error_info.get("error_message", "")
|
||
fixed = cypher
|
||
|
||
# ========== 根据错误类型进行修复 ==========
|
||
|
||
# 1. 语法错误：多余的括号
|
||
if "unexpected" in error_message.lower() and ("(" in error_message or ")" in error_message):
|
||
# 尝试修复多余的括号
|
||
# 匹配 `-[:REL]- ( (n:Label)` 这种模式
|
||
fixed = re.sub(r'(\[:[^\]]+\])-\s+\(\s*\(', r'\1-(', fixed)
|
||
fixed = re.sub(r'\)\s*-\s*(\[:[^\]]+\])-\s+\(\s*\(', r')-\1-(', fixed)
|
||
if fixed != cypher:
|
||
logger.info(f"[智能修复] 修复多余括号错误")
|
||
return fixed
|
||
|
||
# 2. 语法错误：缺少括号
|
||
if "missing" in error_message.lower() and ("(" in error_message or ")" in error_message):
|
||
# 检查括号是否匹配
|
||
open_count = fixed.count('(')
|
||
close_count = fixed.count(')')
|
||
if open_count > close_count:
|
||
# 缺少右括号，在末尾添加
|
||
fixed = fixed + ')' * (open_count - close_count)
|
||
logger.info(f"[智能修复] 添加缺失的右括号: {open_count - close_count} 个")
|
||
return fixed
|
||
elif close_count > open_count:
|
||
# 多余的右括号，但这种情况较难自动修复，返回None
|
||
logger.warning(f"[智能修复] 检测到多余的右括号，无法自动修复")
|
||
return None
|
||
|
||
# 3. 语法错误：关系定义问题
|
||
if "relationship" in error_message.lower() or "relationship type" in error_message.lower():
|
||
# 检查关系定义格式
|
||
# 修复 `-[:REL]- (n:Label)` -> `-[:REL]-(n:Label)`
|
||
fixed = re.sub(r'(\[:[^\]]+\])-\s+\(', r'\1-(', fixed)
|
||
if fixed != cypher:
|
||
logger.info(f"[智能修复] 修复关系定义格式")
|
||
return fixed
|
||
|
||
# 4. Schema错误：节点标签或关系类型不存在
|
||
if error_info["error_type"] == "schema_error":
|
||
keywords = error_info.get("keywords", [])
|
||
# 这里可以尝试其他修复策略
|
||
logger.debug(f"[智能修复] Schema错误，关键词: {keywords}")
|
||
# Schema错误通常需要LLM修正，这里不自动修复
|
||
return None
|
||
|
||
# 5. 语法错误：位置相关的修复
|
||
if error_info.get("error_position"):
|
||
pos = error_info["error_position"]
|
||
line_num = pos.get("line", 1)
|
||
col_num = pos.get("column", 1)
|
||
|
||
lines = fixed.split('\n')
|
||
if line_num <= len(lines):
|
||
error_line = lines[line_num - 1]
|
||
# 检查错误位置附近的字符
|
||
if col_num <= len(error_line):
|
||
char_at_error = error_line[col_num - 1] if col_num > 0 else ''
|
||
context = error_line[max(0, col_num-10):min(len(error_line), col_num+10)]
|
||
|
||
# 如果是多余的字符，尝试删除
|
||
if char_at_error in ['(', ')', ' ', '-']:
|
||
# 检查是否是明显的多余字符
|
||
before = error_line[:col_num-1] if col_num > 0 else ''
|
||
after = error_line[col_num:] if col_num < len(error_line) else ''
|
||
|
||
# 如果前后都是关系连接符，中间的字符可能是多余的
|
||
if re.search(r'[-\s]*$', before) and re.search(r'^[-\s]*', after):
|
||
# 尝试删除错误位置的字符
|
||
new_line = error_line[:col_num-1] + error_line[col_num:]
|
||
lines[line_num - 1] = new_line
|
||
fixed = '\n'.join(lines)
|
||
logger.info(f"[智能修复] 删除错误位置的字符: '{char_at_error}'")
|
||
return fixed
|
||
|
||
# 6. 通用修复：尝试修复常见的格式问题
|
||
# 修复关系后的多余空格
|
||
fixed = re.sub(r'(\[:[^\]]+\])-\s+\(', r'\1-(', fixed)
|
||
# 修复节点定义前的多余括号
|
||
fixed = re.sub(r'\(\s+\(([a-zA-Z_][a-zA-Z0-9_]*\s*:)', r'(\1', fixed)
|
||
|
||
if fixed != cypher:
|
||
logger.info(f"[智能修复] 应用通用修复")
|
||
return fixed
|
||
|
||
# 无法自动修复
|
||
return None
|
||
|
||
|
||
def fix_common_cypher_errors(cypher: str) -> str:
|
||
"""
|
||
修复常见的Cypher语法错误
|
||
"""
|
||
if not cypher:
|
||
return cypher
|
||
|
||
fixed = cypher
|
||
|
||
# ========== 修复1：关系后多余的空格和括号 ==========
|
||
# 匹配模式：`-[:REL]- ( (n:Label)` -> `-[:REL]-(n:Label)`
|
||
# 或者 `-[:REL]- (n:Label)` -> `-[:REL]-(n:Label)`
|
||
fixed = re.sub(r'(\[:[^\]]+\])-\s+\(\s*\(', r'\1-(', fixed) # `-[:REL]- ( (` -> `-[:REL]-(`
|
||
fixed = re.sub(r'(\[:[^\]]+\])-\s+\(([a-zA-Z_][a-zA-Z0-9_]*\s*:)', r'\1-(\2', fixed) # `-[:REL]- (n:` -> `-[:REL]-(n:`
|
||
|
||
# ========== 修复2：节点定义前多余的括号 ==========
|
||
# 匹配模式：`( (n:Label)` -> `(n:Label)`
|
||
fixed = re.sub(r'\(\s+\(([a-zA-Z_][a-zA-Z0-9_]*\s*:)', r'(\1', fixed)
|
||
|
||
# ========== 修复3：MATCH语句中的多余空格和括号 ==========
|
||
# 匹配模式：`MATCH ...)-[:REL]- ( (n:Label)` -> `MATCH ...)-[:REL]-(n:Label)`
|
||
fixed = re.sub(r'\)\s*-\s*(\[:[^\]]+\])-\s+\(\s*\(', r')-\1-(', fixed)
|
||
fixed = re.sub(r'\)\s*-\s*(\[:[^\]]+\])-\s+\(([a-zA-Z_][a-zA-Z0-9_]*\s*:)', r')-\1-(\2', fixed)
|
||
|
||
# ========== 修复4：规范化关系定义后的空格 ==========
|
||
# 确保关系定义后直接跟节点定义，如 `-[:REL]-(n:Label)` 而不是 `-[:REL]- (n:Label)`
|
||
fixed = re.sub(r'(\[:[^\]]+\])-\s+\(', r'\1-(', fixed)
|
||
|
||
if fixed != cypher:
|
||
logger.info(f"[Cypher修复] 检测到并修复了常见语法错误")
|
||
logger.debug(f"[Cypher修复] 原始: {cypher[:200]}...")
|
||
logger.debug(f"[Cypher修复] 修复后: {fixed[:200]}...")
|
||
|
||
return fixed
|