<reponame>Awannaphasch2016/CDKFAUCovid19Cralwer<gh_stars>0
import boto3
import json
import time
import sys
from tweepy.streaming import StreamListener
from tweepy import OAuthHandler
from tweepy import Stream

# Variables that contains the user credentials to access Twitter API
consumer_key = 'M2dcKnRZGqBWTrPBXeefFHHjZ'
consumer_secret = '<KEY>'
access_token = '<KEY>'
access_token_secret = '<KEY>'

class TweetStreamListener(StreamListener):
    print('runing...')

    # on success
    def on_data(self, data):
        print('streaming...')
        tweet = json.loads(data)
        try:
            if 'text' in tweet.keys():
                # print (tweet['text'])
                # message = str(tweet)+',\n'
                message = json.dumps(tweet)
                message = message + ",\n"
                print(message)

                timestamp_ms = tweet['timestamp_ms']
                # print(timestamp_ms)
                # print(type(timestamp_ms))

                kinesis_input_data = bytes(message, 'utf-8')

                response = kinesis_client.put_record(
                    StreamName=stream_name,
                    Data=kinesis_input_data,
                    PartitionKey=timestamp_ms,
                )
                print(response)

                # kinesis_client.put_record(
                #     DeliveryStreamName=stream_name,
                #     Record={
                #         'Data': message
                #     }
                # )

        except (AttributeError, Exception) as e:
            print(e)

        print('--------')
        print('work fine')
        return True

    # on failure
    def on_error(self, status):
        print(status)


stream_name = 'faucovidstream_input'  # fill the name of Kinesis data stream you created
# stream_name = 'transforminputtoS3'  # fill the name of Kinesis data stream you created

if __name__ == '__main__':

    # # # create kinesis client connection
    # # kinesis_client = boto3.client('firehose',
    # #                               region_name='us-east-2',  # enter the region
    # #                               aws_access_key_id='<KEY>',
    # #                               # fill your AWS access key id
    # #                               aws_secret_access_key='<KEY>')  # fill you aws secret access key
    #
    # # create kinesis client connection
    # kinesis_client = boto3.client('kinesis',
    #                               region_name='us-east-2',  # enter the region
    #                               aws_access_key_id='<KEY>',
    #                               # fill your AWS access key id
    #                               aws_secret_access_key='<KEY>')  # fill you aws secret access key
    #
    # # create instance of the tweepy tweet stream listener
    # listener = TweetStreamListener()
    # # set twitter keys/tokens
    # auth = OAuthHandler(consumer_key, consumer_secret)
    # auth.set_access_token(access_token, access_token_secret)
    # # create instance of the tweepy stream
    # stream = Stream(auth, listener)
    # # search twitter for tags or keywords from cli parameters
    # query = sys.argv[1:]  # list of CLI arguments
    # query_fname = ' '.join(query)  # string
    # stream.filter(track=query)

    sleep_time = 3600

    while True:
        try:
            # # create kinesis client connection
            # kinesis_client = boto3.client('firehose',
            #                               region_name='us-east-2',  # enter the region
            #                               aws_access_key_id='<KEY>',
            #                               # fill your AWS access key id
            #                               aws_secret_access_key='<KEY>')  # fill you aws secret access key

            # create kinesis client connection
            kinesis_client = boto3.client('kinesis',
                                          region_name='us-east-2',  # enter the region
                                          aws_access_key_id='<KEY>',
                                          # fill your AWS access key id
                                          aws_secret_access_key='<KEY>')  # fill you aws secret access key

            # create instance of the tweepy tweet stream listener
            listener = TweetStreamListener()
            # set twitter keys/tokens
            auth = OAuthHandler(consumer_key, consumer_secret)
            auth.set_access_token(access_token, access_token_secret)
            # create instance of the tweepy stream
            stream = Stream(auth, listener)
            # search twitter for tags or keywords from cli parameters
            query = sys.argv[1:]  # list of CLI arguments
            query_fname = ' '.join(query)  # string
            stream.filter(track=query)

        except Exception as e:

            print(f'the following exception is foud:{e}')
            time.sleep(sleep_time)
            print(f'sleeping for {sleep_time} section')

import boto3
import json
import time
import sys
from tweepy.streaming import StreamListener
from tweepy import OAuthHandler
from tweepy import Stream

# Variables that contains the user credentials to access Twitter API
consumer_key = 'M2dcKnRZGqBWTrPBXeefFHHjZ'
consumer_secret = '<KEY>'
access_token = '<KEY>'
access_token_secret = '<KEY>'

class TweetStreamListener(StreamListener):
    print('runing...')

    # on success
    def on_data(self, data):
        print('streaming...')
        tweet = json.loads(data)
        try:
            if 'text' in tweet.keys():
                # print (tweet['text'])
                # message = str(tweet)+',\n'
                message = json.dumps(tweet)
                message = message + ",\n"
                print(message)

                timestamp_ms = tweet['timestamp_ms']
                # print(timestamp_ms)
                # print(type(timestamp_ms))

                kinesis_input_data = bytes(message, 'utf-8')

                response = kinesis_client.put_record(
                    StreamName=stream_name,
                    Data=kinesis_input_data,
                    PartitionKey=timestamp_ms,
                )
                print(response)

                # kinesis_client.put_record(
                #     DeliveryStreamName=stream_name,
                #     Record={
                #         'Data': message
                #     }
                # )

        except (AttributeError, Exception) as e:
            print(e)

        print('--------')
        print('work fine')
        return True

    # on failure
    def on_error(self, status):
        print(status)


