twitter.py

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

import os
import yaml
import urllib
import logging
import simplejson

# for OAuth
import time
import hmac
import random
import hashlib


# yaml for twitter api is the below format
# 'http://api.twitter.com/1':
#   'statuses':
#     - [ 'show'  , 'GET' , 'show'  , ''       ]
#     - [ 'update', 'POST', 'update'. 'status' ]
#   'account':
#     - [ 'verify_credentials', 'GET', 'verify_credentials.json', '' ]
#
# Twitter API Documentation:
#   http://apiwiki.twitter.com/Twitter-API-Documentation
#
def get_api_list(filename):
    result = []
    def makelist(d, prefix=''):
        for k, v in d.iteritems():
            prefix_parts = [ prefix, k ]
            if k == '': prefix_parts = prefix_parts[0]
            elif k[-1] != '/': prefix_parts.append('/')
            new_prefix = ''.join(prefix_parts)
            
            if isinstance(v, dict):
                makelist(v, new_prefix)
            elif isinstance(v, list):
                for name, http_method, url, require in v:
                    url = ''.join([new_prefix, url, '.json'])
                    result.append((name, http_method.upper(), url, require))
    if os.path.exists(filename):
        makelist(yaml.load(open(filename).read()))
    return result


def get_api_callee(method_p, name, url, require):

    def call_api(**argv):
        if require:
            require_set = set( x.strip() for x in require.split(',') )
            
            lack = list(require_set - set(argv.keys()))
            if lack:
                if len(lack) == 1:
                    params = lack[0]
                else:
                    params = '%s and %s' % (', '.join(lack[:-1]), lack[-1])
                error_message = '%s requires parameters %s' % (name, params)
                raise KeyError, error_message
        
        request_url = url % argv
        logging.debug('api: %s (%s)' % (name, request_url))
        return method_p(request_url, **argv)
    return call_api


class OAuth:
    def __init__(self, consumer, token):
        self.consumer_key    = consumer.get('consumer_key')
        self.consumer_secret = consumer.get('consumer_secret')
        self.token           = token.get('token')
        self.token_secret    = token.get('token_secret')


class TwitterAPI:
    def __init__(self, oauth):
        self.oauth = oauth
        self.impl_api = {}
        self.bind_api_methods()
    
    
    # bind methods for twitter api defined twitterapi.yaml
    def bind_api_methods(self):
        apilist  = get_api_list('twitterapi.yaml')
        method_dict = dict( GET=self.GET, POST=self.POST, DELETE=self.DELETE)
        
        for name, http_method, url, require in apilist:
            if name in self.impl_api: continue
            
            method = method_dict.get(http_method)
            if method:
                setattr(self, name, get_api_callee(method, name, url, require))
                self.impl_api[name] = getattr(self, name)
    
    
    def call(self, api_name, **argv):
        if api_name in self.impl_api:
            return self.impl_api[api_name](**argv)
        else:
            raise UnboundLocalError, 'API is not implemented: "%s"' % api_name
    
    
    def GET(self, url, **extra_params):
        req_url = self.get_signed_url(url, self.oauth, 'GET', **extra_params);
        result = urllib.urlopen(req_url)
        content = '\n'.join(result.readlines())
        return simplejson.loads(content)
    
    
    def POST(self, url, **extra_params):
        payload = self.get_signed_body(url, self.oauth, 'POST', **extra_params)
        result = urllib.urlopen(url, payload)
        content = '\n'.join(result.readlines())
        return simplejson.loads(content)
    
    
    def DELETE(self, url, **extra_params):
        extra_params['_method'] = 'DELETE'
        return self.POST(url, **extra_params)
    
    
    def get_signed_url(self, url, oauth, method, **extra_params):
        query_string = self.get_signed_body(url, oauth, method, **extra_params)
        return '%s?%s' % (url, query_string)
    
    
    def get_signed_body(self, url, oauth, method, **extra_params):
        
        def encode(text):
            return urllib.quote(str(text), '')
        
        kwargs = {
            'oauth_consumer_key': oauth.consumer_key,
            'oauth_signature_method': 'HMAC-SHA1',
            'oauth_version': '1.0',
            'oauth_timestamp': int(time.time()),
            'oauth_nonce': random.getrandbits(64),
            'oauth_token': oauth.token,
            }
        for k, v in extra_params.iteritems():
            if isinstance(v, unicode):
                extra_params[k] = v.encode('utf-8')
        kwargs.update(extra_params)
        
        kv = ['%s=%s' % (encode(k), encode(kwargs[k])) for k in sorted(kwargs)]
        q  = '&'.join(map(encode, [ method.upper(), url, '&'.join(kv) ]))
        
        service_key  = encode(oauth.consumer_secret)
        token_secret = encode(oauth.token_secret)
        oauth_key    = '%s&%s'% (service_key, token_secret)
        
        kwargs['oauth_signature'] = \
            hmac.new(oauth_key, q, hashlib.sha1).digest().encode('base64')[:-1]
        
        return urllib.urlencode(kwargs)

