Create app.py

app.py

@@ -0,0 +1,440 @@
+"""
+MHMisinfo — Mental Health Misinformation Detector
+Gradio Space: paste a YouTube URL → fetch metadata + transcripts → run 4-stream SeTa-Attention model → show verdict
+"""
+
+import os, re, json, sys, warnings
+warnings.filterwarnings("ignore")
+
+import numpy as np
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+import gradio as gr
+from dataclasses import dataclass
+from typing import Dict, List, Optional
+from huggingface_hub import hf_hub_download
+
+# ── YouTube helpers ────────────────────────────────────────────────────────────
+from googleapiclient.discovery import build as yt_build
+from youtube_transcript_api import YouTubeTranscriptApi, NoTranscriptFound, TranscriptsDisabled
+
+# ── Model + Data (inline, no src/ import needed) ──────────────────────────────
+import re as _re
+
+TAG_SPLIT_RE = _re.compile(r"[\s,]+")
+TEXT_RE = _re.compile(r"[A-Za-z0-9']+")
+
+@dataclass
+class Vocab:
+    token_to_idx: Dict[str, int]
+    idx_to_token: List[str]
+    pad_token: str = "<pad>"
+    unk_token: str = "<unk>"
+
+    @property
+    def pad_idx(self): return self.token_to_idx[self.pad_token]
+    @property
+    def unk_idx(self): return self.token_to_idx[self.unk_token]
+
+    def encode(self, tokens, max_len):
+        ids = [self.token_to_idx.get(t, self.unk_idx) for t in tokens]
+        if len(ids) >= max_len: return ids[:max_len]
+        return ids + [self.pad_idx] * (max_len - len(ids))
+
+    @staticmethod
+    def from_serializable(obj):
+        return Vocab(token_to_idx=obj["token_to_idx"],
+                     idx_to_token=obj["idx_to_token"],
+                     pad_token=obj.get("pad_token","<pad>"),
+                     unk_token=obj.get("unk_token","<unk>"))
+
+def tokenize_tags(text):
+    if not isinstance(text, str): return []
+    cleaned = text.replace("#"," ")
+    return [t for t in TAG_SPLIT_RE.split(cleaned.lower()) if t]
+
+def tokenize_text(text):
+    if not isinstance(text, str): return []
+    return [t.lower() for t in TEXT_RE.findall(text)]
+
+
+# ── Model Architecture (identical to src/model.py) ────────────────────────────
+class SeTaAttention(nn.Module):
+    def __init__(self, input_dim, attn_dim, dropout=0.1):
+        super().__init__()
+        self.proj = nn.Linear(input_dim, attn_dim)
+        self.sem_query = nn.Parameter(torch.randn(attn_dim))
+        self.task_query = nn.Parameter(torch.randn(attn_dim))
+        self.out = nn.Linear(input_dim * 2, input_dim)
+        self.dropout = nn.Dropout(dropout)
+
+    def _attend(self, h, query, mask):
+        proj = torch.tanh(self.proj(h))
+        scores = torch.matmul(proj, query)
+        scores = scores.masked_fill(~mask, -1e9)
+        weights = torch.softmax(scores, dim=1)
+        return torch.sum(h * weights.unsqueeze(-1), dim=1)
+
+    def forward(self, h, mask):
+        sem = self._attend(h, self.sem_query, mask)
+        task = self._attend(h, self.task_query, mask)
+        return self.dropout(torch.tanh(self.out(torch.cat([sem, task], dim=-1))))
+
+
+class StreamEncoder(nn.Module):
+    def __init__(self, vocab_size, emb_dim, hidden_dim, attn_dim, proj_dim, mlp_dim, dropout=0.2):
+        super().__init__()
+        self.embedding = nn.Embedding(vocab_size, emb_dim, padding_idx=0)
+        self.gru = nn.GRU(emb_dim, hidden_dim, batch_first=True, bidirectional=True)
+        self.attn = SeTaAttention(hidden_dim*2, attn_dim, dropout=dropout)
+        self.proj = nn.Sequential(
+            nn.Linear(hidden_dim*2, mlp_dim), nn.ReLU(), nn.Dropout(dropout), nn.Linear(mlp_dim, proj_dim)
+        )
+        self.dropout = nn.Dropout(dropout)
+
+    def forward(self, x):
+        mask = x != 0
+        emb = self.dropout(self.embedding(x))
+        h, _ = self.gru(emb)
+        attn_vec = self.attn(h, mask)
+        proj = self.dropout(torch.tanh(self.proj(attn_vec)))
+        return attn_vec, proj
+
+
+class MultiStreamModel(nn.Module):
+    def __init__(self, vocab_sizes, num_classes, emb_dim=128, hidden_dim=128, attn_dim=128,
+                 proj_dim=128, mlp_dim=256, dropout=0.2, include_tags_ccm=False, per_modality_trust=False):
+        super().__init__()
+        self.include_tags_ccm = include_tags_ccm
+        self.per_modality_trust = per_modality_trust
+        self.num_classes = num_classes
+        h_dim = hidden_dim * 2
+        self.encoders = nn.ModuleDict({
+            "tags": StreamEncoder(vocab_sizes["tags"], emb_dim, hidden_dim, attn_dim, proj_dim, mlp_dim, dropout),
+            "text": StreamEncoder(vocab_sizes["text"], emb_dim, hidden_dim, attn_dim, proj_dim, mlp_dim, dropout),
+            "audio_transcript": StreamEncoder(vocab_sizes["audio_transcript"], emb_dim, hidden_dim, attn_dim, proj_dim, mlp_dim, dropout),
+            "video_transcript": StreamEncoder(vocab_sizes["video_transcript"], emb_dim, hidden_dim, attn_dim, proj_dim, mlp_dim, dropout),
+        })
+        ccm_dim = 3 + (3 if include_tags_ccm else 0)
+        trust_in = h_dim + ccm_dim
+        if per_modality_trust:
+            self.trust_mlps = nn.ModuleDict({k: self._make_mlp(trust_in, mlp_dim, 1, dropout)
+                                              for k in ["text","audio_transcript","video_transcript","tags"]})
+            self.trust_mlp = None
+        else:
+            self.trust_mlp = self._make_mlp(trust_in, mlp_dim, 1, dropout)
+            self.trust_mlps = None
+        self.uncertainty_mlp = self._make_mlp(h_dim, mlp_dim, 1, dropout)
+        classifier_in = proj_dim * 5 + ccm_dim + 4 + 4
+        out_dim = 1 if num_classes == 2 else num_classes
+        self.mlp = nn.Sequential(
+            nn.Linear(classifier_in, mlp_dim), nn.ReLU(), nn.Dropout(dropout), nn.Linear(mlp_dim, out_dim)
+        )
+
+    @staticmethod
+    def _make_mlp(in_dim, hidden_dim, out_dim, dropout):
+        return nn.Sequential(nn.Linear(in_dim, hidden_dim), nn.ReLU(), nn.Dropout(dropout), nn.Linear(hidden_dim, out_dim))
+
+    @staticmethod
+    def _cosine(a, b):
+        return F.cosine_similarity(a, b, dim=-1, eps=1e-8).unsqueeze(-1)
+
+    def _compute_ccm(self, h_text, h_audio, h_video, h_tags):
+        sims = [self._cosine(h_text, h_audio), self._cosine(h_text, h_video), self._cosine(h_audio, h_video)]
+        if self.include_tags_ccm:
+            sims += [self._cosine(h_text, h_tags), self._cosine(h_audio, h_tags), self._cosine(h_video, h_tags)]
+        return torch.cat(sims, dim=-1)
+
+    def _trust_logit(self, key, h_i, ccm):
+        x = torch.cat([h_i, ccm], dim=-1)
+        return self.trust_mlps[key](x) if self.per_modality_trust else self.trust_mlp(x)
+
+    def forward(self, batch, return_details=False):
+        h_tags, p_tags = self.encoders["tags"](batch["tags"])
+        h_text, p_text = self.encoders["text"](batch["text"])
+        h_audio, p_audio = self.encoders["audio_transcript"](batch["audio_transcript"])
+        h_video, p_video = self.encoders["video_transcript"](batch["video_transcript"])
+        ccm = self._compute_ccm(h_text, h_audio, h_video, h_tags)
+        trust_logits = torch.cat([self._trust_logit("text", h_text, ccm),
+                                   self._trust_logit("audio_transcript", h_audio, ccm),
+                                   self._trust_logit("video_transcript", h_video, ccm),
+                                   self._trust_logit("tags", h_tags, ccm)], dim=-1)
+        trust_w = torch.softmax(trust_logits, dim=-1)
+        sigmas = torch.cat([F.softplus(self.uncertainty_mlp(h)) + 1e-6
+                             for h in [h_text, h_audio, h_video, h_tags]], dim=-1)
+        confidence = 1.0 / sigmas
+        fusion_w = trust_w * confidence
+        fusion_w = fusion_w / (fusion_w.sum(dim=-1, keepdim=True) + 1e-8)
+        proj_stack = torch.stack([p_text, p_audio, p_video, p_tags], dim=1)
+        fused = torch.sum(proj_stack * fusion_w.unsqueeze(-1), dim=1)
+        combined = torch.cat([p_text, p_audio, p_video, p_tags, fused, ccm, trust_w, sigmas], dim=-1)
+        logits = self.mlp(combined)
+        if not return_details: return logits
+        return logits, {"ccm": ccm, "trust_w": trust_w, "sigma": sigmas, "fusion_w": fusion_w}
+
+
+# ── Globals ────────────────────────────────────────────────────────────────────
+_model = None
+_vocabs = None
+_max_lens = None
+_config = None
+_device = "cpu"
+
+REPO_ID = "rocky250/MHMisinfo"
+YT_API_KEY = os.environ.get("YT_API_KEY", "")
+
+
+def _load_model():
+    global _model, _vocabs, _max_lens, _config
+    if _model is not None:
+        return
+    ckpt_path = hf_hub_download(repo_id=REPO_ID, filename="best_multimodal.pt")
+    ckpt = torch.load(ckpt_path, map_location=_device, weights_only=False)
+    vocabs_raw = ckpt["vocabs"]
+    _vocabs = {k: Vocab.from_serializable(v) for k, v in vocabs_raw.items()}
+    _max_lens = ckpt["max_lens"]
+    _config = ckpt["config"]
+    num_classes = ckpt["num_classes"]
+    _model = MultiStreamModel(
+        vocab_sizes={k: len(v.token_to_idx) for k, v in _vocabs.items()},
+        num_classes=num_classes,
+        emb_dim=_config["emb_dim"], hidden_dim=_config["hidden_dim"],
+        attn_dim=_config["attn_dim"], proj_dim=_config["proj_dim"],
+        mlp_dim=_config["mlp_dim"], dropout=_config["dropout"],
+        include_tags_ccm=_config.get("include_tags_ccm", False),
+        per_modality_trust=_config.get("per_modality_trust", False),
+    ).to(_device)
+    _model.load_state_dict(ckpt["model_state"])
+    _model.eval()
+
+
+def _extract_video_id(url: str) -> Optional[str]:
+    patterns = [
+        r"(?:v=|youtu\.be/|embed/|shorts/)([A-Za-z0-9_-]{11})",
+    ]
+    for p in patterns:
+        m = re.search(p, url)
+        if m: return m.group(1)
+    return None
+
+
+def _fetch_yt_metadata(video_id: str):
+    """Fetch title, description, tags via YouTube Data API v3."""
+    if not YT_API_KEY:
+        return None, None, None, "⚠️ No YouTube API key set. Set the YT_API_KEY secret in Space settings."
+    try:
+        yt = yt_build("youtube", "v3", developerKey=YT_API_KEY, cache_discovery=False)
+        resp = yt.videos().list(part="snippet", id=video_id).execute()
+        if not resp.get("items"):
+            return None, None, None, "❌ Video not found or unavailable."
+        snippet = resp["items"][0]["snippet"]
+        title = snippet.get("title", "")
+        desc = snippet.get("description", "")
+        tags = " ".join(snippet.get("tags", []))
+        return title, desc, tags, None
+    except Exception as e:
+        return None, None, None, f"❌ YouTube API error: {e}"
+
+
+def _fetch_transcript(video_id: str, field: str) -> str:
+    """Fetch transcript text (same text used for both audio & video transcript streams)."""
+    try:
+        transcript_list = YouTubeTranscriptApi.get_transcript(video_id, languages=["en"])
+        return " ".join(t["text"] for t in transcript_list)
+    except (NoTranscriptFound, TranscriptsDisabled):
+        return ""
+    except Exception:
+        return ""
+
+
+def _encode_single(text: str, tags: str, audio_t: str, video_t: str) -> Dict[str, torch.Tensor]:
+    _load_model()
+    streams = {
+        "tags": tokenize_tags(tags),
+        "text": tokenize_text(text),
+        "audio_transcript": tokenize_text(audio_t),
+        "video_transcript": tokenize_text(video_t),
+    }
+    batch = {}
+    for s, tokens in streams.items():
+        ids = _vocabs[s].encode(tokens, _max_lens[s])
+        batch[s] = torch.tensor([ids], dtype=torch.long).to(_device)
+    return batch
+
+
+def _run_inference(text, tags, audio_t, video_t):
+    batch = _encode_single(text, tags, audio_t, video_t)
+    with torch.no_grad():
+        logits, details = _model(batch, return_details=True)
+    prob = float(torch.sigmoid(logits).squeeze())
+    pred = int(prob >= 0.5)
+    trust = details["trust_w"][0].cpu().numpy().tolist()
+    sigma = details["sigma"][0].cpu().numpy().tolist()
+    ccm = details["ccm"][0].cpu().numpy().tolist()
+    return prob, pred, trust, sigma, ccm
+
+
+# ── Gradio logic ───────────────────────────────────────────────────────────────
+MODALITIES = ["text", "audio_transcript", "video_transcript", "tags"]
+CCM_LABELS_3 = ["text↔audio", "text↔video", "audio↔video"]
+CCM_LABELS_6 = CCM_LABELS_3 + ["text↔tags", "audio↔tags", "video↔tags"]
+
+LABEL_COLORS = {0: "#22c55e", 1: "#ef4444"}
+LABEL_NAMES = {0: "✅ Credible / Not Misinformation", 1: "⚠️ Potential Misinformation"}
+
+
+def _bar(value: float, color: str) -> str:
+    pct = int(value * 100)
+    return (
+        f'<div style="background:#e5e7eb;border-radius:6px;height:14px;width:100%;margin:2px 0">'
+        f'<div style="background:{color};width:{pct}%;height:100%;border-radius:6px;transition:width 0.4s"></div>'
+        f'</div><small style="color:#6b7280">{value:.3f}</small>'
+    )
+
+
+def analyze_url(url: str):
+    if not url.strip():
+        return [gr.update(visible=False)] * 4 + ["Please enter a YouTube URL."]
+
+    video_id = _extract_video_id(url.strip())
+    if not video_id:
+        return [gr.update(visible=False)] * 4 + ["❌ Could not extract a valid YouTube video ID from that URL."]
+
+    # Fetch metadata
+    title, desc, tags, err = _fetch_yt_metadata(video_id)
+    if err:
+        return [gr.update(visible=False)] * 4 + [err]
+
+    # Fetch transcript
+    transcript = _fetch_transcript(video_id, "transcript")
+    text_field = f"{title} {desc}".strip()
+
+    # Run model
+    try:
+        _load_model()
+        prob, pred, trust, sigma, ccm = _run_inference(text_field, tags, transcript, transcript)
+    except Exception as e:
+        return [gr.update(visible=False)] * 4 + [f"❌ Model error: {e}"]
+
+    # ── Verdict card ──────────────────────────────────────────────────────────
+    color = LABEL_COLORS[pred]
+    label_text = LABEL_NAMES[pred]
+    conf_pct = int(prob * 100) if pred == 1 else int((1 - prob) * 100)
+    verdict_html = f"""
+    <div style="border:2px solid {color};border-radius:12px;padding:20px 24px;background:{color}18;margin-bottom:8px">
+      <div style="font-size:1.5rem;font-weight:700;color:{color}">{label_text}</div>
+      <div style="font-size:2.5rem;font-weight:800;color:{color};margin:6px 0">{conf_pct}% confident</div>
+      <div style="color:#6b7280;font-size:0.9rem">Raw misinfo probability: <b>{prob:.4f}</b></div>
+    </div>
+    <div style="background:#f9fafb;border-radius:10px;padding:14px 16px;margin-top:6px">
+      <b>🎬 Video:</b> <a href="{url}" target="_blank">{title}</a><br>
+      <b>🏷️ Tags:</b> {tags[:120] + '…' if len(tags)>120 else (tags or '(none)')}<br>
+      <b>📝 Transcript:</b> {('Available (' + str(len(transcript.split())) + ' words)') if transcript else '(not available — model used title/description only)'}
+    </div>
+    """
+
+    # ── Modality trust weights ─────────────────────────────────────────────────
+    trust_html = "<h4 style='margin-bottom:8px'>Modality Trust Weights</h4>"
+    trust_html += "<small style='color:#6b7280'>How much the model relied on each stream</small><br><br>"
+    for m, t in zip(MODALITIES, trust):
+        trust_html += f"<b>{m.replace('_',' ').title()}</b>{_bar(t, '#3b82f6')}"
+
+    # ── Uncertainty (sigma) ───────────────────────────────────────────────────
+    sigma_html = "<h4 style='margin-bottom:8px'>Uncertainty (σ)</h4>"
+    sigma_html += "<small style='color:#6b7280'>Higher = encoder less certain about this stream</small><br><br>"
+    max_s = max(sigma) if sigma else 1
+    for m, s in zip(MODALITIES, sigma):
+        sigma_html += f"<b>{m.replace('_',' ').title()}</b>{_bar(s/max_s, '#f59e0b')}"
+
+    # ── CCM ───────────────────────────────────────────────────────────────────
+    ccm_labels = CCM_LABELS_6 if len(ccm) == 6 else CCM_LABELS_3
+    ccm_html = "<h4 style='margin-bottom:8px'>Cross-Channel Consistency (CCM)</h4>"
+    ccm_html += "<small style='color:#6b7280'>Cosine similarity between modality representations (−1 to 1)</small><br><br>"
+    for lbl, val in zip(ccm_labels, ccm):
+        norm = (val + 1) / 2  # map [-1,1] → [0,1]
+        ccm_html += f"<b>{lbl}</b>{_bar(norm, '#8b5cf6')}<small style='color:#9ca3af'>raw: {val:.3f}</small><br>"
+
+    status = "✅ Analysis complete."
+    return (
+        gr.update(value=verdict_html, visible=True),
+        gr.update(value=trust_html, visible=True),
+        gr.update(value=sigma_html, visible=True),
+        gr.update(value=ccm_html, visible=True),
+        status,
+    )
+
+
+# ── UI ─────────────────────────────────────────────────────────────────────────
+CSS = """
+#header { text-align:center; margin-bottom: 20px; }
+#header h1 { font-size: 2rem; font-weight: 800; margin: 0; }
+#header p  { color: #6b7280; margin: 4px 0 0; }
+.panel { border-radius:12px !important; }
+footer { display:none !important; }
+"""
+
+with gr.Blocks(css=CSS, title="MHMisinfo — Mental Health Misinformation Detector") as demo:
+    gr.HTML("""
+    <div id="header">
+      <h1>🧠 MHMisinfo</h1>
+      <p>4-Stream SeTa-Attention model for detecting mental health misinformation on YouTube</p>
+      <p style="font-size:0.8rem;color:#9ca3af">
+        Based on: <i>"Supporters and Skeptics: LLM-based Analysis of Engagement with Mental Health (Mis)Information Content on Video-sharing Platforms"</i>
+      </p>
+    </div>
+    """)
+
+    with gr.Row():
+        with gr.Column(scale=3):
+            url_input = gr.Textbox(
+                placeholder="Paste a YouTube URL here, e.g. https://www.youtube.com/watch?v=...",
+                label="YouTube Video URL",
+                lines=1,
+            )
+        with gr.Column(scale=1, min_width=120):
+            analyze_btn = gr.Button("🔍 Analyze", variant="primary", size="lg")
+
+    status_box = gr.Textbox(label="Status", interactive=False, lines=1, visible=True)
+
+    with gr.Row():
+        verdict_out = gr.HTML(visible=False, elem_classes="panel")
+
+    with gr.Row():
+        with gr.Column():
+            trust_out = gr.HTML(visible=False, elem_classes="panel")
+        with gr.Column():
+            sigma_out = gr.HTML(visible=False, elem_classes="panel")
+
+    with gr.Row():
+        ccm_out = gr.HTML(visible=False, elem_classes="panel")
+
+    gr.HTML("""
+    <hr style="margin:28px 0 16px">
+    <details>
+      <summary style="cursor:pointer;font-weight:600;color:#374151">ℹ️ How it works</summary>
+      <div style="padding:12px 0;color:#6b7280;font-size:0.9rem">
+        <b>4 streams:</b> video title+description (text), hashtags/tags, audio transcript, video transcript.<br>
+        Each stream is encoded by a BiGRU with SeTa dual-attention. The model computes:<br>
+        &nbsp;• <b>CCM</b> (Cross-Channel Consistency Matrix) — cosine similarity between stream representations<br>
+        &nbsp;• <b>Trust weights</b> — learned per-stream reliability given CCM context<br>
+        &nbsp;• <b>Uncertainty (σ)</b> — calibrated confidence per stream via DMTE<br>
+        These are fused into a single classification head.<br><br>
+        <b>Note:</b> The model was trained on short YouTube mental health videos. Results on other content types may vary.
+        ROC-AUC on held-out test: <b>0.967</b>. Positive-class F1: <b>0.828</b>.
+      </div>
+    </details>
+    """)
+
+    analyze_btn.click(
+        fn=analyze_url,
+        inputs=[url_input],
+        outputs=[verdict_out, trust_out, sigma_out, ccm_out, status_box],
+    )
+    url_input.submit(
+        fn=analyze_url,
+        inputs=[url_input],
+        outputs=[verdict_out, trust_out, sigma_out, ccm_out, status_box],
+    )
+
+demo.launch()