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.

This commit is contained in:
Ryan McGrath 2009-05-28 01:58:56 -04:00
parent 60c6aae346
commit b6a03f918c

106
tango.py
View file

@ -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*
try:
searchTimeline = simplejson.load(urllib2.urlopen("http://search.twitter.com/search.json", params))
formattedTimeline = []
for tweet in searchTimeline['results']:
formattedTimeline.append(tweet['text'])
return formattedTimeline
# 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