Compare commits
7 Commits
door/issue
...
feat/136-c
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a5eb54161f | ||
| 07c582aa08 | |||
| 5f95dc1e39 | |||
| b1f3cac36d | |||
| 07b3f67845 | |||
| c22bbbaf65 | |||
| 543cb1d40f |
133
crisis/metrics.py
Normal file
133
crisis/metrics.py
Normal file
@@ -0,0 +1,133 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Crisis Metrics CLI — View crisis detection health from the command line.
|
||||
|
||||
Usage:
|
||||
python3 -m crisis.metrics --summary # weekly report
|
||||
python3 -m crisis.metrics --json # raw JSON export
|
||||
python3 -m crisis.metrics --last 24h # last 24 hours
|
||||
|
||||
Ref: #136
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from datetime import datetime, timezone, timedelta
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List
|
||||
|
||||
METRICS_DIR = os.environ.get("CRISIS_METRICS_DIR", str(Path.home() / ".the-door" / "metrics"))
|
||||
|
||||
|
||||
def load_metrics(hours: int = 168) -> List[dict]:
|
||||
"""Load metrics entries from the last N hours."""
|
||||
cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)
|
||||
entries = []
|
||||
metrics_path = Path(METRICS_DIR)
|
||||
|
||||
if not metrics_path.exists():
|
||||
return entries
|
||||
|
||||
for f in sorted(metrics_path.glob("*.json")):
|
||||
try:
|
||||
with open(f) as fh:
|
||||
data = json.load(fh)
|
||||
if isinstance(data, list):
|
||||
entries.extend(data)
|
||||
elif isinstance(data, dict):
|
||||
entries.append(data)
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
# Filter by timestamp
|
||||
filtered = []
|
||||
for e in entries:
|
||||
ts = e.get("timestamp", "")
|
||||
if ts:
|
||||
try:
|
||||
t = datetime.fromisoformat(ts.replace("Z", "+00:00"))
|
||||
if t >= cutoff:
|
||||
filtered.append(e)
|
||||
except Exception:
|
||||
filtered.append(e)
|
||||
|
||||
return filtered
|
||||
|
||||
|
||||
def summarize(entries: List[dict]) -> dict:
|
||||
"""Summarize metrics entries."""
|
||||
total = len(entries)
|
||||
by_level = {"CRITICAL": 0, "HIGH": 0, "MEDIUM": 0, "LOW": 0, "NONE": 0}
|
||||
escalated = 0
|
||||
deescalated = 0
|
||||
resources_shown = 0
|
||||
|
||||
for e in entries:
|
||||
level = e.get("level", "NONE")
|
||||
by_level[level] = by_level.get(level, 0) + 1
|
||||
if e.get("escalated"):
|
||||
escalated += 1
|
||||
if e.get("deescalation_confirmed"):
|
||||
deescalated += 1
|
||||
if e.get("resources_shown"):
|
||||
resources_shown += 1
|
||||
|
||||
return {
|
||||
"period_hours": 168,
|
||||
"total_interactions": total,
|
||||
"by_level": by_level,
|
||||
"escalated_sessions": escalated,
|
||||
"deescalated_sessions": deescalated,
|
||||
"resources_shown": resources_shown,
|
||||
"crisis_rate": round((by_level["CRITICAL"] + by_level["HIGH"]) / max(total, 1) * 100, 1),
|
||||
}
|
||||
|
||||
|
||||
def print_summary(summary: dict):
|
||||
print(f"\n{'='*50}")
|
||||
print(f" CRISIS METRICS SUMMARY")
|
||||
print(f" {datetime.now().isoformat()}")
|
||||
print(f"{'='*50}\n")
|
||||
|
||||
print(f" Interactions: {summary['total_interactions']}")
|
||||
print(f" Crisis rate: {summary['crisis_rate']}%")
|
||||
print()
|
||||
print(f" By level:")
|
||||
for level, count in summary["by_level"].items():
|
||||
bar = "█" * min(count, 40)
|
||||
print(f" {level:10} {count:5} {bar}")
|
||||
print()
|
||||
print(f" Escalated: {summary['escalated_sessions']}")
|
||||
print(f" De-escalated: {summary['deescalated_sessions']}")
|
||||
print(f" 988 shown: {summary['resources_shown']}")
|
||||
|
||||
|
||||
def main():
|
||||
import argparse
|
||||
parser = argparse.ArgumentParser(description="Crisis Metrics CLI")
|
||||
parser.add_argument("--summary", action="store_true", help="Weekly summary")
|
||||
parser.add_argument("--json", action="store_true", help="JSON export")
|
||||
parser.add_argument("--last", default="168h", help="Time window (e.g., 24h, 7d)")
|
||||
args = parser.parse_args()
|
||||
|
||||
# Parse time window
|
||||
last = args.last
|
||||
if last.endswith("h"):
|
||||
hours = int(last[:-1])
|
||||
elif last.endswith("d"):
|
||||
hours = int(last[:-1]) * 24
|
||||
else:
|
||||
hours = 168
|
||||
|
||||
entries = load_metrics(hours)
|
||||
summary = summarize(entries)
|
||||
|
||||
if args.json:
|
||||
print(json.dumps(summary, indent=2))
|
||||
else:
|
||||
print_summary(summary)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -808,6 +808,7 @@ Sovereignty and service always.`;
|
||||
var crisisPanel = document.getElementById('crisis-panel');
|
||||
var crisisOverlay = document.getElementById('crisis-overlay');
|
||||
var overlayDismissBtn = document.getElementById('overlay-dismiss-btn');
|
||||
var overlayCallLink = document.querySelector('.overlay-call');
|
||||
var statusDot = document.querySelector('.status-dot');
|
||||
var statusText = document.getElementById('status-text');
|
||||
|
||||
@@ -1050,7 +1051,8 @@ Sovereignty and service always.`;
|
||||
}
|
||||
}, 1000);
|
||||
|
||||
overlayDismissBtn.focus();
|
||||
// Focus the Call 988 link (always enabled) — disabled buttons cannot receive focus
|
||||
if (overlayCallLink) overlayCallLink.focus();
|
||||
}
|
||||
|
||||
// Register focus trap on document (always listening, gated by class check)
|
||||
|
||||
@@ -52,6 +52,34 @@ class TestCrisisOverlayFocusTrap(unittest.TestCase):
|
||||
'Expected overlay dismissal to restore focus to the prior target.',
|
||||
)
|
||||
|
||||
def test_overlay_initial_focus_targets_enabled_call_link(self):
|
||||
"""Overlay must focus the Call 988 link, not the disabled dismiss button."""
|
||||
# Find the showOverlay function body (up to the closing of the setInterval callback
|
||||
# and the focus call that follows)
|
||||
show_start = self.html.find('function showOverlay()')
|
||||
self.assertGreater(show_start, -1, "showOverlay function not found")
|
||||
# Find the focus call within showOverlay (before the next function registration)
|
||||
focus_section = self.html[show_start:show_start + 2000]
|
||||
self.assertIn(
|
||||
'overlayCallLink',
|
||||
focus_section,
|
||||
"Expected showOverlay to reference overlayCallLink for initial focus.",
|
||||
)
|
||||
# Ensure the old buggy pattern is gone
|
||||
focus_line_region = self.html[show_start + 800:show_start + 1200]
|
||||
self.assertNotIn(
|
||||
'overlayDismissBtn.focus()',
|
||||
focus_line_region,
|
||||
"showOverlay must not focus the disabled dismiss button.",
|
||||
)
|
||||
|
||||
def test_overlay_call_link_variable_is_declared(self):
|
||||
self.assertIn(
|
||||
"querySelector('.overlay-call')",
|
||||
self.html,
|
||||
"Expected a JS reference to the .overlay-call link element.",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
||||
@@ -50,6 +50,22 @@ class TestCrisisOfflinePage(unittest.TestCase):
|
||||
for phrase in required_phrases:
|
||||
self.assertIn(phrase, self.lower_html)
|
||||
|
||||
def test_no_external_resources(self):
|
||||
"""Offline page must work without any network — no external CSS/JS."""
|
||||
import re
|
||||
html = self.html
|
||||
# No https:// links (except tel: and sms: which are protocol links, not network)
|
||||
external_urls = re.findall(r'href=["\']https://|src=["\']https://', html)
|
||||
self.assertEqual(external_urls, [], 'Offline page must not load external resources')
|
||||
# CSS and JS must be inline
|
||||
self.assertIn('<style>', html, 'CSS must be inline')
|
||||
self.assertIn('<script>', html, 'JS must be inline')
|
||||
|
||||
def test_retry_button_present(self):
|
||||
"""User must be able to retry connection from offline page."""
|
||||
self.assertIn('retry-connection', self.html)
|
||||
self.assertIn('Retry connection', self.html)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user