app.py

# -*- coding: utf-8 -*-
"""app

Automatically generated by Colab.

Original file is located at
    https://colab.research.google.com/drive/13NwFd1gWSQc7FMNtnVRoN5FxSOQ6CQ5d
"""

import re
import torch
import spaces          # required for ZeroGPU — HF Spaces free-tier GPU option
import gradio as gr
from transformers import T5ForConditionalGeneration, T5Tokenizer

# ── CONFIG ────────────────────────────────────────────────────
MODEL_PATH = "malaika971/distill-hate"
MAX_INPUT  = 96
MAX_NEW    = 150

# On ZeroGPU, keep the model on CPU at import time — a GPU is only attached
# to the process during a @spaces.GPU-decorated call, not for the whole app.
print("Loading model...")
tokenizer = T5Tokenizer.from_pretrained(MODEL_PATH)
model     = T5ForConditionalGeneration.from_pretrained(MODEL_PATH)
model.eval()
print(f"{sum(p.numel() for p in model.parameters())/1e6:.1f}M params loaded (CPU)")

def make_prompt(post):
    return (f"Determine whether the following post is hate speech.  "
            f"Post: {post}  Options:  (A) Hate (B) Not hate  Answer: ")

# ── label extraction — IDENTICAL to the training script's extract_label(),
#    which is what actually produced the reported 82.74%/82.54% numbers.
#    No pre-guard regex overrides on the raw post (that was v4-only).
def extract_label(text):
    t = text.lower().strip()
    if re.match(r'^\(a\)', t): return "hate"
    if re.match(r'^\(b\)', t): return "non_hate"
    if "this is hate speech" in t and "not" not in t[:35]:
        return "hate"
    if "this is not hate speech" in t:
        return "non_hate"
    return "non_hate"

def clean(text):
    text = re.sub(r'^RATIONALE\s*\d+\s*:\s*\n?', '', text.strip(), flags=re.IGNORECASE)
    text = re.sub(r'\b(\w{4,})\s+(or|and)\s+\1\b', r'\1', text, flags=re.IGNORECASE)
    for pat in [r'Based on (this|these) (analysis|steps|information)[,\s]*', r'it can be concluded that\s*']:
        text = re.sub(pat, '', text, flags=re.IGNORECASE)
    text = re.sub(r'\d+\.\s*(Final Decision)', r'\1', text)
    text = re.sub(r'(Final Decision[^\.]+\.)\s*Final Decision[^\n]*', r'\1', text, flags=re.IGNORECASE)
    return re.sub(r'  +', ' ', text).strip()

def keep_3_steps(text):
    parts = re.split(r'(?=\s*\d+\.)', text)
    label = parts[0].strip(); steps = []
    for p in parts[1:]:
        p = p.strip()
        num = re.match(r'^(\d+)\.', p)
        if num and int(num.group(1)) > 3: break
        if p: steps.append(p)
    return (label + " " + " ".join(steps)).strip()

PROTECTED = {
    "black people","black","white people","white","asian","latino","latina","hispanic",
    "african american","african","caucasian","arab","jewish","jew","indian","indigenous",
    "native american","aboriginal","pacific islander","middle eastern","north african",
    "mexican","chinese","korean","japanese","pakistani","bangladeshi","turkish",
    "muslim","muslims","islam","christian","christians","hindu","hindus","buddhist","sikh",
    "atheist","catholic","protestant","women","woman","female","females",
    "gay","lesbian","bisexual","transgender","trans","queer","lgbtq","lgbt","non-binary",
    "disabled","autistic","deaf","blind","immigrants","immigrant","refugees","refugee",
    "migrants","migrant","asylum seeker","foreigner","nigga","nigger","infidel",
}
HATE_CTX = {
    "kill","attack","hate","harm","deport","inferior","criminal","dangerous","dirty",
    "stupid","violent","replace","invade","destroy","threat","evil","duty","belong",
    "go back","dehumanize","subhuman","trash","sexist","burden","lazy","ruining",
    "taking over","invasion","crime","disease","half a brain","kitchen","always want",
}

def htc(explanation):
    t = explanation.lower()
    patterns = [r'\(a\)\s*(not hate|non.hate)', r'\(a\)\s*hate', r'\(b\)\s*not hate',
                r'final decision[:\s]+', r'(is|constitutes|qualifies as) hate speech',
                r'(is not|cannot be considered|does not constitute) hate speech',
                r'can be considered (hate|offensive)']
    return 1 if any(re.search(p, t) for p in patterns) else 0

def qf(post, explanation, prediction):
    is_hate = "hate" in prediction and "non" not in prediction
    if not is_hate: return 0.0
    exp_low, post_low = explanation.lower(), post.lower()
    quoted = re.findall(r'["\u201c\u201d\u2018\u2019]([^"\u201c\u201d\u2018\u2019]{3,})["\u201c\u201d\u2018\u2019]', explanation)
    inline = re.findall(r'(?:phrase|term|word|statement|use of|joke|post)\s+[\'"]?([a-z][^\'\"\.]{2,30})[\'"]?', exp_low)
    post_words = set(re.findall(r'\b[a-z]{4,}\b', post_low))
    exp_words = set(re.findall(r'\b[a-z]{4,}\b', exp_low))
    stop = {"this","that","with","from","they","them","their","have","been","will","would",
            "could","should","which","there","about","other","more","some","into","than",
            "also","just","very","when","what","such","these","those"}
    post_words -= stop; exp_words -= stop
    overlap = len(post_words & exp_words) / max(len(post_words), 1)
    if quoted:
        covered = sum(len(s) for s in quoted if s.lower() in post_low)
        score = min(covered / max(len(post), 1) + 0.5, 1.0)
    elif inline:
        score = min(overlap + 0.3, 1.0)
    elif overlap > 0.3:
        score = min(overlap + 0.15, 0.75)
    else:
        score = 0.2
    return round(float(score), 4)

