Last active
March 7, 2018 13:44
-
-
Save cvkmohan/7e0af8723d2b00f618baad8013662cba to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# -*- coding: utf-8 -*- | |
import re | |
import tweepy | |
from tweepy import OAuthHandler | |
from textblob import TextBlob | |
class TwitterClient(object): | |
''' | |
Generic Twitter Class for sentiment analysis. | |
''' | |
def __init__(self): | |
# keys and tokens from the Twitter Dev Console | |
consumer_key = 'xxxxxxxxxxxxxxxxxxxxxxxxx' | |
consumer_secret = 'xxxxxxxxxxxxxxxxxxxxxxxxx' | |
access_token = 'xxxxxxxxxxxxxxxxxxxxxxxxx' | |
access_token_secret = 'xxxxxxxxxxxxxxxxxxxxxxxxx' | |
# attempt authentication | |
try: | |
# create OAuthHandler object | |
self.auth = OAuthHandler(consumer_key, consumer_secret) | |
# set access token and secret | |
self.auth.set_access_token(access_token, access_token_secret) | |
# create tweepy API object to fetch tweets | |
self.api = tweepy.API(self.auth) | |
except: | |
print("Error: Authentication Failed") | |
def trim_tweet(self, tweet): | |
return ' '.join(re.sub("(@[A-Za-z0-9]+)|([^0-9A-Za-z \t]) | (\w+:\/\/\S+)", " ", tweet).split()) | |
def get_tweet_sentiment(self, tweet): | |
# create TextBlob object of passed tweet text | |
analysis = TextBlob(self.trim_tweet(tweet)) | |
# set sentiment | |
if analysis.sentiment.polarity > 0: | |
return 'positive' | |
elif analysis.sentiment.polarity == 0: | |
return 'neutral' | |
else: | |
return 'negative' | |
def get_tweets(self, query, count = 10): | |
# empty list to store parsed tweets | |
tweets = [] | |
try: | |
# call twitter api to fetch tweets | |
fetched_tweets = self.api.search(q = query, count = count) | |
# parsing tweets one by one | |
for tweet in fetched_tweets: | |
# empty dictionary to store required params of a tweet | |
parsed_tweet = {} | |
parsed_tweet['text'] = tweet.text | |
# saving sentiment of tweet | |
parsed_tweet['sentiment'] = self.get_tweet_sentiment(tweet.text) | |
# appending parsed tweet to tweets list | |
if tweet.retweet_count > 0: | |
# if tweet has retweets, ensure that it is appended only once | |
if parsed_tweet not in tweets: | |
tweets.append(parsed_tweet) | |
else: | |
tweets.append(parsed_tweet) | |
# return parsed tweets | |
return tweets | |
except tweepy.TweepError as e: | |
# print error (if any) | |
print("Error : " + str(e)) | |
def main(): | |
# creating object of TwitterClient Class | |
api = TwitterClient() | |
# calling function to get tweets | |
tweets = api.get_tweets(query = 'Narendra Modi', count = 1000) | |
# picking positive tweets from tweets | |
ptweets = [tweet for tweet in tweets if tweet['sentiment'] == 'positive'] | |
# percentage of positive tweets | |
print("Positive tweets percentage: {} %".format(100*len(ptweets)/len(tweets))) | |
# picking negative tweets from tweets | |
ntweets = [tweet for tweet in tweets if tweet['sentiment'] == 'negative'] | |
# percentage of negative tweets | |
print("Negative tweets percentage: {} %".format(100*len(ntweets)/len(tweets))) | |
# printing first 5 positive tweets | |
print("\n\nPositive tweets:") | |
for tweet in ptweets[:5]: | |
print(tweet['text']) | |
# printing first 5 negative tweets | |
print("\n\nNegative tweets:") | |
for tweet in ntweets[:5]: | |
print(tweet['text']) | |
if __name__ == "__main__": | |
# calling main function | |
main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment