spark/stream_tweet.py at master · tejadata/spark · GitHub

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
# -*- coding: utf-8 -*-
"""
Created on Tue Aug  4 19:17:14 2020

@author: vishwateja.r
"""


from tweepy import Stream
from tweepy import OAuthHandler
from tweepy.streaming import StreamListener
import json
import boto3
from datetime import date
import uuid


#consumer key, consumer secret, access token, access secret.

ckey= '********************************************'
csecret= ''********************************************''
atoken= ''********************************************''
asecret= ''********************************************''


s3 = boto3.client('s3')


class listener(StreamListener):

    def on_data(self, data):
        all_data = json.loads(data)

        tweet = all_data["text"]
        #print(all_data)
        username = all_data["user"]["screen_name"]
        print(username)

        values = {"uname":username,"id":all_data['id'],"data":tweet}
        #values=json.dumps(values)
        today = date.today()
        today = today.strftime("%Y/%m/%d")
        u_id=uuid.uuid1()
        key_name= today+'/'+ str(u_id)+'_'+username+'.json'
        s3.put_object(Body=str(values), Bucket='twitter-s3', Key=key_name)


        return True

    def on_error(self, status):
        print(status)

auth = OAuthHandler(ckey, csecret)
auth.set_access_token(atoken, asecret)

twitterStream = Stream(auth, listener())
tags = ['#emo']
twitterStream.filter(track=tags)