Spaces:

User1342
/

Bubble-Check-In

Build error

File size: 22,272 Bytes

import gc
import os
import random
import sys
import time

import gradio as gr
import plotly.graph_objects as go
import tweepy
from detoxify import Detoxify
from transformers import pipeline

try:
    from news_classification.news_topic_text_classifier import news_topic_text_classifier
except:
    os.system(
        "{} -m pip install git+https://github.com/user1342/News-Article-Text-Classification.git".format(sys.executable))
    from news_classification.news_topic_text_classifier import news_topic_text_classifier
news_model = news_topic_text_classifier()

# Twitter API keys
consumer_token = os.getenv('consumer_token')
consumer_secret = os.getenv('consumer_secret')
my_access_token = os.getenv('my_access_token')
my_access_secret = os.getenv('my_access_secret')
bearer = os.getenv('bearer')

html_data = '''<!DOCTYPE html> <html> <head> <meta charset="UTF-8"> <meta name="viewport" 
content="width=device-width, initial-scale=1"> <link rel="stylesheet" 
href="https://www.w3schools.com/w3css/4/w3.css"> <link rel="stylesheet" 
href="https://fonts.googleapis.com/css?family=Poppins"> <link rel="stylesheet" 
href="https://cdnjs.cloudflare.com/ajax/libs/font-awesome/4.7.0/css/font-awesome.min.css"> <style> body,h1,h2,h3,h4,
h5 {font-family: "Poppins", sans-serif} body {font-size: 16px;} img {margin-bottom: -8px;} .mySlides {display: none;} 
</style> </head> <body class="w3-content w3-black" style="max-width:1500px;"> <!-- The App Section --> <div 
class="w3-padding-large w3-white"> <div class="w3-row-padding-large"> <div class="w3-col"> <h1 
class="w3-jumbo"><b>Bubble Check-In🐦💭</b></h1> <h1 class="w3-xxxlarge w3-text-blue"><b>Check-in-on someone's Twitter 'bubble'.</b></h1> <p><span class="w3-xlarge">Scroll down to use Bubble Check-In 1.0. ⬇ 
</span> Bubble Check-In is a tool designed to allow you to check-in-on the type of content someone on Twitter is 
being exposed to - be that yourself, a friend, loved one, etc. The goal here is to empower users to look out for 
each-other and identify early if someone is experiencing activity such as hate speech or extremism. We use a queue 
system, which means <b> you may need to wait your turn to run Bubble Check-In</b>. Bubble Check-In is simple to use simply enter the username of the Twitter account you want to check-in-on and click run!</p> 
<a href="https://www.jamesstevenson.me/cartographer-labs/"><button class="w3-button w3-light-grey w3-padding-large w3-section 
" onclick="document.getElementById('download').style.display='block'"> <i class=""></i> Find Out More! 💬 
</button></a> <a href="https://ko-fi.com/jamesstevenson"><button class="w3-button w3-light-grey w3-padding-large 
w3-section " onclick="document.getElementById('download').style.display='block'"> <i class=""></i> Support The 
Creator! ❤ </button></a> <a href="https://twitter.com/CartographerLab"><button class="w3-button w3-light-grey 
w3-padding-large w3-section " onclick="document.getElementById('download').style.display='block'"> <i class=""></i> 
Follow Us! 🐦 </button></a> </div> </div> </div> <!-- Modal --> <script> // Slideshow var slideIndex = 1; showDivs(
slideIndex); function plusDivs(n) { showDivs(slideIndex += n); } function showDivs(n) { var i; var x = 
document.getElementsByClassName("mySlides"); if (n > x.length) {slideIndex = 1} if (n < 1) {slideIndex = x.length} 
for (i = 0; i < x.length; i++) { x[i].style.display = "none"; } x[slideIndex-1].style.display = "block"; } </script> 
<br> </body> </html> '''

# Setup the gradio block and add some generic CSS
block = gr.Blocks(
    css=".container { max-width: 800px; margin: auto; } h1 { margin: 0px; padding: 5px 0; line-height: 50px; font-size: 60pt; }.close-heading {margin: 0px; padding: 0px;} .close-heading p { margin: 0px; padding: 0px;}",
    title="Bubble Check-In")


