Year: 2015

Top 500 StackOverflow contributors from Bangladesh

Post author By masnun
Post date December 12, 2015

Update: The result returned from the StackExchange is slightly outdated. So it might not display the latest reputation or other profile changes and thus slightly affecting the ranking.

Update: Because the large list was affecting the site performance, I have moved it to Github Gist.

This post uses the StackExchange Data Explorer to query the StackOverflow users and grab their data. The Python script to query and parse the data is attached below the ranking. So let’s wait no further and meet the top 500 people on SO from Bangladesh:

View List on Gist

Script:

import requests
import csv

CSV_URL = "http://data.stackexchange.com/stackoverflow/csv/521550"

response = requests.get(CSV_URL)
data = csv.DictReader(response.content.splitlines())

for rank, row in enumerate(data):
    print "#{} - {} - {} - {}".format((rank+1),row['DisplayName'], row['Reputation'], row['WebsiteUrl'])
    
    """
    print "<li><a target='_blank' href='http://stackoverflow.com/users/{}'>{}</a> ({}) </li>".format(
            row['User Link'],
            row['DisplayName'],
            row['Reputation']
        )
    """

import requests

import csv

CSV_URL = "http://data.stackexchange.com/stackoverflow/csv/521550"

response = requests.get(CSV_URL)

data = csv.DictReader(response.content.splitlines())

for rank, row in enumerate(data):

print "#{} - {} - {} - {}".format((rank+1),row['DisplayName'], row['Reputation'], row['WebsiteUrl'])

"""

print "<li><a target='_blank' href='http://stackoverflow.com/users/{}'>{}</a> ({}) </li>".format(

row['User Link'],

row['DisplayName'],

row['Reputation']

)

"""

Python

Python 3: Using blocking functions or codes with asyncio

Post author By masnun
Post date December 7, 2015

We know we can do a lot of async stuff with asyncio but have you ever wondered how to execute blocking codes with it? It’s pretty simple actually, asyncio allows us to run blocking code using BaseEventLoop.run_in_executor method. It will run our functions in parallel and provide us with Future objects which we can await or yield from.

Let’s see an example with the popular requests library:

import asyncio
import requests

loop = asyncio.get_event_loop()

def get_html(url):
	return loop.run_in_executor(None, requests.get, url)

@asyncio.coroutine
def main():
	resp1 = yield from get_html("https://masnun.com")
	resp2 = yield from get_html("http://python.org")
	
	print(resp2, resp1)



loop.run_until_complete(main())

import asyncio

import requests

loop = asyncio.get_event_loop()

def get_html(url):

return loop.run_in_executor(None, requests.get, url)

@asyncio.coroutine

def main():

resp1 = yield from get_html("https://masnun.com")

resp2 = yield from get_html("http://python.org")

print(resp2, resp1)

loop.run_until_complete(main())

If you run the code snippet, you can see how the two responses are fetched asynchronously 🙂

Python

Creating a Twitter Retweet Bot in Python

Post author By masnun
Post date December 5, 2015

We want to create a bot that will track specific topics and retweet them. We shall use the Twitter Streaming API to track topics. We will use the popular tweepy package to interact with Twitter.

Let’s first install Tweepy

pip install tweepy

1	pip install tweepy

We need to create a Twitter app and get the tokens. We can do that from : https://apps.twitter.com/.

Now let’s see the codes:

from tweepy.streaming import StreamListener
from tweepy import OAuthHandler, API
from tweepy import Stream
import json
import logging
import warnings
from pprint import pprint

warnings.filterwarnings("ignore")

access_token = "4464294380-GHJ3PY...lEqzZ8FikULPnkxaS4huT"
access_token_secret = "L033NEsDOA...tZVoaU8prpFbiREWhcVROw2"
consumer_key = "D1G0bn...9Y88p"
consumer_secret = "rGJjCU...FRISsiMURYlqCXJvOP"

auth_handler = OAuthHandler(consumer_key, consumer_secret)
auth_handler.set_access_token(access_token, access_token_secret)

twitter_client = API(auth_handler)

logging.getLogger("main").setLevel(logging.INFO)

AVOID = ["monty", "leather", "skin", "bag", "blood", "bite"]


class PyStreamListener(StreamListener):
    def on_data(self, data):
        tweet = json.loads(data)
        try:
            publish = True
            for word in AVOID:
                if word in tweet['text'].lower():
                    logging.info("SKIPPED FOR {}".format(word))
                    publish = False

            if tweet.get('lang') and tweet.get('lang') != 'en':
                publish = False

            if publish:
                twitter_client.retweet(tweet['id'])
                logging.debug("RT: {}".format(tweet['text']))

        except Exception as ex:
            logging.error(ex)

        return True

    def on_error(self, status):
        print status


if __name__ == '__main__':
    listener = PyStreamListener()
    stream = Stream(auth_handler, listener)
    stream.filter(track=['python', 'django', 'kivy', 'scrapy'])

from tweepy.streaming import StreamListener

from tweepy import OAuthHandler, API

from tweepy import Stream

import json

import logging

import warnings

from pprint import pprint

warnings.filterwarnings("ignore")

access_token = "4464294380-GHJ3PY...lEqzZ8FikULPnkxaS4huT"

access_token_secret = "L033NEsDOA...tZVoaU8prpFbiREWhcVROw2"

consumer_key = "D1G0bn...9Y88p"

consumer_secret = "rGJjCU...FRISsiMURYlqCXJvOP"

auth_handler = OAuthHandler(consumer_key, consumer_secret)

auth_handler.set_access_token(access_token, access_token_secret)

twitter_client = API(auth_handler)

logging.getLogger("main").setLevel(logging.INFO)

AVOID = ["monty", "leather", "skin", "bag", "blood", "bite"]

class PyStreamListener(StreamListener):

def on_data(self, data):

tweet = json.loads(data)

try:

publish = True

for word in AVOID:

if word in tweet['text'].lower():

logging.info("SKIPPED FOR {}".format(word))

publish = False

if tweet.get('lang') and tweet.get('lang') != 'en':

publish = False

if publish:

twitter_client.retweet(tweet['id'])

logging.debug("RT: {}".format(tweet['text']))

except Exception as ex:

logging.error(ex)

return True

def on_error(self, status):

print status

if __name__ == '__main__':

listener = PyStreamListener()

stream = Stream(auth_handler, listener)

stream.filter(track=['python', 'django', 'kivy', 'scrapy'])

The code is pretty much self explanatory:

We create a Twitter API client using the oAuth details we got earlier
We subclass StreamListener to implement our own on_data method
We create an instance of this class, then create a new Stream by passing the auth handler and the listener
We use the track method to track a number of topics we are interested in
When we start to track the topics, it will pass the data to on_data method where we parse the tweet, check some common words to avoid, check language and then retweet it.

Recent Posts

Recent Comments

Archives

Categories