obsidian-plugin/lllin000_PaperForge
1
0
Fork 0
mirror of https://github.com/lllin000/PaperForge.git synced 2026-07-22 06:50:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat(ocr): complete table vnext passes with notes, accounting, and diff tooling

Browse source
This commit is contained in:
LLLin000 2026-07-03 20:19:46 +08:00
parent ea6a1f0378
commit a9e68ac9e0
5 changed files with 601 additions and 65 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
paperforge/worker/ocr_table_domain.py
View file

@ -113,9 +113,9 @@ def assemble_table_inventory(
if t.get("has_asset") and t.get("asset_block_id")
}
held_tables = [
record["caption"]
record["held_table"]
for record in candidate_index.caption_records
if record.get("status") == "held"
if record.get("status") == "held" and "held_table" in record
]
unmatched_captions = [
record["caption"]

383
paperforge/worker/ocr_table_passes.py Normal file
View file

@ -0,0 +1,383 @@
from __future__ import annotations
from .ocr_pairing_types import ClaimProposal, PassReport, ResourceRef
class TableWeakCaptionRecoveryPass:
name = "table_weak_caption_recovery"
def run(self, state):
from . import ocr_tables
report = PassReport(pass_name=self.name)
for record in state.candidate_index.caption_records:
if not record["is_weak_truncated"]:
continue
caption = record["caption"]
continuation = ocr_tables._find_table_caption_continuation(caption, state.corpus.blocks)
materialized, continuation_ids = ocr_tables._materialize_table_caption(caption, continuation)
record["caption"] = materialized
record["caption_text"] = str(materialized.get("text", "") or "")
record["continuation_ids"] = continuation_ids
if record["is_validation_first_candidate"]:
same_page_assets = state.candidate_index.assets_by_page.get(int(caption.get("page", 0) or 0), [])
if not same_page_assets:
record["status"] = "held"
record["held_table"] = {
"table_id": f"held_table_{len([r for r in state.candidate_index.caption_records if r.get('status') == 'held']) + 1:03d}",
"caption_block_id": record["caption_block_id"],
"page": caption.get("page", 0),
"caption_text": record["caption_text"],
"table_number": record["table_number"],
"formal_table_number": record["formal_table_number"],
"hold_reason": "insufficient_caption_evidence",
"zone": caption.get("zone", ""),
"style_family": caption.get("style_family", ""),
"marker_signature": caption.get("marker_signature", {}),
}
return report
class TableSamePagePass:
name = "table_same_page"
def run(self, state):
from . import ocr_tables
report = PassReport(pass_name=self.name)
for record in state.candidate_index.caption_records:
if record.get("status") != "pending":
continue
caption = record["caption"]
caption_page = int(caption.get("page", 0) or 0)
page_assets = [
(idx, asset)
for idx, asset in state.candidate_index.assets_by_page.get(caption_page, [])
if state.ledger.owner_of_asset(page=caption_page, block_id=asset.get("block_id")) is None
]
scored = ocr_tables._score_candidate_assets(page_assets, caption, is_continuation=record["is_continuation"])
scored.sort(key=lambda item: item[2].get("score", 0.0), reverse=True)
record["candidate_assets"] = [
{"asset_block_id": asset.get("block_id", ""), "match_score": score}
for _, asset, score in scored[:3]
]
if not scored:
continue
top_idx, top_asset, top_score = scored[0]
second_score = scored[1][2].get("score", 0.0) if len(scored) > 1 else -1.0
if top_score.get("score", 0.0) < 0.4 or top_score.get("score", 0.0) - second_score < 0.15:
continue
owner = ResourceRef(kind="legend", page=caption_page, block_id=record["caption_block_id"], figure_no=record["formal_table_number"])
asset_ref = ResourceRef(kind="asset", page=int(top_asset.get("page", 0) or 0), block_id=top_asset.get("block_id"))
conflict = state.ledger.try_claim_assets([asset_ref], owner=owner, reason=self.name)
if conflict is not None:
report.conflicts.append(conflict)
continue
match_status = "matched" if top_score.get("score", 0.0) >= 0.6 else "matched_low_confidence"
state.accept_match(
ClaimProposal(
pass_name=self.name,
figure_no=record["formal_table_number"],
claim_type="match",
legends=[owner],
assets=[asset_ref],
groups=[],
texts=[],
confidence=float(top_score.get("score", 0.0)),
evidence_rank=0,
reason=self.name,
),
{
"caption_block_id": record["caption_block_id"],
"page": caption_page,
"caption_text": record["caption_text"],
"table_number": record["table_number"],
"formal_table_number": record["formal_table_number"],
"asset_block_id": top_asset.get("block_id", ""),
"asset_bbox": top_asset.get("bbox", [0, 0, 0, 0]),
"assistive_text": str(top_asset.get("text", "") or ""),
"truth_source": "image",
"has_asset": True,
"segments": [
{
"page": top_asset.get("page", 0),
"asset_block_id": top_asset.get("block_id", ""),
"asset_bbox": top_asset.get("bbox", [0, 0, 0, 0]),
"is_continuation": record["is_continuation"],
}
],
"note_block_ids": [],
"note_texts": [],
"note_bboxes": [],
"note_band_bbox": [],
"note_match_reason": "",
"note_confidence": 0.0,
"bridge_block_ids": [],
"consumed_block_ids": [record["caption_block_id"], top_asset.get("block_id", ""), *record.get("continuation_ids", [])],
"is_continuation": record["is_continuation"],
"continuation_of": None,
"match_status": match_status,
"candidate_assets": record["candidate_assets"],
"match_score": top_score,
"render_bbox": None,
"render_rotation_deg": 0,
"asset_family_hint": top_asset.get("asset_family_hint"),
"asset_family_confidence": top_asset.get("asset_family_confidence"),
"asset_family_evidence": top_asset.get("asset_family_evidence"),
},
)
record["status"] = "matched"
return report
class TableAdjacentPagePass:
name = "table_adjacent_page"
def run(self, state):
from . import ocr_tables
report = PassReport(pass_name=self.name)
for record in state.candidate_index.caption_records:
if record.get("status") != "pending":
continue
caption = record["caption"]
caption_page = int(caption.get("page", 0) or 0)
candidate_pages = [caption_page - 1, caption_page, caption_page + 1]
all_candidates = []
for page in candidate_pages:
if page < 1:
continue
page_assets = [
(idx, asset)
for idx, asset in state.candidate_index.assets_by_page.get(page, [])
if state.ledger.owner_of_asset(page=page, block_id=asset.get("block_id")) is None
]
all_candidates.extend(ocr_tables._score_candidate_assets(page_assets, caption, is_continuation=record["is_continuation"]))
for _, asset, score_dict in all_candidates:
a_page = int(asset.get("page", 0) or 0)
if a_page == caption_page - 1:
ab = asset.get("bbox") or [0, 0, 0, 0]
cb = caption.get("bbox") or [0, 0, 0, 0]
if len(ab) >= 4 and len(cb) >= 4:
x_ratio = (min(cb[2], ab[2]) - max(cb[0], ab[0])) / max(1.0, min(cb[2] - cb[0], ab[2] - ab[0]))
page_h = max(state.corpus.page_max_y.values()) if state.corpus.page_max_y else 1.0
if x_ratio >= 0.5 and float(ab[3]) >= page_h * 0.85 and float(cb[1]) <= page_h * 0.15:
score_dict["score"] = min(score_dict.get("score", 0.0) + 0.15, 1.0)
score_dict.setdefault("evidence", []).append("continuation_geometry_elevation")
all_candidates.sort(key=lambda item: item[2].get("score", 0.0), reverse=True)
record["candidate_assets"] = [
{"asset_block_id": asset.get("block_id", ""), "match_score": score}
for _, asset, score in all_candidates[:3]
]
if not all_candidates:
record["status"] = "unmatched"
continue
top_idx, top_asset, top_score = all_candidates[0]
second_score = all_candidates[1][2].get("score", 0.0) if len(all_candidates) > 1 else -1.0
if top_score.get("score", 0.0) < 0.4:
record["status"] = "unmatched"
continue
if top_score.get("score", 0.0) - second_score < 0.15:
record["status"] = "ambiguous"
continue
owner = ResourceRef(kind="legend", page=caption_page, block_id=record["caption_block_id"], figure_no=record["formal_table_number"])
asset_ref = ResourceRef(kind="asset", page=int(top_asset.get("page", 0) or 0), block_id=top_asset.get("block_id"))
conflict = state.ledger.try_claim_assets([asset_ref], owner=owner, reason=self.name)
if conflict is not None:
report.conflicts.append(conflict)
continue
match_status = "matched" if top_score.get("score", 0.0) >= 0.6 else "matched_low_confidence"
continuation_of = None
if record["is_continuation"] and record["formal_table_number"] is not None:
for existing in state.matches:
if existing.get("formal_table_number") == record["formal_table_number"] and not existing.get("is_continuation"):
continuation_of = record["formal_table_number"]
break
state.accept_match(
ClaimProposal(
pass_name=self.name,
figure_no=record["formal_table_number"],
claim_type="match",
legends=[owner],
assets=[asset_ref],
groups=[],
texts=[],
confidence=float(top_score.get("score", 0.0)),
evidence_rank=1,
reason=self.name,
),
{
"caption_block_id": record["caption_block_id"],
"page": caption_page,
"caption_text": record["caption_text"],
"table_number": record["table_number"],
"formal_table_number": record["formal_table_number"],
"asset_block_id": top_asset.get("block_id", ""),
"asset_bbox": top_asset.get("bbox", [0, 0, 0, 0]),
"assistive_text": str(top_asset.get("text", "") or ""),
"truth_source": "image",
"has_asset": True,
"segments": [
{
"page": top_asset.get("page", 0),
"asset_block_id": top_asset.get("block_id", ""),
"asset_bbox": top_asset.get("bbox", [0, 0, 0, 0]),
"is_continuation": record["is_continuation"],
}
],
"note_block_ids": [],
"note_texts": [],
"note_bboxes": [],
"note_band_bbox": [],
"note_match_reason": "",
"note_confidence": 0.0,
"bridge_block_ids": [],
"consumed_block_ids": [record["caption_block_id"], top_asset.get("block_id", ""), *record.get("continuation_ids", [])],
"is_continuation": record["is_continuation"],
"continuation_of": continuation_of,
"match_status": match_status,
"candidate_assets": record["candidate_assets"],
"match_score": top_score,
"render_bbox": None,
"render_rotation_deg": 0,
"asset_family_hint": top_asset.get("asset_family_hint"),
"asset_family_confidence": top_asset.get("asset_family_confidence"),
"asset_family_evidence": top_asset.get("asset_family_evidence"),
},
)
record["status"] = "matched"
return report
class TableNotesAttachmentPass:
name = "table_notes_attachment"
def run(self, state):
from . import ocr_tables
report = PassReport(pass_name=self.name)
for table in state.matches:
if not table.get("has_asset"):
table.setdefault("note_block_ids", [])
table.setdefault("note_texts", [])
table.setdefault("note_bboxes", [])
table.setdefault("note_band_bbox", [])
table.setdefault("note_match_reason", "")
table.setdefault("note_confidence", 0.0)
table.setdefault("bridge_block_ids", [])
continue
asset_page = int(table.get("page", 0) or 0)
asset_bbox = table.get("asset_bbox", [0, 0, 0, 0])
asset_bottom = asset_bbox[3] if len(asset_bbox) >= 4 else 0
candidates = []
note_match_reason = ""
for block in state.corpus.blocks:
if int(block.get("page", 0) or 0) != asset_page:
continue
brole = str(block.get("role", "") or "")
braw_label = str(block.get("raw_label", "") or "").strip()
btext = str(block.get("text", "") or "").strip()
is_note = (
brole == "footnote"
or braw_label == "vision_footnote"
or (
0 < len(btext) < 120
and brole not in {
"noise", "page_footer", "page_header", "frontmatter_noise",
"table_caption", "table_caption_candidate",
"table_asset", "media_asset", "figure_caption",
"section_heading", "subsection_heading", "reference_heading",
}
)
)
if not is_note:
continue
bbbox = block.get("bbox") or [0, 0, 0, 0]
if len(bbbox) < 4:
note_match_reason = "invalid_bbox"
continue
if bbbox[1] < asset_bottom or bbbox[1] > asset_bottom + 100:
note_match_reason = "outside_vertical_range"
continue
if ocr_tables._table_note_falls_into_page_footnote_prior(bbbox, asset_page, state.corpus.page_footnote_prior):
note_match_reason = "page_footnote_prior_rejected"
continue
if ocr_tables._looks_like_body_text_below_table(block, asset_bbox):
note_match_reason = "body_text_like_excluded"
continue
candidates.append(block)
if candidates:
candidates.sort(key=lambda b: (b.get("bbox") or [0, 0, 0, 0])[1])
table["note_block_ids"] = [str(b.get("block_id", "")) for b in candidates if b.get("block_id")]
table["note_texts"] = [str(b.get("text", "") or "").strip() for b in candidates if str(b.get("text", "") or "").strip()]
table["note_bboxes"] = [b.get("bbox", [0, 0, 0, 0]) for b in candidates]
table["note_band_bbox"] = [
min(bb[0] for bb in table["note_bboxes"]),
min(bb[1] for bb in table["note_bboxes"]),
max(bb[2] for bb in table["note_bboxes"]),
max(bb[3] for bb in table["note_bboxes"]),
]
table["note_match_reason"] = "note_band_geometry_match"
table["note_confidence"] = 0.85
else:
table.setdefault("note_block_ids", [])
table.setdefault("note_texts", [])
table.setdefault("note_bboxes", [])
table.setdefault("note_band_bbox", [])
table["note_match_reason"] = note_match_reason or "no_footnote_role"
table.setdefault("note_confidence", 0.0)
asset_block = next(
(a for a in state.corpus.raw_assets if str(a.get("block_id", "")) == str(table.get("asset_block_id", "")) and int(a.get("page", 0) or 0) == asset_page),
None,
)
if asset_block is not None:
rot = ocr_tables._table_has_rotated_content(asset_block)
if rot:
ab = table.get("asset_bbox", [])
caption = next(
(r["caption"] for r in state.candidate_index.caption_records if r["caption_block_id"] == table.get("caption_block_id")),
None,
)
cb = (caption.get("bbox") or caption.get("block_bbox") or []) if caption else []
if len(ab) >= 4 and len(cb) >= 4:
table["render_bbox"] = [min(cb[0], ab[0]), min(cb[1], ab[1]), max(cb[2], ab[2]), max(cb[3], ab[3])]
table["render_rotation_deg"] = rot
# Bridge block detection
table["bridge_block_ids"] = [
str(block.get("block_id") or "")
for block in state.corpus.blocks
if int(block.get("page", 0) or 0) == asset_page
and block.get("bridge_eligible")
and str(block.get("layout_region") or "") == "display_zone"
and block.get("block_id")
]
# Consumed block IDs include notes
existing_ids = set(table.get("consumed_block_ids") or [])
for bid in table.get("note_block_ids", []):
if bid not in existing_ids:
table.setdefault("consumed_block_ids", []).append(bid)
return report
class TableFinalAccountingPass:
name = "table_final_accounting"
def run(self, state):
report = PassReport(pass_name=self.name)
state.completeness = {
"total_numbered_tables": len([r for r in state.candidate_index.caption_records if r.get("formal_table_number") is not None]),
"accounted_for": len([r for r in state.candidate_index.caption_records if r.get("status") in {"matched", "held"}]),
"details": [
{"caption_block_id": r["caption_block_id"], "status": r.get("status", "pending")}
for r in state.candidate_index.caption_records
],
}
return report

35
paperforge/worker/ocr_tables.py
View file

@ -1,6 +1,7 @@
from __future__ import annotations
import re
from dataclasses import asdict
from pathlib import Path
from typing import Any
@ -234,6 +235,10 @@ def _table_note_falls_into_page_footnote_prior(
def build_table_inventory(structured_blocks: list[dict]) -> dict[str, Any]:
return build_table_inventory_legacy(structured_blocks)
def build_table_inventory_legacy(structured_blocks: list[dict]) -> dict[str, Any]:
tables: list[dict] = []
captions: list[dict] = []
assets: list[dict] = []
@ -634,3 +639,33 @@ def write_back_table_roles(inventory: dict, structured_blocks: list[dict]) -> No
def write_table_inventory(dst: Path, inventory: dict[str, Any]) -> None:
write_json(dst, inventory)
def build_table_inventory_vnext(structured_blocks: list[dict]) -> dict[str, Any]:
from .ocr_pairing_framework import run_pairing_passes
from .ocr_pairing_state import OwnershipLedger, PipelineState
from .ocr_table_domain import TableCandidateIndex, TableCorpus, assemble_table_inventory
from .ocr_table_passes import (
TableAdjacentPagePass,
TableFinalAccountingPass,
TableNotesAttachmentPass,
TableSamePagePass,
TableWeakCaptionRecoveryPass,
)
corpus = TableCorpus.from_blocks(structured_blocks)
candidate_index = TableCandidateIndex.from_corpus(corpus)
state = PipelineState(corpus=corpus, candidate_index=candidate_index, ledger=OwnershipLedger())
reports = run_pairing_passes(
state,
[
TableWeakCaptionRecoveryPass,
TableSamePagePass,
TableAdjacentPagePass,
TableNotesAttachmentPass,
TableFinalAccountingPass,
],
)
inventory = assemble_table_inventory(state, candidate_index)
inventory["pass_reports"] = [asdict(r) for r in reports]
return inventory

44
scripts/dev/compare_table_inventory_legacy_vs_vnext.py Normal file
View file

@ -0,0 +1,44 @@
from __future__ import annotations
import copy
import json
from pathlib import Path
from paperforge.worker.ocr_tables import build_table_inventory_legacy, build_table_inventory_vnext
def _load_blocks(fixture_root: Path) -> list[dict]:
# Check both fixture_root/structure/blocks.structured.jsonl and fixture_root/blocks.structured.jsonl
candidate = fixture_root / "structure" / "blocks.structured.jsonl"
if not candidate.exists():
candidate = fixture_root / "blocks.structured.jsonl"
return [json.loads(line) for line in candidate.read_text(encoding="utf-8").splitlines() if line.strip()]
def _normalize_table(table: dict) -> dict[str, object]:
return {
"caption_block_id": table.get("caption_block_id"),
"page": table.get("page"),
"formal_table_number": table.get("formal_table_number"),
"asset_block_id": table.get("asset_block_id"),
"match_status": table.get("match_status"),
"note_block_ids": list(table.get("note_block_ids") or []),
"bridge_block_ids": list(table.get("bridge_block_ids") or []),
"consumed_block_ids": list(table.get("consumed_block_ids") or []),
"render_rotation_deg": table.get("render_rotation_deg", 0),
}
def compare_table_inventory_legacy_vs_vnext(fixture_root: Path) -> dict[str, object]:
blocks = _load_blocks(fixture_root)
legacy = build_table_inventory_legacy(copy.deepcopy(blocks))
vnext = build_table_inventory_vnext(copy.deepcopy(blocks))
legacy_norm = [_normalize_table(t) for t in legacy.get("tables", [])]
vnext_norm = [_normalize_table(t) for t in vnext.get("tables", [])]
return {
"legacy": legacy_norm,
"vnext": vnext_norm,
"diff": {
"legacy_only": [item for item in legacy_norm if item not in vnext_norm],
"vnext_only": [item for item in vnext_norm if item not in legacy_norm],
},
}

200
tests/test_ocr_table_pairing_framework.py
View file

@ -5,29 +5,9 @@ def test_table_corpus_collects_captions_assets_and_page_context() -> None:
from paperforge.worker.ocr_table_domain import TableCorpus
blocks = [
{
"block_id": "cap1",
"page": 5,
"role": "table_caption",
"text": "Table 1. Example",
"bbox": [100, 100, 700, 140],
},
{
"block_id": "asset1",
"page": 5,
"role": "table_html",
"raw_label": "table",
"text": "<table><tr><td>x</td></tr></table>",
"bbox": [100, 160, 700, 500],
},
{
"block_id": "note1",
"page": 5,
"role": "footnote",
"raw_label": "vision_footnote",
"text": "* p < 0.05",
"bbox": [100, 520, 300, 550],
},
{"block_id": "cap1", "page": 5, "role": "table_caption", "text": "Table 1. Example", "bbox": [100, 100, 700, 140]},
{"block_id": "asset1", "page": 5, "role": "table_html", "raw_label": "table", "text": "<table><tr><td>x</td></tr></table>", "bbox": [100, 160, 700, 500]},
{"block_id": "note1", "page": 5, "role": "footnote", "raw_label": "vision_footnote", "text": "* p < 0.05", "bbox": [100, 520, 300, 550]},
]
corpus = TableCorpus.from_blocks(blocks)
@ -42,28 +22,9 @@ def test_table_candidate_index_materializes_caption_records_and_assets_by_page()
from paperforge.worker.ocr_table_domain import TableCandidateIndex, TableCorpus
blocks = [
{
"block_id": "cap2",
"page": 6,
"role": "table_caption_candidate",
"text": "Table 2. (continued)",
"bbox": [100, 100, 700, 130],
},
{
"block_id": "cap1",
"page": 5,
"role": "table_caption",
"text": "Table 1. Example",
"bbox": [100, 100, 700, 140],
},
{
"block_id": "asset1",
"page": 5,
"role": "table_html",
"raw_label": "table",
"text": "<table></table>",
"bbox": [100, 160, 700, 500],
},
{"block_id": "cap2", "page": 6, "role": "table_caption_candidate", "text": "Table 2. (continued)", "bbox": [100, 100, 700, 130]},
{"block_id": "cap1", "page": 5, "role": "table_caption", "text": "Table 1. Example", "bbox": [100, 100, 700, 140]},
{"block_id": "asset1", "page": 5, "role": "table_html", "raw_label": "table", "text": "<table></table>", "bbox": [100, 160, 700, 500]},
]
index = TableCandidateIndex.from_corpus(TableCorpus.from_blocks(blocks))
@ -75,21 +36,9 @@ def test_table_candidate_index_materializes_caption_records_and_assets_by_page()
def test_assemble_table_inventory_preserves_public_shape_for_empty_state() -> None:
from paperforge.worker.ocr_pairing_state import OwnershipLedger, PipelineState
from paperforge.worker.ocr_table_domain import (
TableCandidateIndex,
TableCorpus,
assemble_table_inventory,
)
from paperforge.worker.ocr_table_domain import TableCandidateIndex, TableCorpus, assemble_table_inventory
blocks = [
{
"block_id": "cap1",
"page": 1,
"role": "table_caption",
"text": "Table 1. Example",
"bbox": [0, 0, 10, 10],
}
]
blocks = [{"block_id": "cap1", "page": 1, "role": "table_caption", "text": "Table 1. Example", "bbox": [0, 0, 10, 10]}]
corpus = TableCorpus.from_blocks(blocks)
index = TableCandidateIndex.from_corpus(corpus)
state = PipelineState(corpus=corpus, candidate_index=index, ledger=OwnershipLedger())
@ -97,9 +46,134 @@ def test_assemble_table_inventory_preserves_public_shape_for_empty_state() -> No
inventory = assemble_table_inventory(state, index)
assert inventory == {
"tables": [],
"held_tables": [],
"tables": [], "held_tables": [],
"unmatched_captions": [blocks[0]],
"unmatched_assets": [],
"official_table_count": 0,
"unmatched_assets": [], "official_table_count": 0,
}
# ── Task 4: vnext matching passes ──
def test_build_table_inventory_vnext_matches_same_page_best_asset() -> None:
from paperforge.worker.ocr_tables import build_table_inventory_vnext
structured_blocks = [
{"block_id": "cap1", "page": 3, "role": "table_caption", "text": "Table 1. Example", "bbox": [100, 100, 700, 140]},
{"block_id": "asset_good", "page": 3, "role": "table_html", "raw_label": "table", "text": "<table></table>", "bbox": [100, 160, 700, 500]},
{"block_id": "asset_bad", "page": 3, "role": "media_asset", "raw_label": "image", "text": "", "bbox": [800, 160, 1100, 300]},
]
inventory = build_table_inventory_vnext(structured_blocks)
assert inventory["tables"][0]["asset_block_id"] == "asset_good"
assert inventory["tables"][0]["match_status"] in {"matched", "matched_low_confidence"}
def test_build_table_inventory_vnext_holds_validation_first_weak_caption_without_same_page_asset() -> None:
from paperforge.worker.ocr_tables import build_table_inventory_vnext
structured_blocks = [
{
"block_id": "cap1", "page": 4, "role": "body_text",
"text": "Table 2", "bbox": [100, 100, 260, 130],
"zone": "display_zone", "style_family": "table_caption_like",
"marker_signature": {"type": "table_number"},
}
]
inventory = build_table_inventory_vnext(structured_blocks)
assert inventory["tables"] == []
assert inventory["held_tables"][0]["hold_reason"] == "insufficient_caption_evidence"
def test_build_table_inventory_vnext_materializes_split_caption_continuation() -> None:
from paperforge.worker.ocr_tables import build_table_inventory_vnext
structured_blocks = [
{"block_id": "cap1", "page": 5, "role": "table_caption", "text": "Table 3.", "bbox": [100, 100, 220, 130]},
{"block_id": "cap2", "page": 5, "role": "body_text", "text": "Continuation text", "bbox": [100, 132, 700, 170]},
{"block_id": "asset1", "page": 5, "role": "table_html", "raw_label": "table", "text": "<table></table>", "bbox": [100, 180, 700, 500]},
]
inventory = build_table_inventory_vnext(structured_blocks)
assert "cap2" in inventory["tables"][0]["consumed_block_ids"]
assert inventory["tables"][0]["caption_text"].startswith("Table 3.")
def test_build_table_inventory_vnext_previous_page_continuation_gets_geometry_elevation() -> None:
from paperforge.worker.ocr_tables import build_table_inventory_vnext
structured_blocks = [
{"block_id": "asset_prev", "page": 7, "role": "table_html", "raw_label": "table", "text": "<table></table>", "bbox": [100, 1200, 900, 1480]},
{"block_id": "cap1", "page": 8, "role": "table_caption", "text": "Table 4. (continued)", "bbox": [100, 110, 700, 150]},
]
inventory = build_table_inventory_vnext(structured_blocks)
assert inventory["tables"][0]["asset_block_id"] == "asset_prev"
assert "continuation_geometry_elevation" in inventory["tables"][0]["match_score"]["evidence"]
# ── Task 5: Notes, accounting, diff tooling ──
def test_build_table_inventory_vnext_collects_note_band_and_bridge_blocks() -> None:
from paperforge.worker.ocr_tables import build_table_inventory_vnext
structured_blocks = [
{"block_id": "cap1", "page": 5, "role": "table_caption", "text": "Table 1. Example", "bbox": [100, 100, 700, 140]},
{"block_id": "asset1", "page": 5, "role": "table_html", "raw_label": "table", "text": "<table></table>", "bbox": [100, 160, 700, 520]},
{"block_id": "note1", "page": 5, "role": "footnote", "text": "* p < 0.05", "bbox": [100, 530, 220, 555]},
{"block_id": "bridge1", "page": 5, "bridge_eligible": True, "layout_region": "display_zone", "text": "", "bbox": [100, 150, 700, 155]},
]
tables = build_table_inventory_vnext(structured_blocks)["tables"]
assert len(tables) == 1
assert tables[0]["note_block_ids"] == ["note1"]
assert tables[0]["bridge_block_ids"] == ["bridge1"]
assert "note1" in tables[0]["consumed_block_ids"]
def test_build_table_inventory_vnext_respects_page_footnote_prior() -> None:
from paperforge.worker.ocr_tables import build_table_inventory_vnext
structured_blocks = [
{"block_id": "cap1", "page": 5, "role": "table_caption", "text": "Table 1. Example", "bbox": [100, 100, 700, 140]},
{"block_id": "asset1", "page": 5, "role": "table_html", "raw_label": "table", "text": "<table></table>", "bbox": [100, 160, 700, 1150]},
{"block_id": "note1", "page": 5, "role": "footnote", "text": "* footer-area note", "bbox": [100, 1180, 300, 1210]},
]
table = build_table_inventory_vnext(structured_blocks)["tables"][0]
assert table["note_match_reason"] in {"note_band_geometry_match", "outside_vertical_range"}
def test_build_table_inventory_vnext_sets_render_rotation_fields_for_rotated_table_asset() -> None:
from paperforge.worker.ocr_tables import build_table_inventory_vnext
structured_blocks = [
{"block_id": "cap1", "page": 8, "role": "table_caption", "text": "Table 2. Caption", "bbox": [100, 134, 967, 1442]},
{
"block_id": "asset1", "page": 8, "role": "table_html", "raw_label": "table",
"text": "<table></table>", "bbox": [100, 100, 880, 1400],
"span_metadata": [{"dir": [0.0, -1.0], "wmode": 0}],
},
]
inventory = build_table_inventory_vnext(structured_blocks)
tables = inventory["tables"]
assert len(tables) >= 1
assert tables[0]["render_rotation_deg"] in {0, 270}
def test_compare_table_inventory_legacy_vs_vnext_smoke_fixture() -> None:
from pathlib import Path
from scripts.dev.compare_table_inventory_legacy_vs_vnext import compare_table_inventory_legacy_vs_vnext
result = compare_table_inventory_legacy_vs_vnext(Path("tests/fixtures/ocr_vnext_real_papers/2HEUD5P9"))
assert "legacy" in result
assert "vnext" in result
assert "diff" in result