Tweepy streamer

In [1]:

    

import logging # python logging module

# basic format for logging
logFormat = "%(asctime)s - [%(levelname)s] (%(funcName)s:%(lineno)d) %(message)s"

# logs will be stored in tweepy.log
logging.basicConfig(filename='tweepylang.log', level=logging.INFO, 
                   format=logFormat, datefmt="%Y-%m-%d %H:%M:%S")


    

In [2]:

    

import tweepy  # importing all the modules required
import socket  # will be used to create sockets
import json    # manipulate json

from httplib import IncompleteRead


    

In [3]:

    

# Keep these tokens secret, as anyone can have full access to your
# twitter account, using these tokens

consumerKey = "#"
consumerSecret = "#"

accessToken = "#-#"
accessTokenSecret = "#"


    

In [4]:

    

# Performing the authentication and authorization, post this step 
# we will have full access to twitter api's
def connectToTwitter():
    """Connect to twitter."""
    try:
        auth = tweepy.OAuthHandler(consumerKey, consumerSecret)
        auth.set_access_token(accessToken, accessTokenSecret)

        api = tweepy.API(auth)
        logging.info("Successfully logged in to twitter.")
        return api, auth
    except Exception as e:
        logging.info("Something went wrong in oauth, please check your tokens.")
        logging.error(e)


    

In [5]:

    

# Tweet listner class which subclasses from tweepy.StreamListener
class TweetListner(tweepy.StreamListener):
    """Twitter stream listner"""
    
    def __init__(self, csocket):
        self.clientSocket = csocket
        
    def dataProcessing(self, data):
        """Process the data, before sending to spark streaming
        """
        sendData = {}  # data that is sent to spark streamer
        user = data.get("user", {})
        name = user.get("name", "undefined").encode('utf-8')
        lang = user.get("lang", "undefined").encode('utf-8')
        sendData["name"] = name
        sendData["lang"] = lang
        #data_string = "{}:{}".format(name, followersCount) 
        self.clientSocket.send(json.dumps(sendData) + u"\n")  # append new line character, so that spark recognizes it
        logging.debug(json.dumps(sendData))
    
    def on_data(self, raw_data):
        """ Called when raw data is received from connection.
            return False to stop stream and close connection.
        """
        try:
            data = json.loads(raw_data)
            self.dataProcessing(data)
            #self.clientSocket.send(json.dumps(sendData) + u"\n") # Because the connection was breaking
            return True
        except Exception as e:
            logging.error("An unhandled exception has occured, check your data processing")
            logging.error(e)
            raise e
    
    def on_error(self, status_code):
        """Called when a non-200 status code is returned"""
        logging.error("A non-200 status code is returned: {}".format(status_code))
        return True


    

In [6]:

    

# Creating a proxy socket
def createProxySocket(host, port):
    """ Returns a socket which can be used to connect
        to spark.
    """
    try:
        s = socket.socket()   # initialize socket instance
        s.bind((host, port))  # bind to the given host and port 
        s.listen(5)           # Enable a server to accept connections.
        logging.info("Listening on the port {}".format(port))
        cSocket, address = s.accept()  # waiting for a connection
        logging.info("Received Request from: {}".format(address))
        return cSocket
    except socket.error as e:  
        if e.errno == socket.errno.EADDRINUSE:  # Address in use
            logging.error("The given host:port {}:{} is already in use"\
                         .format(host, port))
            logging.info("Trying on port: {}".format(port + 1))
            return createProxySocket(host, port + 1)


    

In [7]:

    

if __name__ == "__main__":
    try:
        api, auth = connectToTwitter()  # connecting to twitter
        # Global information is available by using 1 as the WOEID
        # woeid = getWOEIDForTrendsAvailable(api, "Worldwide")  # get the woeid of the worldwide
        
        host = "localhost"
        port = 8500
        cSocket = createProxySocket(host, port)  # Creating a socket
        
        while True:
            try:
                # Connect/reconnect the stream
                tweetStream = tweepy.Stream(auth, TweetListner(cSocket))  # Stream the twitter data
                # DON'T run this approach async or you'll just create a ton of streams!
                tweetStream.filter(track=["iphone", "iPhone", "iphoneX", "iphonex"])                  # Filter on trending topics
            except IncompleteRead:
                # Oh well, reconnect and keep trucking
                continue
            except KeyboardInterrupt:
                # Or however you want to exit this loop
                tweetStream.disconnect()
                break
            except Exception as e:
                logging.error("Unhandled exception has occured")
                logging.error(e)
                continue
                       
    except KeyboardInterrupt:                   # Keyboard interrupt called
        logging.error("KeyboardInterrupt was hit")
    except Exception as e:
        logging.error("Unhandled exception has occured")
        logging.error(e)