Created
March 3, 2013 06:28
-
-
Save omalley/5074976 to your computer and use it in GitHub Desktop.
A python script that uses the twitter api to find the most prolific tweeters in your feed.
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/python | |
import base64 | |
import hmac | |
import hashlib | |
import httplib | |
import json | |
import random | |
import string | |
import sys | |
import time | |
import urllib | |
import urlparse | |
# Register on https://dev.twitter.com/apps to get yours | |
oauth_consumer_key = "yours" | |
access_token = "yours" | |
oauth_consumer_secret = "yours" | |
access_token_secret = "yours" | |
def quote(str): | |
return urllib.quote(str, '') | |
def computeOauthSignature(method, url, authParams, params): | |
params = params.copy() | |
params.update(authParams) | |
msg = "&".join(map(lambda p: "%s=%s" % p, | |
sorted(map(lambda p: tuple(map(quote,p)), params.items()), | |
key=lambda p: p[0]))) | |
msg = "&".join(map(quote, [method, url, msg])) | |
signKey = "&".join(map(quote, [oauth_consumer_secret, access_token_secret])) | |
return base64.b64encode(hmac.new(signKey, msg, hashlib.sha1).digest()) | |
def callTwitter(method, url, params): | |
nonce = ''.join(random.choice(string.ascii_letters + string.digits) | |
for x in range(32)) | |
authParams = {'oauth_consumer_key': oauth_consumer_key, | |
'oauth_nonce': nonce, | |
'oauth_signature_method': 'HMAC-SHA1', | |
'oauth_timestamp': str(int(time.time())), | |
'oauth_token': access_token, | |
'oauth_version': '1.0'} | |
authParams['oauth_signature'] = \ | |
computeOauthSignature(method, url, authParams, params) | |
auth = "OAuth " + ', '.join(map(lambda p: '%s="%s"' % tuple(map(quote, p)), | |
authParams.items())) | |
paramStr = urllib.urlencode(params) | |
parsedUrl = urlparse.urlparse(url) | |
connect = httplib.HTTPSConnection(parsedUrl.hostname) | |
# connect.set_debuglevel(1) | |
connect.request(method, parsedUrl.path + "?" + paramStr, None, | |
{'Authorization': auth, | |
'Content-Type': 'application/json', | |
'Accept': '*/*'}) | |
response = connect.getresponse() | |
if response.status != 200: | |
print "status", response.status | |
print "reason", response.reason | |
sys.exit(1) | |
return response.read() | |
timelineUrl = "https://api.twitter.com/1.1/statuses/home_timeline.json" | |
minId = 1<<62 | |
counts = {} | |
for i in range(5): | |
data = callTwitter("GET", timelineUrl, {'count': '200', | |
'max_id': str(minId)}) | |
tweets = json.loads(data) | |
for tweet in tweets: | |
counts.setdefault(tweet["user"]["screen_name"], set()).add(tweet["id"]) | |
if tweet['id'] < minId: | |
minId = tweet['id'] | |
users = sorted(counts.items(), key=lambda u: -len(u[1])) | |
for (user,tweets) in users: | |
print "%d\t%s" % (len(tweets), user) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment