538 lines
24 KiB
Python
538 lines
24 KiB
Python
|
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
|