libtwitter.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

import time
import hashlib
import random
import base64
import hmac
import urllib2
import urlparse
import socket
import ssl

# timeline: https://api.twitter.com/1/statuses/home_timeline.xml
# update: https://api.twitter.com/1/statuses/update.xml
# user streams: https://userstream.twitter.com/2/user.json
# 
# Usage:
#   t = libtwitter.Twitter(consumer_key, consumer_secret
#                         [, oauth_token, oauth_token_secret])
#   response = t.request_oauth(<str>URL, <str>METHOD, <dict>POSTDATA)
# 
#   t.output_response = True / False # switch on/off of output

def dict2query(params):
    return '&'.join([str(k + '=' + str(params[k])) for k in sorted(params)])

class Twitter:
    """Twitter sample class
    
    """
    consumer = {'key': None, 'secret': None}
    oauth = {'token': None, 'token_secret': None}
    output_response = True
    
    def __init__(self, consumer_key, consumer_secret, oauth_token=None,
                  oauth_token_secret=None):
        self.set_consumer({'key': consumer_key, 'secret': consumer_secret})
        self.set_token({'token': oauth_token, 'token_secret': oauth_token_secret})
    
    def set_consumer(self, values):
        self.consumer['key'] = values['key']
        self.consumer['secret'] = values['secret']
    
    def set_token(self, values):
        self.oauth['token'] = values['token']
        self.oauth['token_secret'] = values['token_secret']
    
    def nonce(self):
        return hashlib.md5('p' + str(random.random())).hexdigest()
    
    def create_signature(self, url, method, params):
        quoted_query = urllib2.quote(dict2query(params), '')
        quoted_url = urllib2.quote(url, '')
        key = '&'.join([self.consumer['secret'], self.oauth['token_secret']])
        base_string = method + '&' + quoted_url + '&' + quoted_query
        value = base64.b64encode(hmac.new(key, base_string, hashlib.sha1).digest())
        return urllib2.quote(value, '')
    
    def create_parameters(self, url, method, post):
        params = {
            'oauth_consumer_key': self.consumer['key'],
            'oauth_signature_method': 'HMAC-SHA1',
            'oauth_timestamp': str(int(time.time())),
            'oauth_nonce': self.nonce(),
            'oauth_version': '1.0a'
        }
        if self.oauth['token']: params['oauth_token'] = self.oauth['token']
        if post: params.update(post)
        params['oauth_signature'] = self.create_signature(url, method, params)
        return params
    
    def request_oauth(self, url, method, post={}):
        params = self.create_parameters(url, method, post)
        return self.request(url, method, params)
    
    def request(self, url, method, params):
        urldict = urlparse.urlparse(url)
        if urldict.path: path = urldict.path
        else: path = '/'
        head = []
        head.append(method + ' ' + path + ' HTTP/1.1')
        head.append('Host: ' + urldict.hostname)
        body = ''
        if urldict.hostname != 'userstream.twitter.com':
            head.append('Expect:')
        if method is 'GET':
            path = path + '?' + urldict.query
            realm = urldict.scheme + '://' + urldict.hostname + '/'
            auth = ', '.join('{0}="{1}"'.format(k, params[k]) for k in params)
            head.append('Authorization: OAuth realm="' + realm + '", ' + auth)
        elif method is 'POST':
            body = dict2query(params)
            head.append('Content-Length: ' + str(len(body)))
        data = ''
        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        except:
            raise
        else:
            ssl_sock = ssl.wrap_socket(s)
            hostname = urldict.hostname
            if urldict.port:
                port = urldict.port
            elif urldict.scheme == 'https':
                port = 443
            else:
                port = 80
            ssl_sock.connect((hostname, port))
            ssl_sock.write('\r\n'.join(head) + '\r\n\r\n' + body)
            http_header = ssl_sock.read(1024*100)
            while True:
                buf = ssl_sock.read(1024*10)
                if len(buf) is 0: break
                if self.output_response: print buf
                data += buf
        finally:
            s.close()
        return data