diff --git a/paperforge/worker/ocr_document.py b/paperforge/worker/ocr_document.py index c95a03fc..425a66d4 100644 --- a/paperforge/worker/ocr_document.py +++ b/paperforge/worker/ocr_document.py @@ -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 diff --git a/paperforge/worker/ocr_health.py b/paperforge/worker/ocr_health.py index 77e272cd..ca02d891 100644 --- a/paperforge/worker/ocr_health.py +++ b/paperforge/worker/ocr_health.py @@ -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 diff --git a/tests/test_ocr_document.py b/tests/test_ocr_document.py index 722e9c87..6acd9950 100644 --- a/tests/test_ocr_document.py +++ b/tests/test_ocr_document.py @@ -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 diff --git a/tests/test_ocr_health.py b/tests/test_ocr_health.py index 2b1e86b1..a4e7298f 100644 --- a/tests/test_ocr_health.py +++ b/tests/test_ocr_health.py @@ -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