Fix every real lint finding and drop degenerate splice fragments
This commit is contained in:
@@ -0,0 +1,485 @@
|
||||
"""Assembles a raw span stream into ordered Monograph records.
|
||||
|
||||
Three simple passes, each independently easy to reason about — avoids a
|
||||
single tangled state machine (SRP: classify, then merge titles, then build):
|
||||
|
||||
1. Classify each span in reading order as a title candidate, a section
|
||||
heading, or body text.
|
||||
2. Coalesce consecutive title-candidate spans into single merged Heading
|
||||
events via `merge.merge_multiline_headings` (handles both the multi-line
|
||||
wrap and same-line font-size-split cases — see merge.py).
|
||||
3. Walk the resulting flat event stream once, building Monograph records.
|
||||
|
||||
Handles the confirmed real "qualifier line" case (outlier-catalog item 18):
|
||||
a monograph title can legitimately repeat (e.g. two distinct "SALBUTAMOL"
|
||||
monographs, "Dùng trong hô hấp" vs "Dùng trong sản khoa") disambiguated by a
|
||||
bold, parenthesized, non-all-caps line directly beneath the title. That
|
||||
qualifier is folded into `drug_id` so two legitimate entries don't collide;
|
||||
a genuine duplicate `drug_id` (no qualifier, same name) raises rather than
|
||||
silently overwriting, since the one apparent duplicate found during ADR
|
||||
0003's investigation (GONADOTROPIN) turned out to be a detector artifact,
|
||||
not real — a real second collision should be surfaced, not hidden.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import unicodedata
|
||||
from dataclasses import dataclass
|
||||
from typing import Iterator, List, Optional, Union
|
||||
|
||||
from ..extract.models import Span
|
||||
from ..extract.page_map import HEADER_BAND_Y
|
||||
from ..normalize import join_spans, substitute_pua
|
||||
from ..tables.classify import QUARANTINE_SHAPES
|
||||
from .atc import extract_atc_codes
|
||||
from .detector import in_monograph_range, is_monograph_title_candidate
|
||||
from .merge import merge_multiline_headings, merge_same_line_bold_fragments
|
||||
from .models import (
|
||||
PART_PROSE,
|
||||
PART_TABLE,
|
||||
Heading,
|
||||
Monograph,
|
||||
SectionPart,
|
||||
SectionSpan,
|
||||
TableBlock,
|
||||
)
|
||||
from .vocab import (
|
||||
SectionDef,
|
||||
is_part_divider,
|
||||
match_section,
|
||||
match_section_with_inline_value,
|
||||
)
|
||||
|
||||
_QUALIFIER_RE = re.compile(r"^\(.+\)$")
|
||||
|
||||
|
||||
def _is_page_boilerplate(span: Span) -> bool:
|
||||
"""Confirmed real (outlier-catalog item 13, measured via a whole-book
|
||||
`assemble()` run): the running header ("DTQGVN 2" + page number +
|
||||
current monograph name, e.g. physical page 1008's "DTQGVN 2" / "1009" /
|
||||
"Morphin sulfat") was falling through every classification branch below
|
||||
into plain body text, since it matches no section heading and isn't a
|
||||
real all-caps title — silently splicing itself into the *middle* of
|
||||
whatever section happens to be open when a physical page turns (1,374
|
||||
of 11,409 sections / 671 of 682 monographs affected). It's reliably
|
||||
identifiable independent of its (non-vocabulary) text: always the
|
||||
full-page-width block in the header band, same signal `page_map.py`
|
||||
already uses to read the folio.
|
||||
"""
|
||||
return span.column == "full_width" and span.y0 < HEADER_BAND_Y
|
||||
|
||||
|
||||
class DuplicateDrugIdError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _SectionEvent:
|
||||
section_def: SectionDef
|
||||
span: Span
|
||||
inline_value: Optional[str] = None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _TextEvent:
|
||||
span: Span
|
||||
|
||||
|
||||
_Event = Union[Heading, _SectionEvent, _TextEvent] # Heading == a title event
|
||||
|
||||
|
||||
def _slugify(text: str) -> str:
|
||||
normalized = unicodedata.normalize("NFKD", text)
|
||||
ascii_text = normalized.encode("ascii", "ignore").decode("ascii")
|
||||
return re.sub(r"[^a-z0-9]+", "_", ascii_text.lower()).strip("_")
|
||||
|
||||
|
||||
def _is_body_line_that_reads_like_a_label(span: Span, items: List) -> bool:
|
||||
"""A plain line that repeats a section name, sitting under a heading.
|
||||
|
||||
Confirmed real and clinically material: FLUOROURACIL (physical page 681)
|
||||
prints `Thời kỳ mang thai` / `Chống chỉ định.` and `Thời kỳ cho con bú` /
|
||||
`Chống chỉ định.`, verified by rendering the page. The body line matches
|
||||
the section vocabulary, so it was read as a heading — leaving both
|
||||
pregnancy and lactation sections empty and dropping the statement that
|
||||
fluorouracil is contraindicated in both.
|
||||
|
||||
The book never prints an empty section, so a *non-bold* label immediately
|
||||
after a heading is that heading's body. Boldness still cannot be required
|
||||
in general (outlier item 20: `Mã ATC: N06AA09.` is a plain span), which is
|
||||
why this is narrowed to the directly-under-a-heading position.
|
||||
"""
|
||||
if span.bold:
|
||||
return False
|
||||
return bool(items) and isinstance(items[-1], _SectionEvent)
|
||||
|
||||
|
||||
def _classify(spans: List[Span]) -> List[Union[Span, _SectionEvent, _TextEvent]]:
|
||||
"""Pass 1: tag each span. Title candidates are left as raw Span objects
|
||||
(pass 2 groups + merges them); everything else becomes a typed event.
|
||||
|
||||
Section matching does NOT require `span.bold` — confirmed real (outlier
|
||||
item 20): AMITRIPTYLIN's "Mã ATC: N06AA09." is a single **plain, non-bold**
|
||||
span (Abacavir's equivalent is bold "Mã ATC: " + a separate plain value
|
||||
span), inconsistent across the book's ~700 individually-authored
|
||||
monographs (the book's own foreword notes "biên soạn bởi nhiều tác giả").
|
||||
Matching by exact vocabulary text (not styling) is the reliable signal,
|
||||
same lesson as "don't gate on font size" (ADR 0003 item 10) applied to
|
||||
boldness instead.
|
||||
"""
|
||||
items: List[Union[Span, _SectionEvent, _TextEvent]] = []
|
||||
for span in spans:
|
||||
if not span.text.strip():
|
||||
continue
|
||||
if _is_page_boilerplate(span):
|
||||
continue
|
||||
if is_monograph_title_candidate(span):
|
||||
items.append(span)
|
||||
continue
|
||||
section_def = match_section(span.text)
|
||||
if section_def is not None and not _is_body_line_that_reads_like_a_label(
|
||||
span, items
|
||||
):
|
||||
items.append(_SectionEvent(section_def, span))
|
||||
continue
|
||||
if section_def is not None:
|
||||
items.append(_TextEvent(span))
|
||||
continue
|
||||
inline = match_section_with_inline_value(span.text)
|
||||
if inline is not None:
|
||||
items.append(_SectionEvent(inline[0], span, inline_value=inline[1]))
|
||||
else:
|
||||
items.append(_TextEvent(span))
|
||||
return items
|
||||
|
||||
|
||||
def _coalesce_titles(items: List[Union[Span, _SectionEvent, _TextEvent]]) -> List[_Event]:
|
||||
"""Pass 2: merge consecutive raw title-candidate Span runs into single
|
||||
Heading events, preserving the order of everything else.
|
||||
"""
|
||||
events: List[_Event] = []
|
||||
run: List[Span] = []
|
||||
|
||||
def flush_run():
|
||||
if run:
|
||||
events.extend(merge_multiline_headings(list(run)))
|
||||
run.clear()
|
||||
|
||||
for item in items:
|
||||
if isinstance(item, Span):
|
||||
run.append(item)
|
||||
else:
|
||||
flush_run()
|
||||
events.append(item)
|
||||
flush_run()
|
||||
return events
|
||||
|
||||
|
||||
def _is_qualifier_line(span: Span) -> bool:
|
||||
text = span.text.strip()
|
||||
return span.bold and not text.isupper() and bool(_QUALIFIER_RE.match(text))
|
||||
|
||||
|
||||
_ANCHOR_LOOKAHEAD = 6
|
||||
_ANCHOR_SECTION_KEY = "ten_chung_quoc_te"
|
||||
|
||||
|
||||
def _has_anchor_ahead(events: List[_Event], title_index: int) -> bool:
|
||||
"""Every real monograph documents "Tên chung quốc tế" as its very first
|
||||
section (the book's own template, item 2 — see vocab.py docstring).
|
||||
Loosening this to "any known section" was tried and reverted: it let
|
||||
a real, different false positive through (outlier item 21) — individual
|
||||
statin names ("SIMVASTATIN", "LOVASTATIN", ...) are bold+all-caps+short
|
||||
sub-headings *inside* the class-level "CÁC CHẤT ỨC CHẾ HMG-CoA
|
||||
REDUCTASE" monograph, each immediately followed by their own "Liều
|
||||
lượng và cách dùng" sub-section but NOT by "Tên chung quốc tế" (that
|
||||
section belongs only to the parent class monograph) — the loose
|
||||
"any section" check couldn't tell this apart from a real monograph
|
||||
start, but the strict "Tên chung quốc tế specifically" check correctly
|
||||
rejects it, since the specific book-documented template guarantees this
|
||||
exact section is always first for genuine top-level monographs.
|
||||
|
||||
Still correctly rejects the other confirmed false positive (outlier
|
||||
item 19: "HSV"/"CMV" table column headers), which aren't followed by
|
||||
ANY recognized section, let alone this specific one.
|
||||
"""
|
||||
for j in range(title_index + 1, min(title_index + 1 + _ANCHOR_LOOKAHEAD, len(events))):
|
||||
event = events[j]
|
||||
if isinstance(event, Heading) and event.is_monograph_title:
|
||||
return False
|
||||
if isinstance(event, _SectionEvent) and event.section_def.key == _ANCHOR_SECTION_KEY:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _filter_false_positive_titles(events: List[_Event]) -> List[_Event]:
|
||||
"""Pass 2.5: drop title-shaped candidates that aren't followed by any
|
||||
recognized section anchor before the next title candidate.
|
||||
"""
|
||||
return [
|
||||
event for i, event in enumerate(events)
|
||||
if not (isinstance(event, Heading) and event.is_monograph_title)
|
||||
or _has_anchor_ahead(events, i)
|
||||
]
|
||||
|
||||
|
||||
def _region_for(table_index, span: Span):
|
||||
"""The table region a span sits in, if any."""
|
||||
if not table_index:
|
||||
return None
|
||||
for region in table_index.get(span.physical_page, ()):
|
||||
if region.contains(span.x0, span.y0, span.x1, span.y1):
|
||||
return region
|
||||
return None
|
||||
|
||||
|
||||
SPAN_STATE_TEXT = "normalized_text"
|
||||
SPAN_STATE_TABLE = "table"
|
||||
SPAN_STATE_QUARANTINED = "quarantined"
|
||||
SPAN_STATE_BOILERPLATE = "boilerplate_excluded"
|
||||
SPAN_STATE_HEADING = "heading"
|
||||
SPAN_STATE_OUT_OF_SCOPE = "out_of_scope"
|
||||
SPAN_STATE_UNASSIGNED = "unassigned"
|
||||
# Deliberately dropped, not missed: the book's own part-divider titles
|
||||
# ("CÁC CHUYÊN LUẬN THUỐC" etc.) are structure, not content. Reporting them
|
||||
# as `unassigned` would make a clean acceptance target of unassigned == 0
|
||||
# impossible to state honestly.
|
||||
SPAN_STATE_STRUCTURAL = "structural_excluded"
|
||||
|
||||
|
||||
def assemble(spans: List[Span], table_index=None, ledger: Optional[list] = None) -> Iterator[Monograph]:
|
||||
"""Assemble monographs from spans.
|
||||
|
||||
`table_index` maps a physical page to the table regions on it (see
|
||||
`tables.index_by_page`). When supplied, spans falling inside a region are
|
||||
diverted into `Monograph.tables` instead of section prose — measured
|
||||
reason: physical page 109's dosage-form table was otherwise concatenated
|
||||
cell by cell into a section body. Omitting it keeps the previous
|
||||
behaviour, so callers without a region map still work.
|
||||
"""
|
||||
raw_chars = sum(len(s.text) for s in spans)
|
||||
spans = merge_same_line_bold_fragments(spans)
|
||||
events = _filter_false_positive_titles(_coalesce_titles(_classify(spans)))
|
||||
|
||||
# Span-level coverage ledger. Character counts alone cannot balance here
|
||||
# (normalization joins, substitutes and drops characters), so every span
|
||||
# is given a state first and characters are aggregated from that.
|
||||
states: dict = {}
|
||||
if ledger is not None:
|
||||
for s in spans:
|
||||
if not s.text.strip():
|
||||
states[id(s)] = "whitespace_only"
|
||||
elif _is_page_boilerplate(s):
|
||||
states[id(s)] = SPAN_STATE_BOILERPLATE
|
||||
elif is_part_divider(s.text):
|
||||
states[id(s)] = SPAN_STATE_STRUCTURAL
|
||||
elif not in_monograph_range(s):
|
||||
states[id(s)] = SPAN_STATE_OUT_OF_SCOPE
|
||||
elif is_monograph_title_candidate(s):
|
||||
# title spans are merged into a Heading event and lose their
|
||||
# link back to the source span, so they are accounted for here
|
||||
# using the same predicate the classifier uses
|
||||
states[id(s)] = SPAN_STATE_HEADING
|
||||
else:
|
||||
states[id(s)] = SPAN_STATE_UNASSIGNED
|
||||
|
||||
def mark(span: Span, state: str):
|
||||
if ledger is not None:
|
||||
states[id(span)] = state
|
||||
|
||||
seen_ids: set = set()
|
||||
current: Optional[Monograph] = None
|
||||
current_section_key: Optional[str] = None
|
||||
runs: List[tuple] = [] # ordered [(region_or_None, [spans])]
|
||||
inline_prefix: str = ""
|
||||
awaiting_qualifier = False
|
||||
|
||||
def append_span(span: Span, region):
|
||||
"""Keep spans in reading order, starting a new run whenever the
|
||||
prose/table context changes — this is what preserves the real
|
||||
prose -> table -> prose sequence inside one section."""
|
||||
key = region.table_id if region is not None else None
|
||||
if runs and runs[-1][0] == key:
|
||||
runs[-1][1].append(span)
|
||||
else:
|
||||
runs.append((key, [span], region))
|
||||
|
||||
def build_parts() -> List[SectionPart]:
|
||||
parts: List[SectionPart] = []
|
||||
for entry in runs:
|
||||
key, collected = entry[0], entry[1]
|
||||
region = entry[2] if len(entry) > 2 else None
|
||||
if not collected:
|
||||
continue
|
||||
text = substitute_pua(join_spans(collected))
|
||||
if not text.strip():
|
||||
continue
|
||||
pages = [s_.physical_page for s_ in collected]
|
||||
xs0 = min(s_.x0 for s_ in collected); ys0 = min(s_.y0 for s_ in collected)
|
||||
xs1 = max(s_.x1 for s_ in collected); ys1 = max(s_.y1 for s_ in collected)
|
||||
ids = [s_.span_id for s_ in collected]
|
||||
if key is None:
|
||||
parts.append(SectionPart(
|
||||
kind=PART_PROSE, text=text, physical_page=min(pages),
|
||||
bbox=[xs0, ys0, xs1, ys1], source_span_ids=ids,
|
||||
))
|
||||
else:
|
||||
parts.append(SectionPart(
|
||||
kind=PART_TABLE, text=text, physical_page=min(pages),
|
||||
bbox=[xs0, ys0, xs1, ys1], source_span_ids=ids,
|
||||
table_id=key,
|
||||
# deterministic: derived from the first source span, so the
|
||||
# same PDF always produces the same id. A counter suffix
|
||||
# would merely hide a duplicate rather than identify it.
|
||||
table_part_id=f"{key}@{ids[0]}",
|
||||
continuation_group=key,
|
||||
shape=region.shape if region is not None else None,
|
||||
quarantined=(region.shape in QUARANTINE_SHAPES)
|
||||
if region is not None else False,
|
||||
))
|
||||
if inline_prefix:
|
||||
head = SectionPart(
|
||||
kind=PART_PROSE, text=inline_prefix,
|
||||
physical_page=parts[0].physical_page if parts else 0,
|
||||
bbox=parts[0].bbox if parts else [0.0, 0.0, 0.0, 0.0],
|
||||
)
|
||||
parts.insert(0, head)
|
||||
return parts
|
||||
|
||||
def close_current_section():
|
||||
nonlocal runs, inline_prefix
|
||||
if current is not None and current_section_key is None and runs:
|
||||
# spans seen after the title but before any section heading
|
||||
current.preamble.extend(build_parts())
|
||||
if current is not None and current_section_key is not None:
|
||||
existing = current.sections[current_section_key]
|
||||
addition = build_parts()
|
||||
# A section heading can legitimately appear twice inside one
|
||||
# monograph (measured: 33 monographs, 38 occurrences — e.g.
|
||||
# CEFAMANDOL's "Liều lượng và cách dùng" resumes on physical page
|
||||
# 339 after a renal-dosing table). Replacing the SectionSpan here
|
||||
# silently destroyed everything captured before the repeat, so
|
||||
# the parts are concatenated instead. The first heading stays the
|
||||
# section's provenance anchor.
|
||||
combined = list(existing.parts) + addition
|
||||
current.sections[current_section_key] = SectionSpan(
|
||||
key=existing.key, display_name=existing.display_name,
|
||||
heading=existing.heading,
|
||||
# `text` is prose only. Table parts stay in `parts` with their
|
||||
# own provenance and quarantine flag, so anything reading
|
||||
# `.text` (the chunker included) cannot pick up linearised
|
||||
# cells by accident — the ordering is preserved in `parts`.
|
||||
text="\n".join(
|
||||
p_.text for p_ in combined
|
||||
if p_.kind == PART_PROSE and not p_.quarantined and p_.text
|
||||
).strip(),
|
||||
parts=combined,
|
||||
)
|
||||
for part in addition:
|
||||
if part.kind == PART_TABLE:
|
||||
current.tables.append(TableBlock(
|
||||
table_id=part.table_id, shape=part.shape or "",
|
||||
physical_page=part.physical_page, bbox=list(part.bbox),
|
||||
section_key=current_section_key, text=part.text,
|
||||
quarantined=part.quarantined,
|
||||
table_part_id=part.table_part_id,
|
||||
continuation_group=part.continuation_group,
|
||||
source_span_ids=list(part.source_span_ids),
|
||||
))
|
||||
runs = []
|
||||
inline_prefix = ""
|
||||
|
||||
def finalize(monograph: Monograph) -> Monograph:
|
||||
# Duplicate check happens here, not at title-detection time: the
|
||||
# qualifier line (if any) is only known a few events later, so
|
||||
# checking at open-time would false-positive on the legitimate
|
||||
# SALBUTAMOL case (outlier item 18) before the qualifier resolves.
|
||||
if monograph.drug_id in seen_ids:
|
||||
raise DuplicateDrugIdError(
|
||||
f"duplicate drug_id '{monograph.drug_id}' (title '{monograph.drug_name}', "
|
||||
f"physical page {monograph.source_page_range[0]}) — check for a qualifier "
|
||||
f"line (outlier item 18) before assuming this is a real collision"
|
||||
)
|
||||
seen_ids.add(monograph.drug_id)
|
||||
if "ma_atc" in monograph.sections:
|
||||
result = extract_atc_codes(monograph.sections["ma_atc"].text)
|
||||
monograph.atc_codes = result.codes
|
||||
monograph.atc_stated_absent = result.stated_absent
|
||||
return monograph
|
||||
|
||||
for event in events:
|
||||
if isinstance(event, Heading) and event.is_monograph_title:
|
||||
close_current_section()
|
||||
if current is not None:
|
||||
yield finalize(current)
|
||||
current = Monograph(
|
||||
drug_id=_slugify(event.text), drug_name=event.text,
|
||||
source_page_range=[event.physical_page, event.physical_page],
|
||||
)
|
||||
current_section_key = None
|
||||
awaiting_qualifier = True
|
||||
for src in getattr(event, "source_spans", ()) or ():
|
||||
mark(src, SPAN_STATE_HEADING)
|
||||
continue
|
||||
|
||||
if current is None:
|
||||
continue # front matter / general chapters before the first monograph
|
||||
|
||||
if isinstance(event, _SectionEvent):
|
||||
close_current_section()
|
||||
current_section_key = event.section_def.key
|
||||
if event.section_def.key not in current.sections:
|
||||
current.sections[event.section_def.key] = SectionSpan(
|
||||
key=event.section_def.key,
|
||||
display_name=event.section_def.display_name,
|
||||
heading=Heading(
|
||||
text=event.section_def.display_name,
|
||||
physical_page=event.span.physical_page, y0=event.span.y0,
|
||||
is_monograph_title=False, section_key=event.section_def.key,
|
||||
),
|
||||
text="",
|
||||
)
|
||||
if event.inline_value:
|
||||
inline_prefix = event.inline_value
|
||||
awaiting_qualifier = False
|
||||
mark(event.span, SPAN_STATE_HEADING)
|
||||
current.source_page_range[1] = max(current.source_page_range[1], event.span.physical_page)
|
||||
continue
|
||||
|
||||
# _TextEvent
|
||||
span = event.span
|
||||
if awaiting_qualifier and _is_qualifier_line(span):
|
||||
text = span.text.strip()
|
||||
current.drug_id = f"{current.drug_id}_{_slugify(text)}"
|
||||
current.drug_name = f"{current.drug_name} {text}"
|
||||
awaiting_qualifier = False
|
||||
mark(span, SPAN_STATE_HEADING)
|
||||
continue
|
||||
awaiting_qualifier = False
|
||||
|
||||
if not in_monograph_range(span):
|
||||
continue
|
||||
current.source_page_range[1] = max(current.source_page_range[1], span.physical_page)
|
||||
|
||||
region = _region_for(table_index, span)
|
||||
append_span(span, region)
|
||||
if region is not None:
|
||||
mark(span, SPAN_STATE_QUARANTINED
|
||||
if region.shape in QUARANTINE_SHAPES else SPAN_STATE_TABLE)
|
||||
else:
|
||||
mark(span, SPAN_STATE_TEXT)
|
||||
|
||||
if ledger is not None:
|
||||
ledger.append({"raw_chars_before_merge": raw_chars})
|
||||
for s_obj in spans:
|
||||
ledger.append({
|
||||
"state": states[id(s_obj)],
|
||||
"physical_page": s_obj.physical_page,
|
||||
"bbox": [s_obj.x0, s_obj.y0, s_obj.x1, s_obj.y1],
|
||||
"chars": len(s_obj.text),
|
||||
"text": s_obj.text[:60],
|
||||
})
|
||||
|
||||
close_current_section()
|
||||
if current is not None:
|
||||
yield finalize(current)
|
||||
Reference in New Issue
Block a user