|
|
import logging
|
|
|
from typing import List
|
|
|
|
|
|
from fastapi import HTTPException
|
|
|
|
|
|
from app.database import neo4j_conn
|
|
|
from app.schemas.user import ChapterRelationCreate, ChapterRelationResponse
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
async def get_graph_relations():
|
|
|
"""
|
|
|
查询知识图谱中的所有节点及关系的详细信息
|
|
|
:return: 列表，每一项为 ChapterRelationResponse 对象
|
|
|
"""
|
|
|
query = """
|
|
|
MATCH (start)-[r]->(end)
|
|
|
RETURN
|
|
|
properties(start) AS start_properties,
|
|
|
labels(start) AS start_labels,
|
|
|
TYPE(r) AS relation,
|
|
|
properties(end) AS end_properties,
|
|
|
labels(end) AS end_labels
|
|
|
"""
|
|
|
async with neo4j_conn.get_session() as session:
|
|
|
result = await session.run(query)
|
|
|
records = await result.data()
|
|
|
|
|
|
responses = []
|
|
|
for record in records:
|
|
|
response = ChapterRelationResponse(
|
|
|
# start_id=record["start_id"],
|
|
|
start_labels=record["start_labels"],
|
|
|
start_properties=record["start_properties"],
|
|
|
relation=record["relation"],
|
|
|
# relation_properties=record["relation_properties"],
|
|
|
#
|
|
|
# end_id=record["end_id"],
|
|
|
end_labels=record["end_labels"],
|
|
|
end_properties=record["end_properties"]
|
|
|
)
|
|
|
responses.append(response)
|
|
|
|
|
|
return responses
|
|
|
|
|
|
|
|
|
async def search_chapters(search_term: str) -> List[ChapterRelationResponse]:
|
|
|
"""
|
|
|
根据搜索关键字模糊查询章节及其相关章节
|
|
|
:param search_term: 搜索关键字
|
|
|
:return: 列表，每一项为 ChapterRelationResponse 对象
|
|
|
"""
|
|
|
# 使用参数化查询以防止Cypher注入
|
|
|
query = """
|
|
|
MATCH (start)-[r]->(end)
|
|
|
WHERE toLower(start.name) CONTAINS toLower($search_term)
|
|
|
or toLower(end.name) CONTAINS toLower($search_term)
|
|
|
RETURN
|
|
|
properties(start) AS start_properties,
|
|
|
labels(start) AS start_labels,
|
|
|
TYPE(r) AS relation,
|
|
|
properties(end) AS end_properties,
|
|
|
labels(end) AS end_labels
|
|
|
"""
|
|
|
try:
|
|
|
async with neo4j_conn.get_session() as session:
|
|
|
result = await session.run(query, parameters={"search_term": search_term})
|
|
|
records = await result.data()
|
|
|
|
|
|
responses = [
|
|
|
ChapterRelationResponse(
|
|
|
start_labels=record.get("start_labels", []),
|
|
|
start_properties=record.get("start_properties", {}),
|
|
|
relation=record.get("relation", ""),
|
|
|
end_labels=record.get("end_labels", []),
|
|
|
end_properties=record.get("end_properties", {}),
|
|
|
)
|
|
|
for record in records
|
|
|
]
|
|
|
|
|
|
return responses
|
|
|
except Exception as e:
|
|
|
logging.error(f"Error during search_chapters: {e}")
|
|
|
return []
|
|
|
|
|
|
|
|
|
async def get_relations_to_level_simple(level: int) -> List[ChapterRelationResponse]:
|
|
|
"""
|
|
|
简化版：根据目标层级查询从 Root 到目标层级的所有节点和关系
|
|
|
:param level: 目标层级（1 -> Root & Subject, 2 -> Root, Subject, Topic, ..., 5 -> Problem）
|
|
|
:return: 列表，每一项为 ChapterRelationResponse 对象
|
|
|
"""
|
|
|
hierarchy = ["Root", "Subject", "Topic", "Section", "SubSection", "Problem"]
|
|
|
|
|
|
if level < 0 or level > len(hierarchy):
|
|
|
raise ValueError("层级参数必须在 0 到 5 之间")
|
|
|
# 特殊处理 level=0 的情况
|
|
|
if level == 0:
|
|
|
query = """
|
|
|
MATCH (start)
|
|
|
WHERE ANY(label IN labels(start) WHERE label = 'Root')
|
|
|
RETURN
|
|
|
properties(start) AS start_properties,
|
|
|
labels(start) AS start_labels
|
|
|
"""
|
|
|
|
|
|
try:
|
|
|
async with neo4j_conn.get_session() as session:
|
|
|
result = await session.run(query)
|
|
|
records = await result.data()
|
|
|
|
|
|
return [
|
|
|
ChapterRelationResponse(
|
|
|
start_labels=record.get("start_labels", []),
|
|
|
start_properties=record.get("start_properties", {}),
|
|
|
relation="", # 没有关系信息
|
|
|
end_labels=[], # 没有终点节点
|
|
|
end_properties={}, # 没有终点节点属性
|
|
|
)
|
|
|
for record in records
|
|
|
]
|
|
|
except Exception as e:
|
|
|
logging.error(f"Error during get_relations_to_level_simple (level=0): {e}")
|
|
|
return []
|
|
|
# 获取从 Root 到目标层级的所有标签
|
|
|
target_labels = hierarchy[:level + 1]
|
|
|
|
|
|
query = """
|
|
|
MATCH (start)-[r]->(end)
|
|
|
WHERE ANY(label IN labels(start) WHERE label IN $target_labels)
|
|
|
AND ANY(label IN labels(end) WHERE label IN $target_labels)
|
|
|
RETURN
|
|
|
properties(start) AS start_properties,
|
|
|
labels(start) AS start_labels,
|
|
|
TYPE(r) AS relation,
|
|
|
properties(end) AS end_properties,
|
|
|
labels(end) AS end_labels
|
|
|
"""
|
|
|
|
|
|
try:
|
|
|
async with neo4j_conn.get_session() as session:
|
|
|
result = await session.run(query, parameters={"target_labels": target_labels})
|
|
|
records = await result.data()
|
|
|
|
|
|
return [
|
|
|
ChapterRelationResponse(
|
|
|
start_labels=record.get("start_labels", []),
|
|
|
start_properties=record.get("start_properties", {}),
|
|
|
relation=record.get("relation", ""),
|
|
|
end_labels=record.get("end_labels", []),
|
|
|
end_properties=record.get("end_properties", {}),
|
|
|
)
|
|
|
for record in records
|
|
|
]
|
|
|
except Exception as e:
|
|
|
logging.error(f"Error during get_relations_to_level_simple: {e}")
|
|
|
return []
|
|
|
async def get_chapter_relations(chapter: str) -> List[ChapterRelationResponse]:
|
|
|
"""
|
|
|
根据章节名称查找该章节及其相关节点和关系
|
|
|
:param chapter_name: 章节名称
|
|
|
:return: 列表，每一项为 ChapterRelationResponse 对象
|
|
|
"""
|
|
|
query = """
|
|
|
MATCH (start)-[r]->(end)
|
|
|
WHERE start.chapter = $chapter and end.chapter = $chapter
|
|
|
RETURN
|
|
|
properties(start) AS start_properties,
|
|
|
labels(start) AS start_labels,
|
|
|
TYPE(r) AS relation,
|
|
|
properties(end) AS end_properties,
|
|
|
labels(end) AS end_labels
|
|
|
"""
|
|
|
try:
|
|
|
async with neo4j_conn.get_session() as session:
|
|
|
result = await session.run(query, parameters={"chapter": chapter})
|
|
|
records = await result.data()
|
|
|
return [
|
|
|
ChapterRelationResponse(
|
|
|
start_labels=record.get("start_labels", []),
|
|
|
start_properties=record.get("start_properties", {}),
|
|
|
relation=record.get("relation", ""),
|
|
|
end_labels=record.get("end_labels", []),
|
|
|
end_properties=record.get("end_properties", {}),
|
|
|
)
|
|
|
for record in records
|
|
|
]
|
|
|
except Exception as e:
|
|
|
logging.error(f"Error during get_chapter_relations: {e}")
|
|
|
return [] |
