mirror of
https://github.com/lllin000/PaperForge.git
synced 2026-07-22 06:50:53 +00:00
feat: add final OCR PDF layout consistency audit
This commit is contained in:
Research Assistant
parent
ef73b3f0a1
commit
fa418b218f
4 changed files with 293 additions and 25 deletions
|
|
@ -81,6 +81,7 @@ class DocumentStructure:
|
|||
tail_reading_order: list[dict] | None = None
|
||||
reference_zones: list[dict] | None = None
|
||||
span_coverage: dict | None = None
|
||||
layout_audit: dict | None = None
|
||||
|
||||
|
||||
def _compute_span_coverage(blocks: list[dict]) -> dict:
|
||||
|
|
@ -1782,11 +1783,20 @@ def _detect_non_body_insert_clusters(
|
|||
return str(span.get("font", "") or "").lower() or None
|
||||
return None
|
||||
|
||||
raw_candidates: dict[int, list[tuple[int, bool, bool]]] = {}
|
||||
# Read spine trust quality from _meta (not per-page quality, which is
|
||||
# contaminated on early pages). Strong spine: width/font baseline is
|
||||
# reliable enough to trust either signal in isolation. Moderate spine:
|
||||
# need both signals or cluster support. Weak spine: no anchor pages
|
||||
# existed, so width/font data are untrustworthy — require both signals.
|
||||
spine_meta = body_spine.get("_meta", {})
|
||||
quality = spine_meta.get("quality", "weak")
|
||||
|
||||
candidates_by_page: dict[int, list[int]] = {}
|
||||
for i, block in enumerate(blocks):
|
||||
page = block.get("page", 1)
|
||||
if page > max_early_page:
|
||||
continue
|
||||
|
||||
# body_paragraph, figure_caption, figure_caption_candidate, and
|
||||
# unknown_structural can be non-body inserts — bio/profile blocks that
|
||||
# OCR misclassified as body text or figure titles. frontmatter_noise
|
||||
|
|
@ -1795,6 +1805,12 @@ def _detect_non_body_insert_clusters(
|
|||
if block.get("role") not in _INSERT_CANDIDATE_ROLES:
|
||||
continue
|
||||
|
||||
# Never mark page-1 title-like blocks as non_body_insert
|
||||
if page == 1:
|
||||
text = (block.get("text") or "").strip()
|
||||
if len(text) > 20 and not any(m in text.lower() for m in ["\u2022", "-", "*"]):
|
||||
continue
|
||||
|
||||
bbox = block.get("bbox", [0, 0, 0, 0])
|
||||
|
||||
# Skip blocks without valid bbox (text-less spacers, rule lines)
|
||||
|
|
@ -1820,31 +1836,15 @@ def _detect_non_body_insert_clusters(
|
|||
spine_fonts = set(spine_fonts) if spine_fonts else set()
|
||||
font_mismatch = bool(block_font and spine_fonts and block_font not in spine_fonts)
|
||||
|
||||
if is_narrow or font_mismatch:
|
||||
raw_candidates.setdefault(page, []).append((i, is_narrow, font_mismatch))
|
||||
if quality == "strong":
|
||||
passes = is_narrow or font_mismatch
|
||||
elif quality == "moderate":
|
||||
passes = is_narrow and (font_mismatch or len(candidates_by_page.get(page, [])) >= 1)
|
||||
else:
|
||||
passes = is_narrow and font_mismatch
|
||||
|
||||
# Quality-gated filtering: font_mismatch signal reliability depends on
|
||||
# whether anchor pages exist (multi-page width estimate) and spine quality.
|
||||
candidates_by_page: dict[int, list[int]] = {}
|
||||
for page, cand_list in raw_candidates.items():
|
||||
spine_key = page if page in body_spine else 1
|
||||
spine = body_spine.get(spine_key, {})
|
||||
spine_fonts = spine.get("all_fonts") or spine.get("fonts", set())
|
||||
if not isinstance(spine_fonts, set):
|
||||
spine_fonts = set(spine_fonts) if spine_fonts else set()
|
||||
has_font_data = bool(spine_fonts)
|
||||
has_anchors = bool(spine.get("anchor_pages", []))
|
||||
spine_quality = spine.get("quality", "weak") if isinstance(spine, dict) else "weak"
|
||||
|
||||
for i, is_narrow, font_mismatch in cand_list:
|
||||
if not has_font_data or has_anchors or spine_quality == "strong":
|
||||
candidates_by_page.setdefault(page, []).append(i)
|
||||
elif spine_quality == "moderate":
|
||||
if is_narrow and (font_mismatch or len(cand_list) >= 2):
|
||||
candidates_by_page.setdefault(page, []).append(i)
|
||||
else: # weak, no anchors
|
||||
if is_narrow and font_mismatch:
|
||||
candidates_by_page.setdefault(page, []).append(i)
|
||||
if passes:
|
||||
candidates_by_page.setdefault(page, []).append(i)
|
||||
|
||||
for candidate_indices in candidates_by_page.values():
|
||||
if len(candidate_indices) >= 2:
|
||||
|
|
@ -2255,6 +2255,72 @@ def _detect_structured_insert_clusters(blocks: list[dict]) -> set[int]:
|
|||
return result
|
||||
|
||||
|
||||
def _run_layout_audit(blocks: list[dict]) -> dict:
|
||||
"""Check resolved structure for obvious geometric contradictions.
|
||||
|
||||
Returns:
|
||||
status: str "pass", "warn", or "fail"
|
||||
page_warnings: dict[int, list[str]] per-page warnings
|
||||
anomaly_count: int
|
||||
anomaly_pages: list[int]
|
||||
"""
|
||||
page_warnings: dict[int, list[str]] = {}
|
||||
|
||||
by_page: dict[int, list[dict]] = {}
|
||||
for b in blocks:
|
||||
p = b.get("page", 1)
|
||||
by_page.setdefault(p, []).append(b)
|
||||
|
||||
for page, page_blocks in by_page.items():
|
||||
warnings: list[str] = []
|
||||
|
||||
# Check 1: heading owns body above it (cross-column)
|
||||
headings = [b for b in page_blocks if b.get("role") in ("section_heading", "subsection_heading")]
|
||||
body_blocks = [b for b in page_blocks if b.get("role") == "body_paragraph"]
|
||||
for h in headings:
|
||||
hb = h.get("bbox") or [0, 0, 0, 0]
|
||||
hx = (hb[0] + hb[2]) / 2
|
||||
hy = hb[1]
|
||||
h_col = 0 if hx < 600 else 1
|
||||
for body in body_blocks:
|
||||
bb = body.get("bbox") or [0, 0, 0, 0]
|
||||
bx = (bb[0] + bb[2]) / 2
|
||||
by_ = bb[1]
|
||||
b_col = 0 if bx < 600 else 1
|
||||
if h_col != b_col and by_ < hy:
|
||||
warnings.append("heading owns body in different column above it")
|
||||
break
|
||||
|
||||
# Check 2: structured_insert overlaps body region
|
||||
inserts = [b for b in page_blocks if b.get("role") in ("non_body_insert", "structured_insert")]
|
||||
body_blocks_p2 = [b for b in page_blocks if b.get("role") == "body_paragraph"]
|
||||
for ins in inserts:
|
||||
ib = ins.get("bbox") or [0, 0, 0, 0]
|
||||
for body in body_blocks_p2:
|
||||
bb = body.get("bbox") or [0, 0, 0, 0]
|
||||
if ib[0] < bb[2] and ib[2] > bb[0] and ib[1] < bb[3] and ib[3] > bb[1]:
|
||||
warnings.append("structured_insert overlaps body region")
|
||||
break
|
||||
|
||||
if warnings:
|
||||
page_warnings[page] = warnings
|
||||
|
||||
total_warnings = sum(len(w) for w in page_warnings.values())
|
||||
if total_warnings == 0:
|
||||
status = "pass"
|
||||
elif total_warnings <= 3:
|
||||
status = "warn"
|
||||
else:
|
||||
status = "fail"
|
||||
|
||||
return {
|
||||
"status": status,
|
||||
"page_warnings": {str(p): w for p, w in page_warnings.items()},
|
||||
"anomaly_count": total_warnings,
|
||||
"anomaly_pages": sorted(page_warnings.keys()),
|
||||
}
|
||||
|
||||
|
||||
def normalize_document_structure(blocks: list[dict]) -> tuple[DocumentStructure, list[dict]]:
|
||||
"""Analyze document structure and normalize roles.
|
||||
|
||||
|
|
@ -2332,4 +2398,8 @@ def normalize_document_structure(blocks: list[dict]) -> tuple[DocumentStructure,
|
|||
if idx < len(blocks):
|
||||
blocks[idx]["role"] = "structured_insert"
|
||||
|
||||
# Run layout audit after all resolution is done
|
||||
layout_audit = _run_layout_audit(blocks)
|
||||
doc_structure.layout_audit = layout_audit
|
||||
|
||||
return doc_structure, blocks
|
||||
|
|
|
|||
|
|
@ -96,6 +96,14 @@ def build_spine_health(body_spine: dict) -> dict:
|
|||
}
|
||||
|
||||
|
||||
def build_layout_audit_health(layout_audit: dict) -> dict:
|
||||
return {
|
||||
"layout_audit_status": layout_audit.get("status", "unknown"),
|
||||
"layout_anomaly_pages": layout_audit.get("anomaly_pages", []),
|
||||
"layout_anomaly_count": layout_audit.get("anomaly_count", 0),
|
||||
}
|
||||
|
||||
|
||||
def write_ocr_health(health_root: Path, report: dict[str, Any]) -> None:
|
||||
from paperforge.core.io import write_json
|
||||
|
||||
|
|
|
|||
|
|
@ -2573,3 +2573,173 @@ def test_span_coverage_strong_when_most_blocks_have_metadata() -> None:
|
|||
assert result["blocks_with_span"] == 8
|
||||
assert result["blocks_without_span"] == 2
|
||||
assert result["degraded_mode_active"] is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Non-body insert: spine quality gating (Task 5)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_non_body_insert_weak_spine_requires_font_mismatch() -> None:
|
||||
"""Moderate spine (2 anchor pages) + narrow width + same font → NOT non_body_insert."""
|
||||
from paperforge.worker.ocr_document import _detect_body_spine, _detect_non_body_insert_clusters
|
||||
|
||||
blocks = []
|
||||
# Pages 2-3: 3 body paragraphs each → 2 anchor pages → moderate meta quality
|
||||
for pg in range(2, 4):
|
||||
for i in range(3):
|
||||
blocks.append({
|
||||
"role": "body_paragraph",
|
||||
"page": pg,
|
||||
"bbox": [80, 100 + i * 100, 590, 160 + i * 100],
|
||||
"span_metadata": [{"size": 10, "font": "Times"}],
|
||||
"page_width": 1200,
|
||||
"page_height": 1700,
|
||||
})
|
||||
# Page 1: 2 body paragraphs
|
||||
for i in range(2):
|
||||
blocks.append({
|
||||
"role": "body_paragraph",
|
||||
"page": 1,
|
||||
"bbox": [80, 100 + i * 100, 590, 160 + i * 100],
|
||||
"span_metadata": [{"size": 10, "font": "Times"}],
|
||||
"page_width": 1200,
|
||||
"page_height": 1700,
|
||||
})
|
||||
# Page 1: 2 narrow blocks with SAME font → should NOT be non_body_insert
|
||||
blocks.append({
|
||||
"role": "body_paragraph",
|
||||
"page": 1,
|
||||
"bbox": [50, 600, 300, 640],
|
||||
"span_metadata": [{"size": 10, "font": "Times"}],
|
||||
"page_width": 1200,
|
||||
"page_height": 1700,
|
||||
})
|
||||
blocks.append({
|
||||
"role": "body_paragraph",
|
||||
"page": 1,
|
||||
"bbox": [50, 680, 310, 720],
|
||||
"span_metadata": [{"size": 10, "font": "Times"}],
|
||||
"page_width": 1200,
|
||||
"page_height": 1700,
|
||||
})
|
||||
|
||||
spine = _detect_body_spine(blocks)
|
||||
result = _detect_non_body_insert_clusters(blocks, spine, page_width=1200)
|
||||
assert len(result) == 0, f"Expected 0 non_body_insert with moderate spine + same font, got {result}"
|
||||
|
||||
|
||||
def test_non_body_insert_strong_spine_or_gate_ok() -> None:
|
||||
"""Strong spine (5 anchor pages) + narrow width or font mismatch → OK to detect."""
|
||||
from paperforge.worker.ocr_document import _detect_body_spine, _detect_non_body_insert_clusters
|
||||
|
||||
blocks = []
|
||||
# Pages 2-6: 3 body paragraphs each → 5 anchor pages → strong meta quality
|
||||
for pg in range(2, 7):
|
||||
for i in range(3):
|
||||
blocks.append({
|
||||
"role": "body_paragraph",
|
||||
"page": pg,
|
||||
"bbox": [80, 100 + i * 100, 590, 160 + i * 100],
|
||||
"span_metadata": [{"size": 10, "font": "Times"}],
|
||||
"page_width": 1200,
|
||||
"page_height": 1700,
|
||||
})
|
||||
# Page 1: 2 narrow blocks with SAME font → should be detected (strong spine trusts narrow alone)
|
||||
blocks.append({
|
||||
"role": "body_paragraph",
|
||||
"page": 1,
|
||||
"bbox": [50, 600, 300, 640],
|
||||
"span_metadata": [{"size": 10, "font": "Times"}],
|
||||
"page_width": 1200,
|
||||
"page_height": 1700,
|
||||
})
|
||||
blocks.append({
|
||||
"role": "body_paragraph",
|
||||
"page": 1,
|
||||
"bbox": [50, 680, 310, 720],
|
||||
"span_metadata": [{"size": 10, "font": "Times"}],
|
||||
"page_width": 1200,
|
||||
"page_height": 1700,
|
||||
})
|
||||
|
||||
spine = _detect_body_spine(blocks)
|
||||
result = _detect_non_body_insert_clusters(blocks, spine, page_width=1200)
|
||||
assert len(result) >= 2, f"Expected at least 2 non_body_insert with strong spine, got {result}"
|
||||
|
||||
|
||||
def test_title_not_swept_into_non_body_insert() -> None:
|
||||
"""Page 1 title-like block (long text, no bullets) is NOT marked as non_body_insert."""
|
||||
from paperforge.worker.ocr_document import _detect_body_spine, _detect_non_body_insert_clusters
|
||||
|
||||
blocks = []
|
||||
# Pages 2-6: 3 body paragraphs each → strong meta quality
|
||||
for pg in range(2, 7):
|
||||
for i in range(3):
|
||||
blocks.append({
|
||||
"role": "body_paragraph",
|
||||
"page": pg,
|
||||
"bbox": [80, 100 + i * 100, 590, 160 + i * 100],
|
||||
"span_metadata": [{"size": 10, "font": "Times"}],
|
||||
"page_width": 1200,
|
||||
"page_height": 1700,
|
||||
})
|
||||
# Page 1: title-like block (long text, narrow width, no bullets)
|
||||
blocks.append({
|
||||
"role": "body_paragraph",
|
||||
"page": 1,
|
||||
"bbox": [50, 100, 400, 140], # width 350 < 357 → narrow
|
||||
"text": "Metabolic regulation of skeletal cell fate and function in development and disease",
|
||||
"span_metadata": [{"size": 14, "font": "Times"}],
|
||||
"page_width": 1200,
|
||||
"page_height": 1700,
|
||||
})
|
||||
# Page 1: companion narrow block (also narrow, same font)
|
||||
blocks.append({
|
||||
"role": "body_paragraph",
|
||||
"page": 1,
|
||||
"bbox": [50, 200, 400, 240], # width 350 < 357 → narrow
|
||||
"text": "Short narrow paragraph that is not a title",
|
||||
"span_metadata": [{"size": 10, "font": "Times"}],
|
||||
"page_width": 1200,
|
||||
"page_height": 1700,
|
||||
})
|
||||
|
||||
spine = _detect_body_spine(blocks)
|
||||
result = _detect_non_body_insert_clusters(blocks, spine, page_width=1200)
|
||||
# Title block must not be in result
|
||||
title_idx = 15 # indices 0-14 are body paragraphs
|
||||
assert title_idx not in result, f"Title block (idx {title_idx}) should not be non_body_insert, got {result}"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Layout audit tests (Task 8)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_layout_audit_heading_owns_wrong_body() -> None:
|
||||
"""Heading in left column, body above it in right column -> audit flags."""
|
||||
from paperforge.worker.ocr_document import _run_layout_audit
|
||||
|
||||
blocks = [
|
||||
{"page": 1, "role": "body_paragraph", "bbox": [700, 100, 1100, 140], "text": "Body above heading"},
|
||||
{"page": 1, "role": "section_heading", "bbox": [100, 200, 500, 230], "text": "Introduction"},
|
||||
]
|
||||
result = _run_layout_audit(blocks)
|
||||
assert result["status"] in ("warn", "fail"), f"Expected warn/fail, got {result['status']}"
|
||||
assert "1" in result["page_warnings"], f"Page 1 should have warnings: {result['page_warnings']}"
|
||||
assert result["anomaly_count"] >= 1
|
||||
|
||||
|
||||
def test_layout_audit_reference_zone_overlaps_body() -> None:
|
||||
"""structured_insert region overlapping body region -> audit flags."""
|
||||
from paperforge.worker.ocr_document import _run_layout_audit
|
||||
|
||||
blocks = [
|
||||
{"page": 1, "role": "body_paragraph", "bbox": [100, 100, 500, 300], "text": "Body text"},
|
||||
{"page": 1, "role": "structured_insert", "bbox": [200, 150, 400, 250], "text": "Insert overlapping body"},
|
||||
]
|
||||
result = _run_layout_audit(blocks)
|
||||
assert result["status"] in ("warn", "fail"), f"Expected warn/fail, got {result['status']}"
|
||||
assert "1" in result["page_warnings"], f"Page 1 should have warnings: {result['page_warnings']}"
|
||||
assert result["anomaly_count"] >= 1
|
||||
|
|
|
|||
|
|
@ -104,3 +104,23 @@ def test_build_span_coverage_health() -> None:
|
|||
empty = build_span_coverage_health([])
|
||||
assert empty["degraded_mode_active"] is True
|
||||
assert empty["coverage_quality"] == "weak"
|
||||
|
||||
|
||||
def test_layout_audit_health_surface() -> None:
|
||||
from paperforge.worker.ocr_health import build_layout_audit_health
|
||||
|
||||
audit = {
|
||||
"status": "warn",
|
||||
"page_warnings": {"3": ["heading owns body in different column above it"]},
|
||||
"anomaly_count": 1,
|
||||
"anomaly_pages": [3],
|
||||
}
|
||||
result = build_layout_audit_health(audit)
|
||||
assert result["layout_audit_status"] == "warn"
|
||||
assert result["layout_anomaly_pages"] == [3]
|
||||
assert result["layout_anomaly_count"] == 1
|
||||
|
||||
empty = build_layout_audit_health({})
|
||||
assert empty["layout_audit_status"] == "unknown"
|
||||
assert empty["layout_anomaly_pages"] == []
|
||||
assert empty["layout_anomaly_count"] == 0
|
||||
|
|
|
|||
Loading…
Reference in a new issue