Master sentiment analysis in Python

import random
import re
import sys
import time
from typing import List, Tuple

import numpy as np
import scipy.sparse as sp

import nltk
from nltk.corpus import movie_reviews, stopwords
from nltk.sentiment.vader import SentimentIntensityAnalyzer

from sklearn.model_selection import train_test_split
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics import accuracy_score, classification_report, confusion_matrix

import torch
from torch import nn
from torch.utils.data import Dataset, DataLoader

# ---------------------- setup ----------------------

SEED = 42
def set_seed(seed=SEED):
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    if torch.cuda.is_available():
        torch.cuda.manual_seed_all(seed)

USE_WANDB = True
try:
    import wandb
except Exception:
    USE_WANDB = False
    wandb = None

def setup_nltk():
    nltk.download("movie_reviews", quiet=True)
    nltk.download("vader_lexicon", quiet=True)
    nltk.download("stopwords", quiet=True)

# ---------------------- data ----------------------

def load_movie_reviews() -> Tuple[List[str], List[str]]:
    docs = []
    for fileid in movie_reviews.fileids():
        text = " ".join(movie_reviews.words(fileid))
        label = movie_reviews.categories(fileid)[0]
        docs.append((text, label))
    random.shuffle(docs)
    texts = [t for t, _ in docs]
    labels = [y for _, y in docs]
    return texts, labels

def preprocess_texts(texts: List[str]) -> List[str]:
    sw = set(stopwords.words("english"))
    for keep in ["no", "not", "nor"]:
        if keep in sw:
            sw.remove(keep)

    def clean(t: str) -> str:
        t = t.lower()
        t = re.sub(r"[^a-z\s]", " ", t)
        words = [w for w in t.split() if w not in sw]
        return " ".join(words)

    return [clean(t) for t in texts]

def vader_baseline(texts_test: List[str], labels_test: List[str]) -> float:
    sia = SentimentIntensityAnalyzer()
    preds = []
    for t in texts_test:
        c = sia.polarity_scores(t)["compound"]
        preds.append("pos" if c >= 0 else "neg")
    return accuracy_score(labels_test, preds)

def vectorize(train_clean: List[str], test_clean: List[str]):
    # sublinear_tf and dtype=float32 keep values small/stable
    vec = TfidfVectorizer(
        ngram_range=(1, 2),
        min_df=2,
        stop_words="english",
        sublinear_tf=True,
        norm="l2",
        dtype=np.float32,
    )
    X_train = vec.fit_transform(train_clean)
    X_test = vec.transform(test_clean)
    return vec, X_train, X_test

def to_binary(labels: List[str]) -> np.ndarray:
    return np.array([1 if y == "pos" else 0 for y in labels], dtype=np.int64)

# ---------------------- dataset ----------------------

class TFIDFDataset(Dataset):
    def __init__(self, X_csr: sp.csr_matrix, y: np.ndarray):
        self.X = X_csr
        self.y = y

    def __len__(self):
        return self.X.shape[0]

    def __getitem__(self, idx):
        row = self.X[idx]
        x = torch.from_numpy(row.toarray().astype(np.float32).ravel())
        y = torch.tensor(self.y[idx], dtype=torch.float32)
        return x, y

# ---------------------- model ----------------------

class MLP(nn.Module):
    def __init__(self, in_dim: int, hidden: int = 256, dropout: float = 0.3):
        super().__init__()
        self.fc1 = nn.Linear(in_dim, hidden)
        self.fc2 = nn.Linear(hidden, hidden)
        self.out = nn.Linear(hidden, 1)
        self.drop = nn.Dropout(dropout)
        self.act = nn.ReLU()
        self._init_weights()

    def _init_weights(self):
        for m in [self.fc1, self.fc2, self.out]:
            nn.init.xavier_uniform_(m.weight)
            nn.init.zeros_(m.bias)

    def forward(self, x):
        x = self.act(self.fc1(x))
        x = self.drop(x)
        x = self.act(self.fc2(x))
        x = self.drop(x)
        return self.out(x).squeeze(1)  # logits

# ---------------------- train/eval ----------------------

