parent
6ae9fd4d6e
commit
e5b714df41
Binary file not shown.
Binary file not shown.
@ -0,0 +1,12 @@
|
||||
from Config.Config import *
|
||||
from Neo4j.Neo4jExecutor import *
|
||||
|
||||
# 使用示例
|
||||
if __name__ == '__main__':
|
||||
executor = Neo4jExecutor(
|
||||
uri=NEO4J_URI,
|
||||
auth=NEO4J_AUTH
|
||||
)
|
||||
# 清库
|
||||
clear(executor.graph)
|
||||
print("清库成功")
|
||||
@ -0,0 +1,61 @@
|
||||
import json
|
||||
|
||||
from Neo4j.Neo4jExecutor import *
|
||||
|
||||
def json_to_cypher(data):
|
||||
"""将知识体系JSON转换为Cypher插入脚本"""
|
||||
cypher = []
|
||||
seen = set()
|
||||
|
||||
def process_node(node, parent_id=None):
|
||||
nonlocal seen
|
||||
|
||||
node_id = node['value']
|
||||
node_name = node['title'].replace("'", "''")
|
||||
|
||||
# 创建当前节点
|
||||
if node_id not in seen:
|
||||
cypher.append(f"MERGE (n:KnowledgePoint {{id: '{node_id}'}}) SET n.name = '{node_name}';")
|
||||
seen.add(node_id)
|
||||
|
||||
# 创建父子关系
|
||||
if parent_id:
|
||||
cypher.append(f"""
|
||||
MATCH (parent:KnowledgePoint {{id: '{parent_id}'}}),
|
||||
(child:KnowledgePoint {{id: '{node_id}'}})
|
||||
MERGE (parent)-[:HAS_SUB_POINT]->(child);""")
|
||||
|
||||
# 递归处理子节点
|
||||
if 'children' in node and not node['isLeaf']:
|
||||
for child in node['children']:
|
||||
process_node(child, parent_id=node_id)
|
||||
|
||||
# 处理根节点
|
||||
for root in data['data']['tree']:
|
||||
process_node(root)
|
||||
|
||||
# 处理根节点的父级关系（如果有）
|
||||
if root.get('parentValue'):
|
||||
cypher.append(f"MERGE (parent:KnowledgePoint {{id: '{root['parentValue']}'}});")
|
||||
cypher.append(f"""
|
||||
MATCH (parent:KnowledgePoint {{id: '{root['parentValue']}'}}),
|
||||
(child:KnowledgePoint {{id: '{root['value']}'}})
|
||||
MERGE (parent)-[:HAS_SUB_POINT]->(child);""")
|
||||
|
||||
return '\n'.join(cypher)
|
||||
|
||||
|
||||
# 使用示例
|
||||
if __name__ == '__main__':
|
||||
executor = Neo4jExecutor(
|
||||
uri=NEO4J_URI,
|
||||
auth=NEO4J_AUTH
|
||||
)
|
||||
# 清库
|
||||
clear(executor.graph)
|
||||
|
||||
# 这里替换成你的JSON数据变量
|
||||
with open('小学数学知识点体系.json', 'r',encoding='utf-8') as f:
|
||||
your_json_data = json.load(f)
|
||||
cypherText=json_to_cypher(your_json_data)
|
||||
executor.execute_cypher_text(cypherText)
|
||||
@ -0,0 +1,61 @@
|
||||
from py2neo import Graph
|
||||
import json
|
||||
from Config import *
|
||||
|
||||
def query_all_questions():
|
||||
# Neo4j连接配置
|
||||
neo4j_config = {
|
||||
"uri": NEO4J_URI, # 从Config导入
|
||||
"auth": NEO4J_AUTH # 从Config导入
|
||||
}
|
||||
|
||||
# 初始化图数据库连接
|
||||
graph = Graph(**neo4j_config)
|
||||
|
||||
try:
|
||||
# 获取所有试题ID
|
||||
question_ids = graph.run(
|
||||
"MATCH (q:Question) RETURN q.id AS question_id"
|
||||
).data()
|
||||
|
||||
results = []
|
||||
|
||||
# 遍历每个试题
|
||||
for record in question_ids:
|
||||
qid = record['question_id']
|
||||
# 修改查询部分
|
||||
data = graph.run("""
|
||||
MATCH (q:Question {id: $qid})
|
||||
OPTIONAL MATCH (q)-[:REQUIRES_KNOWLEDGE]->(kp:KnowledgePoint)
|
||||
OPTIONAL MATCH (q)-[:DEVELOPS_LITERACY]->(lp:LiteracyNode)
|
||||
RETURN
|
||||
q.content AS content,
|
||||
collect(DISTINCT {id: kp.id, name: kp.name}) AS knowledge_points,
|
||||
collect(DISTINCT {id: lp.value, title: lp.title}) AS literacy_points
|
||||
""", qid=qid).data()
|
||||
|
||||
if data:
|
||||
result = {
|
||||
"question_id": qid,
|
||||
"content": data[0]['content'],
|
||||
"knowledge_points": data[0]['knowledge_points'],
|
||||
"literacy_points": data[0]['literacy_points']
|
||||
}
|
||||
results.append(result)
|
||||
|
||||
# 增强版输出展示
|
||||
print(f"\n{'=' * 40} 题目详情 {'=' * 40}")
|
||||
print(f"📚 试题ID: {qid}")
|
||||
print(f"📝 内容全文: {result['content']}") # 新增完整内容输出
|
||||
print(f"🔍 内容摘要: {result['content'][:50]}...") # 保留摘要显示
|
||||
print(f"🧠 知识点: {[kp['name'] for kp in result['knowledge_points']]}")
|
||||
print(f"🆔 知识点ID: {[kp['id'] for kp in result['knowledge_points']]}")
|
||||
print(f"🌟 素养点: {[lp['title'] for lp in result['literacy_points']]}")
|
||||
print(f"🔢 素养点ID: {[lp['id'] for lp in result['literacy_points']]}")
|
||||
print('=' * 90)
|
||||
except Exception as e:
|
||||
print(f"❌ 查询失败: {str(e)}")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
query_all_questions()
|
||||
@ -0,0 +1,107 @@
|
||||
import re
|
||||
from Config import *
|
||||
from py2neo import Graph, Node, Relationship, Subgraph
|
||||
from typing import List, Tuple
|
||||
|
||||
def clear(db):
|
||||
# 清空数据
|
||||
db.run("MATCH (n) DETACH DELETE n")
|
||||
|
||||
# 分步删除约束和索引
|
||||
try:
|
||||
# 删除约束
|
||||
constraints = db.run("SHOW CONSTRAINTS YIELD name").data()
|
||||
for constr in constraints:
|
||||
db.run(f"DROP CONSTRAINT `{constr['name']}`")
|
||||
|
||||
# 删除索引
|
||||
indexes = db.run("SHOW INDEXES YIELD name, type WHERE type <> 'LOOKUP'").data()
|
||||
for idx in indexes:
|
||||
db.run(f"DROP INDEX `{idx['name']}`")
|
||||
except Exception as e:
|
||||
print(f"删除操作失败: {e}")
|
||||
|
||||
|
||||
def create_subgraph(db: Graph, nodes: List[Node], relations: List[Tuple[Node, str, Node]]) -> None:
|
||||
"""统一创建子图"""
|
||||
subgraph = Subgraph(
|
||||
nodes=nodes,
|
||||
relationships=[Relationship(start, rel_type, end) for start, rel_type, end in relations]
|
||||
)
|
||||
db.create(subgraph)
|
||||
|
||||
def tx_create(db: Graph, nodes: List[Node], relations: List[Tuple[Node, str, Node]]) -> None:
|
||||
"""事务方式创建数据"""
|
||||
try:
|
||||
tx = db.begin()
|
||||
subgraph = Subgraph(
|
||||
nodes=nodes,
|
||||
relationships=[Relationship(start, rel_type, end) for start, rel_type, end in relations]
|
||||
)
|
||||
tx.create(subgraph)
|
||||
db.commit(tx)
|
||||
except Exception as e:
|
||||
db.rollback(tx)
|
||||
print(f"事务操作失败: {str(e)}")
|
||||
raise
|
||||
|
||||
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)
|
||||
Binary file not shown.
File diff suppressed because it is too large
Load Diff
Loading…
Reference in new issue