Spaces:
Runtime error
Runtime error
| import tweepy | |
| import time | |
| import pandas as pd | |
| from transformers import pipeline | |
| import matplotlib.pyplot as plt | |
| import gradio as gr | |
| import os | |
| def twitter_auth(consumerkey,consumersecret): | |
| consumer_key = consumerkey | |
| consumer_secret = consumersecret | |
| auth = tweepy.AppAuthHandler(consumer_key,consumer_secret) | |
| api = tweepy.API(auth,wait_on_rate_limit= True) | |
| return api | |
| """## Helper function for handling ratelimit and pagination""" | |
| def limit_handled(cursor): | |
| """ | |
| Function takes the cursor and returns tweets | |
| """ | |
| while True: | |
| try: | |
| yield cursor.next() | |
| except tweepy.errors.TweepyException: | |
| print('reached rate limit, sleeping for > 15 mins') | |
| time.sleep(15*61) | |
| except StopIteration: | |
| break | |
| def tweets_collector(query,count): | |
| consumerkey = os.environ.get('consumerkey') | |
| consumersecret = os.environ.get('consumersecret') | |
| api = twitter_auth(consumerkey,consumersecret) | |
| query = query +' -filter:retweets' | |
| search = limit_handled(tweepy.Cursor(api.search_tweets,q = query,tweet_mode = 'extended',lang ='en',result_type ='recent').items(count)) | |
| sentiment_analysis = pipeline(model = "finiteautomata/bertweet-base-sentiment-analysis") | |
| tweets = [] | |
| for tweet in search: | |
| try: | |
| content = tweet.full_text | |
| sentiment = sentiment_analysis(content) | |
| tweets.append({'tweet' : content ,'sentiment': sentiment[0]['label']}) | |
| except: | |
| pass | |
| return tweets | |
| """## Run sentiment Analysis""" | |
| #tweets = tweets_collector(query,count) | |
| #df = pd.DataFrame(tweets) | |
| import pandas as pd | |
| pd.set_option('max_colwidth',None) | |
| pd.set_option('display.width',3000) | |
| #import matplotlib.pyplot as plt | |
| #sentiment_counts = df.groupby(['sentiment']).size() | |
| #fig = plt.figure(figsize = (6,6),dpi = 100) | |
| #ax = plt.subplot(111) | |
| #sentiment_counts.plot.pie(ax = ax,autopct = '%1.f%%',startangle = 270,fontsize = 12,label = "") | |
| def complaint_analysis(query,count): | |
| tweets = tweets_collector(query,count) | |
| df = pd.DataFrame(tweets) | |
| from wordcloud import WordCloud | |
| from wordcloud import STOPWORDS | |
| sentiment_counts = df.groupby(['sentiment']).size() | |
| fig = plt.figure(figsize = (6,6),dpi = 100) | |
| ax = plt.subplot(111) | |
| sentiment_counts.plot.pie(ax = ax,autopct = '%1.f%%',startangle = 270,fontsize = 12,label = "") | |
| plt.savefig('Overall_satisfaction.png') | |
| positive_tweets = df['tweet'][df['sentiment'] == 'POS'] | |
| stop_words = ["https","co","RT","ola_supports","ola_cabs","customer"] + list(STOPWORDS) | |
| positive_wordcloud = WordCloud(max_font_size=50,max_words = 30,background_color="white",stopwords=stop_words).generate(str(positive_tweets)) | |
| plt.figure() | |
| plt.title("Positive Tweets - Wordcloud") | |
| plt.imshow(positive_wordcloud,interpolation="bilinear") | |
| plt.axis("off") | |
| #plt.show() | |
| plt.savefig('positive_tweet.png') | |
| negative_tweets = df['tweet'][df['sentiment'] == 'NEG'] | |
| stop_words = ["https","co","RT","ola_supports","ola_cabs","customer"] + list(STOPWORDS) | |
| negative_wordcloud = WordCloud(max_font_size=50,max_words = 30,background_color="white",stopwords=stop_words).generate(str(negative_tweets)) | |
| plt.figure() | |
| plt.title("Negative Tweets - Wordcloud") | |
| plt.imshow(negative_wordcloud,interpolation="bilinear") | |
| plt.axis("off") | |
| #plt.show() | |
| plt.savefig('negative_tweet.png') | |
| return ['Overall_satisfaction.png','positive_tweet.png','negative_tweet.png'] | |
| gr.Interface(fn=complaint_analysis, | |
| inputs=[ | |
| gr.inputs.Textbox( | |
| placeholder="Tweet handle please", label="Company support Twitter Handle", lines=5), gr.Slider(100, 1000) ], | |
| outputs= [gr.outputs.Image(type="pil"),gr.outputs.Image(type="pil"),gr.outputs.Image(type="pil")], | |
| examples=[]).launch(debug= True) | |