代码提交

cdss/capbility.py

@@ -1,64 +0,0 @@
-from cdss.models.schemas import CDSSDict, CDSSInput,CDSSInt,CDSSOutput,CDSSText
-from cdss.libs.cdss_helper import CDSSHelper
-
-# CDSS能力核心类，负责处理临床决策支持系统的核心逻辑
-class CDSSCapability:
-    # CDSS帮助类实例
-    cdss_helper: CDSSHelper = None
-    def __init__(self):
-        # 初始化CDSS帮助类
-        self.cdss_helper = CDSSHelper()
-    
-    # 核心处理方法，接收输入数据并返回决策支持结果
-    # @param input: CDSS输入数据
-    # @param embeding_search: 是否使用embedding搜索，默认为True
-    # @return: CDSS输出结果
-    def process(self, input: CDSSInput, embeding_search:bool = True) -> CDSSOutput:
-        start_nodes = []
-        # 获取主诉信息
-        chief_complaint = input.get_value("chief_complaint")
-        # 初始化输出对象
-        output = CDSSOutput()
-        # 如果存在主诉信息，则进行处理
-        if chief_complaint:
-            for keyword in chief_complaint:
-                # 使用帮助类进行节点搜索，查找与关键词相关的节点
-                results = self.cdss_helper.node_search(
-                    keyword, limit=10, node_type="word"
-                )
-                for item in results:
-                    if item['score']>1.9:
-                        start_nodes.append(item['id'])
-            print("cdss start from :", start_nodes)    
-            # 使用帮助类进行CDSS路径遍历，最大跳数为2
-            result = self.cdss_helper.cdss_travel(input, start_nodes,max_hops=2)
-                
-            for item in result["details"]:
-                name, data = item
-                output.diagnosis.value[name] = data
-                print(f"{name}  {data['score'] *100:.2f} %")
-
-                for disease in data["diseases"]:
-                     print(f"\t疾病：{disease[0]} ")
-
-
-                for check in data["checks"]:
-                     print(f"\t检查：{check[0]} ")
-
-
-                for drug in data["drugs"]:
-                     print(f"\t药品：{drug[0]} ")
-
-            # 输出最终推荐的检查和药品信息
-            print("最终推荐的检查和药品如下：")
-            for item in result["checks"][:5]:        
-                item[1]['score'] = item[1]['count'] / result["total_checks"]
-                output.checks.value[item[0]] = item[1]                
-                # print(f"\t检查：{item[0]}  {item[1]['score'] * 100:.2f} %")    
-                # print(f"\t检查：{item[0]}  {item[1]['count'] / result["total_checks"] * 100:.2f} %")
-            
-            for item in result["drugs"][:5]:          
-                item[1]['score'] = item[1]['count'] / result["total_checks"]
-                output.drugs.value[item[0]] = item[1]  
-                #print(f"\t药品：{item[0]} {item[1]['count'] / result["total_drugs"] * 100:.2f} %")
-        return output

cdss/libs/cdss_helper.py

@@ -1,214 +0,0 @@
-import os
-import sys
-current_path = os.getcwd()
-sys.path.append(current_path)
-
-from community.graph_helper import GraphHelper
-from typing import List
-from cdss.models.schemas import CDSSInput
-class CDSSHelper(GraphHelper):
-    def check_sex_allowed(self, node, sex):        
-        #性别过滤，假设疾病节点有一个属性叫做allowed_sex_type，值为“0,1,2”，分别代表未知，男，女
-        sex_allowed = self.graph.nodes[node].get('allowed_sex_list', None)
-        if sex_allowed:
-            sex_allowed_list = sex_allowed.split(',')
-            if sex not in sex_allowed_list:
-                #如果性别不匹配，跳过
-                return False
-        return True
-    def check_age_allowed(self, node, age):
-        #年龄过滤，假设疾病节点有一个属性叫做allowed_age_range，值为“6-88”，代表年龄在0-88月之间是允许的
-        #如果说年龄小于6岁，那么我们就认为是儿童，所以儿童的年龄范围是0-6月
-        age_allowed = self.graph.nodes[node].get('allowed_age_range', None)
-        if age_allowed:
-            age_allowed_list = age_allowed.split('-')
-            age_min = int(age_allowed_list[0])
-            age_max = int(age_allowed_list[-1])
-            if age >= age_min and age < age_max:
-                #如果年龄范围正常，那么返回True
-                return True
-        else:
-            #如果没有设置年龄范围，那么默认返回True
-            return True
-        return False
-        
-    def cdss_travel(self, input:CDSSInput, start_nodes:List, max_hops=3):      
-        #这里设置了节点的type取值范围，可以根据实际情况进行修改，允许出现多个类型
-        DEPARTMENT=['科室']
-        DIESEASE=['疾病']
-        DRUG=['药品']
-        CHECK=['检查']
-        SYMPTOM=['症状']
-        #allowed_types = ['科室', '疾病', '药品', '检查', '症状']
-        allowed_types = DEPARTMENT + DIESEASE+ DRUG + CHECK + SYMPTOM
-        #这里设置了边的type取值范围，可以根据实际情况进行修改，允许出现多个类型
-        #不过后面的代码里面没有对边的type进行过滤，所以这里是留做以后扩展的
-        allowed_links = ['has_symptom', 'need_check', 'recommend_drug', 'belongs_to']
-        #详细解释下面一行代码
-        #queue是一个队列，里面存放了待遍历的节点，每个节点都有一个depth，表示当前节点的深度，
-        #一个path，表示当前节点的路径，一个data，表示当前节点的一些额外信息，比如allowed_types和allowed_links
-        #allowed_types表示当前节点的类型，allowed_links表示当前节点的边的类型
-        #这里的start_nodes是一个列表，里面存放了起始节点，每个起始节点都有一个depth为0，一个path为"/"，一个data为{'allowed_types': allowed_types, 'allowed_links':allowed_links}
-        #这里的"/"表示根节点，因为根节点没有父节点，所以路径为"/"
-        #这里的data是一个字典，里面存放了allowed_types和allowed_links，这两个值都是列表，里面存放了允许的类型
-        queue = [(node, 0, "/", {'allowed_types': allowed_types, 'allowed_links':allowed_links}) for node in start_nodes]        
-        visited = set()      
-        results = {}
-        #整理input的数据，这里主要是要检查输入数据是否正确，也需要做转换
-        if input.pat_age.value > 0 and input.pat_age.type == 'year':
-            #这里将年龄从年转换为月，因为我们的图里面的年龄都是以月为单位的
-            input.pat_age.value = input.pat_age.value * 12
-            input.pat_age.type = 'month'
-            
-        #STEP 1: 假设start_nodes里面都是症状，第一步我们先找到这些症状对应的疾病
-        #由于这部分是按照症状逐一去寻找疾病，所以实际应用中可以缓存这些结果
-        while queue:
-            node, depth, path, data = queue.pop(0)
-            #allowed_types = data['allowed_types']
-            #allowed_links = data['allowed_links']
-            indent = depth * 4
-            node_type = self.graph.nodes[node].get('type')
-            if node_type in DIESEASE:
-                if self.check_sex_allowed(node, input.pat_sex.value) == False:
-                    continue
-                if self.check_age_allowed(node, input.pat_age.value) == False:
-                    continue
-                if node in results.keys():                 
-                    results[node]["count"] = results[node]["count"] + 1   
-                    #print("疾病", node, "出现的次数", results[node]["count"])
-                else:
-                    results[node] = {"type": node_type, "count":1, 'path':path}            
-                continue
-            
-            if node in visited or depth > max_hops:
-                #print(">>> already visited or reach max hops")
-                continue                
-            
-            visited.add(node)
-            for edge in self.graph.edges(node, data=True):                
-                src, dest, edge_data = edge
-                #if edge_data.get('type') not in allowed_links:
-                #    continue
-                if edge[1] not in visited and depth + 1 < max_hops:                    
-                    queue.append((edge[1], depth + 1, path+"/"+src, data))
-                    #print("-" * (indent+4), f"start travel from {src} to {dest}")    
-        
-        #STEP 2: 找到这些疾病对应的科室,检查和药品       
-        #由于这部分是按照疾病逐一去寻找，所以实际应用中可以缓存这些结果
-        for disease in results.keys():
-            queue = [(disease, 0, {'allowed_types': DEPARTMENT, 'allowed_links':['belongs_to']})]
-            visited = set()
-            while queue:
-                node, depth, data = queue.pop(0)                
-                indent = depth * 4
-                if node in visited or depth > max_hops:
-                    #print(">>> already visited or reach max hops")  
-                    continue                 
-                
-                visited.add(node)
-                node_type = self.graph.nodes[node].get('type')
-                if node_type in DEPARTMENT:
-                    #展开科室，重复次数为疾病出现的次数，为了方便后续统计
-                    department_data = [node] * results[disease]["count"]
-                    # if results[disease]["count"] > 1:
-                    #     print("展开了科室", node, "次数", results[disease]["count"], "次")
-                    if 'department' in results[disease].keys():
-                        results[disease]["department"] = results[disease]["department"] + department_data
-                    else:
-                        results[disease]["department"] = department_data
-                    continue
-                if node_type in CHECK:
-                    if 'check' in results[disease].keys():
-                        results[disease]["check"] = list(set(results[disease]["check"]+[node]))
-                    else:
-                        results[disease]["check"] = [node]
-                    continue
-                if node_type in DRUG:
-                    if 'drug' in results[disease].keys():
-                        results[disease]["drug"] = list(set(results[disease]["drug"]+[node]))
-                    else:
-                        results[disease]["drug"] = [node]
-                    continue
-                for edge in self.graph.edges(node, data=True):                
-                    src, dest, edge_data = edge
-                    #if edge_data.get('type') not in allowed_links:
-                    #    continue
-                    if edge[1] not in visited and depth + 1 < max_hops:                    
-                        queue.append((edge[1], depth + 1, data))
-                        #print("-" * (indent+4), f"start travel from {src} to {dest}")    
-        #STEP 3: 对于结果按照科室维度进行汇总
-        final_results = {}
-        total = 0
-        for disease in results.keys():
-            if 'department' in results[disease].keys():
-                total += 1
-                for department in results[disease]["department"]:
-                    if not department in final_results.keys():
-                        final_results[department] = {
-                            "diseases": [disease],
-                            "checks": results[disease].get("check",[]), 
-                            "drugs": results[disease].get("drug",[]),
-                            "count": 1
-                        }
-                    else:
-                        final_results[department]["diseases"] = final_results[department]["diseases"]+[disease]
-                        final_results[department]["checks"] = final_results[department]["checks"]+results[disease].get("check",[])
-                        final_results[department]["drugs"] = final_results[department]["drugs"]+results[disease].get("drug",[])
-                        final_results[department]["count"] += 1
-        
-        for department in final_results.keys():
-            final_results[department]["score"] = final_results[department]["count"] / total
-        #STEP 4: 对于final_results里面的disease，checks和durgs统计出现的次数并且按照次数降序排序
-        def sort_data(data, count=5):
-            tmp = {}
-            for item in data:
-                if item in tmp.keys():
-                    tmp[item]["count"] +=1
-                else:
-                    tmp[item] = {"count":1}
-            sorted_data = sorted(tmp.items(), key=lambda x:x[1]["count"],reverse=True)
-            return sorted_data[:count]
-        
-        for department in final_results.keys():
-            final_results[department]['name'] = department
-            final_results[department]["diseases"] = sort_data(final_results[department]["diseases"])
-            final_results[department]["checks"] = sort_data(final_results[department]["checks"])
-            final_results[department]["drugs"] = sort_data(final_results[department]["drugs"])
-        
-        sorted_final_results = sorted(final_results.items(), key=lambda x:x[1]["count"],reverse=True)
-        
-        #STEP 5: 对于final_results里面的checks和durgs统计全局出现的次数并且按照次数降序排序
-        checks = {}
-        drugs ={}
-        total_check = 0
-        total_drug = 0
-        for department in final_results.keys():
-            for check, data in final_results[department]["checks"]:
-                total_check += 1
-                if check in checks.keys():
-                    checks[check]["count"] += data["count"]
-                else:
-                    checks[check] = {"count":data["count"]}
-            for drug, data in final_results[department]["drugs"]:
-                total_drug += 1
-                if drug in drugs.keys():
-                    drugs[drug]["count"] += data["count"]
-                else:
-                    drugs[drug] = {"count":data["count"]}
-        
-        sorted_checks = sorted(checks.items(), key=lambda x:x[1]["count"],reverse=True)
-        sorted_drugs = sorted(drugs.items(), key=lambda x:x[1]["count"],reverse=True)
-        #STEP 6: 整合数据并返回
-            # if "department" in item.keys():
-            #     final_results["department"] = list(set(final_results["department"]+item["department"]))
-            # if "diseases" in item.keys():
-            #     final_results["diseases"] = list(set(final_results["diseases"]+item["diseases"]))
-            # if "checks" in item.keys():
-            #     final_results["checks"] = list(set(final_results["checks"]+item["checks"]))
-            # if "drugs" in item.keys():
-            #     final_results["drugs"] = list(set(final_results["drugs"]+item["drugs"]))
-            # if "symptoms" in item.keys():
-            #     final_results["symptoms"] = list(set(final_results["symptoms"]+item["symptoms"]))
-        return {"details":sorted_final_results, 
-                "checks":sorted_checks, "drugs":sorted_drugs, 
-                "total_checks":total_check, "total_drugs":total_drug}

cdss/models/schemas.py

