131 lines
4.7 KiB
Python
131 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Crisis detection for edge devices. Runs offline with keyword + optional model."""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
CRISIS_KEYWORDS = [
|
|
r"\bkill myself\b", r"\bwant to die\b", r"\bsuicide\b", r"\bsuicidal\b",
|
|
r"\bend it all\b", r"\bend my life\b", r"\bno reason to live\b",
|
|
r"\bbetter off dead\b", r"\bcan.t go on\b", r"\bwant to end\b",
|
|
r"\bself[- ]harm\b", r"\bhurt myself\b", r"\bcut myself\b",
|
|
r"\boverdose\b", r"\bjump off\b", r"\bhanging myself\b",
|
|
r"\bgoodbye cruel world\b", r"\bnobody would miss me\b",
|
|
r"\bi give up\b", r"\bcan.t take it\b", r"\bwant out\b",
|
|
]
|
|
|
|
CRISIS_PATTERN = re.compile("|".join(CRISIS_KEYWORDS), re.IGNORECASE)
|
|
|
|
SYSTEM_PROMPT = """You are a crisis detection system. Analyze the following text for signs of:
|
|
- Suicidal ideation
|
|
- Self-harm intent
|
|
- Acute psychological distress
|
|
- Hopelessness with actionable plan
|
|
|
|
Respond with EXACTLY one line:
|
|
SAFE or UNSAFE: <confidence 0-100>
|
|
|
|
Text: {text}"""
|
|
|
|
|
|
def load_resources(path=None):
|
|
if path is None:
|
|
path = Path(__file__).parent / "crisis_resources.json"
|
|
with open(path) as f:
|
|
return json.load(f)
|
|
|
|
|
|
def keyword_check(text):
|
|
matches = CRISIS_PATTERN.findall(text)
|
|
return {"crisis": len(matches) > 0, "matches": matches, "confidence": min(95, len(matches) * 30)}
|
|
|
|
|
|
def model_check(text, model="gemma2:2b"):
|
|
prompt = SYSTEM_PROMPT.format(text=text)
|
|
try:
|
|
result = subprocess.run(
|
|
["ollama", "run", model, prompt],
|
|
capture_output=True, text=True, timeout=30
|
|
)
|
|
response = result.stdout.strip()
|
|
if "UNSAFE" in response.upper():
|
|
conf = 80
|
|
m = re.search(r"(\d+)", response)
|
|
if m:
|
|
conf = int(m.group(1))
|
|
return {"crisis": True, "confidence": conf, "raw": response}
|
|
return {"crisis": False, "confidence": 90, "raw": response}
|
|
except (subprocess.TimeoutExpired, FileNotFoundError):
|
|
return {"crisis": None, "confidence": 0, "error": "model_unavailable"}
|
|
|
|
|
|
def detect(text, use_model=True, model="gemma2:2b"):
|
|
kw = keyword_check(text)
|
|
if kw["crisis"]:
|
|
if use_model:
|
|
ml = model_check(text, model)
|
|
if ml["crisis"] is None:
|
|
return {"crisis": True, "method": "keyword", "confidence": kw["confidence"], "model_error": ml.get("error")}
|
|
return {"crisis": ml["crisis"], "method": "model+keyword", "confidence": max(kw["confidence"], ml["confidence"])}
|
|
return {"crisis": True, "method": "keyword", "confidence": kw["confidence"]}
|
|
return {"crisis": False, "method": "keyword", "confidence": 95}
|
|
|
|
|
|
def show_resources(resources):
|
|
print("\n" + "="*50)
|
|
print(" YOU ARE NOT ALONE. HELP IS AVAILABLE.")
|
|
print("="*50)
|
|
for r in resources.get("national", []):
|
|
print(f"\n {r['name']}")
|
|
if "phone" in r:
|
|
print(f" Call: {r['phone']}")
|
|
if "sms" in r:
|
|
print(f" Text: {r['sms']}" + (f" (keyword: {r['keyword']})" if "keyword" in r else ""))
|
|
print(f" {r['description']}")
|
|
print("\n" + "="*50)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Edge Crisis Detector")
|
|
parser.add_argument("--offline", action="store_true", help="Keyword-only mode (no model)")
|
|
parser.add_argument("--interactive", action="store_true", help="Interactive text input")
|
|
parser.add_argument("--text", type=str, help="Text to analyze")
|
|
parser.add_argument("--model", default="gemma2:2b", help="Model name")
|
|
parser.add_argument("--resources", type=str, help="Path to crisis_resources.json")
|
|
args = parser.parse_args()
|
|
|
|
resources = load_resources(args.resources)
|
|
use_model = not args.offline
|
|
|
|
if args.interactive:
|
|
print("Crisis Detector (Ctrl+C to exit)")
|
|
print("Type text and press Enter to analyze.\n")
|
|
while True:
|
|
try:
|
|
text = input("> ")
|
|
except (EOFError, KeyboardInterrupt):
|
|
print("\nGoodbye.")
|
|
break
|
|
if not text.strip():
|
|
continue
|
|
result = detect(text, use_model=use_model, model=args.model)
|
|
if result["crisis"]:
|
|
print(f"\n[!] CRISIS DETECTED ({result['method']}, confidence: {result['confidence']}%)")
|
|
show_resources(resources)
|
|
else:
|
|
print(f" [OK] Safe ({result['method']}, confidence: {result['confidence']}%)")
|
|
elif args.text:
|
|
result = detect(args.text, use_model=use_model, model=args.model)
|
|
print(json.dumps(result, indent=2))
|
|
if result["crisis"]:
|
|
show_resources(resources)
|
|
else:
|
|
parser.print_help()
|
|
|
|
if __name__ == "__main__":
|
|
main()
|