detector.py

"""
detector.py - core engine for SentimentDetector (no UI code).

Pipeline, per sentence
  1. RoBERTa sentiment model on the raw sentence      -> literal polarity (pos - neg)
  2. Same model on the emoji-stripped sentence        -> text-only polarity (isolates emoji effect)
  3. Emoji / emoticon lexicon                         -> emoji valence, blended into (1)
  4. RoBERTa irony model                              -> irony probability
  5. Rule cues (phrases, typography, text-vs-emoji)   -> cue score
  6. Noisy-OR fusion of (4) and (5)                   -> sarcasm probability
  7. Polarity inversion when sarcastic                -> intended polarity
  8. RoBERTa emotion model                            -> anger / joy / optimism / sadness
Sentences are then aggregated into a per-author profile (recency-weighted mood,
trend, volatility, sarcasm rate, dominant emotion).
"""
from __future__ import annotations

import re
from dataclasses import dataclass
from typing import Dict, List, Optional, Tuple

import emoji
import numpy as np
import torch
from transformers import AutoModelForSequenceClassification, AutoTokenizer

# --------------------------------------------------------------------------- config
MODELS = {
    "sentiment": ("cardiffnlp/twitter-roberta-base-sentiment-latest", ["negative", "neutral", "positive"]),
    "irony": ("cardiffnlp/twitter-roberta-base-irony", ["non_irony", "irony"]),
    "emotion": ("cardiffnlp/twitter-roberta-base-emotion", ["anger", "joy", "optimism", "sadness"]),
}
SARCASM_THRESHOLD = 0.55   # sarcasm probability above which polarity is inverted
LABEL_THRESHOLD = 0.25     # |score| below this is "neutral"
MAX_SENTENCES = 60
MAX_EXPLAIN_WORDS = 40
RECENCY_HALF_LIFE = 4      # sentences; recent sentences count more towards "current mood"

# --------------------------------------------------------------------------- lexicons
# Emoji valence in [-1, 1]. Keys have variation selectors / skin tones removed.
EMOJI_VALENCE: Dict[str, float] = {
    "😀": .7, "😃": .7, "😄": .8, "😁": .7, "😊": .8, "🙂": .3, "😍": .9, "🥰": .9, "😘": .7,
    "😂": .6, "🤣": .6, "😎": .6, "🎉": .8, "🎊": .8, "👍": .6, "👏": .7, "❤": .9, "💖": .8,
    "🔥": .6, "✨": .5, "🙏": .4, "💪": .6, "🥳": .9, "😌": .5, "🤩": .9, "😇": .6, "💯": .7,
    "🚀": .6, "✅": .5, "🌟": .6, "😅": .2, "😉": .3, "☺": .7, "🍕": .3,
    "😞": -.7, "😢": -.8, "😭": -.8, "😡": -.9, "🤬": -.9, "😠": -.8, "😤": -.6, "😩": -.7,
    "😫": -.7, "😖": -.6, "😣": -.6, "😔": -.6, "💔": -.9, "👎": -.7, "😱": -.5, "😰": -.6,
    "😨": -.6, "🤮": -.9, "🤢": -.8, "😒": -.6, "🙄": -.6, "😑": -.4, "😐": -.2, "☹": -.6,
    "🙁": -.5, "😟": -.5, "😓": -.5, "🥺": -.2, "💀": -.2, "🤡": -.6, "❌": -.5,
    "🙃": -.1, "😏": -.1,
}
# Emoji that commonly flag irony / eye-rolling when paired with positive words.
IRONIC_EMOJI = {"🙄", "😒", "🙃", "😏", "🤡", "😑"}

EMOTICON_VALENCE = {":)": .6, ":D": .8, ";)": .3, ":P": .4, ":p": .4, "XD": .6,
                    ":(": -.6, ":'(": -.8, ":/": -.3, ":|": -.1, "<3": .8, "</3": -.8}
_EMOTICON_RE = re.compile(r"(?<![\w/])(</3|<3|:'\(|[:;]-?[)(DPp/|]|XD)(?!\w)")

