"""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\d{1,3}(?:\.\d+)?)\s*%\s+(?:of\s+(?:the\s+)?)?myometri", re.IGNORECASE), re.compile(r"myometri\w*[^.%]{0,60}?(?P\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\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), }