Skip to content

Instantly share code, notes, and snippets.

@akash-mitra
Last active October 30, 2022 03:42
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save akash-mitra/9d2ac17c22c04728f027891cdb648d23 to your computer and use it in GitHub Desktop.
Save akash-mitra/9d2ac17c22c04728f027891cdb648d23 to your computer and use it in GitHub Desktop.
Historical Twitter Sentiment Analyzer. Read more here - https://aksmtr.com/articles/9-analyze-twitter-sentiment-trend-using-python-flair
import re
import tweepy
import flair
import datetime
class TwitterSentimentAnalyzer:
twitter_api = None
whitespace = re.compile(r"\s+")
web_address = re.compile(r"(?i)http(s):\/\/[a-z0-9.~_\-\/]+")
user = re.compile(r"(?i)@[a-z0-9_]+")
sentiment_model = flair.models.TextClassifier.load('en-sentiment')
def __init__(self, bearer_token):
auth = tweepy.OAuth2BearerHandler(bearer_token)
self.twitter_api = tweepy.API(auth,
wait_on_rate_limit=True,
retry_count=10,
retry_delay=30)
def __search_twitter_history(self, topic, start_date_str, end_date_str):
start = datetime.datetime.strptime(start_date_str, '%Y-%m-%d')
end = datetime.datetime.strptime(end_date_str, '%Y-%m-%d')
date_obj = start
results = []
while(date_obj <= end):
next_day = date_obj + datetime.timedelta(days=1)
query = topic + " -filter:retweets" \
+ " since:" + date_obj.strftime('%Y-%m-%d') \
+ " until:" + next_day.strftime('%Y-%m-%d')
tweets = tweepy.Cursor(self.twitter_api.search_tweets,
q=query, lang='en', count=100,
tweet_mode='extended').items()
results.append({
'day': date_obj.strftime('%Y-%m-%d'),
'tweets': tweets
})
date_obj = next_day
return results
def __clean_tweet(self, text):
text = self.whitespace.sub(' ', text)
text = self.web_address.sub('', text)
text = self.user.sub('', text)
return text
def __analyze(self, text):
sentence = flair.data.Sentence(text)
self.sentiment_model.predict(sentence)
return sentence
def __get_sentiment(self, tweets, dt):
results = []
for tweet in tweets:
if tweet.lang != 'en':
continue
cleaned_tweet = self.__clean_tweet(tweet.full_text)
s = self.__analyze(cleaned_tweet)
results.append({
"date": dt,
"tweet": tweet.full_text,
"sentiment": s.labels[0].value,
"score": s.labels[0].score * (1 if s.labels[0].value == 'POSITIVE' else -1),
"lang": tweet.lang
})
return results
def historical_sentiment(self, term, start_dt, end_dt):
tweets_across_days = self.__search_twitter_history(term, start_dt, end_dt)
results = {}
for day_tweets in tweets_across_days:
results [day_tweets['day']] = self.__get_sentiment(day_tweets['tweets'], day_tweets['day'])
return results
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment