twitter_sentiment_analysis.py

"""
Twitter Sentiment Analysis - Ukraine-Russia War

This script performs sentiment analysis and topic modeling on tweets related
to the Ukraine-Russia conflict.

Prerequisites:
    pip install pandas numpy nltk gensim spacy scikit-learn pyLDAvis
    python -m spacy download en_core_web_sm

Dataset:
    The analysis uses the 'Ukraine_tweetys.txt' file containing tweets about
    the Ukraine-Russia conflict.
"""

import os
import pandas as pd
import numpy as np
import nltk
from nltk.corpus import stopwords
from nltk.tokenize import word_tokenize
from nltk.sentiment.vader import SentimentIntensityAnalyzer
import gensim
from gensim.utils import simple_preprocess
import gensim.corpora as corpora
import spacy
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.decomposition import NMF
from pprint import pprint

# Configure pandas display options
pd.set_option("display.max_columns", None)
pd.set_option("display.max_rows", None)
pd.set_option('display.max_colwidth', None)


def load_data(file_path='Ukraine_tweetys.txt'):
    """
    Load the Twitter dataset from a CSV file.

    Args:
        file_path: Path to the data file

    Returns:
        DataFrame containing the tweet data
    """
    if not os.path.exists(file_path):
        raise FileNotFoundError(
            f"Data file '{file_path}' not found. "
            "Please ensure the file is in the correct directory."
        )

    df = pd.read_csv(file_path)
    print(f"Loaded {len(df)} tweets from {file_path}")
    return df


def clean_basic_data(df):
    """
    Perform basic data cleaning operations.

    Args:
        df: Input DataFrame

    Returns:
        Cleaned DataFrame with duplicates removed and English tweets only
    """
    # Remove duplicates
    df = df.drop_duplicates()
    print(f"After removing duplicates: {len(df)} tweets")

    # Find tweets in English only
    df_en = df[df['language'] == 'en']
    print(f"English tweets: {len(df_en)}")

    # Remove unnecessary columns
    df_en_clean = df_en.drop(
        columns=['Tweet Id', 'Unnamed: 0', 'Unnamed: 0.1', 'verified'],
        axis=1,
        errors='ignore'
    )

    return df_en_clean


def filter_by_location(df_en_clean, location_keyword, exclude_keywords=None):
    """
    Filter tweets by location keyword.

    Args:
        df_en_clean: Clean English tweets DataFrame
        location_keyword: Keyword to search for in location
        exclude_keywords: List of keywords to exclude

    Returns:
        DataFrame of tweets from specified location and list of positions
    """
    if exclude_keywords is None:
        exclude_keywords = []

    position = 0
    positions = []

    for location in df_en_clean.location:
        if type(location) != float:
            location_lower = location.lower().split()

            # Check if location keyword is present
            if location_keyword in location_lower:
                # Check if none of the exclude keywords are present
                if not any(keyword in location_lower for keyword in exclude_keywords):
                    positions.append(position)
        position += 1

    # Efficiently collect rows using list comprehension
    rows_list = [df_en_clean.iloc[pos, :] for pos in positions]
    # Concatenate all at once instead of in a loop
    df_location = pd.concat(rows_list, axis=1).transpose() if rows_list else pd.DataFrame()

    return df_location, positions


def get_russia_tweets(df_en_clean):
    """Extract tweets from Russia location."""
    exclude_keywords = ['us', 'not', 'but']
    df_russia_en, positions = filter_by_location(
        df_en_clean, 'russia', exclude_keywords
    )
    print(f"Tweets from Russia: {len(df_russia_en)}")
    return df_russia_en


def get_ukraine_tweets(df_en_clean):
    """Extract tweets from Ukraine location."""
    exclude_keywords = [
        'taxpayer', 'and', 'in', 'us', 'stand', 'is', 'free', 'earth',
        'uk', 'with', 'germany', 'ireland', 'to', 'support', 'ca', 'ny',
        'all', 'or', 'denmark'
    ]
    df_ukraine_en, positions = filter_by_location(
        df_en_clean, 'ukraine', exclude_keywords
    )
    print(f"Tweets from Ukraine: {len(df_ukraine_en)}")
    return df_ukraine_en


