张胜宇
e239eb778b
1) 客服 Agent 四份交付文档 + 构建脚手架：品牌由包装占位 XX科技 / 旧名 南方财富 统一为南方基金（热线 400-889-8899 / 官网 nffund.com），系统名改为「智能服务系统」； 同步追加 §0.4 修订记录行，工程记录行保留原占位字面以支撑硬编码扫描验收。 2) 开发文档：清理 28 份已作废/残留文档（14 份移出归档 + 14 份仓库副本）， 新增《文档规整方案与开发前待决事项-2026-09-17》。 3) 客服agent 四份交付文档首次纳入本分支。
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
|