app.py

"""
Defense demo: Multi-Layered Sarcasm-Aware Harmful Content Detection.

One codebase, two deployments:
  - HF Spaces (16GB): live XLM-R sarcasm + live baseline + live Mechanism B
  - Local laptop (4GB): live XLM-R sarcasm + curated showcase (Qwen models skipped)

The app auto-detects what it can load. Set env vars to control:
  LOAD_QWEN=1        attempt to load the two Qwen classifiers (default on Spaces)
  SARCASM_MODEL=...  HF repo or local dir for the XLM-R sarcasm detector
  BASELINE_ADAPTER=... / MECHB_ADAPTER=...   HF repos or local dirs for adapters

Run:  python app.py
"""

import os
import json
import torch
import spaces
import gradio as gr
import numpy as np

# ----------------------------- configuration ------------------------------ #
SARCASM_MODEL = os.environ.get("SARCASM_MODEL", "mlaskri/sarcasm-xlmr-hyperparams-only")
BASELINE_ADAPTER = os.environ.get("BASELINE_ADAPTER", "mlaskri/baseline-combined-seed42")   
MECHB_ADAPTER = os.environ.get("MECHB_ADAPTER", "mlaskri/harmful-content-mechanism-b-v2")               
QWEN_BASE = "Qwen/Qwen2.5-3B-Instruct"

LOAD_QWEN = os.environ.get("LOAD_QWEN", "1") == "1"

SARCASM_TEMPERATURE = 1.953   # validation-fit temperature (thesis Table 4.5)
SARCASM_THRESHOLD = 0.48      # tuned decision threshold
BASELINE_THRESHOLD = 0.35     # validation-selected (thesis Table 5.1)
MECHB_THRESHOLD = 0.60        # validation-selected (thesis Table 5.4)

SHOWCASE_FILE = "curated_examples.json"

# ----------------------------- model loading ------------------------------ #
print("Loading sarcasm detector (XLM-R)...")
from transformers import AutoTokenizer, AutoModelForSequenceClassification

sar_tok = AutoTokenizer.from_pretrained(SARCASM_MODEL)
sar_model = AutoModelForSequenceClassification.from_pretrained(SARCASM_MODEL)
sar_model.eval()  # small; runs on CPU inside the GPU-decorated fn or CPU fallback
print("Sarcasm detector ready.")

qwen_models = {}
if LOAD_QWEN:
    try:
        from peft import PeftModel
        print("Loading Qwen classifiers (baseline + Mechanism B)... this takes a few minutes on CPU.")
        qtok = AutoTokenizer.from_pretrained(QWEN_BASE)
        if qtok.pad_token is None:
            qtok.pad_token = qtok.eos_token
        for name, adapter in [("baseline", BASELINE_ADAPTER), ("mechanism_b", MECHB_ADAPTER)]:
            base = AutoModelForSequenceClassification.from_pretrained(
                QWEN_BASE,
                num_labels=2,
                torch_dtype=torch.float16,
                device_map="cpu",            # <-- force CPU at load; GPU only inside @spaces.GPU
                low_cpu_mem_usage=True,
            )
            base.config.pad_token_id = qtok.pad_token_id
            m = PeftModel.from_pretrained(base, adapter, device_map="cpu")
            m.eval()
            qwen_models[name] = m
            print(f"  {name} ready.")
    except Exception as e:
        print(f"Qwen loading failed ({e}) -- running in sarcasm-only mode.")
        qwen_models = {}
else:
    print("LOAD_QWEN=0 -- sarcasm-only mode (local laptop).")

LIVE_HARMFUL = bool(qwen_models)

# ----------------------------- inference fns ------------------------------ #
@torch.no_grad()
def sarcasm_score(text):
    dev = "cuda" if torch.cuda.is_available() else "cpu"
    model = sar_model.to(dev)
    enc = sar_tok(text, truncation=True, max_length=128, return_tensors="pt").to(dev)
    logits = model(**enc).logits
    # temperature scaling on logits, then softmax
    p = torch.softmax(logits / SARCASM_TEMPERATURE, dim=-1)[0, 1].item()
    return p

@torch.no_grad()
def harmful_pred(name, text):
    m = qwen_models[name]
    dev = "cuda" if torch.cuda.is_available() else "cpu"
    m = m.to(dev)
    enc = qtok(text, truncation=True, max_length=128, return_tensors="pt").to(dev)
    logits = m(**enc).logits
    return torch.softmax(logits.float(), dim=-1)[0, 1].item()

def fmt_pct(p):
    return f"{p:.3f}"