twitterapi.yaml

'http://api.twitter.com/1':
  'statuses':
    - [ 'public_timeline' , 'GET', 'public_timeline', '' ]
    - [ 'home_timeline'   , 'GET', 'home_timeline'   , '' ]
    - [ 'friends_timeline', 'GET', 'friends_timeline', '' ]
    - [ 'user_timeline'   , 'GET', 'user_timeline'   , '' ]
    - [ 'mentions'        , 'GET', 'mentions'        , '' ]
    - [ 'replies'         , 'GET', 'mentions'        , '' ]
    - [ 'retweeted_by_me' , 'GET', 'retweeted_by_me' , '' ]
    - [ 'retweeted_to_me' , 'GET', 'retweeted_to_me' , '' ]
    - [ 'retweets_of_me'  , 'GET', 'retweets_of_me'  , '' ]
    
    - [ 'show'    , 'GET' , 'show/%(id)s'    , 'id'     ]
    - [ 'update'  , 'POST', 'update'         , 'status' ]
    - [ 'tweet'   , 'POST', 'update'         , 'status' ]
    - [ 'destroy' , 'POST', 'destroy/%(id)s' , 'id'     ]
    - [ 'retweet' , 'POST', 'retweet/%(id)s' , 'id'     ]
    - [ 'retweets', 'GET' , 'retweets/%(id)s', 'id'     ]
    - [ 'retweeted_by'    , 'GET', '%(id)s/retweeted_by'    , 'id' ]
    - [ 'retweeted_by_ids', 'GET', '%(id)s/retweeted_by/ids', 'id' ]

    - [ 'friends'  , 'GET', 'friends'  , '' ]
    - [ 'followers', 'GET', 'followers', '' ]

  'users':
    - [ 'users_show'  , 'GET' , 'show'  , '' ]
    - [ 'users_lookup', 'GET' , 'lookup', '' ]
    - [ 'users_search', 'GET' , 'search', '' ]
    - [ 'users_suggestions', 'GET' , 'suggestions', '' ]
    - [ 'users_suggestions_category', 'GET' , 'suggestions/slug', '' ]

  'user':
    - [ 'create_list'  , 'POST'  , 'lists'       , 'name' ]
    - [ 'update_list'  , 'POST'  , 'lists/%(id)s', 'id'   ]
    - [ 'get_list'     , 'GET'   , 'lists'       , ''     ]
    - [ 'show_list'    , 'GET'   , 'lists/%(id)s', 'id'   ]
    - [ 'destroy_list' , 'DELETE', 'lists/%(id)s', 'id'   ]
    - [ 'list_statuses', 'GET', 'lists/%(list_id)s/statuses', 'list_id' ]
    - [ 'list_memberships'  , 'GET', 'lists/memberships'  , '' ]
    - [ 'list_subscriptions', 'GET', 'lists/subscriptions', '' ]

    - [ 'list_get_members', 'GET'   , '%(list_id)s/members', 'list_id' ]
    - [ 'list_add_members', 'POST'  , '%(list_id)s/members', 'list_id,id' ]
    - [ 'list_del_members', 'DELETE', '%(list_id)s/members', 'list_id,id' ]
    - [ 'list_chk_members', 'GET'   , '%(list_id)s/members/%(id)s', 'list_id,id' ]
    
  '':
    - [ 'direct_messages', 'GET' , 'direct_messages', '' ]
    - [ 'favorites'      , 'GET' , 'favorites'      , '' ]
    - [ 'report_spam'    , 'POST', 'report_spam'    , '' ]
    
    
  'direct_messages':
    - [ 'sent_direct_messages'   , 'GET' , 'sent'          , ''          ]
    - [ 'new_direct_messages'    , 'POST', 'new'           , 'user,text' ]
    - [ 'destroy_direct_messages', 'POST', 'destroy/%(id)s', 'id'        ]

  'favorites':
    - [ 'create_favorites' , 'POST', 'create/%(id)' , 'id' ]
    - [ 'fav'              , 'POST', 'create/%(id)' , 'id' ]
    - [ 'destroy_favorites', 'POST', 'destroy/%(id)', 'id' ]
    - [ 'unfav'            , 'POST', 'create/%(id)' , 'id' ]

  'notifications':
    - [ 'notifications_follow', 'POST', 'follow/%(id)s', 'id' ]
    - [ 'notifications_leave' , 'POST', 'leave/%(id)s' , 'id' ]

  'friendships':
    - [ 'create_friendships' , 'POST' , '%(id)s'          , 'id' ]
    - [ 'follow'             , 'POST' , '%(id)s'          , 'id' ]
    - [ 'destroy_friendships', 'POST' , 'destroy/%(id)s'  , 'id' ]
    - [ 'unfollow'           , 'POST' , 'destroy/%(id)s'  , 'id' ]
    - [ 'show_friendships'   , 'GET'  , 'show'            , ''   ]
    - [ 'incoming'           , 'GET'  , 'incoming', 'cursor' ]
    - [ 'outgoing'           , 'GET'  , 'outgoint', 'cursor' ]

  'friends':
    - [ 'friends_ids'        , 'GET' , 'ids'  , '' ]

  'followers':
    - [ 'followers_ids'      , 'GET' , 'ids'  , '' ]
      
  'account':
    - [ 'verify_credentials'    , 'GET' , 'verify_credentials', '' ]
    - [ 'rate_limit_status'     , 'GET' , 'rate_limit_status' , '' ]
    - [ 'end_session'           , 'POST', 'end_session'       , '' ]
    - [ 'update_delivery_device', 'POST', 'update_delivery_device', 'device' ]
    - [ 'update_profile_colors' , 'POST', 'update_profile_colors', '' ]
    - [ 'update_profile_image'  , 'POST', 'update_profile_image', 'image' ]
    - [ 'update_profile_background_image', 'POST', 'update_profile_background_image', 'image' ]
    - [ 'update_profile' , 'POST', 'update_profile', '' ]

  'blocks':
    - [ 'create_blocks' , 'POST', 'create/%(id)s' , 'id' ]
    - [ 'block'         , 'POST', 'create/%(id)s' , 'id' ]
    - [ 'destroy_blocks', 'POST', 'destroy/%(id)s', 'id' ]
    - [ 'exists_blocks' , 'GET' , 'exists/%(id)s' , 'id' ]
    - [ 'blocking'      , 'GET' , 'blocking'      , ''   ]
    - [ 'blocking_ids'  , 'GET' , 'blocking/ids'  , ''   ]

    
'http://search.twitter.com/':
  - [ 'search'        , 'GET', 'search'        , '' ]
  - [ 'trends'        , 'GET', 'trends'        , '' ]
  - [ 'trends_current', 'GET', 'trends/current', '' ]
  - [ 'trends_daily'  , 'GET', 'trends/daily'  , '' ]
  - [ 'trends_weekly' , 'GET', 'trends/weekly' , '' ]