app.py

import glob
from pathlib import Path

import numpy as np
import pandas as pd
import regex as re
from matplotlib import pyplot as plt, use as plt_use

from textclassifier import TextClassifier as tc

plt_use('Agg')

# from functions import functions as f
# import time
TOPIC = "merged_topic"
SELECTED_COLUMN_DICT = {
    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': TOPIC, 'Sentiment': 'sentiment', 'Target': 'merged_target'}
PLOT_CHOICES_REVERSE_DICT = {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_TWEETS = 1000
LIMIT = 0.04


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 fix_choices_correct_order(choices):
    list_choices = [x for x in Columns if x in choices]
    return list_choices


def match_name_lower_case(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_date,
         to_date,
         usr_name_choices,
         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

    # Describe what save_file_bool is for: if you want to save the dataframe to a file, this is the boolean for that

    def add_pie_chart(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]))
                elif col == 'sentiment':
                    pie_charts.append(pie_chart(db[0], col, col + ": " + db[1]))
                elif col == TOPIC:
                    pie_charts.append(nested_pie_chart(db[0]))
        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
            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:
            # df = df[col_name].value_counts()[:5] # Lägg till "Others sedan"
            db = pie_chart_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

    def nested_pie_chart(df):
        """
        This method adds a nested pie chart. The pie chart shows the sentiment for each target.
        :param df: dataframe with data. The selected leaders have non-empty entries in the dataframe.
        :return: figure with the nested pie chart.
        """
        if df.empty:
            return None
        else:
            count_dict = {}
            sent_dict = {'positive': 0, 'negative': 1, 'neutral': 2, 'other': 3}
            tot_sum = len(df)
            for i in range(df.shape[0]):
                topic = df.iloc[i][TOPIC]
                sentiment = df.iloc[i]['sentiment'] if df.iloc[i]['sentiment'] in sent_dict else 'other'
                if topic not in count_dict:
                    count_dict[topic] = [0, 0, 0, 0]
                    count_dict[topic][sent_dict[sentiment]] += 1
                else:
                    count_dict[topic][sent_dict[sentiment]] += 1
            count_list = []
            other_list = np.array([0, 0, 0, 0])
            labels = []
            for topic in count_dict:
                if tot_sum > 0 and np.sum(count_dict[topic]) / tot_sum > LIMIT:
                    count_list.append(count_dict[topic])
                    labels.append(topic)
                else:
                    other_list += np.array(count_dict[topic])

            count_list.append(list(other_list))
            labels.append('Other')
            fig, ax = plt.subplots()

            size = 0.3
            vals = np.array(count_list)
            inner_colors = ['green', 'red', 'blue', 'yellow'] * len(count_dict)
            if vals.shape[0] == 0:
                pass
            else:
                ax.pie(vals.sum(axis=1), radius=1, labels=labels, pctdistance=0.9,
                       wedgeprops=dict(width=size, edgecolor='w'))
                ax.pie(vals.flatten(), radius=1 - size, colors=inner_colors,
                       wedgeprops=dict(width=size, edgecolor='w'))
                ax.set(aspect='equal', title='Topic (outer circle) and its corresponding sentiment (inner circle)')
            return fig

    # text_classifier = tc.TextClassifier(from_date=from_date, to_date=to_date,
    #                                     user_list=match_name_lower_case(usr_name_choices),
    #                                     num_tweets=NUM_TWEETS)
    # text_classifier.run_main_pipeline()
    # dataframe = text_classifier.get_dataframe()

    dataframe = pd.read_csv("{}/data/twitterdata.csv".format(tc.ROOT_PATH))
    # choose subset between from_date and to_date and username is in usr_name_choices
    df = dataframe.loc[(dataframe['date'] >= from_date) & (dataframe['date'] <= to_date) & \
                       (dataframe['username'].isin(match_name_lower_case(usr_name_choices)))].copy()
    # Sort df by date
    df.sort_values(by=['date'], inplace=True)
    # Remove entries from df where 'tweet' starts with '@'
    df = df[df['tweet'].str.startswith('@') == False]
    # change 'merged_topic' to 'Other' if it is 'ERROR_9000' or 'ERROR_496'
    df[TOPIC] = df[TOPIC].apply(lambda x: "N/A" if x == "ERROR_9000" or x == "ERROR_496" else x)
    # change 'merged_topic' to 'Government' if it is 's'
    df[TOPIC] = df[TOPIC].apply(lambda x: "The Government" if x == "s" else x)
    if save_selected:
        user_list = match_name_lower_case(usr_name_choices)
        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)]

    pie_charts = add_pie_chart(df, usr_name_choices, 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_example_df(user_df, PLOT_CHOICES_DICT[radio[i]]))
            else:
                df_list.append(None)

        return df_list + number_tweets + save_file_components_list

    return pie_charts + save_selected_checkbox + get_selected_df_list(df, save_file_bool, list(usr_name_choices),
                                                                      rb_components, df_visibility_check)


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


def get_example_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)

    example_df = pd.concat(stat)

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

    return example_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 pie_chart_input(df, column, limit):
    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 = {column: labels, "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
    test_frame = pd.DataFrame.from_dict(freq_dict)
    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_visibility_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_visibility_bool[i]))
            update_df.append(gr.DataFrame.update(visible=df_visibility_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():
            instructions = gr.Markdown(
                'How to use Politweet: \n\nFirst select the time interval you want to analyze, '
                'then the party leaders you want to focus on. After this you pick which '
                'statistics you want to see and click "Apply" to apply your choices. When '
                'you see empty boxes appearing, you can click "Run" to run the program.\n\nYou should now be able to '
                'see the visualisations of the collected statistics. Under the visualisations there are a few fields '
                'and buttons. If you want to see the full presentation of a specific statistic you can press one of '
                'the radio buttons, click "Show stats" and then once again press "Apply" and "Run" at the top. This '
                'presents a dataframe that accounts for the whole classification process. For "Topic" and "Target" '
                'there is a column called cos_sim_... which says how similar the algorithm found the classification '
                'and "synonym_..." to be. The field "Number" tells how many tweets the party leader made during the '
                'selected time period. If you do not see any tweets for a leader you might have to increase the time '
                'span.')
            with gr.Row():
                with gr.Column():
                    with gr.Row():
                        date1 = gr.Textbox(label="from_date", value='2022-01-01')
                        date2 = gr.Textbox(label="to_date", value='2022-05-31')
                    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=True)

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

# https://51285.gradio.app