app.py

import numpy as np

from textclassifier import TextClassifier as tc
import pandas as pd
import regex as re

from pathlib import Path
import glob
from math import sqrt
import os
import matplotlib

matplotlib.use('Agg')
import matplotlib.pyplot as plt
from functions import functions as f
import time

SELECTED_COLUMN_DICT = {
    'merged_topic': ['tweet', 'main_topic', 'sub_topic', 'synonym_topic', 'cos_sim_topic', 'merged_topic'],
    'sentiment': ['tweet', 'sentiment'],
    'merged_target': ['tweet', 'target', 'synonym_target', 'cos_sim_target', 'merged_target']
    }

USER_LIST = ['jimmieakesson', 'BuschEbba', 'annieloof', 'JohanPehrson', 'bolund', 'martastenevi', 'SwedishPM',
             'dadgostarnooshi']

USER_NAMES = ['Jimmie Åkesson', 'Ebba Busch', 'Annie Lööf', 'Johan Pehrson', 'Per Bolund', 'Märta Stenevi',
              'Magdalena Andersson', 'Nooshi Dadgostar']

CHOICE_LIST = ['Topic', 'Sentiment', 'Target']

# PLOT_CHOICES_DICT = {'Topic': 'sub_topic', 'Sentiment': 'sentiment', 'Target': 'target'}  I just changed its pavue to merged target and merged topic
PLOT_CHOICES_DICT = {'Topic': 'merged_topic', 'Sentiment': 'sentiment', 'Target': 'merged_target'}
PLOT_CHOICES_REVERSE_DICT = {'merged_topic': 'Topic', 'sentiment': 'Sentiment', 'merged_target': 'Target'}
# PLOT_CHOICES_REVERSE_DICT= {'sub_topic':'Topic', 'sentiment':'Sentiment' , 'target':'Target'}
UserNameDict = dict(zip(['Jimmie Åkesson', 'Ebba Busch', 'Annie Lööf', 'Johan Pehrson', 'Per Bolund',
                         'Märta Stenevi', 'Magdalena Andersson', 'Nooshi Dadgostar'], USER_LIST))

Columns = ['username', 'nlikes', 'nreplies', 'nretweets', 'main_topic', 'sub_topic', 'sentiment', 'target', 'tweet',
           'date', 'urls', 'id', 'class_tuple', 'user_id']
num_tweet = 1000
LIMIT = 0.05


def show_all_stats(see_full_stats):
    dataframe = pd.read_csv("{}/data/twitterdata.csv".format(tc.ROOT_PATH))
    if see_full_stats:
        return dataframe
    else:
        return pd.DataFrame()


def fixChoicesCorrectOrder(choices):
    ListChoices = [x for x in Columns if x in choices]
    return ListChoices


def MatchNameToUser(user_names):
    users = []
    for N in user_names:
        users.append(UserNameDict[N])
    return users


def convert_plot_choices(plot_choices):
    return [PLOT_CHOICES_DICT[choice] for choice in plot_choices]


def convert_back_plot_choices(plot_choices_raw):
    return [PLOT_CHOICES_REVERSE_DICT[choice] for choice in plot_choices_raw]


