from sqlalchemy.orm import Session from typing import Optional from model.kg_node import KGNode from db.session import get_db import logging from sqlalchemy.exc import IntegrityError from service.kg_node_service import KGNodeService from tests.service.test_kg_node_service import kg_node_service from utils import vectorizer from utils.vectorizer import Vectorizer from sqlalchemy import func from service.kg_prop_service import KGPropService from service.kg_edge_service import KGEdgeService from cachetools import TTLCache from cachetools.keys import hashkey logger = logging.getLogger(__name__) DISTANCE_THRESHOLD = 0.65 class CdssService: _cache = TTLCache(maxsize=100000, ttl=60*60*24*30) def get_disease_detail(self, diease_name: str,category: str): nodeService = KGNodeService(next(get_db())) diseases = nodeService.search_title_index("", diease_name, category, 1, 0) if diseases is None: return None disease_id = diseases[0]["id"] edgeService = KGEdgeService(next(get_db())) #category='疾病相关实验室检查项目' laboratoryTests = edgeService.get_edges_by_nodes(src_id=disease_id, category='疾病相关实验室检查项目') #category='疾病相关辅助检查项目' examinations = edgeService.get_edges_by_nodes(src_id=disease_id, category='疾病相关辅助检查项目') # category='疾病相关鉴别诊断' differential_diagnosis = edgeService.get_edges_by_nodes(src_id=disease_id, category='疾病相关鉴别诊断') test_nodes = [] exam_nodes = [] ddx_nodes = [] if laboratoryTests: test_nodes = [test['dest_node'] for test in laboratoryTests] self.add_props_to_nodes(test_nodes) if examinations: exam_nodes = [exam['dest_node'] for exam in examinations] self.add_props_to_nodes(exam_nodes) if differential_diagnosis: ddx_nodes = [ddx['dest_node'] for ddx in differential_diagnosis] self.add_props_to_nodes(ddx_nodes,['Differential Diagnosis']) return { 'disease':diseases[0], 'laboratory_tests': test_nodes, 'examinations': exam_nodes, 'differential_diagnosis': ddx_nodes } def add_props_to_nodes(self, dest_nodes,prop_names=None): kgPropService = KGPropService(next(get_db())) for node in dest_nodes: props = kgPropService.get_props_by_ref_id(node['id'],prop_names) if props: node['props'] = props