"""Streamlit demo - IndoBERT (paper repro) vs IBT-hybrid (proposed). Run from project root: streamlit run app.py Two tabs: * **Test set viewer** - paged through the 299 held-out test comments with pre-computed predictions from IndoBERT + 3 IBT-hybrid variants (``results/demo/test_predictions_*.json``). No model loading needed; works off-line as long as the JSONs are committed. * **Try your own** - live inference. Loads IndoBERT + the selected IBT variant from local ``models//`` (synced from Drive after the Colab notebooks run). Falls back to a notice if checkpoints are missing. The full cross-model benchmark report (per-model overview / fairness / multi-seed / tuning) is archived as a standalone script at ``analysis/benchmark_report.py`` - run separately; kept out of this demo. """ from __future__ import annotations import json import logging import os import sys import warnings from pathlib import Path os.environ.setdefault("TRANSFORMERS_VERBOSITY", "error") warnings.filterwarnings("ignore") logging.getLogger("huggingface_hub").setLevel(logging.ERROR) import numpy as np import pandas as pd import streamlit as st ROOT = Path(__file__).resolve().parent if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from src.paths import DEMO_DIR, METRICS_DIR, MODELS_DIR from src.preprocessing import preprocess_minimal logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Registry - IndoBERT (baseline) + 3 IBT-hybrid variants (proposed) # --------------------------------------------------------------------------- # The ``tag`` field is the local directory name under ``models/`` (which mirrors # the Drive-synced ``thesis-cyberbullying/models//``) and the suffix of the # corresponding ``results/demo/test_predictions_.json`` file. INDOBERT = { "name": "IndoBERT (paper repro)", "tag": "indobert_tiktok", "category": "Transformer (baseline)", "demo_json": DEMO_DIR / "test_predictions_indobert.json", } IBT_VARIANTS = { "IBT-CNN-BiLSTM (proposed, best)": { "tag": "ibt_cnnbilstm", "demo_json": DEMO_DIR / "test_predictions_ibt_cnnbilstm.json", }, "IBT-BiLSTM (proposed)": { "tag": "ibt_bilstm", "demo_json": DEMO_DIR / "test_predictions_ibt_bilstm.json", }, "IBT-CNN (proposed)": { "tag": "ibt_cnn", "demo_json": DEMO_DIR / "test_predictions_ibt_cnn.json", }, } DISCLAIMER_MD = """ Academic demo for model comparison on Indonesian TikTok cyberbullying detection. **Not a production moderation tool.** Predictions reflect model outputs, not editorial judgement. Single-prediction use for content moderation, sanctions, or other consequential decisions is **not recommended**. """ ABOUT_MD = """ **Project:** A research demo on single-stage binary cyberbullying detection for Indonesian TikTok comments. It reproduces an IndoBERT baseline and proposes an IBT-hybrid family that pairs a Twitter-domain encoder with a richer classification head, with the goal of testing whether these two changes improve detection on noisy, slang-rich text. **Dataset:** Cyberbullying Bahasa Indonesia, with Slang ([Kaggle](https://www.kaggle.com/datasets/hushian/cyberbullying-dataset-with-slang), CC0). 2,109 raw rows; 1,993 after mojibake fix and deduplication. **Baseline reference:** Jayanti & Rohman (2026), *Journal of Information Systems and Informatics* 8(1), IndoBERT with fairness evaluation. **Proposed:** IBT-hybrid family (IndoBERTweet backbone with CNN, BiLSTM, and stacked CNN+BiLSTM heads), tuned over a 720-config grid and re-fitted across five random seeds. **Label convention:** `0 = cyberbullying`, `1 = non-cyberbullying` (per Kaggle). """ # --------------------------------------------------------------------------- # Cached loaders (JSON for viewer, model checkpoints for live inference) # --------------------------------------------------------------------------- @st.cache_data(show_spinner="Loading test-set predictions...") def _load_demo_payload(path_str: str) -> dict | None: """Cache the per-row predictions JSON. ``None`` if the file is missing.""" p = Path(path_str) if not p.exists(): return None return json.loads(p.read_text(encoding="utf-8")) @st.cache_data(show_spinner=False) def _load_metrics(path_str: str) -> dict | None: p = Path(path_str) if not p.exists(): return None return json.loads(p.read_text(encoding="utf-8")) @st.cache_resource(show_spinner="Loading IndoBERT checkpoint...") def _load_indobert(): from src.models.indobert import IndoBERTModel ckpt = MODELS_DIR / INDOBERT["tag"] if not ckpt.exists(): raise FileNotFoundError(f"Model checkpoint not available: {ckpt}.") return IndoBERTModel.load(ckpt) @st.cache_resource(show_spinner="Loading IBT-hybrid checkpoint...") def _load_ibt(tag: str): if tag == "ibt_cnn": from src.models.indobertweet_cnn import IBTCNNModel as cls elif tag == "ibt_bilstm": from src.models.indobertweet_bilstm import IBTBiLSTMModel as cls elif tag == "ibt_cnnbilstm": from src.models.indobertweet_cnn_bilstm import IBTCNNBiLSTMModel as cls else: raise ValueError(f"Unknown IBT variant tag: {tag}") ckpt = MODELS_DIR / tag if not ckpt.exists(): raise FileNotFoundError(f"Model checkpoint not available: {ckpt}.") return cls.load(ckpt) def _ensure_checkpoints() -> None: """Download checkpoints into ``models/`` from a HuggingFace model repo. Cloud deploy (HF Spaces) ships the ~1.75GB checkpoints out-of-band: set ``HF_CHECKPOINTS_REPO=/`` (and ``HF_TOKEN`` if private). On first boot this pulls them in; with persistent storage it runs once. No-op when the env var is unset (local runs already have ``models/``) or files are present. """ repo = os.environ.get("HF_CHECKPOINTS_REPO") if not repo: return tags = [INDOBERT["tag"], *[v["tag"] for v in IBT_VARIANTS.values()]] if all((MODELS_DIR / t).exists() for t in tags): return from huggingface_hub import snapshot_download with st.spinner(f"Fetching checkpoints from {repo} (first boot only)..."): snapshot_download( repo_id=repo, repo_type="model", local_dir=str(MODELS_DIR), token=os.environ.get("HF_TOKEN"), ) def _warm_all_models(status_cb=None) -> list[str]: """Pre-load every checkpoint into ``st.cache_resource`` + run one dummy forward pass, so the first live prediction during a demo is instant. ``st.cache_resource`` is process-wide and shared across browser sessions, so warming once keeps every model hot for all later visitors. Returns the list of display names whose checkpoint is missing (skipped, not fatal). """ jobs: list[tuple[str, str, str | None]] = [ (INDOBERT["name"], "indobert", None), *[(name, "ibt", spec["tag"]) for name, spec in IBT_VARIANTS.items()], ] failed: list[str] = [] for i, (name, kind, tag) in enumerate(jobs): if status_cb: status_cb(i, len(jobs), name) try: model = _load_indobert() if kind == "indobert" else _load_ibt(tag) _predict(model, "halo") # warm the forward path (JIT/alloc) except FileNotFoundError: failed.append(name) return failed # --------------------------------------------------------------------------- # Inference helpers (Pipeline A - preprocess_minimal for both families) # --------------------------------------------------------------------------- def _predict(model, text: str) -> dict: processed = preprocess_minimal(text) proba = np.asarray(model.predict_proba([processed])[0]) # Match the offline `predict()`: threshold 0.5 on P(non-CB). proba[0]=CB, # proba[1]=non-CB and the two sum to 1, so label is just `proba[1] >= 0.5`. pred = int(proba[1] >= 0.5) return { "prediction": pred, "label": "cyberbullying" if pred == 0 else "non-cyberbullying", "confidence": float(np.max(proba)), "p_cb": float(proba[0]), "p_non_cb": float(proba[1]), "preprocessed": processed, } # --------------------------------------------------------------------------- # Test-set viewer (Phase B - pre-computed predictions) # --------------------------------------------------------------------------- def _build_viewer_frame(indobert_blob: dict, ibt_blob: dict) -> pd.DataFrame: """Inner-join IndoBERT + IBT predictions row-by-row on ``idx``.""" indo_rows = {r["idx"]: r for r in indobert_blob["rows"]} ibt_rows = {r["idx"]: r for r in ibt_blob["rows"]} common = sorted(set(indo_rows) & set(ibt_rows)) rows = [] for i in common: a, b = indo_rows[i], ibt_rows[i] rows.append({ "idx": i, "komentar": a["komentar"], "y_true": a["y_true"], "indobert_pred": a["y_pred"], "indobert_p_cb": a["p_cb"], "ibt_pred": b["y_pred"], "ibt_p_cb": b["p_cb"], "agree": a["y_pred"] == b["y_pred"], "indobert_correct": a["y_pred"] == a["y_true"], "ibt_correct": b["y_pred"] == b["y_true"], }) return pd.DataFrame(rows) def _filter_frame(df: pd.DataFrame, mode: str) -> pd.DataFrame: if mode == "All": return df if mode == "Both agree": return df[df["agree"]] if mode == "Disagree": return df[~df["agree"]] if mode == "Both wrong": return df[(~df["indobert_correct"]) & (~df["ibt_correct"])] if mode == "Only IndoBERT wrong": return df[(~df["indobert_correct"]) & (df["ibt_correct"])] if mode == "Only IBT wrong": return df[(df["indobert_correct"]) & (~df["ibt_correct"])] return df def _label_pill(pred: int, correct: bool) -> str: """Compact label + agreement marker, rendered as HTML.""" label = "CB" if pred == 0 else "non-CB" bg = "#fde2e2" if pred == 0 else "#e3f5e1" fg = "#a01919" if pred == 0 else "#1b5e20" check = "✓" if correct else "✗" return (f"{label} {check}") def _render_row(row: pd.Series, ibt_name: str) -> None: true_label = "cyberbullying" if row["y_true"] == 0 else "non-cyberbullying" st.markdown(f"**Comment idx {int(row['idx'])}** · ground truth: `{true_label}`") st.markdown(f"> {row['komentar']}") cols = st.columns(2) with cols[0]: st.markdown(f"**IndoBERT**: " + _label_pill(int(row["indobert_pred"]), bool(row["indobert_correct"])), unsafe_allow_html=True) st.caption(f"P(CB) = {row['indobert_p_cb']:.4f}") with cols[1]: st.markdown(f"**{ibt_name}**: " + _label_pill(int(row["ibt_pred"]), bool(row["ibt_correct"])), unsafe_allow_html=True) st.caption(f"P(CB) = {row['ibt_p_cb']:.4f}") st.markdown( f"agreement: {'✓ both agree' if row['agree'] else '✗ disagree'}", unsafe_allow_html=True, ) def _tab_viewer() -> None: st.subheader("Test set viewer - IndoBERT vs IBT-hybrid") st.caption( "299 held-out test comments (70/15/15 stratified split, seed=42). " "Predictions are pre-computed, so this view loads instantly." ) ibt_choice = st.selectbox("IBT-hybrid variant", list(IBT_VARIANTS.keys()), index=0) ibt_spec = IBT_VARIANTS[ibt_choice] indo_blob = _load_demo_payload(str(INDOBERT["demo_json"])) ibt_blob = _load_demo_payload(str(ibt_spec["demo_json"])) if indo_blob is None or ibt_blob is None: missing = [str(p) for p, b in [(INDOBERT["demo_json"], indo_blob), (ibt_spec["demo_json"], ibt_blob)] if b is None] st.error("Pre-computed predictions are unavailable: " + ", ".join(missing) + ".") return df = _build_viewer_frame(indo_blob, ibt_blob) # Headline counts n = len(df) n_agree = int(df["agree"].sum()) n_both_right = int((df["indobert_correct"] & df["ibt_correct"]).sum()) n_only_ibt = int((df["ibt_correct"] & (~df["indobert_correct"])).sum()) n_only_indo = int(((~df["ibt_correct"]) & df["indobert_correct"]).sum()) n_both_wrong = int(((~df["ibt_correct"]) & (~df["indobert_correct"])).sum()) mcols = st.columns(5) mcols[0].metric("rows", n) mcols[1].metric("agree", f"{n_agree} ({n_agree/n*100:.1f}%)") mcols[2].metric("both correct", n_both_right) mcols[3].metric("only IBT correct", n_only_ibt) mcols[4].metric("only IndoBERT correct", n_only_indo) st.caption(f"both wrong: {n_both_wrong}") # Filter + table-of-contents filt = st.radio( "Filter", ["All", "Both agree", "Disagree", "Both wrong", "Only IndoBERT wrong", "Only IBT wrong"], horizontal=True, index=2, ) view = _filter_frame(df, filt).reset_index(drop=True) if view.empty: st.info("No rows match this filter.") return page_size = 10 n_pages = (len(view) - 1) // page_size + 1 page = st.number_input( f"Page (1–{n_pages}, {len(view)} rows)", min_value=1, max_value=n_pages, value=1, step=1, ) lo, hi = (page - 1) * page_size, page * page_size for _, row in view.iloc[lo:hi].iterrows(): _render_row(row, ibt_choice.split(" ")[0]) st.divider() # --------------------------------------------------------------------------- # Try-your-own (Phase A - live inference, needs checkpoints) # --------------------------------------------------------------------------- SAMPLE_COMMENTS: dict[str, str] = { "(none)": "", "Clear CB - hate speech": '"Babi g pantes nyentuh anjingnya bidadari, nanti kotor."', "Clear non-CB - positive": '"smga semua kelg dan crew sehat semua.. dan rejeki trs lancar ya kak gigi.."', "Slang-heavy": "pretttt bgt lo bang", "Math-bold styled (synthetic)": "𝐬𝐢𝐥𝐚𝐧𝐠 𝐛𝐞𝐝𝐚 𝐛𝐧𝐞𝐫 ke langit dan tanah", } def _render_live_card(label: str, model_name: str, result: dict) -> None: pct = result["confidence"] * 100 if result["prediction"] == 0: st.error(f"🚨 **CYBERBULLYING** ({pct:.1f}% confidence)") else: st.success(f"✅ **NON-CYBERBULLYING** ({pct:.1f}% confidence)") st.caption(f"{label}: **{model_name}**") c = st.columns(2) c[0].metric("P(CB)", f"{result['p_cb']*100:.2f}%") c[1].metric("P(non-CB)", f"{result['p_non_cb']*100:.2f}%") def _tab_try() -> None: st.subheader("Try your own comment") st.caption( "Live inference: IndoBERT (paper baseline) vs your chosen IBT-hybrid " "variant, side by side." ) ibt_choice = st.selectbox( "IBT-hybrid variant", list(IBT_VARIANTS.keys()), index=0, key="ibt_live", ) ibt_spec = IBT_VARIANTS[ibt_choice] input_col, sample_col = st.columns([3, 1]) with sample_col: sample_label = st.selectbox("Sample preset", list(SAMPLE_COMMENTS.keys()), index=0) with input_col: text_input = st.text_area( "Comment to analyze:", value=SAMPLE_COMMENTS[sample_label], height=140, key=f"text_input_{sample_label}", placeholder="Paste an Indonesian comment here, or pick a sample on the right.", ) if not st.button("Analyze", type="primary"): return text = (text_input or "").strip() if not text: st.warning("Please enter or select a comment first.") return try: with st.spinner("Loading IndoBERT..."): indobert = _load_indobert() with st.spinner(f"Loading {ibt_choice}..."): ibt = _load_ibt(ibt_spec["tag"]) except FileNotFoundError as exc: st.error(str(exc)) return with st.spinner("Predicting..."): a = _predict(indobert, text) b = _predict(ibt, text) cols = st.columns(2) with cols[0]: _render_live_card("Baseline", "IndoBERT", a) with cols[1]: _render_live_card("Proposed", ibt_choice, b) if a["prediction"] == b["prediction"]: st.success(f"**Both agree:** {a['label']}") else: st.warning( f"**Disagreement** - IndoBERT: {a['label']} · {ibt_choice}: {b['label']}" ) with st.expander("Show preprocessed text (Pipeline A - `preprocess_minimal`)"): st.code(a["preprocessed"] or "(empty after preprocessing)", language="text") # --------------------------------------------------------------------------- # Sidebar + main # --------------------------------------------------------------------------- def _sidebar_status() -> None: with st.sidebar: # Dev/ops panel: artifact presence + manual warm-up. Hidden from the # public demo (noise + leaks internal file layout); set DEMO_DEBUG=1 to # show it for local pre-demo checks. if os.environ.get("DEMO_DEBUG") == "1": st.header("Artifact status") st.markdown("**Pre-computed predictions**") for p in (INDOBERT["demo_json"], *[v["demo_json"] for v in IBT_VARIANTS.values()]): present = Path(p).exists() st.markdown(f"- {'✅' if present else '⚪'} `{p.name}`") st.divider() st.markdown("**Local checkpoints**") all_present = True for tag in (INDOBERT["tag"], *[v["tag"] for v in IBT_VARIANTS.values()]): present = (MODELS_DIR / tag).exists() all_present = all_present and present st.markdown(f"- {'✅' if present else '⚪'} `models/{tag}/`") st.divider() # Pre-warm: load all 4 checkpoints into st.cache_resource *before* # the demo so the first "Analyze" click is instant. st.markdown("**Demo readiness**") if st.button("🔥 Warm up all models", use_container_width=True, disabled=not all_present, help="Pre-load all 4 checkpoints now so live inference is instant."): prog = st.progress(0.0, text="Warming up...") failed = _warm_all_models( lambda i, n, name: prog.progress(i / n, text=f"Loading {name} ({i+1}/{n})...") ) prog.progress(1.0, text="Done") if failed: st.warning("Missing checkpoints (skipped): " + ", ".join(failed)) else: st.success("✅ All 4 models warm — demo ready.") elif not all_present: st.caption("Warm-up needs all 4 local checkpoints.") st.divider() with st.expander("⚠️ Disclaimer", expanded=False): st.markdown(DISCLAIMER_MD) with st.expander("About"): st.markdown(ABOUT_MD) def main() -> None: st.set_page_config( page_title="Cyberbullying Demo - IndoBERT vs IBT-hybrid", page_icon="🎯", layout="wide", ) st.title("🎯 Cyberbullying Detection - IndoBERT vs IBT-hybrid") st.caption( "Paper baseline (Jayanti & Rohman 2026, IndoBERT) vs proposed IBT-hybrid family. " "Single-stage binary classification on Indonesian TikTok comments." ) # Cloud (HF Spaces): pull checkpoints from the model repo on first boot, # before the sidebar status renders, so it reflects what's actually present. _ensure_checkpoints() _sidebar_status() # Cloud/headless pre-warm: set DEMO_AUTOWARM=1 so all checkpoints load at # process start and stay hot for every visitor (st.cache_resource is shared # across sessions). Runs once; later reruns hit the cache cheaply. if os.environ.get("DEMO_AUTOWARM") == "1": _warm_all_models() tab_viewer, tab_try = st.tabs(["📊 Test set viewer", "✍️ Try your own"]) with tab_viewer: _tab_viewer() with tab_try: _tab_try() if __name__ == "__main__": main()