def check_connected_users(username):
    '''
    This function retrieves all of the mentions for the given user and all of the tweets from their following.
    :param username: the target user
    :return: a dict of user information relating to the following and mentions of the target user.
    '''

    client = tweepy.Client(
        bearer_token=bearer,
        consumer_key=consumer_token,
        consumer_secret=consumer_secret,
        access_token=my_access_token,
        access_token_secret=my_access_secret
    )

    user_id = client.get_user(username=username).data.data["id"]
    tweet_data_dict = {}
    user_count = 0

    # Get users that have mentioned the target user
    success = False
    users_mentions = []
    while not success:
        try:
            users_mentions = client.get_users_mentions(id=user_id, tweet_fields=["author_id"], max_results=10).data
            if users_mentions == None:
                users_mentions = []
            success = True
        except tweepy.errors.TooManyRequests as e:

            print("sleeping")
            print(e)
            time.sleep(120)
            success = False
            continue

    mention_count = 0

    for tweet in users_mentions:
        success = False
        while not success:
            try:
                mention_count = mention_count + 1
                user = client.get_user(id=tweet.author_id).data
                print("Processing user {}'s mentions. Mention {} of {}. Mention from user {}".format(username,
                                                                                                     mention_count,
                                                                                                     len(users_mentions),
                                                                                                     user))

                # Is this the first time adding a tweet from this user, if so act accordingly
                if user not in tweet_data_dict:
                    tweet_data_dict[user] = {}
                    tweet_data_dict[user]["tweets"] = []

                tweet_data_dict[user]["tweets"].append(tweet.data["text"])

                # Adds the mention type to the user data
                tweet_data_dict[user]["type"] = ["mentioned"]

                # Used for wrapping error handling
                success = True

            except tweepy.errors.TooManyRequests as e:

                print("sleeping")
                print(e)
                time.sleep(120)
                success = False
                continue

    # Loop through all users that the target user is following
    following = client.get_users_following(id=user_id, max_results=1000).data
    # Only take at a maximum the last x following
    if len(following) >= 50:
        following = following[:50]

    for user in following:
        success = False
        while not success:
            try:
                user_count = user_count + 1

                # If the user hasn't already been observed in mentions then create a new list for tweets (if not it would have been created previously)
                if user not in tweet_data_dict:
                    tweet_data_dict[user] = {}
                    tweet_data_dict[user]["tweets"] = []

                # Adds the following type to the user data
                if "type" not in tweet_data_dict[user]:
                    tweet_data_dict[user]["type"] = ["following"]
                else:
                    tweet_data_dict[user]["type"].append("following")

                tweets = client.get_users_tweets(id=user.id, max_results=5)
                tweets = tweets[0]

                if tweets is not None:
                    print("Processing user {}'s followers. {}, number {} of {}. Total user tweets {}.".format(username,
                                                                                                              user,
                                                                                                              user_count,
                                                                                                              len(following),
                                                                                                              len(tweets)))

                    for users_tweet in tweets:
                        tweet_data = str(users_tweet.text)
                        tweet_data_dict[user]["tweets"].append(tweet_data)

                success = True
            except tweepy.errors.TooManyRequests as e:

                print("sleeping")
                time.sleep(120)
                print(e)
                success = False
                continue

                # toxicity_score = Detoxify('original').predict(tweet_data)["toxicity"]
                # toxicities.append(toxicity_score)

                # tweet_data_dict[user]["average_toxicity"] = sum(toxicities) / len(toxicities)

    # do processing such as sentiment, centrality, hate speech, etc
    sentiment_pipeline = pipeline("sentiment-analysis")
    for current_username in tweet_data_dict:
        current_user_data = tweet_data_dict[current_username]
        toxicities = {}
        sentiments = {}
        types = {}
        user_tweets = current_user_data["tweets"]

        # Only consider users with posts for analysis
        if len(user_tweets) == 0:
            continue
        print("Processing metadata for {}'s tweets".format(current_username))
        for tweet in user_tweets:

            # Do hate speech average

            if 'toxicity' not in toxicities:
                toxicities['toxicity'] = []
                toxicities['severe_toxicity'] = []
                toxicities['obscene'] = []
                toxicities['identity_attack'] = []
                toxicities['insult'] = []
                toxicities['threat'] = []
                toxicities['sexual_explicit'] = []

            scores = Detoxify('unbiased').predict([tweet])
            toxicities['toxicity'].append(scores['toxicity'][0])
            toxicities['severe_toxicity'].append(scores['severe_toxicity'][0])
            toxicities['obscene'].append(scores['obscene'][0])
            toxicities['identity_attack'].append(scores['identity_attack'][0])
            toxicities['insult'].append(scores['insult'][0])
            toxicities['threat'].append(scores['threat'][0])
            toxicities['sexual_explicit'].append(scores['sexual_explicit'][0])

            # Do sentiment analysis
            sentiment_score = sentiment_pipeline(tweet)
            sentiment_score = sentiment_score[0]
            if "NEGATIVE" == sentiment_score["label"]:
                if "NEGATIVE" not in sentiments:
                    sentiments["NEGATIVE"] = []
                sentiments["NEGATIVE"].append(sentiment_score["score"])

            elif "POSITIVE" == sentiment_score["label"]:
                if "POSITIVE" not in sentiments:
                    sentiments["POSITIVE"] = []
                sentiments["POSITIVE"].append(sentiment_score["score"])

            # Do type of post (news)
            type = news_model.get_category(tweet)
            if type in types:
                types[type] = types[type] + 1
            else:
                types[type] = 1

        tweet_data_dict[current_username]["average_toxicity"] = sum(toxicities['toxicity']) / len(
            toxicities['toxicity'])
        tweet_data_dict[current_username]["average_severe_toxicity"] = sum(toxicities['severe_toxicity']) / len(
            toxicities['severe_toxicity'])
        tweet_data_dict[current_username]["average_obscene"] = sum(toxicities['obscene']) / len(toxicities['obscene'])
        tweet_data_dict[current_username]["average_identity_attack"] = sum(toxicities['identity_attack']) / len(
            toxicities['identity_attack'])
        tweet_data_dict[current_username]["average_insult"] = sum(toxicities['insult']) / len(toxicities['insult'])
        tweet_data_dict[current_username]["average_threat"] = sum(toxicities['threat']) / len(toxicities['threat'])
        tweet_data_dict[current_username]["average_sexual_explicit"] = sum(toxicities['sexual_explicit']) / len(
            toxicities['sexual_explicit'])
        tweet_data_dict[current_username]["types"] = types
        tweet_data_dict[current_username]["sentiments"] = sentiments

        gc.collect()

    return tweet_data_dict