def epoch_run(model, loader, device, criterion, optimizer=None, max_grad_norm=1.0):
    is_train = optimizer is not None
    model.train() if is_train else model.eval()
    total_loss = 0.0
    total_correct = 0
    total = 0
    with torch.set_grad_enabled(is_train):
        for xb, yb in loader:
            xb = xb.to(device, non_blocking=True)
            yb = yb.to(device, non_blocking=True)
            logits = model(xb)

            # clamp logits to avoid inf in loss (rare but safe)
            logits = torch.clamp(logits, -20.0, 20.0)

            loss = criterion(logits, yb)
            if torch.isnan(loss) or torch.isinf(loss):
                # skip pathological batch
                continue
            if is_train:
                optimizer.zero_grad(set_to_none=True)
                loss.backward()
                nn.utils.clip_grad_norm_(model.parameters(), max_grad_norm)
                optimizer.step()

            total_loss += loss.item() * xb.size(0)
            preds = (torch.sigmoid(logits) >= 0.5).long()
            total_correct += (preds == yb.long()).sum().item()
            total += xb.size(0)
    avg_loss = total_loss / max(total, 1)
    acc = total_correct / max(total, 1)
    return avg_loss, acc

def evaluate_final(model, loader, device, y_true_labels: List[str]):
    model.eval()
    preds_all = []
    with torch.no_grad():
        for xb, _ in loader:
            xb = xb.to(device)
            logits = model(xb)
            preds = (torch.sigmoid(logits) >= 0.5).long().cpu().numpy().tolist()
            preds_all.extend(preds)
    pred_labels = ["pos" if p == 1 else "neg" for p in preds_all]
    acc = accuracy_score(y_true_labels, pred_labels)
    rep = classification_report(y_true_labels, pred_labels, target_names=["neg", "pos"])
    cm = confusion_matrix(y_true_labels, pred_labels, labels=["neg", "pos"])
    return acc, rep, cm

def demo_predictions(vec, model, device):
    examples = [
        "I absolutely loved this movie. Brilliant performances!",
        "I really hate this film. It was a waste of time.",
        "I don't hate it, but it's not good either.",
        "The ambiance was great, but the service was slow.",
        "I love waiting in long lines"
    ]
    sw = set(stopwords.words("english"))
    for keep in ["no", "not", "nor"]:
        if keep in sw:
            sw.remove(keep)

    def clean(t: str) -> str:
        t = t.lower()
        t = re.sub(r"[^a-z\s]", " ", t)
        return " ".join([w for w in t.split() if w not in sw])

    X = vec.transform([clean(t) for t in examples])
    X = torch.from_numpy(X.toarray().astype(np.float32)).to(device)
    model.eval()
    with torch.no_grad():
        logits = model(X)
        preds = (torch.sigmoid(logits) >= 0.5).long().cpu().numpy().tolist()
    labels = ["pos" if p == 1 else "neg" for p in preds]
    print("\nDemo predictions:")
    for t, p in zip(examples, labels):
        print(f"[{p}] {t}")

# ---------------------- main ----------------------

