src/endopath/rules.py

"""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),
    }