def get_other_location_tweets(df_en_clean, df_ukraine_en, df_russia_en):
    """Extract tweets from all other locations."""
    df_other_en = pd.concat([df_en_clean, df_ukraine_en, df_russia_en]).drop_duplicates(keep=False)
    print(f"Tweets from other locations: {len(df_other_en)}")
    return df_other_en


def get_high_impact_tweets(df_en_clean):
    """
    Filter high impact tweets (like, retweet and reply > 1).

    Args:
        df_en_clean: Clean English tweets DataFrame

    Returns:
        DataFrame with high impact tweets
    """
    df_en_highimpact = df_en_clean[df_en_clean['like count'] > 1]
    df_en_highimpact = df_en_highimpact[df_en_highimpact['retweet count'] > 1]
    df_en_highimpact = df_en_highimpact[df_en_highimpact['reply count'] > 1]
    print(f"High impact tweets: {len(df_en_highimpact)}")
    return df_en_highimpact


def clean_text_nltk(pd_column):
    """
    Clean text by removing special characters, stopwords, and Twitter elements.

    Args:
        pd_column: Pandas Series containing text data

    Returns:
        List of cleaned text strings
    """
    stop_words = list(set(stopwords.words('english')))
    clean = []

    for i in pd_column:
        i = i.lower()

        # Preprocess to split the words
        i = i.split('-')
        i = ' '.join(i)
        i = i.split()

        split_sent2 = []

        # Remove stopwords and special characters
        for word in i:
            new_word = []

            # Check each word to remove stop words, @username, website, hash-tag
            if (word not in stop_words and
                word[0] != '@' and
                word[0:4] != 'http' and
                word[0] != '#'):

                # Check each letter in word to remove special characters
                for letter in word:
                    if letter.isalpha():
                        new_word.append(letter)

                word2 = ''.join(new_word)

                # Check stopwords again and minimum length
                if word2 not in stop_words and len(word2) >= 3:
                    split_sent2.append(word2)

        sent2 = ' '.join(split_sent2)
        clean.append(sent2)

    return clean


def nltk_tokenizer(pd_column):
    """
    Tokenize text using NLTK word tokenizer.

    Args:
        pd_column: Pandas Series with full sentences

    Returns:
        List of tokenized sentences
    """
    tokenized = []
    for i in pd_column:
        word_tokens = word_tokenize(i)
        tokenized.append(word_tokens)
    return tokenized


def create_bigrams_trigrams(df_tokenized):
    """
    Create bigram and trigram models.

    Args:
        df_tokenized: Tokenized data

    Returns:
        Bigram and trigram models, and bigrammed data
    """
    # Build the bigram and trigram models
    bigram = gensim.models.Phrases(df_tokenized, min_count=5, threshold=100)
    trigram = gensim.models.Phrases(bigram[df_tokenized], threshold=100)

    # Faster way to get a sentence clubbed as a trigram/bigram
    bigram_mod = gensim.models.phrases.Phraser(bigram)
    trigram_mod = gensim.models.phrases.Phraser(trigram)

    # Create bigrams
    data_words_bigrams = [bigram_mod[doc] for doc in df_tokenized]

    return bigram_mod, trigram_mod, data_words_bigrams


def lemmatization(texts, allowed_postags=['NOUN', 'ADJ', 'VERB', 'ADV']):
    """
    Lemmatize text using spaCy.

    Args:
        texts: List of tokenized texts
        allowed_postags: Parts of speech to keep

    Returns:
        List of lemmatized texts
    """
    nlp = spacy.load('en_core_web_sm')
    texts_out = []

    for sent in texts:
        doc = nlp(" ".join(sent))
        texts_out.append([token.lemma_ for token in doc if token.pos_ in allowed_postags])

    return texts_out


