<s> import time <EOL> from datetime import datetime <EOL> import random <EOL> import urllib <EOL> from urlparse import urlparse , urlunparse , parse_qs , urlsplit , urlunsplit <EOL> from auth import Token , Consumer <EOL> from auth import to_utf8 , escape <EOL> from auth import SignatureMethod_HMAC_SHA1 <EOL> class CustomSignatureMethod_HMAC_SHA1 ( SignatureMethod_HMAC_SHA1 ) : <EOL> def signing_base ( self , request , consumer , token ) : <EOL> """<STR_LIT>""" <EOL> sig = ( <EOL> escape ( request . method ) , <EOL> escape ( OAuthHook . get_normalized_url ( request . url ) ) , <EOL> escape ( OAuthHook . get_normalized_parameters ( request ) ) , <EOL> ) <EOL> key = '<STR_LIT>' % escape ( consumer . secret ) <EOL> if token is not None : <EOL> key += escape ( token . secret ) <EOL> raw = '<STR_LIT:&>' . join ( sig ) <EOL> return key , raw <EOL> class OAuthHook ( object ) : <EOL> OAUTH_VERSION = '<STR_LIT:1.0>' <EOL> header_auth = False <EOL> signature = CustomSignatureMethod_HMAC_SHA1 ( ) <EOL> consumer_key = None <EOL> consumer_secret = None <EOL> def __init__ ( self , access_token = None , access_token_secret = None , consumer_key = None , consumer_secret = None , header_auth = None ) : <EOL> """<STR_LIT>""" <EOL> if access_token is not None : <EOL> self . token = Token ( access_token , access_token_secret ) <EOL> else : <EOL> self . token = None <EOL> if consumer_key is None and consumer_secret is None : <EOL> consumer_key = self . consumer_key <EOL> consumer_secret = self . consumer_secret <EOL> if header_auth is not None : <EOL> self . header_auth = header_auth <EOL> self . consumer = Consumer ( consumer_key , consumer_secret ) <EOL> @ staticmethod <EOL> def _split_url_string ( query_string ) : <EOL> """<STR_LIT>""" <EOL> parameters = parse_qs ( to_utf8 ( query_string ) , keep_blank_values = True ) <EOL> for k , v in parameters . iteritems ( ) : <EOL> parameters [ k ] = urllib . unquote ( v [ <NUM_LIT:0> ] ) <EOL> return parameters <EOL> @ staticmethod <EOL> def get_normalized_parameters ( request ) : <EOL> """<STR_LIT>""" <EOL> if ( '<STR_LIT:Content-Type>' not in request . headers or request . headers . get ( '<STR_LIT:Content-Type>' ) . startswith ( '<STR_LIT>' ) ) and not isinstance ( request . data , basestring ) : <EOL> data_and_params = dict ( request . data . items ( ) + request . params . items ( ) ) <EOL> for key , value in data_and_params . items ( ) : <EOL> request . data_and_params [ to_utf8 ( key ) ] = to_utf8 ( value ) <EOL> if request . data_and_params . has_key ( '<STR_LIT>' ) : <EOL> del request . data_and_params [ '<STR_LIT>' ] <EOL> items = [ ] <EOL> for key , value in request . data_and_params . iteritems ( ) : <EOL> if isinstance ( value , basestring ) : <EOL> items . append ( ( key , value ) ) <EOL> else : <EOL> try : <EOL> value = list ( value ) <EOL> except TypeError , e : <EOL> assert '<STR_LIT>' in str ( e ) <EOL> items . append ( ( key , value ) ) <EOL> else : <EOL> items . extend ( ( key , item ) for item in value ) <EOL> query_string = urlparse ( request . url ) [ <NUM_LIT:4> ] <EOL> items . extend ( [ ( to_utf8 ( k ) , to_utf8 ( v ) ) for k , v in OAuthHook . _split_url_string ( query_string ) . items ( ) ] ) <EOL> items . sort ( ) <EOL> return urllib . urlencode ( items ) . replace ( '<STR_LIT:+>' , '<STR_LIT>' ) . replace ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> @ staticmethod <EOL> def get_normalized_url ( url ) : <EOL> """<STR_LIT>""" <EOL> scheme , netloc , path , params , query , fragment = urlparse ( url ) <EOL> if scheme == '<STR_LIT:http>' and netloc [ - <NUM_LIT:3> : ] == '<STR_LIT>' : <EOL> netloc = netloc [ : - <NUM_LIT:3> ] <EOL> elif scheme == '<STR_LIT>' and netloc [ - <NUM_LIT:4> : ] == '<STR_LIT>' : <EOL> netloc = netloc [ : - <NUM_LIT:4> ] <EOL> if scheme not in ( '<STR_LIT:http>' , '<STR_LIT>' ) : <EOL> raise ValueError ( "<STR_LIT>" % ( url , scheme ) ) <EOL> return urlunparse ( ( scheme , netloc , path , None , None , None ) ) <EOL> @ staticmethod <EOL> def to_url ( request ) : <EOL> """<STR_LIT>""" <EOL> scheme , netloc , path , query , fragment = urlsplit ( to_utf8 ( request . url ) ) <EOL> query = parse_qs ( query ) <EOL> for key , value in request . data_and_params . iteritems ( ) : <EOL> query . setdefault ( key , [ ] ) . append ( value ) <EOL> query = urllib . urlencode ( query , True ) <EOL> return urlunsplit ( ( scheme , netloc , path , query , fragment ) ) <EOL> @ staticmethod <EOL> def to_postdata ( request ) : <EOL> """<STR_LIT>""" <EOL> return urllib . urlencode ( request . data_and_params , True ) . replace ( '<STR_LIT:+>' , '<STR_LIT>' ) <EOL> @ staticmethod <EOL> def authorization_header ( oauth_params ) : <EOL> """<STR_LIT>""" <EOL> authorization_headers = '<STR_LIT>' <EOL> authorization_headers += '<STR_LIT:U+002C>' . join ( [ '<STR_LIT>' . format ( k , urllib . quote ( str ( v ) ) ) <EOL> for k , v in oauth_params . items ( ) ] ) <EOL> return authorization_headers <EOL> def __call__ ( self , request ) : <EOL> """<STR_LIT>""" <EOL> if not request . params : <EOL> request . params = { } <EOL> if not request . data : <EOL> request . data = { } <EOL> if isinstance ( request . params , list ) : <EOL> request . params = dict ( request . params ) <EOL> if isinstance ( request . data , list ) : <EOL> request . data = dict ( request . data ) <EOL> request . oauth_params = { } <EOL> request . oauth_params [ '<STR_LIT>' ] = self . consumer . key <EOL> request . oauth_params [ '<STR_LIT>' ] = str ( int ( time . time ( ) ) ) <EOL> request . oauth_params [ '<STR_LIT>' ] = str ( random . randint ( <NUM_LIT:0> , <NUM_LIT> ) ) <EOL> request . oauth_params [ '<STR_LIT>' ] = self . OAUTH_VERSION <EOL> if self . token : <EOL> request . oauth_params [ '<STR_LIT>' ] = self . token . key <EOL> if '<STR_LIT>' in request . data : <EOL> request . oauth_params [ '<STR_LIT>' ] = request . data . pop ( '<STR_LIT>' ) <EOL> request . oauth_params [ '<STR_LIT>' ] = self . signature . name <EOL> if '<STR_LIT>' in request . data : <EOL> request . oauth_params [ '<STR_LIT>' ] = request . data . pop ( '<STR_LIT>' ) <EOL> if '<STR_LIT>' in request . params : <EOL> request . oauth_params [ '<STR_LIT>' ] = request . params . pop ( '<STR_LIT>' ) <EOL> request . data_and_params = request . oauth_params . copy ( ) <EOL> request . oauth_params [ '<STR_LIT>' ] = self . signature . sign ( request , self . consumer , self . token ) <EOL> request . data_and_params [ '<STR_LIT>' ] = request . oauth_params [ '<STR_LIT>' ] <EOL> if self . header_auth : <EOL> request . headers [ '<STR_LIT>' ] = self . authorization_header ( request . oauth_params ) <EOL> elif request . method in ( "<STR_LIT:GET>" , "<STR_LIT>" ) : <EOL> request . url = self . to_url ( request ) <EOL> elif ( '<STR_LIT:Content-Type>' not in request . headers or request . headers [ '<STR_LIT:Content-Type>' ] != '<STR_LIT>' ) and not isinstance ( request . data , basestring ) : <EOL> request . url = self . to_url ( request ) <EOL> request . data = { } <EOL> else : <EOL> request . data = request . data_and_params <EOL> return request </s>
