-
Notifications
You must be signed in to change notification settings - Fork 0
/
TwitterClient.py
68 lines (56 loc) · 2.44 KB
/
TwitterClient.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
import re
import tweepy
from tweepy import OAuthHandler
from textblob import TextBlob
from textblob.sentiments import NaiveBayesAnalyzer
class TwitterClient(object):
def __init__(self):
try:
import os
consumer_key = os.environ.get("CONSUMER_KEY")
consumer_secret = os.environ.get("CONSUMER_SECRET")
access_token = os.environ.get("ACCESS_TOKEN")
access_secret = os.environ.get("ACCESS_SECRET")
self.auth = OAuthHandler(consumer_key, consumer_secret)
self.auth.set_access_token(access_token, access_secret)
self.api = tweepy.API(self.auth)
except:
print("Authentication error")
def clean_tweet(self, tweet):
return ' '.join(re.sub("(@[A-Za-z0-9]+)|([^0-9A-Za-z \t]) | (\w+:\ / \ / \S+)", " ", tweet).split())
def get_sentiment(self, tweet):
analysis = TextBlob(self.clean_tweet(tweet), analyzer=NaiveBayesAnalyzer())
if analysis.polarity > 0:
return "positive"
elif analysis.polarity == 0:
return "neutral"
else:
return "negative"
def get_polarity(self, tweet):
analysis = TextBlob(self.clean_tweet(tweet))
# ramp up the intensity
return analysis.polarity * 2
def get_tweets(self, query, geocode, city='', count=20):
tweets = []
try:
fetched_tweets = self.api.search(q=query, count=count, geocode=geocode,
lang="en", tweet_mode="extended", truncated=False)
for tweet in fetched_tweets:
parsed_tweet = {}
parsed_tweet['text'] = tweet.full_text
parsed_tweet['sentiment'] = self.get_sentiment(tweet.full_text)
parsed_tweet['polarity'] = self.get_polarity(tweet.full_text)
parsed_tweet['city'] = city
# get the coordinate info
coordinates = geocode.split(",")
parsed_tweet['lat'] = float(coordinates[0])
parsed_tweet['long'] = float(coordinates[1])
if tweet.retweet_count > 0:
if parsed_tweet not in tweets:
tweets.append(parsed_tweet)
else:
tweets.append(parsed_tweet)
return tweets
except tweepy.TweepError as e:
"""print error if there is one"""
print("Error: " + str(e))