File size: 5,252 Bytes
eea689d | 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 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 | """Deterministic rule-based extractors for the CAP/ICCR fields that don't
require narrative interpretation (docs/prd.md section 7: "Deterministic rules
handle unambiguous fields, identifiers, dates, specimen type.").
Only myometrial invasion percentage is currently a modeled CAP/ICCR
checklist field (see schema.py) that a regex can reliably populate; report
date is a Case-level field. Specimen type isn't part of the nine-to-
fourteen-field CAP/ICCR checklist itself (it's report header metadata), so
it's returned as an auxiliary, evidence-backed finding rather than written
into the checklist.
Deliberately NOT attempted here, and why: histologic grade, margin status,
LVSI, cervical stromal invasion, regional lymph node status, and molecular
classification all require interpreting free narrative language (e.g.
distinguishing "no lymphovascular invasion identified" from "focal
lymphovascular invasion present" from a dozen real phrasings across this
corpus) -- routed to the LLM extraction issue instead, per the hybrid
architecture in section 6.
Every extractor here returns fields at NEEDS_REVIEW, never auto-confirmed: a
person confirms every value before it counts as data (docs/prd.md section 8.3).
"""
from __future__ import annotations
import dataclasses
import re
from datetime import date
from typing import Optional
import pandas as pd
from endopath.schema import ChecklistField, EvidenceSpan, FieldStatus
def _make_evidence(text: str, start: int, end: int, context: int = 40) -> EvidenceSpan:
quote_start = max(0, start - context)
quote_end = min(len(text), end + context)
return EvidenceSpan(
quote=text[quote_start:quote_end].strip(),
char_start=start,
char_end=end,
source="text",
)
# --- Myometrial invasion percent (a real EndometrialChecklist field) -------
_MYOMETRIAL_PERCENT_PATTERNS = [
re.compile(r"(?P<pct>\d{1,3}(?:\.\d+)?)\s*%\s+(?:of\s+(?:the\s+)?)?myometri", re.IGNORECASE),
re.compile(r"myometri\w*[^.%]{0,60}?(?P<pct>\d{1,3}(?:\.\d+)?)\s*%", re.IGNORECASE),
]
def extract_myometrial_invasion_percent(text: str) -> Optional[ChecklistField]:
for pattern in _MYOMETRIAL_PERCENT_PATTERNS:
match = pattern.search(text)
if not match:
continue
pct = float(match.group("pct"))
if not (0.0 <= pct <= 100.0):
continue
start, end = match.span()
return ChecklistField(
value=pct,
confidence=0.9,
evidence=_make_evidence(text, start, end),
status=FieldStatus.NEEDS_REVIEW,
)
return None
# --- Report date (Case-level metadata, best-effort) -------------------------
_DATE_CONTEXT_PATTERN = re.compile(
r"(?:DATE OF (?:RECEIPT|SERVICE|COLLECTION|REPORT)|COLLECTED|RECEIVED|SIGNED OUT|REPORTED)"
r"\s*:?\s*(?P<date_str>\d{1,2}[/-]\d{1,2}[/-]\d{2,4}|[A-Za-z]+\.?\s+\d{1,2},?\s+\d{4})",
re.IGNORECASE,
)
@dataclasses.dataclass
class RuleFinding:
value: object
evidence: EvidenceSpan
confidence: float
def extract_report_date(text: str) -> Optional[RuleFinding]:
match = _DATE_CONTEXT_PATTERN.search(text)
if not match:
return None
parsed = pd.to_datetime(match.group("date_str"), errors="coerce")
if pd.isna(parsed):
return None
start, end = match.span()
return RuleFinding(
value=parsed.date(),
evidence=_make_evidence(text, start, end),
confidence=0.6, # context-keyword match, not a structured field -- lower confidence than the regex-only myometrial-percent extractor
)
# --- Specimen type (report metadata, not a CAP/ICCR checklist field) -------
_SPECIMEN_TYPE_VOCAB = [
"total laparoscopic hysterectomy",
"total abdominal hysterectomy",
"radical hysterectomy",
"supracervical hysterectomy",
"total hysterectomy",
"endometrial biopsy",
"dilation and curettage",
"omentectomy",
"salpingo-oophorectomy",
]
def extract_specimen_type(text: str) -> Optional[RuleFinding]:
lowered = text.lower()
best: Optional[tuple[int, int, str]] = None
for term in _SPECIMEN_TYPE_VOCAB:
idx = lowered.find(term)
if idx == -1:
continue
# Prefer the longest (most specific) vocabulary term that matches,
# not just the first one found in the text.
if best is None or len(term) > len(best[2]):
best = (idx, idx + len(term), term)
if best is None:
return None
start, end, term = best
return RuleFinding(
value=term,
evidence=_make_evidence(text, start, end),
confidence=0.85,
)
def apply_rule_based_extraction(text: str) -> dict:
"""Run all deterministic extractors over one report's text.
Returns a dict with a `checklist` sub-dict (fields ready to assign onto
an EndometrialChecklist), a `report_date` finding for the Case, and a
`specimen_type` finding kept separate since it has no schema slot yet.
"""
return {
"checklist": {
"myometrial_invasion_percent": extract_myometrial_invasion_percent(text),
},
"report_date": extract_report_date(text),
"specimen_type": extract_specimen_type(text),
}
|