def perform_lda_modeling(corpus, id2word, num_topics=5):
    """
    Perform LDA topic modeling.

    Args:
        corpus: Document-term matrix
        id2word: Dictionary mapping
        num_topics: Number of topics to extract

    Returns:
        Trained LDA model
    """
    lda_model = gensim.models.ldamodel.LdaModel(
        corpus=corpus,
        id2word=id2word,
        num_topics=num_topics,
        random_state=100
    )

    print(f"\n=== LDA Model with {num_topics} Topics ===")
    pprint(lda_model.print_topics())

    return lda_model


def perform_nmf_modeling(lemmatized_tweets, n_topics=5):
    """
    Perform NMF topic modeling.

    Args:
        lemmatized_tweets: List of lemmatized tweet strings
        n_topics: Number of topics to extract

    Returns:
        NMF model, TF-IDF vectorizer, and document-term matrix
    """
    # Create TF-IDF matrix
    tfidf = TfidfVectorizer(max_df=0.95, min_df=2, stop_words='english')
    dtm = tfidf.fit_transform(lemmatized_tweets)

    # Create NMF model
    nmf_model = NMF(n_components=n_topics, random_state=42)
    nmf_model.fit(dtm)

    # Print topics
    print(f"\n=== NMF Model with {n_topics} Topics ===")
    topic_lists = []
    for index, topic in enumerate(nmf_model.components_):
        print(f'THE TOP 30 WORDS FOR TOPIC #{index}')
        topic_list = [tfidf.get_feature_names_out()[i] for i in topic.argsort()[-30:]]
        print(topic_list)
        topic_lists.append(topic_list)
        print()

    return nmf_model, tfidf, dtm, topic_lists


def apply_sentiment_analysis(df, text_column='Text'):
    """
    Apply sentiment analysis to a DataFrame's text column.

    Args:
        df: DataFrame to analyze
        text_column: Name of the column containing text (default: 'Text')

    Returns:
        DataFrame with added 'sentiment_nltk' column
    """
    # Download VADER lexicon if not already present
    try:
        nltk.data.find('vader_lexicon')
    except LookupError:
        nltk.download('vader_lexicon')

    sid = SentimentIntensityAnalyzer()

    def sentiment_rating(string):
        sid_dict = sid.polarity_scores(string)
        if sid_dict['compound'] >= 0.33:
            return 'Positive'
        elif sid_dict['compound'] <= -0.33:
            return 'Negative'
        else:
            return 'Neutral'

    sentiment_results = [sentiment_rating(tweet) for tweet in df[text_column]]
    df['sentiment_nltk'] = sentiment_results

    return df


def save_dataframes(df_en_clean, df_russia_en, df_ukraine_en, df_other_en,
                   df_en_highimpact):
    """Save all processed DataFrames to CSV files."""
    df_en_clean.to_csv('cleaned_tweets.csv', index=False)
    df_russia_en.to_csv('Russia_loc.csv', index=False)
    df_ukraine_en.to_csv('Ukraine_loc.csv', index=False)
    df_other_en.to_csv('Other_loc.csv', index=False)
    df_en_highimpact.to_csv('highimpact_nmf.csv', index=False)

    print("\n=== Files Saved ===")
    print("- cleaned_tweets.csv")
    print("- Russia_loc.csv")
    print("- Ukraine_loc.csv")
    print("- Other_loc.csv")
    print("- highimpact_nmf.csv")