def main(From,
         To,
         UserNameChoices,
         plot_choice,
         save_selected,
         rb1, rb2, rb3, rb4, rb5, rb6, rb7, rb8,
         v1, v2, v3, v4, v5, v6, v7, v8,
         s1, s2, s3, s4, s5, s6, s7, s8

         ):
    save_file_bool = s1, s2, s3, s4, s5, s6, s7, s8

    def Add_Pychart(df, leaders, plot_choices):
        df_list = []
        pie_charts = []
        return_list = []
        leader_bool_list, plot_bool_list = convert_to_boolean(leaders, convert_back_plot_choices(plot_choices))
        bool_list = []
        for leader in leader_bool_list:
            if leader:
                for choice in plot_bool_list:
                    bool_list.append(choice)
            else:
                for i in range(len(plot_bool_list)):
                    bool_list.append(False)
        for user in USER_NAMES:  # leaders:
            df_list.append((df.loc[df["username"] == UserNameDict[user]], user))

        for db in df_list:
            for col in PLOT_CHOICES_REVERSE_DICT:  # plot_choices:
                if col == 'merged_target':
                    pie_charts.append(bar(db[0], col + ": " + db[1]))
                else:
                    pie_charts.append(pie_chart(db[0], col, col + ": " + db[1]))
        return pie_charts

    def bar(db: pd.DataFrame, title):
        '''This method adds a stacked bar diagram for each target and each sentiment
        NOTE:  The tweets without any target are not shown in the plot, we just show distribution of tweets that have a
        target.
        '''
        if db.empty:
            return None
        else:
            db['merged_target'] = db["merged_target"].apply(lambda
                                                                x: "other" if x == "ERROR_9000" or x == "ERROR_496" else x)  # replacing Different Error type with string "other"
            db['sentiment'] = db['sentiment'].apply(
                lambda x: re.sub('\s+', "", x))  # removing extra spaces in  at the end and beginning of the sentiments.
            # This can be removed after we remove all unnessary spaces from twitter data
            all_targets = ['v', 'mp', 's', 'c', 'l', 'kd', 'm', 'sd', 'Red-Greens', 'The opposition']
            db_new = db.loc[db["merged_target"] != "other"]  # dataframe with other category removed
            percent_target = (len(db_new) / len(db)) * 100
            targets = db_new["merged_target"].value_counts().keys().to_list()
            positive = [0] * len(all_targets)
            negative = [0] * len(all_targets)
            neutral = [0] * len(all_targets)
            other = [0] * len(all_targets)
            for i, target in enumerate(all_targets):
                temp_db = db_new.loc[db_new["merged_target"] == target]
                if temp_db.empty:
                    pass
                else:
                    sent = temp_db['sentiment'].to_list()
                    positive[i] += sent.count('positive')
                    negative[i] += sent.count('negative')
                    neutral[i] += sent.count('neutral')
                    other[i] += sent.count('other')
            font1 = {'family': 'serif', 'color': 'blue', 'size': 10}
            fig = plt.figure()
            y1 = np.array(positive) / len(db_new) if len(db_new) > 0 else np.array(positive)
            y2 = np.array(negative) / len(db_new) if len(db_new) > 0 else np.array(negative)
            y3 = np.array(neutral) / len(db_new) if len(db_new) > 0 else np.array(neutral)
            y4 = np.array(other) / len(db_new) if len(db_new) > 0 else np.array(other)
            plt.bar(all_targets, y1, color='g')
            plt.bar(all_targets, y2, bottom=y1, color='r')
            plt.bar(all_targets, y3, bottom=(y1 + y2), color='yellow')
            plt.bar(all_targets, y4, bottom=(y1 + y2 + y3), color='b')
            plt.xticks(rotation=15)
            plt.ylim(0, 1)
            plt.title(
                str(percent_target)[0:4] + "% " + " of tweets have  target. " + "Number of tweets with target:" + str(
                    len(db_new)), loc='right', fontdict=font1)
            # plt.xlabel("Targets")
            plt.ylabel("Procent")
            plt.legend(["positive", "negative", "neutral", "other"])
            return fig

    def pie_chart(db, col_name, title):
        if db.empty:
            return None
        else:
            # db = db[col_name].value_counts()[:5] # Lägg till "Others sedan"
            db = piechart_input(db, col_name, LIMIT)
            labels = db[col_name].to_list()
            sizes = db['frequency'].values
            # explode = (0, 0.1, 0, 0, 0)  # only "explode" the 2nd slice (i.e. 'Hogs')
            font1 = {'family': 'serif', 'color': 'blue', 'size': 20}
            fig = plt.figure()
            plt.pie(sizes, labels=labels, radius=1, autopct='%1.1f%%')
            plt.title(title, fontdict=font1)
            return fig

    text_classifier = tc.TextClassifier(from_date=From, to_date=To, user_list=MatchNameToUser(UserNameChoices),
                                        num_tweets=num_tweet)
    text_classifier.run_main_pipeline()
    dataframe = text_classifier.get_dataframe()
    # dataframe= pd.read_csv(os.path.dirname(
    #      os.path.dirname(os.path.abspath(__file__))) + "/politweet/data/twitterdata.csv")  #
    df = dataframe

    if save_selected:
        user_list = MatchNameToUser(UserNameChoices)
        df_l = []
        for user in user_list:
            df_l.append(pd.DataFrame(df.loc[df['username'] == user]))

        selected_df = pd.concat(df_l).reset_index(drop=True)
        export_to_download(selected_df, "selected_leaders")
        save_selected_checkbox = [gr.Checkbox.update(interactive=False)]

    else:
        save_selected_checkbox = [gr.Checkbox.update(interactive=True)]

    pycharts = Add_Pychart(df, UserNameChoices, convert_plot_choices(plot_choice))

    rb_components = [rb1, rb2, rb3, rb4, rb5, rb6, rb7, rb8]  # radio_buttons
    df_visibility_check = [v1, v2, v3, v4, v5, v6, v7, v8]

    def get_selected_df_list(d_frame, save_or_no, selected_users, radio, visibility):

        leader_bool_list = [True if leader in selected_users else False for leader in USER_NAMES]
        df_list = []
        number_tweets = []
        save_file_components_list = []
        for i, u_bool in enumerate(leader_bool_list):
            user_df = d_frame.loc[d_frame['username'] == USER_LIST[i]]
            number_tweets.append(gr.Number.update(value=len(user_df), visible=u_bool))

            if save_or_no[i]:
                export_to_download(pd.DataFrame(user_df), "one_leader")
                save_file_components_list.append(gr.Checkbox.update(visible=u_bool, interactive=False))
            else:
                save_file_components_list.append(gr.Checkbox.update(visible=u_bool))

            if u_bool and visibility[i]:
                df_list.append(get_exemple_df(user_df, PLOT_CHOICES_DICT[radio[i]]))
            else:
                df_list.append(None)

        return df_list + number_tweets + save_file_components_list

    return pycharts + save_selected_checkbox + get_selected_df_list(df, save_file_bool, list(UserNameChoices),
                                                                    rb_components, df_visibility_check)


