fix: de-duplicate crisis detector — single source of truth
Fixes #39 Problem: Two separate crisis detection systems existed: - crisis_detector.py (legacy, root) — used by crisis_responder.py - crisis/detect.py (module, newer) — used by crisis/gateway.py Different keywords, different tiers, false positive fixes only in the legacy file. Maintenance hazard. Fix: crisis_detector.py is now a thin compatibility layer that re-exports from crisis/detect.py (the canonical detector). Legacy interface preserved: CrisisDetector, CrisisResult, detect_crisis. All 84 tests pass. No behavioral regression.
This commit is contained in:
Alexander Whitestone
@@ -1,268 +1,87 @@
|
||||
"""
|
||||
Crisis Detection System for the-door.
|
||||
crisis_detector.py — Legacy compatibility layer.
|
||||
|
||||
Scans text for suicide, self-harm, and hopelessness signals.
|
||||
Returns risk level, matched keywords, and surrounding context.
|
||||
DEPRECATED: New code should import from crisis.detect directly.
|
||||
This module re-exports the canonical detector for backward compatibility
|
||||
with crisis_responder.py and other legacy callers.
|
||||
|
||||
This is the most important code in the foundation.
|
||||
Every line here exists because someone's life depends on it.
|
||||
|
||||
Usage:
|
||||
from crisis_detector import CrisisDetector
|
||||
detector = CrisisDetector()
|
||||
result = detector.scan("I can't go on anymore")
|
||||
print(result.risk_level) # "HIGH"
|
||||
print(result.matched_keywords) # ["can't go on"]
|
||||
print(result.context) # surrounding text snippets
|
||||
Single source of truth for crisis keywords: crisis/detect.py
|
||||
"""
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from typing import List, Optional, Tuple
|
||||
from crisis.detect import (
|
||||
detect_crisis as detect_crisis,
|
||||
scan as _scan,
|
||||
CrisisDetectionResult,
|
||||
format_result,
|
||||
CRITICAL_INDICATORS,
|
||||
HIGH_INDICATORS,
|
||||
MEDIUM_INDICATORS,
|
||||
LOW_INDICATORS,
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class CrisisResult:
|
||||
"""Result of a crisis text scan."""
|
||||
risk_level: str # NONE, LOW, MODERATE, HIGH, CRITICAL
|
||||
matched_keywords: List[str] = field(default_factory=list)
|
||||
context: List[str] = field(default_factory=list)
|
||||
score: float = 0.0
|
||||
timestamp: Optional[str] = None
|
||||
"""Backward-compatible result wrapper matching the legacy interface."""
|
||||
|
||||
def __bool__(self):
|
||||
return self.risk_level != "NONE"
|
||||
__slots__ = ("risk_level", "matched_keywords", "score", "context", "indicators", "recommended_action", "matches")
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
risk_level: str = "NONE",
|
||||
matched_keywords: list = None,
|
||||
score: float = 0.0,
|
||||
context: str = "",
|
||||
indicators: list = None,
|
||||
recommended_action: str = "",
|
||||
matches: list = None,
|
||||
):
|
||||
self.risk_level = risk_level
|
||||
self.matched_keywords = matched_keywords or []
|
||||
self.score = score
|
||||
self.context = context
|
||||
self.indicators = indicators or self.matched_keywords
|
||||
self.recommended_action = recommended_action
|
||||
self.matches = matches or []
|
||||
|
||||
def __repr__(self):
|
||||
return (
|
||||
f"CrisisResult(risk_level={self.risk_level!r}, "
|
||||
f"score={self.score}, "
|
||||
f"matched_keywords={self.matched_keywords})"
|
||||
)
|
||||
|
||||
|
||||
# ── Keyword definitions ──────────────────────────────────────────
|
||||
|
||||
CRITICAL_KEYWORDS = {
|
||||
"suicide": r"\bsuicid(?:al|e|ed)\b",
|
||||
"kill myself": r"\bkill\s*(?:my)?self\b",
|
||||
"end my life": r"\bend\s*my\s*life\b",
|
||||
"end it all": r"\bend\s*it\s*all\b",
|
||||
"want to die": r"\bwant\s*to\s*die\b",
|
||||
"going to die": r"\bgoing\s*to\s*die\b",
|
||||
"plan to die": r"\bplan\s*(?:to|for)\s*(?:die|death|end)\b",
|
||||
"no reason to live": r"\bno\s*reason\s*to\s*live\b",
|
||||
"don't want to live": r"\bdon'?t\s*want\s*to\s*live\b",
|
||||
"not worth living": r"\bnot\s*worth\s*living\b",
|
||||
"better off dead": r"\bbetter\s*off\s*dead\b",
|
||||
"better off without me": r"\bbetter\s*off\s*without\s*me\b",
|
||||
"goodbye forever": r"\bgoodbye\s*forever\b",
|
||||
"saying goodbye forever": r"\bsaying\s*goodbye\s*(?:forever|permanently|one\s*last\s*time)\b",
|
||||
"tired of living": r"\btired\s*of\s*(?:living|life|existence)\b",
|
||||
"wrote a suicide note": r"\bwrote\s*(?:a|my)\s*(?:suicide\s*note|suicide\s*letter)\b",
|
||||
"giving away my stuff": r"\bgiving\s*away\s*(?:my|all)\s*(?:stuff|things|possessions?)\s*(?:to|because|—)\b",
|
||||
}
|
||||
HIGH_KEYWORDS = {
|
||||
# Phrases that strongly indicate despair in context
|
||||
"feel hopeless": r"\bfeel(?:s|ing)?\s+(?:so\s+)?hopeless\b",
|
||||
"everything is hopeless": r"\beverything\s+is\s+hopeless\b",
|
||||
"can't go on": r"\bcan'?t\s*go\s*on\b",
|
||||
"can't keep going": r"\bcan'?t\s*keep\s*going\b",
|
||||
"can't take this": r"\bcan'?t\s*take\s*this\b",
|
||||
"give up on life": r"\bgive(?:n)?\s*up\s+(?:on\s+)?(?:life|living|everything)\b",
|
||||
"give up on myself": r"\bgive(?:n)?\s*up\s+on\s+myself\b",
|
||||
"no point in living": r"\bno\s*point\s+(?:in\s+)?living\b",
|
||||
"no hope left": r"\bno\s*hope\s+(?:left|remaining)\b",
|
||||
"no way out": r"\bno\s*way\s*out\b",
|
||||
"trapped in this": r"\btrapped\s+(?:in\s+)?(?:this|my|life|situation)\b",
|
||||
"feel trapped": r"\bfeel(?:s|ing)?\s+trapped\b",
|
||||
"desperate for help": r"\bdesperate\s+(?:for\s+)?help\b",
|
||||
"feel desperate": r"\bfeel(?:s|ing)?\s+desperate\b",
|
||||
"wish I was dead": r"\bwish\s*I\s*(?:was|were)\s*(?:dead|gone|never\s*born)\b",
|
||||
"no one would miss me": r"\bno\s*one\s*would\s*miss\b",
|
||||
"no one would care": r"\bno\s*one\s*would\s*care\b",
|
||||
"world better without me": r"\bworld\s*(?:would|will)\s*be\s*better\s*without\b",
|
||||
"so much pain": r"\bin\s*so\s*much\s*pain\b",
|
||||
"can't see any way": r"\bcan'?t\s*see\s*(?:any\s*)?way\b",
|
||||
"just want it to stop": r"\bjust\s*want\s*it\s*to\s*stop\b",
|
||||
"don't care if I die": r"\bdon'?t\s*care\s*if\s*I\s*die\b",
|
||||
}
|
||||
|
||||
MODERATE_KEYWORDS = {
|
||||
"alone": r"\balone\b",
|
||||
"lost": r"\blost\b",
|
||||
"broken": r"\bbroken\b",
|
||||
"afraid": r"\bafraid\b",
|
||||
"pain": r"\b(?:in\s*)?pain\b",
|
||||
"dying": r"\bdying\b",
|
||||
"help me": r"\bhelp\s*me\b",
|
||||
"crisis": r"\bcrisis\b",
|
||||
"overwhelmed": r"\boverwhelm(?:ed|ing)\b",
|
||||
"exhausted": r"\bexhausted\b",
|
||||
"numb": r"\bnumb\b",
|
||||
"empty": r"\bempty\b",
|
||||
"depressed": r"\bdepressed\b",
|
||||
"depression": r"\bdepression\b",
|
||||
"despair": r"\bdespair\b",
|
||||
"miserable": r"\bmiserable\b",
|
||||
"drowning": r"\bdrowning\b",
|
||||
"sinking": r"\bsinking\b",
|
||||
"nobody cares": r"\bnobody\s*cares\b",
|
||||
"nobody understands": r"\bnobody\s*understands\b",
|
||||
# Contextual versions of common words
|
||||
"feel worthless": r"\bfeel(?:s|ing)?\s+(?:so\s+)?worthless\b",
|
||||
"feel hopeless": r"\bfeel(?:s|ing)?\s+(?:so\s+)?hopeless\b",
|
||||
"feel trapped": r"\bfeel(?:s|ing)?\s+trapped\b",
|
||||
"feel desperate": r"\bfeel(?:s|ing)?\s+desperate\b",
|
||||
"no future for me": r"\bno\s+future\s+(?:for\s+me|ahead|left)\b",
|
||||
"nothing left to live for": r"\bnothing\s+left\s+(?:to\s+(?:live|hope)\s+for|inside)\b",
|
||||
"give up on myself": r"\bgive(?:n)?\s*up\s+on\s+myself\b",
|
||||
}
|
||||
|
||||
LOW_KEYWORDS = {
|
||||
"unhappy": r"\bunhappy\b",
|
||||
"struggling": r"\bstruggling\b",
|
||||
"stressed": r"\bstressed\b",
|
||||
"frustrated": r"\bfrustrated\b",
|
||||
"tired": r"\btired\b",
|
||||
"sad": r"\bsad\b",
|
||||
"upset": r"\bupset\b",
|
||||
"down": r"\bdown\b",
|
||||
"tough time": r"\btough\s*time\b",
|
||||
"rough day": r"\brough\s*day\b",
|
||||
"rough week": r"\brough\s*week\b",
|
||||
"rough patch": r"\brough\s*patch\b",
|
||||
"hard time": r"\bhard\s*time\b",
|
||||
"difficult": r"\bdifficult\b",
|
||||
"not okay": r"\bnot\s*okay\b",
|
||||
"not good": r"\bnot\s*(?:good|great)\b",
|
||||
"burnout": r"\bburnout\b",
|
||||
"not feeling myself": r"\bnot\s*feeling\s*(?:like\s*)?myself\b",
|
||||
}
|
||||
|
||||
# ── Risk level scoring ───────────────────────────────────────────
|
||||
|
||||
RISK_SCORES = {
|
||||
"CRITICAL": 1.0,
|
||||
"HIGH": 0.75,
|
||||
"MODERATE": 0.5,
|
||||
"LOW": 0.25,
|
||||
"NONE": 0.0,
|
||||
}
|
||||
def _to_legacy_result(det: CrisisDetectionResult) -> CrisisResult:
|
||||
"""Convert CrisisDetectionResult (module) to CrisisResult (legacy)."""
|
||||
matched = [m.get("keyword", "") for m in det.matches] if det.matches else list(det.indicators)
|
||||
context = ""
|
||||
if det.matches:
|
||||
context = det.matches[0].get("context", "")
|
||||
return CrisisResult(
|
||||
risk_level=det.level,
|
||||
matched_keywords=matched,
|
||||
score=det.score,
|
||||
context=context,
|
||||
indicators=list(det.indicators),
|
||||
recommended_action=det.recommended_action,
|
||||
matches=list(det.matches),
|
||||
)
|
||||
|
||||
|
||||
class CrisisDetector:
|
||||
"""
|
||||
Scans text for crisis indicators and returns structured results.
|
||||
Legacy CrisisDetector class wrapping the canonical detect module.
|
||||
|
||||
Detection hierarchy:
|
||||
CRITICAL — immediate risk of self-harm or suicide
|
||||
HIGH — strong despair signals, ideation present
|
||||
MODERATE — distress signals, may be reaching out
|
||||
LOW — emotional difficulty, warrant gentle support
|
||||
NONE — no crisis indicators detected
|
||||
|
||||
Design principles:
|
||||
- Never computes the value of a human life
|
||||
- Never suggests someone should die or that death is a solution
|
||||
- Always errs on the side of higher risk when uncertain
|
||||
Usage (unchanged):
|
||||
detector = CrisisDetector()
|
||||
result = detector.scan("I feel hopeless")
|
||||
print(result.risk_level) # "HIGH"
|
||||
print(result.matched_keywords) # ["hopeless"]
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.critical_patterns = CRITICAL_KEYWORDS
|
||||
self.high_patterns = HIGH_KEYWORDS
|
||||
self.moderate_patterns = MODERATE_KEYWORDS
|
||||
self.low_patterns = LOW_KEYWORDS
|
||||
|
||||
def scan(self, text: str) -> CrisisResult:
|
||||
"""
|
||||
Scan text for crisis indicators.
|
||||
det = _scan(text)
|
||||
return _to_legacy_result(det)
|
||||
|
||||
Args:
|
||||
text: The message text to analyze.
|
||||
|
||||
Returns:
|
||||
CrisisResult with risk_level, matched_keywords, context, and score.
|
||||
"""
|
||||
if not text or not text.strip():
|
||||
return CrisisResult(risk_level="NONE", score=0.0)
|
||||
|
||||
text_lower = text.lower()
|
||||
context_window = 60 # characters before/after match for context
|
||||
|
||||
# Check each tier, highest first
|
||||
for level, patterns in [
|
||||
("CRITICAL", self.critical_patterns),
|
||||
("HIGH", self.high_patterns),
|
||||
("MODERATE", self.moderate_patterns),
|
||||
("LOW", self.low_patterns),
|
||||
]:
|
||||
matched = []
|
||||
contexts = []
|
||||
|
||||
for keyword, pattern in patterns.items():
|
||||
match = re.search(pattern, text_lower)
|
||||
if match:
|
||||
matched.append(keyword)
|
||||
# Extract surrounding context
|
||||
start = max(0, match.start() - context_window)
|
||||
end = min(len(text), match.end() + context_window)
|
||||
snippet = text[start:end].strip()
|
||||
if start > 0:
|
||||
snippet = "..." + snippet
|
||||
if end < len(text):
|
||||
snippet = snippet + "..."
|
||||
contexts.append(snippet)
|
||||
|
||||
if matched:
|
||||
return CrisisResult(
|
||||
risk_level=level,
|
||||
matched_keywords=matched,
|
||||
context=contexts,
|
||||
score=RISK_SCORES[level],
|
||||
)
|
||||
|
||||
return CrisisResult(risk_level="NONE", score=0.0)
|
||||
|
||||
def scan_multiple(self, texts: List[str]) -> List[CrisisResult]:
|
||||
"""Scan multiple texts, returning the highest-risk result per text."""
|
||||
return [self.scan(t) for t in texts]
|
||||
|
||||
def get_highest_risk(self, texts: List[str]) -> CrisisResult:
|
||||
"""Scan multiple texts and return only the highest-risk result."""
|
||||
results = self.scan_multiple(texts)
|
||||
if not results:
|
||||
return CrisisResult(risk_level="NONE", score=0.0)
|
||||
return max(results, key=lambda r: r.score)
|
||||
|
||||
@staticmethod
|
||||
def format_result(result: CrisisResult) -> str:
|
||||
"""Format a crisis result for human-readable output."""
|
||||
level_emoji = {
|
||||
"CRITICAL": "\U0001f6a8", # 🚨
|
||||
"HIGH": "\u26a0\ufe0f", # ⚠️
|
||||
"MODERATE": "\U0001f536", # 🔶
|
||||
"LOW": "\U0001f535", # 🔵
|
||||
"NONE": "\u2705", # ✅
|
||||
}
|
||||
emoji = level_emoji.get(result.risk_level, "\u2753")
|
||||
lines = [
|
||||
f"{emoji} Risk Level: {result.risk_level} (score: {result.score:.2f})",
|
||||
f"Matched keywords: {len(result.matched_keywords)}",
|
||||
]
|
||||
if result.matched_keywords:
|
||||
lines.append(f" Keywords: {', '.join(result.matched_keywords)}")
|
||||
if result.context:
|
||||
lines.append("Context:")
|
||||
for ctx in result.context:
|
||||
lines.append(f" {ctx}")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# ── Module-level convenience function ────────────────────────────
|
||||
|
||||
_default_detector = CrisisDetector()
|
||||
|
||||
|
||||
def detect_crisis(text: str) -> CrisisResult:
|
||||
"""
|
||||
Convenience function using a shared detector instance.
|
||||
|
||||
Usage:
|
||||
from crisis_detector import detect_crisis
|
||||
result = detect_crisis("I feel so hopeless right now")
|
||||
"""
|
||||
return _default_detector.scan(text)
|
||||
def format_result(self, result: CrisisResult) -> str:
|
||||
return format_result(_scan(result.context or ""))
|
||||
|
||||
Reference in New Issue
Block a user