app.py

# -*- coding: utf-8 -*-
"""Untitled3.ipynb

Automatically generated by Colab.

Original file is located at
    https://colab.research.google.com/drive/1BTaF9lue6oXAqEx5zFRq1cnWWQ9YKCiQ
"""

import pandas as pd
import numpy as np
import torch
from transformers import BertTokenizer
import seaborn as sns
import matplotlib.pyplot as plt
from sklearn.feature_extraction.text import CountVectorizer


# Load dataset
file_path = 'spam_ham_dataset.csv'
df = pd.read_csv(file_path)
df.head()

# Preprocessing
#.str.replace(r'[^\w\s]', '', regex=True) removes everthing except letters, numbers, and spaces
# df['text'].str.lower() converts everything in the text column to lower case only
df['text'] = df['text'].str.lower().str.replace(r'[^\w\s]', '', regex=True)
df['text'].head()


sns.countplot(x=df['label'])
plt.title("Spam vs Ham Distribution")
plt.show()

# Calculate text length metrics
df['char_count'] = df['text'].apply(len)
df['word_count'] = df['text'].apply(lambda x: len(x.split()))
# Plot word count distribution for spam and ham
plt.figure(figsize=(12, 5))
sns.histplot(data=df, x='word_count', hue='label', bins=30, kde=True)
plt.xlim(0, 1000)
plt.title("Word Count Distribution by Label")
plt.xlabel("Number of Words")
plt.ylabel("Frequency")
plt.show()

def get_top_words(corpus, n=None):
    vec = CountVectorizer(stop_words='english').fit(corpus)
    bag_of_words = vec.transform(corpus)
    sum_words = bag_of_words.sum(axis=0)
    words_freq = [(word, sum_words[0, idx]) for word, idx in vec.vocabulary_.items()]
    words_freq = sorted(words_freq, key=lambda x: x[1], reverse=True)
    return words_freq[:n]

# Top 10 words for spam
top_spam_words = get_top_words(df[df['label'] == "spam"]['text'], n=10)
print("Top spam words:", top_spam_words)

# Top 10 words for ham
top_ham_words = get_top_words(df[df['label'] == "ham"]['text'], n=10)
print("Top ham words:", top_ham_words)

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.metrics import classification_report

# TF-IDF Vectorization
vectorizer = TfidfVectorizer()
X = vectorizer.fit_transform(df['text'])
y = df['label_num']

# Train-Test Split
from sklearn.model_selection import train_test_split
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# Train Naïve Bayes Model
nb_model = MultinomialNB()
nb_model.fit(X_train, y_train)

# Predictions
y_pred = nb_model.predict(X_test)
print(classification_report(y_test, y_pred))

import pandas as pd
import torch
import torch.nn as nn
import torch.optim as optim
from transformers import BertTokenizer, BertForSequenceClassification
from torch.utils.data import Dataset, DataLoader

# Load dataset
file_path = 'spam_ham_dataset.csv'
df = pd.read_csv(file_path)

# Convert label column to numeric (0 for ham, 1 for spam)
df['label_num'] = df['label'].astype('category').cat.codes

# Load tokenizer
tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')

# Tokenize dataset
encodings = tokenizer(df['text'].tolist(), padding=True, truncation=True, max_length=128, return_tensors="pt")
labels = torch.tensor(df['label_num'].values)

# Custom Dataset
class SpamDataset(Dataset):
    def __init__(self, encodings, labels):
        self.encodings = encodings
        self.labels = labels

    def __len__(self):
        return len(self.labels)

    def __getitem__(self, idx):
        item = {key: val[idx] for key, val in self.encodings.items()}  # Keep as PyTorch tensors
        item['labels'] = torch.tensor(self.labels[idx], dtype=torch.long)  # Ensure labels are `long`
        return item

# Create dataset
dataset = SpamDataset(encodings, labels)

# Split dataset (80% train, 20% validation)
train_size = int(0.8 * len(dataset))
val_size = len(dataset) - train_size
train_dataset, val_dataset = torch.utils.data.random_split(dataset, [train_size, val_size])