''' END OF MAIN 
####
#####
####
####
'''


def get_exemple_df(df: pd.DataFrame, column: str):
    print(column)
    df = df[SELECTED_COLUMN_DICT[column]]
    unique_labels = df[column].value_counts().keys()
    stat = []
    for label in unique_labels:
        df_temp = df.loc[df[column] == label]
        if len(df_temp) > 5:
            df_temp = df_temp[0:5]
        stat.append(df_temp)

    exemple_df = pd.concat(stat)

    # stat =stat.reset_index(drop=True)     just in case u want to reset indexing

    return exemple_df


def export_to_download(_data_frame, _type: str):
    downloads_path = str(Path.home()) + "/Downloads/"
    if _type == "one_leader":
        file_name = _data_frame['username'].to_list()[0]  # df['username'][0] + "_data"
    else:
        file_name = "selected_leaders"
    full_path = downloads_path + file_name + ".csv"

    while full_path in glob.glob(downloads_path + "*"):
        search_list = re.findall('\p{N}+', full_path)
        if search_list:
            index = search_list[0]
            full_path = re.sub(index, str(int(index) + 1), full_path)
        else:
            suffix = " (1).csv"
            full_path = re.sub('\.csv', suffix, full_path)

    _data_frame.to_csv(full_path, index=False)

    # , pie_chart(df, "main_topic"), pie_chart("target")


def piechart_input(df, column, limit):
    df_len = len(df)
    df_v = df[column].value_counts()
    df_len = len(df)
    if column == "sentiment":
        ds_sentiment = df[column].apply(lambda x: re.sub("\s+", "", str(x)))
        df_v = ds_sentiment.apply(lambda x: x if str(x).lower() == "positive" or str(x).lower() == "negative" or str(
            x).lower() == "neutral" else "other").value_counts()
    elif column == "merged_target":
        ds_target = df[column].apply(lambda x: "other" if x == "ERROR_9000" or x == "ERROR_496" else x)
        df_v = ds_target.value_counts()
        freq = df_v.to_list()
        labels = df_v.keys().to_list
        freq_dict = {}
        freq_dict[column] = labels
        freq_dict["frequency"] = freq
        return pd.DataFrame.from_dict(freq_dict)

    else:
        df_v = df[column].value_counts()
    freq = df_v.to_list()
    labels = df_v.keys().to_list()
    freq_other = 0
    freq_dict = {column: [], "frequency": []}
    for i in range(len(df_v)):
        if freq[i] / df_len < limit:
            freq_other += freq[i]
        else:
            freq_dict[column].append(labels[i])
            freq_dict["frequency"].append(freq[i])

    if "other" not in freq_dict[column]:
        freq_dict[column].append("other")
        freq_dict["frequency"].append(freq_other)
    else:
        ind_other = freq_dict[column].index("other")
        freq_dict["frequency"][ind_other] += freq_other

    return pd.DataFrame.from_dict(freq_dict)


def convert_to_boolean(leaders, plot_choices):
    leaders_converted = [True if leader in leaders else False for leader in USER_NAMES]
    plot_converted = [True if choice in plot_choices else False for choice in CHOICE_LIST]

    return leaders_converted, plot_converted


