feat(ocr): ParserService in the layout job + Checkpoint 1 (ENG-85)

extralit-server/scripts/checkpoint1_dump.py

@@ -0,0 +1,166 @@
+"""Checkpoint 1: run `ParserService` on the gold PDFs and check table header rows against `gold.md`.
+
+Usage: uv run python scripts/checkpoint1_dump.py [--gold-dir DIR] [--out DIR]
+Exit code 1 when a gold table header row is missing from the markdown (adageba Table 2 excepted)
+or the scanned paper does not route through liteparse with OCR.
+"""
+
+from __future__ import annotations
+
+import argparse
+import itertools
+import json
+import re
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+
+from extralit_server.contexts.ocr.parsers.registry import default_service
+from extralit_server.contexts.ocr.parsers.service import OCR_PARSER
+
+PAPERS = ("robert1991-bullwho-scanned", "adageba2022-plosone-digital")
+EXCEPTIONS = {("adageba2022-plosone-digital", "Table 2")}
+SCANNED = "robert1991-bullwho-scanned"
+
+
+@dataclass
+class HeaderCheck:
+    """Whether one gold table's header row survived into the parsed markdown."""
+
+    table: str
+    header: list[str]
+    status: str  # ok | cells-elsewhere | missing | excepted
+    absent: tuple[str, ...] = ()
+
+
+HEADER_ROWS_SCANNED = 3
+
+
+def _norm(text: str) -> str:
+    """Letters and digits only, so spacing, hyphens and markdown padding do not matter."""
+    return re.sub(r"[^a-z0-9]", "", text.lower())
+
+
+def markdown_tables(markdown: str) -> list[list[str]]:
+    """Each markdown table as its non-separator rows."""
+    tables: list[list[str]] = []
+    current: list[str] = []
+    for line in [*markdown.splitlines(), ""]:
+        if line.lstrip().startswith("|"):
+            if not re.fullmatch(r"\|[\s:|-]+\|?", line.strip()):
+                current.append(line)
+        elif current:
+            tables.append(current)
+            current = []
+    return tables
+
+
+def gold_headers(gold_md: str) -> list[tuple[str, list[str]]]:
+    """Header row of each `### Table N` section: the pipe row followed by a `|---|` separator."""
+    headers: list[tuple[str, list[str]]] = []
+    for section in re.split(r"^### ", gold_md, flags=re.MULTILINE)[1:]:
+        title = re.match(r"(Table \d+)", section)
+        lines = section.splitlines()
+        for line, nxt in itertools.pairwise(lines):
+            if line.startswith("|") and re.fullmatch(r"\|[\s:|-]+\|?", nxt.strip()):
+                cells = [c.strip() for c in line.strip().strip("|").split("|")]
+                headers.append((title.group(1) if title else "?", [c for c in cells if c]))
+                break
+    return headers
+
+
+def check_headers(paper: str, gold_md: str, markdown: str) -> list[HeaderCheck]:
+    """A gold header is `ok` when all its cells sit in the first rows of one markdown table."""
+    heads = ["".join(_norm(row) for row in table[:HEADER_ROWS_SCANNED]) for table in markdown_tables(markdown)]
+    flat = _norm(markdown)
+    results = []
+    for table, header in gold_headers(gold_md):
+        cells = [_norm(c) for c in header]
+        if any(all(c in head for c in cells) for head in heads):
+            status = "ok"
+        elif all(c in flat for c in cells):
+            status = "cells-elsewhere"
+        else:
+            status = "missing"
+        if (paper, table) in EXCEPTIONS and status != "ok":
+            status = "excepted"
+        absent = tuple(c for c, n in zip(header, cells, strict=True) if n not in flat)
+        results.append(HeaderCheck(table, header, status, absent))
+    return results
+
+
+def dump_paper(paper: str, gold_dir: Path, out_dir: Path) -> tuple[list[str], list[HeaderCheck], bool]:
+    """Parse one gold paper, write its artifacts, and return its report lines, header checks and routing verdict."""
+    pdf_bytes = (gold_dir / paper / "source.pdf").read_bytes()
+    service = default_service()
+    plan = service.plan(pdf_bytes)
+    doc = service.parse(pdf_bytes, plan, name=paper, filename="source.pdf")
+    markdown = doc.export_to_markdown()
+
+    target = out_dir / paper
+    target.mkdir(parents=True, exist_ok=True)
+    (target / "markdown.md").write_text(markdown)
+    lines = ["ref\tlabel\tpage\tl\tt\tr\tb"]
+    for item, _ in doc.iterate_items(with_groups=False):
+        for prov in item.prov or [None]:
+            box = prov.bbox if prov else None
+            coords = f"{box.l:.1f}\t{box.t:.1f}\t{box.r:.1f}\t{box.b:.1f}" if box else "\t\t\t"
+            lines.append(f"{item.self_ref}\t{item.label.value}\t{prov.page_no if prov else ''}\t{coords}")
+    (target / "items.tsv").write_text("\n".join(lines) + "\n")
+    plan_json = {
+        "tasks": [{"pages": list(t.pages), "parser": t.parser, "ocr": t.ocr} for t in plan.tasks],
+        "parsers": plan.pages_by_parser,
+        "pages_needing_ocr": list(plan.pages_needing_ocr),
+        "ocr_pages": plan.ocr_pages,
+        "ocr_skipped_pages": plan.ocr_skipped_pages,
+    }
+    (target / "plan.json").write_text(json.dumps(plan_json, indent=2))
+
+    checks = check_headers(paper, (gold_dir / paper / "gold.md").read_text(), markdown)
+    routed = True
+    if paper == SCANNED:
+        ocr_covered = {p for t in plan.tasks if t.parser == OCR_PARSER and t.ocr for p in t.pages}
+        routed = bool(plan.pages_needing_ocr) and set(plan.pages_needing_ocr) <= ocr_covered
+
+    labels = {}
+    for item, _ in doc.iterate_items(with_groups=False):
+        labels[item.label.value] = labels.get(item.label.value, 0) + 1
+    summary = [
+        f"## {paper}",
+        f"- plan: `{json.dumps(plan_json['parsers'])}`; ocr_pages={plan.ocr_pages}; "
+        f"ocr_skipped_pages (text layer read, no OCR)={plan.ocr_skipped_pages}; "
+        f"pages_needing_ocr={list(plan.pages_needing_ocr)}",
+        f"- items by label: `{json.dumps(labels, sort_keys=True)}`",
+    ]
+    if paper == SCANNED:
+        summary.append(f"- every page needing OCR routed through liteparse+OCR: **{routed}**")
+    summary.append("- gold table header rows:")
+    summary += [
+        f"  - {c.table}: **{c.status}** — {' | '.join(c.header)}"
+        + (f" (absent cells: {list(c.absent)})" if c.absent else "")
+        for c in checks
+    ]
+    return summary, checks, routed
+
+
+def main() -> int:
+    """Dump every checkpoint paper and return non-zero if any gate fails."""
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--gold-dir", type=Path, default=Path("~/Projects/Extralit/data/gold").expanduser())
+    parser.add_argument("--out", type=Path, default=Path("/tmp/eng85-checkpoint1"))
+    args = parser.parse_args()
+    args.out.mkdir(parents=True, exist_ok=True)
+
+    report: list[str] = ["# Checkpoint 1 report", ""]
+    failed = False
+    for paper in PAPERS:
+        summary, checks, routed = dump_paper(paper, args.gold_dir, args.out)
+        report += [*summary, ""]
+        failed |= not routed or any(c.status in ("missing", "cells-elsewhere") for c in checks)
+    (args.out / "report.md").write_text("\n".join(report))
+    print("\n".join(report))
+    return 1 if failed else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

