From 23e95a72fd4a75f94b397251d536d41320f0c428 Mon Sep 17 00:00:00 2001 From: Research Assistant Date: Fri, 12 Jun 2026 12:10:57 +0800 Subject: [PATCH] feat: add OCR structural gate safe defaults --- paperforge/worker/ocr_structural_gate.py | 107 +++++++++++++++++++++++ tests/test_ocr_structural_gate.py | 68 ++++++++++++++ 2 files changed, 175 insertions(+) create mode 100644 paperforge/worker/ocr_structural_gate.py create mode 100644 tests/test_ocr_structural_gate.py diff --git a/paperforge/worker/ocr_structural_gate.py b/paperforge/worker/ocr_structural_gate.py new file mode 100644 index 00000000..03c7b890 --- /dev/null +++ b/paperforge/worker/ocr_structural_gate.py @@ -0,0 +1,107 @@ +from __future__ import annotations + +from dataclasses import dataclass, field + + +VERIFY_REQUIRED = { + "paper_title", + "authors", + "abstract_heading", + "abstract_body", + "keywords", + "section_heading", + "subsection_heading", + "reference_heading", + "reference_item", + "figure_caption", + "table_caption", + "table_html", +} + + +@dataclass +class VerifiedRoleDecision: + role: str + status: str + source: str + evidence: list[str] = field(default_factory=list) + seed_role: str = "unknown_structural" + role_candidate: str | None = None + render_default: bool | None = None + + def as_block_fields(self) -> dict: + fields = { + "role": self.role, + "role_verification_status": self.status, + "role_source": self.source, + "role_evidence": list(self.evidence), + "seed_role": self.seed_role, + } + if self.role_candidate is not None: + fields["role_candidate"] = self.role_candidate + if self.render_default is not None: + fields["render_default"] = self.render_default + return fields + + +@dataclass +class RoleGateContext: + source_frontmatter_anchor_ids: dict[str, set[str | int]] = field(default_factory=dict) + abstract_span: dict | None = None + reference_zone: dict | None = None + accepted_heading_block_ids: set[str | int] = field(default_factory=set) + accepted_caption_block_ids: set[str | int] = field(default_factory=set) + accepted_table_block_ids: set[str | int] = field(default_factory=set) + + +def _bid(block: dict) -> str | int | None: + return block.get("block_id") + + +def accept_role(role: str, seed_role: str, source: str, evidence: list[str]) -> VerifiedRoleDecision: + return VerifiedRoleDecision(role=role, status="ACCEPT", source=source, evidence=evidence, seed_role=seed_role) + + +def hold_role(seed_role: str, reason: str) -> VerifiedRoleDecision: + return VerifiedRoleDecision( + role="unknown_structural", + status="HOLD", + source="structural_gate", + evidence=[reason], + seed_role=seed_role, + role_candidate=seed_role, + render_default=False, + ) + + +def resolve_verified_role(block: dict, context: RoleGateContext) -> VerifiedRoleDecision: + current_role = str(block.get("role") or "unassigned") + seed_role = str(block.get("seed_role") or current_role or "unknown_structural") + proposal = seed_role if current_role in {"", "unassigned"} else current_role + block_id = _bid(block) + + if current_role not in {"", "unassigned"} and current_role not in VERIFY_REQUIRED: + return accept_role( + current_role, + seed_role, + "non_structural_normalized_role", + ["pre-gate normalized non-structural role preserved"], + ) + + if proposal == "paper_title" or seed_role == "paper_title": + if block_id in context.source_frontmatter_anchor_ids.get("title", set()): + return accept_role("paper_title", seed_role, "source_frontmatter_title_anchor", ["matched source title anchor"]) + return hold_role(seed_role, "paper title seed lacks source-backed title anchor") + if proposal == "authors" or seed_role == "authors": + if block_id in context.source_frontmatter_anchor_ids.get("authors", set()): + return accept_role("authors", seed_role, "source_frontmatter_authors_anchor", ["matched source authors anchor"]) + return hold_role(seed_role, "authors seed lacks source-backed authors anchor") + if proposal == "keywords" or seed_role == "keywords": + if block_id in set((context.abstract_span or {}).get("keyword_block_ids", [])): + return accept_role("keywords", seed_role, "abstract_span_keyword_boundary", ["keywords follow accepted abstract span"]) + return hold_role(seed_role, "keywords seed lacks abstract-span keyword boundary") + + if proposal in VERIFY_REQUIRED or seed_role in VERIFY_REQUIRED: + return hold_role(seed_role, f"{proposal} requires structural verifier") + source = "non_structural_seed" if current_role in {"", "unassigned"} else "non_structural_normalized_role" + return accept_role(proposal, seed_role, source, ["non-structural role accepted"]) diff --git a/tests/test_ocr_structural_gate.py b/tests/test_ocr_structural_gate.py new file mode 100644 index 00000000..d18d6988 --- /dev/null +++ b/tests/test_ocr_structural_gate.py @@ -0,0 +1,68 @@ +from __future__ import annotations + + +def test_required_seed_without_verifier_holds_not_accepts() -> None: + from paperforge.worker.ocr_structural_gate import RoleGateContext, resolve_verified_role + + block = {"block_id": "a1", "seed_role": "authors", "text": "Author One Author Two"} + + decision = resolve_verified_role(block, RoleGateContext()) + + assert decision.role == "unknown_structural" + assert decision.status == "HOLD" + assert decision.seed_role == "authors" + assert decision.source == "structural_gate" + + +def test_non_required_seed_can_accept_as_non_structural() -> None: + from paperforge.worker.ocr_structural_gate import RoleGateContext, resolve_verified_role + + block = {"block_id": "b1", "seed_role": "body_paragraph", "text": "Regular body paragraph."} + + decision = resolve_verified_role(block, RoleGateContext()) + + assert decision.role == "body_paragraph" + assert decision.status == "ACCEPT" + assert decision.source == "non_structural_seed" + + +def test_hold_preserves_candidate_and_hides_from_render_by_default() -> None: + from paperforge.worker.ocr_structural_gate import RoleGateContext, resolve_verified_role + + block = {"block_id": "r1", "seed_role": "reference_item", "text": "[1] Not actually in references."} + + decision = resolve_verified_role(block, RoleGateContext()) + fields = decision.as_block_fields() + + assert fields["role"] == "unknown_structural" + assert fields["role_candidate"] == "reference_item" + assert fields["role_verification_status"] == "HOLD" + assert fields["render_default"] is False + + +def test_gate_preserves_pre_normalized_non_structural_role() -> None: + from paperforge.worker.ocr_structural_gate import RoleGateContext, resolve_verified_role + + block = {"block_id": "i1", "role": "structured_insert", "seed_role": "body_paragraph", "text": "Side note."} + + decision = resolve_verified_role(block, RoleGateContext()) + + assert decision.role == "structured_insert" + assert decision.seed_role == "body_paragraph" + assert decision.source == "non_structural_normalized_role" + + +def test_gate_preserves_pre_normalized_safe_role_even_if_seed_is_required() -> None: + from paperforge.worker.ocr_structural_gate import RoleGateContext, resolve_verified_role + + block = { + "block_id": "x", + "role": "frontmatter_noise", + "seed_role": "authors", + "text": "Author-like sidebar", + } + + decision = resolve_verified_role(block, RoleGateContext()) + + assert decision.role == "frontmatter_noise" + assert decision.source == "non_structural_normalized_role"