# DataLoader Function (Fix Collate)
def collate_fn(batch):
    keys = batch[0].keys()
    collated = {key: torch.stack([b[key] for b in batch]) for key in keys}
    return collated

# Create DataLoader
train_loader = DataLoader(train_dataset, batch_size=8, shuffle=True, collate_fn=collate_fn)
val_loader = DataLoader(val_dataset, batch_size=8, shuffle=False, collate_fn=collate_fn)

# Load BERT model
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = BertForSequenceClassification.from_pretrained("bert-base-uncased", num_labels=2)
model.to(device)

# Define optimizer and loss function
optimizer = optim.AdamW(model.parameters(), lr=5e-5)
loss_fn = nn.CrossEntropyLoss()

# Training Loop
EPOCHS = 10

for epoch in range(EPOCHS):
    model.train()
    total_loss = 0

    for batch in train_loader:
        optimizer.zero_grad()

        # Move batch to device
        inputs = {key: val.to(device) for key, val in batch.items()}
        labels = inputs.pop("labels").to(device)  # Move labels to device

        # Forward pass
        outputs = model(**inputs)
        loss = loss_fn(outputs.logits, labels)

        # Backward pass
        loss.backward()
        optimizer.step()

        total_loss += loss.item()

    avg_loss = total_loss / len(train_loader)
    print(f"Epoch {epoch+1}, Loss: {avg_loss:.4f}")

print("Training complete!")

from sklearn.metrics import classification_report
from transformers import BertTokenizer
import torch
import torch.nn.functional as F

# Classification function
def classify_email(email_text):
    model.eval()  # Set model to evaluation mode

    with torch.no_grad():
        # Tokenize and convert input text to tensor
        inputs = tokenizer(email_text, padding=True, truncation=True, max_length=256, return_tensors="pt")

        # Move inputs to the appropriate device
        inputs = {key: val.to(device) for key, val in inputs.items()}

        # Get model predictions
        outputs = model(**inputs)
        logits = outputs.logits

        # Convert logits to predicted class
        predictions = torch.argmax(logits, dim=1)

        # Convert logits to probabilities using softmax
        probs = F.softmax(logits, dim=1)
        confidence = torch.max(probs).item() * 100  # Convert to percentage

    # Convert numeric prediction to label
    result = "Spam" if predictions.item() == 1 else "Ham"

    return {
        "result": result,
        "confidence": f"{confidence:.2f}%",
    }

# Evaluation function with detailed classification report
def evaluate_model_with_report(val_loader):
    model.eval()  # Set model to evaluation mode
    y_true = []
    y_pred = []
    correct = 0
    total = 0

    with torch.no_grad():
        for batch in val_loader:
            inputs = {key: val.to(device) for key, val in batch.items()}
            labels = inputs.pop("labels").to(device)

            outputs = model(**inputs)
            predictions = torch.argmax(outputs.logits, dim=1)

            # Collect labels and predictions
            y_true.extend(labels.cpu().numpy())
            y_pred.extend(predictions.cpu().numpy())

            # Calculate accuracy
            correct += (predictions == labels).sum().item()
            total += labels.size(0)

    # Calculate accuracy
    accuracy = correct / total if total > 0 else 0
    print(f"Validation Accuracy: {accuracy:.4f}")

    # Print classification report
    print("\nClassification Report:")
    print(classification_report(y_true, y_pred, target_names=["Ham", "Spam"]))

    return accuracy

# Run evaluation with classification report
accuracy = evaluate_model_with_report(val_loader)
print(f"Model Validation Accuracy: {accuracy:.4f}")

## App Deployment Functions

def generate_performance_metrics():
    y_pred = model.predict(X_test)
    accuracy = evaluate_model_with_report(val_loader)
    report = classification_report(y_true, y_pred, target_names=["Ham", "Spam"])
    return {
        "accuracy": f"{accuracy:.2%}",
        "precision": f"{report['1']['precision']:.2%}",
        "recall": f"{report['1']['recall']:.2%}",
        "f1_score": f"{report['1']['f1-score']:.2%}"
    }