extralit-server/src/extralit_server/api/schemas/v1/document/metadata.py

@@ -79,6 +79,10 @@ class LayoutMetadata(BaseModel):
 
     Only pointers and counts — the layout itself lives in object storage, because
     `documents.metadata_` is returned in full by every `GET /documents` listing.
+
+    `parsers` is the executed plan (parser name -> pages). `ocr_pages` are pages where OCR actually
+    ran; `ocr_skipped_pages` are pages planned for OCR (triage flagged them) whose existing text
+    layer the parser read instead, so `pages_needing_ocr` = planned, `ocr_pages` = actual.
     """
 
     layout_url: str = Field(..., description="S3 object path of the canonical DoclingDocument JSON")
@@ -93,6 +97,14 @@ class LayoutMetadata(BaseModel):
     pages_needing_ocr: list[int] = Field(
         default_factory=list, description="1-indexed pages with no reliable text layer"
     )
+    parsers: dict[str, list[int]] = Field(
+        default_factory=dict, description="Executed plan: parser name -> 1-indexed pages it parsed"
+    )
+    ocr_pages: list[int] = Field(default_factory=list, description="1-indexed pages where OCR actually ran")
+    ocr_skipped_pages: list[int] = Field(
+        default_factory=list,
+        description="Pages planned for OCR that already had a text layer, which the parser read instead of running OCR",
+    )
 
 
 class DocumentProcessingMetadata(BaseModel):

extralit-server/src/extralit_server/contexts/ocr/docling_builder.py

@@ -6,6 +6,8 @@
 
 from __future__ import annotations
 
+import re
+from collections.abc import Sequence
 from dataclasses import dataclass
 from typing import Optional
 
@@ -27,6 +29,9 @@
 #: Fraction of a text block that must fall inside a table/picture for it to count as duplicated.
 CONTAINMENT_THRESHOLD = 0.6
 
+#: Share of the cell text under a block that the block's own words must account for to count as a duplicate.
+CELL_SHARE_THRESHOLD = 0.5
+
 #: Labels routed to `add_table` / `add_picture` rather than `add_text`.
 TABLE_LABELS = frozenset({DocItemLabel.TABLE, DocItemLabel.DOCUMENT_INDEX})
 PICTURE_LABELS = frozenset({DocItemLabel.PICTURE, DocItemLabel.CHART})
@@ -61,10 +66,12 @@ class PageContext:
 
     @property
     def height(self) -> float:
+        """Page height in points."""
         return self.size.height
 
     @property
     def width(self) -> float:
+        """Page width in points."""
         return self.size.width
 
 
@@ -121,41 +128,49 @@ def make_prov(ctx: PageContext, bbox: BoundingBox, text: Optional[str]) -> Prove
     )
 
 
-def is_contained(
-    doc: DoclingDocument,
-    bbox: BoundingBox,
-    page_no: int,
-    threshold: float = CONTAINMENT_THRESHOLD,
-) -> bool:
-    """Whether `bbox` is mostly swallowed by a table or picture already on this page."""
-    for item in [*doc.tables, *doc.pictures]:
-        for prov in item.prov:
-            if prov.page_no != page_no:
-                continue
-            if bbox.intersection_over_self(prov.bbox) >= threshold:
-                return True
-    return False
+def _tokens(text: str) -> list[str]:
+    return re.findall(r"\w+", text.lower())
+
+
+def _carries_text(block_words: list[str], container: LayoutBlock, box: BoundingBox, ctx: PageContext) -> bool:
+    """Whether the cells under `box` hold this block's words and little else.
+
+    A cell much longer than the block is a false table swallowing prose, so the block is kept.
+    Cells without a bbox count only if they share a word with the block.
+    """
+
+    block_set = set(block_words)
 
+    def touches(cell: TableCell) -> bool:
+        if cell.bbox is None:
+            return not block_set.isdisjoint(_tokens(cell.text))
+        return box.intersection_over_self(clamp_to_page(flip_to_top_left(cell.bbox, ctx.height), ctx)) > 0
 
-def _sort_key(doc: DoclingDocument, node: NodeItem) -> tuple[float, float, float]:
-    """Position a body child by the earliest provenance anywhere in its subtree."""
-    positions = []
-    stack = [node]
-    while stack:
-        current = stack.pop()
-        for prov in getattr(current, "prov", []) or []:
-            positions.append((float(prov.page_no), prov.bbox.t, prov.bbox.l))
-        stack.extend(child.resolve(doc) for child in current.children)
-    return min(positions) if positions else (float("inf"),) * 3
+    under = [cell for cell in container.cells or [] if touches(cell)]
+    cell_words = [w for cell in under for w in _tokens(cell.text)]
+    if not cell_words:
+        return False
+    known = set(cell_words)
+    covered = sum(w in known for w in block_words) / len(block_words)
+    return covered >= CONTAINMENT_THRESHOLD and len(block_words) >= CELL_SHARE_THRESHOLD * len(cell_words)
 
 
-def sort_body_by_position(doc: DoclingDocument) -> None:
-    """Restore geometric reading order across text, tables and pictures alike.
+def is_contained(block: LayoutBlock, bbox: BoundingBox, containers: Sequence[LayoutBlock], ctx: PageContext) -> bool:
+    """Whether `block` duplicates a table or picture on its page.
 
-    The dedup pass has to add tables and pictures first, which leaves them ahead of every
-    paragraph in the body; this puts the page back in the order a reader would see it.
+    The bbox must be mostly inside the container; for a table the block's words must also be
+    what the cells under it hold, so a false table over real prose or a row it missed drops nothing.
     """
