import gc
import json
import os
import re
from datetime import datetime
from pathlib import Path
import streamlit as st
from hatespeech_model import (
predict_hatespeech,
load_model_from_hf,
predict_hatespeech_from_file,
get_rationale_from_mistral,
preprocess_rationale_mistral,
predict_hatespeech_from_file_batched,
merge_subword_token_importance,
get_rationale_from_list,
)
import plotly.graph_objects as go
import pandas as pd
import time
import torch
# ── Page config ───────────────────────────────────────────────────────────────
st.set_page_config(
page_title="🛡️ Hate Speech Detector",
page_icon="🛡️",
layout="wide",
)
# ── Model loading ─────────────────────────────────────────────────────────────
@st.cache_resource
def load_cached_model(model_type="altered"):
model, tokenizer_hatebert, tokenizer_rationale, config, device = load_model_from_hf(model_type=model_type)
return {
"model": model,
"tokenizer_hatebert": tokenizer_hatebert,
"tokenizer_rationale": tokenizer_rationale,
"config": config,
"device": device,
}
def clean_user_input(text):
text = re.sub(r"https?://\S+|www\.\S+", "", text)
text = re.sub(r"[^a-zA-Z0-9\s!]", "", text)
text = re.sub(r"\s+", " ", text).strip()
return text
@st.cache_data
def load_example_samples():
examples_path = Path(__file__).resolve().parent / "hate_samples_sentence_case.csv"
return pd.read_csv(examples_path)
def get_random_example(column_name):
try:
samples_df = load_example_samples()
candidates = samples_df[column_name].dropna().astype(str).str.strip()
candidates = candidates[candidates != ""]
if not candidates.empty:
return candidates.sample(n=1).iloc[0]
except Exception:
pass
fallback_examples = {
"Hate": "You people are worthless and should leave this country!",
"Non-hate": "I disagree with your opinion, but I respect your right to express it.",
"Implicit-hate": "This policy is terrible and will hurt everyone involved.",
}
return fallback_examples.get(column_name, "")
def render_quick_stat(label, value):
st.markdown(
f"""
""",
unsafe_allow_html=True,
)
# ── Custom CSS ────────────────────────────────────────────────────────────────
st.markdown("""
""", unsafe_allow_html=True)
is_basemodel = True
is_enhanced = True
# ── Header ────────────────────────────────────────────────────────────────────
st.markdown('🛡️ Hate Speech Detection System
', unsafe_allow_html=True)
st.markdown('', unsafe_allow_html=True)
# ── Load models ───────────────────────────────────────────────────────────────
with st.spinner("🔄 Loading models... This may take a moment on first run."):
try:
base_data = load_cached_model("base")
enhanced_data = load_cached_model("altered")
base_model = base_data["model"]
base_tokenizer_hatebert = base_data["tokenizer_hatebert"]
base_tokenizer_rationale = base_data["tokenizer_rationale"]
base_config = base_data["config"]
base_device = base_data["device"]
enhanced_model = enhanced_data["model"]
enhanced_tokenizer_hatebert = enhanced_data["tokenizer_hatebert"]
enhanced_tokenizer_rationale = enhanced_data["tokenizer_rationale"]
enhanced_config = enhanced_data["config"]
enhanced_device = enhanced_data["device"]
if not st.session_state.get("models_loaded_toast_shown", False):
st.toast("Base Bert Ensemble and Enhanced Bert Ensemble models loaded successfully!", icon="✅")
st.session_state.models_loaded_toast_shown = True
except Exception as e:
st.error(f"❌ Error loading models: {str(e)}")
st.stop()
# ── Session state initialisation ─────────────────────────────────────────────
if "single_cache" not in st.session_state:
st.session_state.single_cache = None
if "batch_cache" not in st.session_state:
st.session_state.batch_cache = None
if "show_stats" not in st.session_state:
st.session_state.show_stats = True
if "examples_expanded" not in st.session_state:
st.session_state.examples_expanded = True
def _clear_single():
st.session_state.single_cache = None
def _clear_batch():
st.session_state.batch_cache = None
def _clear_all():
st.session_state.single_cache = None
st.session_state.batch_cache = None
# ── Sidebar ───────────────────────────────────────────────────────────────────
with st.sidebar:
st.divider()
st.subheader("⚙️ System Information")
st.markdown(f"**Device:** {'GPU' if torch.cuda.is_available() else 'CPU'}")
st.markdown(f"**GPU Available:** {'Yes' if torch.cuda.is_available() else 'No'}")
try:
def _cgroup_cpu():
# cgroup v2
try:
with open("/sys/fs/cgroup/cpu.max") as f:
quota, period = f.read().strip().split()
if quota != "max":
return max(1, int(quota) // int(period))
except Exception:
pass
# cgroup v1
try:
with open("/sys/fs/cgroup/cpu/cpu.cfs_quota_us") as f:
quota = int(f.read().strip())
with open("/sys/fs/cgroup/cpu/cpu.cfs_period_us") as f:
period = int(f.read().strip())
if quota > 0:
return max(1, quota // period)
except Exception:
pass
return os.cpu_count() or 1
def _cgroup_ram_gb():
# cgroup v2
try:
with open("/sys/fs/cgroup/memory.max") as f:
val = f.read().strip()
if val != "max":
return int(val) / (1024 ** 3)
except Exception:
pass
# cgroup v1
try:
with open("/sys/fs/cgroup/memory/memory.limit_in_bytes") as f:
val = int(f.read().strip())
if val < (1 << 62): # large sentinel = no limit set
return val / (1024 ** 3)
except Exception:
pass
import psutil
return psutil.virtual_memory().total / (1024 ** 3)
_cpu_count = _cgroup_cpu()
_ram_gb = _cgroup_ram_gb()
st.markdown(f"**CPU Cores:** {_cpu_count}")
st.markdown(f"**Total RAM:** {_ram_gb:.1f} GB")
except Exception:
pass
show_stats = st.session_state.show_stats
show_rationale_viz = True
show_probabilities = True
show_details = False
st.divider()
st.subheader("💡 About")
st.markdown("""
Enhanced Bert Ensemble (English) uses:
- **HateBERT** for hate speech understanding
- **Mistral 3.1** for Rationale generation for enhanced context
- **Freezed Bert Base Uncased** for Rationale processing and encoding
- **Temporal CNN** for capturing sequential patterns
- **Multi-Scale CNN** for capturing features at different granularities and high-level semantic patterns
- **Gumbel Sigmoid Token Selector** for identifying important tokens contributing to the prediction
""")
# ── Column ratio helper ───────────────────────────────────────────────────────
col_ratios = [2, 1] if show_stats else [1]
# ── Single-result renderer (reads from session_state.single_cache) ────────────
def render_single_results():
sc = st.session_state.single_cache
if sc is None:
return
res_header_col, clear_col = st.columns([6, 1])
with res_header_col:
st.header("📈 Analysis Results")
with clear_col:
st.write("")
if st.button("🗑️ Clear", key="clear_single_results", help="Clear these results"):
_clear_single()
st.rerun()
st.caption(f"**Input:** {sc['cleaned_input']}")
if sc.get("cleaned_rationale") and sc["cleaned_rationale"] != sc["cleaned_input"]:
with st.expander("📝 Generated Rationale"):
st.write(sc["cleaned_rationale"])
_is_base = sc["is_basemodel"]
_is_enh = sc["is_enhanced"]
if _is_base and _is_enh:
_cols = st.columns(2)
base_col = _cols[0]
enhanced_col = _cols[1]
elif _is_base:
base_col = st.container()
enhanced_col = None
else:
base_col = None
enhanced_col = st.container()
# Base model column
if _is_base:
with base_col:
with st.container(border=True):
st.subheader("🔵 Bert Ensemble")
if sc["base_prediction"] == 1:
st.markdown('🚨 HATE SPEECH DETECTED
', unsafe_allow_html=True)
else:
st.markdown('✅ NOT HATE SPEECH
', unsafe_allow_html=True)
base_tab_metrics, base_tab_probs = st.tabs(["📊 Metrics", "📈 Prediction Probabilities"])
with base_tab_metrics:
st.metric("Confidence", f"{sc['base_confidence']:.1%}")
bm1, bm2 = st.columns(2)
with bm1:
st.metric("Not Hate Speech", f"{sc['base_probabilities'][0]:.1%}")
with bm2:
st.metric("Hate Speech", f"{sc['base_probabilities'][1]:.1%}")
st.metric("Processing Time", f"{sc['base_processing_time']:.3f}s")
with base_tab_probs:
fig_base = go.Figure(data=[go.Bar(
x=["Not Hate Speech", "Hate Speech"],
y=sc["base_probabilities"],
marker_color=["#66bb6a", "#ef5350"],
text=[f"{p:.1%}" for p in sc["base_probabilities"]],
textposition="auto",
)])
fig_base.update_layout(yaxis_title="Probability", yaxis_range=[0, 1], height=300, showlegend=False)
st.plotly_chart(fig_base, use_container_width=True)
if show_details:
with st.expander("View Base Model Outputs"):
st.json(sc["base_technical_details"])
# Enhanced model column
if _is_enh:
with enhanced_col:
with st.container(border=True):
st.subheader("🟢 Enhanced Bert Ensemble")
if sc["enhanced_prediction"] == 1:
st.markdown('🚨 HATE SPEECH DETECTED
', unsafe_allow_html=True)
else:
st.markdown('✅ NOT HATE SPEECH
', unsafe_allow_html=True)
enh_tab_metrics, enh_tab_probs, enh_tab_tokens = st.tabs(["📊 Metrics", "📈 Prediction Probabilities", "🔍 Token Importance"])
with enh_tab_metrics:
st.metric("Confidence", f"{sc['enhanced_confidence']:.1%}")
em1, em2 = st.columns(2)
with em1:
st.metric("Not Hate Speech", f"{sc['enhanced_probabilities'][0]:.1%}")
with em2:
st.metric("Hate Speech", f"{sc['enhanced_probabilities'][1]:.1%}")
st.metric("Processing Time", f"{sc['enhanced_processing_time']:.3f}s")
with enh_tab_probs:
fig_enh = go.Figure(data=[go.Bar(
x=["Not Hate Speech", "Hate Speech"],
y=sc["enhanced_probabilities"],
marker_color=["#66bb6a", "#ef5350"],
text=[f"{p:.1%}" for p in sc["enhanced_probabilities"]],
textposition="auto",
)])
fig_enh.update_layout(yaxis_title="Probability", yaxis_range=[0, 1], height=300, showlegend=False)
st.plotly_chart(fig_enh, use_container_width=True)
with enh_tab_tokens:
if show_rationale_viz and sc.get("enhanced_token_importance"):
st.caption("Highlighted words show which parts influenced the prediction")
html_output = ""
for item in sc["enhanced_token_importance"]:
alpha = min(item["Importance"] * 1.5, 1.0)
color = f"rgba(239, 83, 80, {alpha:.2f})"
html_output += f"{item['Token']} "
html_output += "
"
st.markdown(html_output, unsafe_allow_html=True)
if sc["enhanced_prediction"] == 1:
st.caption("🔴 Darker red = More influence on hate speech detection.")
st.markdown("**📋 Top Important Words**")
df_importance = pd.DataFrame(sc["enhanced_token_importance"])
df_importance = df_importance.sort_values("Importance", ascending=False).head(10)
df_importance["Importance"] = df_importance["Importance"].apply(lambda x: f"{x:.4f}")
st.dataframe(df_importance, use_container_width=True, hide_index=True)
else:
st.info("No token importance data available.")
if show_details:
with st.expander("View Enhanced Model Outputs"):
st.json(sc["enhanced_technical_details"])
# ── Batch-result renderer (reads from session_state.batch_cache) ──────────────
def render_batch_results():
_bc = st.session_state.batch_cache
if _bc is None:
return
_base_result = _bc["base_result"]
_enhanced_result = _bc["enhanced_result"]
_file_rows = _bc["file_rows"]
_is_base = _bc["is_basemodel"]
_is_enh = _bc["is_enhanced"]
_csv_bytes = _bc["csv_bytes"]
_csv_filename = _bc["csv_filename"]
st.success("✅ File analysis complete!")
res_header_col, clear_col = st.columns([6, 1])
with res_header_col:
st.header("📊 Analysis Results - Model Comparison")
with clear_col:
st.write("")
if st.button("🗑️ Clear", key="clear_batch_results", help="Clear these results"):
_clear_batch()
st.rerun()
if _is_base and _is_enh:
_file_cols = st.columns(2)
base_file_col = _file_cols[0]
enhanced_file_col = _file_cols[1]
elif _is_base:
base_file_col = st.container()
enhanced_file_col = None
else:
base_file_col = None
enhanced_file_col = st.container()
if _is_base:
with base_file_col:
with st.container(border=True):
st.subheader("🔵 Bert Ensemble")
base_tab_metrics, base_tab_cm, base_tab_resources, base_tab_perf = st.tabs(["📈 Classification Metrics", "🎯 Confusion Matrix", "⚙️ Resource Usage", "⏱️ Performance"])
with base_tab_metrics:
base_fm1, base_fm2 = st.columns(2)
with base_fm1:
st.metric("F1 Score", f"{_base_result['f1_score']:.4f}")
st.metric("Precision", f"{_base_result['precision']:.4f}")
with base_fm2:
st.metric("Accuracy", f"{_base_result['accuracy']:.4f}")
st.metric("Recall", f"{_base_result['recall']:.4f}")
with base_tab_cm:
base_cm = _base_result["confusion_matrix"]
fig_base_cm = go.Figure(data=go.Heatmap(
z=base_cm, x=["Pred Not Hate", "Pred Hate"], y=["True Not Hate", "True Hate"],
colorscale="Blues", text=base_cm, texttemplate="%{text}", textfont={"size": 14}, showscale=False,
))
fig_base_cm.update_layout(height=300)
st.plotly_chart(fig_base_cm, use_container_width=True)
with base_tab_resources:
base_cpu_col, base_mem_col = st.columns(2)
with base_cpu_col:
st.metric("Total CPU", f"{_base_result['cpu_usage']:.2f}%")
st.metric("Peak CPU", f"{_base_result['peak_cpu_usage']:.2f}%")
with base_mem_col:
st.metric("Allocated Memory", f"{_base_result['memory_usage']:.2f} MB")
st.metric("Peak Memory", f"{_base_result['peak_memory_usage']:.2f} MB")
with base_tab_perf:
st.metric("Total Runtime", f"{_base_result['runtime']:.2f}s")
st.metric("Avg Time/Sample", f"{_base_result['runtime'] / _file_rows:.3f}s")
if _is_enh:
with enhanced_file_col:
with st.container(border=True):
st.subheader("🟢 Enhanced Bert Ensemble")
enh_tab_metrics, enh_tab_cm, enh_tab_resources, enh_tab_perf = st.tabs(["📈 Classification Metrics", "🎯 Confusion Matrix", "⚙️ Resource Usage", "⏱️ Performance"])
with enh_tab_metrics:
enh_fm1, enh_fm2 = st.columns(2)
with enh_fm1:
st.metric("F1 Score", f"{_enhanced_result['f1_score']:.4f}")
st.metric("Precision", f"{_enhanced_result['precision']:.4f}")
with enh_fm2:
st.metric("Accuracy", f"{_enhanced_result['accuracy']:.4f}")
st.metric("Recall", f"{_enhanced_result['recall']:.4f}")
with enh_tab_cm:
enhanced_cm = _enhanced_result["confusion_matrix"]
fig_enhanced_cm = go.Figure(data=go.Heatmap(
z=enhanced_cm, x=["Pred Not Hate", "Pred Hate"], y=["True Not Hate", "True Hate"],
colorscale="Greens", text=enhanced_cm, texttemplate="%{text}", textfont={"size": 14}, showscale=False,
))
fig_enhanced_cm.update_layout(height=300)
st.plotly_chart(fig_enhanced_cm, use_container_width=True)
with enh_tab_resources:
enh_cpu_col, enh_mem_col = st.columns(2)
with enh_cpu_col:
st.metric("Total CPU", f"{_enhanced_result['cpu_usage']:.2f}%")
st.metric("Peak CPU", f"{_enhanced_result['peak_cpu_usage']:.2f}%")
with enh_mem_col:
st.metric("Allocated Memory", f"{_enhanced_result['memory_usage']:.2f} MB")
st.metric("Peak Memory", f"{_enhanced_result['peak_memory_usage']:.2f} MB")
with enh_tab_perf:
st.metric("Total Runtime", f"{_enhanced_result['runtime']:.2f}s")
st.metric("Avg Time/Sample", f"{_enhanced_result['runtime'] / _file_rows:.3f}s")
st.download_button(
"↓ Download Additional Results",
data=_csv_bytes,
file_name=_csv_filename,
mime="text/csv",
use_container_width=False,
help="Download per-sample predictions as CSV",
key=f"download_results_csv_{_csv_filename}",
)
# ── Tabs ──────────────────────────────────────────────────────────────────────
tab_text, tab_file = st.tabs(["📝 Text Analysis", "📂 File Upload"])
# ════════════════════════════════════════════════════════════════════════════════
# TAB 1 — Text Analysis
# ════════════════════════════════════════════════════════════════════════════════
with tab_text:
_col_ratio = [2, 1] if st.session_state.show_stats else [9, 1]
input_col, stats_col = st.columns(_col_ratio)
with input_col:
st.subheader("📝 Input Text")
user_input = st.text_area(
"Enter text to analyze:",
placeholder="Type or paste text here to check for hate speech...",
height=200,
help="Enter up to 128 words",
key="user_input_area",
)
def onClick():
st.session_state.user_input_area = ""
st.session_state.pop("optional_rationale_area", None)
btn_col1, btn_col2 = st.columns([4, 1])
with btn_col1:
analyze_text_btn = st.button("🔍 Analyze Text", type="primary", use_container_width=True, key="analyze_text")
with btn_col2:
st.button("🗑️ Clear", use_container_width=True, on_click=onClick, key="clear_text_input")
with stats_col:
st.write("")
st.write("")
toggle_label = "📊 Quick Stats" if st.session_state.show_stats else "📊"
if st.button(toggle_label, key="stats_toggle_text", use_container_width=True):
st.session_state.show_stats = not st.session_state.show_stats
st.rerun()
if st.session_state.show_stats:
current_text = user_input or ""
word_count = len(current_text.split())
char_count = len(current_text)
stat_col1, stat_col2 = st.columns(2)
with stat_col1:
render_quick_stat("Words", word_count)
with stat_col2:
render_quick_stat("Characters", char_count)
# Examples section
with st.expander("💡 Try Example Texts", expanded=st.session_state.examples_expanded):
ex_col1, ex_col2 = st.columns(2)
with ex_col1:
with st.container(border=True):
st.markdown("🚨 **Hate Speech**")
st.caption("Load a random hate speech sample")
if st.button("🎲 Random Example", key="ex_hate", use_container_width=True, type="primary"):
st.session_state.example_text = get_random_example("Hate")
st.session_state.example_type = "Hate"
st.rerun()
with ex_col2:
with st.container(border=True):
st.markdown("✅ **Non-Hate Speech**")
st.caption("Load a random non-hate speech sample")
if st.button("🎲 Random Example", key="ex_nonhate", use_container_width=True):
st.session_state.example_text = get_random_example("Non-hate")
st.session_state.example_type = "Non-hate"
st.rerun()
if st.session_state.get("example_text"):
st.divider()
_ex_type = st.session_state.get("example_type", "")
_badge = "🚨 Hate Speech" if _ex_type == "Hate" else "✅ Non-Hate Speech"
st.markdown(f"**Loaded · {_badge}**")
st.code(st.session_state.example_text, language=None)
st.caption("↑ Copy and paste into the input box above, then click Analyze Text")
if analyze_text_btn:
if not user_input or not user_input.strip():
st.toast("Please enter some text to analyze.", icon="⚠️")
elif not is_basemodel and not is_enhanced:
st.warning("⚠️ Please enable at least one model in the sidebar.")
else:
st.session_state.examples_expanded = False
cleaned_input = clean_user_input(user_input)
with st.spinner("🔄 Generating rationale from Mistral AI..."):
try:
raw_rationale = get_rationale_from_mistral(cleaned_input)
cleaned_rationale = preprocess_rationale_mistral(raw_rationale)
print(f"Raw rationale from Mistral: {raw_rationale}")
except Exception as e:
st.error(f"❌ Error generating/processing rationale: {str(e)}")
cleaned_rationale = cleaned_input
with st.spinner("🔄 Analyzing text with models..."):
enhanced_prediction = enhanced_confidence = enhanced_probabilities = None
enhanced_processing_time = enhanced_token_importance = enhanced_technical_details = None
base_prediction = base_confidence = base_probabilities = None
base_processing_time = base_technical_details = None
if is_enhanced:
enhanced_start = time.time()
enhanced_model_result = predict_hatespeech(
text=cleaned_input,
rationale=cleaned_rationale,
model=enhanced_model,
tokenizer_hatebert=enhanced_tokenizer_hatebert,
tokenizer_rationale=enhanced_tokenizer_rationale,
config=enhanced_config,
device=enhanced_device,
model_type="altered",
)
enhanced_end = time.time()
enhanced_prediction = enhanced_model_result["prediction"]
enhanced_confidence = enhanced_model_result["confidence"]
enhanced_probabilities = enhanced_model_result["probabilities"]
enhanced_rationale_scores = enhanced_model_result["rationale_scores"]
enhanced_tokens = enhanced_model_result["tokens"]
enhanced_token_importance = merge_subword_token_importance(enhanced_tokens, enhanced_rationale_scores)
enhanced_processing_time = enhanced_end - enhanced_start
enhanced_technical_details = {
"prediction": int(enhanced_prediction),
"confidence": float(enhanced_confidence),
"probability_not_hate": float(enhanced_probabilities[0]),
"probability_hate": float(enhanced_probabilities[1]),
"num_tokens": len([t for t in enhanced_tokens if t not in ["[CLS]", "[SEP]", "[PAD]"]]),
"device": "cpu",
"model_config": {
"max_length": str(enhanced_config.get("max_length", 128)),
"cnn_filters": "128",
},
}
if is_basemodel:
base_start = time.time()
base_model_result = predict_hatespeech(
text=cleaned_input,
rationale=cleaned_rationale,
model=base_model,
tokenizer_hatebert=base_tokenizer_hatebert,
tokenizer_rationale=base_tokenizer_rationale,
config=base_config,
device=base_device,
model_type="base",
)
base_end = time.time()
base_prediction = base_model_result["prediction"]
base_confidence = base_model_result["confidence"]
base_probabilities = base_model_result["probabilities"]
base_processing_time = base_end - base_start
base_technical_details = {
"prediction": int(base_prediction),
"confidence": float(base_confidence),
"probability_not_hate": float(base_probabilities[0]),
"probability_hate": float(base_probabilities[1]),
"device": "cpu",
"model_config": {"max_length": str(base_config.get("max_length", 512))},
}
gc.collect()
# Store in session_state + persist to disk
st.session_state.single_cache = {
"cleaned_input": cleaned_input,
"cleaned_rationale": cleaned_rationale,
"is_basemodel": is_basemodel,
"is_enhanced": is_enhanced,
"base_prediction": int(base_prediction) if base_prediction is not None else None,
"base_confidence": float(base_confidence) if base_confidence is not None else None,
"base_probabilities": [float(p) for p in base_probabilities] if base_probabilities is not None else None,
"base_processing_time": float(base_processing_time) if base_processing_time is not None else None,
"base_technical_details": base_technical_details,
"enhanced_prediction": int(enhanced_prediction) if enhanced_prediction is not None else None,
"enhanced_confidence": float(enhanced_confidence) if enhanced_confidence is not None else None,
"enhanced_probabilities": [float(p) for p in enhanced_probabilities] if enhanced_probabilities is not None else None,
"enhanced_processing_time": float(enhanced_processing_time) if enhanced_processing_time is not None else None,
"enhanced_token_importance": enhanced_token_importance,
"enhanced_technical_details": enhanced_technical_details,
}
# Render persisted single results (every run)
st.divider()
render_single_results()
# ════════════════════════════════════════════════════════════════════════════════
# TAB 2 — File Upload
# ════════════════════════════════════════════════════════════════════════════════
with tab_file:
uploaded_file = None
file_content = None
_file_col_ratio = [2, 1] if st.session_state.show_stats else [9, 1]
upload_col, file_stats_col = st.columns(_file_col_ratio)
with upload_col:
st.subheader("📂 Upload File")
uploaded_file = st.file_uploader(
"Choose a CSV file to analyze:",
type=["csv"],
help="Upload a CSV file containing the content you want to analyze for hate speech",
)
if uploaded_file is not None:
try:
file_content = pd.read_csv(uploaded_file)
st.success("✅ File loaded successfully!")
except Exception as e:
st.error(f"❌ Error reading file: {str(e)}")
with file_stats_col:
st.write("")
st.write("")
toggle_label = "📊 Quick Stats" if st.session_state.show_stats else "📊"
if st.button(toggle_label, key="stats_toggle_file", use_container_width=True):
st.session_state.show_stats = not st.session_state.show_stats
st.rerun()
if st.session_state.show_stats:
if uploaded_file is not None and file_content is not None:
st.markdown(f"**Filename:** {uploaded_file.name}")
st.markdown(f"**Size:** {uploaded_file.size / 1024:.2f} KB")
render_quick_stat("Rows in File", len(file_content))
else:
st.info("Upload a CSV file to see statistics.")
analyze_file_btn = st.button("🔍 Analyze File", type="primary", use_container_width=False, key="analyze_file")
if analyze_file_btn:
if uploaded_file is None or file_content is None:
st.toast("Please upload a CSV file first.", icon="⚠️")
elif not is_basemodel and not is_enhanced:
st.warning("⚠️ Please enable at least one model in the sidebar.")
else:
file_rows = len(file_content)
st.markdown("**Preview:**")
st.dataframe(file_content.head(3), use_container_width=True)
with st.spinner("🔄 Analyzing file with models... This may take a while for large files."):
rationales = None
if "CF_Rationales" in file_content.columns and file_content["CF_Rationales"].notna().any():
rationales = file_content["CF_Rationales"].tolist()
if rationales is None:
rationales = get_rationale_from_list(file_content["text"].tolist())
enhanced_result = None
base_result = None
if is_enhanced:
enhanced_result = predict_hatespeech_from_file_batched(
text_list=file_content["text"].tolist(),
rationale_list=rationales,
true_label=file_content["label"].tolist(),
model=enhanced_model,
tokenizer_hatebert=enhanced_tokenizer_hatebert,
tokenizer_rationale=enhanced_tokenizer_rationale,
config=enhanced_config,
device=enhanced_device,
model_type="altered",
)
if is_basemodel:
base_result = predict_hatespeech_from_file(
text_list=file_content["text"].tolist(),
rationale_list=rationales,
true_label=file_content["label"].tolist(),
model=base_model,
tokenizer_hatebert=base_tokenizer_hatebert,
tokenizer_rationale=base_tokenizer_rationale,
config=base_config,
device=base_device,
model_type="base",
)
gc.collect()
# Build CSV rows
rows_for_csv = []
if is_basemodel and is_enhanced:
for base_row, enh_row in zip(base_result["per_sample_results"], enhanced_result["per_sample_results"]):
rows_for_csv.append({
"text": enh_row["text"],
"true_label": enh_row["true_label"],
"rationale": enh_row["rationale"],
"base_prediction": base_row["predicted_label"],
"enhanced_prediction": enh_row["predicted_label"],
"token_importance": json.dumps(enh_row["token_importance"]) if enh_row["token_importance"] else None,
})
elif is_enhanced:
for enh_row in enhanced_result["per_sample_results"]:
rows_for_csv.append({
"text": enh_row["text"],
"true_label": enh_row["true_label"],
"rationale": enh_row["rationale"],
"enhanced_prediction": enh_row["predicted_label"],
"token_importance": json.dumps(enh_row["token_importance"]) if enh_row["token_importance"] else None,
})
elif is_basemodel:
for base_row in base_result["per_sample_results"]:
rows_for_csv.append({
"text": base_row["text"],
"true_label": base_row["true_label"],
"rationale": base_row["rationale"],
"base_prediction": base_row["predicted_label"],
})
df_per_sample = pd.DataFrame(rows_for_csv)
csv_bytes = df_per_sample.to_csv(index=False).encode("utf-8")
csv_filename = f"batch_predictions_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
# Build export payload
file_export_payload = {
"mode": "batch",
"filename": uploaded_file.name,
"rows": int(file_rows),
"models": {},
}
if is_basemodel:
file_export_payload["models"]["base"] = {
"label": "Base Bert Ensemble",
"f1_score": float(base_result["f1_score"]),
"precision": float(base_result["precision"]),
"accuracy": float(base_result["accuracy"]),
"recall": float(base_result["recall"]),
"confusion_matrix": base_result["confusion_matrix"],
"cpu_usage": float(base_result["cpu_usage"]),
"peak_cpu_usage": float(base_result["peak_cpu_usage"]),
"memory_usage": float(base_result["memory_usage"]),
"peak_memory_usage": float(base_result["peak_memory_usage"]),
"runtime": float(base_result["runtime"]),
"avg_time_per_sample": float(base_result["runtime"] / file_rows),
}
if is_enhanced:
file_export_payload["models"]["enhanced"] = {
"label": "Enhanced Bert Ensemble",
"f1_score": float(enhanced_result["f1_score"]),
"precision": float(enhanced_result["precision"]),
"accuracy": float(enhanced_result["accuracy"]),
"recall": float(enhanced_result["recall"]),
"confusion_matrix": enhanced_result["confusion_matrix"],
"cpu_usage": float(enhanced_result["cpu_usage"]),
"peak_cpu_usage": float(enhanced_result["peak_cpu_usage"]),
"memory_usage": float(enhanced_result["memory_usage"]),
"peak_memory_usage": float(enhanced_result["peak_memory_usage"]),
"runtime": float(enhanced_result["runtime"]),
"avg_time_per_sample": float(enhanced_result["runtime"] / file_rows),
}
st.session_state.batch_cache = {
"base_result": base_result,
"enhanced_result": enhanced_result,
"file_rows": file_rows,
"is_basemodel": is_basemodel,
"is_enhanced": is_enhanced,
"csv_bytes": csv_bytes,
"csv_filename": csv_filename,
"file_export_payload": file_export_payload,
}
# Render persisted batch results (every run)
st.divider()
render_batch_results()
# ── Footer ────────────────────────────────────────────────────────────────────
st.divider()
st.markdown("""
Enhanced Bert Ensemble For Interpretable Hate Speech Detection
""", unsafe_allow_html=True)