opensource/MinerU
3
0
Fork 0
mirror of https://github.com/opendatalab/MinerU.git synced 2026-10-28 02:36:45 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: enhance formula detection by adding support for detached above body detection and refining title barrier identification

Browse Source
This commit is contained in:
myhloli
2026-07-28 14:41:28 +08:00
parent 1d01204ba0
commit c57364f9b7
13 changed files with 1908 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mineru/backend/flash/native_pdf/formulas.py
+41 -5
View File
@@ -757,7 +757,12 @@ def _find_formula_spatial_anchors(
continue
center_y = _bbox_center_y(bbox)
detached_below_body = body_bottom < center_y <= body_bottom + 6.0 * median_height
if not body_top <= center_y <= body_bottom and not detached_below_body:
detached_above_body = body_top - 6.0 * median_height <= center_y < body_top
if (
not body_top <= center_y <= body_bottom
and not detached_below_body
and not detached_above_body
):
continue
left_peers = [
(other_line, other_bbox)
@@ -772,7 +777,7 @@ def _find_formula_spatial_anchors(
other_bbox,
_line_effective_height(other_line, other_bbox),
)
if detached_below_body
if detached_below_body or detached_above_body
else _formula_seed_vertical_match(
bbox,
line_height,
@@ -800,6 +805,7 @@ def _find_formula_spatial_anchors(
line=line,
bbox=bbox,
detached_below_body=detached_below_body,
detached_above_body=detached_above_body,
)
)
return _deduplicate_formula_anchors(anchors, median_height)
@@ -908,6 +914,12 @@ def _grow_formula_spatial_component(
dominant_body_font,
median_height,
)
and not _is_formula_title_barrier(
item,
lane,
dominant_body_font,
median_height,
)
and not _is_formula_body_prefix(
item,
lane,
@@ -928,7 +940,7 @@ def _grow_formula_spatial_component(
item[1],
_line_effective_height(*item),
)
if anchor.detached_below_body
if anchor.detached_below_body or anchor.detached_above_body
else _formula_seed_vertical_match(
anchor_bbox,
_line_effective_height(anchor_line, anchor_bbox),
@@ -972,7 +984,7 @@ def _is_formula_body_barrier(
dominant_body_font: tuple[str, int] | None,
median_height: float,
) -> bool:
"""识别近满栏、高置信正文行，阻止公式连通分量跨越正文边界。"""
"""识别具有稳定正文排版的行，阻止公式分量吸收正文尾行。"""
if dominant_body_font is None:
return False
@@ -981,11 +993,35 @@ def _is_formula_body_barrier(
line_height = _line_effective_height(line, bbox)
return (
line.font_signature == dominant_body_font
and bbox[2] - bbox[0] >= 0.85 * lane_width
and line.font_coverage >= 0.75
and bbox[2] - bbox[0] >= 0.3 * lane_width
and 0.8 * median_height <= line_height <= 1.25 * median_height
)
def _is_formula_title_barrier(
candidate: tuple[_LineItem, BBox],
lane: _TextLane,
dominant_body_font: tuple[str, int] | None,
median_height: float,
) -> bool:
"""用左对齐、字号突变和字体变化隔离公式下方的章节标题。"""
if dominant_body_font is None:
return False
line, bbox = candidate
lane_width = max(0.1, lane.right - lane.left)
line_height = _line_effective_height(line, bbox)
return (
line.font_signature is not None
and line.font_signature != dominant_body_font
and line.font_coverage >= 0.75
and 1.1 * median_height <= line_height <= 1.6 * median_height
and bbox[2] - bbox[0] >= 0.25 * lane_width
and abs(bbox[0] - lane.left) <= median_height
)
def _is_formula_body_prefix(
candidate: tuple[_LineItem, BBox],
lane: _TextLane,
mineru/backend/flash/native_pdf/line_layout.py
+268 -41
View File
@@ -74,16 +74,30 @@ def _should_connect_semantic_rows(
return False
if (
previous_line.semantic_type == "paragraph_title"
and vertical_gap > 0.65 * pair_height
and vertical_gap > 0.5 * pair_height
):
return False
if (
font_conflicts = (
previous_line.font_signature is not None
and current_line.font_signature is not None
and previous_line.font_coverage >= 0.75
and current_line.font_coverage >= 0.75
and previous_line.font_signature != current_line.font_signature
):
)
uncertain_document_title_font = (
previous_line.semantic_type == "doc_title"
and min(previous_line.font_coverage, current_line.font_coverage) < 0.85
and (
previous_line.dominant_font_weight is None
or current_line.dominant_font_weight is None
or abs(
previous_line.dominant_font_weight
- current_line.dominant_font_weight
)
< 100.0
)
)
if font_conflicts and not uncertain_document_title_font:
return False
lane_width = max(0.1, lane.right - lane.left)
centered_pair = (
@@ -127,9 +141,15 @@ def _infer_text_lanes(
"""从重复左右边缘推断稳定栏带，并把跨栏行放入独立 span lane。"""
anchor_tolerance = max(3.0, 0.75 * median_height)
regular_lines = [
anchor_geometry = [
item
for item in line_geometry
if item[0].semantic_type
not in {"header", "footer", "page_number", "page_footnote", "aside_text"}
]
regular_lines = [
item
for item in anchor_geometry
if item[1][2] - item[1][0] >= max(4.0 * _line_effective_height(*item), 0.15 * local_page_width)
]
left_clusters: list[list[tuple[_LineItem, BBox]]] = []
@@ -175,6 +195,67 @@ def _infer_text_lanes(
)
]
nested_column_band = None
if len(filtered_intervals) == 1:
nested_column_band = _infer_nested_column_band(
anchor_geometry,
local_page_width,
median_height,
filtered_intervals[0][:2],
)
if nested_column_band is not None:
nested_lanes, band_top, band_bottom = nested_column_band
fallback_lane = _TextLane(
left=filtered_intervals[0][0],
right=filtered_intervals[0][1],
)
span_lines: list[tuple[_LineItem, BBox]] = []
for item in line_geometry:
line, bbox = item
center_y = _bbox_center_y(bbox)
if (
line.semantic_type
in {"header", "footer", "page_number", "page_footnote", "aside_text"}
or not band_top <= center_y <= band_bottom
):
fallback_lane.lines.append(item)
continue
line_width = max(0.1, bbox[2] - bbox[0])
scored_lanes = [
(
max(0.0, min(bbox[2], lane.right) - max(bbox[0], lane.left))
/ line_width,
lane,
)
for lane in nested_lanes
]
coverage_scores = sorted(
(coverage for coverage, _lane in scored_lanes),
reverse=True,
)
if len(coverage_scores) > 1 and coverage_scores[1] >= 0.2:
span_lines.append(item)
continue
best_coverage, best_lane = max(scored_lanes, key=lambda value: value[0])
if best_coverage >= 0.5:
best_lane.lines.append(item)
else:
span_lines.append(item)
lanes = [lane for lane in nested_lanes if lane.lines]
if fallback_lane.lines:
lanes.append(fallback_lane)
if span_lines:
lanes.append(
_TextLane(
left=min(item[1][0] for item in span_lines),
right=max(item[1][2] for item in span_lines),
lines=span_lines,
is_span=True,
)
)
_reattach_span_lane_continuations(lanes, median_height)
return lanes
lanes = [_TextLane(left=left, right=right) for left, right, _support in filtered_intervals]
span_lines: list[tuple[_LineItem, BBox]] = []
for item in line_geometry:
@@ -219,6 +300,113 @@ def _infer_text_lanes(
return lanes
def _infer_nested_column_band(
line_geometry: list[tuple[_LineItem, BBox]],
local_page_width: float,
median_height: float,
outer_interval: tuple[float, float],
) -> tuple[list[_TextLane], float, float] | None:
"""在全宽版心内查找仅占局部纵向区间的并列正文栏。"""
outer_width = max(0.1, outer_interval[1] - outer_interval[0])
candidates = [
item
for item in line_geometry
if item[0].semantic_type is None
and max(4.0 * _line_effective_height(*item), 0.12 * local_page_width)
<= item[1][2] - item[1][0]
<= 0.62 * outer_width
]
if len(candidates) < 6:
return None
center_tolerance = max(2.0 * median_height, 0.06 * local_page_width)
center_clusters: list[list[tuple[_LineItem, BBox]]] = []
for item in sorted(candidates, key=lambda value: _bbox_center_x(value[1])):
center = _bbox_center_x(item[1])
target = next(
(
cluster
for cluster in center_clusters
if abs(
center
- statistics.median(_bbox_center_x(member[1]) for member in cluster)
)
<= center_tolerance
),
None,
)
if target is None:
center_clusters.append([item])
else:
target.append(item)
supported = [cluster for cluster in center_clusters if len(cluster) >= 3]
supported.sort(
key=lambda cluster: statistics.median(_bbox_center_x(item[1]) for item in cluster)
)
best_pair: tuple[
tuple[int, float, float],
list[tuple[_LineItem, BBox]],
list[tuple[_LineItem, BBox]],
tuple[float, float],
tuple[float, float],
] | None = None
for left_cluster, right_cluster in zip(supported, supported[1:]):
left_interval = (
statistics.median(item[1][0] for item in left_cluster),
statistics.median(item[1][2] for item in left_cluster),
)
right_interval = (
statistics.median(item[1][0] for item in right_cluster),
statistics.median(item[1][2] for item in right_cluster),
)
gutter = right_interval[0] - left_interval[1]
common_top = max(
min(item[1][1] for item in left_cluster),
min(item[1][1] for item in right_cluster),
)
common_bottom = min(
max(item[1][3] for item in left_cluster),
max(item[1][3] for item in right_cluster),
)
combined_width = right_interval[1] - left_interval[0]
if (
gutter < max(6.0, 0.75 * median_height)
or common_bottom - common_top < 2.0 * median_height
or combined_width < 0.55 * outer_width
):
continue
score = (
min(len(left_cluster), len(right_cluster)),
common_bottom - common_top,
gutter,
)
candidate_pair = (
score,
left_cluster,
right_cluster,
left_interval,
right_interval,
)
if best_pair is None or candidate_pair[0] > best_pair[0]:
best_pair = candidate_pair
if best_pair is None:
return None
_score, left_cluster, right_cluster, left_interval, right_interval = best_pair
band_top = min(item[1][1] for item in [*left_cluster, *right_cluster]) - 0.5 * median_height
band_bottom = max(item[1][3] for item in [*left_cluster, *right_cluster]) + 0.5 * median_height
return (
[
_TextLane(left=left_interval[0], right=left_interval[1]),
_TextLane(left=right_interval[0], right=right_interval[1]),
],
band_top,
band_bottom,
)
def _reattach_span_lane_continuations(
lanes: list[_TextLane],
median_height: float,
@@ -231,26 +419,10 @@ def _reattach_span_lane_continuations(
return
for span_lane in span_lanes:
span_lane.lines.sort(
key=lambda item: (item[1][1], item[1][0], item[0].source_index)
)
if len(span_lane.lines) < 2:
continue
previous, last = span_lane.lines[-2:]
previous_height = _line_effective_height(*previous)
last_height = _line_effective_height(*last)
if (
previous[0].semantic_type != last[0].semantic_type
or abs(previous[1][0] - last[1][0]) > 0.75 * median_height
or max(previous_height, last_height) / min(previous_height, last_height) > 1.35
or not _title_fonts_compatible(previous[0], last[0])
or not -0.25 * median_height
<= _effective_text_row_gap(previous, last)
<= 0.75 * median_height
):
continue
while True:
span_lane.lines.sort(
key=lambda item: (item[1][1], item[1][0], item[0].source_index)
)
candidates: list[
tuple[
float,
@@ -262,9 +434,26 @@ def _reattach_span_lane_continuations(
for regular_lane in regular_lanes:
for candidate in regular_lane.lines:
candidate_line, candidate_bbox = candidate
preceding = [
item
for item in span_lane.lines
if item[0].semantic_type == candidate_line.semantic_type
and item[1][1] < candidate_bbox[1]
]
if len(preceding) < 2:
continue
previous, last = preceding[-2:]
previous_height = _line_effective_height(*previous)
last_height = _line_effective_height(*last)
if (
candidate_line.semantic_type != last[0].semantic_type
or candidate_bbox[1] <= last[1][1]
abs(previous[1][0] - last[1][0]) > 0.75 * median_height
or max(previous_height, last_height)
/ min(previous_height, last_height)
> 1.35
or not _title_fonts_compatible(previous[0], last[0])
or not -0.25 * median_height
<= _effective_text_row_gap(previous, last)
<= 0.75 * median_height
):
continue
gap = _effective_text_row_gap(last, candidate)
@@ -293,15 +482,15 @@ def _reattach_span_lane_continuations(
continue
candidates.append(
(
max(0.0, gap),
candidate_bbox[1],
max(0.0, gap),
regular_lane,
candidate,
)
)
if not candidates:
break
_gap, _top, regular_lane, candidate = min(
_top, _gap, regular_lane, candidate = min(
candidates,
key=lambda item: (item[0], item[1]),
)
@@ -312,8 +501,6 @@ def _reattach_span_lane_continuations(
span_lane.lines.sort(
key=lambda item: (item[1][1], item[1][0], item[0].source_index)
)
last = candidate
last_height = _line_effective_height(*last)
def _estimate_lane_gap(lane: _TextLane) -> tuple[float, float]:
@@ -381,18 +568,52 @@ def _should_connect_text_rows(
return False
if (
current_height < 0.88 * previous_height
and vertical_gap > regular_gap + 0.25 * previous_height
and vertical_gap
> regular_gap + max(0.25 * previous_height, 3.0 * gap_mad)
):
return False
height_ratio = max(previous_height, current_height) / min(previous_height, current_height)
if height_ratio > 1.35:
both_fill_lane = previous_width >= 0.8 * lane_width and current_width >= 0.8 * lane_width
aligned_left_edges = abs(previous_bbox[0] - current_bbox[0]) <= 0.5 * pair_height
font_style_changed = (
previous_line.font_signature is not None
and current_line.font_signature is not None
and previous_line.font_signature[1] != current_line.font_signature[1]
both_fill_lane = previous_width >= 0.8 * lane_width and current_width >= 0.8 * lane_width
aligned_left_edges = abs(previous_bbox[0] - current_bbox[0]) <= 0.5 * pair_height
current_returns_to_lane_left = (
abs(current_bbox[0] - lane.left) <= 0.75 * pair_height
and -0.5 * pair_height
<= previous_bbox[0] - lane.left
<= 2.0 * pair_height
)
reliable_font_match = (
previous_line.font_signature is None
or current_line.font_signature is None
or previous_line.font_coverage < 0.75
or current_line.font_coverage < 0.75
or previous_line.font_signature == current_line.font_signature
or (
current_width <= 0.5 * lane_width
and previous_line.font_signature[1] == current_line.font_signature[1]
)
)
safe_short_tail = (
previous_width >= 0.75 * lane_width
and current_width <= 0.7 * lane_width
and (aligned_left_edges or current_returns_to_lane_left)
and reliable_font_match
and -0.25 * pair_height
<= vertical_gap
<= regular_gap + max(0.75 * pair_height, 3.0 * gap_mad)
)
height_ratio = max(previous_height, current_height) / min(previous_height, current_height)
font_style_changed = (
previous_line.font_signature is not None
and current_line.font_signature is not None
and previous_line.font_signature[1] != current_line.font_signature[1]
)
full_width_continuation = (
both_fill_lane
and aligned_left_edges
and not font_style_changed
and vertical_gap
<= regular_gap + max(0.75 * min(previous_height, current_height), 3.0 * gap_mad)
)
if height_ratio > 1.35 and not safe_short_tail and not full_width_continuation:
# 满栏混合字体可跨字号续接，但显式正体/斜体等样式边界仍保持原分段语义。
if not both_fill_lane or not aligned_left_edges or font_style_changed:
return False
@@ -401,7 +622,7 @@ def _should_connect_text_rows(
return False
if _bbox_axis_overlap_ratio(previous_bbox, current_bbox, axis="x") < 0.5 and abs(
previous_bbox[0] - current_bbox[0]
) > 1.5 * pair_height:
) > 1.5 * pair_height and not safe_short_tail:
return False
if _connection_crosses_table(previous_line.bbox, current_line.bbox, table_bboxes):
return False
@@ -431,7 +652,7 @@ def _should_connect_text_rows(
if next_indent >= max(5.0, 0.65 * pair_height) and (previous_fill <= 0.8 or terminal_previous):
return False
abnormal_gap = vertical_gap > regular_gap + 0.25 * pair_height
abnormal_gap = vertical_gap > regular_gap + max(0.25 * pair_height, 3.0 * gap_mad)
if (
previous_line.font_signature is not None
and current_line.font_signature is not None
@@ -439,9 +660,15 @@ def _should_connect_text_rows(
and current_line.font_coverage >= 0.75
and previous_line.font_signature != current_line.font_signature
and (abnormal_gap or min(previous_width, current_width) <= 0.7 * lane_width)
and not both_fill_lane
and not safe_short_tail
):
return False
if abnormal_gap and min(previous_width, current_width) <= 0.65 * lane_width:
if (
abnormal_gap
and min(previous_width, current_width) <= 0.65 * lane_width
and not safe_short_tail
):
return False
return True
mineru/backend/flash/native_pdf/models.py
+2 -1
View File
@@ -109,11 +109,12 @@ class _TextLane:
@dataclass(slots=True)
class _FormulaAnchor:
"""保存公式右缘锚点及其是否落在正文密集区下方。"""
"""保存公式右缘锚点及其相对正文密集区的上下位置。"""
line: _LineItem
bbox: BBox
detached_below_body: bool = False
detached_above_body: bool = False
@dataclass(slots=True)
mineru/backend/flash/native_pdf/pipeline.py
+173 -4
View File
@@ -16,6 +16,8 @@ from .models import (
_PreparedPage,
)
from .geometry import (
_bbox_axis_overlap_ratio,
_bbox_center_y,
_bbox_overlap_in_smaller,
_bbox_union_many,
_clip_bbox,
@@ -59,7 +61,11 @@ from .auxiliary_text import (
_classify_repeated_visual_headers,
)
from .titles import _classify_page_titles
from .text_blocks import _build_text_blocks
from .text_blocks import (
_build_text_blocks,
_merge_fragmented_header_blocks,
_merge_image_caption_text_blocks,
)
_TEXT_SEMANTIC_TYPES = {
@@ -249,6 +255,15 @@ def _finalize_prepared_page(
prepared.drawing_lines,
page_footnote_groups=prepared.page_footnote_groups,
)
text_blocks = _merge_image_caption_text_blocks(
text_blocks,
[
block["bbox"]
for block in prepared.fixed_blocks
if block.get("type") == "image"
],
)
text_blocks = _merge_fragmented_header_blocks(text_blocks)
absolute_blocks = prepared.fixed_blocks + formula_blocks + text_blocks
sorted_blocks = _sort_blocks_with_visual_row_groups(absolute_blocks, prepared.page_size)
return [
@@ -272,8 +287,26 @@ def _sort_blocks_with_visual_row_groups(
) -> list[dict[str, Any]]:
"""把同一粗行拆出的单行 block 包装成虚拟项排序，再按局部 x 顺序展开。"""
top_marginals: list[dict[str, Any]] = []
bottom_marginals: list[dict[str, Any]] = []
body_blocks: list[dict[str, Any]] = []
local_page_height = page_size[1]
for block in blocks:
block_type = block.get("type")
bbox = block.get("bbox")
if block_type == "header" or (
block_type == "page_number"
and isinstance(bbox, (list, tuple))
and _bbox_center_y(bbox) <= 0.5 * local_page_height
):
top_marginals.append(block)
elif block_type in {"footer", "page_footnote"} or block_type == "page_number":
bottom_marginals.append(block)
else:
body_blocks.append(block)
grouped_indices: dict[int, list[int]] = {}
for index, block in enumerate(blocks):
for index, block in enumerate(body_blocks):
row_id = block.get("_single_run_row_id")
if isinstance(row_id, int):
grouped_indices.setdefault(row_id, []).append(index)
@@ -283,7 +316,7 @@ def _sort_blocks_with_visual_row_groups(
for row_id, indices in grouped_indices.items():
if len(indices) < 2:
continue
members = [blocks[index] for index in indices]
members = [body_blocks[index] for index in indices]
virtual_group = {
"type": "_xycut_visual_row_group",
"bbox": _bbox_union_many([member["bbox"] for member in members]),
@@ -294,7 +327,11 @@ def _sort_blocks_with_visual_row_groups(
virtual_groups[row_id] = virtual_group
consumed_indices.update(indices)
sortable_blocks = [block for index, block in enumerate(blocks) if index not in consumed_indices]
sortable_blocks = [
block
for index, block in enumerate(body_blocks)
if index not in consumed_indices
]
sortable_blocks.extend(virtual_groups.values())
sorted_payloads = sort_entries(sortable_blocks)
output: list[dict[str, Any]] = []
@@ -311,9 +348,141 @@ def _sort_blocks_with_visual_row_groups(
)
)
output.extend(members)
output = _stabilize_overlapping_lane_order(output, page_size)
return [
*_sort_marginal_blocks(top_marginals, page_size),
*output,
*_sort_marginal_blocks(bottom_marginals, page_size),
]
def _sort_marginal_blocks(
blocks: list[dict[str, Any]],
page_size: tuple[float, float],
) -> list[dict[str, Any]]:
"""先按视觉中心聚合边缘同排块，再按行内 x 排序以消除字体框顶边抖动。"""
if len(blocks) < 2:
return list(blocks)
geometry = [
(
block,
_rotate_bbox_to_upright(
block["bbox"],
page_size,
int(block.get("angle", 0) or 0) % 360,
),
)
for block in blocks
]
heights = [
float(height)
for block in blocks
for height in block.get("_line_heights", [])
if isinstance(height, (int, float)) and height > 0
]
median_height = sorted(heights)[len(heights) // 2] if heights else 1.0
rows: list[list[tuple[dict[str, Any], tuple[float, float, float, float]]]] = []
for item in sorted(geometry, key=lambda value: _bbox_center_y(value[1])):
target = next(
(
row
for row in rows
if abs(
_bbox_center_y(item[1])
- sum(_bbox_center_y(member[1]) for member in row) / len(row)
)
<= 0.75 * median_height
),
None,
)
if target is None:
rows.append([item])
else:
target.append(item)
rows.sort(
key=lambda row: sum(_bbox_center_y(member[1]) for member in row) / len(row)
)
return [
block
for row in rows
for block, _bbox in sorted(row, key=lambda member: member[1][0])
]
def _stabilize_overlapping_lane_order(
blocks: list[dict[str, Any]],
page_size: tuple[float, float],
) -> list[dict[str, Any]]:
"""对同栏轻微重叠块按视觉中心纠正局部逆序，不改变跨栏主阅读顺序。"""
output = list(blocks)
for _pass_index in range(len(output)):
changed = False
for index in range(len(output) - 1):
first = output[index]
second = output[index + 1]
if not _overlapping_lane_pair_is_inverted(first, second, page_size):
continue
output[index], output[index + 1] = second, first
changed = True
if not changed:
break
return output
def _overlapping_lane_pair_is_inverted(
first: dict[str, Any],
second: dict[str, Any],
page_size: tuple[float, float],
) -> bool:
"""判断相邻块是否属于同一内部栏带且视觉中心顺序与当前结果相反。"""
first_interval = first.get("_lane_interval")
second_interval = second.get("_lane_interval")
if (
not isinstance(first_interval, (list, tuple))
or not isinstance(second_interval, (list, tuple))
or len(first_interval) != 2
or len(second_interval) != 2
or first.get("_lane_is_span") != second.get("_lane_is_span")
or int(first.get("angle", 0) or 0) % 360
!= int(second.get("angle", 0) or 0) % 360
):
return False
first_bbox = _rotate_bbox_to_upright(
first["bbox"],
page_size,
int(first.get("angle", 0) or 0) % 360,
)
second_bbox = _rotate_bbox_to_upright(
second["bbox"],
page_size,
int(second.get("angle", 0) or 0) % 360,
)
line_heights = [
float(height)
for block in (first, second)
for height in block.get("_line_heights", [])
if isinstance(height, (int, float)) and height > 0
]
tolerance = 0.75 * (min(line_heights) if line_heights else 1.0)
same_lane = (
abs(float(first_interval[0]) - float(second_interval[0])) <= tolerance
and abs(float(first_interval[1]) - float(second_interval[1])) <= tolerance
)
vertical_overlap = min(first_bbox[3], second_bbox[3]) - max(
first_bbox[1],
second_bbox[1],
)
return (
same_lane
and _bbox_center_y(first_bbox) > _bbox_center_y(second_bbox) + 0.1 * tolerance
and _bbox_axis_overlap_ratio(first_bbox, second_bbox, axis="x") >= 0.35
and vertical_overlap >= 0.0
)
def _normalize_output_block(
block: dict[str, Any],
page_size: tuple[float, float],
mineru/backend/flash/native_pdf/tables.py
+75 -15
View File
@@ -48,8 +48,7 @@ _TABLE_CAPTION_RE = re.compile(
_TABLE_NOTE_RE = re.compile(
r"^(?:notes?|sources?)\b|^(?:注释?|说明)\s*[:：]?|^for\s+[*†‡]"
r"|^(?:\d+|[*†‡])\s+\S"
r"|^(?:[*†‡]|[a-z]|p|t|ns|na)\s+(?:indicates?|denotes?|rainfall\b|total\b|low\b|for\b)",
r"|^[*†‡]\s*\S",
re.IGNORECASE,
)
@@ -561,6 +560,7 @@ def _build_rule_table_candidates(
boundary_rules,
accepted_rows,
rows,
lines,
page_size,
angle,
median_height,
@@ -696,7 +696,7 @@ def _rule_intervals_are_column_compatible(
) -> bool:
"""拒绝跨过长篇栏式正文、导致稳定列数明显塌缩的多表合并区间。"""
profiles: list[tuple[int, float]] = []
profiles: list[tuple[int, float, int, float]] = []
for top_rule, bottom_rule in zip(rule_group, rule_group[1:]):
top = _bbox_center_y(top_rule.bbox)
bottom = _bbox_center_y(bottom_rule.bbox)
@@ -705,20 +705,33 @@ def _rule_intervals_are_column_compatible(
for row in rows
if top <= row.center_y <= bottom and len(row.fragments) >= 2
]
stable_columns, _coverage = _count_stable_columns(
stable_columns, column_coverage = _count_stable_columns(
interval_rows,
median_height,
)
profiles.append((stable_columns, bottom - top))
maximum_columns = max((columns for columns, _height in profiles), default=0)
if maximum_columns < 4:
return True
for interval_index, (columns, interval_height) in enumerate(profiles):
# 首个区间可能只是跨列表头；后续长区间若退化成普通双栏正文，
# 就不能把前后两张表合成一个候选。
if interval_index == 0:
profiles.append(
(
stable_columns,
bottom - top,
len(interval_rows),
column_coverage,
)
)
maximum_columns = max(
(columns for columns, _height, _row_count, _coverage in profiles),
default=0,
)
for interval_index, (columns, interval_height, row_count, coverage) in enumerate(profiles):
# 紧凑首区间可能只是跨列表头；一旦区间明显高于普通表头，
# 也必须具有连续多单元格行，不能无条件跨过正文连接两张表。
if interval_index == 0 and interval_height <= 2.5 * median_height:
continue
if columns < max(2, int(0.5 * maximum_columns)) and interval_height > 4.0 * median_height:
if interval_height <= 6.0 * median_height:
continue
minimum_rows = max(2, int(interval_height / max(8.0 * median_height, 0.1)))
if row_count < minimum_rows or coverage < 0.5:
return False
if columns < max(2, int(0.5 * maximum_columns)):
return False
return True
@@ -915,6 +928,7 @@ def _expand_rule_table_candidate(
rule_group: list[_LocalAxisLine],
core_rows: list[_VisualRow],
all_rows: list[_VisualRow],
all_lines: list[_LineItem],
page_size: tuple[float, float],
angle: int,
median_height: float,
@@ -927,6 +941,7 @@ def _expand_rule_table_candidate(
caption_rows = _collect_caption_rows(all_rows, caption_line, rule_bbox, median_height)
footnote_rows = _collect_footnote_rows(
all_rows,
all_lines,
rule_bbox,
median_height,
core_line_indices,
@@ -986,6 +1001,7 @@ def _collect_caption_rows(
def _collect_footnote_rows(
rows: list[_VisualRow],
lines: list[_LineItem],
rule_bbox: BBox,
median_height: float,
core_line_indices: set[int],
@@ -997,6 +1013,10 @@ def _collect_footnote_rows(
note_chain_started = False
margin = 2.0 * median_height
selected_line_indices = set(core_line_indices)
line_by_index = {line.source_index: line for line in lines}
note_left: float | None = None
note_height: float | None = None
note_fonts: set[tuple[str, int]] = set()
for row in rows:
clipped_row = _clip_visual_row_to_corridor(row, rule_bbox, margin=margin)
if clipped_row is None or clipped_row.bbox[3] <= bottom:
@@ -1005,9 +1025,49 @@ def _collect_footnote_rows(
if line_indices.issubset(selected_line_indices):
bottom = max(bottom, clipped_row.bbox[3])
continue
if max(0.0, clipped_row.bbox[1] - bottom) > 1.5 * median_height:
row_gap = max(0.0, clipped_row.bbox[1] - bottom)
row_lines = [
line_by_index[line_index]
for line_index in line_indices
if line_index in line_by_index
]
row_heights = [
_line_effective_height(line, line.bbox)
for line in row_lines
]
row_height = statistics.median(row_heights) if row_heights else clipped_row.bbox[3] - clipped_row.bbox[1]
row_fonts = {
line.font_signature
for line in row_lines
if line.font_signature is not None and line.font_coverage >= 0.75
}
if row_gap > (1.25 if not note_chain_started else 1.0) * median_height:
break
if not note_chain_started and not _is_table_note_text(_visual_row_text(clipped_row)):
if not note_chain_started:
if not _is_table_note_text(_visual_row_text(clipped_row)):
break
if (
_bbox_axis_overlap_ratio(row.bbox, rule_bbox, axis="x") < 0.35
or abs(clipped_row.bbox[0] - rule_bbox[0]) > 2.0 * median_height
or row_height > 1.15 * median_height
):
break
note_left = clipped_row.bbox[0]
note_height = max(0.1, row_height)
note_fonts = row_fonts
elif (
note_left is None
or note_height is None
or abs(clipped_row.bbox[0] - note_left) > 1.5 * median_height
or not 0.75 <= row_height / note_height <= 1.25
or (
note_fonts
and row_fonts
and note_fonts.isdisjoint(row_fonts)
)
or _bbox_axis_overlap_ratio(row.bbox, rule_bbox, axis="x") < 0.35
):
# 字号、字体或缩进突变表明已进入标题/正文，表注链必须立即终止。
break
output.append(clipped_row)
selected_line_indices.update(line_indices)
mineru/backend/flash/native_pdf/text_blocks.py
+413 -2
View File
@@ -4,6 +4,7 @@
from __future__ import annotations
import re
import statistics
from typing import Any, Sequence
@@ -23,6 +24,7 @@ from .models import (
)
from .geometry import (
_bbox_axis_overlap_ratio,
_bbox_center_x,
_bbox_center_y,
_bbox_union_many,
_rotate_bbox_from_upright,
@@ -40,6 +42,34 @@ from .line_layout import (
_should_connect_semantic_rows,
_should_connect_text_rows,
)
_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,
)
def _starts_structural_reference_entry(
previous: tuple[_LineItem, BBox],
current: tuple[_LineItem, BBox],
) -> bool:
"""仅在编号行相对续行明显左突时确认新的参考文献条目。"""
if _REFERENCE_ENTRY_RE.match(current[0].text.strip()) is None:
return False
previous_height = _line_effective_height(*previous)
current_height = _line_effective_height(*current)
pair_height = max(previous_height, current_height)
return (
current[1][0] <= previous[1][0] - max(5.0, 0.6 * min(previous_height, current_height))
and -0.75 * pair_height
<= _effective_text_row_gap(previous, current)
<= 1.5 * pair_height
)
def _build_hanging_indent_group_map(
lane: _TextLane,
table_bboxes: list[BBox],
@@ -437,7 +467,10 @@ def _build_text_blocks(
else:
previous_group = hanging_indent_groups.get(previous[0].source_index)
current_group = hanging_indent_groups.get(current[0].source_index)
if is_hyphen_at_line_end(previous[0].text):
if _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,
@@ -469,7 +502,17 @@ def _build_text_blocks(
for component_geometry in components:
component_lines = [item[0] for item in component_geometry]
content = _merge_text_line_content([line.text for line in component_lines])
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 = {
@@ -495,6 +538,12 @@ def _build_text_blocks(
_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
},
"_lane_interval": (lane.left, lane.right),
"_lane_is_span": lane.is_span,
}
@@ -502,6 +551,256 @@ def _build_text_blocks(
return _merge_spatial_text_components(blocks, page_size)
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
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 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_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_tail_matches_seed(
seed: dict[str, Any],
candidate: dict[str, Any],
median_height: float,
) -> bool:
"""只用同栏角色、字体、邻接和投影把无标记的图注续行接回锚点。"""
if not _components_share_lane_role(seed, candidate, median_height):
return False
seed_bbox = seed["bbox"]
candidate_bbox = candidate["bbox"]
if _bbox_center_y(candidate_bbox) <= _bbox_center_y(seed_bbox):
return False
vertical_gap = max(0.0, candidate_bbox[1] - seed_bbox[3])
if (
vertical_gap > 1.25 * median_height
or _bbox_center_y(candidate_bbox) - _bbox_center_y(seed_bbox)
> 2.0 * 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_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") == "header" 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 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_internal_text_block_group(
blocks: list[dict[str, Any]],
indices: list[int],
*,
preserve_visual_spaces: bool = False,
) -> dict[str, Any]:
"""合并内部文本块及其版面元数据，最终输出阶段仍会统一移除这些字段。"""
ordered_indices = sorted(
indices,
key=lambda index: _text_component_sort_key(blocks[index]),
)
merged = dict(blocks[ordered_indices[0]])
merged["bbox"] = _bbox_union_many([blocks[index]["bbox"] for index in ordered_indices])
contents = [str(blocks[index].get("content", "")) for index in ordered_indices]
merged["content"] = (
" ".join(content.strip() for content in contents if content.strip())
if preserve_visual_spaces
else _merge_text_line_content(contents)
)
merged["_visual_row_ids"] = set().union(
*[
row_ids
for index in ordered_indices
if isinstance((row_ids := blocks[index].get("_visual_row_ids")), set)
]
)
merged["_single_run_row_id"] = None
merged["_local_line_bboxes"] = [
bbox
for index in ordered_indices
for bbox in blocks[index].get("_local_line_bboxes", [])
]
merged["_line_heights"] = [
height
for index in ordered_indices
for height in blocks[index].get("_line_heights", [])
]
merged["_font_signatures"] = set().union(
*[
signatures
for index in ordered_indices
if isinstance(
(signatures := blocks[index].get("_font_signatures")),
set,
)
]
)
return merged
def _merge_spatial_text_components(
blocks: list[dict[str, Any]],
page_size: tuple[float, float],
@@ -643,6 +942,16 @@ def _merge_spatial_text_components(
for index in ordered_indices
for height in blocks[index].get("_line_heights", [])
]
merged["_font_signatures"] = set().union(
*[
signatures
for index in ordered_indices
if isinstance(
(signatures := blocks[index].get("_font_signatures")),
set,
)
]
)
output.append(merged)
return output
@@ -699,6 +1008,33 @@ def _compatible_component_lane_width(
)
def _components_share_lane_role(
first_block: dict[str, Any],
second_block: dict[str, Any],
median_height: float,
) -> bool:
"""要求二次合并组件同为跨栏或属于同一普通栏带。"""
first_role = first_block.get("_lane_is_span")
second_role = second_block.get("_lane_is_span")
if isinstance(first_role, bool) and isinstance(second_role, bool):
if first_role != second_role:
return False
if first_role:
return True
first_interval = _component_lane_interval(first_block)
second_interval = _component_lane_interval(second_block)
if first_interval is None or second_interval is None:
return False
tolerance = 0.75 * median_height
return (
abs(first_interval[0] - second_interval[0]) <= tolerance
and abs(first_interval[1] - second_interval[1]) <= tolerance
)
# 兼容缺少内部栏元数据的旧调用；生产路径始终会携带该字段。
return not isinstance(first_role, bool) and not isinstance(second_role, bool)
def _block_starts_with_short_wide_rows(
block: dict[str, Any],
local_page_width: float,
@@ -737,6 +1073,12 @@ def _find_short_opener_pairs(
for body_index in candidate_indices:
if body_index == opener_index:
continue
if not _components_share_lane_role(
blocks[opener_index],
blocks[body_index],
median_height,
):
continue
body_bbox = blocks[body_index]["_local_line_bboxes"][0]
gap = body_bbox[1] - opener_bbox[3]
body_width = body_bbox[2] - body_bbox[0]
@@ -751,6 +1093,12 @@ def _find_short_opener_pairs(
or abs(body_bbox[0] - opener_bbox[0]) > 0.75 * median_height
or opener_width > 0.35 * reference_width
or body_width < 0.6 * reference_width
or _component_connection_skips_block(
blocks,
opener_index,
body_index,
median_height,
)
):
continue
matches.append((gap, body_index))
@@ -774,6 +1122,12 @@ def _nearest_following_text_component(
for candidate_index in candidate_indices:
if candidate_index == current_index or candidate_index in section_starts:
continue
if not _components_share_lane_role(
blocks[current_index],
blocks[candidate_index],
maximum_gap / 0.75,
):
continue
candidate_bbox = blocks[candidate_index]["_local_line_bboxes"][0]
gap = candidate_bbox[1] - current_bbox[3]
if not -0.25 * maximum_gap <= gap <= maximum_gap:
@@ -782,6 +1136,13 @@ def _nearest_following_text_component(
left_gap = abs(candidate_bbox[0] - current_bbox[0])
if overlap < 0.5 and left_gap > 3.0 * maximum_gap:
continue
if _component_connection_skips_block(
blocks,
current_index,
candidate_index,
maximum_gap / 0.75,
):
continue
matches.append((max(0.0, gap), left_gap, candidate_index))
return min(matches)[2] if matches else None
@@ -827,6 +1188,12 @@ def _nearest_tapered_tail_component(
for candidate_index in candidate_indices:
if candidate_index == current_index or candidate_index in section_starts:
continue
if not _components_share_lane_role(
blocks[current_index],
blocks[candidate_index],
median_height,
):
continue
candidate_bboxes = blocks[candidate_index]["_local_line_bboxes"]
first_bbox = candidate_bboxes[0]
last_bbox = candidate_bboxes[-1]
@@ -839,12 +1206,56 @@ def _nearest_tapered_tail_component(
or abs(first_bbox[0] - previous_bbox[0]) > 0.75 * median_height
or first_width > 0.85 * previous_width
or last_width > first_width + 0.5 * median_height
or _component_connection_skips_block(
blocks,
current_index,
candidate_index,
median_height,
)
):
continue
matches.append((gap, candidate_index))
return min(matches)[1] if matches else None
def _component_connection_skips_block(
blocks: list[dict[str, Any]],
first_index: int,
second_index: int,
median_height: float,
) -> bool:
"""检查两个正文组件之间是否已有同水平流的中间块，禁止二次合并跨越它。"""
first_bbox = blocks[first_index]["_local_line_bboxes"][-1]
second_bbox = blocks[second_index]["_local_line_bboxes"][0]
first_center = _bbox_center_y(first_bbox)
second_center = _bbox_center_y(second_bbox)
corridor_top, corridor_bottom = sorted((first_center, second_center))
corridor_bbox = (
min(first_bbox[0], second_bbox[0]),
corridor_top,
max(first_bbox[2], second_bbox[2]),
corridor_bottom,
)
for index, block in enumerate(blocks):
if index in {first_index, second_index}:
continue
local_bboxes = block.get("_local_line_bboxes")
if not isinstance(local_bboxes, list) or not local_bboxes:
continue
other_bbox = local_bboxes[0]
other_center = _bbox_center_y(other_bbox)
if not (
corridor_top + 0.1 * median_height
< other_center
< corridor_bottom - 0.1 * median_height
):
continue
if _bbox_axis_overlap_ratio(corridor_bbox, other_bbox, axis="x") >= 0.2:
return True
return False
def _text_component_sort_key(block: dict[str, Any]) -> tuple[float, float]:
"""返回正文组件按局部首行位置排序的稳定键。"""
mineru/backend/flash/native_pdf/titles.py
+166 -11
View File
@@ -19,6 +19,7 @@ from .geometry import (
_bbox_axis_overlap_ratio,
_bbox_center_x,
_bbox_center_y,
_bbox_union_many,
_rotate_bbox_to_upright,
)
from .line_layout import (
@@ -61,21 +62,29 @@ def _classify_page_titles(
lanes,
median_height,
)
local_container_bboxes = [
_rotate_bbox_to_upright(bbox, page_size, angle)
for bbox in container_bboxes
]
grid_title_suppressions.update(
_find_container_visual_row_title_suppressions(
line_geometry,
local_container_bboxes,
median_height,
)
)
document_title_bottom: float | None = None
if page_index == 0:
document_title_bottom = _classify_document_title(
lanes,
local_page_height,
local_page_width,
)
preserve_front_matter_boundaries = _document_title_uses_page_fallback(
lanes,
)
local_container_bboxes = [
_rotate_bbox_to_upright(bbox, page_size, angle)
for bbox in container_bboxes
]
for lane in lanes:
profile = _infer_lane_body_profile(lane)
_classify_paragraph_titles_in_lane(
@@ -194,6 +203,37 @@ def _find_repeated_grid_title_suppressions(
return set().union(*paired_bands)
def _find_container_visual_row_title_suppressions(
line_geometry: list[tuple[_LineItem, BBox]],
container_bboxes: list[BBox],
median_height: float,
) -> set[int]:
"""用完整视觉行与图表容器的邻接关系抑制拆分 caption 标题误报。"""
visual_rows: dict[int, list[tuple[_LineItem, BBox]]] = {}
for item in line_geometry:
row_id = item[0].visual_row_id
if row_id is not None:
visual_rows.setdefault(row_id, []).append(item)
suppressed: set[int] = set()
for members in visual_rows.values():
if len(members) < 2:
continue
row_bbox = _bbox_union_many([bbox for _line, bbox in members])
if any(
_bbox_axis_overlap_ratio(row_bbox, container_bbox, axis="x") >= 0.35
and max(
row_bbox[1] - container_bbox[3],
container_bbox[1] - row_bbox[3],
0.0,
)
<= 1.5 * median_height
for container_bbox in container_bboxes
):
suppressed.update(line.source_index for line, _bbox in members)
return suppressed
def _infer_lane_body_profile(lane: _TextLane) -> _LaneBodyProfile:
"""从栏带的长行主体估计正文行高、主字体、字重、常规行距和样式占比。"""
@@ -241,6 +281,7 @@ def _infer_lane_body_profile(lane: _TextLane) -> _LaneBodyProfile:
def _classify_document_title(
lanes: list[_TextLane],
local_page_height: float,
local_page_width: float,
) -> float | None:
"""从首页上部选取显著大字号锚点，并用同版式邻行扩展多行文档标题。"""
@@ -255,7 +296,7 @@ def _classify_document_title(
for lane, profile in lane_profiles:
lane_width = max(0.1, lane.right - lane.left)
available = [item for item in lane.lines if item[0].semantic_type is None]
for row_index, item in enumerate(available[:5]):
for row_index, item in enumerate(available):
line, bbox = item
reference_height = min(profile.body_height, 1.25 * document_body_height)
height_ratio = _line_effective_height(line, bbox) / max(0.1, reference_height)
@@ -265,6 +306,8 @@ def _classify_document_title(
)
width_ratio = (bbox[2] - bbox[0]) / lane_width
centered = abs(_bbox_center_x(bbox) - (lane.left + lane.right) / 2.0) <= 0.15 * lane_width
page_centered = abs(_bbox_center_x(bbox) - 0.5 * local_page_width) <= 0.15 * local_page_width
page_width_ratio = (bbox[2] - bbox[0]) / max(0.1, local_page_width)
spans_columns_fallback = (
lane.is_span
and centered
@@ -276,10 +319,22 @@ def _classify_document_title(
or (height_ratio < 1.4 and not spans_columns_fallback)
or width_ratio < 0.2
or (not centered and height_ratio < 1.7)
or (
not page_centered
and page_width_ratio < 0.45
and height_ratio < 1.8
)
):
continue
top_preference = max(0.0, 0.45 - _bbox_center_y(bbox) / local_page_height)
score = height_ratio + (0.75 if centered else 0.0) + top_preference - 0.05 * row_index
score = (
height_ratio
+ (0.75 if centered else 0.0)
+ (1.25 if page_centered else 0.0)
+ (0.75 if page_width_ratio >= 0.55 else 0.0)
+ top_preference
- 0.02 * row_index
)
candidates.append((score, lane, row_index, item))
if not candidates:
return None
@@ -301,7 +356,7 @@ def _classify_document_title(
candidate_height = _line_effective_height(candidate_line, candidate_bbox)
if not 0.8 <= candidate_height / anchor_height <= 1.25:
break
if not _title_fonts_compatible(anchor_line, candidate_line):
if not _document_title_fonts_compatible(anchor_line, candidate_line):
break
vertical_gap = max(candidate_bbox[1] - previous_bbox[3], previous_bbox[1] - candidate_bbox[3], 0.0)
lane_width = max(0.1, lane.right - lane.left)
@@ -316,6 +371,25 @@ def _classify_document_title(
return max(bbox[3] for _line, bbox in selected)
def _document_title_fonts_compatible(
first: _LineItem,
second: _LineItem,
) -> bool:
"""允许混排标题因主字体覆盖不足而切换字体，同时保留可靠字重屏障。"""
if _title_fonts_compatible(first, second):
return True
uncertain_dominant_font = min(first.font_coverage, second.font_coverage) < 0.85
weights_compatible = (
first.dominant_font_weight is None
or second.dominant_font_weight is None
or abs(first.dominant_font_weight - second.dominant_font_weight) < 100.0
or max(first.dominant_font_weight, second.dominant_font_weight)
< 1.15 * min(first.dominant_font_weight, second.dominant_font_weight)
)
return uncertain_dominant_font and weights_compatible
def _document_title_uses_page_fallback(lanes: list[_TextLane]) -> bool:
"""判断首页标题是否依赖跨栏 1.30 倍全文正文行高兜底。"""
@@ -388,6 +462,12 @@ def _classify_paragraph_titles_in_lane(
and line.font_coverage >= 0.75
and line.font_signature != profile.body_font
)
low_coverage_style_differs = (
profile.body_font is not None
and line.font_signature is not None
and 0.5 <= line.font_coverage < 0.75
and line.font_signature != profile.body_font
)
style_support = profile.style_support.get(line.font_signature, 1.0) if line.font_signature is not None else 1.0
weight_emphasized = (
profile.body_weight is not None
@@ -427,6 +507,27 @@ def _classify_paragraph_titles_in_lane(
and not _title_fonts_compatible(rows[index - 1][0], line)
):
continue
if _continues_local_body_row(
rows,
index,
lane_width,
profile,
):
continue
compact_local_transition = (
low_coverage_style_differs
and left_aligned
and width_ratio <= 0.65
and gap_above >= 0.1
and gap_below >= 0.2
and _has_following_body_row(
rows,
index,
lane_width,
lane.left,
profile,
)
)
compact_text_section = (
height_ratio < 0.9
and centered
@@ -457,9 +558,9 @@ def _classify_paragraph_titles_in_lane(
):
continue
score = 0.0
if style_differs:
if style_differs or compact_local_transition:
score += 2.0
if style_support <= 0.2:
if compact_local_transition or style_support <= 0.2:
score += 0.75
if weight_emphasized:
score += 1.0
@@ -483,10 +584,15 @@ def _classify_paragraph_titles_in_lane(
strong_layout_signal = (
height_ratio >= 1.18
or style_differs
or compact_local_transition
or weight_emphasized
or (centered and width_ratio <= 0.7 and gap_above >= 0.35 and gap_below >= 0.2)
)
if score >= 4.0 and has_spacing_signal and strong_layout_signal:
if (
score >= 4.0
and (has_spacing_signal or compact_local_transition)
and strong_layout_signal
):
if _is_full_width_inline_heading(
rows,
index,
@@ -589,6 +695,40 @@ def _is_near_full_mixed_inline_row(
)
def _continues_local_body_row(
rows: list[tuple[_LineItem, BBox]],
index: int,
lane_width: float,
profile: _LaneBodyProfile,
) -> bool:
"""识别紧随同字体满行的正文尾行，阻止全局小字号基线造成标题误判。"""
if index <= 0:
return False
previous_line, previous_bbox = rows[index - 1]
line, bbox = rows[index]
if previous_line.semantic_type is not None:
return False
if (
previous_line.font_signature is not None
and line.font_signature is not None
and previous_line.font_signature != line.font_signature
):
return False
previous_height = _line_effective_height(previous_line, previous_bbox)
line_height = _line_effective_height(line, bbox)
pair_height = max(previous_height, line_height)
gap = _effective_text_row_gap(rows[index - 1], rows[index])
return (
0.8 <= line_height / previous_height <= 1.25
and previous_bbox[2] - previous_bbox[0] >= 0.75 * lane_width
and abs(bbox[0] - previous_bbox[0]) <= 2.0 * pair_height
and -0.25 * pair_height
<= gap
<= profile.regular_gap + pair_height
)
def _is_full_width_inline_heading(
rows: list[tuple[_LineItem, BBox]],
index: int,
@@ -645,7 +785,13 @@ def _has_following_body_row(
current_centered = abs(
_bbox_center_x(current[1]) - (lane_left + lane_width / 2.0)
) <= 0.12 * lane_width
maximum_gap = (1.5 if current_centered else 0.75) * profile.body_height
compact_left_aligned = (
current[1][2] - current[1][0] <= 0.35 * lane_width
and abs(current[1][0] - lane_left) <= 0.65 * profile.body_height
)
maximum_gap = (
1.5 if current_centered else 1.25 if compact_left_aligned else 0.75
) * profile.body_height
height_ratio = _line_effective_height(line, bbox) / profile.body_height
width_ratio = (bbox[2] - bbox[0]) / lane_width
return (
@@ -798,6 +944,9 @@ def _line_near_visual_container(
for container_bbox in container_bboxes:
if _bbox_axis_overlap_ratio(line_bbox, container_bbox, axis="x") < 0.35:
continue
container_width = max(0.1, container_bbox[2] - container_bbox[0])
if line_bbox[2] - line_bbox[0] > 0.8 * container_width:
continue
vertical_gap = max(line_bbox[1] - container_bbox[3], container_bbox[1] - line_bbox[3], 0.0)
if vertical_gap <= 1.5 * body_height:
return True
@@ -831,6 +980,12 @@ def _expand_paragraph_title_neighbors(
continue
if candidate_bbox[2] - candidate_bbox[0] > 0.8 * lane_width:
continue
if (
selected_line.font_signature is not None
and candidate_line.font_signature is not None
and selected_line.font_signature != candidate_line.font_signature
):
continue
if not _title_fonts_compatible(selected_line, candidate_line):
continue
vertical_gap = max(
mineru/backend/flash/pdf_extractor.py
+1 -1
View File
@@ -82,7 +82,7 @@ if __name__ == '__main__':
if __name__ == "__main__":
# pdf_path = "/Users/myhloli/pdf/截断合并/demo1-3.pdf"
# pdf_path = "/Users/myhloli/pdf/png/seal4.png" # shubiao.png
pdf_path = "/Users/myhloli/pdf/caibao1.pdf"
pdf_path = "/Users/myhloli/pdf/基于Boosting、Bagging、Stacking与Voting四大集成学习模型的粉质砂土冻土抗剪强度预测研究.pdf"
pdf_bytes = read_fn(pdf_path)
model_list = doc_analyze(pdf_bytes, parse_mode="auto")
logger.info(f"model_list: {model_list}")
tests/unittest/test_flash_pdf_extractor.py
+50
View File
@@ -149,3 +149,53 @@ def test_page_index_map_mismatch_is_rejected_before_hybrid_call(monkeypatch: pyt
)
hybrid_doc_analyze.assert_not_called()
def test_marginal_header_row_does_not_break_two_column_reading_order() -> None:
"""验证页眉页码独立按视觉行排序，正文仍保持左栏后右栏的阅读顺序。"""
blocks = [
{"type": "header", "bbox": (80.0, 2.0, 120.0, 12.0), "angle": 0, "content": "center"},
{"type": "page_number", "bbox": (2.0, 0.0, 12.0, 14.0), "angle": 0, "content": "page"},
{"type": "header", "bbox": (175.0, 1.0, 198.0, 13.0), "angle": 0, "content": "volume"},
{"type": "text", "bbox": (0.0, 20.0, 80.0, 40.0), "angle": 0, "content": "left one"},
{"type": "text", "bbox": (0.0, 45.0, 80.0, 65.0), "angle": 0, "content": "left two"},
{"type": "text", "bbox": (120.0, 20.0, 200.0, 40.0), "angle": 0, "content": "right one"},
{"type": "text", "bbox": (120.0, 45.0, 200.0, 65.0), "angle": 0, "content": "right two"},
]
sorted_blocks = pipeline._sort_blocks_with_visual_row_groups(
blocks,
(200.0, 100.0),
)
assert [block["content"] for block in sorted_blocks] == [
"page",
"center",
"volume",
"left one",
"left two",
"right one",
"right two",
]
def test_overlapping_span_captions_follow_visual_center_order() -> None:
"""验证同一跨栏带中轻微重叠的图注按视觉中心由上到下稳定排序。"""
common = {
"type": "text",
"angle": 0,
"_lane_interval": (20.0, 180.0),
"_lane_is_span": True,
"_line_heights": [10.0],
}
later = {**common, "bbox": (30.0, 58.0, 170.0, 70.0), "content": "later"}
earlier = {**common, "bbox": (70.0, 50.0, 130.0, 62.0), "content": "earlier"}
stabilized = pipeline._stabilize_overlapping_lane_order(
[later, earlier],
(200.0, 100.0),
)
assert [block["content"] for block in stabilized] == ["earlier", "later"]
tests/unittest/test_flash_pdf_formulas.py
+111
View File
@@ -478,6 +478,117 @@ def test_detached_formula_anchor_collects_multiline_formula_but_not_body_prefix(
assert punctuation.text not in member_texts
def test_formula_above_dense_body_collects_math_but_stops_at_title_barrier() -> None:
"""验证正文密集区上方的公式可聚合，且不会吸收紧邻的章节标题。"""
body_font = ("Body", 0)
formula_lines = [
_text_line(
"formula numerator",
(20.0, 8.0, 58.0, 18.0),
0,
effective_height=10.0,
font_signature=("Math", 0),
font_coverage=0.6,
),
_text_line(
"formula denominator",
(25.0, 19.0, 55.0, 29.0),
1,
effective_height=10.0,
font_signature=("Math", 0),
font_coverage=0.6,
),
]
number = _text_line("(5)", (91.0, 17.0, 100.0, 27.0), 2, effective_height=10.0)
heading = _text_line(
"neutral section heading",
(0.0, 28.0, 40.0, 42.0),
3,
effective_height=14.0,
font_signature=("Heading", 0),
font_coverage=1.0,
)
body_lines = [
_text_line(
f"body-{index}",
(0.0, top, 100.0, top + 10.0),
4 + index,
effective_height=10.0,
font_signature=body_font,
font_coverage=1.0,
)
for index, top in enumerate((44.0, 56.0, 68.0, 80.0))
]
lane = models._TextLane(
left=0.0,
right=100.0,
lines=[
*((line, line.bbox) for line in formula_lines),
(number, number.bbox),
(heading, heading.bbox),
*((line, line.bbox) for line in body_lines),
],
)
anchors = formulas._find_formula_spatial_anchors(lane, 10.0, body_font)
assert len(anchors) == 1
assert anchors[0].detached_above_body
members = formulas._grow_formula_spatial_component(
lane,
anchors[0],
0.0,
100.0,
set(),
[],
body_font,
10.0,
)
assert {line.text for line, _bbox in members} == {
"formula numerator",
"formula denominator",
"(5)",
}
def test_formula_number_cannot_upgrade_ordinary_body_row() -> None:
"""验证括号编号缺少独立公式主体时不能把普通正文升级为公式。"""
body_font = ("Body", 0)
lines = [
_text_line(
f"body-{index}",
(0.0, top, 100.0, top + 10.0),
index,
effective_height=10.0,
font_signature=body_font,
font_coverage=1.0,
)
for index, top in enumerate((0.0, 12.0, 48.0, 60.0))
]
ordinary = _text_line(
"ordinary body row",
(0.0, 30.0, 70.0, 40.0),
4,
effective_height=10.0,
font_signature=body_font,
font_coverage=1.0,
)
number = _text_line("(9)", (91.0, 30.0, 100.0, 40.0), 5, effective_height=10.0)
blocks, remaining = formulas._build_formula_like_blocks(
[*lines, ordinary, number],
[],
(100.0, 80.0),
)
assert blocks == []
assert ordinary in remaining
assert number in remaining
def test_overlapping_denominator_cannot_become_short_formula_anchor() -> None:
"""验证与正文横向重叠的右缘分母字符不会成为非编号公式锚点。"""
tests/unittest/test_flash_pdf_tables.py
+127 -1
View File
@@ -525,6 +525,50 @@ def test_chart_tick_rows_fail_dense_multi_cell_distribution() -> None:
assert candidates == []
def test_long_sparse_rule_interval_cannot_bridge_two_low_column_tables() -> None:
"""验证低列数表格之间仅有一行稀疏文本时不能跨长区间合并。"""
rows: list[models._VisualRow] = []
source_index = 0
for row_index, (top, anchors) in enumerate(
(
(10.0, (10.0, 50.0, 90.0)),
(20.0, (10.0, 50.0, 90.0)),
(30.0, (10.0, 50.0, 90.0)),
(80.0, (10.0, 90.0)),
(130.0, (10.0, 50.0, 90.0)),
(140.0, (10.0, 50.0, 90.0)),
)
):
fragments = []
for anchor in anchors:
bbox = (anchor, top, anchor + 5.0, top + 5.0)
fragments.append(
models._Fragment(
text=f"cell-{source_index}",
bbox=bbox,
local_bbox=bbox,
line_index=source_index,
visual_row_id=row_index,
)
)
source_index += 1
rows.append(
models._VisualRow(
fragments=fragments,
center_y=top + 2.5,
bbox=geometry._bbox_union_many([fragment.bbox for fragment in fragments]),
visual_row_id=row_index,
)
)
rules = [
_axis_line("horizontal", (0.0, top, 110.0, top + 0.1))
for top in (0.0, 40.0, 120.0, 150.0)
]
assert not tables._rule_intervals_are_column_compatible(rows, rules, 5.0)
def test_split_table_footnote_marker_is_joined_before_matching() -> None:
"""验证旋转表拆开的 For 与星号脚注在视觉行拼接后可被识别。"""
@@ -538,7 +582,89 @@ def test_split_table_footnote_marker_is_joined_before_matching() -> None:
)
assert tables._is_table_note_text(tables._visual_row_text(row))
assert tables._is_table_note_text("1 Numeric table footnote")
assert not tables._is_table_note_text("1 Numeric table footnote")
def test_table_note_chain_stops_at_font_and_size_transition() -> None:
"""验证表注续行不能跨越字体字号突变的标题或后续正文。"""
specs = [
("Note: neutral marker", (0.0, 52.0, 60.0, 60.0), ("Note", 0), 8.0),
("neutral continuation", (0.0, 61.0, 65.0, 69.0), ("Note", 0), 8.0),
("section barrier", (0.0, 70.0, 45.0, 84.0), ("Heading", 0), 14.0),
("ordinary body", (10.0, 85.0, 100.0, 95.0), ("Body", 0), 10.0),
]
lines: list[models._LineItem] = []
rows: list[models._VisualRow] = []
for source_index, (text, bbox, font, height) in enumerate(specs):
lines.append(
models._LineItem(
text=text,
bbox=bbox,
angle=0,
source_index=source_index,
effective_height=height,
font_signature=font,
font_coverage=1.0,
)
)
fragment = models._Fragment(
text=text,
bbox=bbox,
local_bbox=bbox,
line_index=source_index,
visual_row_id=source_index,
)
rows.append(
models._VisualRow(
fragments=[fragment],
center_y=geometry._bbox_center_y(bbox),
bbox=bbox,
visual_row_id=source_index,
)
)
selected = tables._collect_footnote_rows(
rows,
lines,
(0.0, 0.0, 100.0, 50.0),
10.0,
set(),
)
assert [tables._visual_row_text(row) for row in selected] == [
"Note: neutral marker",
"neutral continuation",
]
def test_numeric_body_row_cannot_start_table_note_chain() -> None:
"""验证数字开头正文即使紧邻表格下边界也不能独立启动表注扩张。"""
bbox = (0.0, 52.0, 80.0, 62.0)
line = models._LineItem(
text="5 ordinary numbered body",
bbox=bbox,
angle=0,
source_index=0,
effective_height=10.0,
font_signature=("Body", 0),
font_coverage=1.0,
)
row = models._VisualRow(
fragments=[models._Fragment(line.text, bbox, bbox, 0, 0)],
center_y=geometry._bbox_center_y(bbox),
bbox=bbox,
visual_row_id=0,
)
assert tables._collect_footnote_rows(
[row],
[line],
(0.0, 0.0, 100.0, 50.0),
10.0,
set(),
) == []
@pytest.mark.parametrize("projection_mode", ["empty", "error"])
tests/unittest/test_flash_pdf_text_blocks.py
+332
View File
@@ -737,6 +737,60 @@ def test_hyphen_continuation_cannot_start_false_hanging_indent_entry() -> None:
]
def test_local_two_column_band_survives_full_width_body_below() -> None:
"""验证页面顶部局部双栏不会被下方通栏正文覆盖。"""
lines: list[models._LineItem] = []
for row_index, top in enumerate((0.0, 12.0, 24.0, 36.0)):
lines.extend(
[
_text_line(f"left-{row_index}", (10.0, top, 90.0, top + 10.0), row_index * 2),
_text_line(f"right-{row_index}", (110.0, top, 190.0, top + 10.0), row_index * 2 + 1),
]
)
lines.extend(
_text_line(f"wide-{row_index}", (10.0, top, 190.0, top + 10.0), 20 + row_index)
for row_index, top in enumerate((70.0, 82.0, 94.0, 106.0))
)
lanes = line_layout._infer_text_lanes(
[(line, line.bbox) for line in lines],
200.0,
10.0,
)
lane_texts = [{line.text for line, _bbox in lane.lines} for lane in lanes]
assert {f"left-{index}" for index in range(4)} in lane_texts
assert {f"right-{index}" for index in range(4)} in lane_texts
assert {f"wide-{index}" for index in range(4)} in lane_texts
def test_short_span_tail_reattaches_before_later_cross_layout_region() -> None:
"""验证局部通栏段的短尾行不会因页面后续另有通栏行而留在单栏。"""
first = _text_line("wide one", (10.0, 0.0, 190.0, 10.0), 0)
second = _text_line("wide two", (10.0, 10.0, 190.0, 20.0), 1)
tail = _text_line("tail", (10.0, 20.0, 50.0, 30.0), 2)
later = _text_line("later wide", (10.0, 80.0, 190.0, 90.0), 3)
parallel = _text_line("parallel", (110.0, 50.0, 190.0, 60.0), 4)
left_lane = models._TextLane(left=10.0, right=90.0, lines=[(tail, tail.bbox)])
right_lane = models._TextLane(left=110.0, right=190.0, lines=[(parallel, parallel.bbox)])
span_lane = models._TextLane(
left=10.0,
right=190.0,
lines=[(first, first.bbox), (second, second.bbox), (later, later.bbox)],
is_span=True,
)
line_layout._reattach_span_lane_continuations(
[left_lane, right_lane, span_lane],
10.0,
)
assert tail not in [line for line, _bbox in left_lane.lines]
assert tail in [line for line, _bbox in span_lane.lines]
def test_spatial_post_merge_connects_short_opener_wide_body_and_tail() -> None:
"""验证跨栏拆开的短首行、满宽正文和紧邻尾行仅按空间关系重新连接。"""
@@ -917,6 +971,284 @@ def test_spatial_post_merge_does_not_share_incompatible_lane_width() -> None:
]
def test_spatial_post_merge_does_not_join_span_caption_to_column_body() -> None:
"""验证跨栏图注即使紧邻单栏正文也不会在二次阶段合并。"""
blocks = [
{
"type": "text",
"bbox": (40.0, 10.0, 560.0, 30.0),
"angle": 0,
"content": "wide caption",
"_visual_row_ids": {0, 1},
"_local_line_bboxes": [
(200.0, 10.0, 360.0, 20.0),
(40.0, 20.0, 560.0, 30.0),
],
"_line_heights": [10.0, 10.0],
"_lane_interval": (40.0, 560.0),
"_lane_is_span": True,
},
{
"type": "text",
"bbox": (40.0, 35.0, 290.0, 65.0),
"angle": 0,
"content": "left column body",
"_visual_row_ids": {2, 3, 4},
"_local_line_bboxes": [
(40.0, 35.0, 290.0, 45.0),
(40.0, 45.0, 290.0, 55.0),
(40.0, 55.0, 290.0, 65.0),
],
"_line_heights": [10.0, 10.0, 10.0],
"_lane_interval": (40.0, 290.0),
"_lane_is_span": False,
},
]
merged = text_blocks._merge_spatial_text_components(blocks, (600.0, 100.0))
assert [block["content"] for block in merged] == [
"wide caption",
"left column body",
]
def test_image_caption_marker_only_attaches_same_font_spatial_tail() -> None:
"""验证通用图注标记仅确认图像邻近候选，并把同字体续行接回对应图注。"""
common = {
"type": "text",
"angle": 0,
"_single_run_row_id": None,
"_lane_interval": (10.0, 90.0),
"_lane_is_span": True,
}
blocks = [
{
**common,
"bbox": (30.0, 50.0, 70.0, 60.0),
"content": "图1 neutral caption",
"_visual_row_ids": {1},
"_local_line_bboxes": [(30.0, 50.0, 70.0, 60.0)],
"_line_heights": [10.0],
"_font_signatures": {("Chinese", 0)},
},
{
**common,
"bbox": (10.0, 58.0, 90.0, 68.0),
"content": "Fig. 1 neutral caption;",
"_visual_row_ids": {2},
"_local_line_bboxes": [(10.0, 58.0, 90.0, 68.0)],
"_line_heights": [10.0],
"_font_signatures": {("English", 0)},
},
{
**common,
"bbox": (20.0, 67.0, 80.0, 77.0),
"content": "caption continuation",
"_visual_row_ids": {3},
"_local_line_bboxes": [(20.0, 67.0, 80.0, 77.0)],
"_line_heights": [10.0],
"_font_signatures": {("English", 0)},
},
]
merged = text_blocks._merge_image_caption_text_blocks(
blocks,
[(20.0, 10.0, 80.0, 50.0)],
)
assert len(merged) == 2
chinese = next(block for block in merged if block["content"].startswith("图1"))
english = next(block for block in merged if block["content"].startswith("Fig. 1"))
assert "continuation" not in chinese["content"]
assert "caption continuation" in english["content"]
assert english["bbox"] == (10.0, 58.0, 90.0, 77.0)
def test_fragmented_center_header_merges_but_remote_volume_stays_separate() -> None:
"""验证等距窄页眉字形形成一个逻辑块，远端卷期块不被吸收。"""
blocks = []
for index, left in enumerate((40.0, 60.0, 80.0, 100.0)):
blocks.append(
{
"type": "header",
"bbox": (left, 5.0, left + 6.0, 15.0),
"angle": 0,
"content": f"h{index}",
"_visual_row_ids": {0},
"_single_run_row_id": 0,
"_local_line_bboxes": [(left, 5.0, left + 6.0, 15.0)],
"_line_heights": [10.0],
"_font_signatures": {("Header", 0)},
"_lane_interval": (0.0, 200.0),
"_lane_is_span": True,
}
)
blocks.append(
{
"type": "header",
"bbox": (170.0, 5.0, 190.0, 15.0),
"angle": 0,
"content": "volume",
"_visual_row_ids": {0},
"_single_run_row_id": 0,
"_local_line_bboxes": [(170.0, 5.0, 190.0, 15.0)],
"_line_heights": [10.0],
"_font_signatures": {("Header", 0)},
"_lane_interval": (0.0, 200.0),
"_lane_is_span": True,
}
)
merged = text_blocks._merge_fragmented_header_blocks(blocks)
assert len(merged) == 2
center = next(block for block in merged if block["content"].startswith("h0"))
assert center["content"] == "h0 h1 h2 h3"
assert center["bbox"] == (40.0, 5.0, 106.0, 15.0)
assert next(block for block in merged if block["content"] == "volume")
def test_spatial_post_merge_cannot_skip_intervening_title_block() -> None:
"""验证二次组件合并不能越过同一水平流中的中间标题块。"""
common = {
"angle": 0,
"_single_run_row_id": None,
"_lane_interval": (0.0, 100.0),
"_lane_is_span": False,
"_font_signatures": {("Body", 0)},
}
blocks = [
{
**common,
"type": "text",
"bbox": (0.0, 0.0, 25.0, 10.0),
"content": "short opener",
"_visual_row_ids": {0},
"_local_line_bboxes": [(0.0, 0.0, 25.0, 10.0)],
"_line_heights": [10.0],
},
{
**common,
"type": "paragraph_title",
"bbox": (0.0, 9.0, 45.0, 19.0),
"content": "intervening title",
"_visual_row_ids": {1},
"_local_line_bboxes": [(0.0, 9.0, 45.0, 19.0)],
"_line_heights": [10.0],
},
{
**common,
"type": "text",
"bbox": (0.0, 12.0, 100.0, 22.0),
"content": "wide body",
"_visual_row_ids": {2},
"_local_line_bboxes": [(0.0, 12.0, 100.0, 22.0)],
"_line_heights": [10.0],
},
]
merged = text_blocks._merge_spatial_text_components(blocks, (100.0, 50.0))
assert [block["content"] for block in merged] == [
"short opener",
"intervening title",
"wide body",
]
def test_exaggerated_bbox_short_tail_stays_with_full_width_previous_row() -> None:
"""验证同左边界短尾行不会因异常字体框高度而从正文段落脱落。"""
previous = _text_line(
"full width previous row",
(0.0, 0.0, 100.0, 10.0),
0,
effective_height=10.0,
font_signature=("Body", 0),
font_coverage=1.0,
)
current = _text_line(
"short tail",
(0.0, 12.0, 25.0, 30.0),
1,
effective_height=18.0,
font_signature=("Number", 0),
font_coverage=1.0,
)
lane = models._TextLane(
left=0.0,
right=100.0,
lines=[(previous, previous.bbox), (current, current.bbox)],
)
assert line_layout._should_connect_text_rows(
(previous, previous.bbox),
(current, current.bbox),
lane,
regular_gap=2.0,
gap_mad=0.5,
table_bboxes=[],
axis_lines=[],
)
def test_indented_list_row_can_return_to_lane_left_for_short_tail() -> None:
"""验证列表首行轻度缩进时，回到栏左边的同字体短尾行仍可续接。"""
body_font = ("Body", 0)
previous = _text_line(
"indented list row",
(15.0, 0.0, 100.0, 10.0),
0,
font_signature=body_font,
font_coverage=1.0,
)
current = _text_line(
"short tail",
(0.0, 10.0, 65.0, 20.0),
1,
font_signature=body_font,
font_coverage=1.0,
)
lane = models._TextLane(
left=0.0,
right=100.0,
lines=[(previous, previous.bbox), (current, current.bbox)],
)
assert line_layout._should_connect_text_rows(
(previous, previous.bbox),
(current, current.bbox),
lane,
regular_gap=0.0,
gap_mad=0.0,
table_bboxes=[],
axis_lines=[],
)
def test_outdented_reference_number_starts_new_structural_entry() -> None:
"""验证通用参考文献编号仅在左突几何成立时切开上一条续行。"""
previous = _text_line("previous continuation", (25.0, 0.0, 100.0, 10.0), 0)
current = _text_line("［23］ next reference", (0.0, 10.0, 100.0, 20.0), 1)
aligned = _text_line("［23］ inline marker", (25.0, 10.0, 100.0, 20.0), 2)
assert text_blocks._starts_structural_reference_entry(
(previous, previous.bbox),
(current, current.bbox),
)
assert not text_blocks._starts_structural_reference_entry(
(previous, previous.bbox),
(aligned, aligned.bbox),
)
def test_spatial_post_merge_limits_tapered_tail_to_parallel_information_grid() -> None:
"""验证略大间距的递减尾行只在并列信息网格中并回其左对齐主体。"""
tests/unittest/test_flash_pdf_titles.py
+149
View File
@@ -274,6 +274,152 @@ def test_multiline_document_title_does_not_absorb_author_line() -> None:
assert [line.semantic_type for line in lines[:3]] == ["doc_title", "doc_title", None]
def test_multiline_document_title_accepts_uncertain_mixed_dominant_font() -> None:
"""验证混排标题主字体覆盖不稳定时仍可按字号、居中和字重合并。"""
lines = [
_text_line(
"wide mixed title",
(5.0, 20.0, 95.0, 35.0),
0,
effective_height=15.0,
font_signature=("MixedDominant", 0),
font_coverage=0.8,
dominant_font_weight=500.0,
),
_text_line(
"wide title continuation",
(10.0, 36.0, 90.0, 50.0),
1,
effective_height=14.0,
font_signature=("LocalTitle", 0),
font_coverage=1.0,
dominant_font_weight=500.0,
),
_text_line(
"author",
(35.0, 60.0, 65.0, 69.0),
2,
effective_height=9.0,
font_signature=("LocalTitle", 0),
font_coverage=1.0,
dominant_font_weight=500.0,
),
_text_line("body one", (0.0, 110.0, 100.0, 120.0), 3),
_text_line("body two", (0.0, 122.0, 100.0, 132.0), 4),
_text_line("body three", (0.0, 134.0, 100.0, 144.0), 5),
]
titles._classify_page_titles(
lines,
(100.0, 200.0),
page_index=0,
container_bboxes=[],
)
assert [line.semantic_type for line in lines[:3]] == ["doc_title", "doc_title", None]
blocks = text_blocks._build_text_blocks(lines, [], (100.0, 200.0))
title_blocks = [block for block in blocks if block["type"] == "doc_title"]
assert len(title_blocks) == 1
assert title_blocks[0]["content"] == "wide mixed title\nwide title continuation"
def test_compact_left_heading_accepts_one_body_height_following_gap() -> None:
"""验证短标题与后继正文相隔约一行时仍可由局部样式过渡确认。"""
body_font = ("Body", 0)
heading = _text_line(
"neutral heading",
(0.0, 45.0, 25.0, 56.0),
3,
effective_height=11.0,
font_signature=("Heading", 0),
font_coverage=0.65,
)
lines = [
_text_line(
"body one",
(0.0, 0.0, 100.0, 10.0),
0,
font_signature=body_font,
font_coverage=1.0,
),
_text_line(
"body two",
(0.0, 12.0, 100.0, 22.0),
1,
font_signature=body_font,
font_coverage=1.0,
),
_text_line(
"body three",
(0.0, 24.0, 100.0, 34.0),
2,
font_signature=body_font,
font_coverage=1.0,
),
heading,
_text_line(
"next body",
(0.0, 67.0, 100.0, 77.0),
4,
font_signature=body_font,
font_coverage=1.0,
),
_text_line(
"next body two",
(0.0, 79.0, 100.0, 89.0),
5,
font_signature=body_font,
font_coverage=1.0,
),
]
titles._classify_page_titles(
lines,
(100.0, 120.0),
page_index=1,
container_bboxes=[],
)
assert heading.semantic_type == "paragraph_title"
def test_same_font_body_tail_with_moderate_gap_is_not_title() -> None:
"""验证同字体满行之后的短正文尾行不会因中等间距被升级为标题。"""
body_font = ("Body", 0)
lines = [
_text_line("reference body", (0.0, 0.0, 100.0, 10.0), 0, effective_height=8.0),
_text_line("reference body two", (0.0, 10.0, 100.0, 20.0), 1, effective_height=8.0),
_text_line(
"local full body",
(0.0, 35.0, 100.0, 45.0),
2,
font_signature=body_font,
font_coverage=1.0,
),
_text_line(
"local body tail",
(0.0, 54.0, 60.0, 64.0),
3,
font_signature=body_font,
font_coverage=1.0,
),
_text_line("different section", (0.0, 80.0, 100.0, 90.0), 4),
_text_line("different continuation", (0.0, 92.0, 100.0, 102.0), 5),
]
titles._classify_page_titles(
lines,
(100.0, 120.0),
page_index=1,
container_bboxes=[],
)
assert lines[3].semantic_type is None
def test_cross_column_document_title_uses_thirteen_tenths_body_height_fallback() -> None:
"""验证首页跨栏居中标题达到正文 1.30 倍时可命中，作者行保持正文类型。"""
@@ -617,12 +763,15 @@ def test_paragraph_title_detector_does_not_read_line_text() -> None:
titles._classify_page_titles,
titles._build_physical_title_gap_map,
titles._find_repeated_grid_title_suppressions,
titles._find_container_visual_row_title_suppressions,
titles._infer_lane_body_profile,
titles._classify_document_title,
titles._document_title_fonts_compatible,
titles._document_title_uses_page_fallback,
titles._classify_paragraph_titles_in_lane,
titles._visual_row_has_body_style_sibling,
titles._is_near_full_mixed_inline_row,
titles._continues_local_body_row,
titles._is_full_width_inline_heading,
titles._has_following_body_row,
titles._has_following_compact_text_section,