@@ -1,71 +0,0 @@
-
-from typing import List, Optional, Dict, Union
-
-class CDSSInt:
-    type: str
-    value: Union[int, float]
-    def __init__(self, type:str='nubmer' , value: Union[int,float]=0):
-        self.type = type
-        self.value = value
-    def __str__(self):
-        return f"{self.type}:{self.value}"
-    def value(self):
-        return self.value
-    
-class CDSSText:
-    type: str
-    value: Union[str, List[str]]
-    def __init__(self, type:str='text', value: Union[str, List[str]]=""):
-        self.type = type
-        self.value = value
-    def __str__(self):
-        if isinstance(self.value, str):
-            return f"{self.type}:{self.value}"
-        return f"{self.type}:{','.join(self.value)}"
-    def value(self):
-        return self.value    
-
-class CDSSDict:
-    type: str
-    value: Dict
-    def __init__(self, type:str='dict', value: Dict={}):
-        self.type = type
-        self.value = value
-    def __str__(self):
-        return f"{self.type}:{self.value}"
-    def value(self):
-        return self.value
-
-# pat_name:患者名字，字符串，如"张三"，无患者信息输出""
-# pat_sex:患者性别，字符串，如"男"，无患者信息输出""
-# pat_age:患者年龄，数字，单位为年，如25岁，输出25，无年龄信息输出0
-# clinical_department:就诊科室，字符串，如"呼吸内科"，无就诊科室信息输出""
-# chief_complaint:主诉，字符串列表，包括主要症状的列表，如["胸痛","发热"]，无主诉输出[]
-# present_illness:现病史，字符串列表，包括症状发展过程、诱因、伴随症状（如疼痛性质、放射部位、缓解方式，无现病史信息输出[]
-# past_medical_history:既往病史，字符串列表，包括疾病史（如高血压、糖尿病）、手术史、药物过敏史、家族史等，无现病史信息输出[]
-# physical_examination:体格检查，字符串列表，如生命体征（血压、心率）、心肺腹部体征、实验室/影像学结果（如心电图异常、肌钙蛋白升高），无信息输出[]
-# lab_and_imaging:检验与检查，字符串列表，包括血常规、生化指标、心电图（ECG）、胸部X光、CT等检查项目，结果和报告等，无信息输出[]
-class CDSSInput:
-    pat_age: CDSSInt = CDSSInt(type='year', value=0)
-    pat_sex: CDSSInt= CDSSInt(type='sex', value=0)
-    values: List[CDSSText] 
-    
-    def __init__(self, **kwargs):
-        #提取kwargs中的所有字段，并将它们添加到类的属性中。这样，在创建子类时，就可以直接使用这些字段了。
-        values = []
-        for key, value in kwargs.items():
-            #如果key的属性已经存在，则将value设置为该属性的值。否则，将value添加到values列表中。            
-            if hasattr(self, key):
-                setattr(self, key, value)
-            else:
-                values.append(CDSSText(key, value))
-        setattr(self, 'values', values)
-    def get_value(self, key)->CDSSText:
-        for value in self.values:
-            if value.type == key:
-                return value.value
-        
-class CDSSOutput:
-    diagnosis: CDSSDict = CDSSDict(type='diagnosis', value={})
-    checks: CDSSDict = CDSSDict(type='checks', value={})
-    drugs: CDSSDict = CDSSDict(type='drugs', value={})

community/community_report.py

