"""
SafeChat — Real-Time Toxic Chat Moderation Dashboard
=====================================================
A comprehensive Streamlit frontend that showcases:
• Live chat moderation with per-category toxicity probabilities
• Gemini 2.0 Flash detoxified alternative suggestions
• Severity badge system (SAFE / LOW / MEDIUM / HIGH)
• Conversation context awareness
• Continuous learning feedback loop
• Session analytics sidebar
Run: streamlit run app.py
"""
import streamlit as st
import uuid
import time
import json
from datetime import datetime
from api_client import check_health, moderate, detoxify, get_feedback_stats, submit_feedback
# ─── Page Configuration ──────────────────────────────────────────────────────
st.set_page_config(
page_title="SafeChat · Real-Time Moderation",
page_icon="🛡️",
layout="wide",
initial_sidebar_state="expanded",
)
# ─── Premium Dark-Mode Glassmorphism CSS ──────────────────────────────────────
CUSTOM_CSS = """
"""
st.markdown(CUSTOM_CSS, unsafe_allow_html=True)
# ─── Session State ────────────────────────────────────────────────────────────
if "messages" not in st.session_state:
st.session_state.messages = []
if "session_id" not in st.session_state:
st.session_state.session_id = str(uuid.uuid4())[:8]
if "total_safe" not in st.session_state:
st.session_state.total_safe = 0
if "total_toxic" not in st.session_state:
st.session_state.total_toxic = 0
if "total_time" not in st.session_state:
st.session_state.total_time = 0
# ─── Helper Functions ─────────────────────────────────────────────────────────
def severity_badge(severity: str) -> str:
cls = f"badge-{severity.lower()}"
return f'{severity}'
def prob_bar(label: str, value: float) -> str:
"""Render a single category probability bar."""
pct = value * 100
if pct > 75:
color = "#ef4444"
elif pct > 50:
color = "#f97316"
elif pct > 30:
color = "#eab308"
else:
color = "#22c55e"
return f"""
"""
def render_categories(categories: dict) -> str:
"""Render all 6 category probability bars."""
nice_names = {
"toxic": "Toxic",
"severe_toxic": "Severe Toxic",
"obscene": "Obscene",
"identity_hate": "Identity Hate",
"insult": "Insult",
"threat": "Threat",
}
bars = ""
for key in ["toxic", "severe_toxic", "obscene", "insult", "identity_hate", "threat"]:
val = categories.get(key, 0.0)
bars += prob_bar(nice_names.get(key, key), val)
return bars
# ─── Sidebar ──────────────────────────────────────────────────────────────────
with st.sidebar:
st.markdown('🛡️ SafeChat
', unsafe_allow_html=True)
st.caption("Context-Aware Multilingual Content Safety Platform")
st.divider()
# Health check
health = check_health()
if health:
st.success("ML Service Online", icon="✅")
else:
st.error("ML Service Offline", icon="🔴")
st.caption("Start the ML service on port 8001")
st.divider()
# Session analytics
st.markdown("### 📊 Session Analytics")
c1, c2 = st.columns(2)
with c1:
st.markdown(f"""
{st.session_state.total_safe}
Safe
""", unsafe_allow_html=True)
with c2:
st.markdown(f"""
{st.session_state.total_toxic}
Toxic
""", unsafe_allow_html=True)
total = st.session_state.total_safe + st.session_state.total_toxic
if total > 0:
safe_pct = (st.session_state.total_safe / total) * 100
st.markdown(f"""
{safe_pct:.0f}%
Safety Rate
""", unsafe_allow_html=True)
if st.session_state.total_time > 0 and total > 0:
avg_ms = st.session_state.total_time / total
st.markdown(f"""
{avg_ms:.0f}ms
Avg Latency
""", unsafe_allow_html=True)
st.divider()
# Model info
st.markdown("### 🧠 Model Stack")
st.markdown("""
- **Classifier:** HingBERT (fine-tuned)
- **Gatekeeper:** HingBERT / TextDetox
- **Detoxifier:** Gemini 2.0 Flash
- **Categories:** 6 multi-label
""")
st.divider()
# Quick test messages
st.markdown("### 🧪 Quick Test Messages")
test_messages = {
"🟢 Safe English": "Hey, how are you doing today?",
"🟢 Safe Hindi": "नमस्ते भाई, कैसे हो?",
"🟢 Safe Hinglish": "Bhai party kab de raha hai tu?",
"🔴 Toxic English": "You're such an idiot, shut up!",
"🔴 Toxic Hindi": "तू बहुत बड़ा बेवकूफ है, चुप रह साले",
"🔴 Toxic Hinglish": "bhenchod bakwas mat kar harami",
}
for label, msg in test_messages.items():
if st.button(label, key=f"test_{label}", use_container_width=True):
st.session_state["_inject_msg"] = msg
st.divider()
if st.button("🗑️ Clear Chat", use_container_width=True):
st.session_state.messages = []
st.session_state.total_safe = 0
st.session_state.total_toxic = 0
st.session_state.total_time = 0
st.rerun()
# ─── Main Content ─────────────────────────────────────────────────────────────
# Header
st.markdown("""
🛡️ SafeChat · Real-Time Moderation
Context-aware multilingual toxicity detection with intent-preserving style transfer
""", unsafe_allow_html=True)
# ─── Render Chat History ──────────────────────────────────────────────────────
for idx, msg in enumerate(st.session_state.messages):
mod = msg.get("moderation")
if mod and mod.get("is_toxic"):
# ── TOXIC MESSAGE ──
flagged = [k for k, v in mod.get("categories", {}).items() if v > 0.3]
flagged_str = ", ".join(flagged) if flagged else "general"
st.markdown(f"""
{msg.get('timestamp', '')} · {mod.get('detected_language', '?')} · {mod.get('inference_time_ms', '?')}ms
{severity_badge(mod.get('severity', 'HIGH'))}
⚠️ {msg['content']}
CATEGORY PROBABILITIES
""", unsafe_allow_html=True)
st.markdown(render_categories(mod.get('categories', dict())), unsafe_allow_html=True)
# Show suggestion
suggestion = mod.get("suggestion") or msg.get("detoxified")
if suggestion:
st.markdown(f"""
✨ Suggested Alternative (Gemini 2.0 Flash):
"{suggestion}"
""", unsafe_allow_html=True)
else:
# ── SAFE MESSAGE ──
lang = mod.get("detected_language", "?") if mod else "?"
latency = mod.get("inference_time_ms", "?") if mod else "?"
score = mod.get("overall_score", 0) if mod else 0
st.markdown(f"""
{msg.get('timestamp', '')} · {lang} · {latency}ms
{severity_badge('SAFE')}
✅ {msg['content']}
Overall toxicity: {score:.2%}
""", unsafe_allow_html=True)
# ─── Chat Input ───────────────────────────────────────────────────────────────
# Check if a test message was injected from sidebar
injected = st.session_state.pop("_inject_msg", None)
prompt = st.chat_input("Type a message in English, Hindi, or Hinglish...")
input_text = injected or prompt
if input_text:
timestamp = datetime.now().strftime("%H:%M:%S")
with st.spinner("🔍 Analyzing message..."):
mod_result = moderate(input_text, context=None)
msg_obj = {
"id": str(uuid.uuid4()),
"content": input_text,
"timestamp": timestamp,
"moderation": mod_result,
}
if mod_result:
if mod_result.get("is_toxic"):
st.session_state.total_toxic += 1
# If suggestion wasn't provided by the /moderate endpoint, explicitly detoxify
if not mod_result.get("suggestion"):
with st.spinner("✨ Generating polite alternative..."):
detox_result = detoxify(input_text, context=context_msgs)
if detox_result and detox_result.get("detoxified"):
msg_obj["detoxified"] = detox_result["detoxified"]
else:
st.session_state.total_safe += 1
st.session_state.total_time += mod_result.get("inference_time_ms", 0)
else:
# API unreachable — still add message
st.session_state.total_safe += 1
st.session_state.messages.append(msg_obj)
st.rerun()