refactor: simplify get_level_block parameters and enhance regex handling in SplitModel

【【知识库】markdown文件中前后有相同内容的标题但层级不同，分段时标题错误】
https://www.tapd.cn/tapd_fe/62980211/bug/detail/1162980211001077656

Author: liuruibin

apps/common/tests_split_model.py

@@ -0,0 +1,122 @@
+import re
+import unittest
+from types import SimpleNamespace
+
+from common.handle.impl.text.text_split_handle import TextSplitHandle
+from common.utils.split_model import SplitModel, get_split_model
+
+
+class SplitModelTestCase(unittest.TestCase):
+    @staticmethod
+    def paragraphs(text, filename='test.md'):
+        return [
+            (' '.join(row['title'].split()), row['content'].strip())
+            for row in get_split_model(filename).parse(text)
+        ]
+
+    def test_same_heading_at_different_levels(self):
+        for level in range(1, 6):
+            with self.subTest(level=level):
+                parent, child = '#' * level, '#' * (level + 1)
+                text = (
+                    f'{parent} 员工手册目录\n\n'
+                    f'{child} 第一部分：公司概述\n目录条目一\n\n'
+                    f'{child} 第二部分：人力资源\n目录条目二\n\n'
+                    f'{parent} 第一部分：公司概述\n正文内容\n'
+                )
+                self.assertEqual(self.paragraphs(text), [
+                    ('员工手册目录 第一部分：公司概述', '目录条目一'),
+                    ('员工手册目录 第二部分：人力资源', '目录条目二'),
+                    ('第一部分：公司概述', '正文内容'),
+                ])
+
+    def test_heading_text_in_body_does_not_start_section(self):
+        text = '# 目录\n引用：# 正文\n\n# 正文\n正文内容'
+        self.assertEqual(self.paragraphs(text), [
+            ('目录', '引用：# 正文'),
+            ('正文', '正文内容'),
+        ])
+
+    def test_heading_text_in_code_block_does_not_start_section(self):
+        text = '# 目录\n```markdown\n# 正文\n```\n\n# 正文\n正文内容'
+        self.assertEqual(self.paragraphs(text), [
+            ('目录', '```markdown\n# 正文\n```'),
+            ('正文', '正文内容'),
+        ])
+
+    def test_repeated_headings_keep_each_body(self):
+        text = '# 章节\n第一段提到 # 章节\n\n# 章节\n第二段'
+        self.assertEqual(self.paragraphs(text), [
+            ('章节', '第一段提到 # 章节'),
+            ('章节', '第二段'),
+        ])
+
+    def test_heading_prefix_does_not_end_section(self):
+        text = '# 前言\n引用：# 公司概述扩展\n\n# 公司概述扩展\n扩展内容\n\n# 公司概述\n正文'
+        self.assertEqual(self.paragraphs(text), [
+            ('前言', '引用：# 公司概述扩展'),
+            ('公司概述扩展', '扩展内容'),
+            ('公司概述', '正文'),
+        ])
+
+    def test_preamble_is_preserved(self):
+        text = '前言提到 # 正文\n\n# 正文\n正文内容'
+        self.assertEqual(self.paragraphs(text), [
+            ('', '前言提到 # 正文'),
+            ('正文', '正文内容'),
+        ])
+
+    def test_custom_patterns_with_capture_groups(self):
+        pattern = re.compile(r'(?m)^(# .+)$|^(## .+)$')
+        text = '# 目录\n引用：## 公司概述\n\n## 公司概述\n正文'
+        result = SplitModel([pattern], with_filter=False).parse(text)
+        self.assertEqual([(row['title'].strip(), row['content'].strip()) for row in result], [
+            ('目录', '引用：## 公司概述'),
+            ('公司概述', '正文'),
+        ])
+
+    def test_invalid_patterns_fall_back_to_plain_text(self):
+        for pattern in (None, '', '   ', '['):
+            with self.subTest(pattern=pattern):
+                self.assertEqual(SplitModel([pattern]).parse('普通文本'), [
+                    {'title': '', 'content': '普通文本'},
+                ])
+
+    def test_long_heading_keeps_text_after_title_limit(self):
+        heading = '# ' + '标题' * 150
+        rows = get_split_model('test.md').parse(heading + '\n正文')
+        self.assertEqual(len(rows), 1)
+        self.assertEqual(rows[0]['title'], heading[:255].replace('#', ''))
+        self.assertEqual(rows[0]['content'], heading[255:] + '\n正文')
+
+    def test_plain_text_splitting(self):
+        self.assertEqual(self.paragraphs('第一段\n\n第二段', 'test.txt'), [
+            ('', '第一段'),
+            ('', '第二段'),
+        ])
+
+    def test_text_upload_handler_keeps_directory_under_its_heading(self):
+        text = (
+            '# 员工手册目录\n\n'
+            '## 第一部分：公司概述\n目录条目一\n\n'
+            '## 第二部分：人力资源\n目录条目二\n\n'
+            '# 第一部分：公司概述\n\n'
+            '## 公司简介\n正文内容\n'
+        )
+        for newline in ('\n', '\r\n', '\r'):
+            with self.subTest(newline=repr(newline)):
+                result = TextSplitHandle().handle(
+                    SimpleNamespace(name='员工手册目录.md'), [], False, 100000,
+                    lambda _: text.replace('\n', newline).encode('utf-8'), None,
+                )
+                self.assertEqual([
+                    (' '.join(row['title'].split()), row['content'].strip()) for row in result['content']
+                ], [
+                    ('员工手册目录 第一部分：公司概述', '目录条目一'),
+                    ('员工手册目录 第二部分：人力资源', '目录条目二'),
+                    ('第一部分：公司概述 公司简介', '正文内容'),
+                ])
+
+
+if __name__ == '__main__':
+    unittest.main()

