Clean and analyze social media usage data with Python (Coursera)

import pandas
from collections import Counter
from os import path
import json
from datetime import datetime

cwd = '../data'
data_path = path.join(cwd, '01.json')
student_path = path.join(cwd, 'students.txt')

exclude_author = ['AutoModerator', 'BlueArchive-ModTeam']

with open(data_path, "r", encoding='utf-8') as outfile:
    data = json.load(outfile)

data_df = pandas.DataFrame.from_dict(data, orient='index')
# [print(col) for col in data_df.columns]
filtered_df = data_df[['title', 'link_flair_text', 'ups', 'created_utc', 'comments', 'num_comments']]
filtered_df.created_utc = filtered_df.created_utc.apply(lambda x: datetime.fromtimestamp(x))
filtered_df.comments = filtered_df.comments.apply(lambda x: [comment['body'] for comment in x if comment['author'] not in exclude_author])
filtered_df.num_comments = filtered_df.comments.apply(lambda x: len(x))
filtered_df = filtered_df.loc[filtered_df['link_flair_text'].isin(['OC ART ', 'NON OC ART ', 'Comic/TL'])]

display(filtered_df.nlargest(10, 'num_comments'))
Counter(filtered_df.link_flair_text)

import spacy
from spacy.tokens import Token
from spacymoji import Emoji
from tqdm.notebook import tqdm

print(spacy.require_gpu())

ba_character_list = list()
with open(student_path, 'r', encoding='utf-8') as f:
    for l in f.readlines():
        ba_character_list.append(l.strip())

# alias groups - (main name, alias-1, alias-2, alias-3, ...)
alias_groups = [
    ('aris', 'alice', 'arisu', 'fridge', 'refrigerator'),
    ('hoshino', 'oji-san', 'ojisan', 'oji san'),
    ('yuuka', 'yuka', 'calculator', '100kg', '100 kg'),
    # ('sensei', 'senseis'), # cannot be detected as plural
    ('hina', 'hinature', 'hiniature')
]

ba_character_list.extend([j for i in alias_groups for j in i])

print('ba_character_list:', len(ba_character_list))
print(ba_character_list)


def BA_char_finder(tk):
    if (student := tk.text.lower()) in ba_character_list:
        for group in alias_groups:
            if student in group:
                student = group[0]
        return student
    else:
        return None


def get_nbors_(tk: spacy.tokens.token.Token, step: int, N: int):
    # step usually should be 1 or -1 - searching range step
    # N is the Nth search result to be returned

    if tk.is_sent_start and step < 0:
        return
    elif tk.is_sent_end and step > 0:
        return
    
    n = 1
    i = step
    while True:
        nbor = tk.nbor(i)
        if nbor.is_stop or nbor._.is_emoji or not nbor.is_alpha:
            pass
        else:
            if n == N:
                return nbor
            else:
                n += 1
                pass
        
        if nbor.is_sent_end or nbor.is_sent_start:
            return

        i += step


def get_nbors(tk):
    return ((
        (
        get_nbors_(tk, -1, 1), # l1
        get_nbors_(tk, 1, 1), # r1
        ),
        (
        get_nbors_(tk, -1, 2), # l2
        get_nbors_(tk, 1, 2), # r2
        ),
        # tk.sent
    ))

# for splitting sentences
@spacy.language.Language.component("custom_boundary")
def custom_boundary(doc):

    delimiters=['...', '\n']

    for token in doc[:-1]:
        if token.text in delimiters:
            doc[token.i+1].is_sent_start = True
    
    return doc


nlp = spacy.load('en_core_web_trf')
nlp.add_pipe("emoji", first=True)
nlp.add_pipe('custom_boundary', before='parser')

Token.set_extension("nbors", getter=get_nbors, force=True)
Token.set_extension("ba_characters", getter=BA_char_finder, force=True)

analysis = nlp.analyze_pipes(pretty=True)

print(spacy.require_gpu())
"""
Cuda compilation tools, release 11.5, V11.5.119
Build cuda_11.5.r11.5/compiler.30672275_0
"""

docs = list()
for t in tqdm(data_text.values()):
    docs.append(nlp(t))

def is_cleanable(tk: spacy.tokens.token.Token):
    # might(?) try and add emoji if that seems plausible
    if (tk.is_alpha and not tk.is_stop and not tk._.is_emoji) or tk.is_sent_end:
        return True
    else:
        return False

def preprocess_tk(tk: spacy.tokens.token.Token):
    if not tk.is_alpha:
        tk = '\n'
    elif tk.is_sent_end:
        tk = tk.lemma_.lower().strip()
        for group in alias_groups:
            if tk in group:
                tk = group[0]
        tk += '\n'
    else:
        tk = tk.lemma_.lower().strip()
        for group in alias_groups:
            if tk in group:
                tk = group[0]
        tk += ' '
    
    return tk