stream_name = 'faucovidstream_input'  # fill the name of Kinesis data stream you created
# stream_name = 'transforminputtoS3'  # fill the name of Kinesis data stream you created

if __name__ == '__main__':

    # # # create kinesis client connection
    # # kinesis_client = boto3.client('firehose',
    # #                               region_name='us-east-2',  # enter the region
    # #                               aws_access_key_id='<KEY>',
    # #                               # fill your AWS access key id
    # #                               aws_secret_access_key='<KEY>')  # fill you aws secret access key
    #
    # # create kinesis client connection
    # kinesis_client = boto3.client('kinesis',
    #                               region_name='us-east-2',  # enter the region
    #                               aws_access_key_id='<KEY>',
    #                               # fill your AWS access key id
    #                               aws_secret_access_key='<KEY>')  # fill you aws secret access key
    #
    # # create instance of the tweepy tweet stream listener
    # listener = TweetStreamListener()
    # # set twitter keys/tokens
    # auth = OAuthHandler(consumer_key, consumer_secret)
    # auth.set_access_token(access_token, access_token_secret)
    # # create instance of the tweepy stream
    # stream = Stream(auth, listener)
    # # search twitter for tags or keywords from cli parameters
    # query = sys.argv[1:]  # list of CLI arguments
    # query_fname = ' '.join(query)  # string
    # stream.filter(track=query)

    sleep_time = 3600

    while True:
        try:
            # # create kinesis client connection
            # kinesis_client = boto3.client('firehose',
            #                               region_name='us-east-2',  # enter the region
            #                               aws_access_key_id='<KEY>',
            #                               # fill your AWS access key id
            #                               aws_secret_access_key='<KEY>')  # fill you aws secret access key

            # create kinesis client connection
            kinesis_client = boto3.client('kinesis',
                                          region_name='us-east-2',  # enter the region
                                          aws_access_key_id='<KEY>',
                                          # fill your AWS access key id
                                          aws_secret_access_key='<KEY>')  # fill you aws secret access key

            # create instance of the tweepy tweet stream listener
            listener = TweetStreamListener()
            # set twitter keys/tokens
            auth = OAuthHandler(consumer_key, consumer_secret)
            auth.set_access_token(access_token, access_token_secret)
            # create instance of the tweepy stream
            stream = Stream(auth, listener)
            # search twitter for tags or keywords from cli parameters
            query = sys.argv[1:]  # list of CLI arguments
            query_fname = ' '.join(query)  # string
            stream.filter(track=query)

        except Exception as e:

            print(f'the following exception is foud:{e}')
            time.sleep(sleep_time)
            print(f'sleeping for {sleep_time} section')
# Variables that contains the user credentials to access Twitter API # on success # print (tweet['text']) # message = str(tweet)+',\n' # print(timestamp_ms) # print(type(timestamp_ms)) # kinesis_client.put_record( #     DeliveryStreamName=stream_name, #     Record={ #         'Data': message #     } # ) # on failure # fill the name of Kinesis data stream you created # stream_name = 'transforminputtoS3'  # fill the name of Kinesis data stream you created # # # create kinesis client connection # # kinesis_client = boto3.client('firehose', # #                               region_name='us-east-2',  # enter the region # #                               aws_access_key_id='<KEY>', # #                               # fill your AWS access key id # #                               aws_secret_access_key='<KEY>')  # fill you aws secret access key # # # create kinesis client connection # kinesis_client = boto3.client('kinesis', #                               region_name='us-east-2',  # enter the region #                               aws_access_key_id='<KEY>', #                               # fill your AWS access key id #                               aws_secret_access_key='<KEY>')  # fill you aws secret access key # # # create instance of the tweepy tweet stream listener # listener = TweetStreamListener() # # set twitter keys/tokens # auth = OAuthHandler(consumer_key, consumer_secret) # auth.set_access_token(access_token, access_token_secret) # # create instance of the tweepy stream # stream = Stream(auth, listener) # # search twitter for tags or keywords from cli parameters # query = sys.argv[1:]  # list of CLI arguments # query_fname = ' '.join(query)  # string # stream.filter(track=query) # # create kinesis client connection # kinesis_client = boto3.client('firehose', #                               region_name='us-east-2',  # enter the region #                               aws_access_key_id='<KEY>', #                               # fill your AWS access key id #                               aws_secret_access_key='<KEY>')  # fill you aws secret access key # create kinesis client connection # enter the region # fill your AWS access key id # fill you aws secret access key # create instance of the tweepy tweet stream listener # set twitter keys/tokens # create instance of the tweepy stream # search twitter for tags or keywords from cli parameters # list of CLI arguments # string