knowledge/service/cdss_service.py

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