def button_pressed(text_box):
    '''
    A function that is called when the 'run' button is pressed
    :param text_box: a string which should relate to a Twitter users username
    :return: several gradio elements used to populate plots and a summary label field
    '''

    tweet_data = check_connected_users(text_box)

    total_types_count = {}
    total_average_toxicity = []
    total_average_severe_toxicity = []
    total_average_obscene = []
    total_average_identity_attack = []
    total_identity_attack = []
    total_average_insult = []
    total_average_threat = []
    total_average_sexual_explicit = []
    total_average_pos_sentiment = []
    total_average_neg_sentiment = []

    mentions = 0
    following = 0

    tweets = 0

    user_data = {}
    for user in tweet_data:
        data = tweet_data[user]

        tweets = tweets + len(data["tweets"])

        if len(data["tweets"]) < 1:
            continue

        if "mentioned" in data["type"]:
            mentions = mentions + 1
        if "following" in data["type"]:
            following = following + 1

        types = data["types"]

        # Get types
        for type in types:
            if type not in total_types_count:
                total_types_count[type] = 1
            else:
                total_types_count[type] = total_types_count[type] + 1

            total_average_toxicity.append(data["average_toxicity"])
            user_data[user.name] = data["average_toxicity"]
            total_average_severe_toxicity.append(data["average_severe_toxicity"])
            total_average_obscene.append(data["average_obscene"])
            total_average_identity_attack.append(data["average_identity_attack"])
            total_average_insult.append(data["average_insult"])
            total_average_threat.append(data["average_threat"])
            total_average_sexual_explicit.append(data["average_sexual_explicit"])

            if 'NEGATIVE' in data["sentiments"]:
                for sentiment in data["sentiments"]["NEGATIVE"]:
                    total_average_neg_sentiment.append(sentiment)

            if 'POSITIVE' in data["sentiments"]:
                for sentiment in data["sentiments"]["POSITIVE"]:
                    total_average_pos_sentiment.append(sentiment)

    # Comprise elements for hate speech plot
    total_average_toxicity = sum(total_average_toxicity) / len(total_average_toxicity)
    total_average_severe_toxicity = sum(total_average_severe_toxicity) / len(total_average_severe_toxicity)
    total_average_obscene = sum(total_average_obscene) / len(total_average_obscene)
    total_average_identity_attack = sum(total_average_identity_attack) / len(total_average_identity_attack)
    total_average_insult = sum(total_average_insult) / len(total_average_insult)
    total_average_threat = sum(total_average_threat) / len(total_average_threat)
    total_average_sexual_explicit = sum(total_average_sexual_explicit) / len(total_average_sexual_explicit)

    total_average_neg_sentiment = sum(total_average_neg_sentiment) / len(total_average_neg_sentiment)
    total_average_pos_sentiment = sum(total_average_pos_sentiment) / len(total_average_pos_sentiment)

    toxicity_plot = dict({
        "data": [{"type": "bar",
                  "x": ["Average Toxicity", "Average Severe Toxicity", "Average Obscene", "Average Identity Attack",
                        "Average Insult", "Average Threat", "Average Sexual Explicit"],
                  "y": [total_average_toxicity, total_average_severe_toxicity, total_average_obscene,
                        total_average_identity_attack, total_average_insult, total_average_threat,
                        total_average_sexual_explicit]}],
        "layout": {"title": {"text": "Hate Speech"}}
    })

    toxicity_plot_fig = go.Figure(toxicity_plot)

    # Comprise elements for sentiment plot
    sentiment_plot = dict({
        "data": [{"type": "bar",
                  "x": ["Positive Sentiment Average", "Negative Sentiment Average"],
                  "y": [total_average_pos_sentiment, total_average_neg_sentiment]}],
        "layout": {"title": {"text": "Sentiment"}}
    })

    sentiment_plot_fig = go.Figure(sentiment_plot)

    # User distrabution plot
    user_plot = dict({
        "data": [{"type": "bar",
                  "x": list(user_data.keys()),
                  "y": list(user_data.values())}],
        "layout": {"title": {"text": "Hate Speech By Observed User"}}
    })

    user_plot_fig = go.Figure(user_plot)

    # Distrabution Pie
    labels = ['Timeline', 'Mentions'.format(text_box)]
    values = [following,mentions]

    distrabution_fig = go.Figure(data=[go.Pie(labels=labels, values=values, title="Distribution Of Observed Users")])

    # Comprise elements for 'type' plot
    colours = []
    keys = list(total_types_count.keys())
    x_list = []
    for key in keys:
        x_list.append(key.replace("_", " ").title())

    for iterator in range(0, len(keys)):
        colours.append('rgb({}, {}, {})'.format(random.randint(1, 255), random.randint(1, 255), random.randint(1, 255)))

    sizes = []
    for value in total_types_count.values():
        sizes.append(value * 20)

    fig = go.Figure(data=[go.Scatter(
        x=x_list, y=list(total_types_count.values()),
        mode='markers',
        marker=dict(
            color=colours,
            size=sizes
        )
    )])

    # Comprise text for summary label
    original_text = "A total number of {} recent tweets in @{}'s mentions and timeline were reviewed, of which @{} was exposed to {} users via mentions and " \
                    "{} directly via following them.".format(tweets, text_box, text_box, mentions, following)

    text = original_text

    high_identifiers = []
    extreme_identifiers = []

    if total_average_toxicity > 75:
        extreme_identifiers.append("toxic")
    elif total_average_toxicity > 50:
        high_identifiers.append("toxic")

    if total_average_severe_toxicity > 75:
        extreme_identifiers.append("severe toxic")
    elif total_average_severe_toxicity > 50:
        high_identifiers.append("severe toxic")

    if total_average_obscene > 75:
        extreme_identifiers.append("obscene")
    elif total_average_obscene > 50:
        high_identifiers.append("obscene")

    if total_average_identity_attack > 75:
        extreme_identifiers.append("identity based hate")
    elif total_average_identity_attack > 50:
        high_identifiers.append("identity based hate")

    if total_average_insult > 75:
        extreme_identifiers.append("insulting")
    elif total_average_insult > 50:
        high_identifiers.append("insulting")

    if total_average_threat > 75:
        extreme_identifiers.append("threatening")
    elif total_average_threat > 50:
        high_identifiers.append("threatening")

    if total_average_sexual_explicit > 75:
        extreme_identifiers.append("sexually explicit")
    elif total_average_sexual_explicit > 50:
        high_identifiers.append("sexually explicit")

    if len(high_identifiers) > 0:
        text = text + " @{} is observing a high amount of "
        for identifier in high_identifiers:
            text = text + " {},".format(identifier)

        text = text[:len(text - 1)] + " language."

    if len(extreme_identifiers) > 0:
        text = text + " @{} is observing an extremely high amount of".format(text_box)
        for identifier in extreme_identifiers:
            text = text + " {},".format(identifier)

        text = text[:len(text - 1)] + " language."

    if total_average_neg_sentiment > 0.7 and total_average_neg_sentiment > total_average_pos_sentiment:
        text = text + " @{} is experiencing a high amount of negative sentiment content.".format(text_box)
    elif total_average_neg_sentiment > 0.9 and total_average_neg_sentiment > total_average_pos_sentiment:
        text = text + " '{} is experiencing a significantly high amount of negative sentiment content.".format(text_box)

    if len(text) == len(original_text):
        text = text + " No excessive hate speech or low sentiment was observed in @{}'s mentions or timeline.".format(
            text_box)

    return [toxicity_plot_fig, sentiment_plot_fig, fig, text,user_plot_fig,distrabution_fig]


