Alexander Whitestone
da31288525
All checks were successful
Smoke Test / smoke (push) Successful in 4s
Merge PR #76 (squash)
90 lines
2.6 KiB
Python
90 lines
2.6 KiB
Python
"""
|
|
DEPRECATED — Use crisis.detect instead.
|
|
|
|
This module is a thin wrapper around crisis.detect for backward compatibility.
|
|
All unique patterns have been merged into crisis/detect.py (see issue #40).
|
|
|
|
This module will be removed in a future release.
|
|
"""
|
|
|
|
import warnings
|
|
from dataclasses import dataclass, field, asdict
|
|
from typing import List, Optional, Dict
|
|
import json
|
|
import hashlib
|
|
|
|
# Re-export the canonical detection
|
|
from crisis.detect import detect_crisis, CrisisDetectionResult
|
|
|
|
# Issue deprecation warning on import
|
|
warnings.warn(
|
|
"dying_detection is deprecated. Use 'from crisis.detect import detect_crisis' instead. "
|
|
"All patterns have been consolidated into crisis/detect.py. "
|
|
"See issue #40.",
|
|
DeprecationWarning,
|
|
stacklevel=2,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class DetectionResult:
|
|
"""Backward-compatible result type matching the old dying_detection API."""
|
|
level: str
|
|
indicators: List[str] = field(default_factory=list)
|
|
recommended_action: str = ""
|
|
raw_matched_patterns: List[str] = field(default_factory=list)
|
|
confidence: float = 0.0
|
|
session_hash: str = ""
|
|
|
|
|
|
def detect(text: str) -> DetectionResult:
|
|
"""
|
|
Primary detection function — delegates to crisis.detect.
|
|
|
|
Args:
|
|
text: User message to analyze
|
|
|
|
Returns:
|
|
DetectionResult with level, indicators, recommended_action, confidence
|
|
"""
|
|
result = detect_crisis(text)
|
|
|
|
# Extract raw patterns from matches
|
|
raw_patterns = [m["pattern"] for m in result.matches] if result.matches else []
|
|
|
|
return DetectionResult(
|
|
level=result.level,
|
|
indicators=result.indicators,
|
|
recommended_action=result.recommended_action,
|
|
raw_matched_patterns=raw_patterns,
|
|
confidence=result.score,
|
|
session_hash=hashlib.sha256(text.encode()).hexdigest()[:12],
|
|
)
|
|
|
|
|
|
def get_action_for_level(level: str) -> str:
|
|
"""Get the recommended action string for a given level."""
|
|
from crisis.detect import ACTIONS
|
|
return ACTIONS.get(level, "Unknown level.")
|
|
|
|
|
|
def as_json(result: DetectionResult, indent: int = 2) -> str:
|
|
"""Return the DetectionResult as a JSON string."""
|
|
return json.dumps(asdict(result), indent=indent)
|
|
|
|
|
|
def process(text: str) -> dict:
|
|
"""
|
|
Full pipeline: detect and return a dict.
|
|
This is the primary API function for other modules.
|
|
"""
|
|
result = detect(text)
|
|
return {
|
|
"level": result.level,
|
|
"indicators": result.indicators,
|
|
"recommended_action": result.recommended_action,
|
|
"confidence": result.confidence,
|
|
"raw_patterns": result.raw_matched_patterns,
|
|
"action": get_action_for_level(result.level),
|
|
}
|