def main():
    """Main execution function."""
    print("=" * 60)
    print("Twitter Sentiment Analysis - Ukraine-Russia War")
    print("=" * 60)

    # Download NLTK resources if needed
    try:
        nltk.data.find('stopwords')
    except LookupError:
        print("Downloading NLTK stopwords...")
        nltk.download('stopwords')

    try:
        nltk.data.find('vader_lexicon')
    except LookupError:
        print("Downloading NLTK vader_lexicon...")
        nltk.download('vader_lexicon')

    # Load and clean data
    print("\n=== Step 1: Loading Data ===")
    df = load_data('Ukraine_tweetys.txt')

    print("\n=== Step 2: Basic Data Cleaning ===")
    df_en_clean = clean_basic_data(df)

    # Filter by location
    print("\n=== Step 3: Location-Based Filtering ===")
    df_russia_en = get_russia_tweets(df_en_clean)
    df_ukraine_en = get_ukraine_tweets(df_en_clean)
    df_other_en = get_other_location_tweets(df_en_clean, df_ukraine_en, df_russia_en)

    # Get high impact tweets
    print("\n=== Step 4: High Impact Tweet Analysis ===")
    df_en_highimpact = get_high_impact_tweets(df_en_clean)

    # Text preprocessing for topic modeling
    print("\n=== Step 5: Text Preprocessing ===")
    cleaned_text = clean_text_nltk(df_en_highimpact['Text'])
    df_tokenized = pd.Series(nltk_tokenizer(cleaned_text))

    # Create bigrams and trigrams
    print("Creating bigrams and trigrams...")
    bigram_mod, trigram_mod, data_words_bigrams = create_bigrams_trigrams(df_tokenized)

    # Lemmatization
    print("Performing lemmatization...")
    data_lemmatized = lemmatization(data_words_bigrams,
                                   allowed_postags=['NOUN', 'ADJ', 'VERB', 'ADV'])

    # Create Dictionary and Corpus for LDA
    print("\n=== Step 6: Topic Modeling ===")
    id2word = corpora.Dictionary(data_lemmatized)
    texts = data_lemmatized
    corpus = [id2word.doc2bow(text) for text in texts]

    # LDA topic modeling with different numbers of topics
    lda_model_5 = perform_lda_modeling(corpus, id2word, num_topics=5)
    lda_model_7 = perform_lda_modeling(corpus, id2word, num_topics=7)
    lda_model_10 = perform_lda_modeling(corpus, id2word, num_topics=10)

    # Convert lemmatized tweets to strings for NMF
    lemmatized_tweets = [' '.join(ls) for ls in data_lemmatized]

    # NMF topic modeling
    nmf_model, tfidf, dtm, topic_lists = perform_nmf_modeling(lemmatized_tweets, n_topics=5)

    # Apply topic labels to high impact tweets
    topic_results = nmf_model.transform(dtm)
    df_en_highimpact['Topic'] = topic_results.argmax(axis=1)

    # Define topic labels
    mytopic_dict = {
        0: 'War cause',
        1: 'Military progress',
        2: 'Refugee and human rights',
        3: 'Economic impact of the war',
        4: 'News and journalism in war'
    }
    df_en_highimpact['Topic Label'] = df_en_highimpact['Topic'].map(mytopic_dict)

    print("\nTopic distribution:")
    print(df_en_highimpact['Topic'].value_counts())

    # Sentiment analysis
    print("\n=== Step 7: Sentiment Analysis ===")

    print("\nApplying sentiment analysis to high impact tweets...")
    df_en_highimpact = apply_sentiment_analysis(df_en_highimpact.copy())
    print(df_en_highimpact['sentiment_nltk'].value_counts())

    print("\nApplying sentiment analysis to Russia location tweets...")
    df_russia_en = apply_sentiment_analysis(df_russia_en.copy())
    print(df_russia_en['sentiment_nltk'].value_counts())

    print("\nApplying sentiment analysis to Ukraine location tweets...")
    df_ukraine_en = apply_sentiment_analysis(df_ukraine_en.copy())
    print(df_ukraine_en['sentiment_nltk'].value_counts())

    # Save results
    print("\n=== Step 8: Saving Results ===")
    save_dataframes(df_en_clean, df_russia_en, df_ukraine_en, df_other_en,
                   df_en_highimpact)

    print("\n" + "=" * 60)
    print("Analysis Complete!")
    print("=" * 60)

    return df_en_clean, df_russia_en, df_ukraine_en, df_other_en, df_en_highimpact


if __name__ == "__main__":
    main()