# The main chunk of code that uses Gradio blocks to create the UI
html_button = None
with block:
    gr.HTML('''
<meta name="viewport" content="width=device-width, initial-scale=1">
<link rel="stylesheet" href="https://www.w3schools.com/w3css/4/w3.css">
''')

    # todo check if user signed in
    gr.HTML(value=html_data)
    with gr.Group():
        with gr.Row().style(equal_height=True):
            with gr.Box():
                with gr.Row().style(equal_height=True):
                    text_input = gr.Text(label="Username", visible=True, max_lines=1)
                    btn = gr.Button("Run Bubble Check-In").style(full_width=True).style()
                gr.HTML(value="<br>")
                output_label = gr.Label(label="Summary")
                gr.HTML(value="<br>")
                with gr.Row().style(equal_height=True):
                    toxicity_plot = gr.Plot(label="Hate Speech Graph")
                    sentiment_plot = gr.Plot(label="Sentiment Graph")
                gr.HTML(value="<br>")
                type_plot = gr.Plot(label="Content Type Graph")
                gr.HTML(value="<br>")
                with gr.Row().style(equal_height=True):
                    user_plot = gr.Plot(label="Observed Users")
                    format_type_plot = gr.Plot(label="Distribution")
    btn.click(fn=button_pressed, inputs=[text_input], outputs=[toxicity_plot, sentiment_plot, type_plot, output_label,user_plot,format_type_plot])
    gr.Markdown(
        """___
   <p style='text-align: center'>
   Created by <a href="https://twitter.com/_JamesStevenson" target="_blank"</a> James Stevenson
   <br/>
   </p>"""
    )

# block.attach_load_events()

# Launcg the page
block.launch(enable_queue=True,show_api=False)