@spaces.GPU(duration=90)
def analyze(text):
    text = (text or "").strip()
    if not text:
        return "Please enter some text.", "", ""
    s = sarcasm_score(text)
    s_label = "SARCASTIC" if s >= SARCASM_THRESHOLD else "not sarcastic"
    sar_out = (f"**Calibrated sarcasm probability: {fmt_pct(s)}**  \n"
               f"Decision (threshold {SARCASM_THRESHOLD}): **{s_label}**  \n"
               f"*Twitter-XLM-RoBERTa, hyperparameters-only configuration; "
               f"temperature-scaled (T={SARCASM_TEMPERATURE}).*")

    if not LIVE_HARMFUL:
        harm_out = ("*Harmful-content classifiers (3B parameters each) exceed this machine's memory. "
                    "See the Showcase tab for committed evaluation results comparing all mechanisms — "
                    "every number traces to results files in the project repository.*")
        return sar_out, harm_out, ""

    pb = harmful_pred("baseline", text)
    pm = harmful_pred("mechanism_b", text)
    lb = "HARMFUL" if pb >= BASELINE_THRESHOLD else "non-harmful"
    lm = "HARMFUL" if pm >= MECHB_THRESHOLD else "non-harmful"
    harm_out = (f"| Model | p(harmful) | Decision |\n|---|---|---|\n"
                f"| Baseline (no sarcasm signal) | {fmt_pct(pb)} | **{lb}** |\n"
                f"| Mechanism B (sarcasm-aware) | {fmt_pct(pm)} | **{lm}** |")
    note = ""
    if lb != lm:
        note = ("**The two models disagree on this example** — the sarcasm-aware model's decision "
                "differs from the baseline's, illustrating the effect studied in this thesis.")
    return sar_out, harm_out, note

# ----------------------------- showcase tab ------------------------------- #
def load_showcase():
    if os.path.exists(SHOWCASE_FILE):
        with open(SHOWCASE_FILE, encoding="utf-8") as f:
            return json.load(f)
    return []

SHOWCASE = load_showcase()

def showcase_view(idx):
    if not SHOWCASE:
        return "No curated examples loaded (curated_examples.json missing)."
    ex = SHOWCASE[int(idx)]
    rows = "\n".join(
        f"| {m['name']} | {m['prob']} | **{m['decision']}** |"
        for m in ex["models"])
    return (f"### {ex['title']}\n\n"
            f"**Text ({ex['language']}):** {ex['text']}\n\n"
            f"**True label:** {ex['true_label']}   |   **Sarcasm score:** {ex['sarcasm_score']}\n\n"
            f"| Model | p(harmful) | Decision |\n|---|---|---|\n{rows}\n\n"
            f"**Why this example matters:** {ex['story']}\n\n"
            f"*Source: {ex['source']} (held-out test set; predictions from committed evaluation artifacts).*")

# ----------------------------- UI ----------------------------------------- #
with gr.Blocks(title="Sarcasm-Aware Harmful Content Detection — PFE Demo") as demo:
    gr.Markdown(
        "# Multi-Layered Multilingual Sarcasm-Aware Harmful Content Detection\n"
        "**PFE defense demo — ESTIN 2025/2026** · Arabic + English · "
        + ("**Live mode: full pipeline**" if LIVE_HARMFUL else "**Local mode: live sarcasm layer + committed results**"))

    with gr.Tab("Live analysis"):
        inp = gr.Textbox(label="Enter any text (Arabic or English)", lines=3,
                         placeholder="Type here — e.g. a tweet, a comment...")
        btn = gr.Button("Analyze", variant="primary")
        out_sar = gr.Markdown(label="Layer 1 — Sarcasm detection")
        out_harm = gr.Markdown(label="Layer 2 — Harmful-content detection")
        out_note = gr.Markdown()
        btn.click(analyze, inputs=inp, outputs=[out_sar, out_harm, out_note])

    with gr.Tab("Showcase: findings on the evaluation set"):
        gr.Markdown("Hand-picked held-out test examples illustrating the thesis's main findings. "
                    "All predictions are committed evaluation artifacts, reproducible from the repository.")
        if SHOWCASE:
            slider = gr.Slider(0, len(SHOWCASE) - 1, step=1, value=0, label="Example")
            view = gr.Markdown(showcase_view(0))
            slider.change(showcase_view, inputs=slider, outputs=view)
        else:
            gr.Markdown("*curated_examples.json not found — add it next to app.py.*")

    with gr.Tab("About"):
        gr.Markdown(
            "**Architecture:** Layer 1 (Twitter-XLM-RoBERTa sarcasm detector, temperature-calibrated) → "
            "integration mechanisms → Layer 2 (Qwen2.5-3B + LoRA harmful-content classifier).\n\n"
            "**Mechanism B** transfers the sarcasm layer's adapter weights into the harmful-content "
            "classifier's initialization — the best-performing mechanism (macro-F1 0.8366 vs 0.8301 baseline; "
            "harmful-class McNemar p < 10⁻³⁴).\n\n"
            "Thresholds shown are the validation-selected values frozen before test evaluation, "
            "as documented in the thesis.")

demo.launch()