def main():
    set_seed()
    setup_nltk()

    texts, labels = load_movie_reviews()
    print(f"Total reviews: {len(texts)}")
    print(f"Positive reviews: {labels.count('pos')}")
    print(f"Negative reviews: {labels.count('neg')}")

    X_train_raw, X_test_raw, y_train_labels, y_test_labels = train_test_split(
        texts, labels, test_size=0.2, random_state=SEED, stratify=labels
    )

    # quick baseline (print only)
    vader_acc = vader_baseline(X_test_raw, y_test_labels)
    print(f"\nVADER baseline accuracy: {vader_acc:.3f}")

    # tf-idf
    X_train_clean = preprocess_texts(X_train_raw)
    X_test_clean = preprocess_texts(X_test_raw)
    vec, X_train_tfidf, X_test_tfidf = vectorize(X_train_clean, X_test_clean)

    y_train_bin = to_binary(y_train_labels)
    y_test_bin = to_binary(y_test_labels)

    # loaders
    train_ds = TFIDFDataset(X_train_tfidf, y_train_bin)
    val_ds = TFIDFDataset(X_test_tfidf, y_test_bin)
    train_loader = DataLoader(train_ds, batch_size=128, shuffle=True, num_workers=0)
    val_loader = DataLoader(val_ds, batch_size=256, shuffle=False, num_workers=0)

    # device
    # device = torch.device("cuda" if torch.cuda.is_available()
                        #   else ("mps" if torch.backends.mps.is_available() else "cpu"))
    device = 'cpu'
    # model/opt/loss with safer settings
    model = MLP(in_dim=X_train_tfidf.shape[1], hidden=256, dropout=0.3).to(device)
    criterion = nn.BCEWithLogitsLoss()
    optimizer = torch.optim.AdamW(model.parameters(), lr=5e-4, weight_decay=1e-2)

    # wandb
    run = None
    if USE_WANDB:
        try:
            run = wandb.init(
                project="sentiment_analysis_tutorial",
                name="MLP_TFIDF_Stable",
                mode="online",
                config={
                    "model": "MLP",
                    "hidden": 256,
                    "dropout": 0.3,
                    "optimizer": "AdamW",
                    "lr": 5e-4,
                    "weight_decay": 1e-2,
                    "batch_size": 128,
                    "epochs": 12,
                    "num_features": int(X_train_tfidf.shape[1]),
                    "num_train_samples": int(X_train_tfidf.shape[0]),
                    "num_val_samples": int(X_test_tfidf.shape[0]),
                },
            )
        except Exception:
            try:
                run = wandb.init(project="sentiment_analysis_tutorial", name="MLP_TFIDF_Stable", mode="offline")
            except Exception:
                run = None

    epochs = 12
    for epoch in range(1, epochs + 1):
        t0 = time.time()
        train_loss, train_acc = epoch_run(model, train_loader, device, criterion, optimizer, max_grad_norm=1.0)
        val_loss, val_acc = epoch_run(model, val_loader, device, criterion, optimizer=None)
        dt = time.time() - t0

        if run is not None:
            wandb.log(
                {
                    "epoch": epoch,
                    "train_loss": float(train_loss),
                    "train_accuracy": float(train_acc),
                    "val_loss": float(val_loss),
                    "val_accuracy": float(val_acc),
                    "epoch_time_sec": float(dt),
                }
            )

        print(f"epoch {epoch:02d}  train_loss {train_loss:.4f}  train_acc {train_acc:.3f}  "
              f"val_loss {val_loss:.4f}  val_acc {val_acc:.3f}  time {dt:.1f}s")

    if run is not None:
        wandb.finish()

    acc, report, cm = evaluate_final(model, val_loader, device, y_test_labels)
    print(f"\nMLP TF IDF accuracy: {acc:.3f}")
    print("\nClassification report:\n" + report)
    print("Confusion matrix [rows true, cols pred] order neg pos:")
    print(cm)

    demo_predictions(vec, model, device)

if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        print("Error:", e, file=sys.stderr)
        raise

import time
import re
from typing import List, Tuple

import nltk
from nltk.corpus import movie_reviews
from nltk.sentiment.vader import SentimentIntensityAnalyzer

from sklearn.model_selection import train_test_split
from sklearn import metrics

from openai import OpenAI


# 1. NLTK setup
def setup_nltk():
    nltk.download("movie_reviews", quiet=True)
    nltk.download("vader_lexicon", quiet=True)


# 2. Data loading and split
def load_data() -> Tuple[List[str], List[str]]:
    texts, labels = [], []
    for fid in movie_reviews.fileids():
        texts.append(" ".join(movie_reviews.words(fid)))
        labels.append(movie_reviews.categories(fid)[0])
    return texts, labels


# 3. Ollama OpenAI compatible client
OLLAMA_BASE_URL = "http://localhost:11434/v1"
MODEL_NAME = "gemma3:270m"

client = OpenAI(base_url=OLLAMA_BASE_URL, api_key="ollama")  # required by client, unused by Ollama

SYSTEM_PROMPT = (
    "You are a rigorous sentiment classifier.\n"
    "Output ONLY one word: positive or negative.\n"
    "No punctuation, no extra words, no JSON."
)

