Last active
October 30, 2022 03:42
-
-
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
This file contains 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
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