Compare commits
1 Commits
fix/132
...
door/issue
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1f744e846d |
@@ -1,155 +0,0 @@
|
||||
"""
|
||||
Image Content Screening for Self-Harm Indicators
|
||||
|
||||
Screens images sent in chat for self-harm content using vision models.
|
||||
Privacy-preserving: analyzes in-memory, discards after scoring. No image retention.
|
||||
"""
|
||||
|
||||
import base64
|
||||
import json
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
from typing import Optional
|
||||
|
||||
logger = logging.getLogger("the-door.image_screening")
|
||||
|
||||
|
||||
class RiskLevel(str, Enum):
|
||||
SAFE = "safe"
|
||||
CONCERNING = "concerning"
|
||||
CRITICAL = "critical"
|
||||
|
||||
|
||||
@dataclass
|
||||
class ImageScreenResult:
|
||||
risk_level: RiskLevel
|
||||
confidence: float
|
||||
categories: list
|
||||
reasoning: str
|
||||
model_used: str
|
||||
|
||||
@property
|
||||
def needs_crisis_response(self):
|
||||
return self.risk_level == RiskLevel.CRITICAL
|
||||
|
||||
@property
|
||||
def needs_followup(self):
|
||||
return self.risk_level in (RiskLevel.CONCERNING, RiskLevel.CRITICAL)
|
||||
|
||||
def to_dict(self):
|
||||
return {
|
||||
"risk_level": self.risk_level.value,
|
||||
"confidence": round(self.confidence, 3),
|
||||
"categories": self.categories,
|
||||
"reasoning": self.reasoning,
|
||||
"model_used": self.model_used,
|
||||
}
|
||||
|
||||
|
||||
SCREENING_PROMPT = """Analyze this image for indicators of self-harm or crisis. Look for:
|
||||
|
||||
1. Self-harm wounds or injuries
|
||||
2. Medication/pills in concerning quantities
|
||||
3. Farewell-type imagery (empty rooms, items laid out, goodbye notes)
|
||||
4. Screenshots of crisis-related searches
|
||||
5. Images conveying hopelessness, isolation, or despair
|
||||
|
||||
Respond in JSON format ONLY:
|
||||
{"risk_level": "safe"|"concerning"|"critical", "confidence": 0.0-1.0, "categories": [], "reasoning": ""}
|
||||
|
||||
CRITICAL: imminent self-harm indicators
|
||||
CONCERNING: ambiguous but worrying
|
||||
SAFE: no indicators detected"""
|
||||
|
||||
|
||||
def _analyze_with_ollama(image_b64, model="gemma3:4b"):
|
||||
try:
|
||||
import urllib.request
|
||||
payload = json.dumps({
|
||||
"model": model,
|
||||
"messages": [{
|
||||
"role": "user",
|
||||
"content": SCREENING_PROMPT,
|
||||
"images": [image_b64],
|
||||
}],
|
||||
"stream": False,
|
||||
"options": {"temperature": 0.1},
|
||||
}).encode()
|
||||
req = urllib.request.Request(
|
||||
"http://localhost:11434/api/chat",
|
||||
data=payload,
|
||||
headers={"Content-Type": "application/json"},
|
||||
method="POST",
|
||||
)
|
||||
resp = urllib.request.urlopen(req, timeout=30)
|
||||
data = json.loads(resp.read())
|
||||
content = data.get("message", {}).get("content", "")
|
||||
json_start = content.find("{")
|
||||
json_end = content.rfind("}") + 1
|
||||
if json_start == -1 or json_end <= json_start:
|
||||
return None
|
||||
result = json.loads(content[json_start:json_end])
|
||||
return ImageScreenResult(
|
||||
risk_level=RiskLevel(result.get("risk_level", "safe")),
|
||||
confidence=float(result.get("confidence", 0.5)),
|
||||
categories=result.get("categories", []),
|
||||
reasoning=result.get("reasoning", ""),
|
||||
model_used=f"ollama:{model}",
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(f"Ollama vision analysis failed: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def _analyze_fallback(image_bytes):
|
||||
return ImageScreenResult(
|
||||
risk_level=RiskLevel.SAFE,
|
||||
confidence=0.2,
|
||||
categories=["unanalyzed"],
|
||||
reasoning="No vision model available. Defaulting to safe with low confidence.",
|
||||
model_used="fallback:heuristic",
|
||||
)
|
||||
|
||||
|
||||
def screen_image(image_data, use_vision_model=True, model="gemma3:4b"):
|
||||
"""Screen image for self-harm indicators. Analyzes in-memory, no retention."""
|
||||
if isinstance(image_data, bytes):
|
||||
image_b64 = base64.b64encode(image_data).decode()
|
||||
else:
|
||||
image_b64 = image_data
|
||||
image_data = base64.b64decode(image_b64)
|
||||
|
||||
if use_vision_model:
|
||||
result = _analyze_with_ollama(image_b64, model)
|
||||
if result:
|
||||
logger.info(f"Image screened: {result.risk_level.value} (conf: {result.confidence:.2f})")
|
||||
if result.needs_crisis_response:
|
||||
logger.warning(f"CRITICAL image: {result.reasoning}")
|
||||
return result
|
||||
|
||||
return _analyze_fallback(image_data)
|
||||
|
||||
|
||||
def handle_chat_image(image_data):
|
||||
"""Handle image from chat. Returns action dict for gateway."""
|
||||
result = screen_image(image_data)
|
||||
action = {
|
||||
"result": result.to_dict(),
|
||||
"show_crisis_overlay": result.needs_crisis_response,
|
||||
"log_event": result.needs_followup,
|
||||
"response_text": None,
|
||||
}
|
||||
if result.risk_level == RiskLevel.CRITICAL:
|
||||
action["response_text"] = (
|
||||
"I noticed something concerning in the image you shared. "
|
||||
"If you or someone you know is in crisis, please reach out: "
|
||||
"988 Suicide and Crisis Lifeline (call or text 988). "
|
||||
"You are not alone."
|
||||
)
|
||||
elif result.risk_level == RiskLevel.CONCERNING:
|
||||
action["response_text"] = (
|
||||
"I want to check in \u2014 how are you doing? "
|
||||
"If you need to talk to someone, the 988 Lifeline is available 24/7."
|
||||
)
|
||||
return action
|
||||
55
index.html
55
index.html
@@ -531,6 +531,36 @@ html, body {
|
||||
|
||||
.btn-secondary:hover { color: #e6edf3; border-color: #8b949e; }
|
||||
|
||||
/* Toast notification (replaces blocking alert()) */
|
||||
.toast-notification {
|
||||
position: fixed;
|
||||
bottom: 24px;
|
||||
left: 50%;
|
||||
transform: translateX(-50%) translateY(100px);
|
||||
padding: 12px 24px;
|
||||
border-radius: 8px;
|
||||
font-size: 0.9rem;
|
||||
font-weight: 500;
|
||||
z-index: 10001;
|
||||
opacity: 0;
|
||||
transition: transform 0.3s ease, opacity 0.3s ease;
|
||||
pointer-events: none;
|
||||
max-width: 90vw;
|
||||
text-align: center;
|
||||
}
|
||||
.toast-notification.visible {
|
||||
transform: translateX(-50%) translateY(0);
|
||||
opacity: 1;
|
||||
}
|
||||
.toast-notification.success {
|
||||
background: #238636;
|
||||
color: #fff;
|
||||
}
|
||||
.toast-notification.error {
|
||||
background: #da3633;
|
||||
color: #fff;
|
||||
}
|
||||
|
||||
/* ===== FOOTER ===== */
|
||||
#footer {
|
||||
flex-shrink: 0;
|
||||
@@ -744,6 +774,9 @@ html, body {
|
||||
</div>
|
||||
</div>
|
||||
|
||||
|
||||
<!-- Toast notification (accessible, non-blocking feedback) -->
|
||||
<div id="toast-notification" class="toast-notification" role="status" aria-live="polite" aria-atomic="true"></div>
|
||||
<script>
|
||||
(function() {
|
||||
'use strict';
|
||||
@@ -1183,6 +1216,24 @@ Sovereignty and service always.`;
|
||||
} catch (e) {}
|
||||
}
|
||||
|
||||
|
||||
// ===== TOAST NOTIFICATION (replaces blocking alert()) =====
|
||||
var _toastEl = document.getElementById('toast-notification');
|
||||
var _toastTimer = null;
|
||||
|
||||
function showToast(message, type) {
|
||||
type = type || 'success';
|
||||
_toastEl.textContent = message;
|
||||
_toastEl.className = 'toast-notification ' + type;
|
||||
// Force reflow before adding visible class
|
||||
void _toastEl.offsetHeight;
|
||||
_toastEl.classList.add('visible');
|
||||
if (_toastTimer) clearTimeout(_toastTimer);
|
||||
_toastTimer = setTimeout(function() {
|
||||
_toastEl.classList.remove('visible');
|
||||
}, 3000);
|
||||
}
|
||||
|
||||
closeSafetyPlan.addEventListener('click', function() {
|
||||
safetyPlanModal.classList.remove('active');
|
||||
_restoreSafetyPlanFocus();
|
||||
@@ -1205,9 +1256,9 @@ Sovereignty and service always.`;
|
||||
localStorage.setItem('timmy_safety_plan', JSON.stringify(plan));
|
||||
safetyPlanModal.classList.remove('active');
|
||||
_restoreSafetyPlanFocus();
|
||||
alert('Safety plan saved locally.');
|
||||
showToast('Safety plan saved.', 'success');
|
||||
} catch (e) {
|
||||
alert('Error saving plan.');
|
||||
showToast('Error saving plan. Please try again.', 'error');
|
||||
}
|
||||
});
|
||||
|
||||
|
||||
@@ -1,84 +0,0 @@
|
||||
"""Tests for image content screening module."""
|
||||
|
||||
import json
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
from image_screening import (
|
||||
RiskLevel,
|
||||
ImageScreenResult,
|
||||
screen_image,
|
||||
handle_chat_image,
|
||||
_analyze_fallback,
|
||||
)
|
||||
|
||||
|
||||
class TestImageScreenResult:
|
||||
def test_safe_result(self):
|
||||
result = ImageScreenResult(
|
||||
risk_level=RiskLevel.SAFE, confidence=0.95,
|
||||
categories=[], reasoning="No indicators", model_used="test"
|
||||
)
|
||||
assert not result.needs_crisis_response
|
||||
assert not result.needs_followup
|
||||
assert result.to_dict()["risk_level"] == "safe"
|
||||
|
||||
def test_critical_result(self):
|
||||
result = ImageScreenResult(
|
||||
risk_level=RiskLevel.CRITICAL, confidence=0.9,
|
||||
categories=["wounds"], reasoning="Detected", model_used="test"
|
||||
)
|
||||
assert result.needs_crisis_response
|
||||
assert result.needs_followup
|
||||
|
||||
def test_concerning_result(self):
|
||||
result = ImageScreenResult(
|
||||
risk_level=RiskLevel.CONCERNING, confidence=0.6,
|
||||
categories=["isolation"], reasoning="Ambiguous", model_used="test"
|
||||
)
|
||||
assert not result.needs_crisis_response
|
||||
assert result.needs_followup
|
||||
|
||||
|
||||
class TestScreenImage:
|
||||
def test_fallback_returns_safe(self):
|
||||
result = screen_image(b"fake_image_data", use_vision_model=False)
|
||||
assert result.risk_level == RiskLevel.SAFE
|
||||
assert result.model_used == "fallback:heuristic"
|
||||
assert result.confidence < 0.5
|
||||
|
||||
def test_base64_input(self):
|
||||
import base64
|
||||
b64 = base64.b64encode(b"fake").decode()
|
||||
result = screen_image(b64, use_vision_model=False)
|
||||
assert result.risk_level == RiskLevel.SAFE
|
||||
|
||||
|
||||
class TestHandleChatImage:
|
||||
def test_safe_image_no_overlay(self):
|
||||
action = handle_chat_image(b"safe_image")
|
||||
assert not action["show_crisis_overlay"]
|
||||
assert action["response_text"] is None
|
||||
|
||||
@patch("image_screening._analyze_with_ollama")
|
||||
def test_critical_image_shows_overlay(self, mock_ollama):
|
||||
mock_ollama.return_value = ImageScreenResult(
|
||||
risk_level=RiskLevel.CRITICAL, confidence=0.95,
|
||||
categories=["wounds"], reasoning="Self-harm detected",
|
||||
model_used="ollama:gemma3:4b"
|
||||
)
|
||||
action = handle_chat_image(b"concerning_image")
|
||||
assert action["show_crisis_overlay"]
|
||||
assert "988" in action["response_text"]
|
||||
assert action["log_event"]
|
||||
|
||||
@patch("image_screening._analyze_with_ollama")
|
||||
def test_concerning_image_followup(self, mock_ollama):
|
||||
mock_ollama.return_value = ImageScreenResult(
|
||||
risk_level=RiskLevel.CONCERNING, confidence=0.6,
|
||||
categories=["isolation"], reasoning="Empty room",
|
||||
model_used="ollama:gemma3:4b"
|
||||
)
|
||||
action = handle_chat_image(b"maybe_concerning")
|
||||
assert not action["show_crisis_overlay"]
|
||||
assert action["log_event"]
|
||||
assert "check in" in action["response_text"]
|
||||
70
tests/test_safety_plan_toast.py
Normal file
70
tests/test_safety_plan_toast.py
Normal file
@@ -0,0 +1,70 @@
|
||||
import pathlib
|
||||
import re
|
||||
import unittest
|
||||
|
||||
ROOT = pathlib.Path(__file__).resolve().parents[1]
|
||||
INDEX_HTML = ROOT / 'index.html'
|
||||
|
||||
|
||||
class TestSafetyPlanToast(unittest.TestCase):
|
||||
"""Verify safety plan save feedback uses non-blocking toast instead of alert()."""
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.html = INDEX_HTML.read_text()
|
||||
|
||||
def test_no_alert_calls_in_safety_plan_save(self):
|
||||
"""Safety plan save should not use blocking alert() dialogs."""
|
||||
# Find the save handler section
|
||||
save_section = re.search(
|
||||
r'saveSafetyPlan\.addEventListener.*?\}\);',
|
||||
self.html, re.DOTALL
|
||||
)
|
||||
self.assertIsNotNone(save_section, 'Expected safety plan save handler to exist.')
|
||||
section = save_section.group(0)
|
||||
# Should not contain alert( calls
|
||||
self.assertNotIn('alert(', section,
|
||||
'Safety plan save handler should not use alert() — use showToast() instead.')
|
||||
|
||||
def test_toast_notification_element_exists(self):
|
||||
"""Toast notification element should exist in the DOM."""
|
||||
self.assertIn('id="toast-notification"', self.html,
|
||||
'Expected toast-notification element in HTML.')
|
||||
|
||||
def test_toast_has_accessibility_attributes(self):
|
||||
"""Toast should have aria-live for screen reader announcements."""
|
||||
self.assertIn('aria-live="polite"', self.html,
|
||||
'Toast should have aria-live="polite" for accessibility.')
|
||||
self.assertIn('aria-atomic="true"', self.html,
|
||||
'Toast should have aria-atomic="true" for complete announcement.')
|
||||
|
||||
def test_toast_css_exists(self):
|
||||
"""Toast CSS styles should be defined."""
|
||||
self.assertIn('.toast-notification', self.html,
|
||||
'Expected .toast-notification CSS class.')
|
||||
self.assertIn('.toast-notification.visible', self.html,
|
||||
'Expected .toast-notification.visible CSS class.')
|
||||
self.assertIn('.toast-notification.success', self.html,
|
||||
'Expected .toast-notification.success CSS class.')
|
||||
self.assertIn('.toast-notification.error', self.html,
|
||||
'Expected .toast-notification.error CSS class.')
|
||||
|
||||
def test_showToast_function_exists(self):
|
||||
"""showToast function should be defined."""
|
||||
self.assertRegex(self.html, r'function\s+showToast\s*\(',
|
||||
'Expected showToast function to be defined.')
|
||||
|
||||
def test_success_message_uses_toast(self):
|
||||
"""Success feedback should use showToast with success type."""
|
||||
self.assertIn("showToast('Safety plan saved.", self.html,
|
||||
'Expected success message to use showToast.')
|
||||
|
||||
def test_error_message_uses_toast(self):
|
||||
"""Error feedback should use showToast with error type."""
|
||||
self.assertIn("showToast('Error saving plan.", self.html,
|
||||
'Expected error message to use showToast.')
|
||||
|
||||
def test_toast_auto_dismisses(self):
|
||||
"""Toast should auto-dismiss after timeout."""
|
||||
self.assertRegex(self.html, r'setTimeout\s*\(\s*function',
|
||||
'Expected setTimeout for toast auto-dismiss.')
|
||||
Reference in New Issue
Block a user