# preprocessed text for FastText
ft_preprocessed = [preprocess_tk(tk) for doc in docs for tk in doc if is_cleanable(tk)]
print('no. of tokens:', len(ft_preprocessed))
ft_preprocessed = ''.join(ft_preprocessed)
ft_preprocessed = re.sub(" +", " ", ft_preprocessed) # fixes double(or more) spaces
ft_preprocessed = re.sub("<[^>]*>", "", ft_preprocessed) # remove html tags

print('no. of comments:', ft_preprocessed.count('\n'))

ft_path = path.join(cwd, 'ft_preprocessed.txt')
with open(ft_path, 'w', encoding='utf-8') as f:
    f.write(ft_preprocessed)

import fasttext
print(ft_path)
model = fasttext.train_unsupervised(ft_path, dim=300)  # pretrainedVectors=path.join(cwd, 'crawl-300d-2M.vec')

Counter([tk._.ba_characters for doc in docs for tk in doc if tk._.ba_characters]).most_common(10)

from sklearn import preprocessing
import numpy as np

# will be the target ba_caharcter to calculate the similarity/score
target_character = 'hina'

# `cm` as in correlation matrix (not exactly a matrix but...)
cm_original = [tk._.nbors for doc in tqdm(docs) for tk in doc if tk._.ba_characters == target_character]

cm1_original = [j for i in (cm_original) for j in i[0] if j and (j._.ba_characters != target_character)] # bigram range
cm2_original = [j for i in (cm_original) for j in i[1] if j and (j._.ba_characters != target_character)] # trigram range

print('cm_1', cm1 := Counter([i.text for i in cm1_original]).most_common(500))
print('cm_2', cm2 := Counter([i.text for i in cm2_original]).most_common(500))

cm1_words = [i[0] for i in cm1]
cm2_words = [i[0] for i in cm2]

cm1_count = np.array([i[1] for i in cm1])
cm2_count = np.array([i[1] for i in cm2])

cm1_normalized_count = {k: v for k, v in zip(cm1_words, preprocessing.normalize([cm1_count])[0])}
cm2_normalized_count = {k: v for k, v in zip(cm2_words, preprocessing.normalize([cm2_count])[0])}

# TODO: make it so that it'll compare the raw vector values. - use cosine similarity
from scipy import spatial

def sort_dict(inp: dict) -> dict:
    return dict(sorted(inp.items(), key=lambda item: item[1], reverse=True))

def get_vec_similarity(word1, word2):
    return 1 - spatial.distance.cosine(model.get_word_vector(word1), model.get_word_vector(word2))\

def penalty(k: str, v): 
    # give a penalty to the score of students since students would naturally have higher similarity scores
    return v/2 if k in ba_character_list + ['chan', 'probably'] else v

cm1_vec = {word: get_vec_similarity(word, target_character) for word in cm1_words}
cm1_vec = {k: (v + cm1_normalized_count[k] * 10) for k, v in cm1_vec.items()}
cm1_vec = sort_dict(cm1_vec)
# display(cm1_vec)

cm2_vec = {word: get_vec_similarity(word, target_character) for word in cm2_words}
cm2_vec = {k: (v + cm2_normalized_count[k] * 10)/4 for k, v in cm2_vec.items()}
cm2_vec = sort_dict(cm2_vec)
# display(cm2_vec)

score = {k: penalty(k, v + cm2_vec.get(k, 0)) for k, v in cm1_vec.items()} # add up cm1 and cm2
score = sort_dict(score)

top_n = 10
for k in score:
    print(f'{k}: {score[k]}')
    top_n -= 1
    if top_n == 0: 
        break

from PIL import Image
import numpy as np
import matplotlib.pyplot as plt

from wordcloud import WordCloud, ImageColorGenerator

mask = np.array(Image.open(path.join(cwd, 'hina.png')))

wordcloud = WordCloud(random_state=1,
                      width = 2000, height = 2000,
                      background_color='black',
                      mask = mask,
                      max_words = 1000
                      ).generate_from_frequencies(score)

# Plot
image_colors = ImageColorGenerator(mask)
plt.figure(figsize=[10,10])
plt.imshow(wordcloud.recolor(color_func=image_colors), interpolation="bilinear") # wordcloud.recolor(color_func = lambda *args, **kwargs: "black") for single color
plt.axis("off")
plt.savefig(path.join(cwd, 'wordcloud.png'), dpi=2000)
plt.show()