wehub-resource-sync
910 lines
34 KiB
Python
910 lines
34 KiB
Python
"""MinerU IR builder tests: content_list.json → IR translation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from lightrag.parser.external.mineru import MinerUIRBuilder
|
|
|
|
|
|
def _write_bundle(tmp_path: Path, content_list: list[dict]) -> Path:
|
|
"""Build a minimal *.mineru_raw/ directory."""
|
|
raw = tmp_path / "doc.mineru_raw"
|
|
raw.mkdir()
|
|
(raw / "content_list.json").write_text(json.dumps(content_list, ensure_ascii=False))
|
|
return raw
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_simple_text_and_heading(tmp_path: Path) -> None:
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
{"type": "text", "text": "1 Introduction", "text_level": 1},
|
|
{"type": "text", "text": "Body paragraph."},
|
|
{"type": "text", "text": "1.1 Sub", "text_level": 2},
|
|
{"type": "text", "text": "Sub body."},
|
|
],
|
|
)
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="x.pdf")
|
|
|
|
assert ir.doc_title == "1 Introduction"
|
|
assert ir.document_format == "pdf"
|
|
# Heading + body merge into a single block per heading.
|
|
assert len(ir.blocks) == 2
|
|
assert ir.blocks[0].heading == "1 Introduction"
|
|
assert ir.blocks[0].level == 1
|
|
# Heading line is rendered with markdown ``#`` prefix matching the level.
|
|
assert ir.blocks[0].content_template == "# 1 Introduction\nBody paragraph."
|
|
# Sub-heading updates stack and records parent.
|
|
assert ir.blocks[1].heading == "1.1 Sub"
|
|
assert ir.blocks[1].level == 2
|
|
assert ir.blocks[1].parent_headings == ["1 Introduction"]
|
|
assert ir.blocks[1].content_template == "## 1.1 Sub\nSub body."
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_preface_block_for_pre_heading_content(tmp_path: Path) -> None:
|
|
"""Items emitted before the first heading land in a synthetic
|
|
``Preface/Uncategorized`` block at level 0."""
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
{"type": "text", "text": "Floating intro line."},
|
|
{"type": "list", "list_items": ["a", "b"]},
|
|
{"type": "text", "text": "Section A", "text_level": 1},
|
|
{"type": "text", "text": "A body."},
|
|
],
|
|
)
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="p.pdf")
|
|
|
|
assert len(ir.blocks) == 2
|
|
preface = ir.blocks[0]
|
|
assert preface.heading == "Preface/Uncategorized"
|
|
assert preface.level == 0
|
|
assert preface.parent_headings == []
|
|
assert preface.content_template == "Floating intro line.\na\nb"
|
|
|
|
section = ir.blocks[1]
|
|
assert section.heading == "Section A"
|
|
assert section.level == 1
|
|
assert section.content_template == "# Section A\nA body."
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_merges_mixed_payloads_under_heading(tmp_path: Path) -> None:
|
|
"""Tables / images / equations / code under the same heading merge into
|
|
one block; their placeholders appear in document order."""
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
{"type": "text", "text": "Methods", "text_level": 1},
|
|
{"type": "text", "text": "We did stuff."},
|
|
{
|
|
"type": "table",
|
|
"table_body": [["a", "b"], ["1", "2"]],
|
|
"num_rows": 2,
|
|
"num_cols": 2,
|
|
},
|
|
{"type": "image", "img_path": "images/fig1.png"},
|
|
{"type": "equation", "text": "$$E = mc^2$$"},
|
|
{"type": "code", "code_body": "print('ok')"},
|
|
],
|
|
)
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="m.pdf")
|
|
assert len(ir.blocks) == 1
|
|
block = ir.blocks[0]
|
|
assert block.heading == "Methods"
|
|
assert block.level == 1
|
|
assert len(block.tables) == 1
|
|
assert len(block.drawings) == 1
|
|
assert len(block.equations) == 1
|
|
# Lines are joined in source order; the heading carries its ``#`` prefix.
|
|
expected_lines = [
|
|
"# Methods",
|
|
"We did stuff.",
|
|
f"{{{{TBL:{block.tables[0].placeholder_key}}}}}",
|
|
f"{{{{IMG:{block.drawings[0].placeholder_key}}}}}",
|
|
f"{{{{EQ:{block.equations[0].placeholder_key}}}}}",
|
|
"print('ok')",
|
|
]
|
|
assert block.content_template == "\n".join(expected_lines)
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_table_and_drawing_and_equation(tmp_path: Path) -> None:
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
{
|
|
"type": "table",
|
|
"table_body": [["a", "b"], ["1", "2"]],
|
|
"num_rows": 2,
|
|
"num_cols": 2,
|
|
"table_caption": ["Tbl"],
|
|
"header": [["a", "b"]],
|
|
},
|
|
{
|
|
"type": "image",
|
|
"img_path": "images/img_001.jpg",
|
|
"image_caption": ["Fig 1"],
|
|
"page_idx": 1,
|
|
"bbox": [10, 20, 30, 40],
|
|
},
|
|
{"type": "equation", "text": "$E = mc^2$", "caption": "Eq 1"},
|
|
],
|
|
)
|
|
# The drawing references images/img_001.jpg — adapter accepts missing
|
|
# files and produces an AssetSpec with source=None.
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="d.pdf")
|
|
|
|
table_block = next(b for b in ir.blocks if b.tables)
|
|
table = table_block.tables[0]
|
|
assert table.rows == [["a", "b"], ["1", "2"]]
|
|
assert table.num_rows == 2 and table.num_cols == 2
|
|
assert table.caption == "Tbl"
|
|
assert table.table_header == [["a", "b"]]
|
|
assert table.self_ref == "content_list.json#/0"
|
|
|
|
drawing_block = next(b for b in ir.blocks if b.drawings)
|
|
drawing = drawing_block.drawings[0]
|
|
assert drawing.fmt == "jpg"
|
|
assert drawing.caption == "Fig 1"
|
|
assert drawing.self_ref == "content_list.json#/1"
|
|
# Position carried through. The bbox-bearing item produces exactly one
|
|
# fine-grained position (anchor + range) and is NOT also rolled into the
|
|
# page-only summary channel — so the block has a single position entry,
|
|
# not a duplicate summary + bbox pair.
|
|
assert len(drawing_block.positions) == 1
|
|
assert drawing_block.positions[0].type == "bbox"
|
|
# Anchor is always serialized as a string (uniform on-disk format,
|
|
# accommodates book pagination labels like Roman "ii").
|
|
assert drawing_block.positions[0].anchor == "2" # page_idx+1
|
|
assert drawing_block.positions[0].range == [10.0, 20.0, 30.0, 40.0]
|
|
|
|
# Asset is declared with the relative path as ref.
|
|
assert any(a.ref == "images/img_001.jpg" for a in ir.assets)
|
|
|
|
equation_block = next(b for b in ir.blocks if b.equations)
|
|
eq = equation_block.equations[0]
|
|
# IREquation.latex preserves MinerU's raw form so blocks.jsonl shows it
|
|
# verbatim; equations.json strips the ``$`` wrappers downstream (writer).
|
|
assert eq.latex == "$E = mc^2$"
|
|
assert eq.is_block is True
|
|
assert eq.caption == "Eq 1"
|
|
assert eq.self_ref == "content_list.json#/2"
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_page_idx_aggregated_and_deduped_when_no_bbox(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Real MinerU output carries ``page_idx`` on every item but rarely a
|
|
``bbox``. Each unique page contributing to a merged block must surface as
|
|
one anchor-only ``{type:"bbox", anchor:<page+1>}`` entry, sorted, no
|
|
duplicates, no ``range``.
|
|
"""
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
{"type": "text", "text": "Section", "text_level": 1, "page_idx": 0},
|
|
{"type": "text", "text": "line A", "page_idx": 0},
|
|
{"type": "text", "text": "line B", "page_idx": 1},
|
|
{"type": "text", "text": "line C", "page_idx": 1},
|
|
{"type": "text", "text": "line D", "page_idx": 2},
|
|
],
|
|
)
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="p.pdf")
|
|
|
|
assert len(ir.blocks) == 1
|
|
block = ir.blocks[0]
|
|
# Pages 0, 1, 2 → anchors "1", "2", "3" — one entry per unique page.
|
|
# Anchors are persisted as strings for on-disk uniformity.
|
|
assert len(block.positions) == 3
|
|
anchors = [p.anchor for p in block.positions]
|
|
assert anchors == ["1", "2", "3"]
|
|
for pos in block.positions:
|
|
assert pos.type == "bbox"
|
|
# Page-only summary entries have no range; ``to_jsonable`` must omit
|
|
# the key entirely.
|
|
assert pos.range is None
|
|
assert "range" not in pos.to_jsonable()
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_bbox_items_and_page_only_items_coexist(tmp_path: Path) -> None:
|
|
"""When a block merges both bbox-bearing and bbox-less items, the bbox
|
|
items are emitted per-item (no dedupe, with ``range``) and only the
|
|
bbox-less items contribute to the page-only summary. Ordering: summary
|
|
first (sorted by anchor), bbox entries after (source order).
|
|
"""
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
{"type": "text", "text": "Mixed", "text_level": 1, "page_idx": 1},
|
|
{
|
|
"type": "image",
|
|
"img_path": "images/fig.png",
|
|
"page_idx": 1,
|
|
"bbox": [10, 20, 30, 40],
|
|
},
|
|
{"type": "text", "text": "tail line", "page_idx": 2},
|
|
],
|
|
)
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="m.pdf")
|
|
|
|
assert len(ir.blocks) == 1
|
|
positions = ir.blocks[0].positions
|
|
# One page-only summary for page 3 (the bbox-less tail line) and one
|
|
# bbox entry for page 2 (the image). The heading item has page_idx=1
|
|
# but no bbox, so it adds anchor 2 to the page set — combined with the
|
|
# tail item's anchor 3 the summary section has TWO anchors (1+1, 2+1).
|
|
assert [(p.anchor, p.range) for p in positions] == [
|
|
("2", None),
|
|
("3", None),
|
|
("2", [10.0, 20.0, 30.0, 40.0]),
|
|
]
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_page_sort_books_convention_with_mixed_anchors(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Block merges items with Roman preface labels and Arabic numerals.
|
|
|
|
Two guarantees:
|
|
|
|
1. The adapter must not crash when sorting heterogeneous anchors — a
|
|
previous bug surfaced ``TypeError: '<' not supported between
|
|
instances of 'str' and 'int'`` whenever ``page_idx`` mixed types.
|
|
2. Output order follows book pagination convention: Roman / letter
|
|
labels first (lexical), then numeric pages by integer value, so
|
|
``"2"`` precedes ``"10"`` (not ``"10"`` before ``"2"`` as a naive
|
|
lexical sort would do).
|
|
"""
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
{
|
|
"type": "text",
|
|
"text": "Mixed Pagination",
|
|
"text_level": 1,
|
|
"page_idx": "i",
|
|
},
|
|
{"type": "text", "text": "preface intro", "page_idx": "i"},
|
|
{"type": "text", "text": "preface tail", "page_idx": "ii"},
|
|
{"type": "text", "text": "chapter line A", "page_idx": 1}, # → "2"
|
|
{"type": "text", "text": "chapter line B", "page_idx": 9}, # → "10"
|
|
],
|
|
)
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="mix.pdf")
|
|
|
|
assert len(ir.blocks) == 1
|
|
anchors = [p.anchor for p in ir.blocks[0].positions]
|
|
# Roman labels first (lex order), then numerics by int value.
|
|
assert anchors == ["i", "ii", "2", "10"]
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_empty_text_item_does_not_leak_page_to_block(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""An item whose body is empty must NOT contribute its ``page_idx`` to
|
|
the current block's positions — otherwise spurious pages from
|
|
content-less items poison provenance.
|
|
|
|
Regression: empty text on page 99 sits between two real headings; its
|
|
page must not appear under either block.
|
|
"""
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
{"type": "text", "text": "Section A", "text_level": 1, "page_idx": 0},
|
|
{"type": "text", "text": "real body", "page_idx": 0},
|
|
# Empty body — should be silently dropped, page_idx not recorded.
|
|
{"type": "text", "text": "", "page_idx": 98},
|
|
{"type": "text", "text": "Section B", "text_level": 1, "page_idx": 1},
|
|
{"type": "text", "text": "next body", "page_idx": 1},
|
|
],
|
|
)
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="leak.pdf")
|
|
|
|
assert len(ir.blocks) == 2
|
|
a_anchors = [p.anchor for p in ir.blocks[0].positions]
|
|
b_anchors = [p.anchor for p in ir.blocks[1].positions]
|
|
# Section A only mentions page 1 (page_idx 0 + 1) — NOT 99 from the
|
|
# dropped empty item.
|
|
assert a_anchors == ["1"]
|
|
assert "99" not in a_anchors and "99" not in b_anchors
|
|
# Section B only mentions page 2 (page_idx 1 + 1).
|
|
assert b_anchors == ["2"]
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_each_heading_starts_its_own_block(tmp_path: Path) -> None:
|
|
"""Every recognized heading starts its own block. Back-to-back headings
|
|
with no body between them are NOT folded — each becomes a standalone
|
|
block whose content is just the heading line; the heading that does have
|
|
a following body merges that body in. Mirrors the native docx parser.
|
|
"""
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
{"type": "text", "text": "1 Top", "text_level": 1},
|
|
{"type": "text", "text": "1.1 Mid", "text_level": 2},
|
|
{"type": "text", "text": "1.1.1 Deep", "text_level": 3},
|
|
{"type": "text", "text": "Body for deep."},
|
|
{"type": "text", "text": "2 Top Again", "text_level": 1},
|
|
{"type": "text", "text": "More body."},
|
|
],
|
|
)
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="m.pdf")
|
|
|
|
summary = [
|
|
(b.heading, b.content_template, b.level, b.parent_headings) for b in ir.blocks
|
|
]
|
|
assert summary == [
|
|
("1 Top", "# 1 Top", 1, []),
|
|
("1.1 Mid", "## 1.1 Mid", 2, ["1 Top"]),
|
|
(
|
|
"1.1.1 Deep",
|
|
"### 1.1.1 Deep\nBody for deep.",
|
|
3,
|
|
["1 Top", "1.1 Mid"],
|
|
),
|
|
("2 Top Again", "# 2 Top Again\nMore body.", 1, []),
|
|
]
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_heading_markdown_prefix_cap_and_existing_marker(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Heading prefix is capped at six ``#`` and a heading whose text already
|
|
carries a markdown prefix is not double-prefixed in content but is stored
|
|
cleanly in metadata."""
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
# level 7 → clamped to six "#".
|
|
{"type": "text", "text": "Deep", "text_level": 7},
|
|
{"type": "text", "text": "deep body."},
|
|
# Source text already a markdown heading → kept verbatim.
|
|
{"type": "text", "text": "# Already MD", "text_level": 1},
|
|
{"type": "text", "text": "md body."},
|
|
{"type": "text", "text": "## Child MD", "text_level": 2},
|
|
{"type": "text", "text": "child body."},
|
|
],
|
|
)
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="m.pdf")
|
|
|
|
first_lines = [b.content_template.split("\n")[0] for b in ir.blocks]
|
|
assert first_lines == ["###### Deep", "# Already MD", "## Child MD"]
|
|
assert [(b.heading, b.parent_headings) for b in ir.blocks] == [
|
|
("Deep", []),
|
|
("Already MD", []),
|
|
("Child MD", ["Already MD"]),
|
|
]
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_adjacent_shallower_heading_starts_new_block(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Inverse case: when the second adjacent heading is shallower (level
|
|
number smaller or equal), it must NOT merge — it starts a new block.
|
|
"""
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
{"type": "text", "text": "1.1 Mid first", "text_level": 2},
|
|
{"type": "text", "text": "2 Top after", "text_level": 1},
|
|
{"type": "text", "text": "body"},
|
|
],
|
|
)
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="m.pdf")
|
|
|
|
# The first block is heading-only; the writer downstream will keep it
|
|
# (the merged-heading rule only forwards DEEPER headings).
|
|
assert len(ir.blocks) == 2
|
|
assert ir.blocks[0].heading == "1.1 Mid first"
|
|
assert ir.blocks[0].level == 2
|
|
assert ir.blocks[0].content_template == "## 1.1 Mid first"
|
|
|
|
assert ir.blocks[1].heading == "2 Top after"
|
|
assert ir.blocks[1].level == 1
|
|
assert ir.blocks[1].content_template == "# 2 Top after\nbody"
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_body_breaks_adjacent_heading_merge(tmp_path: Path) -> None:
|
|
"""Once any body content lands in the current block, the next heading —
|
|
even a deeper one — must flush and open a fresh block (no merge)."""
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
{"type": "text", "text": "1 Top", "text_level": 1},
|
|
{"type": "text", "text": "Intro line under 1."},
|
|
{"type": "text", "text": "1.1 Mid", "text_level": 2},
|
|
{"type": "text", "text": "Mid body."},
|
|
],
|
|
)
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="m.pdf")
|
|
|
|
assert len(ir.blocks) == 2
|
|
assert ir.blocks[0].content_template == "# 1 Top\nIntro line under 1."
|
|
assert ir.blocks[1].heading == "1.1 Mid"
|
|
assert ir.blocks[1].parent_headings == ["1 Top"]
|
|
assert ir.blocks[1].content_template == "## 1.1 Mid\nMid body."
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_block_equation_preserves_dollar_wrappers(tmp_path: Path) -> None:
|
|
"""Block equations keep the ``$$`` markers verbatim on IREquation.latex
|
|
so the writer renders blocks.jsonl's ``<equation>`` body byte-identical
|
|
to MinerU's source. The downstream writer is responsible for stripping
|
|
them when generating equations.json."""
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
{
|
|
"type": "equation",
|
|
"text": "$$\n\\int_0^1 x dx = \\tfrac{1}{2}\n$$",
|
|
"text_format": "block",
|
|
"caption": "Eq A",
|
|
},
|
|
],
|
|
)
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="b.pdf")
|
|
eq = ir.blocks[0].equations[0]
|
|
assert eq.is_block is True
|
|
# No stripping in the adapter; whitespace.strip() only.
|
|
assert eq.latex == "$$\n\\int_0^1 x dx = \\tfrac{1}{2}\n$$"
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_empty_equation_dropped(tmp_path: Path) -> None:
|
|
"""Fix 2: equation items with empty text MUST NOT enter the IR (and
|
|
consequently not the sidecar). They previously left dangling sidecar
|
|
entries."""
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
{"type": "equation", "text": "", "caption": "ghost"},
|
|
{"type": "equation", "text": " ", "caption": "ghost"},
|
|
{"type": "text", "text": "kept"},
|
|
],
|
|
)
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="g.pdf")
|
|
eq_count = sum(len(b.equations) for b in ir.blocks)
|
|
assert eq_count == 0
|
|
assert any(b.content_template == "kept" for b in ir.blocks)
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_empty_table_dropped(tmp_path: Path) -> None:
|
|
"""Table items with no usable body MUST NOT enter the IR.
|
|
|
|
MinerU sometimes misidentifies a page-number / blank region as a table
|
|
and emits a body-less ``table`` item (missing ``table_body``/``rows``,
|
|
or with an empty string / empty grid). Leaving such items in the IR
|
|
would later trip the analyze worker's hard-failure path on empty
|
|
``content``. The IR builder filters them upstream.
|
|
"""
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
# 1) Body field completely absent.
|
|
{"type": "table", "num_rows": 0, "num_cols": 0},
|
|
# 2) Empty string body (matches the real m012-manual.pdf bug).
|
|
{"type": "table", "table_body": ""},
|
|
# 3) Empty list body.
|
|
{"type": "table", "rows": []},
|
|
# 4) Grid with only blank cells.
|
|
{"type": "table", "rows": [["", " "], ["\t", ""]]},
|
|
# 5) A real text item so the IR is not entirely empty.
|
|
{"type": "text", "text": "kept"},
|
|
],
|
|
)
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="t.pdf")
|
|
table_count = sum(len(b.tables) for b in ir.blocks)
|
|
assert table_count == 0
|
|
# No table placeholder should leak into the rendered content either.
|
|
joined = "\n".join(b.content_template for b in ir.blocks)
|
|
assert "TBL:" not in joined
|
|
assert "kept" in joined
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_page_number_dropped(tmp_path: Path) -> None:
|
|
"""``page_number`` items are layout noise and MUST NOT enter the IR.
|
|
|
|
MinerU emits a ``page_number`` item per page. Empty-text page numbers were
|
|
already dropped by the fallback's _append_text guard, but page numbers that
|
|
carry real text (``"12"``, ``"iii"``) previously leaked into block content
|
|
and contaminated the block's positions with their page_idx. The IR builder
|
|
now skips them outright regardless of text.
|
|
"""
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
{"type": "text", "text": "kept", "page_idx": 0},
|
|
# Page number carrying real text — the regression case.
|
|
{"type": "page_number", "text": "12", "page_idx": 7},
|
|
# Roman-numeral page number.
|
|
{"type": "page_number", "text": "iii", "page_idx": 8},
|
|
# Blank/whitespace page number.
|
|
{"type": "page_number", "text": "- ", "page_idx": 9},
|
|
],
|
|
)
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="p.pdf")
|
|
joined = "\n".join(b.content_template for b in ir.blocks)
|
|
assert "12" not in joined
|
|
assert "iii" not in joined
|
|
assert "kept" in joined
|
|
# The page_idx of skipped page numbers must not leak into block positions.
|
|
# Anchors are 1-based strings, so page_idx 7/8/9 would surface as 8/9/10.
|
|
anchors = {pos.anchor for b in ir.blocks for pos in b.positions}
|
|
assert anchors.isdisjoint({"8", "9", "10"})
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_bbox_attributes_default_and_override(tmp_path: Path) -> None:
|
|
raw = _write_bundle(tmp_path, [{"type": "text", "text": "x"}])
|
|
adapter = MinerUIRBuilder()
|
|
ir = adapter.normalize_from_workdir(raw, document_name="x.pdf")
|
|
assert ir.bbox_attributes == {"origin": "LEFTTOP", "max": 1000}
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_bbox_attributes_env_override(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.setenv(
|
|
"MINERU_BBOX_ATTRIBUTES",
|
|
'{"origin": "LEFTBOTTOM", "max": 612}',
|
|
)
|
|
raw = _write_bundle(tmp_path, [{"type": "text", "text": "x"}])
|
|
adapter = MinerUIRBuilder()
|
|
ir = adapter.normalize_from_workdir(raw, document_name="x.pdf")
|
|
assert ir.bbox_attributes == {"origin": "LEFTBOTTOM", "max": 612}
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_engine_version_recorded_in_split_option(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.setenv("MINERU_ENGINE_VERSION", "magic-pdf 1.5.4")
|
|
raw = _write_bundle(tmp_path, [{"type": "text", "text": "x"}])
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="x.pdf")
|
|
assert ir.split_option == {"engine_version": "magic-pdf 1.5.4"}
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_missing_content_list_raises(tmp_path: Path) -> None:
|
|
raw_dir = tmp_path / "bad.mineru_raw"
|
|
raw_dir.mkdir()
|
|
with pytest.raises(FileNotFoundError):
|
|
MinerUIRBuilder().normalize_from_workdir(raw_dir, document_name="x.pdf")
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_html_table_preserves_merged_cells_and_extracts_thead(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""MinerU HTML table_body must stay HTML so rowspan/colspan survive."""
|
|
table_html = (
|
|
'<table><thead><tr><th rowspan="2">Metric</th>'
|
|
'<th colspan="2">Group</th></tr><tr><th>A</th><th>B</th></tr></thead>'
|
|
"<tbody><tr><td>Accuracy</td><td>91%</td><td>93%</td></tr></tbody></table>"
|
|
)
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
{
|
|
"type": "table",
|
|
"table_body": table_html,
|
|
"table_caption": ["Tbl"],
|
|
}
|
|
],
|
|
)
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="h.pdf")
|
|
table = ir.blocks[0].tables[0]
|
|
|
|
assert table.rows is None
|
|
assert table.html == table_html
|
|
assert 'rowspan="2"' in table.html
|
|
assert 'colspan="2"' in table.html
|
|
assert table.body_override == table_html.removeprefix("<table>").removesuffix(
|
|
"</table>"
|
|
)
|
|
assert table.num_rows == 3
|
|
assert table.num_cols == 3
|
|
assert table.caption == "Tbl"
|
|
# The <thead> is preserved verbatim (raw HTML) so rowspan/colspan survive —
|
|
# not flattened into a grid.
|
|
assert table.table_header == (
|
|
'<thead><tr><th rowspan="2">Metric</th>'
|
|
'<th colspan="2">Group</th></tr><tr><th>A</th><th>B</th></tr></thead>'
|
|
)
|
|
assert 'rowspan="2"' in table.table_header
|
|
assert 'colspan="2"' in table.table_header
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_html_table_header_preserves_spans(tmp_path: Path) -> None:
|
|
"""A spanned <thead> is stored verbatim — colspan/rowspan markup is kept
|
|
intact rather than flattened into a rectangular grid (so merged-cell
|
|
semantics survive into every repeated header chunk downstream)."""
|
|
thead = (
|
|
"<thead>"
|
|
'<tr><th>ID</th><th colspan="3">Scores</th></tr>'
|
|
'<tr><th>n</th><th>A</th><th colspan="2">BC</th></tr>'
|
|
"</thead>"
|
|
)
|
|
table_html = (
|
|
f"<table>{thead}<tbody>"
|
|
"<tr><td>1</td><td>x</td><td>y</td><td>z</td></tr>"
|
|
"</tbody></table>"
|
|
)
|
|
raw = _write_bundle(tmp_path, [{"type": "table", "table_body": table_html}])
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="h.pdf")
|
|
table = ir.blocks[0].tables[0]
|
|
assert table.num_cols == 4
|
|
assert table.table_header == thead
|
|
assert table.table_header.count('colspan="3"') == 1
|
|
assert table.table_header.count('colspan="2"') == 1
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_html_table_without_thead_does_not_guess_header(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Only a real HTML <thead> should populate table_header."""
|
|
table_html = (
|
|
"<table><tbody><tr><td>H1</td><td>H2</td></tr>"
|
|
"<tr><td>a</td><td>b</td></tr></tbody></table>"
|
|
)
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
{
|
|
"type": "table",
|
|
"table_body": table_html,
|
|
"num_rows": 1,
|
|
"num_cols": 2,
|
|
}
|
|
],
|
|
)
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="h.pdf")
|
|
table = ir.blocks[0].tables[0]
|
|
assert table.rows is None
|
|
assert table.html == table_html
|
|
assert table.table_header is None
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_html_table_without_thead_falls_back_to_header_grid(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""An HTML table whose markup carries no <thead> but for which MinerU
|
|
supplied a separate ``header`` grid keeps that grid (the writer renders it to
|
|
a span-less <thead>) rather than silently dropping the recovered header."""
|
|
table_html = (
|
|
"<table><tbody><tr><td>a</td><td>b</td></tr></tbody></table>" # no <thead>
|
|
)
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
{
|
|
"type": "table",
|
|
"table_body": table_html,
|
|
"header": [["H1", "H2"]],
|
|
"num_rows": 1,
|
|
"num_cols": 2,
|
|
}
|
|
],
|
|
)
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="h.pdf")
|
|
table = ir.blocks[0].tables[0]
|
|
assert table.html == table_html
|
|
assert table.table_header == [["H1", "H2"]] # grid kept, not dropped
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_html_table_attr_with_gt_keeps_inner_body(tmp_path: Path) -> None:
|
|
"""A ``>`` inside a ``<table>`` attribute must not truncate the open tag,
|
|
so ``body_override`` still strips the full ``<table …>`` wrapper."""
|
|
table_html = '<table data-x="a>b"><tbody><tr><td>v</td></tr></tbody></table>'
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[{"type": "table", "table_body": table_html}],
|
|
)
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="h.pdf")
|
|
table = ir.blocks[0].tables[0]
|
|
assert table.html == table_html
|
|
assert table.body_override == "<tbody><tr><td>v</td></tr></tbody>"
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_empty_html_table_falls_back_to_full_html(tmp_path: Path) -> None:
|
|
"""A degenerate ``<table></table>`` has no inner body, so ``body_override``
|
|
stays None and the writer renders ``table.html`` verbatim."""
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[{"type": "table", "table_body": "<table></table>"}],
|
|
)
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="h.pdf")
|
|
table = ir.blocks[0].tables[0]
|
|
assert table.html == "<table></table>"
|
|
assert table.body_override is None
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_html_table_strips_html_body_wrapper(tmp_path: Path) -> None:
|
|
"""MinerU's table model may wrap output in ``<html><body>``; the adapter
|
|
must unwrap to a single ``<table>`` so the writer does not nest tables and
|
|
``TABLE_TAG_RE`` consumers are not truncated at the inner ``</table>``."""
|
|
inner = (
|
|
'<thead><tr><th colspan="2">G</th></tr></thead>'
|
|
"<tbody><tr><td>a</td><td>b</td></tr></tbody>"
|
|
)
|
|
payload = f"<html><body><table>{inner}</table></body></html>"
|
|
raw = _write_bundle(tmp_path, [{"type": "table", "table_body": payload}])
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="h.pdf")
|
|
table = ir.blocks[0].tables[0]
|
|
assert table.html == f"<table>{inner}</table>" # <html>/<body> wrapper gone
|
|
assert table.body_override == inner # block text renders only the inner body
|
|
assert table.num_cols == 2
|
|
# The <thead> is kept verbatim (colspan preserved), not flattened to a grid.
|
|
assert table.table_header == '<thead><tr><th colspan="2">G</th></tr></thead>'
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_list_items_joined_with_newline(tmp_path: Path) -> None:
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
{"type": "list", "list_items": ["one", "two", "three"]},
|
|
],
|
|
)
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="l.pdf")
|
|
assert ir.blocks[0].content_template == "one\ntwo\nthree"
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_drawing_asset_source_only_when_file_exists(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""The adapter should declare an AssetSpec for the drawing in both
|
|
cases, but ``source`` is set only when the bytes are on disk; the
|
|
writer then warns and skips a missing-source asset."""
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
{"type": "image", "img_path": "images/exists.png"},
|
|
{"type": "image", "img_path": "images/missing.png"},
|
|
],
|
|
)
|
|
(raw / "images").mkdir()
|
|
(raw / "images" / "exists.png").write_bytes(b"\x89PNG")
|
|
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="a.pdf")
|
|
by_ref = {a.ref: a for a in ir.assets}
|
|
assert by_ref["images/exists.png"].source is not None
|
|
assert by_ref["images/missing.png"].source is None
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_refuses_path_traversal_img_path(tmp_path: Path) -> None:
|
|
"""Untrusted img_path with ``..`` or absolute filesystem segments must
|
|
not be allowed to point ``AssetSpec.source`` outside ``raw_dir``.
|
|
|
|
Otherwise the writer would copy attacker-named files from the host into
|
|
the sidecar's ``*.blocks.assets/`` directory (file-disclosure path).
|
|
"""
|
|
# Place a "secret" file outside the raw bundle that should never be
|
|
# selectable as an asset source.
|
|
secret = tmp_path / "secret.txt"
|
|
secret.write_bytes(b"private")
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
{"type": "image", "img_path": "../secret.txt"},
|
|
{"type": "image", "img_path": str(secret)}, # absolute path
|
|
],
|
|
)
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="x.pdf")
|
|
by_ref = {a.ref: a for a in ir.assets}
|
|
|
|
# Relative ``..`` escape is rejected outright.
|
|
assert by_ref["../secret.txt"].source is None
|
|
|
|
# Absolute filesystem path is reinterpreted as ``images/<basename>``
|
|
# inside raw_dir. Since no such file exists, source must remain None
|
|
# (and crucially must not point at the original secret file).
|
|
abs_asset = by_ref[str(secret)]
|
|
assert abs_asset.source is None
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_absolute_url_img_path_resolves_to_images_basename(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""When MinerU emits an absolute URL in img_path, the downloader saves
|
|
it as ``images/<basename>``; the adapter must look there too."""
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
{
|
|
"type": "image",
|
|
"img_path": "https://cdn.example.com/imgs/figure_42.png",
|
|
},
|
|
],
|
|
)
|
|
(raw / "images").mkdir()
|
|
(raw / "images" / "figure_42.png").write_bytes(b"\x89PNGfake")
|
|
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="u.pdf")
|
|
asset = ir.assets[0]
|
|
assert asset.ref == "https://cdn.example.com/imgs/figure_42.png"
|
|
assert asset.suggested_name == "figure_42.png"
|
|
assert asset.source is not None
|
|
assert asset.source.read_bytes() == b"\x89PNGfake"
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_image_url_template_mode_maps_relative_to_images_basename(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""When MINERU_IMAGE_URL_TEMPLATE is set, MinerURawClient stores every
|
|
image reference — including relative ones — at ``images/<basename>``.
|
|
The adapter must mirror that lookup so the asset is wired up, otherwise
|
|
the downloaded bytes are silently dropped from the sidecar."""
|
|
monkeypatch.setenv(
|
|
"MINERU_IMAGE_URL_TEMPLATE",
|
|
"http://mineru.internal/assets/{name}",
|
|
)
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
{"type": "image", "img_path": "page/img.png"},
|
|
],
|
|
)
|
|
# Downloader's actual landing spot in template mode.
|
|
(raw / "images").mkdir()
|
|
(raw / "images" / "img.png").write_bytes(b"\x89PNGtemplate")
|
|
# The "naive" location (raw_dir/page/img.png) does NOT exist; in
|
|
# template mode the downloader does not write there.
|
|
assert not (raw / "page" / "img.png").exists()
|
|
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="t.pdf")
|
|
asset = ir.assets[0]
|
|
assert asset.source is not None
|
|
assert asset.source.read_bytes() == b"\x89PNGtemplate"
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_adapter_no_template_keeps_relative_path_lookup(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Sanity: without MINERU_IMAGE_URL_TEMPLATE, a relative img_path still
|
|
resolves under raw_dir at its original location (regression guard for
|
|
the template-mode change above)."""
|
|
raw = _write_bundle(
|
|
tmp_path,
|
|
[
|
|
{"type": "image", "img_path": "page/img.png"},
|
|
],
|
|
)
|
|
(raw / "page").mkdir()
|
|
(raw / "page" / "img.png").write_bytes(b"\x89PNGrel")
|
|
|
|
ir = MinerUIRBuilder().normalize_from_workdir(raw, document_name="r.pdf")
|
|
asset = ir.assets[0]
|
|
assert asset.source is not None
|
|
assert asset.source.read_bytes() == b"\x89PNGrel"
|