Source code for dynamix.events.twitter

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
Creates Events from Twitter.com
"""

from __future__ import print_function, division, unicode_literals, absolute_import, generators

from datetime import datetime

from twitter import TwitterStream, Twitter

from .tools import make_literals

__author__ = "Pascal Held"
__email__ = "paheld@gmail.com"


TWITTER_TIME_FORMAT_ = '%a %b %d %H:%M:%S %z %Y'


class TwitterEvents(object):
    """Twitter api wrapper"""

    auth = None

    def __init__(self, auth):
        """Uses a twitter auth object to connect to twitter stream api

        Parameters
        ----------
        auth : twitter auth object
            Auth object used to connect.
        """
        self.auth = auth

    def raw_tweets(self):
        """Iterator for basic tweet stream."""
        t = TwitterStream(auth=self.auth)
        return t.statuses.sample()

    def hashtags(self, limit=None, min_tags=2, search_tags=None):
        """Generates events from Twitter Hashtags.

        Every tweet with hashtags generates an event with sender is a list of all containing hashtags
        and receiver not set.

        Parameters
        ----------
        limit : int or None
            Limits the result size. Default is None - no limit.
        min_tags : int
            Minimum number of hashtags (excluding search tags)
        search_tags : list of strings or string
            List of hashtags. At least one hashtag must be in the tweet
        """

        if search_tags and not isinstance(search_tags,(set, list, tuple)):
            search_tags = [search_tags]

        cnt = 0
        if search_tags:
            search_tags = [x.lower() for x in search_tags]
            t = TwitterStream(auth=self.auth)
            query = ",".join(["#"+tag for tag in search_tags])
            it = t.statuses.filter(track=query)
        else:
            t = TwitterStream(auth=self.auth)
            it = t.statuses.sample()

        for tweet in it:
            try:
                if search_tags:
                    tweet["entities"]["hashtags"] = [tweet for tweet in tweet["entities"]["hashtags"] if tweet["text"].lower() not in search_tags]
                if tweet["entities"]["hashtags"] and len(tweet["entities"]["hashtags"]) >= min_tags:
                    yield make_literals({
                        "time": get_tweet_timestamp(tweet),
                        "sender": [x["text"] for x in tweet["entities"]["hashtags"] if x["text"]]
                    })
                    cnt += 1
                    if limit and cnt >= limit:
                        break
            except KeyError:
                pass

    def user_activities(self, limit=None):
        """Generates events from Twitter Tweets.

        Every tweet generates an event with sender is the creator of the tweet.

        Parameters
        ----------
        limit : int or None
            Limits the result size. Default is None - no limit.
        """
        cnt = 0
        t = TwitterStream(auth=self.auth)
        for tweet in t.statuses.sample():
            try:
                yield make_literals({
                    "time": get_tweet_timestamp(tweet),
                    "sender": tweet['user']['screen_name']
                })
                cnt += 1
                if limit and cnt >= limit:
                    break
            except KeyError:
                pass

    def mentions(self, limit=None):
        """Generates events from Twitter mentions.

        Every tweet with mentions generates an event with sender is the creator of the tweed
        and receiver is a list of mentions.

        Parameters
        ----------
        limit : int or None
            Limits the result size. Default is None - no limit.
        """
        cnt = 0
        t = TwitterStream(auth=self.auth)
        for tweet in t.statuses.sample():
            try:
                if tweet["entities"]["user_mentions"]:
                    yield make_literals({
                        "time": get_tweet_timestamp(tweet),
                        "receiver": [x["screen_name"] for x in tweet["entities"]["user_mentions"]],
                        "sender": tweet['user']['screen_name']
                    })
                    cnt += 1
                    if limit and cnt >= limit:
                        break
            except KeyError:
                pass


def get_tweet_timestamp(tweet):
    """Extract the unix timestamp from Tweet

    Parameters
    ----------
    tweet : dict
        Twitter tweet in dict format
    """
    return datetime.strptime(tweet['created_at'], TWITTER_TIME_FORMAT_).timestamp()