feat(data-process): 清理PDF文档级噪声
This commit is contained in:
caoxiaozhu
@@ -175,6 +175,15 @@ class PdfPageText:
|
||||
source_end: int
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class DocumentNoiseSpan:
|
||||
"""PDF 中可安全从展示内容移除的文本范围。"""
|
||||
|
||||
start: int
|
||||
end: int
|
||||
kind: Literal["page_number", "repeated_margin", "table_of_contents"]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class TextChunk:
|
||||
"""带有可追溯来源位置的非结构化文本切片。"""
|
||||
@@ -496,6 +505,240 @@ def extract_pdf_page_texts(raw: bytes) -> tuple[PdfPageText, ...]:
|
||||
return tuple(pages)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _PdfLine:
|
||||
text: str
|
||||
start: int
|
||||
end: int
|
||||
|
||||
|
||||
_PDF_PAGE_NUMBER_LINE_PATTERN = re.compile(
|
||||
r"^(?:页次\s*)?(?:第\s*)?(?P<page>\d+)\s*页\s*"
|
||||
r"(?:(?:[/／]\s*)?共\s*(?P<total>\d+)\s*页)?$"
|
||||
)
|
||||
_PDF_FRACTION_PAGE_LINE_PATTERN = re.compile(
|
||||
r"^[—–-]?\s*(?P<page>\d+)\s*[/／]\s*(?P<total>\d+)\s*[—–-]?$"
|
||||
)
|
||||
_PDF_CLASSIFICATION_LABEL_PATTERN = re.compile(
|
||||
r"^(?:(?:秘密等级|密级)\s*)?(?:商密|秘密|机密|绝密)"
|
||||
r"\s*(?:[【\[(（][^】\])）]{1,8}[】\])）])?$"
|
||||
)
|
||||
_TOC_TITLE_PATTERN = re.compile(r"^(?:目\s*录|contents)$", re.IGNORECASE)
|
||||
_TOC_LEADER_ENTRY_PATTERN = re.compile(
|
||||
r"(?:[.．…·•]\s*){3,}\s*\d{1,4}\s*$"
|
||||
)
|
||||
_TOC_NUMBERED_ENTRY_PATTERN = re.compile(
|
||||
r"^(?:第[\u3400-\u4dbf\u4e00-\u9fff]{1,12}章|附表\s*\d+|\d+(?:\.\d+)+)"
|
||||
r"\s+.+\s+\d{1,4}\s*$",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_MARGIN_TEMPLATE_KEYWORDS = (
|
||||
"页",
|
||||
"页次",
|
||||
"版本",
|
||||
"文件编码",
|
||||
"秘密等级",
|
||||
"密级",
|
||||
"商密",
|
||||
"confidential",
|
||||
)
|
||||
|
||||
|
||||
def _pdf_page_lines(page: PdfPageText) -> tuple[_PdfLine, ...]:
|
||||
lines: list[_PdfLine] = []
|
||||
local_offset = 0
|
||||
for raw_line in page.text.splitlines(keepends=True):
|
||||
content = raw_line.rstrip("\r\n")
|
||||
leading = len(content) - len(content.lstrip())
|
||||
trailing = len(content.rstrip())
|
||||
text = content.strip()
|
||||
if text:
|
||||
lines.append(
|
||||
_PdfLine(
|
||||
text=text,
|
||||
start=page.source_start + local_offset + leading,
|
||||
end=page.source_start + local_offset + trailing,
|
||||
)
|
||||
)
|
||||
local_offset += len(raw_line)
|
||||
return tuple(lines)
|
||||
|
||||
|
||||
def _is_standalone_page_number(
|
||||
text: str,
|
||||
*,
|
||||
physical_page: int,
|
||||
page_count: int,
|
||||
) -> bool:
|
||||
normalized = unicodedata.normalize("NFKC", text).strip()
|
||||
match = _PDF_PAGE_NUMBER_LINE_PATTERN.fullmatch(normalized)
|
||||
if match is None:
|
||||
match = _PDF_FRACTION_PAGE_LINE_PATTERN.fullmatch(normalized)
|
||||
if match is None or int(match.group("page")) != physical_page:
|
||||
return False
|
||||
total = match.groupdict().get("total")
|
||||
return total is None or int(total) == page_count
|
||||
|
||||
|
||||
def _margin_signature(text: str) -> str:
|
||||
normalized = unicodedata.normalize("NFKC", text).casefold()
|
||||
normalized = re.sub(r"\s+", " ", normalized).strip()
|
||||
if any(keyword in normalized for keyword in _MARGIN_TEMPLATE_KEYWORDS):
|
||||
normalized = re.sub(r"\d+", "#", normalized)
|
||||
return normalized
|
||||
|
||||
|
||||
def _has_margin_metadata_keyword(text: str) -> bool:
|
||||
normalized = unicodedata.normalize("NFKC", text).casefold()
|
||||
return any(keyword in normalized for keyword in _MARGIN_TEMPLATE_KEYWORDS)
|
||||
|
||||
|
||||
def _has_meaningful_margin_signature(signature: str) -> bool:
|
||||
return len(re.sub(r"[#\W_]+", "", signature, flags=re.UNICODE)) >= 2
|
||||
|
||||
|
||||
def _is_toc_leader_entry(text: str) -> bool:
|
||||
return bool(_TOC_LEADER_ENTRY_PATTERN.search(text))
|
||||
|
||||
|
||||
def _is_toc_numbered_entry(text: str) -> bool:
|
||||
return bool(_TOC_NUMBERED_ENTRY_PATTERN.fullmatch(text))
|
||||
|
||||
|
||||
def detect_pdf_document_noise(
|
||||
pages: Sequence[PdfPageText],
|
||||
) -> tuple[DocumentNoiseSpan, ...]:
|
||||
"""识别 PDF 中的独立页码、重复页边内容和高置信目录。
|
||||
|
||||
规则只查看每页顶部 5 行和底部 3 行来推断页眉页脚；目录必须有
|
||||
明显的点引导线密度，避免仅因正文中出现“目录”或章节标题而误删。
|
||||
"""
|
||||
|
||||
page_lines = tuple(_pdf_page_lines(page) for page in pages)
|
||||
detected: dict[tuple[int, int], DocumentNoiseSpan] = {}
|
||||
|
||||
def mark(
|
||||
line: _PdfLine,
|
||||
kind: Literal["page_number", "repeated_margin", "table_of_contents"],
|
||||
) -> None:
|
||||
detected.setdefault(
|
||||
(line.start, line.end),
|
||||
DocumentNoiseSpan(
|
||||
start=line.start,
|
||||
end=line.end,
|
||||
kind=kind,
|
||||
),
|
||||
)
|
||||
|
||||
for page, lines in zip(pages, page_lines, strict=True):
|
||||
for line in lines:
|
||||
if _is_standalone_page_number(
|
||||
line.text,
|
||||
physical_page=page.page_number,
|
||||
page_count=len(pages),
|
||||
):
|
||||
mark(line, "page_number")
|
||||
outer_margin_lines = (*lines[:2], *lines[-2:])
|
||||
for line in outer_margin_lines:
|
||||
if _PDF_CLASSIFICATION_LABEL_PATTERN.fullmatch(line.text):
|
||||
mark(line, "repeated_margin")
|
||||
|
||||
# 只在三页及以上文档中推断通用页眉页脚，避免短文档误删。
|
||||
if len(pages) >= 3:
|
||||
signature_pages: dict[str, set[int]] = {}
|
||||
candidate_lines: list[tuple[int, _PdfLine, str]] = []
|
||||
for page_index, lines in enumerate(page_lines):
|
||||
boundary_lines = (
|
||||
*((line, index < 2) for index, line in enumerate(lines[:5])),
|
||||
*((line, index < 2) for index, line in enumerate(reversed(lines[-3:]))),
|
||||
)
|
||||
seen_ranges: set[tuple[int, int]] = set()
|
||||
for line, is_outer_margin in boundary_lines:
|
||||
line_range = (line.start, line.end)
|
||||
if (
|
||||
line_range in seen_ranges
|
||||
or line_range in detected
|
||||
or len(line.text) > 160
|
||||
):
|
||||
continue
|
||||
seen_ranges.add(line_range)
|
||||
if not is_outer_margin and not _has_margin_metadata_keyword(line.text):
|
||||
continue
|
||||
signature = _margin_signature(line.text)
|
||||
if not _has_meaningful_margin_signature(signature):
|
||||
continue
|
||||
signature_pages.setdefault(signature, set()).add(page_index)
|
||||
candidate_lines.append((page_index, line, signature))
|
||||
minimum_pages = max(3, math.ceil(len(pages) * 0.3))
|
||||
repeated_signatures = {
|
||||
signature
|
||||
for signature, matching_pages in signature_pages.items()
|
||||
if len(matching_pages) >= minimum_pages
|
||||
}
|
||||
for _, line, signature in candidate_lines:
|
||||
if signature in repeated_signatures:
|
||||
mark(line, "repeated_margin")
|
||||
|
||||
# 先依据强证据判定目录页，再补充删除少量不带点引导线的编号目录项。
|
||||
toc_active = False
|
||||
for lines in page_lines:
|
||||
content_lines = [
|
||||
line for line in lines if (line.start, line.end) not in detected
|
||||
]
|
||||
leader_entries = [line for line in content_lines if _is_toc_leader_entry(line.text)]
|
||||
titles = [line for line in content_lines if _TOC_TITLE_PATTERN.fullmatch(line.text)]
|
||||
starts_toc = bool(titles and len(leader_entries) >= 2)
|
||||
is_toc_dense = bool(
|
||||
len(leader_entries) >= 3
|
||||
and len(leader_entries) / max(1, len(content_lines)) >= 0.5
|
||||
)
|
||||
if not (starts_toc or (toc_active and is_toc_dense)):
|
||||
toc_active = False
|
||||
continue
|
||||
toc_active = True
|
||||
for line in content_lines:
|
||||
if (
|
||||
line in titles
|
||||
or _is_toc_leader_entry(line.text)
|
||||
or _is_toc_numbered_entry(line.text)
|
||||
):
|
||||
mark(line, "table_of_contents")
|
||||
|
||||
return tuple(sorted(detected.values(), key=lambda span: (span.start, span.end)))
|
||||
|
||||
|
||||
def remove_document_noise(
|
||||
text: str,
|
||||
spans: Sequence[DocumentNoiseSpan],
|
||||
*,
|
||||
source_offset: int = 0,
|
||||
) -> str:
|
||||
"""按原文绝对偏移移除噪声，不改动调用方保留的原文及偏移。"""
|
||||
|
||||
text_end = source_offset + len(text)
|
||||
intersections = sorted(
|
||||
(
|
||||
max(0, span.start - source_offset),
|
||||
min(len(text), span.end - source_offset),
|
||||
)
|
||||
for span in spans
|
||||
if span.start < text_end and span.end > source_offset
|
||||
)
|
||||
if not intersections:
|
||||
return text
|
||||
parts: list[str] = []
|
||||
cursor = 0
|
||||
for start, end in intersections:
|
||||
if end <= cursor:
|
||||
continue
|
||||
if start > cursor:
|
||||
parts.append(text[cursor:start])
|
||||
cursor = end
|
||||
parts.append(text[cursor:])
|
||||
cleaned = normalize_text("".join(parts))
|
||||
return re.sub(r"\n{3,}", "\n\n", cleaned)
|
||||
|
||||
|
||||
def _extract_pdf_text(raw: bytes) -> str:
|
||||
return "\n\n".join(page.text for page in extract_pdf_page_texts(raw) if page.text)
|
||||
|
||||
@@ -2737,6 +2980,7 @@ __all__ = [
|
||||
"ChunkMethod",
|
||||
"DatasetSplit",
|
||||
"DocumentHeading",
|
||||
"DocumentNoiseSpan",
|
||||
"DocumentStructure",
|
||||
"ParsedText",
|
||||
"PdfPageText",
|
||||
@@ -2752,6 +2996,7 @@ __all__ = [
|
||||
"desensitize_pii",
|
||||
"desensitize_structured_record",
|
||||
"detect_document_structure",
|
||||
"detect_pdf_document_noise",
|
||||
"detect_text_format",
|
||||
"deduplicate_structured_records",
|
||||
"estimate_token_count",
|
||||
@@ -2773,6 +3018,7 @@ __all__ = [
|
||||
"preprocess_structured_records",
|
||||
"protected_context_ranges",
|
||||
"record_fingerprint",
|
||||
"remove_document_noise",
|
||||
"score_quality",
|
||||
"stable_split",
|
||||
]
|
||||
|
||||
Reference in New Issue
Block a user