# Five in-context examples
FEW_SHOT = [
    ("I absolutely loved this movie. Brilliant pacing and acting.", "positive"),
    ("This was a complete waste of time. Boring and poorly written.", "negative"),
    ("A delightful surprise with charming performances.", "positive"),
    ("Awful editing and the story made no sense at all.", "negative"),
    ("Not perfect, but I left the theater smiling.", "positive"),
]

# Final user instruction template
USER_TEMPLATE = (
    "Classify the overall sentiment of this text.\n"
    "Respond with exactly one word: positive or negative.\n"
    "Text:\n"
    "{text}"
)

def build_messages_with_fewshot(text: str):
    msgs = [{"role": "system", "content": SYSTEM_PROMPT}]
    for ex_text, ex_label in FEW_SHOT:
        msgs.append({
            "role": "user",
            "content": (
                "Classify the overall sentiment of this text.\n"
                "Respond with exactly one word: positive or negative.\n"
                "Text:\n"
                f"{ex_text}"
            )
        })
        msgs.append({"role": "assistant", "content": ex_label})
    msgs.append({"role": "user", "content": USER_TEMPLATE.format(text=text)})
    return msgs

def gemma_label(text: str) -> str:
    resp = client.chat.completions.create(
        model=MODEL_NAME,
        messages=build_messages_with_fewshot(text),
        temperature=0.0,
    )
    content = resp.choices[0].message.content.strip().lower()
    # Be robust to accidental extra tokens
    word = re.findall(r"[a-z]+", content)
    lbl = word[0] if word else "positive"
    if lbl not in {"positive", "negative"}:
        if "negative" in content:
            lbl = "negative"
        elif "positive" in content:
            lbl = "positive"
        else:
            lbl = "positive"
    return lbl


# 4. Map Gemma labels to dataset labels
def map_label(lbl: str) -> str:
    return "pos" if lbl == "positive" else "neg"


# Optional baseline with VADER
def vader_baseline(texts: List[str], gold: List[str]) -> float:
    sia = SentimentIntensityAnalyzer()
    preds = []
    for t in texts:
        c = sia.polarity_scores(t)["compound"]
        preds.append("pos" if c >= 0 else "neg")
    return metrics.accuracy_score(gold, preds)


def evaluate_gemma(texts: List[str], sleep_sec: float = 0.0) -> List[str]:
    preds = []
    start = time.time()
    for i, t in enumerate(texts, 1):
        gem = gemma_label(t)
        preds.append(map_label(gem))
        if i % 25 == 0:
            elapsed = time.time() - start
            print(f"Processed {i} samples in {elapsed:.1f}s")
        if sleep_sec > 0:
            time.sleep(sleep_sec)
    return preds


def main():
    setup_nltk()
    texts, labels = load_data()

    X_train, X_test, y_train, y_test = train_test_split(
        texts, labels, test_size=0.2, random_state=42, stratify=labels
    )

    print("Evaluating Gemma 270M via Ollama on 400 test reviews with 5-shot prompts...")
    gemma_preds = evaluate_gemma(X_test)

    acc = metrics.accuracy_score(y_test, gemma_preds)
    report = metrics.classification_report(y_test, gemma_preds, target_names=["neg", "pos"])
    cm = metrics.confusion_matrix(y_test, gemma_preds, labels=["neg", "pos"])

    print(f"\nGemma accuracy: {acc:.3f}")
    print("\nClassification report:")
    print(report)
    print("Confusion matrix [rows true, cols pred] order neg pos:")
    print(cm)

    print("\nSample predictions:")
    for i in range(5):
        excerpt = X_test[i][:120].replace("\n", " ") + "..."
        print(f"Gold={y_test[i]} Pred={gemma_preds[i]}  |  {excerpt}")

    try:
        v_acc = vader_baseline(X_test, y_test)
        print(f"\nVADER baseline accuracy: {v_acc:.3f}")
    except Exception as e:
        print(f"\nVADER baseline skipped: {e}")


if __name__ == "__main__":
    main()