-    doc.body.children.sort(key=lambda ref: _sort_key(doc, ref.resolve(doc)))
+    words = _tokens(block.text)
+    for container in containers:
+        container_box = clamp_to_page(flip_to_top_left(container.bbox, ctx.height), ctx)
+        if bbox.intersection_over_self(container_box) < CONTAINMENT_THRESHOLD:
+            continue
+        if container.label in PICTURE_LABELS or not words:
+            return True
+        if _carries_text(words, container, bbox, ctx):
+            return True
+    return False
 
 
 def _page_of(item: NodeItem) -> Optional[int]:
@@ -184,34 +199,24 @@ def link_captions(doc: DoclingDocument) -> None:
 def append_blocks(
     doc: DoclingDocument,
     ctx: PageContext,
-    blocks: list[LayoutBlock],
+    blocks: Sequence[LayoutBlock],
 ) -> list[NodeItem]:
-    """Add one page's blocks, then restore reading order.
-
-    Tables and pictures go in first so the text pass can drop anything they already contain —
-    without that ordering the same words land in the document twice, once as a table cell and
-    once as a stray paragraph.
-    """
+    """Add one page's blocks in the order given, dropping text a table or picture already carries."""
     from extralit_server.contexts.ocr.figures import add_picture_block
     from extralit_server.contexts.ocr.tables import add_table_block
     from extralit_server.contexts.ocr.text import add_text_block
 
     register_page(doc, ctx)
