""" Temporal Preprocessor for Clinical Intent Classification. Rule-based preprocessing layer that runs BEFORE the ML model to handle temporal context that BERT struggles with: 1. BUT-clause reversals: "I felt better but now the pain is back" → Extracts the active complaint, discards the resolved prefix. 2. Proxy/third-party urgency: "My mom says her chest hurts really bad" → Detects active urgency even when reported through a third party. 3. Still-active markers after past tense: "I had pain and it's getting worse" → Flags as active despite past-tense opener. Returns: TemporalSignal with: - processed_text: cleaned text for model input - is_active_emergency: True if active urgency detected despite temporal markers - temporal_tag: "active" | "resolved" | "mixed" | "neutral" - confidence: how confident the preprocessor is in its tag - reason: human-readable explanation """ import re from dataclasses import dataclass, field from typing import Optional, List @dataclass class TemporalSignal: """Result of temporal preprocessing.""" processed_text: str # Text after preprocessing (may be modified) original_text: str # Original unmodified text is_active_emergency: bool # True if active urgency detected temporal_tag: str # "active" | "resolved" | "mixed" | "neutral" confidence: float # 0.0-1.0 confidence in the tag reason: Optional[str] = None # Human-readable explanation proxy_report: bool = False # True if someone reporting for another person matched_rules: List[str] = field(default_factory=list) # ── BUT-clause reversal patterns ── # These indicate the speaker HAD a resolved condition but it's NOW active again BUT_CLAUSE_PATTERNS = [ # "felt better but now..." re.compile( r"(?P.+?)\b(?:but|however|except|although|though|yet)\b\s+" r"(?P(?:now|today|currently|right now|at this moment|this morning|tonight|again)" r".+)", re.IGNORECASE ), # "was fine but started..." re.compile( r"(?P.+?)\b(?:but|however|except)\b\s+" r"(?P(?:it'?s?|the pain|the symptoms?|i'?m|i am|i feel|i have|it has|they)" r"\s+(?:back|returned|worse|coming back|started|acting up|flaring).+)", re.IGNORECASE ), # "...but now I can't..." re.compile( r"(?P.+?)\b(?:but|however|except)\b\s+" r"(?P(?:now\s+)?i\s+(?:can'?t|cannot|am having|have|feel|'m).+)", re.IGNORECASE ), # "...but then I started / but woke up with / but noticed..." re.compile( r"(?P.+?)\b(?:but|however|except)\b\s+" r"(?P(?:then|woke up|noticed|started|began|suddenly|this morning)\b.+)", re.IGNORECASE ), ] # ── General contrastive marker ── # Detects ANY contrastive clause to split and re-evaluate CONTRASTIVE_MARKER = re.compile( r"\b(?:but|however|yet|still|though|although|except)\b", re.IGNORECASE ) # ── Still-active markers ── # If these appear AFTER past-tense language, the condition is STILL active STILL_ACTIVE_PATTERNS = [ re.compile(r"\b(?:still|again|keeps?|won'?t stop|getting worse|not getting better)\b", re.IGNORECASE), re.compile(r"\b(?:came back|come back|coming back|returned|recurring|back again)\b", re.IGNORECASE), re.compile(r"\b(?:right now|at this moment|currently|presently|as we speak)\b", re.IGNORECASE), re.compile(r"\b(?:worse than|worse today|worse now|escalating|intensifying)\b", re.IGNORECASE), re.compile(r"\b(?:i'?m (?:still |now )?(?:having|experiencing))\b", re.IGNORECASE), re.compile(r"\bi'?m feeling\b(?!\s*(?:better|fine|good|okay|ok|great|alright|normal|well))", re.IGNORECASE), re.compile(r"\b(?:it'?s (?:back|worse|not going away|spreading))\b", re.IGNORECASE), re.compile(r"\b(?:hasn'?t (?:gone away|stopped|improved|gotten better))\b", re.IGNORECASE), ] # ── Past-tense / resolved markers ── PAST_TENSE_PATTERNS = [ re.compile(r"\bi (?:had|used to have|was having|experienced|went through|suffered)\b", re.IGNORECASE), re.compile(r"\b(?:last (?:week|month|year|night|time)|yesterday|\d+ (?:days?|weeks?|months?|years?) ago)\b", re.IGNORECASE), re.compile(r"\b(?:in the past|previously|formerly|back in|when i was)\b", re.IGNORECASE), ] RESOLVED_PATTERNS = [ re.compile(r"\b(?:feeling (?:much )?better|improved|resolved|went away|gone now)\b", re.IGNORECASE), re.compile(r"\b(?:no longer|stopped|cleared up|fine now|okay now|recovered)\b", re.IGNORECASE), re.compile(r"\b(?:been treated|got (?:it )?checked|under control|managed|stable)\b", re.IGNORECASE), re.compile(r"\b(?:i'?m fine|i am fine|i'?m okay|i'?m ok|i'?m good|i'?m alright)\b", re.IGNORECASE), re.compile(r"\b(?:checked (?:me |it )?out|doctor (?:checked|said|cleared)|all clear)\b", re.IGNORECASE), ] # ── Proxy/third-party reporting patterns ── PROXY_PATTERNS = [ re.compile( r"\b(?:my\s+)?(?:mom|mother|dad|father|wife|husband|son|daughter|child|kid|" r"grandma|grandmother|grandpa|grandfather|brother|sister|friend|neighbor|" r"partner|spouse|baby|toddler|infant)\b", re.IGNORECASE ), ] # Active urgency markers that override third-party suppression PROXY_ACTIVE_URGENCY = [ re.compile(r"\b(?:says?|told me|telling me|texted|called|screaming)\b.*\b(?:hurts?|pain|can'?t breathe|bleeding|fell|fainted|passed out|unconscious|chest|heart|seizure)\b", re.IGNORECASE), re.compile(r"\b(?:hurts?|pain|can'?t breathe|bleeding|fell|fainted|passed out|unconscious|chest|heart|seizure)\b.*\b(?:really|very|so|extremely|terribly|awful|bad)\b", re.IGNORECASE), re.compile(r"\b(?:need|needs?|rush|hurry|emergency|ambulance|911|help)\b", re.IGNORECASE), re.compile(r"\b(?:right now|just now|just happened|happening)\b", re.IGNORECASE), ] # ── Intermittent/recurring symptom patterns ── # Ambiguous intermittent phrasing with dangerous symptoms should bias toward # escalation unless the patient's record shows chronic, stable baseline. INTERMITTENT_ACTIVE = [ re.compile(r"\b(?:sometimes|occasionally|on and off|comes and goes|every now and then|once in a while|from time to time|intermittent)\b.*\b(?:trouble breathing|chest pain|can'?t breathe|difficulty breathing|hard to breathe|dizzy|faint|pass out|black out|seizure|heart races|heart pounds|palpitation)\b", re.IGNORECASE), re.compile(r"\b(?:trouble breathing|chest pain|can'?t breathe|difficulty breathing|hard to breathe|dizzy|faint|pass out|black out|seizure|heart races|heart pounds|palpitation)\b.*\b(?:sometimes|occasionally|on and off|comes and goes|every now and then|once in a while|from time to time|intermittent)\b", re.IGNORECASE), re.compile(r"\b(?:sometimes|occasionally|on and off|every now and then)\b.*\b(?:pain in my chest|pressure in my chest|tightness in my chest|left arm goes numb|vision goes|lose consciousness|feel like i.{0,10}(?:faint|pass out))\b", re.IGNORECASE), ] def preprocess_temporal(text: str, context: Optional[str] = None) -> TemporalSignal: """ Analyze and preprocess text for temporal context before ML classification. This is the main entry point. Call this BEFORE tokenization/inference. Args: text: Patient's current utterance context: Optional conversation history Returns: TemporalSignal with preprocessing results """ original = text matched_rules = [] # ── Step 1: Check for BUT-clause reversals ── for i, pattern in enumerate(BUT_CLAUSE_PATTERNS): m = pattern.search(text) if m: active_part = m.group("active").strip() if len(active_part) >= 10: # Sanity check: active part is meaningful matched_rules.append(f"but_clause_reversal_{i}") return TemporalSignal( processed_text=active_part, original_text=original, is_active_emergency=True, temporal_tag="active", confidence=0.90, reason=f"BUT-clause reversal detected: resolved prefix discarded, active complaint extracted: '{active_part[:60]}'", matched_rules=matched_rules, ) # ── Step 2: Check for proxy/third-party with active urgency ── is_proxy = any(p.search(text) for p in PROXY_PATTERNS) if is_proxy: has_active_urgency = any(p.search(text) for p in PROXY_ACTIVE_URGENCY) if has_active_urgency: matched_rules.append("proxy_active_urgency") return TemporalSignal( processed_text=text, original_text=original, is_active_emergency=True, temporal_tag="active", confidence=0.85, reason="Third-party report with active urgency — treat as emergency", proxy_report=True, matched_rules=matched_rules, ) # ── Step 3: Check for intermittent symptoms with dangerous conditions ── # Moved BEFORE past-tense analysis: "sometimes I have chest pain" should # bias toward escalation regardless of temporal framing. for p in INTERMITTENT_ACTIVE: if p.search(text): matched_rules.append("intermittent_dangerous") return TemporalSignal( processed_text=text, original_text=original, is_active_emergency=True, temporal_tag="active", confidence=0.75, reason="Intermittent/recurring dangerous symptom — should escalate", matched_rules=matched_rules, ) # ── Step 4: Check for past-tense + still-active markers ── has_past = any(p.search(text) for p in PAST_TENSE_PATTERNS) has_still_active = any(p.search(text) for p in STILL_ACTIVE_PATTERNS) has_resolved = any(p.search(text) for p in RESOLVED_PATTERNS) if has_past and has_still_active: matched_rules.append("past_but_still_active") return TemporalSignal( processed_text=text, original_text=original, is_active_emergency=True, temporal_tag="active", confidence=0.85, reason="Past-tense language detected BUT still-active markers present — active condition", matched_rules=matched_rules, ) # ── Step 5: Contrastive clause analysis ── # When but/however/yet splits a sentence, prioritize the post-contrastive # clause — contrast often reverses clinical polarity. contrastive_match = CONTRASTIVE_MARKER.search(text) if contrastive_match and has_past: post_contrastive = text[contrastive_match.end():].strip() # Check if post-contrastive clause contains resolution markers post_has_resolved = any(p.search(post_contrastive) for p in RESOLVED_PATTERNS) # Check if post-contrastive clause contains active markers post_has_active = any(p.search(post_contrastive) for p in STILL_ACTIVE_PATTERNS) if post_has_active and not post_has_resolved: # "I had chest pain yesterday but it's getting worse" → active matched_rules.append("contrastive_post_active") return TemporalSignal( processed_text=text, original_text=original, is_active_emergency=True, temporal_tag="active", confidence=0.90, reason=f"Contrastive clause reversal: post-contrastive is active — '{post_contrastive[:50]}'", matched_rules=matched_rules, ) if post_has_resolved and not post_has_active: # "I had chest pain yesterday but I'm feeling better now" # → resolved, but app.py enforces R2 minimum for R3 domains matched_rules.append("contrastive_post_resolved") return TemporalSignal( processed_text=text, original_text=original, is_active_emergency=False, temporal_tag="resolved", confidence=0.80, reason="Past symptom with contrastive resolution — resolved but warrants clinical follow-up", matched_rules=matched_rules, ) # ── Step 6: Past + explicit resolution (no contrastive) ── if has_past and has_resolved and not has_still_active: matched_rules.append("past_and_resolved") return TemporalSignal( processed_text=text, original_text=original, is_active_emergency=False, temporal_tag="resolved", confidence=0.80, reason="Past-tense + resolution markers, no active signals — resolved but warrants follow-up", matched_rules=matched_rules, ) # ── Step 7: Past tense only — NO resolution evidence ── # SAFETY: Past tense alone does NOT mean resolved. "I had chest pain # yesterday" without saying they're better is a RED FLAG, not a reason # to suppress. Tag as neutral and let the safety system handle it. # Proxy reports are also NOT automatically suppressed — caregivers are # authoritative reporters of acute symptoms. if has_past and not has_still_active and not has_resolved: matched_rules.append("past_no_resolution") return TemporalSignal( processed_text=text, original_text=original, is_active_emergency=False, temporal_tag="neutral", confidence=0.50, reason="Past-tense language but NO resolution evidence — cannot confirm resolved", proxy_report=is_proxy, matched_rules=matched_rules, ) # ── Step 8: No temporal signals detected — neutral ── return TemporalSignal( processed_text=text, original_text=original, is_active_emergency=False, temporal_tag="neutral", confidence=0.5, reason="No temporal signals detected", matched_rules=matched_rules, )