-
Notifications
You must be signed in to change notification settings - Fork 0
/
streamData.py
31 lines (28 loc) · 978 Bytes
/
streamData.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
from tweepy.streaming import StreamListener
from tweepy import OAuthHandler
from tweepy import Stream
from kafka import SimpleProducer, KafkaClient
import json
access_token = "add token"
access_token_secret = "add token"
consumer_key = "add token"
consumer_secret = "add token"
class StdOutListener(StreamListener):
def on_data(self, data):
resp=json.loads(data)
lang=resp.get("lang")
tweet=resp.get("text")
if(tweet is not None and lang=="en" and "RT" not in tweet):
producer.send_messages("breaking", tweet.encode('utf-8'))
print (tweet)
return True
def on_error(self, status):
print (status)
track=["obama","trump"]
kafka = KafkaClient("localhost:9092")
producer = SimpleProducer(kafka)
listener = StdOutListener()
auth = OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)
stream = Stream(auth, listener)
stream.filter(track=track)