Compare commits
2 Commits
fix/132
...
burn/73-17
| Author | SHA1 | Date | |
|---|---|---|---|
| 8358ad09c3 | |||
| 35174acc19 |
@@ -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
|
||||
50
index.html
50
index.html
@@ -613,6 +613,31 @@ html, body {
|
||||
top: 8px;
|
||||
outline: 2px solid #58a6ff;
|
||||
}
|
||||
|
||||
/* ===== TOAST NOTIFICATION ===== */
|
||||
.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; }
|
||||
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
@@ -744,6 +769,11 @@ 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';
|
||||
@@ -820,6 +850,22 @@ Sovereignty and service always.`;
|
||||
var saveSafetyPlan = document.getElementById('save-safety-plan');
|
||||
var clearChatBtn = document.getElementById('clear-chat-btn');
|
||||
|
||||
|
||||
// ===== TOAST NOTIFICATION =====
|
||||
var _toastEl = document.getElementById('toast-notification');
|
||||
var _toastTimer = null;
|
||||
|
||||
function showToast(message, type) {
|
||||
type = type || 'success';
|
||||
_toastEl.textContent = message;
|
||||
_toastEl.className = 'toast-notification ' + type;
|
||||
void _toastEl.offsetHeight; // force reflow before transition
|
||||
_toastEl.classList.add('visible');
|
||||
if (_toastTimer) clearTimeout(_toastTimer);
|
||||
_toastTimer = setTimeout(function() {
|
||||
_toastEl.classList.remove('visible');
|
||||
}, 3000);
|
||||
}
|
||||
// ===== STATE =====
|
||||
var messages = [];
|
||||
var isStreaming = false;
|
||||
@@ -1205,9 +1251,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 locally.', 'success');
|
||||
} catch (e) {
|
||||
alert('Error saving plan.');
|
||||
showToast('Error saving plan.', '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"]
|
||||
76
tests/test_toast_notification.py
Normal file
76
tests/test_toast_notification.py
Normal file
@@ -0,0 +1,76 @@
|
||||
import pathlib
|
||||
import re
|
||||
import unittest
|
||||
|
||||
ROOT = pathlib.Path(__file__).resolve().parents[1]
|
||||
INDEX_HTML = ROOT / "index.html"
|
||||
|
||||
|
||||
class TestToastNotification(unittest.TestCase):
|
||||
"""Regression tests for toast notification replacing blocking alert(). Issue #73."""
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.html = INDEX_HTML.read_text()
|
||||
|
||||
# -- CSS --
|
||||
def test_toast_css_exists(self):
|
||||
self.assertIn(".toast-notification", self.html,
|
||||
"Expected .toast-notification CSS class.")
|
||||
|
||||
def test_toast_success_error_classes(self):
|
||||
self.assertIn(".toast-notification.success", self.html,
|
||||
"Expected .success variant for green toast.")
|
||||
self.assertIn(".toast-notification.error", self.html,
|
||||
"Expected .error variant for red toast.")
|
||||
|
||||
def test_toast_visible_transition(self):
|
||||
self.assertIn(".toast-notification.visible", self.html,
|
||||
"Expected .visible class to trigger slide-up transition.")
|
||||
|
||||
# -- HTML element --
|
||||
def test_toast_element_exists(self):
|
||||
self.assertIn('id="toast-notification"', self.html,
|
||||
"Expected toast-notification element.")
|
||||
|
||||
def test_toast_aria_live(self):
|
||||
self.assertRegex(self.html,
|
||||
r'aria-live="polite"',
|
||||
"Expected aria-live="polite" for accessible announcements.")
|
||||
|
||||
def test_toast_role_status(self):
|
||||
self.assertRegex(self.html,
|
||||
r'role="status"',
|
||||
"Expected role="status" for toast element.")
|
||||
|
||||
# -- JS function --
|
||||
def test_showToast_function_defined(self):
|
||||
self.assertRegex(self.html,
|
||||
r"function\s+showToast\s*\(",
|
||||
"Expected showToast() function to be defined.")
|
||||
|
||||
def test_showToast_auto_dismiss(self):
|
||||
self.assertRegex(self.html,
|
||||
r"setTimeout.*classList\.remove\(.*visible.*\)",
|
||||
"Expected setTimeout to auto-dismiss toast.")
|
||||
|
||||
# -- alert() replaced --
|
||||
def test_no_alert_in_safety_plan_save(self):
|
||||
lines = self.html.split("\n")
|
||||
for i, line in enumerate(lines):
|
||||
if "alert(" in line:
|
||||
self.fail(
|
||||
f"Blocking alert() still present at line {i+1}: {line.strip()}"
|
||||
)
|
||||
|
||||
def test_showToast_used_for_save_success(self):
|
||||
self.assertIn("showToast('Safety plan saved locally.', 'success')", self.html,
|
||||
"Expected showToast success call for save feedback.")
|
||||
|
||||
def test_showToast_used_for_save_error(self):
|
||||
self.assertIn("showToast('Error saving plan.', 'error')", self.html,
|
||||
"Expected showToast error call for save error feedback.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user