_LEXICAL_CUES = [
    (r"\byeah,?\s+(right|sure)\b", "'yeah right'"),
    (r"\boh,?\s+(great|wonderful|fantastic|perfect|lovely|joy|brilliant)\b", "'oh great'"),
    (r"\bjust what i (needed|wanted|ordered)\b", "'just what I needed'"),
    (r"\bthanks a lot\b|\bthanks for nothing\b|\bthanks,? (genius|captain obvious)\b", "mock thanks"),
    (r"\bwhat a (great|wonderful|brilliant|fantastic|lovely) (idea|day|surprise|way)\b", "'what a great ...'"),
    (r"\b(love|adore) (how|that|it when|waiting|being)\b", "'love how ...'"),
    (r"\bsure,? because\b|\bbecause that (always )?works\b", "'sure, because'"),
    (r"\bas if\b", "'as if'"),
    (r"\bsuch a (surprise|shocker)\b|\bshocker\b", "'shocker'"),
    (r"\bnot like i('m| am) (bitter|complaining|jealous)\b", "'not like I'm bitter'"),
    (r"\b(great|wonderful|fantastic|perfect|brilliant)\b[,.!]?\s+(another|more|now)\b", "positive word + 'another'"),
    (r"\btruly living the dream\b|\bliving the dream\b", "'living the dream'"),
    (r"(^|\s)/s\b", "'/s' tag"),
]
_TYPO_CUES = [
    (r"\.{3,}|…", "ellipsis"),
    (r"\b(?:[a-z][A-Z]){2,}[a-z]?\b", "aLtErNaTiNg case"),
    (r"[\"“](great|nice|helpful|smart|genius|wonderful|perfect|amazing|good)[\"”]", "scare-quoted praise"),
    (r"\b[A-Z]{3,}\b", "ALL-CAPS emphasis"),
    (r"(\w)\1{2,}", "letter elongation"),
    (r"!{3,}", "repeated !"),
]
LEXICAL_CUES = [(re.compile(p, re.I), n) for p, n in _LEXICAL_CUES]
TYPO_CUES = [(re.compile(p), n) for p, n in _TYPO_CUES]

_SKIN_VS = re.compile("[\U0001F3FB-\U0001F3FF\uFE0F\u200D]")


# --------------------------------------------------------------------------- text utils
def preprocess(text: str) -> str:
    """Normalise @mentions and URLs the way the Cardiff models were trained."""
    text = re.sub(r"@\w+", "@user", text)
    text = re.sub(r"https?://\S+", "http", text)
    return text.strip()


def strip_emoji(text: str) -> str:
    return re.sub(r"\s{2,}", " ", emoji.replace_emoji(text, replace="")).strip()


def norm_emoji(e: str) -> str:
    return _SKIN_VS.sub("", e)


def split_sentences(text: str) -> List[str]:
    """Regex splitter; emoji/punctuation-only fragments are glued to the previous sentence."""
    parts: List[str] = []
    for line in re.split(r"\n+", text.strip()):
        parts += re.split(r"(?<=[.!?…])\s+", line.strip())
    merged: List[str] = []
    for p in parts:
        p = p.strip()
        if not p:
            continue
        if merged:  # "…needed... 🙄 Whatever" -> the 🙄 belongs to the previous sentence
            lead, pos = "", 0
            for m in emoji.emoji_list(p):
                if p[pos:m["match_start"]].strip() == "":
                    lead += m["emoji"]
                    pos = m["match_end"]
                else:
                    break
            if lead:
                merged[-1] += " " + lead
                p = p[pos:].strip()
                if not p:
                    continue
        if merged and not re.search(r"[A-Za-z0-9]", strip_emoji(p)):
            merged[-1] += " " + p
        else:
            merged.append(p)
    return merged[:MAX_SENTENCES]


def emoji_signal(text: str) -> Tuple[List[str], List[float]]:
    """Return (emoji/emoticon symbols found, their valences)."""
    symbols, vals = [], []
    for m in emoji.emoji_list(text):
        e = m["emoji"]
        symbols.append(e)
        v = EMOJI_VALENCE.get(norm_emoji(e))
        if v is not None:
            vals.append(v)
    for m in _EMOTICON_RE.finditer(re.sub(r"https?://\S+", "", text)):
        tok = m.group(1).replace("-", "")
        if tok in EMOTICON_VALENCE:
            symbols.append(m.group(1))
            vals.append(EMOTICON_VALENCE[tok])
    return symbols, vals


def detect_cues(text: str) -> Tuple[List[str], float]:
    cues, score = [], 0.0
    for pat, name in LEXICAL_CUES:
        if pat.search(text):
            cues.append(name)
            score += 0.30
    for pat, name in TYPO_CUES:
        if pat.search(text):
            cues.append(name)
            score += 0.12
    return cues, min(score, 1.0)


