From def58ac0828ce50181bf7d76e818271520bfc798 Mon Sep 17 00:00:00 2001 From: myhloli Date: Sat, 5 Sep 2026 14:56:35 +0800 Subject: [PATCH] refactor: make Flash PDF semantic and table assembly stages explicit --- mineru/model/flash/pdf/pipeline.py | 115 +- .../model/flash/pdf/table_recovery/vector.py | 208 +- .../model/flash/pdf/text_assembly/__init__.py | 4 + .../flash/pdf/text_assembly/annotations.py | 589 +++ .../model/flash/pdf/text_assembly/assembly.py | 297 ++ .../model/flash/pdf/text_assembly/common.py | 482 +++ .../flash/pdf/text_assembly/footnotes.py | 405 ++ .../model/flash/pdf/text_assembly/merging.py | 1280 ++++++ mineru/model/flash/pdf/text_assembly/rows.py | 697 ++++ mineru/model/flash/pdf/text_blocks.py | 3593 +---------------- .../flash/pdf/title_analysis/__init__.py | 4 + .../flash/pdf/title_analysis/body_profile.py | 227 ++ .../model/flash/pdf/title_analysis/common.py | 122 + .../pdf/title_analysis/document_profile.py | 174 + .../flash/pdf/title_analysis/lane_titles.py | 765 ++++ .../flash/pdf/title_analysis/page_titles.py | 1030 +++++ .../flash/pdf/title_analysis/prototype.py | 204 + .../flash/pdf/title_analysis/structural.py | 1088 +++++ mineru/model/flash/pdf/titles.py | 3505 +--------------- .../test_flash_pdf_document_profile.py | 30 +- tests/unittest/test_flash_pdf_titles.py | 12 +- 21 files changed, 7777 insertions(+), 7054 deletions(-) create mode 100644 mineru/model/flash/pdf/text_assembly/__init__.py create mode 100644 mineru/model/flash/pdf/text_assembly/annotations.py create mode 100644 mineru/model/flash/pdf/text_assembly/assembly.py create mode 100644 mineru/model/flash/pdf/text_assembly/common.py create mode 100644 mineru/model/flash/pdf/text_assembly/footnotes.py create mode 100644 mineru/model/flash/pdf/text_assembly/merging.py create mode 100644 mineru/model/flash/pdf/text_assembly/rows.py create mode 100644 mineru/model/flash/pdf/title_analysis/__init__.py create mode 100644 mineru/model/flash/pdf/title_analysis/body_profile.py create mode 100644 mineru/model/flash/pdf/title_analysis/common.py create mode 100644 mineru/model/flash/pdf/title_analysis/document_profile.py create mode 100644 mineru/model/flash/pdf/title_analysis/lane_titles.py create mode 100644 mineru/model/flash/pdf/title_analysis/page_titles.py create mode 100644 mineru/model/flash/pdf/title_analysis/prototype.py create mode 100644 mineru/model/flash/pdf/title_analysis/structural.py diff --git a/mineru/model/flash/pdf/pipeline.py b/mineru/model/flash/pdf/pipeline.py index ba78a6393..8f40e7afb 100644 --- a/mineru/model/flash/pdf/pipeline.py +++ b/mineru/model/flash/pdf/pipeline.py @@ -104,25 +104,25 @@ _marginal_geometry_matches, _marginal_text_matches, ) -from .titles import ( +from .title_analysis.body_profile import _infer_document_body_profile +from .title_analysis.document_profile import _infer_document_title_profile +from .title_analysis.page_titles import _classify_page_titles +from .title_analysis.structural import ( _classify_body_height_section_titles, _classify_explicit_section_titles, _classify_inline_typography_reset_titles, _classify_document_structural_titles, - _classify_page_titles, - _infer_document_body_profile, - _infer_document_title_profile, _promote_noninitial_document_title_band, ) -from .text_blocks import ( - _build_text_blocks, +from .text_assembly.annotations import ( _merge_fragmented_header_blocks, _merge_front_matter_column_blocks, _merge_image_caption_text_blocks, - _merge_internal_text_block_group, _merge_multiline_title_blocks, _merge_repeated_compact_title_continuations, ) +from .text_assembly.assembly import _build_text_blocks +from .text_assembly.common import _merge_internal_text_block_group from .visual_annotations import _classify_and_bind_visual_annotations @@ -408,34 +408,17 @@ def _prepare_document_sources( return prepared_pages -def _analyze_native_document( - pdf_doc: PDFDocument, - *, - script_diagnostics: list[dict[str, Any]] | None = None, - geometry_diagnostics: list[dict[str, Any]] | None = None, -) -> list[list[dict[str, Any]]]: - """逐页读取数字 PDF，并在轻量页面上完成跨页文本类型判定。""" +@dataclass(frozen=True, slots=True) +class _DocumentTextProfiles: + """分别保存原始正文尺度、规范正文尺度及全文标题原型。""" + + body: _DocumentBodyProfile | None + canonical_body: _DocumentBodyProfile | None + title: _DocumentTitleProfile | None - sources = _collect_document_sources(pdf_doc) - prepared_pages = _prepare_document_sources(sources, geometry_diagnostics=geometry_diagnostics) - if script_diagnostics is not None: - script_diagnostics.extend( - { - "page_index": page_index, - "page_size": prepared.page_size, - "script_lines": list(prepared.script_lines), - "lines": [ - { - "source_index": line.source_index, - "text": line.text, - "bbox": line.bbox, - "angle": line.angle, - } - for line in prepared.remaining_lines - ], - } - for page_index, prepared in enumerate(prepared_pages) - ) + +def _classify_document_text(prepared_pages: list[_PreparedPage]) -> _DocumentTextProfiles: + """按既有顺序分类跨页辅助文本，再统计正文并确认结构标题。""" _classify_repeated_visual_headers(prepared_pages) _classify_repeated_page_marginals(prepared_pages) @@ -466,16 +449,17 @@ def _analyze_native_document( legacy_body_profile=document_body_profile, document_title_profile=document_title_profile, ) - finalized_pages = [ - _finalize_prepared_page( - prepared, - page_index, - canonical_body_profile=canonical_body_profile, - document_body_profile=document_body_profile, - document_title_profile=document_title_profile, - ) - for page_index, prepared in enumerate(prepared_pages) - ] + return _DocumentTextProfiles(document_body_profile, canonical_body_profile, document_title_profile) + + +def _materialize_document_inline( + finalized_pages: list[list[dict[str, Any]]], + prepared_pages: list[_PreparedPage], + sources: _DocumentSources, + script_diagnostics: list[dict[str, Any]] | None, +) -> None: + """在页面归一化后按链接、样式、上下标顺序物化最终行内语义。""" + for page_index, (page_blocks, prepared, style_lines, link_lines, page_size) in enumerate( zip( finalized_pages, @@ -499,6 +483,49 @@ def _analyze_native_document( materialized_diagnostics=materialized_diagnostics, ) materialize_pdf_inline_spans(page_blocks) + + +def _analyze_native_document( + pdf_doc: PDFDocument, + *, + script_diagnostics: list[dict[str, Any]] | None = None, + geometry_diagnostics: list[dict[str, Any]] | None = None, +) -> list[list[dict[str, Any]]]: + """逐页读取数字 PDF，并在轻量页面上完成跨页文本类型判定。""" + + sources = _collect_document_sources(pdf_doc) + prepared_pages = _prepare_document_sources(sources, geometry_diagnostics=geometry_diagnostics) + if script_diagnostics is not None: + script_diagnostics.extend( + { + "page_index": page_index, + "page_size": prepared.page_size, + "script_lines": list(prepared.script_lines), + "lines": [ + { + "source_index": line.source_index, + "text": line.text, + "bbox": line.bbox, + "angle": line.angle, + } + for line in prepared.remaining_lines + ], + } + for page_index, prepared in enumerate(prepared_pages) + ) + + profiles = _classify_document_text(prepared_pages) + finalized_pages = [ + _finalize_prepared_page( + prepared, + page_index, + canonical_body_profile=profiles.canonical_body, + document_body_profile=profiles.body, + document_title_profile=profiles.title, + ) + for page_index, prepared in enumerate(prepared_pages) + ] + _materialize_document_inline(finalized_pages, prepared_pages, sources, script_diagnostics) return finalized_pages diff --git a/mineru/model/flash/pdf/table_recovery/vector.py b/mineru/model/flash/pdf/table_recovery/vector.py index ffe10a8be..1b90913c1 100644 --- a/mineru/model/flash/pdf/table_recovery/vector.py +++ b/mineru/model/flash/pdf/table_recovery/vector.py @@ -1152,27 +1152,57 @@ def _looks_like_single_column_tracks( ) -def _build_vector_candidate( +@dataclass(frozen=True, slots=True) +class _VectorTracks: + """保存已通过别名、尺寸和物理行数校验的规范轨道。""" + + snap_tolerance: float + local_width: float + rules: list[_MergedRule] + canonical_x_tracks: list[_CanonicalTrack] + canonical_y_tracks: list[_CanonicalTrack] + x_tracks: list[float] + y_tracks: list[float] + narrow_empty_threshold: float + is_line_grid: bool + is_single_row_shape: bool + is_single_column_shape: bool + rows: int + cols: int + + +@dataclass(frozen=True, slots=True) +class _VectorTopology: + """保存隔断连接后的逻辑单元格及独立物理证据。""" + + specs: tuple[GridCellSpec, ...] + separator_decisions: list[float] + ambiguous_ratio: float + alias_separator_recoveries: int + y_alias_separator_recoveries: int + alias_affected_rows: set[int] + single_row_evidence: _SingleRowEvidence | None + single_column_evidence: _SingleColumnEvidence | None + + +def _reject_vector_candidate(diagnostics: dict[str, Any] | None, gate: str) -> None: + """记录当前假设的首个拒绝门，供各显式阶段保留统一诊断行为。""" + + if diagnostics is not None: + diagnostics["first_rejection_gate"] = gate + return None + + +def _build_vector_tracks( table_input: NativeTableInput, text: NativeTableText, *, include_drawing: bool, include_rectangles: bool, - evidence_label: str, - prune_unsupported_horizontal: bool = False, - diagnostics: dict[str, Any] | None = None, -) -> NativeTableCandidate | None: - """按单一证据来源构造一个显式 rowspan/colspan 候选。""" - - if diagnostics is not None: - diagnostics["evidence"] = evidence_label - - def reject(gate: str) -> None: - """记录当前假设的首个拒绝门并返回空候选。""" - - if diagnostics is not None: - diagnostics["first_rejection_gate"] = gate - return None + prune_unsupported_horizontal: bool, + diagnostics: dict[str, Any] | None, +) -> _VectorTracks | None: + """构造并规范化轨道，保持 halo、别名及物理行数的原有拒绝顺序。""" snap_tolerance = clamp( 0.08 * text.median_glyph_height, @@ -1195,7 +1225,7 @@ def reject(gate: str) -> None: ) table_bbox = normalize_bbox(table_input.table_bbox) if table_bbox is None: - return reject("table_geometry") + return _reject_vector_candidate(diagnostics, "table_geometry") local_width, local_height = table_local_size( table_bbox, normalize_angle(table_input.angle), @@ -1208,7 +1238,7 @@ def reject(gate: str) -> None: evidence_halo=0.0, ) if not exact_fragments or len(exact_fragments) > MAX_PRIMITIVES_PER_TABLE: - return reject("raw_fragments") + return _reject_vector_candidate(diagnostics, "raw_fragments") exact_rules = _merge_rule_fragments( exact_fragments, snap_tolerance, @@ -1295,7 +1325,7 @@ def reject(gate: str) -> None: snap_tolerance, ) ): - return reject("rect_lattice") + return _reject_vector_candidate(diagnostics, "rect_lattice") line_widths = [rule.width for rule in table_input.drawing_lines if rule.width > 0] median_line_width = float(statistics.median(line_widths)) if line_widths else 0.0 narrow_empty_threshold = max( @@ -1347,7 +1377,7 @@ def reject(gate: str) -> None: snap_tolerance, ) ): - return reject("canonical_alias") + return _reject_vector_candidate(diagnostics, "canonical_alias") x_tracks = _canonical_track_coordinates(canonical_x_tracks) y_tracks = _canonical_track_coordinates(canonical_y_tracks) if diagnostics is not None: @@ -1373,7 +1403,7 @@ def reject(gate: str) -> None: right - left <= narrow_empty_threshold and not any(left < center < right for center in glyph_centers_x) for left, right in zip(x_tracks, x_tracks[1:]) ): - return reject("remaining_narrow_track") + return _reject_vector_candidate(diagnostics, "remaining_narrow_track") if any( bottom_track.coordinate - top_track.coordinate <= narrow_empty_threshold and not any(top_track.coordinate < center < bottom_track.coordinate for center in glyph_centers_y) @@ -1400,7 +1430,7 @@ def reject(gate: str) -> None: canonical_y_tracks[1:], ) ): - return reject("remaining_narrow_track") + return _reject_vector_candidate(diagnostics, "remaining_narrow_track") is_line_grid = include_drawing and not include_rectangles is_single_row_shape = is_line_grid and len(y_tracks) == 2 and len(x_tracks) >= 3 is_single_column_shape = is_line_grid and len(x_tracks) == 2 and len(y_tracks) >= 3 @@ -1410,13 +1440,13 @@ def reject(gate: str) -> None: or len(x_tracks) > MAX_TRACKS_PER_AXIS or len(y_tracks) > MAX_TRACKS_PER_AXIS ): - return reject("track_count") + return _reject_vector_candidate(diagnostics, "track_count") rows = len(y_tracks) - 1 cols = len(x_tracks) - 1 if diagnostics is not None: diagnostics["grid"] = {"rows": rows, "cols": cols} if rows * cols > MAX_ATOMIC_CELLS: - return reject("atomic_cell_limit") + return _reject_vector_candidate(diagnostics, "atomic_cell_limit") dense_baseline_pairs = ( _physical_row_dense_baseline_pairs( text, @@ -1429,7 +1459,43 @@ def reject(gate: str) -> None: if diagnostics is not None: diagnostics["physical_row_dense_baseline_pairs"] = list(dense_baseline_pairs) if dense_baseline_pairs: - return reject("physical_row_undercount") + return _reject_vector_candidate(diagnostics, "physical_row_undercount") + + return _VectorTracks( + snap_tolerance=snap_tolerance, + local_width=local_width, + rules=rules, + canonical_x_tracks=canonical_x_tracks, + canonical_y_tracks=canonical_y_tracks, + x_tracks=x_tracks, + y_tracks=y_tracks, + narrow_empty_threshold=narrow_empty_threshold, + is_line_grid=is_line_grid, + is_single_row_shape=is_single_row_shape, + is_single_column_shape=is_single_column_shape, + rows=rows, + cols=cols, + ) + + +def _build_vector_topology( + tracks: _VectorTracks, + text: NativeTableText, + diagnostics: dict[str, Any] | None, +) -> _VectorTopology | None: + """连接原子格并验证矩形拓扑，保留单行和单列的独立物理证据。""" + + snap_tolerance = tracks.snap_tolerance + rules = tracks.rules + canonical_x_tracks = tracks.canonical_x_tracks + canonical_y_tracks = tracks.canonical_y_tracks + x_tracks = tracks.x_tracks + y_tracks = tracks.y_tracks + narrow_empty_threshold = tracks.narrow_empty_threshold + is_single_row_shape = tracks.is_single_row_shape + is_single_column_shape = tracks.is_single_column_shape + rows = tracks.rows + cols = tracks.cols union_find = _UnionFind(rows * cols) separator_decisions: list[float] = [] @@ -1517,7 +1583,7 @@ def reject(gate: str) -> None: diagnostics["y_alias_separator_recoveries"] = y_alias_separator_recoveries diagnostics["alias_affected_rows"] = sorted(alias_affected_rows) if ambiguous_ratio > 0.05: - return reject("ambiguous_separator") + return _reject_vector_candidate(diagnostics, "ambiguous_separator") single_row_evidence: _SingleRowEvidence | None = None if is_single_row_shape: @@ -1541,7 +1607,7 @@ def reject(gate: str) -> None: "glyph_crossing": single_row_evidence.glyph_crossing, } if not single_row_evidence.verified: - return reject("single_row_physical_evidence") + return _reject_vector_candidate(diagnostics, "single_row_physical_evidence") single_column_evidence: _SingleColumnEvidence | None = None if is_single_column_shape: @@ -1565,7 +1631,7 @@ def reject(gate: str) -> None: "glyph_crossing": single_column_evidence.glyph_crossing, } if not single_column_evidence.verified: - return reject("single_column_physical_evidence") + return _reject_vector_candidate(diagnostics, "single_column_physical_evidence") specs = _build_component_specs( union_find, @@ -1575,11 +1641,55 @@ def reject(gate: str) -> None: y_tracks, ) if specs is None: - return reject("nonrectangular_topology") + return _reject_vector_candidate(diagnostics, "nonrectangular_topology") maximum_row_cells = max(sum(spec.row <= row_index < spec.row + spec.rowspan for spec in specs) for row_index in range(rows)) maximum_col_cells = max(sum(spec.col <= col_index < spec.col + spec.colspan for spec in specs) for col_index in range(cols)) if (maximum_row_cells < 2 and not is_single_column_shape) or (maximum_col_cells < 2 and not is_single_row_shape): - return reject("degenerate_grid") + return _reject_vector_candidate(diagnostics, "degenerate_grid") + return _VectorTopology( + specs=specs, + separator_decisions=separator_decisions, + ambiguous_ratio=ambiguous_ratio, + alias_separator_recoveries=alias_separator_recoveries, + y_alias_separator_recoveries=y_alias_separator_recoveries, + alias_affected_rows=alias_affected_rows, + single_row_evidence=single_row_evidence, + single_column_evidence=single_column_evidence, + ) + + +def _materialize_vector_candidate( + tracks: _VectorTracks, + topology: _VectorTopology, + text: NativeTableText, + evidence_label: str, + diagnostics: dict[str, Any] | None, +) -> NativeTableCandidate | None: + """将文本落格并评分，按原顺序执行完整性与空行发布门。""" + + snap_tolerance = tracks.snap_tolerance + local_width = tracks.local_width + rules = tracks.rules + canonical_x_tracks = tracks.canonical_x_tracks + canonical_y_tracks = tracks.canonical_y_tracks + x_tracks = tracks.x_tracks + y_tracks = tracks.y_tracks + narrow_empty_threshold = tracks.narrow_empty_threshold + is_line_grid = tracks.is_line_grid + is_single_row_shape = tracks.is_single_row_shape + is_single_column_shape = tracks.is_single_column_shape + rows = tracks.rows + cols = tracks.cols + + specs = topology.specs + separator_decisions = topology.separator_decisions + ambiguous_ratio = topology.ambiguous_ratio + alias_separator_recoveries = topology.alias_separator_recoveries + y_alias_separator_recoveries = topology.y_alias_separator_recoveries + alias_affected_rows = topology.alias_affected_rows + single_row_evidence = topology.single_row_evidence + single_column_evidence = topology.single_column_evidence + decisiveness = float(statistics.mean(separator_decisions)) if separator_decisions else 1.0 if single_row_evidence is not None: decisiveness = max( @@ -1684,11 +1794,11 @@ def reject(gate: str) -> None: if diagnostics is not None and diagnostics.get("candidate_rejection_gate") else "candidate_hard_gate" ) - return reject(candidate_gate) + return _reject_vector_candidate(diagnostics, candidate_gate) if is_single_row_shape and (candidate.text_capture < 1.0 or candidate.order_consistency < 1.0): - return reject("single_row_text_integrity") + return _reject_vector_candidate(diagnostics, "single_row_text_integrity") if is_single_column_shape and (candidate.text_capture < 1.0 or candidate.order_consistency < 1.0): - return reject("single_column_text_integrity") + return _reject_vector_candidate(diagnostics, "single_column_text_integrity") row_content_support = [ sum(bool(cell.content.strip()) for cell in candidate.cells if cell.row <= row_index < cell.row + cell.rowspan) for row_index in range(candidate.rows) @@ -1702,7 +1812,7 @@ def reject(gate: str) -> None: ): if diagnostics is not None: diagnostics["empty_rows"] = sorted(empty_rows) - return reject("empty_row") + return _reject_vector_candidate(diagnostics, "empty_row") if diagnostics is not None: diagnostics["first_rejection_gate"] = None diagnostics["score"] = candidate.score @@ -1710,6 +1820,36 @@ def reject(gate: str) -> None: return candidate +def _build_vector_candidate( + table_input: NativeTableInput, + text: NativeTableText, + *, + include_drawing: bool, + include_rectangles: bool, + evidence_label: str, + prune_unsupported_horizontal: bool = False, + diagnostics: dict[str, Any] | None = None, +) -> NativeTableCandidate | None: + """按轨道、拓扑、文本落格及评分的固定顺序构造矢量候选。""" + + if diagnostics is not None: + diagnostics["evidence"] = evidence_label + tracks = _build_vector_tracks( + table_input, + text, + include_drawing=include_drawing, + include_rectangles=include_rectangles, + prune_unsupported_horizontal=prune_unsupported_horizontal, + diagnostics=diagnostics, + ) + if tracks is None: + return None + topology = _build_vector_topology(tracks, text, diagnostics) + if topology is None: + return None + return _materialize_vector_candidate(tracks, topology, text, evidence_label, diagnostics) + + def build_vector_candidates( table_input: NativeTableInput, text: NativeTableText, diff --git a/mineru/model/flash/pdf/text_assembly/__init__.py b/mineru/model/flash/pdf/text_assembly/__init__.py new file mode 100644 index 000000000..b2b2d2a58 --- /dev/null +++ b/mineru/model/flash/pdf/text_assembly/__init__.py @@ -0,0 +1,4 @@ +# Copyright (c) Opendatalab. All rights reserved. +"""按职责拆分的 Flash PDF 内部实现。""" + +__all__: list[str] = [] diff --git a/mineru/model/flash/pdf/text_assembly/annotations.py b/mineru/model/flash/pdf/text_assembly/annotations.py new file mode 100644 index 000000000..4547824a7 --- /dev/null +++ b/mineru/model/flash/pdf/text_assembly/annotations.py @@ -0,0 +1,589 @@ +# Copyright (c) Opendatalab. All rights reserved. +"""组装跨行标题、图片注释、页眉及首页信息块。""" + +from __future__ import annotations + +import statistics +from typing import Any + +from .....types import BBox +from ..geometry import ( + _bbox_axis_overlap_ratio, + _bbox_center_x, + _bbox_center_y, + _bbox_union_many, +) +from ..line_layout import ( + _is_structural_typography_gap, +) +from .common import _FIGURE_CAPTION_MARKER_RE, _components_share_lane_role, _merge_internal_text_block_group + + +def _merge_image_caption_text_blocks( + blocks: list[dict[str, Any]], + image_bboxes: list[BBox], +) -> list[dict[str, Any]]: + """在图像邻接已成立后，用通用图注标记确认锚点并吸收同字体续行。""" + + if not image_bboxes: + return blocks + text_indices = [ + index + for index, block in enumerate(blocks) + if block.get("type") == "text" + and isinstance(block.get("content"), str) + and isinstance(block.get("bbox"), (list, tuple)) + ] + all_heights = [ + float(height) + for index in text_indices + for height in blocks[index].get("_line_heights", []) + if isinstance(height, (int, float)) and height > 0 + ] + median_height = statistics.median(all_heights) if all_heights else 1.0 + caption_image_bboxes = _caption_image_group_bboxes( + image_bboxes, + median_height, + ) + seed_indices = { + index + for index in text_indices + if _FIGURE_CAPTION_MARKER_RE.match(str(blocks[index]["content"]).strip()) + and any( + _caption_seed_matches_image( + blocks[index], + image_bbox, + median_height, + ) + for image_bbox in caption_image_bboxes + ) + } + if not seed_indices: + return blocks + + assignments: dict[int, list[int]] = {index: [] for index in seed_indices} + for candidate_index in text_indices: + if candidate_index in seed_indices: + continue + candidate = blocks[candidate_index] + matches: list[tuple[float, float, int]] = [] + for seed_index in seed_indices: + seed = blocks[seed_index] + if not _caption_tail_matches_seed( + seed, + candidate, + median_height, + ): + continue + seed_bbox = seed["bbox"] + candidate_bbox = candidate["bbox"] + matches.append( + ( + _bbox_center_y(candidate_bbox) - _bbox_center_y(seed_bbox), + abs(_bbox_center_x(candidate_bbox) - _bbox_center_x(seed_bbox)), + seed_index, + ) + ) + if matches: + assignments[min(matches)[2]].append(candidate_index) + + merged_indices: set[int] = set() + replacements: dict[int, dict[str, Any]] = {} + for seed_index, tail_indices in assignments.items(): + if not tail_indices: + continue + group_indices = [seed_index, *tail_indices] + replacements[seed_index] = _merge_internal_text_block_group( + blocks, + group_indices, + ) + merged_indices.update(tail_indices) + return [replacements.get(index, block) for index, block in enumerate(blocks) if index not in merged_indices] + + +def _caption_image_group_bboxes( + image_bboxes: list[BBox], + median_height: float, +) -> list[BBox]: + """合并同一视觉行的并排图片 bbox，使跨多图的统一图注也能建立邻接。""" + + remaining = list(image_bboxes) + grouped_bboxes = list(image_bboxes) + while remaining: + group = [remaining.pop(0)] + changed = True + while changed: + changed = False + for candidate in list(remaining): + aligned = False + for member in group: + overlap = max( + 0.0, + min(candidate[3], member[3]) - max(candidate[1], member[1]), + ) + minimum_height = max( + 0.1, + min( + candidate[3] - candidate[1], + member[3] - member[1], + ), + ) + horizontal_gap = max( + 0.0, + max(candidate[0], member[0]) - min(candidate[2], member[2]), + ) + if overlap / minimum_height >= 0.7 and horizontal_gap <= 2.0 * median_height: + aligned = True + break + if aligned: + group.append(candidate) + remaining.remove(candidate) + changed = True + if len(group) >= 2: + grouped_bboxes.append(_bbox_union_many(group)) + return grouped_bboxes + + +def _caption_seed_matches_image( + block: dict[str, Any], + image_bbox: BBox, + median_height: float, +) -> bool: + """用上下位置、水平投影和居中关系确认图像下方的图注空间候选。""" + + bbox = block["bbox"] + image_width = max(0.1, image_bbox[2] - image_bbox[0]) + block_width = max(0.1, bbox[2] - bbox[0]) + vertical_gap = max(0.0, bbox[1] - image_bbox[3]) + return ( + _bbox_center_y(bbox) >= image_bbox[3] - 0.25 * median_height + and vertical_gap <= 2.5 * median_height + and _bbox_axis_overlap_ratio(bbox, image_bbox, axis="x") >= 0.35 + and abs(_bbox_center_x(bbox) - _bbox_center_x(image_bbox)) <= 0.35 * max(image_width, block_width) + and block_width <= 1.75 * image_width + ) + + +def _caption_body_has_structural_gap( + seed: dict[str, Any], + candidate: dict[str, Any], +) -> bool: + """用图注末行、候选首行和图注内部行距阻止跨排版层级回并。""" + + seed_bboxes = seed.get("_local_line_bboxes") + seed_heights = seed.get("_line_heights") + candidate_bboxes = candidate.get("_local_line_bboxes") + candidate_heights = candidate.get("_line_heights") + if not ( + isinstance(seed_bboxes, list) + and isinstance(seed_heights, list) + and len(seed_bboxes) == len(seed_heights) + and seed_bboxes + and isinstance(candidate_bboxes, list) + and isinstance(candidate_heights, list) + and len(candidate_bboxes) == len(candidate_heights) + and candidate_bboxes + ): + return False + + seed_rows = sorted( + zip(seed_bboxes, seed_heights, strict=True), + key=lambda item: (item[0][1], item[0][0]), + ) + candidate_rows = sorted( + zip(candidate_bboxes, candidate_heights, strict=True), + key=lambda item: (item[0][1], item[0][0]), + ) + previous_bbox, previous_height = seed_rows[-1] + current_bbox, current_height = candidate_rows[0] + internal_gaps = [ + max(0.0, current[0][1] - (previous[0][1] + float(previous[1]))) for previous, current in zip(seed_rows, seed_rows[1:]) + ] + regular_gap = statistics.median(internal_gaps) if internal_gaps else 0.0 + gap_mad = statistics.median(abs(gap - regular_gap) for gap in internal_gaps) if internal_gaps else 0.0 + seed_fonts = seed.get("_font_signatures") + candidate_fonts = candidate.get("_font_signatures") + reliable_style_change = ( + isinstance(seed_fonts, set) + and bool(seed_fonts) + and isinstance(candidate_fonts, set) + and bool(candidate_fonts) + and seed_fonts.isdisjoint(candidate_fonts) + ) + return _is_structural_typography_gap( + float(previous_height), + float(current_height), + current_bbox[1] - (previous_bbox[1] + float(previous_height)), + regular_gap, + gap_mad, + reliable_style_change=reliable_style_change, + ) + + +def _caption_tail_matches_seed( + seed: dict[str, Any], + candidate: dict[str, Any], + median_height: float, +) -> bool: + """只用同栏角色、字体、邻接和投影把无标记的图注续行接回锚点。""" + + seed_bbox = seed["bbox"] + candidate_bbox = candidate["bbox"] + if not _components_share_lane_role(seed, candidate, median_height) and ( + _bbox_axis_overlap_ratio(seed_bbox, candidate_bbox, axis="x") < 0.75 + or abs(seed_bbox[0] - candidate_bbox[0]) > median_height + ): + return False + if _bbox_center_y(candidate_bbox) <= _bbox_center_y(seed_bbox): + return False + if _caption_body_has_structural_gap(seed, candidate): + return False + vertical_gap = max(0.0, candidate_bbox[1] - seed_bbox[3]) + if vertical_gap > 0.5 * median_height or _bbox_axis_overlap_ratio(seed_bbox, candidate_bbox, axis="x") < 0.35: + return False + seed_fonts = seed.get("_font_signatures") + candidate_fonts = candidate.get("_font_signatures") + return not ( + isinstance(seed_fonts, set) + and seed_fonts + and isinstance(candidate_fonts, set) + and candidate_fonts + and seed_fonts.isdisjoint(candidate_fonts) + ) + + +def _merge_multiline_title_blocks( + blocks: list[dict[str, Any]], +) -> list[dict[str, Any]]: + """跨错误栏带合并紧贴且字体兼容的多行文档标题和段落标题。""" + + replacements: dict[int, dict[str, Any]] = {} + consumed: set[int] = set() + for block_type in ("doc_title", "paragraph_title"): + indices = [ + index + for index, block in enumerate(blocks) + if block.get("type") == block_type and isinstance(block.get("bbox"), (list, tuple)) + ] + indices.sort( + key=lambda index: ( + blocks[index]["bbox"][1], + blocks[index]["bbox"][0], + ) + ) + groups: list[list[int]] = [] + for index in indices: + if not groups: + groups.append([index]) + continue + previous_index = groups[-1][-1] + previous = blocks[previous_index] + current = blocks[index] + previous_bbox = previous["bbox"] + current_bbox = current["bbox"] + previous_heights = previous.get("_line_heights", []) + current_heights = current.get("_line_heights", []) + previous_height = ( + statistics.median(previous_heights) + if isinstance(previous_heights, list) and previous_heights + else previous_bbox[3] - previous_bbox[1] + ) + current_height = ( + statistics.median(current_heights) + if isinstance(current_heights, list) and current_heights + else current_bbox[3] - current_bbox[1] + ) + vertical_gap = current_bbox[1] - previous_bbox[3] + previous_fonts = previous.get("_font_signatures") + current_fonts = current.get("_font_signatures") + fonts_conflict = ( + isinstance(previous_fonts, set) + and previous_fonts + and isinstance(current_fonts, set) + and current_fonts + and previous_fonts.isdisjoint(current_fonts) + ) + if ( + -0.2 * max(previous_height, current_height) <= vertical_gap <= 0.4 * max(previous_height, current_height) + and _bbox_axis_overlap_ratio( + previous_bbox, + current_bbox, + axis="x", + ) + >= 0.2 + and not fonts_conflict + ): + groups[-1].append(index) + else: + groups.append([index]) + for group in groups: + if len(group) < 2: + continue + replacements[group[0]] = _merge_internal_text_block_group( + blocks, + group, + ) + consumed.update(group[1:]) + return [replacements.get(index, block) for index, block in enumerate(blocks) if index not in consumed] + + +def _merge_fragmented_header_blocks( + blocks: list[dict[str, Any]], +) -> list[dict[str, Any]]: + """聚合同一视觉行中等距分散的页眉页脚片段。""" + + grouped: dict[tuple[int, int], list[int]] = {} + for index, block in enumerate(blocks): + row_id = block.get("_single_run_row_id") + angle = int(block.get("angle", 0) or 0) % 360 + if block.get("type") in {"header", "footer"} and isinstance(row_id, int): + grouped.setdefault((angle, row_id), []).append(index) + + replacements: dict[int, dict[str, Any]] = {} + consumed: set[int] = set() + for indices in grouped.values(): + ordered = sorted(indices, key=lambda index: blocks[index]["bbox"][0]) + components: list[list[int]] = [] + for index in ordered: + bbox = blocks[index]["bbox"] + heights = blocks[index].get("_line_heights", []) + effective_height = ( + statistics.median(heights) if isinstance(heights, list) and heights else max(0.1, bbox[3] - bbox[1]) + ) + if blocks[index].get("type") == "header" and bbox[2] - bbox[0] > 1.25 * effective_height: + continue + if not components: + components.append([index]) + continue + previous_index = components[-1][-1] + previous_bbox = blocks[previous_index]["bbox"] + previous_heights = blocks[previous_index].get("_line_heights", []) + previous_height = ( + statistics.median(previous_heights) + if isinstance(previous_heights, list) and previous_heights + else max(0.1, previous_bbox[3] - previous_bbox[1]) + ) + if ( + bbox[0] - previous_bbox[2] <= 4.0 * max(effective_height, previous_height) + and _bbox_axis_overlap_ratio(previous_bbox, bbox, axis="y") >= 0.5 + ): + components[-1].append(index) + else: + components.append([index]) + for component in components: + if len(component) < 2: + continue + replacement = _merge_internal_text_block_group( + blocks, + component, + preserve_visual_spaces=True, + ) + replacement["_single_run_row_id"] = None + replacements[component[0]] = replacement + consumed.update(component[1:]) + return [replacements.get(index, block) for index, block in enumerate(blocks) if index not in consumed] + + +def _merge_front_matter_column_blocks( + blocks: list[dict[str, Any]], + page_size: tuple[float, float], + *, + page_index: int, +) -> list[dict[str, Any]]: + """把首页标题下方规则排列的多列作者信息按列聚合。""" + + if page_index != 0: + return blocks + page_width, page_height = page_size + if page_width <= 0 or page_height <= 0: + return blocks + title_blocks = [ + block for block in blocks if block.get("type") == "doc_title" and isinstance(block.get("bbox"), (list, tuple)) + ] + if not title_blocks: + return blocks + title_bottom = max(block["bbox"][3] for block in title_blocks) + candidates = [ + index + for index, block in enumerate(blocks) + if block.get("type") == "text" + and isinstance(block.get("bbox"), (list, tuple)) + and title_bottom < block["bbox"][1] + and block["bbox"][3] + <= min( + 0.4 * page_height, + title_bottom + 0.22 * page_height, + ) + and block["bbox"][2] - block["bbox"][0] <= 0.32 * page_width + and block["bbox"][3] - block["bbox"][1] <= 0.035 * page_height + ] + if len(candidates) < 9: + return blocks + median_height = statistics.median(blocks[index]["bbox"][3] - blocks[index]["bbox"][1] for index in candidates) + row_groups: list[list[int]] = [] + for index in sorted( + candidates, + key=lambda item: ( + _bbox_center_y(blocks[item]["bbox"]), + blocks[item]["bbox"][0], + ), + ): + center_y = _bbox_center_y(blocks[index]["bbox"]) + target = next( + ( + row + for row in row_groups + if abs(center_y - statistics.median(_bbox_center_y(blocks[member]["bbox"]) for member in row)) + <= 0.6 * median_height + ), + None, + ) + if target is None: + row_groups.append([index]) + else: + target.append(index) + dense_rows = [ + row + for row in row_groups + if 3 <= len(row) <= 6 + and ( + max(blocks[index]["bbox"][2] for index in row) - min(blocks[index]["bbox"][0] for index in row) >= 0.55 * page_width + ) + ] + if len(dense_rows) < 2: + return blocks + anchor_row = min( + dense_rows, + key=lambda row: ( + -len(row), + statistics.median(_bbox_center_y(blocks[index]["bbox"]) for index in row), + ), + ) + anchor_centers = sorted(_bbox_center_x(blocks[index]["bbox"]) for index in anchor_row) + if len(anchor_centers) != 4: + return blocks + boundaries = [0.5 * (left + right) for left, right in zip(anchor_centers, anchor_centers[1:])] + band_top = min(min(blocks[index]["bbox"][1] for index in row) for row in dense_rows) - median_height + band_bottom = max(max(blocks[index]["bbox"][3] for index in row) for row in dense_rows) + median_height + column_groups: list[list[int]] = [[] for _center in anchor_centers] + for index in candidates: + bbox = blocks[index]["bbox"] + if not band_top <= _bbox_center_y(bbox) <= band_bottom: + continue + center_x = _bbox_center_x(bbox) + column_index = sum(center_x > boundary for boundary in boundaries) + if column_index >= len(column_groups): + continue + column_groups[column_index].append(index) + if any(len(group) < 3 for group in column_groups): + return blocks + + replacements: dict[int, dict[str, Any]] = {} + consumed: set[int] = set() + for group in column_groups: + ordered = sorted( + group, + key=lambda index: ( + blocks[index]["bbox"][1], + blocks[index]["bbox"][0], + ), + ) + replacements[ordered[0]] = _merge_internal_text_block_group( + blocks, + ordered, + ) + consumed.update(ordered[1:]) + return [replacements.get(index, block) for index, block in enumerate(blocks) if index not in consumed] + + +def _merge_repeated_compact_title_continuations( + blocks: list[dict[str, Any]], + page_size: tuple[float, float], +) -> list[dict[str, Any]]: + """把重复出现的两行弱标题与紧邻异字体续行恢复为普通文本块。""" + + candidate_pairs: list[tuple[int, int, float]] = [] + for title_index, title in enumerate(blocks): + title_lines = title.get("_local_line_bboxes") + title_fonts = title.get("_font_signatures") + title_bbox = title.get("bbox") + if ( + title.get("type") != "paragraph_title" + or not isinstance(title_bbox, (list, tuple)) + or not isinstance(title_lines, list) + or len(title_lines) < 2 + or not isinstance(title_fonts, set) + or not title_fonts + ): + continue + angle = int(title.get("angle", 0) or 0) % 360 + local_page_width = page_size[1] if angle in {90, 270} else page_size[0] + line_heights = [ + float(height) for height in title.get("_line_heights", []) if isinstance(height, (int, float)) and height > 0 + ] + title_height = statistics.median(line_heights) if line_heights else 0.0 + if title_height <= 0 or title_bbox[2] - title_bbox[0] > 0.55 * local_page_width: + continue + + continuations: list[tuple[float, int]] = [] + for text_index, text_block in enumerate(blocks): + text_bbox = text_block.get("bbox") + text_fonts = text_block.get("_font_signatures") + if ( + text_block.get("type") != "text" + or int(text_block.get("angle", 0) or 0) % 360 != angle + or not isinstance(text_bbox, (list, tuple)) + or not isinstance(text_fonts, set) + or not text_fonts + or not title_fonts.isdisjoint(text_fonts) + or text_bbox[2] - text_bbox[0] > 0.6 * local_page_width + ): + continue + gap = text_bbox[1] - title_bbox[3] + if -0.25 * title_height <= gap <= 0.6 * title_height and abs(text_bbox[0] - title_bbox[0]) <= 0.75 * title_height: + continuations.append((max(0.0, gap), text_index)) + if continuations: + _gap, text_index = min(continuations) + candidate_pairs.append((title_index, text_index, title_height)) + + supported_pairs: list[tuple[int, int]] = [] + for title_index, text_index, title_height in candidate_pairs: + title_bbox = blocks[title_index]["bbox"] + support_count = sum( + abs(blocks[other_title]["bbox"][0] - title_bbox[0]) <= max(title_height, other_height) + and 0.75 <= other_height / title_height <= 1.25 + for other_title, _other_text, other_height in candidate_pairs + ) + if support_count >= 2: + supported_pairs.append((title_index, text_index)) + if not supported_pairs: + return blocks + + replacements: dict[int, dict[str, Any]] = {} + consumed: set[int] = set() + for title_index, text_index in supported_pairs: + if title_index in consumed or text_index in consumed: + continue + merged = _merge_internal_text_block_group( + blocks, + [title_index, text_index], + ) + merged["type"] = "text" + replacements[min(title_index, text_index)] = merged + consumed.update({title_index, text_index}) + return [ + replacements.get(index, block) for index, block in enumerate(blocks) if index not in consumed or index in replacements + ] + + +__all__ = [ + "_merge_image_caption_text_blocks", + "_caption_image_group_bboxes", + "_caption_seed_matches_image", + "_caption_body_has_structural_gap", + "_caption_tail_matches_seed", + "_merge_multiline_title_blocks", + "_merge_fragmented_header_blocks", + "_merge_front_matter_column_blocks", + "_merge_repeated_compact_title_continuations", +] diff --git a/mineru/model/flash/pdf/text_assembly/assembly.py b/mineru/model/flash/pdf/text_assembly/assembly.py new file mode 100644 index 000000000..ba3f5f1f3 --- /dev/null +++ b/mineru/model/flash/pdf/text_assembly/assembly.py @@ -0,0 +1,297 @@ +# Copyright (c) Opendatalab. All rights reserved. +"""保持既有顺序编排正文行分组与块级组装。""" + +from __future__ import annotations + +import statistics +from typing import Any, Sequence + +from .....types import BBox +from .....utils.text import is_hyphen_at_line_end +from ..geometry import ( + _bbox_union_many, + _rotate_bbox_to_upright, + _transform_axis_lines, +) +from ..line_layout import ( + _estimate_lane_gap, + _infer_text_lanes, + _line_effective_height, + _should_connect_semantic_rows, + _should_connect_text_rows, +) +from ..models import _AxisLine, _LineItem +from ..native_text import _normalize_native_run_text +from .common import _PARAGRAPH_FORMULA_CONTEXT_MARKER, _merge_text_line_content +from .footnotes import _build_grouped_page_footnote_blocks +from .merging import ( + _merge_inline_math_fragment_text_blocks, + _merge_list_intro_text_components, + _merge_overlapping_same_line_text_blocks, + _merge_paragraph_formula_context_blocks, + _merge_short_same_baseline_prefix_blocks, + _merge_spatial_text_components, + _merge_unterminated_text_components, +) +from .rows import ( + _build_hanging_indent_group_map, + _centered_visual_reset_break_sources, + _component_starts_with_emphasized_row, + _explicit_text_break_sources, + _formula_style_text_row_break_sources, + _front_matter_keyword_break_sources, + _infer_local_text_lane_map, + _isolated_indented_paragraph_break_sources, + _leading_typography_reset_break_sources, + _local_tight_output_line_bboxes, + _starts_structural_reference_entry, + _structured_text_break_sources, +) + + +def _build_text_blocks( + lines: list[_LineItem], + table_bboxes: list[BBox], + page_size: tuple[float, float], + drawing_lines: list[_AxisLine] | None = None, + *, + page_footnote_groups: Sequence[set[int]] | None = None, + page_index: int | None = None, + visual_bboxes: Sequence[BBox] | None = None, +) -> list[dict[str, Any]]: + """先构建分组脚注，再按类型屏障、栏带和自然段边界聚合其余文本。""" + + blocks, grouped_footnote_indices = _build_grouped_page_footnote_blocks( + lines, + page_footnote_groups or [], + page_size, + ) + lines = [line for line in lines if line.source_index not in grouped_footnote_indices] + for angle in sorted({line.angle for line in lines}): + line_geometry = [(line, _rotate_bbox_to_upright(line.bbox, page_size, angle)) for line in lines if line.angle == angle] + if not line_geometry: + continue + line_geometry.sort(key=lambda item: (item[1][1], item[1][0], item[0].source_index)) + effective_heights = [_line_effective_height(line, bbox) for line, bbox in line_geometry] + median_height = statistics.median(effective_heights) if effective_heights else 1.0 + local_page_width = page_size[1] if angle in {90, 270} else page_size[0] + local_page_height = page_size[0] if angle in {90, 270} else page_size[1] + local_visual_bboxes = [_rotate_bbox_to_upright(bbox, page_size, angle) for bbox in (visual_bboxes or [])] + lanes = _infer_text_lanes(line_geometry, local_page_width, median_height) + local_axis_lines = _transform_axis_lines(drawing_lines or [], page_size, angle) + split_row_counts: dict[int, int] = {} + for line, _bbox in line_geometry: + if line.visual_row_id is not None and line.split_from_row: + split_row_counts[line.visual_row_id] = split_row_counts.get(line.visual_row_id, 0) + 1 + + for lane in lanes: + lane.lines.sort(key=lambda item: (item[1][1], item[1][0], item[0].source_index)) + if not lane.lines: + continue + regular_gap, gap_mad = _estimate_lane_gap(lane) + local_lane_by_source = _infer_local_text_lane_map(lane) + structured_break_sources = _structured_text_break_sources( + lane, + regular_gap, + gap_mad, + ) + isolated_break_sources = _isolated_indented_paragraph_break_sources( + lane, + regular_gap, + gap_mad, + ) + structured_break_sources.update( + isolated_break_sources, + ) + visual_reset_sources = _centered_visual_reset_break_sources( + lane, + local_visual_bboxes, + local_page_height, + ) + typography_reset_sources = _leading_typography_reset_break_sources( + lane, + regular_gap, + gap_mad, + ) + formula_text_break_sources = _formula_style_text_row_break_sources( + lane, + ) + structured_break_sources.update(visual_reset_sources) + structured_break_sources.update(typography_reset_sources) + structured_break_sources.update(formula_text_break_sources) + protected_break_sources: set[int] = set() + protected_break_sources.update(visual_reset_sources) + protected_break_sources.update(typography_reset_sources) + protected_break_sources.update(formula_text_break_sources) + protected_break_sources.update( + _front_matter_keyword_break_sources( + lane, + local_page_height, + page_index, + ) + ) + protected_break_sources.update( + _explicit_text_break_sources(lane), + ) + structured_break_sources.update( + protected_break_sources, + ) + hanging_indent_groups = _build_hanging_indent_group_map( + lane, + table_bboxes, + local_axis_lines, + ) + component: list[tuple[_LineItem, BBox]] = [lane.lines[0]] + components: list[list[tuple[_LineItem, BBox]]] = [] + for previous, current in zip(lane.lines, lane.lines[1:]): + previous_type = previous[0].semantic_type + current_type = current[0].semantic_type + if previous_type != current_type: + should_connect = False + elif previous_type is not None: + should_connect = _should_connect_semantic_rows( + previous, + current, + lane, + regular_gap, + table_bboxes, + local_axis_lines, + ) + else: + previous_group = hanging_indent_groups.get(previous[0].source_index) + current_group = hanging_indent_groups.get(current[0].source_index) + previous_local_lane = local_lane_by_source.get(previous[0].source_index) + current_local_lane = local_lane_by_source.get(current[0].source_index) + connection_lane = ( + current_local_lane + if current_local_lane is not None and previous_local_lane is current_local_lane + else lane + ) + if ( + current[0].style_scale_repaired + and current[0].split_from_row + and current[0].visual_row_id is not None + and split_row_counts.get( + current[0].visual_row_id, + 0, + ) + >= 2 + ): + should_connect = False + elif current[0].source_index in structured_break_sources: + should_connect = False + elif _starts_structural_reference_entry(previous, current): + # 编号只确认已经由悬挂缩进几何形成的新条目，不能单独扩张范围。 + should_connect = False + elif is_hyphen_at_line_end(previous[0].text): + # 断词续行优先于悬挂缩进分组，但仍复用正文连接中的距离和障碍限制。 + should_connect = _should_connect_text_rows( + previous, + current, + connection_lane, + regular_gap, + gap_mad, + table_bboxes, + local_axis_lines, + ) + elif previous_group is not None or current_group is not None: + should_connect = previous_group is not None and previous_group == current_group + else: + should_connect = _should_connect_text_rows( + previous, + current, + connection_lane, + regular_gap, + gap_mad, + table_bboxes, + local_axis_lines, + ) + if should_connect: + component.append(current) + else: + components.append(component) + component = [current] + components.append(component) + + for component_geometry in components: + component_lines = [item[0] for item in component_geometry] + component_local_lane = local_lane_by_source.get(component_lines[0].source_index) + if component_local_lane is None or not all( + local_lane_by_source.get(line.source_index) is component_local_lane for line in component_lines + ): + component_local_lane = lane + if component_lines[0].semantic_type == "doc_title": + # 文档标题保留自然换行，避免混排标题因语言检测在中文折行处插入空格。 + content = "\n".join( + normalized for line in component_lines if (normalized := _normalize_native_run_text(line.text)) + ) + else: + content = _merge_text_line_content([line.text for line in component_lines]) + if not content: + continue + visual_row_ids = {line.visual_row_id for line in component_lines if line.visual_row_id is not None} + single_run_row_id = ( + component_lines[0].visual_row_id + if len(component_lines) == 1 + and component_lines[0].split_from_row + and component_lines[0].visual_row_id is not None + else None + ) + local_output_line_bboxes, output_bbox_repaired = _local_tight_output_line_bboxes( + component_lines, + page_size, + angle, + ) + blocks.append( + { + "type": component_lines[0].semantic_type or "text", + "bbox": _bbox_union_many([line.bbox for line in component_lines]), + "angle": angle, + "content": content, + "_visual_row_ids": visual_row_ids, + "_single_run_row_id": single_run_row_id, + "_local_line_bboxes": [bbox for _line, bbox in component_geometry], + "_local_output_line_bboxes": local_output_line_bboxes, + "_output_bbox_repaired": output_bbox_repaired, + "_line_heights": [_line_effective_height(line, bbox) for line, bbox in component_geometry], + "_font_signatures": { + line.font_signature + for line in component_lines + if line.font_signature is not None and line.font_coverage >= 0.5 + }, + "_inline_math_regions": [region for line in component_lines for region in line.inline_math_regions], + _PARAGRAPH_FORMULA_CONTEXT_MARKER: any(line.paragraph_formula_context for line in component_lines), + "_lane_interval": ( + component_local_lane.left, + component_local_lane.right, + ), + "_lane_is_span": component_local_lane.is_span, + "_hard_break_before": (component_lines[0].source_index in structured_break_sources), + "_protected_hard_break_before": (component_lines[0].source_index in protected_break_sources), + "_hanging_indent_group": hanging_indent_groups.get( + component_lines[0].source_index, + ), + "_leading_emphasis_start": _component_starts_with_emphasized_row( + component_lines, + ), + } + ) + blocks = _merge_short_same_baseline_prefix_blocks( + blocks, + page_size, + ) + blocks = _merge_spatial_text_components(blocks, page_size) + blocks = _merge_list_intro_text_components(blocks) + blocks = _merge_unterminated_text_components(blocks) + blocks = _merge_overlapping_same_line_text_blocks(blocks, page_size) + blocks = _merge_inline_math_fragment_text_blocks( + blocks, + page_size, + ) + return _merge_paragraph_formula_context_blocks( + blocks, + page_size, + ) + + +__all__ = ["_build_text_blocks"] diff --git a/mineru/model/flash/pdf/text_assembly/common.py b/mineru/model/flash/pdf/text_assembly/common.py new file mode 100644 index 000000000..b1d477717 --- /dev/null +++ b/mineru/model/flash/pdf/text_assembly/common.py @@ -0,0 +1,482 @@ +# Copyright (c) Opendatalab. All rights reserved. +"""共享正文块连接的几何、内容拼接和来源规则。""" + +from __future__ import annotations + +import re +from typing import Any, Sequence + +from .....utils.language import detect_lang +from .....utils.text import merge_text_line_contents +from ..geometry import ( + _bbox_axis_overlap_ratio, + _bbox_center_y, + _bbox_union_many, +) +from ..native_text import _normalize_native_run_text + +_REFERENCE_ENTRY_RE = re.compile(r"^[［\[]\s*\d+\s*[］\]]") + + +_FIGURE_CAPTION_MARKER_RE = re.compile( + r"^(?:图\s*[0-9０-９一二三四五六七八九十]|fig(?:ure)?\.?\s*[0-9])", + re.IGNORECASE, +) + + +_INLINE_MATH_RECOVERY_MARKER = "_recovered_inline_math_fragments" + + +_PARAGRAPH_FORMULA_CONTEXT_MARKER = "_paragraph_formula_context" + + +_FRONT_MATTER_FIELD_RE = re.compile( + r"^\s*(?:keywords?|key\s+words?|关键词|中图分类号|文献标识码|文章编号)\s*[:：]", + re.IGNORECASE, +) + + +_LIST_ITEM_RE = re.compile( + r"^\s*(?:[（(]\s*(?:\d+|[ivxlcdm]+)\s*[）)]|[①-⑳]|[•●▪])", + re.IGNORECASE, +) + + +_BULLET_ITEM_RE = re.compile( + r"^\s*[•●▪]", +) + + +_EMAIL_METADATA_RE = re.compile( + r"^\s*e[\s-]*mail\s*[:：]", + re.IGNORECASE, +) + + +_ABSTRACT_METADATA_RE = re.compile( + r"^\s*(?:abstract|摘\s*要)\s*[:：]", + re.IGNORECASE, +) + + +_LABELLED_METADATA_RE = re.compile( + r"^\s*(?P