Created
January 12, 2012 06:59
-
-
Save eyeseast/1599122 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python | |
import argparse | |
import datetime | |
import httplib2 | |
import json | |
import sys | |
import urllib | |
BASE = "https://api.twitter.com/1/statuses/user_timeline.json?" | |
class TwitterException(Exception): | |
pass | |
class TwitterRateLimit(TwitterException): | |
"Exception for when you hit twitter's rate limit" | |
class TwitterClient(object): | |
""" | |
Simple client to grab tweets for a user | |
while managing rate limits | |
""" | |
def __init__(self, screen_name=None): | |
self.screen_name = screen_name | |
self.http = httplib2.Http() | |
self.rate_limit_remaining = 150 | |
self.rate_limit_reset = None | |
def fetch(self, url): | |
""" | |
Get a resource from Twitter, updating rate limits | |
""" | |
r, c = self.http.request(url) | |
try: | |
self.rate_limit_remaining = int(r['x-ratelimit-remaining']) | |
self.rate_limit_reset = datetime.datetime.fromtimestamp( | |
float(r['x-ratelimit-reset']) | |
) | |
except KeyError, ValueError: | |
self.rate_limit_remaining -= 1 | |
return json.loads(c) | |
def get_single_tweet(self, id, **kwargs): | |
url = "http://api.twitter.com/1/statuses/show/%s.json?" % id | |
params = { | |
'trim_user': False, | |
'include_entities': True | |
} | |
params.update(kwargs) | |
url = url + urllib.urlencode(params) | |
return self.fetch(url) | |
def get_tweets(self, screen_name=None, **kwargs): | |
""" | |
Grab a bunch of tweets | |
""" | |
if not self.rate_limit_remaining: | |
if self.rate_limit_reset and self.rate_limit_reset > datetime.datetime.now(): | |
raise TwitterRateLimit('Rate limit exceeded') | |
if screen_name is None: | |
screen_name = self.screen_name | |
params = { | |
'include_entities': True, | |
'include_rts': True, | |
'screen_name': screen_name, | |
'count': 200, | |
'contributor_details': True | |
} | |
params.update(kwargs) | |
if not params['screen_name']: | |
raise TwitterException('No screen name') | |
url = BASE + urllib.urlencode(params) | |
return self.fetch(url) | |
if __name__ == "__main__": | |
parser = argparse.ArgumentParser(description='Grab some tweets') | |
parser.add_argument('screen_name', type=str, | |
help='Twitter user to fetch') | |
args = parser.parse_args() | |
twitter = TwitterClient(args.screen_name) | |
tweets = twitter.get_tweets() | |
for tweet in tweets: | |
try: | |
status = twitter.get_single_tweet(tweet['id_str']) | |
print status.get('text') | |
except TwitterRateLimit, e: | |
print e | |
sys.exit(1) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment