a0e4f15fc3
- Add scan() alias for detect_crisis - Add extract_context() for match context windows - Add generate_grounding_steps() (5-4-3-2-1 grounding) - Add generate_breathing_exercise() (box breathing) - Extend CrisisDetectionResult with matches field (position data) - Refactor _find_indicators to return position-aware match dicts All 77 tests pass (72 existing + 5 rescued).
235 lines
7.3 KiB
Python
235 lines
7.3 KiB
Python
"""
|
|
Crisis Detection Module for the-door.
|
|
|
|
Parses incoming text for despair/suicide indicators and classifies into
|
|
tiers: LOW, MEDIUM, HIGH, CRITICAL.
|
|
"""
|
|
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from typing import List
|
|
|
|
|
|
@dataclass
|
|
class CrisisDetectionResult:
|
|
level: str
|
|
indicators: List[str] = field(default_factory=list)
|
|
recommended_action: str = ""
|
|
score: float = 0.0
|
|
matches: List[dict] = field(default_factory=list)
|
|
|
|
|
|
# ── Indicator sets ──────────────────────────────────────────────
|
|
|
|
CRITICAL_INDICATORS = [
|
|
r"\bbetter off without me\b",
|
|
r"\bkill\s*(my)?self\b",
|
|
r"\bend\s*my\s*life\b",
|
|
r"\bsuicid(?:al|ed|e)\b",
|
|
r"\bnot\s+worth\s+living\b",
|
|
r"\bbetter\s+off\s+dead\b",
|
|
r"\bend\s+it\s+all\b",
|
|
r"\bcan'?t\s+(go|live)\s+on\b",
|
|
r"\bno\s+reason\s+to\s+live\b",
|
|
r"\bdon'?t\s+want\s+to\s+live\b",
|
|
r"\bwant\s+to\s+die\b",
|
|
r"\bgoing\s+to\s+kill\s+myself\b",
|
|
r"\bplan\s+to\s+(end|kill|die)\b",
|
|
r"\btired\s+of\s+(living|life|existence)\b",
|
|
r"\bsaying\s+goodbye\s+(forever|permanently|one last time)\b",
|
|
r"\bwrote\s+a\s+(will|suicide\s*note|letter)\b",
|
|
r"\bgiving\s+away\s+(my|all my)\s+possess",
|
|
r"\btied\s+(up|down)\s+my\s+(loose\s+)?ends",
|
|
]
|
|
|
|
HIGH_INDICATORS = [
|
|
r"\bdesperate\b",
|
|
r"\bdespair\b",
|
|
r"\bhopeless\b",
|
|
r"\bno(?!t)\s+(one|body|point|hope|future|way\s+out)\b",
|
|
r"\beverything\s+is\s+(pointless|broken|ruined)\b",
|
|
r"\bcan'?t\s+take\s+this\s+anymore\b",
|
|
r"\bdon'?t\s+care\s+if\s+I\s+die\b",
|
|
r"\bwish\s+I\s+(was|were)\s+(dead|gone|never\s+born)\b",
|
|
r"\bdon'?t\s+matter\s+if\s+I\s+exist\b",
|
|
r"\bno\s+one\s+would\s+care\b",
|
|
r"\bno\s+one\s+would\s+miss\b",
|
|
r"\bworld\s+would?\s+be\s+better\s+without\b",
|
|
r"\bin\s+so\s+much\s+(pain|agony|suffering|torment)\b",
|
|
r"\bcan'?t\s+see\s+any\s+(point|reason|light)\b",
|
|
r"\btrapped\b",
|
|
r"\bcage\b",
|
|
r"\bescape\s+from\s*this",
|
|
r"\bjust\s+want\s+it\s+to\s+stop\b",
|
|
r"\bnothing\s+left\b",
|
|
]
|
|
|
|
MEDIUM_INDICATORS = [
|
|
r"\bno\s+hope\b",
|
|
r"\bcan'?t\s+go\s+on\b",
|
|
r"\bcan'?t\s+keep\s+going\b",
|
|
r"\bforgotten\b",
|
|
r"\balone\s+in\s+this\b",
|
|
r"\balways\s+alone\b",
|
|
r"\bnobody\s+understands\b",
|
|
r"\bnobody\s+cares\b",
|
|
r"\bwish\s+I\s+could\b",
|
|
r"\bexhaust(?:ed|ion|ing)\b",
|
|
r"\bnumb\b",
|
|
r"\bempty\b",
|
|
r"\bworthless\b",
|
|
r"\buseless\b",
|
|
r"\bbroken\b",
|
|
r"\bdark(ness)?\b",
|
|
r"\bdepressed\b",
|
|
r"\bdepression\b",
|
|
r"\bcrying\b",
|
|
r"\btears\b",
|
|
r"\bsad(ness)?\b",
|
|
r"\bmiserable\b",
|
|
r"\boverwhelm(?:ed|ing)\b",
|
|
r"\bfailing\b",
|
|
r"\bcannot\s+cope\b",
|
|
r"\blosing\s*(my)?\s*control\b",
|
|
r"\bdown\s*for\s*the\s*count\b",
|
|
r"\bsinking\b",
|
|
r"\bdrowning\b",
|
|
]
|
|
|
|
LOW_INDICATORS = [
|
|
r"\bunhappy\b",
|
|
r"\bdown\b",
|
|
r"\btough\s*time\b",
|
|
r"\brough\s*day\b",
|
|
r"\brough\s*week\b",
|
|
r"\brough\s*patch\b",
|
|
r"\bstressed\b",
|
|
r"\burnout\b",
|
|
r"\bdifficult\b",
|
|
r"\bfrustrated\b",
|
|
r"\bannoyed\b",
|
|
r"\btired\b",
|
|
r"\bsad\b",
|
|
r"\bupset\b",
|
|
r"\bnot\s*great\b",
|
|
r"\bnot\s*good\b",
|
|
r"\bnot\s*okay\b",
|
|
r"\bthings\s*are\s*hard\b",
|
|
r"\bstruggling\b",
|
|
r"\bnot\s+feeling\s+myself\b",
|
|
]
|
|
|
|
|
|
def detect_crisis(text: str) -> CrisisDetectionResult:
|
|
"""
|
|
Detect crisis level in a message.
|
|
|
|
Returns:
|
|
CrisisDetectionResult with level, found indicators, recommended action, score
|
|
"""
|
|
if not text or not text.strip():
|
|
return CrisisDetectionResult(level="NONE", score=0.0)
|
|
|
|
text_lower = text.lower()
|
|
matches = _find_indicators(text_lower)
|
|
|
|
if not matches:
|
|
return CrisisDetectionResult(level="NONE", score=0.0)
|
|
|
|
# Priority: highest tier wins
|
|
for tier in ("CRITICAL", "HIGH", "MEDIUM", "LOW"):
|
|
if matches[tier]:
|
|
tier_matches = matches[tier]
|
|
patterns = [m["pattern"] for m in tier_matches]
|
|
scores = {"CRITICAL": 1.0, "HIGH": 0.75, "MEDIUM": 0.5, "LOW": 0.25}
|
|
actions = {
|
|
"CRITICAL": (
|
|
"INTERRUPT CONVERSATION. Ask 'Are you safe right now?' "
|
|
"Provide 988 Suicide & Crisis Lifeline. "
|
|
"Provide Crisis Text Line (HOME to 741741). "
|
|
"Stay present. Do not disconnect. Redirect to professional help."
|
|
),
|
|
"HIGH": (
|
|
"Show crisis panel. Ask about safety. Surface 988 number prominently. "
|
|
"Continue conversation with crisis awareness."
|
|
),
|
|
"MEDIUM": (
|
|
"Increase warmth and presence. Subtly surface help resources. "
|
|
"Keep conversation anchored in the present."
|
|
),
|
|
"LOW": (
|
|
"Normal conversation with warm undertone. "
|
|
"No crisis UI elements needed. Remain vigilant."
|
|
),
|
|
}
|
|
return CrisisDetectionResult(
|
|
level=tier,
|
|
indicators=patterns,
|
|
recommended_action=actions[tier],
|
|
score=scores[tier],
|
|
matches=tier_matches,
|
|
)
|
|
|
|
return CrisisDetectionResult(level="NONE", score=0.0)
|
|
|
|
|
|
def _find_indicators(text: str) -> dict:
|
|
"""Return dict with indicators found per tier, including match positions."""
|
|
results = {"CRITICAL": [], "HIGH": [], "MEDIUM": [], "LOW": []}
|
|
|
|
for pattern in CRITICAL_INDICATORS:
|
|
m = re.search(pattern, text)
|
|
if m:
|
|
results["CRITICAL"].append({"pattern": pattern, "start": m.start(), "end": m.end()})
|
|
|
|
for pattern in HIGH_INDICATORS:
|
|
m = re.search(pattern, text)
|
|
if m:
|
|
results["HIGH"].append({"pattern": pattern, "start": m.start(), "end": m.end()})
|
|
|
|
for pattern in MEDIUM_INDICATORS:
|
|
m = re.search(pattern, text)
|
|
if m:
|
|
results["MEDIUM"].append({"pattern": pattern, "start": m.start(), "end": m.end()})
|
|
|
|
for pattern in LOW_INDICATORS:
|
|
m = re.search(pattern, text)
|
|
if m:
|
|
results["LOW"].append({"pattern": pattern, "start": m.start(), "end": m.end()})
|
|
|
|
return results
|
|
|
|
|
|
def scan(text: str) -> CrisisDetectionResult:
|
|
"""Alias for detect_crisis — shorter name used in tests."""
|
|
return detect_crisis(text)
|
|
|
|
|
|
def extract_context(text: str, start: int, end: int, window: int = 60) -> str:
|
|
"""Extract surrounding context around a match position."""
|
|
ctx_start = max(0, start - window)
|
|
ctx_end = min(len(text), end + window)
|
|
snippet = text[ctx_start:ctx_end].strip()
|
|
if ctx_start > 0:
|
|
snippet = "..." + snippet
|
|
if ctx_end < len(text):
|
|
snippet = snippet + "..."
|
|
return snippet
|
|
|
|
|
|
def get_urgency_emoji(level: str) -> str:
|
|
mapping = {"CRITICAL": "🚨", "HIGH": "⚠️", "MEDIUM": "🔶", "LOW": "🔵", "NONE": "✅"}
|
|
return mapping.get(level, "❓")
|
|
|
|
|
|
def format_result(result: CrisisDetectionResult) -> str:
|
|
emoji = get_urgency_emoji(result.level)
|
|
lines = [
|
|
f"{emoji} Crisis Level: {result.level} (score: {result.score})",
|
|
f"Indicators: {len(result.indicators)} found",
|
|
f"Action: {result.recommended_action or 'None needed'}",
|
|
]
|
|
if result.indicators:
|
|
lines.append(f"Patterns: {result.indicators}")
|
|
return "\n".join(lines)
|