apps/common/utils/split_model.py

@@ -13,20 +13,18 @@
 import jieba
 
 
-def get_level_block(text, level_content_list, level_content_index, cursor):
+def get_level_block(text, level_content_list, level_content_index):
     """
     从文本中获取块数据
     :param text: 文本
     :param level_content_list: 拆分的title数组
     :param level_content_index: 指定的下标
-    :param cursor: 开始的下标位置
     :return: 拆分后的文本数据
     """
     start_content: str = level_content_list[level_content_index].get('content')
-    next_content = level_content_list[level_content_index + 1].get("content") if level_content_index + 1 < len(
-        level_content_list) else None
-    start_index = text.index(start_content, cursor)
-    end_index = text.index(next_content, start_index + 1) if next_content is not None else len(text)
+    start_index = level_content_list[level_content_index]['start']
+    end_index = level_content_list[level_content_index + 1]['start'] if level_content_index + 1 < len(
+        level_content_list) else len(text)
     return text[start_index + len(start_content):end_index], end_index
 
 
@@ -180,8 +178,20 @@ def parse_level(text, pattern: str):
     :param pattern:  正则
     :return: 符合正则的文本
     """
-    masked_text = mask_code_blocks(text)
-    level_content_list = list(map(to_tree_obj, [r[0:255] for r in re_findall(pattern, masked_text) if r is not None]))
+    if pattern is None or (isinstance(pattern, str) and not pattern.strip()):
+        return []
+    try:
+        matches = re.finditer(pattern, mask_code_blocks(text))
+    except re.error:
+        return []
+    level_content_list = []
+    for match in matches:
+        groups = range(1, match.re.groups + 1) if match.re.groups else [0]
+        for group in groups:
+            content = match.group(group)
+            if content:
+                # 保留正则匹配位置，避免同名标题在其他层级、正文或代码块中被错误定位。
+                level_content_list.append({**to_tree_obj(content[:255]), 'start': match.start(group)})
     # 过滤掉空标题或只包含#和空白字符的标题
     filtered_list = [item for item in level_content_list
                      if item['content'].strip(' ') and item['content'].replace('#', '').strip(' ')]
@@ -387,26 +397,24 @@ def parse_to_tree(self, text: str, index=0):
         level_content_list = parse_title_level(text, self.content_level_pattern, index)
         if len(level_content_list) == 0:
             return [to_tree_obj(row, 'block') for row in smart_split_paragraph(text, limit=self.limit)]
-        if index == 0 and text.lstrip().index(level_content_list[0]["content"].lstrip()) != 0:
-            level_content_list.insert(0, to_tree_obj(""))
+        if index == 0 and text[:level_content_list[0]['start']].strip():
+            level_content_list.insert(0, {**to_tree_obj(""), 'start': 0})
 
         cursor = 0
         level_title_content_list = [item for item in level_content_list if item.get('state') == 'title']
         for i in range(len(level_title_content_list)):
-            start_content: str = level_title_content_list[i].get('content')
-            if cursor < text.index(start_content, cursor):
-                for row in smart_split_paragraph(text[cursor:   text.index(start_content, cursor)], limit=self.limit):
-                    level_content_list.insert(0, to_tree_obj(row, 'block'))
-
-            block, cursor = get_level_block(text, level_title_content_list, i, cursor)
+            start_index = level_title_content_list[i]['start']
+            if cursor < start_index:
+                level_content_list[0:0] = [
+                    to_tree_obj(row, 'block')
+                    for row in smart_split_paragraph(text[cursor:start_index], limit=self.limit)
+                ]
+
+            block, cursor = get_level_block(text, level_title_content_list, i)
             if len(block) == 0:
                 continue
             children = self.parse_to_tree(text=block, index=index + 1)
             level_title_content_list[i]['children'] = children
-            first_child_idx_in_block = block.lstrip().index(children[0]["content"].lstrip())
-            if first_child_idx_in_block != 0:
-                inner_children = self.parse_to_tree(block[:first_child_idx_in_block], index + 1)
-                level_title_content_list[i]['children'].extend(inner_children)
         return level_content_list
 
     def parse(self, text: str):