|
| 1 | +import re |
| 2 | +import tweepy |
| 3 | +from tweepy import OAuthHandler |
| 4 | +from textblob import TextBlob |
| 5 | + |
| 6 | +class TwitterClient(object): |
| 7 | + def __init__(self): |
| 8 | + # keys and tokens from the Twitter Dev Console |
| 9 | + consumer_key = 'XXXXXXXXXXXXXXXXXXXXXXXX' |
| 10 | + consumer_secret = 'XXXXXXXXXXXXXXXXXXXXXXXXXXXX' |
| 11 | + access_token = 'XXXXXXXXXXXXXXXXXXXXXXXXXXXX' |
| 12 | + access_token_secret = 'XXXXXXXXXXXXXXXXXXXXXXXXX' |
| 13 | + |
| 14 | + # attempt authentication |
| 15 | + try: |
| 16 | + # create OAuthHandler object |
| 17 | + self.auth = OAuthHandler(consumer_key, consumer_secret) |
| 18 | + # set access token and secret |
| 19 | + self.auth.set_access_token(access_token, access_token_secret) |
| 20 | + # create tweepy API object to fetch tweets |
| 21 | + self.api = tweepy.API(self.auth) |
| 22 | + except: |
| 23 | + print("Error: Authentication Failed") |
| 24 | + |
| 25 | + def clean_tweet(self, tweet): |
| 26 | + ''' |
| 27 | + Utility function to clean tweet text by removing links, special characters |
| 28 | + using simple regex statements. |
| 29 | + ''' |
| 30 | + return ' '.join(re.sub("(@[A-Za-z0-9]+)|([^0-9A-Za-z \t]) |
| 31 | + |(\w+:\/\/\S+)", " ", tweet).split()) |
| 32 | + |
| 33 | + def get_tweet_sentiment(self, tweet): |
| 34 | + ''' |
| 35 | + Utility function to classify sentiment of passed tweet |
| 36 | + using textblob's sentiment method |
| 37 | + ''' |
| 38 | + # create TextBlob object of passed tweet text |
| 39 | + analysis = TextBlob(self.clean_tweet(tweet)) |
| 40 | + # set sentiment |
| 41 | + if analysis.sentiment.polarity > 0: |
| 42 | + return 'positive' |
| 43 | + elif analysis.sentiment.polarity == 0: |
| 44 | + return 'neutral' |
| 45 | + else: |
| 46 | + return 'negative' |
| 47 | + |
| 48 | + def get_tweets(self, query, count = 10): |
| 49 | + ''' |
| 50 | + Main function to fetch tweets and parse them. |
| 51 | + ''' |
| 52 | + # empty list to store parsed tweets |
| 53 | + tweets = [] |
| 54 | + |
| 55 | + try: |
| 56 | + # call twitter api to fetch tweets |
| 57 | + fetched_tweets = self.api.search(q = query, count = count) |
| 58 | + |
| 59 | + # parsing tweets one by one |
| 60 | + for tweet in fetched_tweets: |
| 61 | + # empty dictionary to store required params of a tweet |
| 62 | + parsed_tweet = {} |
| 63 | + |
| 64 | + # saving text of tweet |
| 65 | + parsed_tweet['text'] = tweet.text |
| 66 | + # saving sentiment of tweet |
| 67 | + parsed_tweet['sentiment'] = self.get_tweet_sentiment(tweet.text) |
| 68 | + |
| 69 | + # appending parsed tweet to tweets list |
| 70 | + if tweet.retweet_count > 0: |
| 71 | + # if tweet has retweets, ensure that it is appended only once |
| 72 | + if parsed_tweet not in tweets: |
| 73 | + tweets.append(parsed_tweet) |
| 74 | + else: |
| 75 | + tweets.append(parsed_tweet) |
| 76 | + |
| 77 | + # return parsed tweets |
| 78 | + return tweets |
| 79 | + |
| 80 | + except tweepy.TweepError as e: |
| 81 | + # print error (if any) |
| 82 | + print("Error : " + str(e)) |
| 83 | + |
| 84 | +def main(): |
| 85 | + # creating object of TwitterClient Class |
| 86 | + api = TwitterClient() |
| 87 | + # calling function to get tweets |
| 88 | + tweets = api.get_tweets(query = 'Donald Trump', count = 200) |
| 89 | + |
| 90 | + # picking positive tweets from tweets |
| 91 | + ptweets = [tweet for tweet in tweets if tweet['sentiment'] == 'positive'] |
| 92 | + # percentage of positive tweets |
| 93 | + print("Positive tweets percentage: {} %".format(100*len(ptweets)/len(tweets))) |
| 94 | + # picking negative tweets from tweets |
| 95 | + ntweets = [tweet for tweet in tweets if tweet['sentiment'] == 'negative'] |
| 96 | + # percentage of negative tweets |
| 97 | + print("Negative tweets percentage: {} %".format(100*len(ntweets)/len(tweets))) |
| 98 | + # percentage of neutral tweets |
| 99 | + print("Neutral tweets percentage: {} % \ |
| 100 | + ".format(100*(len(tweets) -(len( ntweets )+len( ptweets)))/len(tweets))) |
| 101 | + |
| 102 | + # printing first 5 positive tweets |
| 103 | + print("\n\nPositive tweets:") |
| 104 | + for tweet in ptweets[:10]: |
| 105 | + print(tweet['text']) |
| 106 | + |
| 107 | + # printing first 5 negative tweets |
| 108 | + print("\n\nNegative tweets:") |
| 109 | + for tweet in ntweets[:10]: |
| 110 | + print(tweet['text']) |
| 111 | + |
| 112 | +if __name__ == "__main__": |
| 113 | + # calling main function |
| 114 | + main() |
0 commit comments