def label_from_score(score: float) -> str:
    if score >= LABEL_THRESHOLD:
        return "positive"
    if score <= -LABEL_THRESHOLD:
        return "negative"
    return "neutral"


# --------------------------------------------------------------------------- result types
@dataclass
class SentenceResult:
    text: str
    literal: float            # what the words + emojis say on the surface
    intended: float           # after sarcasm-aware inversion
    label: str
    probs: Dict[str, float]
    irony: float              # raw irony-model probability
    sarcasm: float            # fused sarcasm probability
    is_sarcastic: bool
    emojis: List[str]
    emoji_valence: Optional[float]
    cues: List[str]
    emotion: str
    emotions: Dict[str, float]


@dataclass
class TextProfile:
    sentences: List[SentenceResult]
    mean: float
    overall: float            # recency-weighted "current mood"
    overall_label: str
    volatility: float
    slope: float
    trend: str
    sarcasm_rate: float
    sarcasm_shift: float      # mean(intended - literal): how much sarcasm-awareness changed the read
    dominant_emotion: str
    emotion_mean: Dict[str, float]
    emoji_count: int


# --------------------------------------------------------------------------- detector
class SentimentDetector:
    def __init__(self, device: Optional[str] = None):
        self.device = device or ("cuda" if torch.cuda.is_available() else "cpu")
        self._cache: Dict[str, tuple] = {}

    # ---- model plumbing
    def _load(self, key: str):
        if key not in self._cache:
            name, default = MODELS[key]
            tok = AutoTokenizer.from_pretrained(name)
            model = AutoModelForSequenceClassification.from_pretrained(name).to(self.device).eval()
            id2label = [str(model.config.id2label[i]).lower() for i in range(model.config.num_labels)]
            labels = default if all(l.startswith("label_") for l in id2label) else id2label
            self._cache[key] = (tok, model, labels)
        return self._cache[key]

    def warmup(self) -> None:
        for k in MODELS:
            self._load(k)

    def _idx(self, key: str, label: str, fallback: int) -> int:
        labels = self._load(key)[2]
        return labels.index(label) if label in labels else fallback

    @torch.inference_mode()
    def _probs(self, key: str, texts: List[str], batch_size: int = 32) -> np.ndarray:
        tok, model, _ = self._load(key)
        out = []
        for i in range(0, len(texts), batch_size):
            enc = tok(texts[i:i + batch_size], padding=True, truncation=True,
                      max_length=128, return_tensors="pt").to(self.device)
            out.append(torch.softmax(model(**enc).logits, dim=-1).cpu().numpy())
        return np.concatenate(out) if out else np.zeros((0, 1))

    # ---- convenience for evaluation scripts
    def sentiment_probs(self, texts: List[str]) -> np.ndarray:
        """Baseline model only. Columns: [negative, neutral, positive]."""
        P = self._probs("sentiment", [preprocess(t) for t in texts])
        return P[:, [self._idx("sentiment", "negative", 0), self._idx("sentiment", "neutral", 1),
                     self._idx("sentiment", "positive", 2)]]

    def irony_probs(self, texts: List[str]) -> np.ndarray:
        """Baseline irony model only."""
        return self._probs("irony", [preprocess(t) for t in texts])[:, self._idx("irony", "irony", 1)]

    # ---- main API
    def analyze(self, text: str, sarcasm_aware: bool = True, emoji_aware: bool = True) -> TextProfile:
        sentences = split_sentences(text)
        if not sentences:
            raise ValueError("No text to analyze.")

        raw = [preprocess(s) for s in sentences]
        stripped = [preprocess(strip_emoji(s)) or preprocess(s) for s in sentences]
        model_in = raw if emoji_aware else stripped

        P_full = self._probs("sentiment", model_in)
        P_txt = self._probs("sentiment", stripped) if emoji_aware else P_full
        irony = self._probs("irony", model_in)[:, self._idx("irony", "irony", 1)]
        emo = self._probs("emotion", model_in)
        emo_labels = self._load("emotion")[2]
        ineg, ineu, ipos = (self._idx("sentiment", "negative", 0), self._idx("sentiment", "neutral", 1),
                            self._idx("sentiment", "positive", 2))

        results: List[SentenceResult] = []
        for i, s in enumerate(sentences):
            neg, neu, pos = float(P_full[i][ineg]), float(P_full[i][ineu]), float(P_full[i][ipos])
            model_score = pos - neg
            txt_score = float(P_txt[i][ipos] - P_txt[i][ineg])

            found, vals = emoji_signal(s) if emoji_aware else ([], [])
            emo_val = float(np.mean(vals)) if vals else None
            if emo_val is not None:
                w = min(0.4, 0.2 * len(vals))          # emojis get up to 40% of the say
                literal = (1 - w) * model_score + w * emo_val
            else:
                literal = model_score
            literal = float(np.clip(literal, -1, 1))

            cues, cue_score = detect_cues(s)
            if emo_val is not None and txt_score > 0.25:
                ironic = any(norm_emoji(e) in IRONIC_EMOJI for e in found)
                if emo_val < -0.25 or ironic:           # "Great job." + eye-roll
                    cues.append("positive words + negative/ironic emoji")
                    cue_score = min(1.0, cue_score + 0.35)

            irony_p = float(irony[i])
            sarcasm = 1 - (1 - 0.85 * irony_p) * (1 - cue_score)   # noisy-OR fusion
            is_sarc = sarcasm >= SARCASM_THRESHOLD

            if sarcasm_aware and is_sarc:
                intended = -literal * (0.4 + 0.6 * sarcasm) if literal > 0.1 else literal - 0.25 * sarcasm
            else:
                intended = literal
            intended = float(np.clip(intended, -1, 1))

            label = label_from_score(intended)
            if not (sarcasm_aware and is_sarc) and min(pos, neg) > 0.3 and abs(intended) < LABEL_THRESHOLD:
                label = "mixed"

            e_probs = {emo_labels[j]: float(emo[i][j]) for j in range(len(emo_labels))}
            results.append(SentenceResult(
                text=s, literal=literal, intended=intended, label=label,
                probs={"negative": neg, "neutral": neu, "positive": pos},
                irony=irony_p, sarcasm=float(sarcasm), is_sarcastic=is_sarc,
                emojis=found, emoji_valence=emo_val, cues=cues,
                emotion=max(e_probs, key=e_probs.get), emotions=e_probs))

        return self._profile(results)

    def _profile(self, results: List[SentenceResult]) -> TextProfile:
        n = len(results)
        scores = np.array([r.intended for r in results])
        weights = 0.5 ** ((n - 1 - np.arange(n)) / RECENCY_HALF_LIFE)
        overall = float((weights * scores).sum() / weights.sum())
        slope = float(np.polyfit(np.arange(n), scores, 1)[0]) if n >= 3 else 0.0
        trend = "improving" if slope > 0.05 else "declining" if slope < -0.05 else "stable"
        emo_mean = {k: float(np.mean([r.emotions[k] for r in results])) for k in results[0].emotions}
        return TextProfile(
            sentences=results, mean=float(scores.mean()), overall=overall,
            overall_label=label_from_score(overall), volatility=float(scores.std()),
            slope=slope, trend=trend,
            sarcasm_rate=float(np.mean([r.is_sarcastic for r in results])),
            sarcasm_shift=float(np.mean([r.intended - r.literal for r in results])),
            dominant_emotion=max(emo_mean, key=emo_mean.get), emotion_mean=emo_mean,
            emoji_count=sum(len(r.emojis) for r in results))

    # ---- explainability: leave-one-word-out occlusion on the literal sentiment score
    def explain(self, sentence: str, emoji_aware: bool = True) -> List[Tuple[str, Optional[str]]]:
        words = sentence.split()
        if len(words) < 2 or len(words) > MAX_EXPLAIN_WORDS:
            return [(sentence + " ", None)]
        variants = [sentence] + [" ".join(words[:i] + words[i + 1:]) for i in range(len(words))]
        prep = (lambda t: preprocess(t)) if emoji_aware else (lambda t: preprocess(strip_emoji(t)))
        P = self._probs("sentiment", [prep(v) for v in variants])
        ipos, ineg = self._idx("sentiment", "positive", 2), self._idx("sentiment", "negative", 0)
        sc = P[:, ipos] - P[:, ineg]
        contrib = sc[0] - sc[1:]          # >0: removing the word lowers the score, i.e. it pushed positive
        out: List[Tuple[str, Optional[str]]] = []
        for w, c in zip(words, contrib):
            lab = "pushes positive" if c > 0.08 else "pushes negative" if c < -0.08 else None
            out.append((w + " ", lab))
        return out