"""Conservative text normalization before chunking knowledge documents.""" from __future__ import annotations import re from dataclasses import dataclass _PAGE_MARKER = re.compile(r"^\s*(?:第\s*\d+\s*页|page\s+\d+)\s*$", re.I) @dataclass(frozen=True) class CleanedDocument: text: str warnings: list[str] changed: bool def clean_document_text(text: str) -> CleanedDocument: """Normalize parser output without changing business values or punctuation.""" if not isinstance(text, str): raise TypeError("document text must be a string") original = text warnings: list[str] = [] text = text.replace("\ufeff", "").replace("\r\n", "\n").replace("\r", "\n") if text != original: warnings.append("已统一编码标记和换行符") cleaned_lines: list[str] = [] removed_page_markers = 0 for line in text.split("\n"): line = line.replace("\u00a0", " ") line = "".join(char for char in line if char in "\t\n" or ord(char) >= 32) if _PAGE_MARKER.match(line): removed_page_markers += 1 continue line = re.sub(r"[ \t]+", " ", line).strip() cleaned_lines.append(line) if removed_page_markers: warnings.append(f"已移除{removed_page_markers}个独立页码标记") normalized = "\n".join(cleaned_lines) normalized = re.sub(r"\n{3,}", "\n\n", normalized).strip() if normalized != text.strip(): warnings.append("已清理多余空白、控制字符或空行") return CleanedDocument( text=normalized, warnings=warnings, changed=normalized != original, )