A quick Python script that implements "paging" -- or at least Twitter's version of it -- to search the social media platform's API for a given hashtag

blogpost.md

I helped the folks at #wjchat a while back by writing a quick django app to pull tweets each week and save them to a database. The tweets were then accessible as formatted HTML that could be posted in the #wjchat Wordpress blog.

But something happened around the middle of April. The script pulled in 389 tweets on April 16, but the following week only caught 100. Same for the week after.

I did some hand-holding for a few weeks, but more and more of the same occurred.

#wjchat-transcript news: @ChrisLKeller & I are aware we’re a couple weeks behind. Pipes running the transcripts are a little clogged. Soon!

— P. Kim Bui (@kimbui) June 20, 2014

<script async src="//platform.twitter.com/widgets.js" charset="utf-8"></script>

The script -- which was using the Tweepy library -- was missing the first hour or so of tweets, and leaving the archive incomplete. Time for a refactor.

I'm not blaming Tweepy but I wanted to see what else was out there. I tried making a go with requests and oauth but I don't think I'm ready yet. But the twitter library seems to work well.

Tell you the truth: the library didn't matter. It comes down to how tweets are returned by the search API -- descending order -- the maximum number of results per page -- 100 -- and this thing called the max_id.

The max_id is what amounts to "paging" it seems for the Twitter API.

The solution to the issue described above is to use a technique for working with streams of data called cursoring. Instead of reading a timeline relative to the top of the list (which changes frequently), an application should read the timeline relative to the IDs of Tweets it has already processed. This is achieved through the use of the max_id request parameter.

So the best I could get in a single search after April 16 was 100. And once I had 100 I had no means of moving on to the next page. Thankfully tonight I figured out pretty fast how the max_id parameter can work.

The script is based on the #wjchat script, and takes a hashtag and a search start date as arguments and outputs the results as a csv. It requires the following packages...

    pip install python-dateutil==2.1
    pip install pytz==2013b
    pip install twitter==1.14.3

twitter_search.py

import csv
import time
import logging
import datetime
from datetime import tzinfo
import pytz
from pytz import timezone
from dateutil import parser
from twitter import *

logging.basicConfig(format='\033[1;36m%(levelname)s:\033[0;37m %(message)s', level=logging.DEBUG)

TWITTER_CONSUMER_KEY = "#"
TWITTER_CONSUMER_SECRET = "#"
TWITTER_ACCESS_TOKEN = "#"
TWITTER_ACCESS_TOKEN_SECRET = "#"

LOCAL_TIMEZONE = pytz.timezone("US/Pacific")
TWITTER_TIMEZONE = timezone("UTC")

# you can really only search back 6 or 7 days
START_DATE_FOR_SEARCH = LOCAL_TIMEZONE.localize(datetime.datetime(2014, 6, 25, 17, 0))

# hashtag to search
HASHTAG = "#wjchat"

def init_search(start_date_to_search, hashtag):
    ''' starts twitter search and loops through the results '''

    # set our date defaults for comparisons
    start_date_utc = start_date_to_search.astimezone(TWITTER_TIMEZONE)

    # default params for our loop
    max_id = None
    search_is_done = False

    csv_output_file = open("%s.csv" % (hashtag), 'wb')
    csv_output_data = csv.writer(csv_output_file, delimiter=',', quoting=csv.QUOTE_ALL)

    csv_headers = [
        "hashtag",
        "tweet_utc_date",
        "user_name",
        "user_screen_name",
        "tweet_text",
        "tweet_url"
        "tweet_id",
        "user_profile_image_url",
        "user_location",
    ]

    csv_output_data.writerow(csv_headers)

    # begin the loop
    while (search_is_done == False):

        # return our tweets
        tweet_results = construct_twitter_search(hashtag, max_id)

        # for each status
        for tweet in tweet_results["statuses"]:

            # get the UTC time for each
            tweet_comparison_date = parser.parse(tweet["created_at"])
            tweet_comparison_date = tweet_comparison_date.replace(tzinfo=TWITTER_TIMEZONE)

            # if the tweet falls between our begin and end range
            if tweet_comparison_date >= start_date_utc:

                search_is_done = False

                # output some information
                logging.debug("%s - %s - %s" % (
                    tweet_comparison_date,
                    tweet["user"]["screen_name"],
                    tweet["id"]
                ))

                csv_row_data = [
                    hashtag,
                    tweet_comparison_date,
                    tweet["user"]["name"].encode('ascii', 'ignore'),
                    tweet["user"]["screen_name"].encode('ascii', 'ignore'),
                    tweet["text"].encode('ascii', 'ignore'),
                    "https://twitter.com/" + tweet["user"]["screen_name"].encode('ascii', 'ignore') + "/status/" + str(tweet["id"]),
                    tweet["id"],
                    tweet["user"]["profile_image_url"].encode('ascii', 'ignore'),
                    tweet["user"]["location"].encode('ascii', 'ignore'),
                ]

                csv_output_data.writerow(csv_row_data)

            # if the tweet falls outside our begin and end range
            else:
                # our search is done
                search_is_done = True

        # if we get through the loop, set the new max id, which is in effect paging
        max_id = get_max_id(tweet_results)
        logging.debug("Retrieving more tweets since %s" % (max_id))

    csv_output_file.close()

def construct_twitter_search(hashtag, max_id):
    ''' reusable function to return tweets '''
    twitter_object = Twitter(
        auth=OAuth(
            TWITTER_ACCESS_TOKEN,
            TWITTER_ACCESS_TOKEN_SECRET,
            TWITTER_CONSUMER_KEY,
            TWITTER_CONSUMER_SECRET
        )
    )

    tweet_results = twitter_object.search.tweets(
        q=hashtag,
        count=1000,
        result_type="recent",
        include_entities=True,
        max_id = max_id,
        lang="en"
    )

    return tweet_results


def get_max_id(results):
    ''' get the max_id of the next twitter search '''
    parsed_string = results["search_metadata"]["next_results"].split("&")
    parsed_string = parsed_string[0].split("?max_id=")
    max_id = parsed_string[1]
    return max_id


if __name__ == "__main__":
    init_search(START_DATE_FOR_SEARCH, HASHTAG)

wjchat.csv