51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
"""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,
|
|
)
|