"""Cypher 合法性检查节点
输入: state['cypher_query']: List[str] 上游生成的 cypher 语句
输出: state['is_all_validate_cypher']: bool 是否全部合法
为什么生成完还要再查一遍：`generate_neo4j_cypher_node` 里的过滤只看"是不是只读"，
拦不住"语法对、但标签/属性图里没有"这类语句 —— 它们能正常执行，只是恒返回空行，
下游会当成"图里没这个知识"，比报错更难查。所以这里做三层检查：
1. 静态只读：复用生成节点的 FORBIDDEN_KEYWORDS，同一份白名单，避免两处漂移
2. 对 schema 核账：标签、关系类型、`变量.属性` 都要在 tcm_metadata.json 里存在
3. Neo4j EXPLAIN：只编译不执行，把语法错误在打库之前拦掉
第 3 层要连库，连不上时只跳过这一层（前两层的结论照样有效），不让整个节点挂掉。
"""
import json
import re
from typing import Dict, List, Set, Tuple
from _004_langgraph_more_nodes.agent_state import AgentState
from _004_langgraph_more_nodes.nodes.generate_neo4j_cypher_node import (
FORBIDDEN_KEYWORDS,
)
from common.config import Config
conf = Config()
# 空的 cypher_query 判 False。空集合"全部合法"在逻辑上成立，但这个标志位是给路由用的：
# 没有语句可执行时，必须走兜底分支而不是查库分支，判 True 会让下游拿着空列表白跑一趟。
EMPTY_IS_VALID = False
# ======================
# schema：从 tcm_metadata.json 读出标签 / 关系 / 属性
# ======================
def _load_schema() -> Tuple[Set[str], Set[str], Set[str]]:
"""返回 (标签集合, 关系类型集合, 所有属性名集合)。
属性不按标签分开：cypher 里 `n.name` 的 `n` 要回溯到它的标签才能精确校验，
而 WITH / OPTIONAL MATCH 之后变量的标签可能已经丢了。这里只判"这个属性名在图里
存在吗"，宁可漏判也不误杀。
"""
meta = json.loads(conf.TCM_METADATA)
labels = {label["name"] for label in meta.get("labels", [])}
rel_types = {rel["type"] for rel in meta.get("relationships", [])}
props = set()
for label in meta.get("labels", []):
props.update(prop["name"] for prop in label.get("properties", []))
for rel in meta.get("relationships", []):
props.update(prop["name"] for prop in rel.get("properties", []))
return labels, rel_types, props
LABELS, REL_TYPES, PROP_KEYS = _load_schema()
# `(n:Herb)`、`(f:Formula:Prescription)` 里的标签。冒号后面允许空格，多标签用 | 或 :
LABEL_PATTERN = re.compile(r"\(\s*\w*\s*:\s*([A-Za-z_][\w|:\s]*?)\s*[){\s]")
# `-[:HAS_EFFECT]->`、`-[r:HAS_EFFECT|HAS_INGREDIENT*1..2]-`
REL_PATTERN = re.compile(r"\[\s*\w*\s*:\s*([A-Za-z_][\w|\s]*)")
# `n.name`、`h.property_flavor`。前面不能是字母/数字，排开 `count(n).x` 这种不存在的写法
PROP_PATTERN = re.compile(r"\b[A-Za-z_]\w*\.([A-Za-z_]\w+)")
# cypher 内建函数和关键字后面也可能跟点号或被当成标签，先排掉常见的
IGNORED_TOKENS = {"DISTINCT", "AS", "AND", "OR", "NOT", "IN", "NULL", "TRUE", "FALSE"}
STRING_LITERAL = re.compile(r"'[^']*'|\"[^\"]*\"")
def _strip_literals(cypher: str) -> str:
"""把字符串字面量换成空串再做正则匹配。
实体名是数据不是代码，`WHERE s.name = '桂枝汤.加减'` 里的 `.加减` 不该被当成属性，
`'Herb'` 这种字面量也不该被当成标签。
"""
return STRING_LITERAL.sub("''", cypher)
def _split_names(raw: str) -> List[str]:
"""`Formula:Prescription`、`HAS_EFFECT|HAS_INGREDIENT*1..2` 拆成名字列表"""
names = re.split(r"[|:]", raw)
# 关系里可能带 `*1..2` 变长路径，星号后面的部分不是名字
return [name.split("*")[0].strip() for name in names if name.split("*")[0].strip()]
# ======================
# 第 1 层：静态只读
# ======================
def _check_readonly(cypher: str) -> List[str]:
"""返回问题列表，空列表表示这一层通过"""
upper = _strip_literals(cypher).upper()
problems = [
f"包含写操作关键字 {keyword}"
for keyword in FORBIDDEN_KEYWORDS
if re.search(rf"\b{re.escape(keyword)}\b", upper)
]
if "RETURN" not in upper:
problems.append("没有 RETURN，查不出任何结果")
if not upper.lstrip().startswith(("MATCH", "OPTIONAL MATCH", "WITH")):
problems.append("不是以 MATCH / OPTIONAL MATCH / WITH 开头")
return problems
# ======================
# 第 2 层：对 schema 核账
# ======================
def _check_schema(cypher: str) -> List[str]:
"""标签、关系类型、属性名都要在 tcm_metadata.json 里存在"""
text = _strip_literals(cypher)
problems = []
for raw in LABEL_PATTERN.findall(text):
for label in _split_names(raw):
if label.upper() in IGNORED_TOKENS:
continue
if label not in LABELS:
problems.append(f"标签不存在: {label}")
for raw in REL_PATTERN.findall(text):
for rel in _split_names(raw):
if rel not in REL_TYPES:
problems.append(f"关系类型不存在: {rel}")
for prop in PROP_PATTERN.findall(text):
if prop not in PROP_KEYS:
problems.append(f"属性不存在: .{prop}")
# 同一个错误可能在一条语句里出现多次，去重保序
return list(dict.fromkeys(problems))
# ======================
# 第 3 层：Neo4j EXPLAIN
# ======================
_explain_available = None
def _check_explain(cypher: str) -> List[str]:
"""用 EXPLAIN 让 Neo4j 自己编译一遍，只查语法不执行。
连不上库时把 _explain_available 置 False，后续语句直接跳过这一层，
不用每条都重试一次连接。
"""
global _explain_available
if _explain_available is False:
return []
try:
from common.neo4j_manager import neo4j_client
neo4j_client.run_cypher(f"EXPLAIN {cypher}")
_explain_available = True
return []
except Exception as exc:
message = str(exc)
# 区分"库连不上"和"这条语句本身有问题"：连接类异常不该算语句不合法
if _explain_available is None and (
"Unable to retrieve routing information" in message
or "ServiceUnavailable" in message
or "AuthError" in type(exc).__name__
or "Couldn't connect" in message
):
_explain_available = False
print(f"[warn] Neo4j 连不上，跳过 EXPLAIN 校验: {type(exc).__name__}")
return []
_explain_available = True
return [f"EXPLAIN 失败: {message.splitlines()[0][:120]}"]
# ======================
# 对外接口
# ======================
def validate_cypher(cypher: str, use_explain: bool = True) -> List[str]:
"""检查单条 cypher，返回问题列表。空列表表示合法。
三层依次短路：静态检查没过就不必再对 schema 和连库了，问题已经确定。
"""
cypher = (cypher or "").strip()
if not cypher:
return ["空语句"]
problems = _check_readonly(cypher)
if problems:
return problems
problems = _check_schema(cypher)
if problems:
return problems
return _check_explain(cypher) if use_explain else []
def check_cypher_node01(state: AgentState, use_explain: bool = True) -> AgentState:
"""检查 cypher_query 里所有语句是否合法，结果写回 is_all_validate_cypher"""
cyphers = state.get("cypher_query") or []
if not cyphers:
state["is_all_validate_cypher"] = EMPTY_IS_VALID
return state
all_valid = True
for cypher in cyphers:
problems = validate_cypher(cypher, use_explain=use_explain)
if problems:
all_valid = False
# 打日志：不合法时下游只看到一个 False，不记下来就无法定位是哪条、错在哪
print(f"[warn] 非法 cypher: {cypher[:80]}")
for problem in problems:
print(f" - {problem}")
state["is_all_validate_cypher"] = all_valid
return state
if __name__ == "__main__":
print(f"schema: {len(LABELS)} 标签 / {len(REL_TYPES)} 关系 / {len(PROP_KEYS)} 属性")
# (语句, 期望合法) —— 前四条来自 generate_neo4j_cypher_node 的真实输出
cases: List[Tuple[str, bool]] = [
(
"MATCH (h:Herb)-[:HAS_EFFECT]->(e:Effect) WHERE h.name IN ['人参'] "
"RETURN h.name AS herb, e.name AS effect LIMIT 20",
True,
),
(
"MATCH (f:Formula)-[:HAS_INGREDIENT]->(h:Herb) WHERE f.name = '四君子汤' "
"RETURN h.name AS herb LIMIT 20",
True,
),
(
"MATCH (s:Source)<-[:FROM_SOURCE]-(f:Formula) WHERE s.name IN ['《伤寒论》'] "
"RETURN f.name AS formula LIMIT 20",
True,
),
(
"MATCH (h:Herb) WHERE h.name = '人参' "
"RETURN h.property_flavor AS flavor, h.meridian AS meridian LIMIT 20",
True,
),
# 标签不在 schema 里：语法合法，执行恒空 —— 第 2 层要抓的就是这种
(
"MATCH (m:Medicine)-[:HAS_EFFECT]->(e:Effect) RETURN m.name AS name LIMIT 20",
False,
),
# 关系名编造
(
"MATCH (h:Herb)-[:CURES]->(d:Disease) RETURN d.name AS disease LIMIT 20",
False,
),
# 属性编造
("MATCH (h:Herb) RETURN h.price AS price LIMIT 20", False),
# 写操作
("MATCH (h:Herb {name: '人参'}) SET h.name = '党参' RETURN h.name AS n", False),
# 没有 RETURN
("MATCH (h:Herb) WHERE h.name = '人参'", False),
# 语法错误：括号不闭合，前两层看不出来，靠 EXPLAIN
("MATCH (h:Herb RETURN h.name AS herb LIMIT 20", False),
# 实体名里带点号和标签同名词，不该被误判
(
"MATCH (h:Herb) WHERE h.name IN ['Herb', '桂枝.加减'] RETURN h.name AS n LIMIT 5",
True,
),
("", False),
]
print("\n=== 单条校验 ===")
wrong = 0
for cypher, expected in cases:
problems = validate_cypher(cypher)
actual = not problems
flag = "OK " if actual == expected else "MISS"
if actual != expected:
wrong += 1
print(f" [{flag}] {'合法' if actual else '非法'} <- {cypher[:64] or '<空>'}")
for problem in problems:
print(f" {problem}")
print(f"\n预期不符 {wrong} 条")
print("\n=== 节点整体 ===")
for cypher_query in [
[cases[0][0], cases[1][0]], # 全合法 -> True
[cases[0][0], cases[4][0]], # 混了一条非法 -> False
[], # 空列表 -> False
]:
result = check_cypher_node01({"cypher_query": list(cypher_query)})
print(f" {len(cypher_query)} 条 -> {result['is_all_validate_cypher']}")
#我们不用这个node了,逻辑有些复杂
#逻辑:
#1. 用正则进行cypher语句的检查如果没问题
#2. 用neo4j的元数据进行cypher语句的检查如果没问题
# 3.用Explain 语句进行检查