refactor: replace dead reorder import with direct column-major fallback

This commit is contained in:
Research Assistant 2026-06-07 01:00:45 +08:00
parent 0ac980ebbe
commit ccff2aef3b
2 changed files with 37 additions and 105 deletions

View file

@ -1419,37 +1419,40 @@ def caption_group_assignments(blocks: list[dict]) -> tuple[dict[int, list[dict]]
def _apply_layered_body_reorder(blocks: list[dict], page_width: int, page_height: int) -> list[dict]:
from paperforge.worker.ocr_orchestrator import reorder_blocks_layered as _rbl
"""Column-major sort for two-column pages.
result = _rbl(blocks, page_width=page_width, page_height=page_height)
if result is not blocks:
return result
Groups blocks by column (left/center first, right second), each
sorted by y-position. Single-column pages pass through unchanged.
"""
if not page_width or len(blocks) < 4:
return blocks
# Fallback: column-major sort when body spine module unavailable.
# Groups by column (left/center first, right second), sorted by y.
if page_width and len(blocks) >= 4:
mid = page_width / 2
mid = page_width / 2
def _col(b):
bb = b.get("block_bbox") or b.get("bbox")
if not bb or len(bb) < 4:
return 0
xc = (bb[0] + bb[2]) / 2
w = bb[2] - bb[0]
if (bb[0] < mid and bb[2] > mid) or w >= page_width * 0.55:
return 0
return 1 if xc >= mid else 0
def _col(b):
bb = b.get("block_bbox") or b.get("bbox")
if not bb or len(bb) < 4:
return 0
xc = (bb[0] + bb[2]) / 2
w = bb[2] - bb[0]
if (bb[0] < mid and bb[2] > mid) or w >= page_width * 0.55:
return 0
return 1 if xc >= mid else 0
left = sorted(
[b for b in blocks if _col(b) == 0],
key=lambda b: (b.get("block_bbox") or b.get("bbox") or [0, 0, 0, 0])[1],
)
right = sorted(
[b for b in blocks if _col(b) == 1],
key=lambda b: (b.get("block_bbox") or b.get("bbox") or [0, 0, 0, 0])[1],
)
return left + right
return blocks
has_left = any(_col(b) == 0 for b in blocks)
has_right = any(_col(b) == 1 for b in blocks)
if not (has_left and has_right):
return blocks
left = sorted(
[b for b in blocks if _col(b) == 0],
key=lambda b: (b.get("block_bbox") or b.get("bbox") or [0, 0, 0, 0])[1],
)
right = sorted(
[b for b in blocks if _col(b) == 1],
key=lambda b: (b.get("block_bbox") or b.get("bbox") or [0, 0, 0, 0])[1],
)
return left + right
def render_page_blocks(

View file

@ -29,82 +29,11 @@ def reorder_blocks_layered(
page_width: int = 0,
page_height: int = 0,
) -> list[dict]:
if not blocks:
return blocks
"""Layout-aware body reordering.
try:
from paperforge.worker.ocr_body_spine import BodySpineNode
from paperforge.worker.ocr_layout import detect_layout_zones, order_body_spine
from paperforge.worker.ocr_roles import assign_block_role
except ImportError:
return blocks
annotated: list[BlockAnnotated] = []
for b in blocks:
role = assign_block_role(b, blocks, page_width=page_width, page_height=page_height)
annotated.append(
BlockAnnotated(block=b, role=role.role, role_confidence=role.confidence)
)
body_annotated = [a for a in annotated if a.role in {"section_heading", "subsection_heading", "body_paragraph", "unknown_structural"}]
non_body = [a for a in annotated if a not in body_annotated]
if not body_annotated:
return blocks
spine_nodes: list[BodySpineNode] = []
for i, a in enumerate(body_annotated):
bbox = tuple(a.block.get("block_bbox", [0, 0, 0, 0]))
node_type = (
a.role
if a.role in {"section_heading", "subsection_heading"}
else "paragraph"
)
spine_nodes.append(
BodySpineNode(
node_id=f"body-{i}",
node_type=node_type,
text=a.block.get("block_content", ""),
bbox=bbox,
role_confidence=a.role_confidence,
block_ids=[i],
)
)
zones = detect_layout_zones(spine_nodes, page_width=page_width, page_height=page_height)
ordered_nodes = order_body_spine(spine_nodes, zones, mode="column_major")
node_id_to_idx: dict[str, int] = {}
for i, node in enumerate(spine_nodes):
node_id_to_idx[node.node_id] = i
body_result: list[dict] = []
for node in ordered_nodes:
idx = node_id_to_idx.get(node.node_id)
if idx is not None and idx < len(body_annotated):
body_result.append(body_annotated[idx].block)
body_iter = iter(body_result)
result: list[dict] = []
for annotated_block in annotated:
if annotated_block in body_annotated:
try:
result.append(next(body_iter))
except StopIteration:
result.append(annotated_block.block)
else:
result.append(annotated_block.block)
result.extend(body_iter)
first_body_y = min((_bbox_top(block) for block in body_result), default=None)
if first_body_y is not None:
leading_roles = {"figure_caption", "table_caption", "media_asset"}
leading_blocks = [
a.block
for a in annotated
if a.role in leading_roles and _bbox_bottom(a.block) <= first_body_y
]
if leading_blocks:
leading_ids = {id(block) for block in leading_blocks}
result = leading_blocks + [block for block in result if id(block) not in leading_ids]
return result
Note: Full body spine reordering is unavailable (ocr_body_spine
module was removed). This function is preserved for signature
compatibility; callers use the column-major fallback in
_apply_layered_body_reorder instead.
"""
return blocks