Spaces:
Sleeping
Sleeping
File size: 5,039 Bytes
88cc76c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 |
import os
import sys
from fastapi import Depends
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from app.core.config import Settings, get_settings
from utils.data_mapping import DataMapping
from utils.extract_entity import extract_entities
from core.type import Node
from neo4j import GraphDatabase
from utils.constant import NEO4J_LABELS, NEO4J_RELATIONS
NEO4J_URI = os.getenv("NEO4J_URI", "neo4j://localhost:7687")
NEO4J_USER = os.getenv("NEO4J_USER", "neo4j")
NEO4J_PASSWORD = os.getenv("NEO4J_PASSWORD", "password")
NEO4J_DATABASE = os.getenv("NEO4J_DATABASE", "neo4j")
class Neo4jConnection:
def __init__(self):
"""Khởi tạo kết nối tới Neo4j"""
self.uri = NEO4J_URI
self.user = NEO4J_USER
self.password = NEO4J_PASSWORD
self.database = NEO4J_DATABASE
self.driver = GraphDatabase.driver(
self.uri,
auth=(self.user, self.password),
database=self.database
)
self.entity_types = []
self.relations = []
with self.driver.session() as session:
result = session.run("CALL db.info()")
self.database_info = result.single().data()
self.entity_types = NEO4J_LABELS
self.relations = NEO4J_RELATIONS
def get_database_info(self):
"""Trả về thông tin về database đang kết nối"""
return self.database_info
def close(self):
"""Đóng kết nối tới Neo4j"""
if self.driver is not None:
self.driver.close()
def execute_query(self, query, parameters=None):
"""Thực thi một truy vấn Cypher bất kỳ"""
with self.driver.session() as session:
result = session.run(query, parameters)
return [record for record in result]
class KnowledgeGraphUtils:
def get_disease_from_env_factors(self, crop_id: str, params: list[Node]):
envFactors = [param.id for param in params if param.label == "EnvironmentalFactor"]
query = f"""
MATCH (c:Crop {{id: "{crop_id}"}})
WITH c
MATCH (d:Disease)-[:AFFECTS]-(c)
OPTIONAL MATCH (ef:EnvironmentalFactor)-[:FAVORS]-(d)
WHERE ef.id IN {envFactors}
OPTIONAL MATCH (ef2:EnvironmentalFactor)-[:FAVORS]-(cause:Cause)-[:CAUSES|AFFECTS]-(d)
WHERE ef2.id IN {envFactors}
WITH d, COLLECT(DISTINCT ef.id) AS direct_env, COLLECT(DISTINCT ef2.id) AS indirect_env
WHERE SIZE(direct_env) > 0 OR SIZE(indirect_env) > 0
RETURN DISTINCT d, direct_env, indirect_env
"""
kg = Neo4jConnection()
result = kg.execute_query(query)
print(result)
final_result = []
for record in result:
record_dict = dict(record)
disease = Node.map_json_to_node(dict(record_dict["d"]), "Disease")
env_ids = list(record_dict["direct_env"]) + list(record_dict["indirect_env"])
print(env_ids)
score = 0
for env_id in env_ids:
for param in params:
if param.id == env_id:
score = max(score, param.score)
disease.score = score
final_result.append({
"disease": disease,
"env_ids": env_ids
})
final_result.sort(key=lambda x: x["disease"].score, reverse=True)
return final_result
def get_disease_from_symptoms(self, crop_id: str, params: list[Node]) -> list:
symptoms = [param.id for param in params if param.label == "Symptom"]
query = f"""
MATCH (c:Crop {{id: "{crop_id}"}})
WITH c
MATCH (d:Disease)-[:AFFECTS]-(c)
OPTIONAL MATCH (sym1:Symptom)-[:HAS_SYMPTOM]-(d)
WHERE sym1.id IN {symptoms}
OPTIONAL MATCH (sym2:Symptom)-[:HAS_SYMPTOM|LOCATED_ON]-(p:PlantPart)-[:CONTAINS]-(d)
WHERE sym2.id IN {symptoms}
WITH d, p, c, sym1, sym2, COLLECT(DISTINCT sym1.id) AS direct_env, COLLECT(DISTINCT sym2.id) AS indirect_env
WHERE SIZE(direct_env) > 0 OR SIZE(indirect_env) > 0
RETURN d, c, p, sym1, sym2
"""
kg = Neo4jConnection()
result = kg.execute_query(query)
final_result = []
for record in result:
record_dict = dict(record)
disease = Node.map_json_to_node(dict(record_dict["d"]), "Disease")
symptom_ids = list(record_dict["sym1"]) + list(record_dict["sym2"])
score = 0
for symptom_id in symptom_ids:
for param in params:
if param.id == symptom_id:
score = max(score, param.score)
disease.score = score
final_result.append({
"disease": disease,
"symptom_ids": symptom_ids
})
final_result.sort(key=lambda x: x["disease"].score, reverse=True)
return final_result
|