def email_analysis_pipeline(email_text):
    results = classify_email(email_text)
    accuracy = evaluate_model_with_report(val_loader)
    return {
        results["result"],
        results["confidence"],
        accuracy
    }

## Gradio Interface

import gradio as gr

# Create Gradio Interface
def create_interface():
    performance_metrics = generate_performance_metrics()

    # Introduction - Title + Brief Description
    with gr.Blocks(css=custom_css) as interface:
        gr.Markdown("Spam Email Classification")
        gr.Markdown(
            """
            Brief description of the project here

            """
        )

        # Email Text Input
        with gr.Row():
            email_input = gr.Textbox(
                lines=8, placeholder="Type or paste your email content here...", label="Email Content"
            )

        # Email Text Results and Analysis
        with gr.Row():
            result_output = gr.HTML(label="Classification Result") # label = [function that prints classification result]
            confidence_output = gr.Textbox(label="Confidence Score", interactive=False)
            accuracy_output = gr.Textbox(label="Accuracy", interactive=False)


        analyze_button = gr.Button("Analyze Email 🕵️‍♂️")

        analyze_button.click(
            fn=email_analysis_pipeline,
            inputs=email_input,
            outputs=[result_output, confidence_output, accuracy_output]
        )

        # Analysis
        gr.Markdown("## 📊 Model Performance Analytics")
        with gr.Row():
            with gr.Column():
                gr.Textbox(value=performance_metrics["accuracy"], label="Accuracy", interactive=False, elem_classes=["metric"])
                gr.Textbox(value=performance_metrics["precision"], label="Precision", interactive=False, elem_classes=["metric"])
                gr.Textbox(value=performance_metrics["recall"], label="Recall", interactive=False, elem_classes=["metric"])
                gr.Textbox(value=performance_metrics["f1_score"], label="F1 Score", interactive=False, elem_classes=["metric"])
            with gr.Column():
                gr.Markdown("### Confusion Matrix")
                gr.HTML(f"<img src='data:image/png;base64,{performance_metrics['confusion_matrix_plot']}' style='max-width: 100%; height: auto;' />")

        gr.Markdown("## 📘 Glossary and Explanation of Labels")
        gr.Markdown(
            """
            ### Labels:
            - **Spam:** Unwanted or harmful emails flagged by the system.
            - **Ham:** Legitimate, safe emails.

            ### Metrics:
            - **Accuracy:** The percentage of correct classifications.
            - **Precision:** Out of predicted Spam, how many are actually Spam.
            - **Recall:** Out of all actual Spam emails, how many are predicted as Spam.
            - **F1 Score:** Harmonic mean of Precision and Recall.
            """
        )

    return interface

# Launch the interface
interface = create_interface()
interface.launch(share=True)

## CSS

# Updated CSS
custom_css = """
body {
    font-family: 'Arial', sans-serif;
    background-image: url('https://cdn.pixabay.com/photo/2016/11/19/15/26/email-1839873_1280.jpg');
    background-size: cover;
    background-position: center;
    background-attachment: fixed;
    color: #333;
}
h1, h2, h3 {
    text-align: center;
    color: #ffffff;
    text-shadow: 2px 2px 4px rgba(0, 0, 0, 0.7);
}
.gradio-container {
    background-color: rgba(255, 255, 255, 0.8);
    border-radius: 10px;
    padding: 20px;
    box-shadow: 0px 4px 10px rgba(0, 0, 0, 0.3);
}
button {
    background-color: #1e90ff;
    color: white;
    padding: 10px 20px;
    border: none;
    border-radius: 5px;
    cursor: pointer;
    font-size: 1.2em;
    transition: transform 0.2s, background-color 0.3s;
}
button:hover {
    background-color: #1c86ee;
    transform: scale(1.05);
}
.highlight {
    background-color: #ffeb3b;
    font-weight: bold;
    padding: 0 3px;
    border-radius: 3px;
}
.metric {
    font-size: 1.2em;
    text-align: center;
    color: #ffffff;
    background-color: #4CAF50;
    border-radius: 8px;
    padding: 10px;
    margin: 10px 0;
    box-shadow: 2px 2px 5px rgba(0, 0, 0, 0.2);
}
"""