put twitter search results to Amazon Kinesis on AWS

kinesis-twitter.py

# -*- coding: utf-8 -*-
from __future__ import unicode_literals

import os
import time
import json
import twitter
import boto3


class TwitterClient():
    def __init__(self, consumer_key, consumer_secret, token, token_secret, count=10, lang='ja'):
        auth = twitter.OAuth(consumer_key=consumer_key,
                             consumer_secret=consumer_secret,
                             token=token,
                             token_secret=token_secret
        )
        self._t = twitter.Twitter(auth = auth)
        self.max_id = 0
        self.count = count
        self.lang=lang
        
    def get_search(self, query):
        results = self._t.search.tweets(q=query, count=self.count, lang=self.lang, max_id=self.max_id)
        if len(results['statuses']) != 0:
            self.max_id = results['statuses'][-1]['id'] - 1
            
        return results['statuses']


def twitter2kinesis(query, max_loop=3, count=10, data_path='data', save_data=False):

    # Initialize Twitter API
    tc = TwitterClient(consumer_key='',
                       consumer_secret='',
                       token='',
                       token_secret='',
                       count=count
    )
    
    # Initialize Boto3
    client = boto3.client('firehose', region_name='us-west-2')
    
    tweet_num = 0
    for i in range(max_loop):
        results = tc.get_search(query)
        if len(results) == 0:
            break
    
        for line in results:
            # show message
            print('%d (%d): %s' % (line['id'], tweet_num, line['text'].replace('\n', '').replace('\r', '')))
    
            # write to local file
            if save_data:
                with open(os.path.join(data_path, str(line['id'])+'.json'), 'w') as f:
                    json_text = json.dumps(line, ensure_ascii=False, indent=4, separators=(',', ': '))
                    f.write(json_text)
                
            # send to kinesis
            json_record = json.dumps(line, ensure_ascii=False)
            response = client.put_record(DeliveryStreamName='stream-twitter', Record={'Data': json_record})
    
            tweet_num += 1

        time.sleep(1)

    return tweet_num


if __name__ == '__main__':
    twitter2kinesis(query='#AWSSummit', max_loop=100, count=100)