4
\$\begingroup\$

I have a fully working Python server that utilizes Tornado, asyncio, websockets and tweepy(Twitter's Streaming api). The main point of this server is to receive a query from the user, call upon tweepy to track the query, perform sentiment analysis on any message received and send it back to the client.

All messages sent by the client are a json message of the form:

{
  track: 'query to track'
}

All messages sent by the server are a json message of the form:

{
  polarityIndex: 'An index where 0 is positive sentiment, 1 is negative sentiment and 2 is neutral',
  hashtags: 'A dictionary of all the hashtags that were found and their frequency.'
}

There are four main modules that make up this entire server. There are also two extra modules but those two don't do anything interesting. One of them simply setup a logger and the other one loads up environment variables.

The first one we are going to look at is the server.py file

import logging
import ssl
from queue import Queue
from setup_logger import ROOT_LOGGER
from tornado import web, ioloop
from tornado.options import define, options, parse_command_line
from websocket_handler import WSHandler
from constants import SETTINGS
from tweet_stream_listener import TweetStreamListener, listen_for_tweets

define("port", default=8000, help="run on the given port.", type=int)
define("debug", default=True, help="run in debug mode.", type=bool)


def main():
    parse_command_line()

    logger = logging.getLogger(ROOT_LOGGER)

    settings = {
        "debug": options.debug,
    }

    app = web.Application(
        [
            (r"/track", WSHandler),
        ],
        **settings
    )

    queue = Queue()
    app.listener = TweetStreamListener(queue)

    context = None

    if SETTINGS['CERTFILE'] and SETTINGS['KEYFILE']:
        context = ssl.SSLContext()
        context.load_cert_chain(SETTINGS['CERTFILE'], SETTINGS['KEYFILE'])

    app.listen(port=options.port, ssl_options=context)
    logger.info("Server listening on port: {0}".format(options.port))

    loop = ioloop.IOLoop.current()
    loop.run_in_executor(None, listen_for_tweets, app.listener, queue)

    try:
        loop.start()
    except KeyboardInterrupt:
        loop.stop()

     # This is a sentinel value to to the consumer queue that we are done.
    queue.put(None)
    app.listener.stop_tracking()

    logger.info('Server shutting down.')


if __name__ == "__main__":
    main()

Nothing too interesting here except for a couple of things. Notice at the end I'm putting none in a queue that is being shared the twitter streamer listener class and a threaded function of some sort. You can think of the queue in the twitter stream listener as my producer queue and the queue that is being passed into to my threaded function as my consumer queue. The second interesting thing that I'm doing is I'm attaching my listener to the main application object app so that I have access to it in my websocket handler class. Now we are going to move on to the websocket handler class.

import logging
import json
from constants import SETTINGS
from urllib.parse import urlparse
from uuid import uuid4
from tornado import ioloop, gen
from tornado.websocket import WebSocketHandler


class WSHandler(WebSocketHandler):

    LOGGER = logging.getLogger(__qualname__)
    WHITELISTED_DOMAINS = SETTINGS['WHITELISTED_DOMAINS'].split(",")

    def check_origin(self, origin):
        parsed_origin = urlparse(origin)

        if parsed_origin.hostname is 'localhost':
            return True

        domain = ".".join(parsed_origin.netloc.split(".")[1:])
        return domain in WSHandler.WHITELISTED_DOMAINS

    def open(self):
        self._sess_id = uuid4().hex
        WSHandler.LOGGER.debug(
            'Websocket with id {0} is now connected.'.format(self._sess_id))
        self.application.listener.websockets[self._sess_id] = self

    @gen.coroutine
    def on_message(self, body):
        WSHandler.LOGGER.debug(
            'Websocket {0} has received a message: {1}'.format(self._sess_id, body))

        message = json.loads(body)
        yield self.wait_still_stream_finishes(message['track'])

    @gen.coroutine
    def wait_still_stream_finishes(self, message):
        # Disconnect the stream momentarily.
        self.application.listener.stop_tracking()

        while self.application.listener.is_stream_running():
            yield gen.sleep(1)

        ioloop.IOLoop.current().run_in_executor(None,
                                                self.application.listener.start_tracking,
                                                self._sess_id, message)

    def on_close(self):
        WSHandler.LOGGER.debug(
            'Websocket with id {0} has disconnected.'.format(self._sess_id))
        self.application.listener.websockets.pop(self._sess_id)

        if(len(self.application.listener.websockets) is 0):
            WSHandler.LOGGER.debug(
                "No more connections to keep track of. Closing stream.")
            self.application.listener.stop_tracking()

There is quite a bit going on in this class so I'm going to try to explain as much as I can. First, because tornado is stateless and I have no way of knowing which websocket made which request, I need to introduce some internal state. Every time there is a new connection, we generate a uuid4 so we can identify any websocket easily in the dictionary via their uuid4. The twitter streamer listener class is the one that is keeping track of all active websockets and anytime one of the websockets close the connection, we remove the websocket from the dictionary and remove the query from the list of queries that we are keeping track of.

Now, there is a limitation with tweepy and maybe some of the maintainers of tweepy can tell me of a better way of doing this but when a user makes query, I need to temporarily stop the stream, recreate the list of query with the new query while keeping all the queries that other users have made intact and then start streaming again. Of course, I can not simply disconnect the stream and start it up again. If I did, there would not be enough time for the previous thread to finish before I could create a new thread and start streaming again. This solution I don't really like at all and would like ideas on how I can do this better. Let us move to the twitter stream listener module.

import logging
import asyncio
from tornado.websocket import WebSocketClosedError
from tweepy import StreamListener, OAuthHandler, Stream, API
from constants import SETTINGS
from tweet_sentiment_analyzer import get_polarity_index_from_tweet, \
    get_hashtag_frequencies_from_tweet, preprocess_tweet

auth = OAuthHandler(
    SETTINGS["TWITTER_CONSUMER_API_KEY"], SETTINGS["TWITTER_CONSUMER_API_SECRET_KEY"])

auth.set_access_token(
    SETTINGS["TWITTER_ACCESS_KEY"], SETTINGS["TWITTER_ACCESS_SECRET_KEY"])

api = API(auth, wait_on_rate_limit=True)


class TweetStreamListener(StreamListener):

    def __init__(self, queue):
        self.api = api
        self.queue = queue
        self.websockets = {}
        self.logger = logging.getLogger(self.__class__.__name__)
        self.stream = Stream(auth=self.api.auth, listener=self,
                             tweet_mode='extended')
        self.current_searches = {}
        self.streaming = False

    def start_tracking(self, sess_id, track):
        self.logger.debug('Stream has started')
        self.streaming = True
        tracking_list = self.get_updated_tracking_list(sess_id, track)
        self.stream.filter(track=tracking_list, languages=['en'])
        self.streaming = False
        self.logger.debug('Stream has stopped')

    def get_updated_tracking_list(self, sess_id, track):
        # New tweet to listen to.
        self.current_searches[sess_id] = track.lower()

        '''
            Create a new search term list and put it all in a set to remove
            potential duplicate search terms.
        '''
        return set(self.current_searches.values())

    def stop_tracking(self):
        self.stream.disconnect()

    def is_stream_running(self):
        return self.streaming

    def on_status(self, status):
        # We are not processing retweets. Only new tweets.
        if getattr(status, 'retweeted_status', None):
            return

        self.queue.put(status)

    def on_timeout(self, status):
        self.logger.error('Stream disconnected. continuing...')
        return True  # Don't kill the stream

    """
    Summary: Callback that executes for any error that may occur. Whenever we get a 420 Error code, we simply
    stop streaming tweets as we have reached our rate limit. This is due to making too many requests.

    Returns: False if we are sending too many tweets, otherwise return true to keep the stream going.
    """

    def on_error(self, status_code):
        if status_code == 420:
            self.logger.error(
                'Encountered error code 420. Disconnecting the stream')
            # returning False in on_data disconnects the stream
            return False
        else:
            self.logger.error('Encountered error with status code: {}'.format(
                status_code))
            return True  # Don't kill the stream


def listen_for_tweets(listener, queue):
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    while True:
        message = queue.get()

        if message is None:
            break

        # We still have websockets to send this message to.
        if len(listener.websockets) > 0:
            task = asyncio.ensure_future(process_tweet(message, listener.current_searches,
                                                       listener.websockets))
            loop.run_until_complete(task)

        queue.task_done()

    loop.close()


async def process_tweet(status, current_searches, websockets):
    tweet = status.text.lower()

    # If extended_tweet exists, this the means that status.text is truncated.
    # We want the entire text.
    if getattr(status, 'extended_tweet', None):
        tweet = status.extended_tweet['full_text'].lower()

    polarity_tweet, filtered_hashtag_list = preprocess_tweet(tweet)

    polarityIndex = get_polarity_index_from_tweet(polarity_tweet)
    hashtag_freqs = get_hashtag_frequencies_from_tweet(filtered_hashtag_list)

    message = {
        'polarityIndex': polarityIndex,
        'hashtags': dict(hashtag_freqs)
    }

    sess_ids = []

    for sess_id, topic in current_searches.items():
        if topic in filtered_hashtag_list:
            sess_ids.append(sess_id)

    for sess_id in sess_ids:
        try:
            # We need this check and exception in case a websocket closes abruptly.
            if sess_id in websockets:
                await websockets[sess_id].write_message(message)
        except WebSocketClosedError:
            continue

So this is the module that listens for tweets and figures out which websockets sends which messages. Now let us talk about the two queues that are being used in this module. Depending if the topic that the user searched for is trending or not, it may be the case that we have to analyze alot of incoming messages. To offload this work, any message that we receive from twitter we simply check that it is not a retweet and put it in the queue. Then once a message has been placed in the queue, the queue in the threaded function listen for tweets will get the message off of the queue, process the tweet and figure out which websocket sends which messages.

Notice that there is another dictionary called current_searches. The key for that dictionary is the same key that the websocket dictionary uses. The whole point of the current_searches dictionary is so that we know which query belongs to which socket. When we receive a message from twitter and have finished analyzing it, we then need to figure out who should receive this message. To keep it simple, all we do is look through the tweet and see which query is in this tweet. If we find a match, we store their key in a list. Of course, a tweet may contain multiple queries and so we simply add in all of their keys in the list as well. The last module that we are going to look at is the twitter sentiment analyzer module.

import re
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
from nltk.corpus import stopwords
from nltk import FreqDist

analyser = SentimentIntensityAnalyzer()
default_stopwords = set(stopwords.words('english'))


def get_polarity_index_from_tweet(text):
    polarity_scores = analyser.polarity_scores(text)
    polarityIndex = 2

    # 0 is positive sentiment, 1 is negative sentiment and 2 is neutral sentiment
    if polarity_scores['compound'] > 0.05:
        polarityIndex = 0
    elif polarity_scores['compound'] < -0.05:
        polarityIndex = 1

    return polarityIndex


def remove_pattern(text, pattern):
    matches = re.findall(pattern, text)

    for match in matches:
        text = re.sub(match, '', text)

    return text


def preprocess_tweet(tweet):
    # remove twitter handles (@xxx)
    tweet = remove_pattern(tweet, r"@[\w]*")

    # remove URL links (httpxxx)
    tweet = remove_pattern(tweet, r"https?://[A-Za-z0-9./]*")

    '''
        remove special characters, numbers and punctuations. Exceptions are #, !, ' and ? as vader uses both ! and ?
        characters for emphasis and will affect our polarity score, vader also takes into account contractions and we
        want to keep hashtags to calculate which hashtags were more frequent. Because of all these exceptions, we are
        going to have two tweets. One tweet called polarity tweet that is feed into the vader analyzer and another tweet
        that only has letters and the hashtag. The second tweet is what we will use to calculate the frequency of a hashtag
        from each tweet that we get and will be fed into a function that ntlk has called FreqDist to calculate the frequency
        distribution in a tweet.
    '''
    polarity_tweet = re.sub(r"[^a-zA-Z#!?']", " ", tweet)
    hashtag_tweet = re.sub(r"[^a-zA-Z#]", " ", tweet)

    tokenize_hashtag_list = hashtag_tweet.split()
    filtered_hashtag_list = [
        word for word in tokenize_hashtag_list if not word in default_stopwords]

    return (polarity_tweet, filtered_hashtag_list)


def extract_hashtags(tweet_list):
    hashtags = []

    for word in tweet_list:
        ht = re.findall(r"#(\w+)", word)
        hashtags.append(ht)

    return hashtags


def get_hashtag_frequencies_from_tweet(tweet_list):
    hashtags = extract_hashtags(tweet_list)
    return FreqDist(sum(hashtags, []))

To perform sentiment analysis, I use a pre train mode that supposedly does well with social media text. I have a simple threshold that determines whether something is positive, negative or neutral. This is a threshold that the maintainers of vader recommend. Then, I also have a couple of preprocessing functions that strips out @ and https text, extract hashtags and count their frequencies. I also have this big comment which you guys should read so you guys know why I did what I did when it came to preprocessing each tweet.

Here is the link to the repo Senti Server for those that want to take a look at the entire repo. Feel free to respond on how I can make this better and make it more testable.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.