def update_window(leaders: list, plot_choices: list,
                  v1, v2, v3, v4, v5, v6, v7, v8
                  ):
    leader_bool_list, plot_bool_list = convert_to_boolean(leaders, plot_choices)

    bool_list = []
    df_visiblity_bool = [v1, v2, v3, v4, v5, v6, v7, v8]

    # this loop sets boolean for plots
    for leader in leader_bool_list:
        if leader:
            for choice in plot_bool_list:
                bool_list.append(choice)
        # bool_list.append(True)  ## this is for radio component
        else:
            for i in range(len(plot_bool_list)):
                bool_list.append(False)
            # bool_list.append(False)

    update_blocks = []
    update_plots = []
    update_radio = []
    update_nr_tweet = []
    update_checkbox = []
    update_save_file_checkboxes = []
    update_df = []

    # all_visual = block_list + plots + radio_list + nr_tweet_list + checkbox_list + saving_file_checkboxes + df_list

    for i, vis_or_not in enumerate(leader_bool_list):
        update_blocks.append(gr.Row.update(visible=vis_or_not))
        update_blocks.append(gr.Row.update(visible=vis_or_not))
        if vis_or_not:
            update_blocks.append(gr.Row.update(visible=df_visiblity_bool[i]))
            update_df.append(gr.DataFrame.update(visible=df_visiblity_bool[i]))
        else:

            update_blocks.append(gr.Row.update(visible=False))
            update_df.append(gr.DataFrame.update(visible=False))

        update_nr_tweet.append(gr.Number.update(visible=vis_or_not))
        update_radio.append(gr.Radio.update(visible=vis_or_not))
        update_checkbox.append(gr.Checkbox.update(visible=vis_or_not))
        update_save_file_checkboxes.append(gr.Checkbox.update(visible=vis_or_not))
    for choice in bool_list:
        update_plots.append(gr.Plot.update(visible=choice))

    return update_blocks + update_plots + update_radio + update_nr_tweet + update_checkbox + update_save_file_checkboxes + update_df


def add_plots(user):
    plot_list = []
    for plot_type in PLOT_CHOICES_DICT:
        plot_list.append(gr.Plot(label=plot_type + " for " + user, visible=False))
    return plot_list


def add_nbr_boxes():
    return [gr.Number(value=0, label='Tweets by ' + user, visible=False) for user in USER_NAMES]


if __name__ == "__main__":
    import gradio as gr

    demo = gr.Blocks(title='Politweet')
    with demo:
        with gr.Column():
            with gr.Row():
                with gr.Column():
                    with gr.Row():
                        date1 = gr.Textbox(label="From", value='2022-05-10')
                        date2 = gr.Textbox(label="To", value='2022-05-30')
                    leaders = gr.Checkboxgroup(choices=USER_NAMES,
                                               label="")
                    plot_choices = gr.CheckboxGroup(choices=CHOICE_LIST, label='Choose what to show')

                    save_selected_data_checkbox = gr.Checkbox(label="Export selected data")
                    with gr.Row():
                        update = gr.Button('Apply')
                        btn = gr.Button("Run")

                        # show_stat = gr.Checkbox(label="Show full statistics", value=True)
                        # show_plots = gr.components.Checkbox(label='Show topics', value=True)
                with gr.Column():
                    selected = gr.DataFrame(label="Summary statistics for the selected choices",
                                            max_rows=None, visible=False)
                    # all_data = gr.components.DataFrame(label="Summary statistics of the total database",
                    # max_rows=None)

            plots = []
            radio_list = []
            checkbox_list = []
            df_list = []
            block_list = []
            saving_file_checkboxes = []
            nr_tweet_list = []
            with gr.Column():
                for i in range(len(USER_NAMES)):
                    block_list += [gr.Row()] * 3
                for i, leader in enumerate(USER_NAMES):
                    with gr.Row():
                        plots += add_plots(leader)
                    with gr.Row():
                        radio_list.append(gr.Radio(list(PLOT_CHOICES_DICT.keys()), visible=False, interactive=True))
                        nr_tweet_list.append(gr.Number(visible=False))
                        checkbox_list.append(gr.Checkbox(label="Show stats ", value=False, visible=False))
                        saving_file_checkboxes.append(gr.Checkbox(label="Export file", value=False, visible=False))

                    with gr.Row():
                        df_list.append(gr.DataFrame(visible=False))

        inp = [date1,
               date2,
               leaders,
               plot_choices, save_selected_data_checkbox] + radio_list + checkbox_list + saving_file_checkboxes

        output = plots + [save_selected_data_checkbox] + df_list + nr_tweet_list + saving_file_checkboxes

        all_visual = block_list + plots + radio_list + nr_tweet_list + checkbox_list + saving_file_checkboxes + df_list  # + df_list # df_comps

        update_inp = [leaders, plot_choices] + checkbox_list

        update.click(fn=update_window, inputs=update_inp, outputs=all_visual)

        btn.click(fn=main, inputs=inp, outputs=output)
        # input.change(fn=main, inputs=input, outputs=output)
    demo.launch(share=False)

# df= pd.read_csv(os.getcwd()+"/data/twitterdata.csv")

# https://51285.gradio.app