File size: 2,854 Bytes
ab5ea78 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 | """The frequency-sorted review queue — the pipeline's actual product.
Ordering is the product decision here, and it targets the bottleneck directly:
the expert is the scarce resource, so they should hit the terms whose definition
propagates furthest first. Conflicts are promoted above frequency regardless,
because a contradiction is a decision only they can make.
Each row carries page, section and the verbatim span so review is a matter of
checking a quote against a page, not a claim against memory. That is what makes
the queue finishable.
"""
from __future__ import annotations
def build_queue(entries: list[dict]) -> list[dict]:
def sort_key(entry: dict):
conflicting = (
entry.get("diff_status") == "conflicting"
or entry.get("definition_conflict") is True
)
return (0 if conflicting else 1, -int(entry.get("mention_count", 0) or 0))
queue = []
for rank, entry in enumerate(sorted(entries, key=sort_key), start=1):
provenance = entry.get("provenance") or {}
queue.append(
{
"rank": rank,
"term": entry.get("term"),
"definition": entry.get("definition"),
"source_wording": entry.get("source_wording"),
"mention_count": entry.get("mention_count", 0),
"extraction_status": entry.get("extraction_status"),
"diff_status": entry.get("diff_status"),
"definition_conflict": entry.get("definition_conflict", False),
"conflict_variants": entry.get("conflict_variants", []),
"page": provenance.get("page"),
"section_no": provenance.get("section_no"),
"span": provenance.get("span"),
"review_reason": _reason(entry),
}
)
return queue
def _reason(entry: dict) -> str:
if entry.get("definition_conflict") or entry.get("diff_status") == "conflicting":
return "conflicting definitions — expert decision required"
if _wording_differs(entry):
return "source wording differs from the expanded name — confirm which is correct"
if entry.get("extraction_status") == "no_definition_found":
return "term found but no definition in document"
if not entry.get("definition"):
return "definition rejected by span check or absent"
return "routine confirmation"
def _wording_differs(entry: dict) -> bool:
"""The document says "Physical of Availability"; the expansion says
"Physical Availability". Surfacing that to the expert is a locked
requirement, so it earns its own review reason."""
source = (entry.get("source_wording") or "").strip().casefold()
full = (entry.get("full_name") or "").strip().casefold()
return bool(source and full) and full not in source
|