From b6a03f918ce8c862d48407d149dd49c9b6918627 Mon Sep 17 00:00:00 2001 From: Ryan McGrath Date: Thu, 28 May 2009 01:58:56 -0400 Subject: [PATCH] Many, many changes. Broke out the url shortener so it's more plug and play; users can now throw in their own desired shortening service, and it should auto work. All timeline methods are now implemented, moving on to user methods next. Refactored some parts of the library that were becoming kludgy, and set an optional debug parameter that may be useful in the future. --- tango.py | 108 +++++++++++++++++++++++++++++++++++++++++-------------- 1 file changed, 81 insertions(+), 27 deletions(-) diff --git a/tango.py b/tango.py index 1a99ed0..0802d5e 100644 --- a/tango.py +++ b/tango.py @@ -1,7 +1,11 @@ #!/usr/bin/python """ - Django-Twitter (Tango) utility functions. Huzzah. + Tango is an up-to-date library for Python that wraps the Twitter API. + Other Python Twitter libraries seem to have fallen a bit behind, and + Twitter's API has evolved a bit. Here's hoping this helps. + + Questions, comments? ryan@venodesigns.net """ import urllib, urllib2 @@ -19,12 +23,13 @@ except: print "Tango requires oauth for authentication purposes. http://oauth.googlecode.com/svn/code/python/oauth/oauth.py" class setup: - def __init__(self, authtype = "OAuth", username = None, password = None, oauth_keys = None): + def __init__(self, authtype = "OAuth", username = None, password = None, oauth_keys = None, debug = False): self.authtype = authtype self.authenticated = False self.username = username self.password = password self.oauth_keys = oauth_keys + self.debug = debug if self.username is not None and self.password is not None: if self.authtype == "OAuth": pass @@ -37,9 +42,26 @@ class setup: else: pass - def shortenURL(self, url_to_shorten): - # Perhaps we should have fallbacks here in case the is.gd API limit gets hit? Maybe allow them to set the host? - shortURL = urllib2.urlopen("http://is.gd/api.php?" + urllib.urlencode({"longurl": url_to_shorten})).read() + def explainOAuthSupport(self): + print "Sorry, OAuth support is still forthcoming. Default back to Basic Authentication for now, or help out on this front!" + pass + + # OAuth functions; shortcuts for verifying the credentials. + def fetch_response_oauth(self, oauth_request): + pass + + def get_unauthorized_request_token(self): + pass + + def get_authorization_url(self, token): + pass + + def exchange_tokens(self, request_token): + pass + + # URL Shortening function huzzah + def shortenURL(self, url_to_shorten, shortener = "http://is.gd/api.php", query = "longurl"): + shortURL = urllib2.urlopen(shortener + "?" + urllib.urlencode({query: url_to_shorten})).read() return shortURL def constructApiURL(self, base_url, params): @@ -51,28 +73,45 @@ class setup: questionMarkUsed = True return queryURL + def createGenericTimeline(self, existingTimeline): + # Many of Twitter's API functions return the same style of data. This function just wraps it if we need it. + genericTimeline = [] + for tweet in existingTimeline: + genericTimeline.append({ + "created_at": tweet["created_at"], + "in_reply_to_screen_name": tweet["in_reply_to_screen_name"], + "in_reply_to_status_id": tweet["in_reply_to_status_id"], + "in_reply_to_user_id": tweet["in_reply_to_user_id"], + "id": tweet["id"], + "text": tweet["text"] + }) + return genericTimeline + def getPublicTimeline(self): - publicTimeline = simplejson.load(urllib2.urlopen("http://twitter.com/statuses/public_timeline.json")) - formattedTimeline = [] - for tweet in publicTimeline: - formattedTimeline.append(tweet['text']) - return formattedTimeline + return self.createGenericTimeline(simplejson.load(urllib2.urlopen("http://twitter.com/statuses/public_timeline.json"))) + + def getFriendsTimeline(self, **kwargs): + if self.authenticated is True: + friendsTimelineURL = self.constructApiURL("http://twitter.com/statuses/friends_timeline.json", kwargs) + return self.createGenericTimeline(simplejson.load(self.opener.open(friendsTimelineURL))) + else: + print "getFriendsTimeline() requires you to be authenticated." + pass def getUserTimeline(self, **kwargs): - # 99% API compliant, I think - need to figure out Gzip compression and auto-getting based on authentication - # By doing this with kwargs and constructing a url outside, we can stay somewhat agnostic of API changes - it's all - # based on what the user decides to pass. We just handle the heavy lifting! :D userTimelineURL = self.constructApiURL("http://twitter.com/statuses/user_timeline/" + self.username + ".json", kwargs) - userTimeline = simplejson.load(urllib2.urlopen(userTimelineURL)) - formattedTimeline = [] - for tweet in userTimeline: - formattedTimeline.append(tweet['text']) - return formattedTimeline + try: + return self.createGenericTimeline(simplejson.load(urllib2.urlopen(userTimelineURL))) + except: + print "Hmmm, that failed. Does this user hide/protect their updates? If so, you'll need to authenticate and be their friend to get their timeline." + pass def getUserMentions(self, **kwargs): if self.authenticated is True: if self.authtype == "Basic": - pass + mentionsFeedURL = self.constructApiURL("http://twitter.com/statuses/mentions.json", kwargs) + mentionsFeed = simplejson.load(self.opener.open(mentionsFeedURL)) + return self.createGenericTimeline(mentionsFeed) else: pass else: @@ -88,26 +127,41 @@ class setup: print e.code print e.headers else: - print "Sorry, OAuth support is still forthcoming. Feel free to help out on this front!" - pass + self.explainOAuthSupport() else: print "updateStatus() requires you to be authenticated." pass def destroyStatus(self, id): if self.authenticated is True: - self.http.request("http://twitter.com/status/destroy/" + id + ".json", "POST") + if self.authtype == "Basic": + self.opener.open("http://twitter.com/status/destroy/" + id + ".json", "POST") + else: + self.explainOAuthSupport() else: print "destroyStatus() requires you to be authenticated." pass def getSearchTimeline(self, search_query, optional_page): params = urllib.urlencode({'q': search_query, 'rpp': optional_page}) # Doesn't hurt to do pages this way. *shrug* - searchTimeline = simplejson.load(urllib2.urlopen("http://search.twitter.com/search.json", params)) - formattedTimeline = [] - for tweet in searchTimeline['results']: - formattedTimeline.append(tweet['text']) - return formattedTimeline + try: + searchTimeline = simplejson.load(urllib2.urlopen("http://search.twitter.com/search.json", params)) + # This one is custom built because the search feed is a bit different than the other feeds. + genericTimeline = [] + for tweet in searchTimeline["results"]: + genericTimeline.append({ + "created_at": tweet["created_at"], + "from_user": tweet["from_user"], + "from_user_id": tweet["from_user_id"], + "profile_image_url": tweet["profile_image_url"], + "id": tweet["id"], + "text": tweet["text"], + "to_user_id": tweet["to_user_id"] + }) + return genericTimeline + except HTTPError, e: + print e.code + print e.headers def getCurrentTrends(self): # Returns an array of dictionary items containing the current trends