import importlib.util import io import re import pytest from app.service.document_parser import ( SUPPORTED_EXTENSIONS, DocumentParser, ParsedChunk, UnsupportedDocumentError, ) DOCX_AVAILABLE = importlib.util.find_spec("docx") is not None CHUNK_SEPARATOR = "\n\n" # 段落分隔符,同时隔开 overlap 窗口与正文(见 document_parser) BLOCK_SEPARATOR = re.compile(r"\n\s*\n") # 解析器的段落切分规则(同 _BLOCK_SEPARATOR) LEADING_NEWLINE = re.compile(r"\n") # 块边界落在同一段落组两行之间时被丢掉的那一个 \n def test_supported_extensions_match_requirement() -> None: assert SUPPORTED_EXTENSIONS == frozenset({".txt", ".md", ".docx"}) def test_txt_is_split_with_overlap_and_indexed() -> None: parser = DocumentParser(chunk_size=10, chunk_overlap=2) text = "\n\n".join(f"段落{i}" + "内容" * 20 for i in range(4)) chunks = parser.parse(filename="a.txt", content=text.encode("utf-8")) assert len(chunks) > 1 assert [c.index for c in chunks] == list(range(len(chunks))) assert all(c.text.strip() for c in chunks) def test_markdown_keeps_heading_path_as_metadata() -> None: parser = DocumentParser(chunk_size=200, chunk_overlap=0) text = "# 第一章\n\n## 第一节\n\n这是正文内容,需要足够长以便成块。\n" chunks = parser.parse(filename="a.md", content=text.encode("utf-8")) assert chunks # brief 原文期望 ("第一章", "第一节"):那来自「把全文标题拼成一个 tuple」的实现缺陷。 # 本实现的 heading_path 是「该块所属」的标题层级,首块属于「第一章」。 assert chunks[0].heading_path == ("第一章",) # 同一层级下的正文块确实携带完整层级: deep = parser.parse( filename="b.md", content=("# 第一章\n\n## 第一节\n\n" + "正文内容" * 30).encode("utf-8"), ) assert any(c.heading_path == ("第一章", "第一节") for c in deep) def test_markdown_heading_path_tracks_section_per_chunk() -> None: parser = DocumentParser(chunk_size=40, chunk_overlap=0) text = ( "# 第一章\n\n" + "甲" * 60 + "\n\n## 第二节\n\n" + "乙" * 60 + "\n\n# 第二章\n\n" + "丙" * 60 ) chunks = parser.parse(filename="a.md", content=text.encode("utf-8")) paths = [c.heading_path for c in chunks] assert paths[0] == ("第一章",) assert paths[-1] == ("第二章",) assert ("第一章", "第二节") in paths assert all(path and path[0] in {"第一章", "第二章"} for path in paths) def test_markdown_heading_hierarchy_pops_back_to_parent_level() -> None: parser = DocumentParser(chunk_size=30, chunk_overlap=0) text = "# 一\n\n## 一A\n\n" + "甲" * 40 + "\n\n### 一A1\n\n" + "乙" * 40 chunks = parser.parse(filename="a.md", content=text.encode("utf-8")) assert chunks[-1].heading_path == ("一", "一A", "一A1") def test_txt_carries_no_heading_path() -> None: parser = DocumentParser(chunk_size=20, chunk_overlap=0) chunks = parser.parse(filename="a.txt", content=("正文" * 30).encode("utf-8")) assert chunks assert all(c.heading_path == () for c in chunks) def assert_union_covers(parser: DocumentParser, chunks: list[ParsedChunk], source: str) -> None: """断言切块的核心不变量(分隔符感知版)。 块的 `text` = 「至多一个 overlap 前导片段」+「原文的连续切片」,两段之间是 `CHUNK_SEPARATOR` (见 `document_parser` 模块 docstring)。因此按 `CHUNK_SEPARATOR` 切开后,**首段可能是刻意的 上下文重复**(前导片段),其余各段才是原文正文。 1. `index` 从 0 连续;每块长度不超过 `chunk_size`——解析器把前导片段与分隔符都算进预算 (`_fits()` 的 overhead、`_hard_split()` 的 budget),所以这个上界是紧的,不是 `chunk_size + chunk_overlap + 分隔符`; 2. 除首块外,同章节的块可以带一个前导片段(上一块的上下文重复):助手**不**钉死 「前导片段 == `previous_body[-chunk_overlap:]`」——那条等式在「重叠窗口自身含 `\\n\\n`」 时必然误报(`split("\\n\\n")` 的首段是空串,而窗口已整体跨过段落分隔符), 在块边界恰逢段首/行首时也不成立,**那正是这条助手误报的根源**。 现在只约束:带前导片段时其长度不超过 `chunk_overlap`;是否真是前导片段由原文走查判定; 跨章节(`heading_path` 变化)的块不得携带前导片段; 3. 正文各段按序在原文中相接:每段起点 == 上一段终点(只允许跨过被解析器规范化掉的段落 分隔符,以及块边界处被丢掉的一个 `\\n`),即无空洞、无乱序、不丢字、不伪造原文不存在的串。 走查对「首段是前导片段 / 首段就是正文」两种读法都做搜索(逐块保留所有可行位置), 不会因为贪心选错读法而误报; 4. 走查终点必须正好落在 `len(source)`(覆盖到原文结尾)。 参数取自 `parser` 实例本身,调用方无法传入与解析器实际参数矛盾的常量。 """ def skip_separator(start: int) -> int | None: """块边界处的分隔符不属于任何块:允许跨过一个空行,或被丢掉的那一个 `\\n`。""" skipped = BLOCK_SEPARATOR.match(source, start) or LEADING_NEWLINE.match(source, start) return skipped.end() if skipped is not None else None def walk(start: int, fragment_list: list[str]) -> int | None: """从 start 起按原文匹配各片段;返回新位置,不相接返回 None。""" position = start for fragment in fragment_list: if not fragment: return None if not source.startswith(fragment, position): advanced = skip_separator(position) if advanced is None or not source.startswith(fragment, advanced): return None position = advanced position += len(fragment) return position chunk_size = parser.chunk_size chunk_overlap = parser.chunk_overlap assert chunks, "解析结果为空,无法校验覆盖性" def split(chunk: ParsedChunk) -> list[str]: """按 `CHUNK_SEPARATOR` 切开,并剥掉前导空片段(窗口以 `\\n\\n` 结尾时切出来的那个)。""" fragments = chunk.text.split(CHUNK_SEPARATOR) while fragments and not fragments[0]: fragments.pop(0) return fragments def validate(number: int, chunk: ParsedChunk) -> None: assert chunk.index == number, f"第 {number} 块 index 不连续:{chunk.index}" assert len(chunk.text) <= chunk_size, ( f"第 {number} 块长度 {len(chunk.text)} 超出 chunk_size={chunk_size}" ) assert split(chunk), f"第 {number} 块为空块" def options_for(number: int, chunk: ParsedChunk) -> list[list[str]]: """两种读法:整块都是正文;若首段像上一块的尾部,则首段还可能是重叠窗口。""" fragments = split(chunk) assert fragments, f"第 {number} 块为空块" assert all(fragments), f"第 {number} 块出现空片段" options: list[list[str]] = [] if ( number and chunk_overlap and len(fragments) > 1 and len(fragments[0]) <= chunk_overlap and chunks[number - 1].heading_path == chunk.heading_path ): options.append(fragments[1:]) options.append(fragments) return options for number, chunk in enumerate(chunks): validate(number, chunk) # 逐块把「所有可行的原文位置」向前推进:首段既可能是前导片段、也可能就是正文(块边界恰逢 # 段首时窗口不会被携带),两种读法都保留;只要最终有路径正好落在 len(source) 即算通过。 # 病理输入下分支会指数膨胀,因此每一层只保留位置最靠前的 32 条(正常文档只有 1-2 条)。 positions: set[int] = {0} for number, chunk in enumerate(chunks): reached: set[int] = set() for start in sorted(positions)[:32]: for candidate in options_for(number, chunk): advanced = walk(start, candidate) if advanced is not None: reached.add(advanced) if len(reached) >= 32: break if len(reached) >= 32: break assert reached, ( f"第 {number} 块与原文接不上(存在空洞 / 乱序 / 伪造内容):{chunk.text[:24]!r}" ) positions = reached assert len(source) in positions, ( f"块序列没有覆盖到原文结尾:{sorted(positions)[:5]} / {len(source)}" ) def test_chunk_text_is_contiguous_and_never_duplicated() -> None: parser = DocumentParser(chunk_size=50, chunk_overlap=10) text = "".join(f"[{i:03d}]" for i in range(60)) assert len(text) == 300 chunks = parser.parse(filename="a.txt", content=text.encode("utf-8")) assert chunks[0].text == text[: parser.chunk_size] # 首块无前导片段,正好占满 chunk_size assert_union_covers(parser, chunks, text) def test_overlap_carries_context_forward_only_once() -> None: parser = DocumentParser(chunk_size=50, chunk_overlap=10) text = "".join(f"<{i:03d}>" for i in range(40)) chunks = parser.parse(filename="a.txt", content=text.encode("utf-8")) first, second = chunks[0].text, chunks[1].text window = first[-parser.chunk_overlap :] # 重叠窗口是刻意保留的上下文:确实会在下一块头部再出现一次(Task 5 据此估向量条数), # 但以分隔符与正文隔开,不会熔接成原文没有的串。 assert second.startswith(window + CHUNK_SEPARATOR) assert_union_covers(parser, chunks, text) def test_overlap_window_is_a_separate_leading_segment() -> None: """I1 回归:overlap 窗口必须与正文分隔,不得熔接出原文不存在的字符串。 段落取 12 字(> chunk_overlap=10),因此任何长度 10 的窗口都不会跨段落边界, 源文本中不可能出现「甲」紧跟「乙」——一旦出现即证明接缝发生了熔接。 """ parser = DocumentParser(chunk_size=40, chunk_overlap=10) source = "甲" * 12 + "\n\n" + "乙" * 12 + "\n\n" chunks = parser.parse(filename="a.txt", content=(source * 20).encode("utf-8")) assert chunks assert "".join(c.text for c in chunks).count("甲乙") == 0 for previous, current in zip(chunks, chunks[1:], strict=False): window = previous.text[-parser.chunk_overlap :] assert current.text.startswith(window + CHUNK_SEPARATOR) assert current.text[len(window)] == "\n" # 接缝处是分隔符,而不是被熔掉的正文 def test_context_window_never_leaks_across_chapters() -> None: """C1 回归:跨章节时不得把上一章的残留字符夹进下一章的块里。""" parser = DocumentParser(chunk_size=100, chunk_overlap=30) text = "# 第一章\n\n" + "甲" * 600 + "\n\n# 第二章\n\n" + "乙" * 200 chunks = parser.parse(filename="a.md", content=text.encode("utf-8")) assert chunks assert chunks[0].heading_path == ("第一章",) assert chunks[-1].heading_path == ("第二章",) for chunk in chunks: if chunk.heading_path == ("第一章",): assert "乙" not in chunk.text assert "第二章" not in chunk.text else: assert "甲" not in chunk.text assert "第一章" not in chunk.text assert all(len(c.text) <= parser.chunk_size for c in chunks) def test_long_single_paragraph_terminates_and_stays_within_budget() -> None: parser = DocumentParser(chunk_size=30, chunk_overlap=5) text = "字" * 200 chunks = parser.parse(filename="a.txt", content=text.encode("utf-8")) assert [c.index for c in chunks] == list(range(len(chunks))) assert all(len(c.text) <= parser.chunk_size for c in chunks) assert all(set(c.text) - set(CHUNK_SEPARATOR) == {"字"} for c in chunks) assert_union_covers(parser, chunks, text) def test_oversized_block_is_split_mid_text_and_every_char_survives() -> None: parser = DocumentParser(chunk_size=40, chunk_overlap=8) body = "".join(f"{i:04d}" for i in range(30)) text = "开头段落\n\n" + body + "\n\n结尾段落" chunks = parser.parse(filename="a.txt", content=text.encode("utf-8")) assert chunks[0].text == "开头段落" assert len(chunks[0].text) < parser.chunk_size # 短段落独立成块,不硬切 assert chunks[-1].text.endswith("结尾段落") # 原文每一段都必须落在某一块里(硬切不许丢字) assert_union_covers(parser, chunks, text) def test_block_start_does_not_gain_an_extra_separator() -> None: """D1 回归:块正文从段首开始,不得在块首伪造出原文没有的 `\\n\\n`。 第二个场景是原文本来就含连续空行(`\\n\\n\\n`):块内只保留一个段落分隔符, 不能因为「空串哨兵」参与拼接而放大成更多换行。 """ parser = DocumentParser() heading_source = "# 第一章\n\n正文内容。" heading_only = parser.parse(filename="a.md", content=heading_source.encode("utf-8")) assert [c.text for c in heading_only] == ["# 第一章\n\n正文内容。"] extra_blank_source = "甲\n\n\n乙" extra_blank = parser.parse(filename="a.txt", content=extra_blank_source.encode("utf-8")) assert [c.text for c in extra_blank] == ["甲\n\n乙"] def test_short_tail_paragraph_does_not_emit_extra_blank_lines() -> None: """D1d 回归:硬切后的短尾段不得被拼出原文没有的四连换行。 原文只有 `\\n\\n`(3000 个「字」+ 空行 + 短尾段),任何一个块里都不该出现 3 个及以上连续换行。 """ parser = DocumentParser() text = "字" * 3000 + "\n\n" + "结尾短段" chunks = parser.parse(filename="a.txt", content=text.encode("utf-8")) assert chunks[-1].text.endswith("结尾短段") assert all("\n\n\n\n" not in c.text for c in chunks) assert not any(re.search(r"\n{3,}", c.text) for c in chunks), [c.text[-16:] for c in chunks] assert_union_covers(parser, chunks, text) def test_helper_tolerates_single_newline_dropped_at_block_boundary() -> None: """I-1b 回归:块边界落在同一段落组两行之间时,被丢掉的那一个 `\\n` 不该让助手误报。 解析器输出 `['abcdefghij', 'klmnopqrst']` 是正确的(块的正文从行首开始,边界上的分隔符 不属于任何块,见模块 docstring 的「已知限制 1」);助手原来只容忍跨 `\\n\\s*\\n` 的丢弃。 """ parser = DocumentParser(chunk_size=12, chunk_overlap=0) source = "abcdefghij\nklmnopqrst" chunks = parser.parse(filename="a.txt", content=source.encode("utf-8")) assert [c.text for c in chunks] == ["abcdefghij", "klmnopqrst"] assert_union_covers(parser, chunks, source) def test_helper_accepts_overlap_window_that_contains_a_paragraph_separator() -> None: """I-1 回归:`chunk_overlap` >= 某个段落长度时,重叠窗口自身含 `\\n\\n`,助手不得误报。 最小反例:源 `'甲甲甲\\n\\n乙乙乙\\n\\n丙丙丙丙'`,`cs=10`/`co=5`——块 1 正文恰好 10 字 (装满),块 1 的窗口 `'乙乙乙\\n\\n'` 跨了段落分隔符;助手对块 2 取 `chunk.text.split("\\n\\n")` 得到的首段是空串(窗口以 `\\n\\n` 结尾),旧助手拿空串去比 overlap 窗口必然失败。块 2、块 3 的正文本身与原文完全相接(无伪造字符、无空洞), 缺口在助手:它假定「块首无空片段」。 """ parser = DocumentParser(chunk_size=10, chunk_overlap=5) source = "甲甲甲\n\n乙乙乙\n\n丙丙丙丙" chunks = parser.parse(filename="a.txt", content=source.encode("utf-8")) assert [c.text for c in chunks] == ["甲甲甲\n\n乙乙乙", "\n\n乙乙乙\n\n丙丙丙", "丙丙丙\n\n丙"] assert_union_covers(parser, chunks, source) def test_helper_still_rejects_tampered_output() -> None: """反证:助手放宽后仍必须抓住真缺陷(伪造字符 / 丢字 / 丢块 / 伪造前导片段)。""" parser = DocumentParser(chunk_size=10, chunk_overlap=5) source = "甲甲甲\n\n乙乙乙\n\n丙丙丙丙" chunks = parser.parse(filename="a.txt", content=source.encode("utf-8")) assert_union_covers(parser, chunks, source) # 未篡改时通过 def tampered(text: str) -> list[ParsedChunk]: return [ chunks[0], ParsedChunk(text=text, heading_path=chunks[1].heading_path, index=1), ] cases = [ tampered(chunks[1].text.replace("丙", "丁", 1)), # 篡改 1 字 tampered(chunks[1].text[:-1]), # 删掉尾部 1 字 tampered("伪造" + CHUNK_SEPARATOR + chunks[1].text), # 伪造前导片段 ] for bad in cases: with pytest.raises(AssertionError): assert_union_covers(parser, bad, source) with pytest.raises(AssertionError): # 少一块:覆盖不到原文结尾 assert_union_covers(parser, chunks[:-1], source) def test_no_chunk_is_pure_overlap_window() -> None: """I-2 回归:`chunk_overlap` 接近 `chunk_size` 时不得产出「零新内容」的纯重叠块。 `cs=10`/`co=9`/源 11 字:旧行为第 2 块 = 窗口 8 字 + 正文 1 字(几乎全是上一块的重复); 现在窗口被缩到 7 字、正文 3 字。判据是**窗口 + 分隔符 + 1 字正文 <= chunk_size**, 即每块至少留得下 1 个新字符(这正是旧实现会突破的界)。 """ for chunk_size, source in [(10, "字" * 11), (512, "字" * 513), (10, "字" * 30)]: parser = DocumentParser(chunk_size=chunk_size, chunk_overlap=chunk_size - 1) chunks = parser.parse(filename="a.txt", content=source.encode("utf-8")) assert len(chunks) > 1 assert all(len(c.text) <= chunk_size for c in chunks) for number, chunk in enumerate(chunks): fragments = chunk.text.split(CHUNK_SEPARATOR) while fragments and not fragments[0]: fragments.pop(0) if number: window = fragments[0] if len(fragments) > 1 else "" body = CHUNK_SEPARATOR.join(fragments[1:]) if len(fragments) > 1 else fragments[0] assert len(window) + len(CHUNK_SEPARATOR) + 1 <= chunk_size, ( f"cs={chunk_size} 第 {number} 块的窗口 {len(window)} 字没有给正文留出位置" ) assert body, f"cs={chunk_size} 第 {number} 块只有窗口、没有正文" assert len(window) < parser.chunk_overlap, ( f"cs={chunk_size} 第 {number} 块窗口仍是满额 {len(window)} 字:正文只剩 " f"{len(body)} 字,等于纯重叠" ) assert_union_covers(parser, chunks, source) def test_no_chunk_exceeds_chunk_size_when_overlap_almost_fills_the_budget() -> None: """I-2b 回归:`chunk_overlap` 接近 `chunk_size` 时**硬切**也不得产出超长块。 `cs=512`/`co=511`/源 1200 字:第 4 块曾切成「窗口 510 字 + 正文 1 字」= 512 字 (窗口 510 与传入的 509 前导片段叠加,超出 `chunk_size` 的预算), 现在硬切会先把传入的前导片段裁到「装得下 1 字正文」。 """ for chunk_size, overlap, source in [ (512, 511, "字" * 1200), (10, 9, "字" * 30), (20, 19, "字" * 60), (64, 63, "字" * 200), ]: parser = DocumentParser(chunk_size=chunk_size, chunk_overlap=overlap) chunks = parser.parse(filename="a.txt", content=source.encode("utf-8")) assert len(chunks) > 2 # 走到硬切路径 assert all(len(c.text) <= chunk_size for c in chunks), [ len(c.text) for c in chunks ] def test_single_newline_inside_a_paragraph_is_preserved() -> None: """D4 回归:段落组内的单换行按原文保留,不得被规范化成 `\\n\\n`。""" parser = DocumentParser() source = "a\nb" chunks = parser.parse(filename="a.txt", content=source.encode("utf-8")) assert [c.text for c in chunks] == ["a\nb"] assert_union_covers(parser, chunks, source) three_lines = "甲\n乙\n丙" assert [c.text for c in parser.parse(filename="a.md", content=three_lines.encode("utf-8"))] == [ "甲\n乙\n丙" ] def test_empty_and_whitespace_only_files_yield_no_chunks() -> None: parser = DocumentParser() assert parser.parse(filename="a.txt", content=b"") == [] assert parser.parse(filename="empty.md", content=b" \n\n\t \n") == [] def test_undecodable_bytes_are_replaced_not_raised() -> None: parser = DocumentParser(chunk_size=50, chunk_overlap=0) chunks = parser.parse(filename="a.txt", content=b"\xff\xfe\x00bad\x80" * 10) assert chunks assert all(c.text.strip() for c in chunks) def test_uppercase_extension_is_accepted() -> None: parser = DocumentParser(chunk_size=50, chunk_overlap=0) chunks = parser.parse(filename="FAQ.TXT", content=("内容" * 40).encode("utf-8")) assert chunks @pytest.mark.parametrize("filename", ["a.pdf", "a.docx.txt.pdf", "noextension", "a.txt.exe"]) def test_unsupported_extension_is_rejected(filename: str) -> None: parser = DocumentParser() with pytest.raises(UnsupportedDocumentError): parser.parse(filename=filename, content=b"%PDF-1.4") def test_unsupported_extension_is_rejected_with_brief_example() -> None: parser = DocumentParser() with pytest.raises(UnsupportedDocumentError): parser.parse(filename="a.pdf", content=b"%PDF-1.4") def test_docx_bytes_are_parsed_into_text_and_heading_path() -> None: if not DOCX_AVAILABLE: pytest.skip("python-docx 未安装,docx 支持降级(见 task-4 报告)") docx = pytest.importorskip("docx") document = docx.Document() document.add_heading("产品说明", level=1) document.add_paragraph("场内基金模拟交易说明。" * 5) document.add_heading("费用", level=2) document.add_paragraph("申购费率与赎回费率。" * 20) buffer = io.BytesIO() document.save(buffer) chunks = DocumentParser(chunk_size=100, chunk_overlap=0).parse( filename="guide.docx", content=buffer.getvalue() ) assert chunks assert chunks[0].heading_path == ("产品说明",) assert chunks[-1].heading_path == ("产品说明", "费用") assert any("费用" in c.text for c in chunks) def test_docx_without_dependency_raises_unsupported() -> None: if DOCX_AVAILABLE: pytest.skip("python-docx 已安装,走真实解析路径") with pytest.raises(UnsupportedDocumentError, match="python-docx"): DocumentParser().parse(filename="guide.docx", content=b"PK\x03\x04not-a-real-docx") def test_unreadable_docx_is_rejected_with_clear_error() -> None: if not DOCX_AVAILABLE: pytest.skip("python-docx 未安装,docx 支持降级(见 task-4 报告)") with pytest.raises(UnsupportedDocumentError): DocumentParser().parse(filename="guide.docx", content=b"PK\x03\x04not-a-real-docx") @pytest.mark.parametrize( ("chunk_size", "chunk_overlap"), [(0, 0), (-1, 0), (10, -1), (10, 10), (10, 20)], ) def test_invalid_chunk_parameters_are_rejected(chunk_size: int, chunk_overlap: int) -> None: with pytest.raises(ValueError): DocumentParser(chunk_size=chunk_size, chunk_overlap=chunk_overlap) def test_defaults_match_requirement() -> None: parser = DocumentParser() assert parser.chunk_size == 512 assert parser.chunk_overlap == 64