+    containers = [b for b in blocks if b.label in TABLE_LABELS or b.label in PICTURE_LABELS]
 
     added: list[NodeItem] = []
     for block in blocks:
         if block.label in TABLE_LABELS:
             added.append(add_table_block(doc, block, ctx))
         elif block.label in PICTURE_LABELS:
             added.append(add_picture_block(doc, block, ctx))
-
-    for block in blocks:
-        if block.label in TABLE_LABELS or block.label in PICTURE_LABELS:
-            continue
-        item = add_text_block(doc, block, ctx)
-        if item is not None:
+        elif (item := add_text_block(doc, block, ctx, containers=containers)) is not None:
             added.append(item)
 
-    sort_body_by_position(doc)
     link_captions(doc)
     return added

extralit-server/src/extralit_server/contexts/ocr/parsers/pdf_inspector.py

@@ -254,6 +254,21 @@ def _blocks_for_page(
     return blocks
 
 
+class PdfInspectorParser:
+    """`PageParser` over pdf-inspector; it reads the text layer, so `ocr` has no effect."""
+
+    def parse_pages(self, pdf_bytes: bytes, pages: Sequence[int], *, ocr: bool) -> dict[int, list[LayoutBlock]]:
+        """Parse the given 1-indexed pages into top-left-origin blocks, keyed by page number."""
+        if not pages:
+            return {}
+        sizes = page_sizes(pdf_bytes)
+        roles = _structure_roles(pdf_bytes)
+        by_page: dict[int, list[Any]] = defaultdict(list)
+        for item in pdf_inspector.extract_text_with_positions_bytes(pdf_bytes):
+            by_page[item.page].append(item)
+        return {p: _blocks_for_page(by_page.get(p, []), roles, p) for p in sorted(set(pages)) if p in sizes}
+
+
 def parse(
     pdf_bytes: bytes,
     *,

extralit-server/src/extralit_server/contexts/ocr/parsers/pymupdf.py

@@ -13,7 +13,7 @@
 import pymupdf
 from docling_core.types.doc import BoundingBox, CoordOrigin, DocItemLabel, DoclingDocument, Size, TableCell
 
-# pymupdf4llm drops its top-level IdentifyHeaders/TocHeaders when pymupdf-layout is installed.
+# pymupdf-layout removes these names from the package namespace; the submodule keeps them.
 from pymupdf4llm.helpers.pymupdf_rag import IdentifyHeaders, TocHeaders
 
 from extralit_server.contexts.ocr.docling_builder import (
@@ -154,6 +154,18 @@ def _blocks_for_page(page: pymupdf.Page, headers: Any) -> list[LayoutBlock]:
     return blocks
 
 
+class PyMuPDFParser:
+    """`PageParser` over pymupdf; it reads the text layer, so `ocr` has no effect."""
+
+    def parse_pages(self, pdf_bytes: bytes, pages: Sequence[int], *, ocr: bool) -> dict[int, list[LayoutBlock]]:
+        """Parse the given 1-indexed pages into top-left-origin blocks, keyed by page number."""
+        if not pages:
+            return {}
+        with pymupdf.open(stream=pdf_bytes, filetype="pdf") as pdf:
+            headers = _header_levels(pdf, pages)
+            return {p: _blocks_for_page(pdf[p - 1], headers) for p in sorted(set(pages)) if 0 < p <= pdf.page_count}
+
+
 def parse(
     pdf_bytes: bytes,
     *,