Skip to content

Embed URL

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Twitter's most popular
#!/usr/bin/env python2.6
import pycurl, json
user = "YOURTWITTERNAME"
password = "YOURTWITTERPASSWORD"
streamurl = "https://stream.twitter.com/1/statuses/filter.json"
track = "locations=-74,40,-73,41" # nyc
class TwitterClient:
# set curl options
def __init__(self):
self.persons = []
self.buffer = ""
self.connection = pycurl.Curl()
self.connection.setopt(pycurl.URL, streamurl)
self.connection.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_ANY)
self.connection.setopt(pycurl.USERPWD, "{0}:{1}".format(user, password))
self.connection.setopt(pycurl.POST, 1)
self.connection.setopt(pycurl.POSTFIELDS, track)
self.connection.setopt(pycurl.WRITEFUNCTION, self.curlWriteback)
# try to talk to the api
try:
self.connection.perform()
except pycurl.error as e:
print("ooops")
finally:
self.connection.close()
# handle respones from api
def curlWriteback(self, data):
self.buffer += data
if data.endswith("\r\n"):
content = json.loads(self.buffer)
user = content["user"]["screen_name"].encode("utf-8").replace("\n", " ")
followers = content["user"]["followers_count"]
self.processTuples(user, followers)
self.buffer = ""
# for now we put the tuples in a list; feel free to improve this
def processTuples(self, user, followers):
self.persons.append((user, followers))
# print the list every time we add 50 new tuples
if(len(self.persons) % 50 == 0):
self.showResults()
# remove duplicates, sort by followers and print the tuples
def showResults(self):
self.persons = sorted(dict(self.persons).items(), key=lambda x: x[1], reverse=True)
print("Twitter's most popular (looked at {0} so far)".format(len(self.persons)))
for name, followers in self.persons[:10]:
print("{0}, {1}".format(followers, name))
# entry
if __name__ == '__main__':
TwitterClient()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Something went wrong with that request. Please try again.