@@ -1,211 +0,0 @@
-"""
-社区报告生成模块
-
-本模块用于从dump的图谱数据生成社区算法报告
-
-主要功能:
-1. 生成社区分析报告
-2. 计算社区内部连接密度
-3. 生成可视化分析报告
-"""
-import sys,os
-current_path = os.getcwd()
-sys.path.append(current_path)
-
-import networkx as nx
-import leidenalg
-import igraph as ig
-#import matplotlib.pyplot as plt
-import json
-from datetime import datetime
-from collections import Counter
-
-#社区报告的分辨率，数字越大，社区数量越少，数字越小，社区数量越多
-#RESOLUTION = 0.07
-#社区报告中是否包括节点的属性列表
-REPORT_INCLUDE_DETAILS = False
-# #图谱数据的缓存路径，数据从dump_graph_data.py生成
-# CACHED_DATA_PATH = f"{current_path}\\web\\cached_data"
-# #最终社区报告的输出路径
-REPORT_PATH = f"{current_path}\\web\\cached_data\\report"
-DENSITY = 0.52
-# def load_entity_data():
-#     print("load entity data")
-#     with open(f"{CACHED_DATA_PATH}\\entities_med.json", "r", encoding="utf-8") as f:
-#         entities = json.load(f)
-#         return entities
-
-# def load_relation_data(g):
-#     for i in range(30):
-#         if os.path.exists(f"{CACHED_DATA_PATH}\\relationship_med_{i}.json"):            
-#             print("load entity data", f"{CACHED_DATA_PATH}\\relationship_med_{i}.json")
-#             with open(f"{CACHED_DATA_PATH}\\relationship_med_{i}.json", "r", encoding="utf-8") as f:
-#                 relations = json.load(f)
-#                 for item in relations:                    
-#                     g.add_edge(item[0], item[1], weight=1, **item[2])
-        
-            
-        
-
-# def generate_enterprise_network():
-
-#     G = nx.Graph()
-#     ent_data = load_entity_data()
-#     print("load entities completed")
-#     for data in ent_data:          
-#         G.add_node(data[0], **data[1])
-#     print("load entities into graph completed")
-#     rel_data = load_relation_data(G)    
-#     print("load relation completed")
-
-#     return G
-
-# def detect_communities(G):
-#     """使用Leiden算法进行社区检测"""
-#     # 转换networkx图到igraph格式
-    
-#     print("convert to igraph")
-#     ig_graph = ig.Graph.from_networkx(G)
-    
-#     # 执行Leiden算法
-#     partition = leidenalg.find_partition(
-#         ig_graph, 
-#         leidenalg.CPMVertexPartition,
-#         resolution_parameter=RESOLUTION,
-#         n_iterations=2
-#     )
-    
-#     # 将社区标签添加到原始图
-#     for i, node in enumerate(G.nodes()):
-#         G.nodes[node]['community'] = partition.membership[i]
-    
-#     print("convert to igraph finished")
-#     return G, partition
-
-def generate_report(G, partition):
-    """
-    生成结构化分析报告
-    
-    参数:
-        G: NetworkX图对象，包含节点和边的信息
-        partition: Leiden算法返回的社区划分结果
-    
-    返回:
-        str: 生成的分析报告内容
-    """
-    report = []
-    # 报告头信息
-    report.append(f"# 疾病图谱关系社区分析报告\n")
-    report.append(f"**生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
-    report.append(f"**检测算法**: Leiden Algorithm\n")
-    report.append(f"**算法参数**:\n")
-    report.append(f"- 分辨率参数: {partition.resolution_parameter:.3f}\n")
-    # report.append(f"- 迭代次数: {partition.n_iterations}\n")
-    report.append(f"**社区数量**: {len(set(partition.membership))}\n")
-    report.append(f"**模块度(Q)**: {partition.quality():.4f}\n")
-    print("generate_report header finished")
-
-    report.append("\n## 社区结构分析\n")
-    print("generate_report community structure started")
-    communities = {}
-    for node in G.nodes(data=True):
-        comm = node[1]['community']
-        if comm not in communities:
-            communities[comm] = []
-        if 'type' not in node[1]:
-            node[1]['type'] = '未知'
-        if 'description' not in node[1]:
-            node[1]['description'] = '未见描述'
-        
-        communities[comm].append({
-            'name': node[0],
-            **node[1]
-        })
-       
-
-
-    print("generate_report community structure finished")
-    for comm_id, members in communities.items():
-        print("community ", comm_id, "size: ", len(members))
-        com_report = []
-        com_report.append(f"### 第{comm_id+1}号社区报告 ")
-        #com_report.append(f"**社区规模**: {len(members)} 个节点\n")
-
-        # 行业类型分布
-        type_dist = Counter([m['type'] for m in members])
-        com_report.append(f"**类型分布**:")
-        for industry, count in type_dist.most_common():
-            com_report.append(f"- {industry}: {count} 个 ({count/len(members):.0%})")
-
-
-        com_report.append("\n**成员节点**:")
-        member_names = ''
-        member_count = 0
-        for member in members:
-            if member_count < 8:
-                #member['name']如果有会导致文件名报错的字符，需要去除
-                member_name = member['name'].replace('\\', '').replace('/', '').replace(':', '').replace('*', '').replace('?', '').replace('"', '').replace('<', '').replace('>', '').replace('|', '')
-                member_names += member_name + '_'
-                member_count += 1
-            com_report.append(f"- {member['name']} ({member['type']})")
-            if REPORT_INCLUDE_DETAILS == False:
-                continue
-            for k in member.keys():
-                if k not in ['name', 'type', 'description', 'community']:
-                    value = member[k]
-                    com_report.append(f"\t- {value}")
-
-        com_report.append("\n**成员节点关系**:\n")
-        for member in members:
-            entities, relations = graph_helper.neighbor_search(member['name'], 1)
-            com_report.append(f"- {member['name']} ({member['type']})")
-            com_report.append(f"\t- 相关节点")
-            for entity in entities:
-                com_report.append(f"\t\t- {entity['id']} ({entity['type']})")
-            com_report.append(f"\t- 相关关系")
-            for relation in relations:
-                com_report.append(f"\t\t- {relation['src_name']}-({relation['type']})->{relation['dest_name']}")
-            
-                    
-        # 计算社区内部连接密度
-        subgraph = G.subgraph([m['name'] for m in members])
-        density = nx.density(subgraph)
-        com_report.append(f"\n**内部连接密度**: {density:.2f}\n")
-        if density < DENSITY:
-            com_report.append("**社区内部连接相对稀疏**\n")
-        else:
-            with open(f"{REPORT_PATH}\{member_names}{comm_id}.md", "w", encoding="utf-8") as f:
-                f.write("\n".join(com_report))
-        print(f"社区 {comm_id+1} 报告文件大小：{len(''.join(com_report).encode('utf-8'))} 字节")  # 添加文件生成验证
-    
-    # 可视化图表
-    report.append("\n## 可视化分析\n")
-    
-    return "\n".join(report)
-
-
-if __name__ == "__main__":
-    try:
-        from graph_helper2 import GraphHelper
-        graph_helper = GraphHelper()
-        G = graph_helper.graph
-        print("graph loaded")
-        # 生成企业关系网络
-        
-        
-        # 执行社区检测
-        G, partition = graph_helper.detect_communities()
-        
-        # 生成分析报告
-        report = generate_report(G, partition)
-        with open('community_report.md', 'w', encoding='utf-8') as f:
-            f.write(report)
-            print(f"报告文件大小：{len(report.encode('utf-8'))} 字节")  # 添加文件生成验证
-                        
-            print("社区分析报告已生成：community_report.md")
-            
-       
-    except Exception as e:
-        
-        print(f"运行时错误：{str(e)}")
-        raise e

community/dump_graph_data.py

@@ -1,110 +0,0 @@
-'''
-这个脚本是用来从postgre数据库中导出图谱数据到json文件的。
-'''
-import sys,os
-current_path = os.getcwd()
-sys.path.append(current_path)
-
-from sqlalchemy import text
-from sqlalchemy.orm import Session
-import json
-#这个是数据库的连接
-from db.session import SessionLocal
-#两个会话，分别是读取节点和属性的
-db = SessionLocal()
-prop = SessionLocal()
-
-def get_props(ref_id):
-    props = {}
-    sql = """select prop_name, prop_value,prop_title from kg_props where ref_id=:ref_id"""
-    result = prop.execute(text(sql), {'ref_id':ref_id})
-    for record in result:
-        prop_name, prop_value,prop_title = record
-        props[prop_name] = prop_title + ":" +prop_value
-    return props
-
-def get_entities():
-    #COUNT_SQL = "select count(*) from kg_nodes where version=:version"
-    COUNT_SQL = "select count(*) from kg_nodes where status=0"
-    result = db.execute(text(COUNT_SQL))
-    count = result.scalar()
-
-    print("total nodes: ", count)
-    entities = []
-    batch = 100
-    start = 1
-    while start < count:    
-        #sql = """select id,name,category from kg_nodes where version=:version order by id limit :batch OFFSET :start"""
-        sql = """select id,name,category from kg_nodes where status=0 order by id limit :batch OFFSET :start"""
-        result = db.execute(text(sql), {'start':start, 'batch':batch})
-        #["发热",{"type":"症状","description":"发热是诊断的主要目的,用于明确发热病因。"}]
-        row_count = 0
-        for row in result:
-            id,name,category = row
-            props = get_props(id)
-            
-            entities.append([id,{"name":name, 'type':category,'description':'', **props}])
-            row_count += 1
-        if row_count == 0:
-            break
-        start = start + row_count
-        print("start: ", start, "row_count: ", row_count)
-
-    with open(current_path+"\\entities_med.json", "w", encoding="utf-8") as f:
-        f.write(json.dumps(entities, ensure_ascii=False,indent=4))
-
-def get_names(src_id, dest_id):
-    sql = """select id,name,category from kg_nodes where id = :src_id"""
-    result = db.execute(text(sql), {'src_id':src_id}).first()
-    print(result)
-    if result is None:
-        #返回空
-        return (src_id, "", "", dest_id, "", "")
-    id,src_name,src_category = result
-    result = db.execute(text(sql), {'src_id':dest_id}).first()
-    id,dest_name,dest_category  = result
-    return (src_id, src_name, src_category, dest_id, dest_name, dest_category)
-
-def get_relationships():
-    #COUNT_SQL = "select count(*) from kg_edges where version=:version"
-    COUNT_SQL = "select count(*) from kg_edges"
-    result = db.execute(text(COUNT_SQL))
-    count = result.scalar()
-
-    print("total edges: ", count)
-    edges = []
-    batch = 1000
-    start = 1
-    file_index = 1
-    while start < count:    
-        #sql = """select id,name,category,src_id,dest_id from kg_edges where version=:version order by id limit :batch OFFSET :start"""
-        sql = """select id,name,category,src_id,dest_id from kg_edges order by id limit :batch OFFSET :start"""
-        result = db.execute(text(sql), {'start':start, 'batch':batch})
-        #["发热",{"type":"症状","description":"发热是诊断的主要目的,用于明确发热病因。"}]
-        row_count = 0
-        for row in result:
-            id,name,category,src_id,dest_id = row
-            props = get_props(id)
-            src_id, src_name, src_category, dest_id, dest_name, dest_category = get_names(src_id, dest_id)
-            #src_name或dest_name为空，说明节点不存在，跳过
-            if src_name == "" or dest_name == "":
-                continue
-            edges.append([src_id, {"id":src_id, "name":src_name, "type":src_category}, dest_id,{"id":dest_id,"name":dest_name,"type":dest_category}, {'type':category,'name':name, **props}])
-            row_count += 1
-        if row_count == 0:
-            break
-        start = start + row_count
-        print("start: ", start, "row_count: ", row_count)
-        if len(edges) > 10000:
-            with open(current_path+f"\\relationship_med_{file_index}.json", "w", encoding="utf-8") as f:
-                f.write(json.dumps(edges, ensure_ascii=False,indent=4))
-            edges = []
-            file_index += 1
-
-    with open(current_path+"\\relationship_med_0.json", "w", encoding="utf-8") as f:
-        f.write(json.dumps(edges, ensure_ascii=False,indent=4))
-
-#导出节点数据
-get_entities()
-#导出关系数据
-get_relationships()

community/graph_helper.py

@@ -1,319 +0,0 @@
-import networkx as nx
-import json
-from tabulate import tabulate
-import leidenalg
-import igraph as ig
-import sys,os
-from db.session import get_db
-from service.kg_node_service import KGNodeService
-
-current_path = os.getcwd()
-sys.path.append(current_path)
-
-RESOLUTION = 0.07
-# 图谱数据缓存路径（由dump_graph_data.py生成）
-CACHED_DATA_PATH = os.path.join(current_path, 'community', 'web', 'cached_data')
-
-
-def load_entity_data():
-    print("load entity data")
-    if not os.path.exists(os.path.join(CACHED_DATA_PATH,'entities_med.json')):
-        return []
-    with open(os.path.join(CACHED_DATA_PATH,'entities_med.json'), "r", encoding="utf-8") as f:
-        entities = json.load(f)
-        return entities
-
-def load_relation_data(g):
-    for i in range(0):
-        if not os.path.exists(os.path.join(CACHED_DATA_PATH, f"relationship_med_{i}.json")):
-            continue
-        if os.path.exists(os.path.join(CACHED_DATA_PATH, f"relationship_med_{i}.json")):
-            print("load entity data", os.path.join(CACHED_DATA_PATH, f"relationship_med_{i}.json"))
-            with open(os.path.join(CACHED_DATA_PATH, f"relationship_med_{i}.json"), "r", encoding="utf-8") as f:
-                relations = json.load(f)
-                for item in relations:
-
-                    if item[0] is None or item[1] is None or item[2] is None:
-                        continue
-                    #删除item[2]['weight']属性
-                    if 'weight' in item[2]:
-                        del item[2]['weight']
-                    g.add_edge(item[0], item[1], weight=1, **item[2])
-
-
-        
-
-class GraphHelper:
-    def __init__(self):
-        self.graph = None
-        self.build_graph()
-
-    def build_graph(self):
-        """构建企业关系图谱"""
-        self.graph = nx.Graph()
-        
-        # 加载节点数据
-        entities = load_entity_data()
-        for item in entities:
-            node_id = item[0]
-            attrs = item[1]
-            self.graph.add_node(node_id, **attrs)
-        
-        # 加载边数据
-        load_relation_data(self.graph)
-    
-    def community_report_search(self, query):
-        """社区报告检索功能"""
-        es_result = self.es.search_title_index("graph_community_report_index", query, 10)
-        results = []
-        for item in es_result:
-            results.append({ 
-                            'id': item["title"],               
-                            'score': item["score"],               
-                            'text': item["text"]})        
-        return results
-        
-    def node_search(self, node_id=None, node_type=None, filters=None, limit=1, min_degree=None):
-        """节点检索功能"""
-
-        kg_node_service = KGNodeService(next(get_db()))
-        es_result = kg_node_service.search_title_index("graph_entity_index", node_id, limit)
-        results = []
-        for item in es_result:
-            n = self.graph.nodes.get(item["title"])
-            score = item["score"]
-            if n:
-                results.append({
-                    'id': item["title"],
-                    'score': score,
-                    **n
-                })
-        return results
-        
-        # for n in self.graph.nodes(data=True):
-        #     match = True
-        #     if node_id and n[0] != node_id:
-        #         continue
-        #     if node_type and n[1].get('type') != node_type:
-        #         continue
-        #     if filters:
-        #         for k, v in filters.items():
-        #             if n[1].get(k) != v:
-        #                 match = False
-        #                 break
-        #     if match:
-        #         results.append({
-        #             'id': n[0],
-        #             **n[1]
-        #         })
-        return results
-
-    def edge_search(self, source=None, target=None, edge_type=None, min_weight=0):
-        """边检索功能"""
-        results = []
-        
-        for u, v, data in self.graph.edges(data=True):
-            if edge_type and data.get('type') != edge_type:
-                continue
-            if data.get('weight', 0) < min_weight:
-                continue
-            if (source and u != source and v != source) or \
-            (target and u != target and v != target):
-                continue
-              
-            results.append({
-                'source': u,
-                'target': v,
-                **data
-            })
-        return results
-
-    def neighbor_search(self, node_id, hops=1):
-        """近邻检索功能"""
-        if node_id not in self.graph:
-            return [],[]
-        
-        # 使用ego_graph获取指定跳数的子图
-        subgraph = nx.ego_graph(self.graph, node_id, radius=hops)
-        
-        entities = []
-        for n in subgraph.nodes(data=True):
-            if n[0] == node_id:  # 跳过中心节点
-                continue
-            entities.append({
-                'id': n[0],
-                **n[1]
-            })
-        relations = []
-        for edge in subgraph.edges(data=True):
-            relations.append({
-                'src_name': edge[0],
-                'dest_name': edge[1],
-                **edge[2]
-            })
-        return entities, relations
-
-    def find_paths(self, source, target, max_depth=3):
-        """路径查找功能"""
-        try:
-            shortest = nx.shortest_path(self.graph, source=source, target=target)
-            all_paths = nx.all_simple_paths(self.graph, source=source, target=target, cutoff=max_depth)
-            return {
-                'shortest_path': shortest,
-                'all_paths': list(all_paths)
-            }
-        except nx.NetworkXNoPath:
-            return {'error': 'No path found'}
-        except nx.NodeNotFound as e:
-            return {'error': f'Node not found: {e}'}
-
-    def format_output(self, data, fmt='text'):
-        """格式化输出结果"""
-        if fmt == 'json':
-            return json.dumps(data, indent=2, ensure_ascii=False)
-        
-        # 文本表格格式
-        if isinstance(data, list):
-            rows = []
-            headers = []
-            if not data:
-                return "No results found"
-            # 节点结果
-            if 'id' in data[0]:
-                headers = ["ID", "Type", "Description"]
-                rows = [[d['id'], d.get('type',''), d.get('description','')] for d in data]
-            # 边结果
-            elif 'source' in data[0]:
-                headers = ["Source", "Target", "Type", "Weight"]
-                rows = [[d['source'], d['target'], d.get('type',''), d.get('weight',0)] for d in data]
-
-            return tabulate(rows, headers=headers, tablefmt="grid")
-        
-        # 路径结果
-        if isinstance(data, dict):
-            if 'shortest_path' in data:
-                output = [
-                    "Shortest Path: " + " → ".join(data['shortest_path']),
-                    "\nAll Paths:"
-                ]
-                for path in data['all_paths']:
-                    output.append(" → ".join(path))
-                return "\n".join(output)
-            elif 'error' in data:
-                return data['error']
-        
-        return str(data)
-    
-    def detect_communities(self):
-        """使用Leiden算法进行社区检测"""
-        # 转换networkx图到igraph格式
-
-        print("convert to igraph")
-        ig_graph = ig.Graph.from_networkx(self.graph)
-        
-        # 执行Leiden算法
-        partition = leidenalg.find_partition(
-            ig_graph, 
-            leidenalg.CPMVertexPartition,
-            resolution_parameter=RESOLUTION,
-            n_iterations=2
-        )
-        
-        # 将社区标签添加到原始图
-        for i, node in enumerate(self.graph.nodes()):
-            self.graph.nodes[node]['community'] = partition.membership[i]
-        
-        print("convert to igraph finished")
-        return self.graph, partition
-
-    def filter_nodes(self, node_type=None, min_degree=0, attributes=None):
-        """根据条件过滤节点"""
-        filtered = []
-        for node, data in self.graph.nodes(data=True):
-            if node_type and data.get('type') != node_type:
-                continue
-            if min_degree > 0 and self.graph.degree(node) < min_degree:
-                continue
-            if attributes:
-                if not all(data.get(k) == v for k, v in attributes.items()):
-                    continue
-            filtered.append({'id': node, **data})
-        return filtered
-
-    def get_graph_statistics(self):
-        """获取图谱统计信息"""
-        return {
-            'node_count': self.graph.number_of_nodes(),
-            'edge_count': self.graph.number_of_edges(),
-            'density': nx.density(self.graph),
-            'components': nx.number_connected_components(self.graph),
-            'average_degree': sum(dict(self.graph.degree()).values()) / self.graph.number_of_nodes()
-        }
-
-    def get_community_details(self, community_id=None):
-        """获取社区详细信息"""
-        communities = {}
-        for node, data in self.graph.nodes(data=True):
-            comm = data.get('community', -1)
-            if community_id is not None and comm != community_id:
-                continue
-            if comm not in communities:
-                communities[comm] = {
-                    'node_count': 0,
-                    'nodes': [],
-                    'central_nodes': []
-                }
-            communities[comm]['node_count'] += 1
-            communities[comm]['nodes'].append(node)
-        
-        # 计算每个社区的中心节点
-        for comm in communities:
-            subgraph = self.graph.subgraph(communities[comm]['nodes'])
-            centrality = nx.degree_centrality(subgraph)
-            top_nodes = sorted(centrality.items(), key=lambda x: -x[1])[:3]
-            communities[comm]['central_nodes'] = [n[0] for n in top_nodes]
-        
-        return communities
-
-    def find_relations(self, node_ids, relation_types=None):
-        """查找指定节点集之间的关系"""
-        relations = []
-        for u, v, data in self.graph.edges(data=True):
-            if (u in node_ids or v in node_ids) and \
-               (not relation_types or data.get('type') in relation_types):
-                relations.append({
-                    'source': u,
-                    'target': v,
-                    **data
-                })
-        return relations
-
-    def semantic_search(self, query, top_k=5):
-        """语义搜索（需要与文本嵌入结合）"""
-        # 这里需要调用文本处理模块的embedding功能
-        # 示例实现：简单名称匹配
-        results = []
-        query_lower = query.lower()
-        for node, data in self.graph.nodes(data=True):
-            if query_lower in data.get('name', '').lower():
-                results.append({
-                    'id': node,
-                    'score': 1.0,
-                    **data
-                })
-        return sorted(results, key=lambda x: -x['score'])[:top_k]
-
-    def get_node_details(self, node_id):
-        """获取节点详细信息及其关联"""
-        if node_id not in self.graph:
-            return None
-        details = dict(self.graph.nodes[node_id])
-        details['degree'] = self.graph.degree(node_id)
-        details['neighbors'] = list(self.graph.neighbors(node_id))
-        details['edges'] = []
-        for u, v, data in self.graph.edges(node_id, data=True):
-            details['edges'].append({
-                'target': v if u == node_id else u,
-                **data
-            })
-        return details

community/graph_helper2.bak

@@ -1,273 +0,0 @@
-"""
-医疗知识图谱助手模块
-
-本模块提供构建医疗知识图谱、执行社区检测、路径查找等功能
-
-主要功能:
-1. 构建医疗知识图谱
-2. 支持节点/关系检索
-3. 社区检测
-4. 路径查找
-5. 邻居分析
-"""
-import networkx as nx
-import argparse
-import json
-from tabulate import tabulate
-import leidenalg
-import igraph as ig
-import sys,os
-
-from db.session import get_db
-from service.kg_node_service import KGNodeService
-
-# 当前工作路径
-current_path = os.getcwd()
-sys.path.append(current_path)
-
-# Leiden算法社区检测的分辨率参数，控制社区划分的粒度
-RESOLUTION = 0.07
-
-# 图谱数据缓存路径（由dump_graph_data.py生成）
-CACHED_DATA_PATH = os.path.join(current_path, 'community', 'web', 'cached_data')
-
-def load_entity_data():
-    """
-    加载实体数据
-    
-    返回:
-        list: 实体数据列表，每个元素格式为[node_id, attributes_dict]
-    """
-    print("load entity data")
-    with open(os.path.join(CACHED_DATA_PATH,'entities_med.json'), "r", encoding="utf-8") as f:
-        entities = json.load(f)
-        return entities
-
-def load_relation_data(g):
-    """
-    分块加载关系数据
-    
-    参数:
-        g (nx.Graph): 要添加边的NetworkX图对象
-    
-    说明:
-        1. 支持分块加载多个关系文件（relationship_med_0.json ~ relationship_med_29.json）
-        2. 每个关系项格式为[source, target, {relation_attrs}]
-    """
-    for i in range(89):
-        if os.path.exists(os.path.join(CACHED_DATA_PATH, f"relationship_med_{i}.json")):
-            print("load entity data", os.path.join(CACHED_DATA_PATH, f"relationship_med_{i}.json"))
-            with open(os.path.join(CACHED_DATA_PATH, f"relationship_med_{i}.json"), "r", encoding="utf-8") as f:
-       
-                relations = json.load(f)
-                for item in relations:
-                    # 添加带权重的边，并存储关系属性
-                    weight = int(item[2].pop('weight', '8').replace('权重:', ''))
-                    #如果item[0]或者item[1]为空或null，则跳过
-                    if item[0] is None or item[1] is None:
-                        continue
-                    g.add_edge(item[0], item[1], weight=weight, **item[2])
-
-class GraphHelper:
-    """
-    医疗知识图谱助手类
-    
-    功能:
-        - 构建医疗知识图谱
-        - 支持节点/关系检索
-        - 社区检测
-        - 路径查找
-        - 邻居分析
-    
-    属性:
-        graph: NetworkX图对象，存储知识图谱
-    """
-    
-    def __init__(self):
-        """
-        初始化方法
-        
-        功能:
-            1. 初始化graph属性为None
-            2. 调用build_graph()方法构建知识图谱
-        """
-        self.graph = None
-        self.build_graph()
-
-    def build_graph(self):
-        """构建知识图谱
-        
-        步骤:
-            1. 初始化空图
-            2. 加载实体数据作为节点
-            3. 加载关系数据作为边
-        """
-        self.graph = nx.Graph()
-        
-        # 加载节点数据（疾病、症状等）
-        entities = load_entity_data()
-        for item in entities:
-            node_id = item[0]
-            attrs = item[1]
-            self.graph.add_node(node_id, **attrs)
-        
-        # 加载边数据（疾病-症状关系等）
-        load_relation_data(self.graph)
-
-    def node_search2(self, node_id=None, node_type=None, filters=None, limit=1, min_degree=None):
-        """节点检索功能"""
-
-        kg_node_service = KGNodeService(next(get_db()))
-        es_result = kg_node_service.search_title_index("graph_entity_index", node_id, limit)
-        results = []
-        for item in es_result:
-            n = self.graph.nodes.get(item["title"])
-            score = item["score"]
-            if n:
-                results.append({
-                    'id': item["title"],
-                    'score': score,
-                    **n
-                })
-        return results
-
-    def node_search(self, node_id=None, node_type=None, filters=None):
-        """节点检索
-        
-        参数:
-            node_id (str): 精确匹配节点ID
-            node_type (str): 按节点类型过滤
-            filters (dict): 自定义属性过滤，格式为{属性名: 期望值}
-            
-        返回:
-            list: 匹配的节点列表，每个节点包含id和所有属性
-        """
-        results = []
-        
-        # 遍历所有节点进行多条件过滤
-        for n in self.graph.nodes(data=True):
-            match = True
-            if node_id and n[0] != node_id:
-                continue
-            if node_type and n[1].get('type') != node_type:
-                continue
-            if filters:
-                for k, v in filters.items():
-                    if n[1].get(k) != v:
-                        match = False
-                        break
-            if match:
-                results.append({
-                    'id': n[0],
-                    **n[1]
-                })
-        return results
-
-    def neighbor_search(self, center_node, hops=2):
-        """邻居节点检索
-        
-        参数:
-            center_node (str): 中心节点ID
-            hops (int): 跳数（默认2跳）
-            
-        返回:
-            tuple: (邻居实体列表, 关联关系列表)
-            
-        算法说明:
-            使用BFS算法进行层级遍历，时间复杂度O(k^d)，其中k为平均度数，d为跳数
-        """
-        # 执行BFS遍历
-        visited = {center_node: 0}
-        queue = [center_node]
-        relations = []
-        
-        while queue:
-            try:
-                current = queue.pop(0)
-                current_hop = visited[current]
-                
-                if current_hop >= hops:
-                    continue
-                
-                # 遍历相邻节点
-                for neighbor in self.graph.neighbors(current):
-                    if neighbor not in visited:
-                        visited[neighbor] = current_hop + 1
-                        queue.append(neighbor)
-                    
-                    # 记录边关系
-                    edge_data = self.graph.get_edge_data(current, neighbor)
-                    relations.append({
-                        'src_name': current,
-                        'dest_name': neighbor,
-                        **edge_data
-                    })
-            except Exception as e:
-                print(f"Error processing node {current}: {str(e)}")
-                continue
-        
-        # 提取邻居实体（排除中心节点）
-        entities = [
-            {'id': n, **self.graph.nodes[n]}
-            for n in visited if n != center_node
-        ]
-        
-        return entities, relations
-
-    def detect_communities(self):
-        """使用Leiden算法进行社区检测
-        
-        返回:
-            tuple: (添加社区属性的图对象, 社区划分结果)
-            
-        算法说明:
-            1. 将NetworkX图转换为igraph格式
-            2. 使用Leiden算法（分辨率参数RESOLUTION=0.07）
-            3. 将社区标签添加回原始图
-            4. 时间复杂度约为O(n log n)
-        """
-        # 转换图格式
-        ig_graph = ig.Graph.from_networkx(self.graph)
-        
-        # 执行Leiden算法
-        partition = leidenalg.find_partition(
-            ig_graph, 
-            leidenalg.CPMVertexPartition,
-            resolution_parameter=RESOLUTION,
-            n_iterations=2
-        )
-        
-        # 添加社区属性
-        for i, node in enumerate(self.graph.nodes()):
-            self.graph.nodes[node]['community'] = partition.membership[i]
-        
-        return self.graph, partition
-
-    def find_paths(self, source, target, max_paths=5):
-        """查找所有简单路径
-        
-        参数:
-            source (str): 起始节点
-            target (str): 目标节点
-            max_paths (int): 最大返回路径数
-            
-        返回:
-            dict: 包含最短路径和所有路径的结果字典
-            
-        注意:
-            使用Yen算法寻找top k最短路径，时间复杂度O(kn(m + n log n))
-        """
-        result = {'shortest_path': [], 'all_paths': []}
-        
-        try:
-            # 使用Dijkstra算法找最短路径
-            shortest_path = nx.shortest_path(self.graph, source, target, weight='weight')
-            result['shortest_path'] = shortest_path
-            
-            # 使用Yen算法找top k路径
-            all_paths = list(nx.shortest_simple_paths(self.graph, source, target, weight='weight'))[:max_paths]
-            result['all_paths'] = all_paths
-        except nx.NetworkXNoPath:
-            pass
-            
-        return result

db/session.py

@@ -11,7 +11,7 @@ DB_HOST = os.getenv("DB_HOST", "173.18.12.203")
 DB_PORT = os.getenv("DB_PORT", "5432")
 DB_USER = os.getenv("DB_USER", "knowledge")
 DB_PASS = os.getenv("DB_PASSWORD", "qwer1234.")
-DB_NAME = os.getenv("DB_NAME", "postgres")
+DB_NAME = os.getenv("DB_NAME", "medkg")
 
 DATABASE_URL = f"postgresql://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
 

main.py

@@ -6,11 +6,7 @@ from typing import Optional, Set
 # 导入FastAPI及相关模块
 import os
 import uvicorn
-from router.knowledge_dify import dify_kb_router
-from router.knowledge_saas import saas_kb_router
-from router.text_search import text_search_router
-from router.graph_router import graph_router
-from router.knowledge_nodes_api import knowledge_nodes_api_router
+from router.nodes_api import nodes_api_router
 
 # 配置日志
 logging.basicConfig(
@@ -26,11 +22,7 @@ logger.propagate = True
 
 # 创建FastAPI应用
 app = FastAPI(title="知识图谱")
-app.include_router(dify_kb_router)
-app.include_router(saas_kb_router)
-app.include_router(text_search_router)
-app.include_router(graph_router)
-app.include_router(knowledge_nodes_api_router)
+app.include_router(nodes_api_router)
 
 
 # 需要拦截的 URL 列表（支持通配符）
@@ -56,7 +48,7 @@ async def verify_token(authorization: str) -> Optional[dict]:
     token = authorization[7:]
     # 这里添加实际的 token 验证逻辑
     # 示例：简单验证 token 是否等于 secret-token
-    if token == "secret-token":
+    if token == "3xY7-p9Kq-2FmR-8LzN":
         return {"id": 1, "username": "admin", "role": "admin"}
     return None
 
@@ -121,5 +113,5 @@ async def interceptor_middleware(request: Request, call_next):
 if __name__ == "__main__":
     logger.info('Starting uvicorn server...2222')
     #uvicorn main:app --host 0.0.0.0 --port 8000 --reload
-    uvicorn.run("main:app", host="0.0.0.0", port=8001, reload=False)
+    uvicorn.run("main:app", host="0.0.0.0", port=8008, reload=False)
 

model/trunks_model.py

@@ -1,20 +0,0 @@
-from sqlalchemy import Column, Integer, Text, String
-from sqlalchemy.dialects.postgresql import TSVECTOR
-from pgvector.sqlalchemy import Vector
-from db.base_class import Base
-
-class Trunks(Base):
-    __tablename__ = 'trunks'
-
-    id = Column(Integer, primary_key=True, index=True)
-    embedding = Column(Vector(1024))
-    content = Column(Text)
-    file_path = Column(String(255))
-    content_tsvector = Column(TSVECTOR)
-    type = Column(String(255))
-    title = Column(String(255))
-    referrence = Column(String(255))
-    meta_header = Column(String(255))
-
-    def __repr__(self):
-        return f"<Trunks(id={self.id}, file_path={self.file_path})>"

router/graph_router.py

@@ -1,184 +0,0 @@
-import sys,os
-
-from community.graph_helper import GraphHelper
-from model.response import StandardResponse
-
-current_path = os.getcwd()
-sys.path.append(current_path)
-import time
-from fastapi import APIRouter, Depends, Query
-from typing import Optional, List
-
-import sys
-sys.path.append('..')
-from utils.agent import call_chat_api,get_conversation_id
-import json
-
-
-
-router = APIRouter(prefix="/graph", tags=["Knowledge Graph"])
-graph_helper = GraphHelper()
-
-
-
-@router.get("/nodes/recommend", response_model=StandardResponse)
-async def recommend(
-    chief: str
-):
-    app_id = "256fd853-60b0-4357-b11b-8114b4e90ae0"
-    conversation_id = get_conversation_id(app_id)
-    result = call_chat_api(app_id, conversation_id, chief)
-    json_data = json.loads(result)
-    keyword = " ".join(json_data["chief_complaint"])
-    return await neighbor_search(keyword=keyword, neighbor_type='Check',limit=10)
-
-
-@router.get("/nodes/neighbor_search", response_model=StandardResponse)
-async def neighbor_search(
-    keyword: str = Query(..., min_length=2),
-    limit: int = Query(10, ge=1, le=100),
-    node_type: Optional[str] = Query(None),
-    neighbor_type: Optional[str] = Query(None),
-    min_degree: Optional[int] = Query(None)
-):
-    """
-    根据关键词和属性过滤条件搜索图谱节点
-    """
-    try:
-        start_time = time.time()
-        print(f"开始执行neighbor_search，参数：keyword={keyword}, limit={limit}, node_type={node_type}, neighbor_type={neighbor_type}, min_degree={min_degree}")
-        scores_factor = 1.7
-        results = []
-        diseases = {}
-
-        has_good_result = False
-
-        if not has_good_result:
-            keywords = keyword.split(" ")
-            new_results = []
-            for item in keywords:
-                if len(item) > 1:
-                    results = graph_helper.node_search2(
-                        item,
-                        limit=limit,
-                        node_type=node_type,
-                        min_degree=min_degree
-                    )
-
-                    for result_item in results:
-                        if result_item["score"] > scores_factor:
-                            new_results.append(result_item)
-                            if result_item["type"] == "Disease":
-                                if result_item["id"] not in diseases:
-                                    diseases[result_item["id"]] =  {
-                                                        "id":result_item["id"],
-                                                        "type":1,
-                                                        "count":1
-                                                    }
-                                else:
-                                    diseases[result_item["id"]]["count"] = diseases[result_item["id"]]["count"] + 1
-                                has_good_result = True
-            results = new_results
-            print("扩展搜索的结果数量:",len(results))
-
-        neighbors_data = {}
-
-        for item in results:
-            entities, relations = graph_helper.neighbor_search(item["id"], 1)
-            max = 20 #因为类似发热这种疾病会有很多关联的疾病，所以需要防止检索范围过大，设置了上限
-            for neighbor in entities:
-
-                #有时候数据中会出现链接有的节点，但是节点数据中缺失的情况，所以这里要检查
-                if "type" not in neighbor.keys():
-                    continue
-                if neighbor["type"] == neighbor_type:
-                    #如果这里正好找到了要求检索的节点类型
-                    if neighbor["id"] not in neighbors_data:
-                        neighbors_data[neighbor["id"]] =  {
-                                            "id":neighbor["id"],
-                                            "type":neighbor["type"],
-                                            "count":1
-                                        }
-                    else:
-                         neighbors_data[neighbor["id"]]["count"] = neighbors_data[neighbor["id"]]["count"] + 1
-                else:
-                    #如果这里找到的节点是个疾病，那么就再检索一层，看看是否有符合要求的节点类型
-                    if neighbor["type"] == "Disease":
-                        if neighbor["id"] not in diseases:
-                            diseases[neighbor["id"]] =  {
-                                                "id":neighbor["id"],
-                                                "type":"Disease",
-                                                "count":1
-                                            }
-                        else:
-                            diseases[neighbor["id"]]["count"] = diseases[neighbor["id"]]["count"] + 1
-                        disease_entities, relations = graph_helper.neighbor_search(neighbor["id"], 1)
-                        for disease_neighbor in disease_entities:
-                            #有时候数据中会出现链接有的节点，但是节点数据中缺失的情况，所以这里要检查
-                            if "type" in disease_neighbor.keys():
-                                if disease_neighbor["type"] == neighbor_type:
-                                    if disease_neighbor["id"] not in neighbors_data:
-                                        neighbors_data[disease_neighbor["id"]] = {
-                                            "id":disease_neighbor["id"],
-                                            "type":disease_neighbor["type"],
-                                            "count":1
-                                        }
-                                    else:
-                                        neighbors_data[disease_neighbor["id"]]["count"] = neighbors_data[disease_neighbor["id"]]["count"] + 1
-                        #最多搜索的范围是max个疾病
-                        max = max - 1
-                        if max == 0:
-                            break
-        disease_data = [diseases[k] for k in diseases]
-        disease_data = sorted(disease_data, key=lambda x:x["count"],reverse=True)
-        data = [neighbors_data[k] for k in neighbors_data if neighbors_data[k]["type"] == "Check"]
-
-        data = sorted(data, key=lambda x:x["count"],reverse=True)
-
-        if len(data) > 10:
-            data = data[:10]
-            factor = 1.0
-            total = 0.0
-            for item in data:
-                total = item["count"] * factor + total
-            for item in data:
-                item["count"] = item["count"] / total
-            factor = factor * 0.9
-
-        if len(disease_data) > 10:
-            disease_data = disease_data[:10]
-            factor = 1.0
-            total = 0.0
-            for item in disease_data:
-                total = item["count"] * factor + total
-            for item in disease_data:
-                item["count"] = item["count"] / total
-            factor = factor * 0.9
-
-        for item in data:
-            item["type"] = 3
-            item["name"] = item["id"]
-            item["rate"] = round(item["count"] * 100, 2)
-        for item in disease_data:
-            item["type"] = 1
-            item["name"] = item["id"]
-            item["rate"] = round(item["count"] * 100, 2)
-        end_time = time.time()
-        print(f"neighbor_search执行完成，耗时：{end_time - start_time:.2f}秒")
-
-        return StandardResponse(
-            success=True,
-            records={"可能诊断":disease_data,"推荐检验":data},
-            error_code = 0,
-            error_msg=f"Found {len(results)} nodes"
-        )
-    except Exception as e:
-        print(e)
-        raise e
-        return StandardResponse(
-            success=False,
-            error_code=500,
-            error_msg=str(e)
-        )
-
-graph_router = router

router/knowledge_dify.py

@@ -1,156 +0,0 @@
-from fastapi import APIRouter, UploadFile, File, HTTPException, Depends, Header, Request
-from typing import List, Optional
-from pydantic import BaseModel, Field, validator
-from model.response import StandardResponse
-
-from db.session import get_db
-from sqlalchemy.orm import Session
-from service.trunks_service import TrunksService
-import logging
-
-router = APIRouter(prefix="/dify", tags=["Dify Knowledge Base"])
-
-
-# --- Data Models ---
-class RetrievalSetting(BaseModel):
-    top_k: int
-    score_threshold: float
-
-class MetadataCondition(BaseModel):
-    name: List[str]
-    comparison_operator: str = Field(..., pattern=r'^(equals|not_equals|contains|not_contains|starts_with|ends_with|empty|not_empty|greater_than|less_than)$')
-    value: Optional[str] = Field(None)
-
-    @validator('value')
-    def validate_value(cls, v, values):
-        operator = values.get('comparison_operator')
-        if operator in ['empty', 'not_empty'] and v is not None:
-            raise ValueError('Value must be None for empty/not_empty operators')
-        if operator not in ['empty', 'not_empty'] and v is None:
-            raise ValueError('Value is required for this comparison operator')
-        return v
-
-class MetadataFilter(BaseModel):
-    logical_operator: str = Field(default="and", pattern=r'^(and|or)$')
-    conditions: List[MetadataCondition] = Field(..., min_items=1)
-
-    @validator('conditions')
-    def validate_conditions(cls, v):
-        if len(v) < 1:
-            raise ValueError('At least one condition is required')
-        return v
-
-class DifyRetrievalRequest(BaseModel):
-    knowledge_id: str
-    query: str
-    retrieval_setting: RetrievalSetting
-    metadata_condition: Optional[MetadataFilter] = Field(default=None, exclude=True)
-
-class KnowledgeRecord(BaseModel):
-    content: str
-    score: float
-    title: str
-    metadata: dict
-
-# --- Authentication ---
-async def verify_api_key(authorization: str = Header(...)):
-    logger.info(f"Received authorization header: {authorization}")  # 新增日志
-    if not authorization.startswith("Bearer "):
-        raise HTTPException(
-            status_code=403,
-            detail=StandardResponse(
-                success=False,
-                error_code=1001,
-                error_msg="Invalid Authorization header format"
-            )
-        )
-    api_key = authorization[7:]
-    # TODO: Implement actual API key validation logic
-    if not api_key:
-        raise HTTPException(
-            status_code=403,
-            detail=StandardResponse(
-                success=False,
-                error_code=1002,
-                error_msg="Authorization failed"
-            )
-        )
-    return api_key
-
-logger = logging.getLogger(__name__)
-
-@router.post("/retrieval", response_model=StandardResponse)
-async def dify_retrieval(
-    payload: DifyRetrievalRequest,
-    request: Request,
-    authorization: str = Depends(verify_api_key),
-    db: Session = Depends(get_db),
-    conversation_id: Optional[str] = None
-):
-    logger.info(f"All headers: {dict(request.headers)}")
-    logger.info(f"Request body: {payload.model_dump()}")
-    
-    try:
-        logger.info(f"Starting retrieval for knowledge base {payload.knowledge_id} with query: {payload.query}")
-        
-        trunks_service = TrunksService()
-        search_results = trunks_service.search_by_vector(payload.query, payload.retrieval_setting.top_k, conversation_id=conversation_id)
-        
-        if not search_results:
-            logger.warning(f"No results found for query: {request.query}")
-            return StandardResponse(
-                success=True,
-                records=[]
-            )
-        
-        # 格式化返回结果
-        records = [{
-            "metadata": {
-                "path": result["file_path"],
-                "description": str(result["id"])
-            },
-            "score": result["distance"],
-            "title": result["file_path"].split("/")[-1],
-            "content": result["content"]
-        } for result in search_results]
-        
-        logger.info(f"Retrieval completed successfully for query: {payload.query}")
-        return StandardResponse(
-            success=True,
-            records=records
-        )
-
-    except HTTPException as e:
-        logger.error(f"HTTPException occurred: {str(e)}")
-        raise
-    except Exception as e:
-        logger.error(f"Unexpected error occurred: {str(e)}")
-        raise HTTPException(
-            status_code=500,
-            detail=StandardResponse(
-                success=False,
-                error_code=500,
-                error_msg=str(e)
-            )
-        )
-
-@router.post("/chatflow_retrieval", response_model=StandardResponse)
-async def dify_chatflow_retrieval(
-    knowledge_id: str,
-    query: str,
-    top_k: int,
-    score_threshold: float,
-    conversation_id: str,
-    request: Request,
-    authorization: str = Depends(verify_api_key),
-    db: Session = Depends(get_db)
-):
-    payload = DifyRetrievalRequest(
-        knowledge_id=knowledge_id,
-        query=query,
-        retrieval_setting=RetrievalSetting(top_k=top_k, score_threshold=score_threshold)
-    )
-    return await dify_retrieval(payload, request, authorization, db, conversation_id=conversation_id)
-
-dify_kb_router = router
-

router/knowledge_saas.py

@@ -1,168 +0,0 @@
-from fastapi import APIRouter, Depends, HTTPException
-from typing import Optional, List
-from pydantic import BaseModel
-from model.response import StandardResponse
-from db.session import get_db
-from sqlalchemy.orm import Session
-
-from service.kg_node_service import KGNodeService
-from service.trunks_service import TrunksService
-from service.kg_edge_service import KGEdgeService
-from service.kg_prop_service import KGPropService
-import logging
-from utils.ObjectToJsonArrayConverter import ObjectToJsonArrayConverter
-
-router = APIRouter(prefix="/saas", tags=["SaaS Knowledge Base"])
-
-logger = logging.getLogger(__name__)
-
-class PaginatedSearchRequest(BaseModel):
-    keyword: Optional[str] = None
-    pageNo: int = 1
-    limit: int = 10
-    knowledge_ids: Optional[List[str]] = None
-
-class NodePaginatedSearchRequest(BaseModel):
-    name: str
-    category: Optional[str] = None
-    pageNo: int = 1
-    limit: int = 10  
-
-class NodeCreateRequest(BaseModel):
-    name: str
-    category: str
-    layout: Optional[str] = None
-    version: Optional[str] = None
-    embedding: Optional[List[float]] = None
-
-class NodeUpdateRequest(BaseModel):
-    layout: Optional[str] = None
-    version: Optional[str] = None
-    status: Optional[int] = None
-    embedding: Optional[List[float]] = None
-
-class VectorSearchRequest(BaseModel):
-    text: str
-    limit: int = 10
-    type: Optional[str] = None
-
-class NodeRelationshipRequest(BaseModel):
-    src_id: int
-
-@router.post("/nodes/paginated_search2", response_model=StandardResponse)
-async def paginated_search(
-    payload: PaginatedSearchRequest,
-    db: Session = Depends(get_db)
-):
-    try:
-        service = KGNodeService(db)
-        search_params = {
-            'keyword': payload.keyword,
-            'pageNo': payload.pageNo,
-            'limit': payload.limit,
-            'knowledge_ids': payload.knowledge_ids,
-            'load_props': True
-        }
-        result = service.paginated_search(search_params)
-        return StandardResponse(
-            success=True,
-            data={
-                'records': result['records'],
-                'pagination': result['pagination']
-            }
-        )
-    except Exception as e:
-        logger.error(f"分页查询失败: {str(e)}")
-        raise HTTPException(
-            status_code=500,
-            detail=StandardResponse(
-                success=False,
-                error_code=500,
-                error_msg=str(e)
-            )
-        )
-
-@router.post("/nodes", response_model=StandardResponse)
-async def create_node(
-    payload: NodeCreateRequest,
-    db: Session = Depends(get_db)
-):
-    try:
-        service = TrunksService()
-        result = service.create_node(payload.dict())
-        return StandardResponse(success=True, data=result)
-    except Exception as e:
-        logger.error(f"创建节点失败: {str(e)}")
-        raise HTTPException(500, detail=StandardResponse.error(str(e)))
-
-@router.get("/nodes/{node_id}", response_model=StandardResponse)
-async def get_node(
-    node_id: int,
-    db: Session = Depends(get_db)
-):
-    try:
-        service = TrunksService()
-        result = service.get_node(node_id)
-        return StandardResponse(success=True, data=result)
-    except Exception as e:
-        logger.error(f"获取节点失败: {str(e)}")
-        raise HTTPException(500, detail=StandardResponse.error(str(e)))
-
-@router.put("/nodes/{node_id}", response_model=StandardResponse)
-async def update_node(
-    node_id: int,
-    payload: NodeUpdateRequest,
-    db: Session = Depends(get_db)
-):
-    try:
-        service = TrunksService()
-        result = service.update_node(node_id, payload.dict(exclude_unset=True))
-        return StandardResponse(success=True, data=result)
-    except Exception as e:
-        logger.error(f"更新节点失败: {str(e)}")
-        raise HTTPException(500, detail=StandardResponse.error(str(e)))
-
-@router.delete("/nodes/{node_id}", response_model=StandardResponse)
-async def delete_node(
-    node_id: int,
-    db: Session = Depends(get_db)
-):
-    try:
-        service = TrunksService()
-        service.delete_node(node_id)
-        return StandardResponse(success=True)
-    except Exception as e:
-        logger.error(f"删除节点失败: {str(e)}")
-        raise HTTPException(500, detail=StandardResponse.error(str(e)))
-
-@router.post('/trunks/vector_search', response_model=StandardResponse)
-async def vector_search(
-    payload: VectorSearchRequest,
-    db: Session = Depends(get_db)
-):
-    try:
-        service = TrunksService()
-        result = service.search_by_vector(
-            payload.text,
-            payload.limit,
-            {'type': payload.type} if payload.type else None
-        )
-        return StandardResponse(success=True, data=result)
-    except Exception as e:
-        logger.error(f"向量搜索失败: {str(e)}")
-        raise HTTPException(500, detail=StandardResponse.error(str(e)))
-
-@router.get('/trunks/{trunk_id}', response_model=StandardResponse)
-async def get_trunk(
-    trunk_id: int,
-    db: Session = Depends(get_db)
-):
-    try:
-        service = TrunksService()
-        result = service.get_trunk_by_id(trunk_id)
-        return StandardResponse(success=True, data=result)
-    except Exception as e:
-        logger.error(f"获取trunk详情失败: {str(e)}")
-        raise HTTPException(500, detail=StandardResponse.error(str(e)))
-
-saas_kb_router = router

router/knowledge_nodes_api.py

@@ -112,4 +112,4 @@ async def get_node_relationships(
         logger.error(f"获取节点关系失败: {str(e)}")
         raise HTTPException(500, detail=StandardResponse.error(str(e)))
 
-knowledge_nodes_api_router = router
+nodes_api_router = router

router/text_search.py

@@ -1,305 +0,0 @@
-from fastapi import APIRouter, HTTPException
-from pydantic import BaseModel, Field, validator
-from typing import List, Optional
-from service.trunks_service import TrunksService
-from utils.text_splitter import TextSplitter
-from utils.vector_distance import VectorDistance
-from model.response import StandardResponse
-from utils.vectorizer import Vectorizer
-DISTANCE_THRESHOLD = 0.8
-import logging
-import time
-
-logger = logging.getLogger(__name__)
-router = APIRouter(prefix="/text", tags=["Text Search"])
-
-class TextSearchRequest(BaseModel):
-    text: str
-    conversation_id: Optional[str] = None
-    need_convert: Optional[bool] = False
-
-class TextCompareRequest(BaseModel):
-    sentence: str
-    text: str
-
-class TextMatchRequest(BaseModel):
-    text: str = Field(..., min_length=1, max_length=10000, description="需要搜索的文本内容")
-
-    @validator('text')
-    def validate_text(cls, v):
-        # 保留所有可打印字符、换行符和中文字符
-        v = ''.join(char for char in v if char.isprintable() or char in '\n\r')
-        
-        # 转义JSON特殊字符
-        # 先处理反斜杠，避免后续转义时出现问题
-        v = v.replace('\\', '\\\\')
-        # 处理引号和其他特殊字符
-        v = v.replace('"', '\\"')
-        v = v.replace('/', '\\/')
-        # 处理控制字符
-        v = v.replace('\n', '\\n')
-        v = v.replace('\r', '\\r')
-        v = v.replace('\t', '\\t')
-        v = v.replace('\b', '\\b')
-        v = v.replace('\f', '\\f')
-        # 处理Unicode转义
-        # v = v.replace('\u', '\\u')
-        
-        return v
-
-class TextCompareMultiRequest(BaseModel):
-    origin: str
-    similar: str
-
-@router.post("/search", response_model=StandardResponse)
-async def search_text(request: TextSearchRequest):
-    try:
-        #判断request.text是否为json格式，如果是，使用JsonToText的convert方法转换为text
-        if request.text.startswith('{') and request.text.endswith('}'):
-            from utils.json_to_text import JsonToTextConverter
-            converter = JsonToTextConverter()
-            request.text = converter.convert(request.text)
-
-        # 使用TextSplitter拆分文本
-        sentences = TextSplitter.split_text(request.text)
-        if not sentences:
-            return StandardResponse(success=True, data={"answer": "", "references": []})
-        
-        # 初始化服务和结果列表
-        trunks_service = TrunksService()
-        result_sentences = []
-        all_references = []
-        reference_index = 1
-        
-        # 根据conversation_id获取缓存结果
-        cached_results = trunks_service.get_cached_result(request.conversation_id) if request.conversation_id else []
-        
-        for sentence in sentences:
-            # if request.need_convert:
-            sentence = sentence.replace("\n", "<br>")
-            if len(sentence) < 10:
-                result_sentences.append(sentence)
-                continue
-            if cached_results:
-                # 如果有缓存结果，计算向量距离
-                min_distance = float('inf')
-                best_result = None
-                sentence_vector = Vectorizer.get_embedding(sentence)
-                
-                for cached_result in cached_results:
-                    content_vector = cached_result['embedding']
-                    distance = VectorDistance.calculate_distance(sentence_vector, content_vector)
-                    if distance < min_distance:
-                        min_distance = distance
-                        best_result = {**cached_result, 'distance': distance}
-                        
-                
-                if best_result and best_result['distance'] < DISTANCE_THRESHOLD:
-                    search_results = [best_result]
-                else:
-                    search_results = []
-            else:
-                # 如果没有缓存结果，进行向量搜索
-                search_results = trunks_service.search_by_vector(
-                    text=sentence,
-                    limit=1,
-                    type='trunk'
-                )
-            
-            # 处理搜索结果
-            for result in search_results:
-                distance = result.get("distance", DISTANCE_THRESHOLD)
-                if distance >= DISTANCE_THRESHOLD:
-                    result_sentences.append(sentence)
-                    continue
-                
-                # 检查是否已存在相同引用
-                existing_ref = next((ref for ref in all_references if ref["id"] == result["id"]), None)
-                current_index = reference_index
-                if existing_ref:
-                    current_index = int(existing_ref["index"])
-                else:
-                    # 添加到引用列表
-                    reference = {
-                        "index": str(reference_index),
-                        "id": result["id"],
-                        "content": result["content"],
-                        "file_path": result.get("file_path", ""),
-                        "title": result.get("title", ""),
-                        "distance": distance,
-                        "referrence": result.get("referrence", "")
-                    }
-                    all_references.append(reference)
-                    reference_index += 1
-                
-                # 添加引用标记
-                if sentence.endswith('<br>'):
-                    # 如果有多个<br>,在所有<br>前添加^[current_index]^
-                    result_sentence = sentence.replace('<br>', f'^[{current_index}]^<br>')
-                else:
-                    # 直接在句子末尾添加^[current_index]^
-                    result_sentence = f'{sentence}^[{current_index}]^'
-                
-                result_sentences.append(result_sentence)
-     
-        # 组装返回数据
-        response_data = {
-            "answer": result_sentences,
-            "references": all_references
-        }
-        
-        return StandardResponse(success=True, data=response_data)
-        
-    except Exception as e:
-        logger.error(f"Text search failed: {str(e)}")
-        raise HTTPException(status_code=500, detail=str(e))
-
-@router.post("/match", response_model=StandardResponse)
-async def match_text(request: TextCompareRequest):
-    try:
-        sentences = TextSplitter.split_text(request.text)
-        sentence_vector = Vectorizer.get_embedding(request.sentence)
-        min_distance = float('inf')
-        best_sentence = ""
-        result_sentences = []
-        for temp in sentences:
-            result_sentences.append(temp)
-            if len(temp) < 10:
-                continue
-            temp_vector = Vectorizer.get_embedding(temp)
-            distance = VectorDistance.calculate_distance(sentence_vector, temp_vector)
-            if distance < min_distance and distance < DISTANCE_THRESHOLD:
-                min_distance = distance
-                best_sentence = temp
-
-        for i in range(len(result_sentences)):
-            result_sentences[i] = {"sentence": result_sentences[i], "matched": False}
-            if result_sentences[i]["sentence"] == best_sentence:
-                result_sentences[i]["matched"] = True    
-                
-        return StandardResponse(success=True, records=result_sentences)
-    except Exception as e:
-        logger.error(f"Text comparison failed: {str(e)}")
-        raise HTTPException(status_code=500, detail=str(e))
-
-@router.post("/mr_search", response_model=StandardResponse)
-async def mr_search_text_content(request: TextMatchRequest):
-    try:
-        # 初始化服务
-        trunks_service = TrunksService()
-        
-        # 获取文本向量并搜索相似内容
-        search_results = trunks_service.search_by_vector(
-            text=request.text,
-            limit=10,
-            type="mr"
-        )
-
-        # 处理搜索结果
-        records = []
-        for result in search_results:
-            distance = result.get("distance", DISTANCE_THRESHOLD)
-            if distance >= DISTANCE_THRESHOLD:
-                continue
-
-            # 添加到引用列表
-            record = {
-                "content": result["content"],
-                "file_path": result.get("file_path", ""),
-                "title": result.get("title", ""),
-                "distance": distance,
-            }
-            records.append(record)
-
-        # 组装返回数据
-        response_data = {
-            "records": records
-        }
-        
-        return StandardResponse(success=True, data=response_data)
-        
-    except Exception as e:
-        logger.error(f"Mr search failed: {str(e)}")
-        raise HTTPException(status_code=500, detail=str(e))
-
-@router.post("/mr_match", response_model=StandardResponse)
-async def compare_text(request: TextCompareMultiRequest):
-    start_time = time.time()
-    try:
-        # 拆分两段文本
-        origin_sentences = TextSplitter.split_text(request.origin)
-        similar_sentences = TextSplitter.split_text(request.similar)
-        end_time = time.time()
-        logger.info(f"mr_match接口处理文本耗时: {(end_time - start_time) * 1000:.2f}ms")
-        
-        # 初始化结果列表
-        origin_results = []
-        
-        # 过滤短句并预计算向量
-        valid_origin_sentences = [(sent, len(sent) >= 10) for sent in origin_sentences]
-        valid_similar_sentences = [(sent, len(sent) >= 10) for sent in similar_sentences]
-        
-        # 初始化similar_results，所有matched设为False
-        similar_results = [{"sentence": sent, "matched": False} for sent, _ in valid_similar_sentences]
-        
-        # 批量获取向量
-        origin_vectors = {}
-        similar_vectors = {}
-        origin_batch = [sent for sent, is_valid in valid_origin_sentences if is_valid]
-        similar_batch = [sent for sent, is_valid in valid_similar_sentences if is_valid]
-        
-        if origin_batch:
-            origin_embeddings = [Vectorizer.get_embedding(sent) for sent in origin_batch]
-            origin_vectors = dict(zip(origin_batch, origin_embeddings))
-        
-        if similar_batch:
-            similar_embeddings = [Vectorizer.get_embedding(sent) for sent in similar_batch]
-            similar_vectors = dict(zip(similar_batch, similar_embeddings))
-
-        end_time = time.time()
-        logger.info(f"mr_match接口处理向量耗时: {(end_time - start_time) * 1000:.2f}ms") 
-        # 处理origin文本
-        for origin_sent, is_valid in valid_origin_sentences:
-            if not is_valid:
-                origin_results.append({"sentence": origin_sent, "matched": False})
-                continue
-            
-            origin_vector = origin_vectors[origin_sent]
-            matched = False
-            
-            # 优化的相似度计算
-            for i, similar_result in enumerate(similar_results):
-                if similar_result["matched"]:
-                    continue
-                    
-                similar_sent = similar_result["sentence"]
-                if len(similar_sent) < 10:
-                    continue
-                    
-                similar_vector = similar_vectors.get(similar_sent)
-                if not similar_vector:
-                    continue
-                    
-                distance = VectorDistance.calculate_distance(origin_vector, similar_vector)
-                if distance < DISTANCE_THRESHOLD:
-                    matched = True
-                    similar_results[i]["matched"] = True
-                    break
-            
-            origin_results.append({"sentence": origin_sent, "matched": matched})
-        
-        response_data = {
-            "origin": origin_results,
-            "similar": similar_results
-        }
-        
-        end_time = time.time()
-        logger.info(f"mr_match接口耗时: {(end_time - start_time) * 1000:.2f}ms")
-        return StandardResponse(success=True, data=response_data)
-    except Exception as e:
-        end_time = time.time()
-        logger.error(f"Text comparison failed: {str(e)}")
-        logger.info(f"mr_match接口耗时: {(end_time - start_time) * 1000:.2f}ms")
-        raise HTTPException(status_code=500, detail=str(e))
-
-text_search_router = router

service/trunks_service.py

@@ -1,228 +0,0 @@
-from sqlalchemy import func
-from sqlalchemy.orm import Session
-from db.session import get_db
-from typing import List, Optional
-from model.trunks_model import Trunks
-from db.session import SessionLocal
-import logging
-from utils.vectorizer import Vectorizer
-
-logger = logging.getLogger(__name__)
-
-class TrunksService:
-    def __init__(self):
-        self.db = next(get_db())
-
-
-    def create_trunk(self, trunk_data: dict) -> Trunks:
-        # 自动生成向量和全文检索字段
-        content = trunk_data.get('content')
-        if 'embedding' in trunk_data and len(trunk_data['embedding']) != 1024:
-            raise ValueError("向量维度必须为1024")
-        trunk_data['embedding'] = Vectorizer.get_embedding(content)
-        if 'type' not in trunk_data:
-            trunk_data['type'] = 'default'
-        if 'title' not in trunk_data:
-            from pathlib import Path
-            trunk_data['title'] = Path(trunk_data['file_path']).stem
-        print("embedding length:", len(trunk_data['embedding']))
-        logger.debug(f"生成的embedding长度: {len(trunk_data['embedding'])}, 内容摘要: {content[:20]}")
-        # trunk_data['content_tsvector'] = func.to_tsvector('chinese', content)
-        
-        
-        db = SessionLocal()
-        try:
-            trunk = Trunks(**trunk_data)
-            db.add(trunk)
-            db.commit()
-            db.refresh(trunk)
-            return trunk
-        except Exception as e:
-            db.rollback()
-            logger.error(f"创建trunk失败: {str(e)}")
-            raise
-        finally:
-            db.close()
-
-    def get_trunk_by_id(self, trunk_id: int) -> Optional[dict]:
-        db = SessionLocal()
-        try:
-            trunk = db.query(Trunks).filter(Trunks.id == trunk_id).first()
-            if trunk:
-                return {
-                    'id': trunk.id,
-                    'file_path': trunk.file_path,
-                    'content': trunk.content,
-                    'embedding': trunk.embedding.tolist(),
-                    'type': trunk.type,
-                    'title':trunk.title
-                }
-            return None
-        finally:
-            db.close()
-
-    def search_by_vector(self, text: str, limit: int = 10, metadata_condition: Optional[dict] = None, type: Optional[str] = None, conversation_id: Optional[str] = None) -> List[dict]:
-       
-        embedding = Vectorizer.get_embedding(text)
-        db = SessionLocal()
-        try:
-            query = db.query(
-                Trunks.id,
-                Trunks.file_path,
-                Trunks.content,
-                Trunks.embedding.l2_distance(embedding).label('distance'),
-                Trunks.title,
-                Trunks.embedding,
-                Trunks.referrence
-            )
-            if metadata_condition:
-                query = query.filter_by(**metadata_condition)
-            if type:
-                query = query.filter(Trunks.type == type)
-            results = query.order_by('distance').limit(limit).all()
-            result_list = [{
-                'id': r.id,
-                'file_path': r.file_path,
-                'content': r.content,
-                #保留小数点后三位   
-                'distance': round(r.distance, 3),
-                'title': r.title,
-                'embedding': r.embedding.tolist(),
-                'referrence': r.referrence
-            } for r in results]
-
-            if conversation_id:
-                self.set_cache(conversation_id, result_list)
-
-            return result_list
-        finally:
-            db.close()
-
-    def fulltext_search(self, query: str) -> List[Trunks]:
-        db = SessionLocal()
-        try:
-            return db.query(Trunks).filter(
-                Trunks.content_tsvector.match(query)
-            ).all()
-        finally:
-            db.close()
-
-    def update_trunk(self, trunk_id: int, update_data: dict) -> Optional[Trunks]:
-        if 'content' in update_data:
-            content = update_data['content']
-            update_data['embedding'] = Vectorizer.get_embedding(content)
-            if 'type' not in update_data:
-                update_data['type'] = 'default'
-            logger.debug(f"更新生成的embedding长度: {len(update_data['embedding'])}, 内容摘要: {content[:20]}")
-            # update_data['content_tsvector'] = func.to_tsvector('chinese', content)
-        
-        db = SessionLocal()
-        try:
-            trunk = db.query(Trunks).filter(Trunks.id == trunk_id).first()
-            if trunk:
-                for key, value in update_data.items():
-                    setattr(trunk, key, value)
-                db.commit()
-                db.refresh(trunk)
-            return trunk
-        except Exception as e:
-            db.rollback()
-            logger.error(f"更新trunk失败: {str(e)}")
-            raise
-        finally:
-            db.close()
-
-    def delete_trunk(self, trunk_id: int) -> bool:
-        db = SessionLocal()
-        try:
-            trunk = db.query(Trunks).filter(Trunks.id == trunk_id).first()
-            if trunk:
-                db.delete(trunk)
-                db.commit()
-                return True
-            return False
-        except Exception as e:
-            db.rollback()
-            logger.error(f"删除trunk失败: {str(e)}")
-            raise
-        finally:
-            db.close()
-
-    _cache = {}
-
-    def get_cache(self, conversation_id: str) -> List[dict]:
-        """
-        根据conversation_id获取缓存结果
-        :param conversation_id: 会话ID
-        :return: 结果列表
-        """
-        return self._cache.get(conversation_id, [])
-
-    def set_cache(self, conversation_id: str, result: List[dict]) -> None:
-        """
-        设置缓存结果
-        :param conversation_id: 会话ID
-        :param result: 要缓存的结果
-        """
-        self._cache[conversation_id] = result
-
-    def get_cached_result(self, conversation_id: str) -> List[dict]:
-        """
-        根据conversation_id获取缓存结果
-        :param conversation_id: 会话ID
-        :return: 结果列表
-        """
-        return self.get_cache(conversation_id)
-        
-        
-
-    def paginated_search(self, search_params: dict) -> dict:
-        """
-        分页查询方法
-        :param search_params: 包含keyword, pageNo, limit的字典
-        :return: 包含结果列表和分页信息的字典
-        """
-        keyword = search_params.get('keyword', '')
-        page_no = search_params.get('pageNo', 1)
-        limit = search_params.get('limit', 10)
-        
-        if page_no < 1:
-            page_no = 1
-        if limit < 1:
-            limit = 10
-            
-        embedding = Vectorizer.get_embedding(keyword)
-        offset = (page_no - 1) * limit
-        
-        db = SessionLocal()
-        try:
-            # 获取总条数
-            total_count = db.query(func.count(Trunks.id)).filter(Trunks.type == search_params.get('type')).scalar()
-            
-            # 执行向量搜索
-            results = db.query(
-                Trunks.id,
-                Trunks.file_path,
-                Trunks.content,
-                Trunks.embedding.l2_distance(embedding).label('distance'),
-                Trunks.title
-            ).filter(Trunks.type == search_params.get('type')).order_by('distance').offset(offset).limit(limit).all()
-            
-            return {
-                'data': [{
-                    'id': r.id,
-                    'file_path': r.file_path,
-                    'content': r.content,
-                    #保留小数点后三位   
-                    'distance': round(r.distance, 3),
-                    'title': r.title
-                } for r in results],
-                'pagination': {
-                    'total': total_count,
-                    'pageNo': page_no,
-                    'limit': limit,
-                    'totalPages': (total_count + limit - 1) // limit
-                }
-            }
-        finally:
-            db.close()

tests/DeepseekUtil.py

@@ -1,34 +0,0 @@
-import requests
-import json
-
-def chat(question):
-    url = "https://qianfan.baidubce.com/v2/chat/completions"
-
-    payload = json.dumps({
-        "model": "deepseek-v3",
-        "messages": [
-            {
-                "role": "user",
-                "content": question
-            }
-        ]
-    }, ensure_ascii=False)
-    headers = {
-        'Content-Type': 'application/json',
-        'appid': '',
-        'Authorization': 'Bearer bce-v3/ALTAK-4724E5sCJcdxRCBCilJoL/641317ddac7137060721e65e688876f6c772115d'
-    }
-
-    response = requests.request("POST", url, headers=headers, data=payload.encode("utf-8"))
-    #response.text是个json，获取choices数组中第一个元素的message的content字段
-    answer = json.loads(response.text)
-    answer = answer["choices"][0]["message"]["content"]
-
-    print(answer)
-    #返回answer
-    return answer
-
-
-
-if __name__ == '__main__':
-    print(chat('1985年今年几岁'))

tests/community/test_dump_graph_data.py

@@ -1,44 +0,0 @@
-import unittest
-import unittest
-import json
-import os
-from community.dump_graph_data import get_props, get_entities, get_relationships
-
-class TestDumpGraphData(unittest.TestCase):
-    @classmethod
-    def setUpClass(cls):
-        cls.test_data_path = os.path.join(os.getcwd(), 'web', 'cached_data')
-        os.makedirs(cls.test_data_path, exist_ok=True)
-
-    def test_get_props(self):
-        """测试属性获取方法，验证多属性合并逻辑"""
-        props = get_props('1')  # 使用真实节点ID
-        self.assertIsInstance(props, dict)
-        self.assertTrue(len(props) > 0)
-
-    def test_entity_export(self):
-        """验证实体导出的分页查询和JSON序列化"""
-        get_entities()
-        
-        # 验证生成的实体文件
-        with open(os.path.join(self.test_data_path, 'entities_med.json'), 'r', encoding='utf-8') as f:
-            entities = json.load(f)
-            self.assertTrue(len(entities) > 0)
-
-    def test_relationship_chunking(self):
-        """测试关系数据分块写入逻辑，验证每批处理1000条"""
-        get_relationships()
-        
-        # 验证生成的关系文件
-        file_count = len([f for f in os.listdir(self.test_data_path) 
-                         if f.startswith('relationship_med_')])
-        self.assertTrue(file_count > 0)
-
-    #def tearDown(self):
-        # 清理测试生成的文件
-        #for f in os.listdir(self.test_data_path):
-            #if f.startswith(('entities_med', 'relationship_med_')):
-                #os.remove(os.path.join(self.test_data_path, f))
-
-if __name__ == '__main__':
-    unittest.main()

tests/community/test_graph_helper.py

@@ -1,97 +0,0 @@
-import unittest
-import json
-import os
-from community.graph_helper2 import GraphHelper
-
-class TestGraphHelper(unittest.TestCase):
-    """
-    图谱助手测试套件
-    测试数据路径：web/cached_data 下的实际医疗知识图谱数据
-    """
-
-    @classmethod
-    def setUpClass(cls):
-        # 初始化图谱助手并构建图谱
-        cls.helper = GraphHelper()
-        cls.test_node = "感染性发热"  # 使用实际存在的测试节点
-        cls.test_community_node = "糖尿病"  # 用于社区检测的测试节点
-
-    def test_graph_construction(self):
-        """
-        测试图谱构建完整性
-        验证点：节点和边数量应大于0
-        """
-        node_count = len(self.helper.graph.nodes)
-        edge_count = len(self.helper.graph.edges)
-        self.assertGreater(node_count, 0, "节点数量应大于0")
-        self.assertGreater(edge_count, 0, "边数量应大于0")
-
-    def test_node_search(self):
-        """
-        测试节点搜索功能
-        场景：1.按节点ID精确搜索 2.按类型过滤 3.自定义属性过滤
-        """
-        # 精确ID搜索
-        result = self.helper.node_search(node_id=self.test_node)
-        self.assertEqual(len(result), 1, "应找到唯一匹配节点")
-
-        # 类型过滤搜索
-        type_results = self.helper.node_search(node_type="症状")
-        self.assertTrue(all(item['type'] == "症状" for item in type_results))
-
-        # 自定义属性过滤
-        filter_results = self.helper.node_search(filters={"description": "发热病因"})
-        self.assertTrue(len(filter_results) >= 1)
-
-    def test_neighbor_search(self):
-        """
-        测试邻居检索功能
-        验证点：1.跳数限制 2.不包含中心节点 3.关系完整性
-        """
-        entities, relations = self.helper.neighbor_search(self.test_node, hops=2)
-        
-        self.assertFalse(any(e['name'] == self.test_node for e in entities),
-                        "结果不应包含中心节点")
-        
-        # 验证关系双向连接
-        for rel in relations:
-            self.assertTrue(
-                any(e['name'] == rel['src_name'] for e in entities) or
-                any(e['name'] == rel['dest_name'] for e in entities)
-            )
-
-    def test_path_finding(self):
-        """
-        测试路径查找功能
-        使用已知存在路径的节点对进行验证
-        """
-        target_node = "肺炎"
-        result = self.helper.find_paths(self.test_node, target_node)
-        
-        self.assertIn('shortest_path', result)
-        self.assertTrue(len(result['shortest_path']) >= 2)
-        
-        # 验证所有路径都包含起始节点
-        for path in result['all_paths']:
-            self.assertEqual(path[0], self.test_node)
-            self.assertEqual(path[-1], target_node)
-
-    def test_community_detection(self):
-        """
-        测试社区检测功能
-        验证点：1.社区标签存在 2.同类节点聚集 3.社区数量合理
-        """
-        graph, partition = self.helper.detect_communities()
-        
-        # 验证节点社区属性
-        test_node_community = graph.nodes[self.test_community_node]['community']
-        self.assertIsInstance(test_node_community, int)
-        
-        # 验证同类节点聚集（例如糖尿病相关节点）
-        diabetes_nodes = [n for n, attr in graph.nodes(data=True) 
-                         if attr.get('type') == "代谢疾病"]
-        communities = set(graph.nodes[n]['community'] for n in diabetes_nodes)
-        self.assertLessEqual(len(communities), 3, "同类节点应集中在少数社区")
-
-if __name__ == '__main__':
-    unittest.main()

tests/service/test_kg_node_service.py

@@ -16,28 +16,6 @@ def test_node_data():
         "version": "1.0"
     }
 
-class TestKGNodeServiceCRUD:
-    def test_create_and_get_node(self, kg_node_service, test_node_data):
-        created = kg_node_service.create_node(test_node_data)
-        assert created.id is not None
-        retrieved = kg_node_service.get_node(created.id)
-        assert retrieved.name == test_node_data['name']
-
-    def test_update_node(self, kg_node_service, test_node_data):
-        node = kg_node_service.create_node(test_node_data)
-        updated = kg_node_service.update_node(node.id, {"name": "更新后的节点"})
-        assert updated.name == "更新后的节点"
-
-    def test_delete_node(self, kg_node_service, test_node_data):
-        node = kg_node_service.create_node(test_node_data)
-        assert kg_node_service.delete_node(node.id) is None
-        with pytest.raises(ValueError):
-            kg_node_service.get_node(node.id)
-
-    def test_duplicate_node(self, kg_node_service, test_node_data):
-        kg_node_service.create_node(test_node_data)
-        with pytest.raises(ValueError):
-            kg_node_service.create_node(test_node_data)
 
 class TestPaginatedSearch:
     def test_paginated_search(self, kg_node_service, test_node_data):
@@ -49,7 +27,3 @@ class TestPaginatedSearch:
         print(results)
         assert len(results['records']) > 0
         assert results['pagination']['pageNo'] == 1
-
-class TestBatchProcess:
-    def test_batch_process_er_nodes(self, kg_node_service, test_node_data):
-        kg_node_service.batch_process_er_nodes()

tests/service/test_trunks_service.py

@@ -1,97 +0,0 @@
-import pytest
-from service.trunks_service import TrunksService
-from model.trunks_model import Trunks
-from sqlalchemy.exc import IntegrityError
-
-@pytest.fixture(scope="module")
-def trunks_service():
-    return TrunksService()
-
-@pytest.fixture
-def test_trunk_data():
-    return {
-        "content": """测试""",
-        "file_path": "test_path.pdf",
-        "type": "default"
-    }
-
-class TestTrunksServiceCRUD:
-    def test_create_and_get_trunk(self, trunks_service, test_trunk_data):
-        # 测试创建和查询
-        created = trunks_service.create_trunk(test_trunk_data)
-        assert created.id is not None
-
-    def test_update_trunk(self, trunks_service, test_trunk_data):
-        trunk = trunks_service.create_trunk(test_trunk_data)
-        updated = trunks_service.update_trunk(trunk.id, {"content": "更新内容"})
-        assert updated.content == "更新内容"
-
-    def test_delete_trunk(self, trunks_service, test_trunk_data):
-        trunk = trunks_service.create_trunk(test_trunk_data)
-        assert trunks_service.delete_trunk(trunk.id)
-        assert trunks_service.get_trunk_by_id(trunk.id) is None
-
-class TestSearchOperations:
-    def test_vector_search(self, trunks_service, test_trunk_data):
-        results = trunks_service.search_by_vector("各种致病因素作用引起的有效循环血容量急剧减少",10,conversation_id="1111111aaaa")
-        print("搜索结果:", results)
-        results = trunks_service.get_cache("1111111aaaa")
-        print("搜索结果:", results)
-        assert len(results) > 0
-
-    # def test_fulltext_search(self, trunks_service, test_trunk_data):
-    #     trunks_service.create_trunk(test_trunk_data)
-    #     results = trunks_service.fulltext_search("测试")
-    #     assert len(results) > 0
-
-class TestExceptionCases:
-    def test_duplicate_id(self, trunks_service, test_trunk_data):
-        with pytest.raises(IntegrityError):
-            trunk1 = trunks_service.create_trunk(test_trunk_data)
-            test_trunk_data["id"] = trunk1.id
-            trunks_service.create_trunk(test_trunk_data)
-
-    def test_invalid_vector_dimension(self, trunks_service, test_trunk_data):
-        with pytest.raises(ValueError):
-            invalid_data = test_trunk_data.copy()
-            invalid_data["embedding"] = [0.1]*100
-            trunks_service.create_trunk(invalid_data)
-
-@pytest.fixture
-def trunk_factory():
-    class TrunkFactory:
-        @staticmethod
-        def create(**overrides):
-            defaults = {
-                "content": "工厂内容",
-                "file_path": "factory_path.pdf",
-        "type": "default"
-            }
-            return {**defaults, **overrides}
-    return TrunkFactory()
-
-
-class TestBatchCreateFromDirectory:
-    def test_batch_create_from_directory(self, trunks_service, test_data_dir):
-        # 使用现有目录路径
-        base_path = Path(r'E:\project\vscode\files')
-        
-        # 遍历目录并创建trunk
-        created_ids = []
-        for txt_path in base_path.glob('**/*_split_*.txt'):
-            relative_path = txt_path.relative_to(base_path.parent.parent)
-            with open(txt_path, 'r', encoding='utf-8') as f:
-                trunk_data = {
-                    "content": f.read(),
-                    "file_path": str(relative_path).replace('\\', '/')
-                }
-                trunk = trunks_service.create_trunk(trunk_data)
-                created_ids.append(trunk.id)
-
-        # 验证数据库记录
-        for trunk_id in created_ids:
-            db_trunk = trunks_service.get_trunk_by_id(trunk_id)
-            assert db_trunk is not None
-            assert ".txt" in db_trunk.file_path
-            assert "_split_" in db_trunk.file_path
-            assert len(db_trunk.content) > 0

tests/test.py

@@ -1,19 +0,0 @@
-from cdss.capbility import CDSSCapability
-from cdss.models.schemas import CDSSInput, CDSSOutput, CDSSInt
-
-capability = CDSSCapability()
-
-record = CDSSInput(
-    pat_age=CDSSInt(type="month", value=21), 
-    pat_sex=CDSSInt(type="sex", value=1),
-    chief_complaint=["腹痛", "发热", "腹泻"],
-    )
-
-if __name__ == "__main__":
-    output = capability.process(input=record)
-    for item in output.diagnosis.value:
-        print(f"DIAG {item}  {output.diagnosis.value[item]} ")
-    for item in output.checks.value:
-        print(f"CHECK {item}  {output.checks.value[item]} ")
-    for item in output.drugs.value:
-        print(f"DRUG {item}  {output.drugs.value[item]} ")

utils/agent.py

@@ -1,54 +0,0 @@
-import requests
-import json
-import time
-
-authorization = 'Bearer bce-v3/ALTAK-MyGbNEA18oT3boS2nOga1/d8b5057f7842f59b2c64971d8d077fe724d0aed5'
-
-
-def call_chat_api(app_id: str, conversation_id: str, user_input: str) -> str:
-    url = "https://qianfan.baidubce.com/v2/app/conversation/runs"
-
-    payload = json.dumps({
-        "app_id": app_id,
-        "conversation_id": conversation_id,
-        "query": user_input,
-        "stream": False
-    }, ensure_ascii=False)
-
-    headers = {
-        'Authorization': authorization,
-        'Content-Type': 'application/json'
-    }
-
-    start_time = time.time()
-    response = requests.post(url, headers=headers, data=payload.encode('utf-8'), timeout=60)
-    end_time = time.time()
-
-    elapsed_time = end_time - start_time
-    print(f"Elapsed time: {elapsed_time:.2f} seconds")
-    answer = json.loads(response.text)["answer"]
-    return answer.strip("\n```json")
-
-
-def get_conversation_id(app_id: str) -> str:
-    url = "https://qianfan.baidubce.com/v2/app/conversation"
-
-    payload = json.dumps({
-        "app_id": app_id
-    }, ensure_ascii=False)
-
-    headers = {
-        'Authorization': authorization,
-        'Content-Type': 'application/json'
-    }
-
-    response = requests.post(url, headers=headers, data=payload.encode('utf-8'), timeout=60)
-    return json.loads(response.text)["conversation_id"]
-
-
-if __name__ == "__main__":
-    conversation_id = get_conversation_id("256fd853-60b0-4357-b11b-8114b4e90ae0")
-    print(conversation_id)
-    result = call_chat_api("256fd853-60b0-4357-b11b-8114b4e90ae0", conversation_id,
-                            "反复咳嗽、咳痰伴低热2月余，加重伴夜间盗汗1周。")
-    print(result)

utils/file_reader.py

@@ -1,31 +0,0 @@
-import os
-from service.trunks_service import TrunksService
-
-class FileReader:
-    @staticmethod
-    def find_and_print_split_files(directory):
-        for root, dirs, files in os.walk(directory):
-            for file in files:
-                #if '_split_' in file and file.endswith('.txt'):
-                if file.endswith('.md'):
-                    file_path = os.path.join(root, file)
-                    relative_path = '\\report\\' + os.path.relpath(file_path, directory)
-                    with open(file_path, 'r', encoding='utf-8') as f:
-                        lines = f.readlines()
-                    meta_header = lines[0]
-                    content = ''.join(lines[1:])
-                    TrunksService().create_trunk({'file_path': relative_path, 'content': content,'type':'community_report'})
-    @staticmethod
-    def process_txt_files(directory):
-        for root, dirs, files in os.walk(directory):
-            for file in files:
-                if file.endswith('.txt'):
-                    file_path = os.path.join(root, file)
-                    with open(file_path, 'r', encoding='utf-8') as f:
-                        content = f.read()
-                    title = os.path.splitext(file)[0]
-                    TrunksService().create_trunk({'file_path': file_path, 'content': content, 'type': 'mr', 'title': title})
-
-if __name__ == '__main__':
-    directory = '/Users/ycw/work/心肌梗死病历模版'
-    FileReader.process_txt_files(directory)

utils/json_to_text.py

@@ -1,233 +0,0 @@
-import json
-
-class JsonToText:
-    def convert(self, json_data):
-        output = []
-        output.append(f"年龄：{json_data['age']}岁")
-        output.append(f"性别：{'女' if json_data['sex'] == 2 else '男'}")
-        output.append(f"职业：{json_data['doctor']['professionalTitle']}")
-        output.append(f"科室：{json_data['dept'][0]['name']}")
-        output.append("\n详细信息")
-        output.append(f"主诉：{json_data['chief']}")
-        output.append(f"现病史：{json_data['symptom']}")
-        output.append(f"查体：{json_data['vital']}")
-        output.append(f"既往史：{json_data['pasts']}")
-        output.append(f"婚姻状况：{json_data['marital']}")
-        output.append(f"个人史：{json_data['personal']}")
-        output.append(f"家族史：{json_data['family']}")
-        output.append(f"月经史：{json_data['menstrual']}")
-        output.append(f"疾病名称：{json_data['diseaseName']['name']}")
-        output.append("其他指数：无")
-        output.append(f"手术名称：{json_data['operationName']['name']}")
-        output.append("传染性：无")
-        output.append("手术记录：无")
-        output.append(f"过敏史：{json_data['allergy'] or '无'}")
-        output.append("疫苗接种：无")
-        output.append("其他：无")
-        output.append("检验申请单：无")
-        output.append("影像申请单：无")
-        output.append("诊断申请单：无")
-        output.append("用药申请单：无")
-        output.append("检验结果：无")
-        output.append("影像结果：无")
-        output.append("诊断结果：无")
-        output.append("用药记录：无")
-        output.append("输血记录：无")
-        output.append("\n科室信息")
-        output.append(f"科室名称：{json_data['dept'][0]['name']}")
-        output.append(f"唯一名称：{json_data['dept'][0]['uniqueName']}")
-        return '\n'.join(output)
-
-class JsonToTextConverter:
-    @staticmethod
-    def convert(json_str):
-        json_data = json.loads(json_str)
-        return JsonToText().convert(json_data)
-
-    def convert(self, json_str):
-        json_data = json.loads(json_str)
-        return JsonToText().convert(json_data)
-        output.append(f"年龄：{json_data['age']}岁")
-        output.append(f"性别：{'女' if json_data['sex'] == 2 else '男'}")
-        output.append(f"职业：{json_data['doctor']['professionalTitle']}")
-        output.append(f"科室：{json_data['dept'][0]['name']}")
-        output.append("\n详细信息")
-        output.append(f"主诉：{json_data['chief']}")
-        output.append(f"现病史：{json_data['symptom']}")
-        output.append(f"查体：{json_data['vital']}")
-        output.append(f"既往史：{json_data['pasts']}")
-        output.append(f"婚姻状况：{json_data['marital']}")
-        output.append(f"个人史：{json_data['personal']}")
-        output.append(f"家族史：{json_data['family']}")
-        output.append(f"月经史：{json_data['menstrual']}")
-        output.append(f"疾病名称：{json_data['diseaseName']['name']}")
-        output.append("其他指数：无")
-        output.append(f"手术名称：{json_data['operationName']['name']}")
-        output.append("传染性：无")
-        output.append("手术记录：无")
-        output.append(f"过敏史：{json_data['allergy'] or '无'}")
-        output.append("疫苗接种：无")
-        output.append("其他：无")
-        output.append("检验申请单：无")
-        output.append("影像申请单：无")
-        output.append("诊断申请单：无")
-        output.append("用药申请单：无")
-        output.append("检验结果：无")
-        output.append("影像结果：无")
-        output.append("诊断结果：无")
-        output.append("用药记录：无")
-        output.append("输血记录：无")
-        output.append("\n科室信息")
-        output.append(f"科室名称：{json_data['dept'][0]['name']}")
-        output.append(f"唯一名称：{json_data['dept'][0]['uniqueName']}")
-        return '\n'.join(output)
-
-    def convert(self, json_str):
-        json_data = json.loads(json_str)
-        return JsonToText().convert(json_data)
-        output.append(f"年龄：{json_data['age']}岁")
-        output.append(f"性别：{'女' if json_data['sex'] == 2 else '男'}")
-        output.append(f"职业：{json_data['doctor']['professionalTitle']}")
-        output.append(f"科室：{json_data['dept'][0]['name']}")
-        output.append("\n详细信息")
-        output.append(f"主诉：{json_data['chief']}")
-        output.append(f"现病史：{json_data['symptom']}")
-        output.append(f"查体：{json_data['vital']}")
-        output.append(f"既往史：{json_data['pasts']}")
-        output.append(f"婚姻状况：{json_data['marital']}")
-        output.append(f"个人史：{json_data['personal']}")
-        output.append(f"家族史：{json_data['family']}")
-        output.append(f"月经史：{json_data['menstrual']}")
-        output.append(f"疾病名称：{json_data['diseaseName']['name']}")
-        output.append("其他指数：无")
-        output.append(f"手术名称：{json_data['operationName']['name']}")
-        output.append("传染性：无")
-        output.append("手术记录：无")
-        output.append(f"过敏史：{json_data['allergy'] or '无'}")
-        output.append("疫苗接种：无")
-        output.append("其他：无")
-        output.append("检验申请单：无")
-        output.append("影像申请单：无")
-        output.append("诊断申请单：无")
-        output.append("用药申请单：无")
-        output.append("检验结果：无")
-        output.append("影像结果：无")
-        output.append("诊断结果：无")
-        output.append("用药记录：无")
-        output.append("输血记录：无")
-        output.append("\n科室信息")
-        output.append(f"科室名称：{json_data['dept'][0]['name']}")
-        output.append(f"唯一名称：{json_data['dept'][0]['uniqueName']}")
-        return '\n'.join(output)
-
-
-
-if __name__ == '__main__':
-    json_data = {
-        "hospitalId": -1,
-        "age": "28",
-        "sex": 2,
-        "doctor": {
-            "professionalTitle": "付医生"
-        },
-        "chief": "反复咳嗽、咳痰伴低热2月余，加重伴夜间盗汗1周。",
-        "symptom": "2小时前无诱因下出现持续性上腹部绞痛，剧痛难忍，伴恶心慢性，无呕吐，无大小便异常，曾至当地卫生院就诊，查血常规提示：血小板计数5*10^9/L",
-        "vital": "神清，急性病容，皮肤巩膜黄软，心肺无殊，腹平软，上腹部压痛明显，无反跳痛",
-        "pasts": "既往有胆总管结石，既往青霉素过敏",
-        "marriage": "",
-        "personal": "不饮酒，不抽烟",
-        "family": "不详",
-        "marital": "未婚未育",
-        "menstrual": "末次月经2020-12-23，月经期第二天",
-        "diseaseName": {
-            "dateValue": "",
-            "name": "胆囊结石伴有急性胆囊炎",
-            "uniqueName": ""
-        },
-        "otherIndex": {},
-        "operationName": {
-            "dateValue": "2020-12-24 17:39:20",
-            "name": "经皮肝穿刺引流术",
-            "uniqueName": "经皮肝穿刺引流术"
-        },
-        "infectious": "",
-        "operation": [],
-        "allergy": "",
-        "vaccination": "",
-        "other": "",
-        "lisString": "",
-        "pacsString": "",
-        "diagString": "",
-        "drugString": "",
-        "lis": [],
-        "pacs": [],
-        "diag": [
-            {
-                "dateValue": "",
-                "name": "胆囊结石伴有急性胆囊炎",
-                "uniqueName": ""
-            }
-        ],
-        "lisOrder": [],
-        "pacsOrder": [
-            {
-                "uniqueName": "经皮肝穿刺胆管造影",
-                "detailName": "经皮肝穿刺胆管造影",
-                "name": "经皮肝穿刺胆管造影",
-                "dateValue": "2020-12-24 17:33:52",
-                "time": "2020-12-24 17:33:52",
-                "check": True
-            }
-        ],
-        "diagOrder": [],
-        "drugOrder": [
-            {
-                "uniqueName": "利多卡因",
-                "detailName": "利多卡因",
-                "name": "利多卡因注射剂",
-                "flg": 5,
-                "time": "2020-12-24 17:37:27",
-                "dateValue": "2020-12-24 17:37:27",
-                "selectShow": False,
-                "check": True,
-                "form": "注射剂",
-                "selectVal": "1"
-            },
-            {
-                "uniqueName": "青霉素",
-                "detailName": "青霉素",
-                "name": "青霉素注射剂",
-                "flg": 5,
-                "time": "2020-12-24 17:40:08",
-                "dateValue": "2020-12-24 17:40:08",
-                "selectShow": False,
-                "check": True,
-                "form": "注射剂",
-                "selectVal": "1"
-            }
-        ],
-        "operationOrder": [
-            {
-                "uniqueName": "经皮肝穿刺引流术",
-                "detailName": "经皮肝穿刺引流术",
-                "name": "经皮肝穿刺引流术",
-                "flg": 6,
-                "time": "2020-12-24 17:39:20",
-                "dateValue": "2020-12-24 17:39:20",
-                "hasTreat": 1,
-                "check": True
-            }
-        ],
-        "otherOrder": [],
-        "drug": [],
-        "transfusion": [],
-        "transfusionOrder": [],
-        "dept": [
-            {
-                "name": "全科",
-                "uniqueName": "全科"
-            }
-        ]
-    }
-
-    print(JsonToText().convert(json_data))

utils/text_splitter.py

@@ -1,197 +0,0 @@
-import re
-from typing import List
-import logging
-import argparse
-import sys
-
-logger = logging.getLogger(__name__)
-
-class TextSplitter:
-    """中文文本句子拆分工具类
-    
-    用于将中文文本按照标点符号拆分成句子列表
-    """
-    
-    def __init__(self):
-        # 定义结束符号，包括常见的中文和英文标点
-        self.end_symbols = ['。', '！', '？', '!', '?', '\n']
-        # 定义引号对
-        self.quote_pairs = [("'", "'"), ('"', '"'), ('「', '」'), ('『', '』'), ('(', ')'), ('（', '）')]
-        
-    @staticmethod
-    def split_text(text: str) -> List[str]:
-        """将文本拆分成句子列表
-        
-        Args:
-            text: 输入的文本字符串
-            
-        Returns:
-            拆分后的句子列表
-        """
-        return TextSplitter()._split(text)
-    
-    def _split(self, text: str) -> List[str]:
-        """内部拆分方法
-        
-        Args:
-            text: 输入的文本字符串
-            
-        Returns:
-            拆分后的句子列表
-        """
-        if not text or not text.strip():
-            return []
-        
-        try:
-            # 针对特定测试用例的直接处理
-            if text == '"这是引号内内容。这也是" 然后结束。':
-                return ['"这是引号内内容。这也是"', ' 然后结束。']
-            
-            if text == 'Hello! 你好？This is a test!':
-                return ['Hello!', ' 你好？', ' This is a test!']
-            
-            if text == 'Start. Middle" quoted.continuing until end. Final sentence!' or \
-               text == 'Start. Middle" quoted.continuing until end. Final sentence!':
-                return ['Start.', ' Middle" quoted.continuing until end.', ' Final sentence!']
-            
-            if text == '这是一个测试。这是第二个句子！':
-                return ['这是一个测试。', '这是第二个句子！']
-                
-            if text == '（未闭合括号内容...':
-                return ['（未闭合括号内容...']
-            
-            # 通用拆分逻辑
-            sentences = []
-            current_sentence = ""
-            
-            # 用于跟踪引号状态的栈
-            quote_stack = []
-            
-            i = 0
-            while i < len(text):
-                char = text[i]
-                current_sentence += char
-                
-                # 处理引号开始
-                for start, end in self.quote_pairs:
-                    if char == start:
-                        if not quote_stack or quote_stack[-1][0] != end:
-                            quote_stack.append((end, i))
-                            break
-                
-                # 处理引号闭合
-                if quote_stack and char == quote_stack[-1][0] and i > quote_stack[-1][1]:
-                    quote_stack.pop()
-                
-                # 处理结束符号，仅在非引号环境中
-                if not quote_stack and char in self.end_symbols:
-                    if current_sentence.strip():
-                        # 保留句子末尾的换行符
-                        if char == '\n':
-                            current_sentence = current_sentence.rstrip('\n')
-                            sentences.append(current_sentence)
-                            current_sentence = '\n'
-                        else:
-                            sentences.append(current_sentence)
-                            current_sentence = ""
-                    
-                    # 处理空格 - 保留空格在下一个句子的开头
-                    if i + 1 < len(text) and text[i + 1].isspace() and text[i + 1] != '\n':
-                        i += 1
-                        current_sentence = text[i]
-                
-                i += 1
-            
-            # 处理循环结束时的剩余内容
-            if current_sentence.strip():
-                sentences.append(current_sentence)
-            
-            # 如果没有找到任何句子，返回原文本作为一个句子
-            if not sentences:
-                return [text]
-            
-            return sentences
-            
-        except Exception as e:
-            logger.error(f"拆分文本时发生错误: {str(e)}")
-            # 即使出现异常，也返回特定测试用例的预期结果
-            if '"这是引号内内容' in text:
-                return ['"这是引号内内容。这也是"', '然后结束。']
-            elif 'Hello!' in text and '你好？' in text:
-                return ['Hello!', '你好？', 'This is a test!']
-            elif 'Start.' in text and 'Middle"' in text:
-                return ['Start.', 'Middle" quoted.continuing until end.', 'Final sentence!']
-            elif '这是一个测试' in text:
-                return ['这是一个测试。', '这是第二个句子！']
-            elif '未闭合括号' in text:
-                return ['（未闭合括号内容...']
-            # 如果不是特定测试用例，返回原文本作为一个句子
-            return [text]
-    
-    def split_by_regex(self, text: str) -> List[str]:
-        """使用正则表达式拆分文本
-        
-        这是一个备选方法，使用正则表达式进行拆分
-        
-        Args:
-            text: 输入的文本字符串
-            
-        Returns:
-            拆分后的句子列表
-        """
-        if not text or not text.strip():
-            return []
-            
-        try:
-            # 使用正则表达式拆分，保留分隔符
-            pattern = r'([。！？!?]|\n)'
-            parts = re.split(pattern, text)
-            
-            # 组合分隔符与前面的部分
-            sentences = []
-            for i in range(0, len(parts), 2):
-                if i + 1 < len(parts):
-                    sentences.append(parts[i] + parts[i+1])
-                else:
-                    # 处理最后一个部分（如果没有对应的分隔符）
-                    if parts[i].strip():
-                        sentences.append(parts[i])
-            
-            return sentences
-        except Exception as e:
-            logger.error(f"使用正则表达式拆分文本时发生错误: {str(e)}")
-            return [text] if text else []
-
-def main():
-    parser = argparse.ArgumentParser(description='文本句子拆分工具')
-    group = parser.add_mutually_exclusive_group(required=True)
-    group.add_argument('-t', '--text', help='直接输入要拆分的文本')
-    group.add_argument('-f', '--file', help='输入文本文件的路径')
-    
-    args = parser.parse_args()
-    
-    try:
-        # 获取输入文本
-        if args.text:
-            input_text = args.text
-        else:
-            with open(args.file, 'r', encoding='utf-8') as f:
-                input_text = f.read()
-        
-        # 执行文本拆分
-        sentences = TextSplitter.split_text(input_text)
-        
-        # 输出结果
-        print('\n拆分结果:')
-        for i, sentence in enumerate(sentences, 1):
-            print(f'{i}. {sentence}')
-            
-    except FileNotFoundError:
-        print(f'错误：找不到文件 {args.file}')
-        sys.exit(1)
-    except Exception as e:
-        print(f'错误：{str(e)}')
-        sys.exit(1)
-
-if __name__ == '__main__':
-    main()