def tgi(post, explanation, prediction):
    combined = (post + " " + explanation).lower()
    tokens = combined.split(); ngrams = set()
    for n in range(1, 4):
        for i in range(len(tokens) - n + 1):
            g = re.sub(r'^[^a-z]+|[^a-z]+$', '', " ".join(tokens[i:i+n]))
            if g: ngrams.add(g)
    matched = ngrams & PROTECTED
    if not matched: return 0
    is_hate = "hate" in prediction and "non" not in prediction
    if not is_hate: return 1
    exp_low, post_low = explanation.lower(), post.lower()
    for grp in matched:
        for text in [exp_low, post_low]:
            pos = text.find(grp)
            if pos == -1: continue
            window = text[max(0, pos-120): pos+120]
            if any(hw in window for hw in HATE_CTX): return 1
    return 1 if (matched and len(explanation) > 100) else 0

def cc(htc_v, qf_v, tgi_v, prediction, tau=0.25):
    is_hate = "hate" in prediction and "non" not in prediction
    if is_hate: return 1 if (qf_v >= tau and tgi_v == 1) else 0
    return 1 if (qf_v < tau and tgi_v == 0) else 0

def hatexscore(h, q, t, c): return round((h + q + t + c) / 4.0, 4)

# ── inference ───────────────────────────────────────────────
@spaces.GPU  # HF ZeroGPU: a GPU is attached only for the duration of this call
def generate_raw(post):
    device = "cuda" if torch.cuda.is_available() else "cpu"
    model.to(device)
    inputs = tokenizer(make_prompt(post), return_tensors="pt",
                        truncation=True, max_length=MAX_INPUT).to(device)
    with torch.no_grad():
        # Greedy decoding, no repetition controls — matches the training
        # script's run_val() generation call exactly (that script used
        # max_new_tokens=20, just enough to catch the (A)/(B) label; we use
        # MAX_NEW=150 here so the demo can display a full rationale, not just
        # the label — extract_label() below only reads the start of the text
        # / specific fixed phrases, so the longer generation doesn't change
        # which label gets extracted).
        out = model.generate(**inputs, max_new_tokens=MAX_NEW, do_sample=False)
    return tokenizer.decode(out[0], skip_special_tokens=True)

def predict(post):
    if not post or not post.strip():
        return "—", "Enter a post above.", "", ""

    raw_full = generate_raw(post)
    pred = extract_label(raw_full)              # classification: on raw text, exactly as in training/eval
    raw_display = clean(keep_3_steps(raw_full))  # cosmetic cleanup for the rationale shown in the UI only

    h = htc(raw_display)
    q = qf(post, raw_display, pred)
    t = tgi(post, raw_display, pred)
    c = cc(h, q, t, pred)
    hx = hatexscore(h, q, t, c)

    verdict = "🔴 HATE SPEECH" if pred == "hate" else "🟢 NON-HATE SPEECH"

    breakdown = (
        f"**HateXScore = {hx:.4f}**  (τ = 0.25)\n\n"
        f"| Component | Value |\n|---|---|\n"
        f"| HTC — Hate-Type Check | {h} |\n"
        f"| QF — Quotation Faithfulness | {q:.4f} |\n"
        f"| TGI — Target-Group ID | {t} |\n"
        f"| CC — Consistency Check | {c} |\n"
    )

    return verdict, raw_display, breakdown, pred

# ── Gradio UI ───────────────────────────────────────────────
with gr.Blocks(title="Distill-Hate — Explainable Implicit Hate Speech Detection") as demo:
    gr.Markdown(
        "# Distill-Hate\n"
        "Knowledge-distilled Flan-T5-base (247M params) — classifies a post as hate / "
        "non-hate and generates a 3-step rationale, scored with HateXScore."
    )
    with gr.Row():
        with gr.Column():
            post_in = gr.Textbox(label="Post", lines=3,
                                  placeholder="Enter a post to analyze...")
            submit = gr.Button("Analyze", variant="primary")
            gr.Examples(
                examples=[
                    ["Women should stay in the kitchen not run companies."],
                    ["I love spending time with my family on weekends."],
                    ["Old people are useless in the workplace and should just retire."],
                ],
                inputs=post_in,
            )
        with gr.Column():
            verdict_out = gr.Textbox(label="Verdict")
            rationale_out = gr.Textbox(label="Rationale", lines=6)
            score_out = gr.Markdown(label="HateXScore breakdown")
            pred_out = gr.Textbox(visible=False)

    submit.click(predict, inputs=post_in,
                 outputs=[verdict_out, rationale_out, score_out, pred_out])
    post_in.submit(predict, inputs=post_in,
                    outputs=[verdict_out, rationale_out, score_out, pred_out])

if __name__ == "__main__":
    demo.launch()