<s> import re <EOL> import sys <EOL> import unicodedata <EOL> import collections <EOL> import functools <EOL> import logging <EOL> import six <EOL> import requests <EOL> import social <EOL> from requests . adapters import HTTPAdapter <EOL> from requests . packages . urllib3 . poolmanager import PoolManager <EOL> from social . exceptions import AuthCanceled , AuthUnreachableProvider <EOL> from social . p3 import urlparse , urlunparse , urlencode , parse_qs as battery_parse_qs <EOL> SETTING_PREFIX = '<STR_LIT>' <EOL> social_logger = logging . getLogger ( '<STR_LIT>' ) <EOL> class SSLHttpAdapter ( HTTPAdapter ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , ssl_protocol ) : <EOL> self . ssl_protocol = ssl_protocol <EOL> super ( SSLHttpAdapter , self ) . __init__ ( ) <EOL> def init_poolmanager ( self , connections , maxsize , block = False ) : <EOL> self . poolmanager = PoolManager ( <EOL> num_pools = connections , <EOL> maxsize = maxsize , <EOL> block = block , <EOL> ssl_version = self . ssl_protocol <EOL> ) <EOL> @ classmethod <EOL> def ssl_adapter_session ( cls , ssl_protocol ) : <EOL> session = requests . Session ( ) <EOL> session . mount ( '<STR_LIT>' , SSLHttpAdapter ( ssl_protocol ) ) <EOL> return session <EOL> def import_module ( name ) : <EOL> __import__ ( name ) <EOL> return sys . modules [ name ] <EOL> def module_member ( name ) : <EOL> mod , member = name . rsplit ( '<STR_LIT:.>' , <NUM_LIT:1> ) <EOL> module = import_module ( mod ) <EOL> return getattr ( module , member ) <EOL> def user_agent ( ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT>' + social . __version__ <EOL> def url_add_parameters ( url , params ) : <EOL> """<STR_LIT>""" <EOL> if params : <EOL> fragments = list ( urlparse ( url ) ) <EOL> value = parse_qs ( fragments [ <NUM_LIT:4> ] ) <EOL> value . update ( params ) <EOL> fragments [ <NUM_LIT:4> ] = urlencode ( value ) <EOL> url = urlunparse ( fragments ) <EOL> return url <EOL> def to_setting_name ( * names ) : <EOL> return '<STR_LIT:_>' . join ( [ name . upper ( ) . replace ( '<STR_LIT:->' , '<STR_LIT:_>' ) for name in names if name ] ) <EOL> def setting_name ( * names ) : <EOL> return to_setting_name ( * ( ( SETTING_PREFIX , ) + names ) ) <EOL> def sanitize_redirect ( host , redirect_to ) : <EOL> """<STR_LIT>""" <EOL> if redirect_to : <EOL> try : <EOL> netloc = urlparse ( redirect_to ) [ <NUM_LIT:1> ] or host <EOL> except ( TypeError , AttributeError ) : <EOL> pass <EOL> else : <EOL> if netloc == host : <EOL> return redirect_to <EOL> def user_is_authenticated ( user ) : <EOL> if user and hasattr ( user , '<STR_LIT>' ) : <EOL> if isinstance ( user . is_authenticated , collections . Callable ) : <EOL> authenticated = user . is_authenticated ( ) <EOL> else : <EOL> authenticated = user . is_authenticated <EOL> elif user : <EOL> authenticated = True <EOL> else : <EOL> authenticated = False <EOL> return authenticated <EOL> def user_is_active ( user ) : <EOL> if user and hasattr ( user , '<STR_LIT>' ) : <EOL> if isinstance ( user . is_active , collections . Callable ) : <EOL> is_active = user . is_active ( ) <EOL> else : <EOL> is_active = user . is_active <EOL> elif user : <EOL> is_active = True <EOL> else : <EOL> is_active = False <EOL> return is_active <EOL> def slugify ( value ) : <EOL> """<STR_LIT>""" <EOL> value = unicodedata . normalize ( '<STR_LIT>' , value ) . encode ( '<STR_LIT:ascii>' , '<STR_LIT:ignore>' ) . decode ( '<STR_LIT:ascii>' ) <EOL> value = re . sub ( '<STR_LIT>' , '<STR_LIT>' , value ) . strip ( ) . lower ( ) <EOL> return re . sub ( '<STR_LIT>' , '<STR_LIT:->' , value ) <EOL> def first ( func , items ) : <EOL> """<STR_LIT>""" <EOL> for item in items : <EOL> if func ( item ) : <EOL> return item <EOL> def parse_qs ( value ) : <EOL> """<STR_LIT>""" <EOL> return drop_lists ( battery_parse_qs ( value ) ) <EOL> def drop_lists ( value ) : <EOL> out = { } <EOL> for key , val in value . items ( ) : <EOL> val = val [ <NUM_LIT:0> ] <EOL> if isinstance ( key , six . binary_type ) : <EOL> key = six . text_type ( key , '<STR_LIT:utf-8>' ) <EOL> if isinstance ( val , six . binary_type ) : <EOL> val = six . text_type ( val , '<STR_LIT:utf-8>' ) <EOL> out [ key ] = val <EOL> return out <EOL> def partial_pipeline_data ( backend , user = None , * args , ** kwargs ) : <EOL> partial = backend . strategy . session_get ( '<STR_LIT>' , None ) <EOL> if partial : <EOL> idx , backend_name , xargs , xkwargs = backend . strategy . partial_from_session ( partial ) <EOL> if backend_name == backend . name : <EOL> kwargs . setdefault ( '<STR_LIT>' , idx ) <EOL> if user : <EOL> kwargs . setdefault ( '<STR_LIT:user>' , user ) <EOL> kwargs . setdefault ( '<STR_LIT>' , backend . strategy . request_data ( ) ) <EOL> xkwargs . update ( kwargs ) <EOL> return xargs , xkwargs <EOL> else : <EOL> backend . strategy . clean_partial_pipeline ( ) <EOL> def build_absolute_uri ( host_url , path = None ) : <EOL> """<STR_LIT>""" <EOL> path = path or '<STR_LIT>' <EOL> if path . startswith ( '<STR_LIT>' ) or path . startswith ( '<STR_LIT>' ) : <EOL> return path <EOL> if host_url . endswith ( '<STR_LIT:/>' ) and path . startswith ( '<STR_LIT:/>' ) : <EOL> path = path [ <NUM_LIT:1> : ] <EOL> return host_url + path <EOL> def constant_time_compare ( val1 , val2 ) : <EOL> """<STR_LIT>""" <EOL> if len ( val1 ) != len ( val2 ) : <EOL> return False <EOL> result = <NUM_LIT:0> <EOL> if six . PY3 and isinstance ( val1 , bytes ) and isinstance ( val2 , bytes ) : <EOL> for x , y in zip ( val1 , val2 ) : <EOL> result |= x ^ y <EOL> else : <EOL> for x , y in zip ( val1 , val2 ) : <EOL> result |= ord ( x ) ^ ord ( y ) <EOL> return result == <NUM_LIT:0> <EOL> def is_url ( value ) : <EOL> return value and ( value . startswith ( '<STR_LIT>' ) or <EOL> value . startswith ( '<STR_LIT>' ) or <EOL> value . startswith ( '<STR_LIT:/>' ) ) <EOL> def setting_url ( backend , * names ) : <EOL> for name in names : <EOL> if is_url ( name ) : <EOL> return name <EOL> else : <EOL> value = backend . setting ( name ) <EOL> if is_url ( value ) : <EOL> return value <EOL> def handle_http_errors ( func ) : <EOL> @ functools . wraps ( func ) <EOL> def wrapper ( * args , ** kwargs ) : <EOL> try : <EOL> return func ( * args , ** kwargs ) <EOL> except requests . HTTPError as err : <EOL> if err . response . status_code == <NUM_LIT> : <EOL> raise AuthCanceled ( args [ <NUM_LIT:0> ] , response = err . response ) <EOL> elif err . response . status_code == <NUM_LIT> : <EOL> raise AuthUnreachableProvider ( args [ <NUM_LIT:0> ] ) <EOL> else : <EOL> raise <EOL> return wrapper <EOL> def append_slash ( url ) : <EOL> """<STR_LIT>""" <EOL> if url and not url . endswith ( '<STR_LIT:/>' ) : <EOL> url = '<STR_LIT>' . format ( url ) <EOL> return url </s>
