knowledge/tests/service/test_trunks_service.py

import regex
from pathlib import Path

import pytest

from service.kg_edge_service import KGEdgeService
from service.kg_node_service import KGNodeService
from service.kg_prop_service import KGPropService
from service.trunks_service import TrunksService
from model.trunks_model import Trunks
from sqlalchemy.exc import IntegrityError

from utils import DeepseekUtil
from db.session import get_db

@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_search2(self, trunks_service):
        page = 1
        limit = 100
        file_path = '急诊医学（第2版'
        while True:
            results = trunks_service.paginated_search_by_type_and_filepath(
                {'pageNo': page, 'limit': limit, 'type': 'trunk', 'file_path': file_path})
            if not results['data']:
                break
            for record in results['data']:
                print(f"{record['id']}{record['type']}{record['title']}{record['file_path']}")
                if record['type'] != 'trunk' or file_path not in record['file_path']:
                    print('出现异常数据')
                    break

                page_no = self.get_page_no(record['content'], trunks_service, file_path)
                if page_no is None:
                    print(f"{record['id']}找到page_no: {page_no}")
                    continue
                trunks_service.update_trunk(record['id'], {'page_no': page_no})
            page += 1

    def test_vector_search(self, trunks_service):
        page = 1
        limit = 100
        file_path='trunk2'
        while True:
            results = trunks_service.paginated_search_by_type_and_filepath({'pageNo': page, 'limit': limit, 'type': 'trunk', 'file_path': file_path})
            if not results['data']:
                break
            for record in results['data']:
                print(f"{record['id']}{record['type']}{record['title']}{record['file_path']}")
                if record['type'] != 'trunk' or file_path not in record['file_path']:
                    print('出现异常数据')
                    break

                page_no = self.get_page_no(record['content'],trunks_service,file_path)
                if page_no is None:
                    print(f"{record['id']}找到page_no: {page_no}")
                    continue
                trunks_service.update_trunk(record['id'], {'page_no': page_no})
            page += 1

    def test_trunk_search(self, trunks_service):
        page = 1
        limit = 100

        while True:
            results = trunks_service.paginated_search_by_type_and_filepath({'pageNo': page, 'limit': limit, 'type': 'trunk'})
            if not results['data']:
                break
    
            for record in results['data']:
                prompt = '''
##角色任务
你是一个医学专家，且精通中文文本的实体和关系标注（自然语言处理），用户将提供两个医学文本：“前置医学文本”和“任务医学文本”，你的任务是从用户输入的“任务医学文本”中抽取所有的“关系型”知识要点，构建符合「头部实体---关系---尾部实体」的三元组集合。

##请严格依次执行以下步骤及其细节要求：
第一步，通读“前置医学文本”和“任务医学文本”，对“任务医学文本”形成一个整体的理解，搞明白其围绕的主题是什么？主要讲的是什么？主要说了哪几块或几点内容？。此处“前置医学文本”是“任务医学文本”的上文语义背景，其有助于你对“任务医学文本”语义的理解，且在你提取三元组时，其可能作为参考或辅助信息。

第二步【实体识别】：
规则1：可参考但不限于如下医学实体类型：疾病、药品、药品剂型、症状、体格检查项目、体格检查结果、手术和操作、实验室检查套餐、实验室检查子项目、辅助检查项目、辅助检查子项目、辅助检查描述、辅助检查结果、输血类型、麻醉、科室、性别、人群、食物、其他过敏原、医疗器械及物品、给药途径、部位、护理、量表、单位、ICD10疾病类别、药品化学物质类别、药品治疗学类别、药品药理学类别、药品解剖学类别、症状类别、手术和操作类别、ICD10疾病类别根节点、科室疾病类别根节点、药品化学物质类别根节点、药品治疗学类别根节点、药品药理学类别根节点、药品解剖学类别根节点、症状类别根节点、手术和操作类别根节点、实验室检查类别根节点、辅助检查类别根节点、年龄、疾病系统分类、性质、中医疾病、中医证候、诱因、政策法规、否定词、疾病集合、药品通用名集合、药品剂型集合、症状集合、体格检查项目集合、体格检查结果集合、手术和操作集合、实验室检查套餐集合、辅助检查项目集合、辅助检查子项目集合、辅助检查描述集合、辅助检查结果集合、麻醉集合、科室集合、食物集合、其他过敏原集合、医疗器械及物品集合、部位集合、中医疾病集合、中医证候集合、诱因集合、给药途径集合、物理治疗、经典病例、历史病例检查、检查结果、手术操作、其他治疗、人群、人体结构或部位、医疗器械、食物、病理机制等。
规则2：复合实体要拆分：如“老年人和免疫功能低下者”不能作为一个实体，需拆分为："老年人"、"免疫功能低下者"两个实体。
规则3：文字中省略了，但语义中暗含的文字内容要进行语义补全：如文本：“胸、腹部检查”中，这里的“胸”其实是指“胸部检查”，按语义补全后应该有两个实体：胸部检查、腹部检查。同理，文本：“注意有无心内膜炎、心肌炎、心包炎体征”中，应该有三个实体：内膜炎体征、心肌炎体征、心包炎体征；文本：“有无肝脏和脾脏肿大”中，应该有两个实体：肝脏肿大、脾脏肿大。

第三步【关系构建】：
规则4：可参考的“基础关系类型”如下（但不仅仅限于这些）：属于（是）、包括（包含）、导致（的结果）、是由…导致（的结果）、的原因是、是…的原因、的病因是、是…的病因、基于（基础是）、是…的基础、推荐、被推荐于、区别于、相似于、关联、疾病常关联、疾病可推荐等等。
规则5：根据上下文整体语义分析并确定出两个实体之间的“基础关系类型”后，可能还不足以描述清楚两个实体间的“详细关系”或“精准的关系”，所以你要尽量构建字数更多的详细的“关系”，需要依据原文中真实的语义，在“基础关系”中增加“修饰词”、“限定条件词”等语义描述，避免关系构建太过粗糙，避免原文关系语义信息的衰减或丢失。
规则6：语义中属于该三元组关系的修饰词、限定条件词等，需要融入到该“关系”中（如：可能、少数、30%、多数、显著、轻微、手术后、治疗无效时、满足XX条件时等），如果归属于该“关系”，则需要组合到该“关系”中使得关系更丰满（如：“可能导致”、“少数由…导致”、“手术后导致”、“满足XX条件时，推荐”等等）。
规则7：“尾部实体”不能为复合实体或并排结构，必须合理的拆分为多个三元组。
规则8：强制质量检测：“头部实体---关系类型---尾部实体”中，三者必须能组合成语法通顺的一句话，不能有语病，且关系指向不能模糊或错误。如三元组1：“A---的原因是---B”中，组合成一句话“A的原因是B”没有语病，且关系明确：B是原因，A是B原因导致的结果；如三元组2：“A---是…的原因---B”中，组合成一句话“A是B的原因”没有语病，且关系明确：A是原因，B是A原因导致的结果。注意：不是所有的关系都是镜像对称的，很多关系是单向关系，请根据语义关系的方向选择正确的关系类型，否则语义会完全相反。

##输出格式要求：
对于每个三元组关系，提取以下信息：头部实体的名称（source_entity）、尾部实体的名称（target_entity）、关系（relationship_type）、头部实体的类型（source_entity_type）、尾部实体的类型（target_entity_type），最后这些信息整理成如下格式的一个“字符串”：“头部实体的名称---关系---尾部实体的名称---头部实体的类型---尾部实体的类型”，不同的三元组“字符串”之间用换行符号连接起来。

##质量红线
1.禁止出现四元组或嵌套结构
2.禁止合并多个“差异点”或“并列实体”到单个三元组
3.每个三元组必须保留原文核心逻辑

##示例：
“前置医学文本”为：
第一节，自发性气胸
“任务医学文本”为：
"三、诊断要点
自发性气胸通过胸部 X 线片确立诊断，条件允许时，应选择直立位拍片。

1. 既往胸部 X 线检查无明显病变或有 COPD、肺结核、哮喘等肺部基础病变。
2. 突发一侧胸痛伴不同程度的胸闷、呼吸困难。患侧胸廓饱满、呼吸运动减弱，叩诊呈鼓音，肝、 肺浊音界消失，听诊呼吸音减弱，甚至消失。
3. 发病时胸部 X 线影像学检查是诊断气胸最为准确和可靠的方法。

典型自发性气胸诊断不难。继发性气胸病人可因原有基础疾病而影响诊断，因此，对临床不能用 其他原因解释或经急诊处理呼吸困难无改善者，需考虑自发性气胸的可能。因病情危重不能立即行 胸部 X 线检查时，可在胸腔积气体征最明显处进行诊断性穿刺。"
输出为：
自发性气胸---通过...确立诊断---胸部X线片---疾病---检查
自发性气胸---条件允许时推荐选择---直立位拍片---疾病---检查操作
自发性气胸---诊断要点需结合---既往胸部X线检查无明显病变---疾病---检查结果
自发性气胸---诊断要点需结合---或有肺部基础病变---疾病---检查结果
肺部基础病变---的原因包括---COPD---检查结果---疾病
肺部基础病变---的原因包括---肺结核---检查结果---疾病
肺部基础病变---的原因包括---哮喘---检查结果---疾病
自发性气胸---诊断要点需结合---突发一侧胸痛伴不同程度的胸闷
自发性气胸---诊断要点需结合---突发一侧胸痛伴不同程度的呼吸困难
突发一侧胸痛---伴发---胸闷---症状---症状
突发一侧胸痛---伴发---呼吸困难---症状---症状
自发性气胸---患侧表现---患侧胸廓饱满---疾病---体征结果
自发性气胸---患侧表现---呼吸运动减弱---疾病---体征结果
自发性气胸---患侧表现---叩诊呈鼓音---疾病---体征结果
自发性气胸---体征结果---肝脏浊音界消失---疾病---体征结果
自发性气胸---体征结果---肺浊音界消失---疾病---体征结果
自发性气胸---患侧表现---肺浊音界消失---疾病---体征结果
自发性气胸---患侧表现---听诊呼吸音减弱---疾病---体征结果
自发性气胸---患侧表现---听诊呼吸音消失---疾病---体征结果
胸部X线影像学检查---是诊断...最准确可靠方法---气胸---检查---疾病
继发性气胸病人---可因...被影响诊断---原有基础疾病---人群---疾病
临床不能用其他原因解释的呼吸困难者---需考虑可能为---自发性气胸---人群---疾病
急诊处理呼吸困难无改善者---需考虑可能为---自发性气胸---人群---疾病
病情危重不能立即行胸部X线检查者---推荐进行---诊断性穿刺---人群---检查操作
诊断性穿刺---实施部位---胸腔积气体征最明显处---检查操作---人体部位

用户输入的“前置医学文本”为：
'''
                prompt+=record['meta_header']+'\n任务文本为：\n'
                prompt = prompt + record['content']
                llm_result = DeepseekUtil.chat(prompt)
                if not llm_result or not isinstance(llm_result, str):
                    print(f"LLM返回结果无效: {type(llm_result)}")
                    continue
                
                for line in llm_result.strip().split('\n'):
                    try:
                        if not line.strip():
                            continue
                            
                        # 验证行格式
                        if line.count('---') != 4:
                            print(f"无效的三元组格式: {line}")
                            continue
                            
                        # 解析结果
                        parts = line.strip().split('---')
                        if len(parts) != 5:
                            print(f"解析失败，部分数量不符: {len(parts)} parts in {line}")
                            continue

                        start_node_name, relation_name, end_node_name, start_category, end_category = parts

                        # 创建或获取起始节点和结束节点
                        start_node_id = self._create_or_get_node(start_node_name, start_category)
                        end_node_id = self._create_or_get_node(end_node_name, end_category)
                        edgeService = KGEdgeService(next(get_db()))
                        #edges = edgeService.get_edges_by_nodes(start_node_id, end_node_id, relation_name)
                        #if len(edges) == 0:
                        edge_data = {}
                        edge_data['src_id'] = start_node_id
                        edge_data['dest_id'] = end_node_id
                        edge_data['name'] = relation_name
                        edge_data['category'] = relation_name
                        edge_data['version'] = 'unk'
                        edge = edgeService.create_edge(edge_data)

                        propService = KGPropService(next(get_db()))
                        props_by_ref_id = propService.get_prop_by_ref_id(edge.id, 'trunk_ids')
                        if props_by_ref_id:
                            if record['id'] not in props_by_ref_id['trunk_ids']:
                                props_by_ref_id['trunk_ids'].append(record['id'])   
                            propService.update_prop(edge.id, {'trunk_ids': props_by_ref_id['trunk_ids']})
                            continue

                        prop_data = {}
                        prop_data['ref_id'] = edge.id
                        prop_data['category'] = 2
                        prop_data['type'] = 2
                        prop_data['prop_name'] = 'trunk_ids'
                        prop_data['prop_value'] = [record['id']]
                        prop_data['prop_title'] = '切片id列表'
                        propService.create_prop(prop_data)
                    except Exception as e:
                        print(f"处理行时发生异常: {str(e)}")
                        continue

        page += 1

    def _create_or_get_node(self, node_name: str, category: str) -> int:
        node_service = KGNodeService(next(get_db()))
        node = node_service.get_node_by_name_category(node_name, category)
        if node:
            return node['id']
        node_data = {}
        node_data['name'] = node_name
        node_data['category'] = category
        node_data['version'] = 'unk'
        node_data['status'] = 0

        return node_service.create_node(node_data).id
    def get_page_no(self, text: str, trunks_service,file_path:str) -> int:
        results = trunks_service.search_by_vector(text,1000,type='page',file_path=file_path,conversation_id="1111111aaaa")
        sentences = self.split_text(text)
        count = 0
        for r in results:
            #将r["content"]的所有空白字符去掉
            content = regex.sub(r'[^\w\d\p{L}]', '', r["content"])
            count+=1
            match_count = 0
            length = len(sentences)/2
            for sentence in sentences:
                sentence = regex.sub(r'[^\w\d\p{L}]', '', sentence)
                if sentence in content:
                    match_count += 1
                    if match_count >= 2:
                        return r["page_no"]

    def test_match_trunk(self,trunks_service) -> int:
        must_matchs = ['心肌梗死']
        keywords = [ '概述']
        text = '''- 主要病因：
1. 冠状动脉粥样硬化（占90%以上）
2. 冠状动脉栓塞（如房颤血栓脱落）
3. 冠状动脉痉挛（可卡因滥用等）
- 危险因素：
1. 吸烟（RR=2.87）
2. 高血压（RR=2.50）
3. LDL-C≥190mg/dL（RR=4.48）
- 遗传因素：
家族性高胆固醇血症（OMIM#143890）'''
        text = regex.sub(r'[^\w\d\p{L}]', '', text)
        results = trunks_service.search_by_vector(text,1000,distance=0.72,type='trunk')
        print(f"原结果: {results[0]["meta_header"]}")
        print(results[0]["content"])
        max_match_count = 0
        best_match = None
        for r in results:
            if all(must_match in r["content"] or must_match in r["meta_header"] for must_match in must_matchs):
                match_count = sum(keyword in r["content"] for keyword in keywords)
                if match_count > max_match_count:
                    max_match_count = match_count
                    best_match = r
                elif best_match is None and max_match_count == 0:
                    best_match = r
        if best_match:
            print(f"最佳匹配: {best_match["title"]}")
            print(best_match["content"])
        return best_match
           
    def split_text(self, text):
        """将文本分割成句子"""
        print(text)
        # 使用常见的标点符号作为分隔符
        delimiters = ['!', '?', '。', '！', '？', '\n', ';', '。', '；']
        sentences = [text]
        for delimiter in delimiters:
            new_sentences = []
            for sentence in sentences:
                parts = sentence.split(delimiter)
                new_sentences.extend([part + delimiter if i < len(parts) - 1 else part for i, part in enumerate(parts)])
            sentences = [s.strip() for s in new_sentences if s.strip()]
        
        # 合并短句子
        merged_sentences = []
        buffer = ""
        for sentence in sentences:
            buffer += " " + sentence if buffer else sentence
            if len(buffer) >= 10:
                merged_sentences.append(buffer)
                buffer = ""
        if buffer:
            merged_sentences.append(buffer)
        
        # 打印最终句子
        for i, sentence in enumerate(merged_sentences):
            print(f"句子{i+1}: {sentence.replace(" ","").replace("\u2003", "").replace("\u2002", "").replace("\u2009", "").replace("\n", "").replace("\r", "")}")
        
        return merged_sentences
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):
        # 使用现有目录路径
        base_path = Path(r'E:\project\vscode\《急诊医学（第2版）》')
        
        # 遍历目录并创建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