<s> """<STR_LIT>""" <EOL> import cgi <EOL> import codecs <EOL> import collections <EOL> import io <EOL> import os <EOL> import re <EOL> import socket <EOL> import struct <EOL> import warnings <EOL> from . import __version__ <EOL> from . import certs <EOL> from . compat import parse_http_list as _parse_list_header <EOL> from . compat import ( quote , urlparse , bytes , str , OrderedDict , unquote , is_py2 , <EOL> builtin_str , getproxies , proxy_bypass , urlunparse , <EOL> basestring ) <EOL> from . cookies import RequestsCookieJar , cookiejar_from_dict <EOL> from . structures import CaseInsensitiveDict <EOL> from . exceptions import InvalidURL , FileModeWarning <EOL> _hush_pyflakes = ( RequestsCookieJar , ) <EOL> NETRC_FILES = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> DEFAULT_CA_BUNDLE_PATH = certs . where ( ) <EOL> def dict_to_sequence ( d ) : <EOL> """<STR_LIT>""" <EOL> if hasattr ( d , '<STR_LIT>' ) : <EOL> d = d . items ( ) <EOL> return d <EOL> def super_len ( o ) : <EOL> total_length = <NUM_LIT:0> <EOL> current_position = <NUM_LIT:0> <EOL> if hasattr ( o , '<STR_LIT>' ) : <EOL> total_length = len ( o ) <EOL> elif hasattr ( o , '<STR_LIT>' ) : <EOL> total_length = o . len <EOL> elif hasattr ( o , '<STR_LIT>' ) : <EOL> total_length = len ( o . getvalue ( ) ) <EOL> elif hasattr ( o , '<STR_LIT>' ) : <EOL> try : <EOL> fileno = o . fileno ( ) <EOL> except io . UnsupportedOperation : <EOL> pass <EOL> else : <EOL> total_length = os . fstat ( fileno ) . st_size <EOL> if '<STR_LIT:b>' not in o . mode : <EOL> warnings . warn ( ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) , <EOL> FileModeWarning <EOL> ) <EOL> if hasattr ( o , '<STR_LIT>' ) : <EOL> try : <EOL> current_position = o . tell ( ) <EOL> except ( OSError , IOError ) : <EOL> current_position = total_length <EOL> return max ( <NUM_LIT:0> , total_length - current_position ) <EOL> def get_netrc_auth ( url , raise_errors = False ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> from netrc import netrc , NetrcParseError <EOL> netrc_path = None <EOL> for f in NETRC_FILES : <EOL> try : <EOL> loc = os . path . expanduser ( '<STR_LIT>' . format ( f ) ) <EOL> except KeyError : <EOL> return <EOL> if os . path . exists ( loc ) : <EOL> netrc_path = loc <EOL> break <EOL> if netrc_path is None : <EOL> return <EOL> ri = urlparse ( url ) <EOL> splitstr = b'<STR_LIT::>' <EOL> if isinstance ( url , str ) : <EOL> splitstr = splitstr . decode ( '<STR_LIT:ascii>' ) <EOL> host = ri . netloc . split ( splitstr ) [ <NUM_LIT:0> ] <EOL> try : <EOL> _netrc = netrc ( netrc_path ) . authenticators ( host ) <EOL> if _netrc : <EOL> login_i = ( <NUM_LIT:0> if _netrc [ <NUM_LIT:0> ] else <NUM_LIT:1> ) <EOL> return ( _netrc [ login_i ] , _netrc [ <NUM_LIT:2> ] ) <EOL> except ( NetrcParseError , IOError ) : <EOL> if raise_errors : <EOL> raise <EOL> except ( ImportError , AttributeError ) : <EOL> pass <EOL> def guess_filename ( obj ) : <EOL> """<STR_LIT>""" <EOL> name = getattr ( obj , '<STR_LIT:name>' , None ) <EOL> if ( name and isinstance ( name , basestring ) and name [ <NUM_LIT:0> ] != '<STR_LIT:<>' and <EOL> name [ - <NUM_LIT:1> ] != '<STR_LIT:>>' ) : <EOL> return os . path . basename ( name ) <EOL> def from_key_val_list ( value ) : <EOL> """<STR_LIT>""" <EOL> if value is None : <EOL> return None <EOL> if isinstance ( value , ( str , bytes , bool , int ) ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> return OrderedDict ( value ) <EOL> def to_key_val_list ( value ) : <EOL> """<STR_LIT>""" <EOL> if value is None : <EOL> return None <EOL> if isinstance ( value , ( str , bytes , bool , int ) ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if isinstance ( value , collections . Mapping ) : <EOL> value = value . items ( ) <EOL> return list ( value ) <EOL> def parse_list_header ( value ) : <EOL> """<STR_LIT>""" <EOL> result = [ ] <EOL> for item in _parse_list_header ( value ) : <EOL> if item [ : <NUM_LIT:1> ] == item [ - <NUM_LIT:1> : ] == '<STR_LIT:">' : <EOL> item = unquote_header_value ( item [ <NUM_LIT:1> : - <NUM_LIT:1> ] ) <EOL> result . append ( item ) <EOL> return result <EOL> def parse_dict_header ( value ) : <EOL> """<STR_LIT>""" <EOL> result = { } <EOL> for item in _parse_list_header ( value ) : <EOL> if '<STR_LIT:=>' not in item : <EOL> result [ item ] = None <EOL> continue <EOL> name , value = item . split ( '<STR_LIT:=>' , <NUM_LIT:1> ) <EOL> if value [ : <NUM_LIT:1> ] == value [ - <NUM_LIT:1> : ] == '<STR_LIT:">' : <EOL> value = unquote_header_value ( value [ <NUM_LIT:1> : - <NUM_LIT:1> ] ) <EOL> result [ name ] = value <EOL> return result <EOL> def unquote_header_value ( value , is_filename = False ) : <EOL> r"""<STR_LIT>""" <EOL> if value and value [ <NUM_LIT:0> ] == value [ - <NUM_LIT:1> ] == '<STR_LIT:">' : <EOL> value = value [ <NUM_LIT:1> : - <NUM_LIT:1> ] <EOL> if not is_filename or value [ : <NUM_LIT:2> ] != '<STR_LIT>' : <EOL> return value . replace ( '<STR_LIT>' , '<STR_LIT:\\>' ) . replace ( '<STR_LIT>' , '<STR_LIT:">' ) <EOL> return value <EOL> def dict_from_cookiejar ( cj ) : <EOL> """<STR_LIT>""" <EOL> cookie_dict = { } <EOL> for cookie in cj : <EOL> cookie_dict [ cookie . name ] = cookie . value <EOL> return cookie_dict <EOL> def add_dict_to_cookiejar ( cj , cookie_dict ) : <EOL> """<STR_LIT>""" <EOL> cj2 = cookiejar_from_dict ( cookie_dict ) <EOL> cj . update ( cj2 ) <EOL> return cj <EOL> def get_encodings_from_content ( content ) : <EOL> """<STR_LIT>""" <EOL> warnings . warn ( ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> DeprecationWarning ) <EOL> charset_re = re . compile ( r'<STR_LIT>' , flags = re . I ) <EOL> pragma_re = re . compile ( r'<STR_LIT>' , flags = re . I ) <EOL> xml_re = re . compile ( r'<STR_LIT>' ) <EOL> return ( charset_re . findall ( content ) + <EOL> pragma_re . findall ( content ) + <EOL> xml_re . findall ( content ) ) <EOL> def get_encoding_from_headers ( headers ) : <EOL> """<STR_LIT>""" <EOL> content_type = headers . get ( '<STR_LIT>' ) <EOL> if not content_type : <EOL> return None <EOL> content_type , params = cgi . parse_header ( content_type ) <EOL> if '<STR_LIT>' in params : <EOL> return params [ '<STR_LIT>' ] . strip ( "<STR_LIT>" ) <EOL> if '<STR_LIT:text>' in content_type : <EOL> return '<STR_LIT>' <EOL> def stream_decode_response_unicode ( iterator , r ) : <EOL> """<STR_LIT>""" <EOL> if r . encoding is None : <EOL> for item in iterator : <EOL> yield item <EOL> return <EOL> decoder = codecs . getincrementaldecoder ( r . encoding ) ( errors = '<STR_LIT:replace>' ) <EOL> for chunk in iterator : <EOL> rv = decoder . decode ( chunk ) <EOL> if rv : <EOL> yield rv <EOL> rv = decoder . decode ( b'<STR_LIT>' , final = True ) <EOL> if rv : <EOL> yield rv <EOL> def iter_slices ( string , slice_length ) : <EOL> """<STR_LIT>""" <EOL> pos = <NUM_LIT:0> <EOL> while pos < len ( string ) : <EOL> yield string [ pos : pos + slice_length ] <EOL> pos += slice_length <EOL> def get_unicode_from_response ( r ) : <EOL> """<STR_LIT>""" <EOL> warnings . warn ( ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> DeprecationWarning ) <EOL> tried_encodings = [ ] <EOL> encoding = get_encoding_from_headers ( r . headers ) <EOL> if encoding : <EOL> try : <EOL> return str ( r . content , encoding ) <EOL> except UnicodeError : <EOL> tried_encodings . append ( encoding ) <EOL> try : <EOL> return str ( r . content , encoding , errors = '<STR_LIT:replace>' ) <EOL> except TypeError : <EOL> return r . content <EOL> UNRESERVED_SET = frozenset ( <EOL> "<STR_LIT>" <EOL> + "<STR_LIT>" ) <EOL> def unquote_unreserved ( uri ) : <EOL> """<STR_LIT>""" <EOL> parts = uri . split ( '<STR_LIT:%>' ) <EOL> for i in range ( <NUM_LIT:1> , len ( parts ) ) : <EOL> h = parts [ i ] [ <NUM_LIT:0> : <NUM_LIT:2> ] <EOL> if len ( h ) == <NUM_LIT:2> and h . isalnum ( ) : <EOL> try : <EOL> c = chr ( int ( h , <NUM_LIT:16> ) ) <EOL> except ValueError : <EOL> raise InvalidURL ( "<STR_LIT>" % h ) <EOL> if c in UNRESERVED_SET : <EOL> parts [ i ] = c + parts [ i ] [ <NUM_LIT:2> : ] <EOL> else : <EOL> parts [ i ] = '<STR_LIT:%>' + parts [ i ] <EOL> else : <EOL> parts [ i ] = '<STR_LIT:%>' + parts [ i ] <EOL> return '<STR_LIT>' . join ( parts ) <EOL> def requote_uri ( uri ) : <EOL> """<STR_LIT>""" <EOL> safe_with_percent = "<STR_LIT>" <EOL> safe_without_percent = "<STR_LIT>" <EOL> try : <EOL> return quote ( unquote_unreserved ( uri ) , safe = safe_with_percent ) <EOL> except InvalidURL : <EOL> return quote ( uri , safe = safe_without_percent ) <EOL> def address_in_network ( ip , net ) : <EOL> """<STR_LIT>""" <EOL> ipaddr = struct . unpack ( '<STR_LIT>' , socket . inet_aton ( ip ) ) [ <NUM_LIT:0> ] <EOL> netaddr , bits = net . split ( '<STR_LIT:/>' ) <EOL> netmask = struct . unpack ( '<STR_LIT>' , socket . inet_aton ( dotted_netmask ( int ( bits ) ) ) ) [ <NUM_LIT:0> ] <EOL> network = struct . unpack ( '<STR_LIT>' , socket . inet_aton ( netaddr ) ) [ <NUM_LIT:0> ] & netmask <EOL> return ( ipaddr & netmask ) == ( network & netmask ) <EOL> def dotted_netmask ( mask ) : <EOL> """<STR_LIT>""" <EOL> bits = <NUM_LIT> ^ ( <NUM_LIT:1> << <NUM_LIT:32> - mask ) - <NUM_LIT:1> <EOL> return socket . inet_ntoa ( struct . pack ( '<STR_LIT>' , bits ) ) <EOL> def is_ipv4_address ( string_ip ) : <EOL> try : <EOL> socket . inet_aton ( string_ip ) <EOL> except socket . error : <EOL> return False <EOL> return True <EOL> def is_valid_cidr ( string_network ) : <EOL> """<STR_LIT>""" <EOL> if string_network . count ( '<STR_LIT:/>' ) == <NUM_LIT:1> : <EOL> try : <EOL> mask = int ( string_network . split ( '<STR_LIT:/>' ) [ <NUM_LIT:1> ] ) <EOL> except ValueError : <EOL> return False <EOL> if mask < <NUM_LIT:1> or mask > <NUM_LIT:32> : <EOL> return False <EOL> try : <EOL> socket . inet_aton ( string_network . split ( '<STR_LIT:/>' ) [ <NUM_LIT:0> ] ) <EOL> except socket . error : <EOL> return False <EOL> else : <EOL> return False <EOL> return True <EOL> def should_bypass_proxies ( url ) : <EOL> """<STR_LIT>""" <EOL> get_proxy = lambda k : os . environ . get ( k ) or os . environ . get ( k . upper ( ) ) <EOL> no_proxy = get_proxy ( '<STR_LIT>' ) <EOL> netloc = urlparse ( url ) . netloc <EOL> if no_proxy : <EOL> no_proxy = ( <EOL> host for host in no_proxy . replace ( '<STR_LIT:U+0020>' , '<STR_LIT>' ) . split ( '<STR_LIT:U+002C>' ) if host <EOL> ) <EOL> ip = netloc . split ( '<STR_LIT::>' ) [ <NUM_LIT:0> ] <EOL> if is_ipv4_address ( ip ) : <EOL> for proxy_ip in no_proxy : <EOL> if is_valid_cidr ( proxy_ip ) : <EOL> if address_in_network ( ip , proxy_ip ) : <EOL> return True <EOL> else : <EOL> for host in no_proxy : <EOL> if netloc . endswith ( host ) or netloc . split ( '<STR_LIT::>' ) [ <NUM_LIT:0> ] . endswith ( host ) : <EOL> return True <EOL> try : <EOL> bypass = proxy_bypass ( netloc ) <EOL> except ( TypeError , socket . gaierror ) : <EOL> bypass = False <EOL> if bypass : <EOL> return True <EOL> return False <EOL> def get_environ_proxies ( url ) : <EOL> """<STR_LIT>""" <EOL> if should_bypass_proxies ( url ) : <EOL> return { } <EOL> else : <EOL> return getproxies ( ) <EOL> def select_proxy ( url , proxies ) : <EOL> """<STR_LIT>""" <EOL> proxies = proxies or { } <EOL> urlparts = urlparse ( url ) <EOL> if urlparts . hostname is None : <EOL> proxy = None <EOL> else : <EOL> proxy = proxies . get ( urlparts . scheme + '<STR_LIT>' + urlparts . hostname ) <EOL> if proxy is None : <EOL> proxy = proxies . get ( urlparts . scheme ) <EOL> return proxy <EOL> def default_user_agent ( name = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT>' % ( name , __version__ ) <EOL> def default_headers ( ) : <EOL> return CaseInsensitiveDict ( { <EOL> '<STR_LIT>' : default_user_agent ( ) , <EOL> '<STR_LIT>' : '<STR_LIT:U+002CU+0020>' . join ( ( '<STR_LIT>' , '<STR_LIT>' ) ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> def parse_header_links ( value ) : <EOL> """<STR_LIT>""" <EOL> links = [ ] <EOL> replace_chars = '<STR_LIT>' <EOL> for val in re . split ( '<STR_LIT>' , value ) : <EOL> try : <EOL> url , params = val . split ( '<STR_LIT:;>' , <NUM_LIT:1> ) <EOL> except ValueError : <EOL> url , params = val , '<STR_LIT>' <EOL> link = { '<STR_LIT:url>' : url . strip ( '<STR_LIT>' ) } <EOL> for param in params . split ( '<STR_LIT:;>' ) : <EOL> try : <EOL> key , value = param . split ( '<STR_LIT:=>' ) <EOL> except ValueError : <EOL> break <EOL> link [ key . strip ( replace_chars ) ] = value . strip ( replace_chars ) <EOL> links . append ( link ) <EOL> return links <EOL> _null = '<STR_LIT:\x00>' . encode ( '<STR_LIT:ascii>' ) <EOL> _null2 = _null * <NUM_LIT:2> <EOL> _null3 = _null * <NUM_LIT:3> <EOL> def guess_json_utf ( data ) : <EOL> sample = data [ : <NUM_LIT:4> ] <EOL> if sample in ( codecs . BOM_UTF32_LE , codecs . BOM32_BE ) : <EOL> return '<STR_LIT>' <EOL> if sample [ : <NUM_LIT:3> ] == codecs . BOM_UTF8 : <EOL> return '<STR_LIT>' <EOL> if sample [ : <NUM_LIT:2> ] in ( codecs . BOM_UTF16_LE , codecs . BOM_UTF16_BE ) : <EOL> return '<STR_LIT>' <EOL> nullcount = sample . count ( _null ) <EOL> if nullcount == <NUM_LIT:0> : <EOL> return '<STR_LIT:utf-8>' <EOL> if nullcount == <NUM_LIT:2> : <EOL> if sample [ : : <NUM_LIT:2> ] == _null2 : <EOL> return '<STR_LIT>' <EOL> if sample [ <NUM_LIT:1> : : <NUM_LIT:2> ] == _null2 : <EOL> return '<STR_LIT>' <EOL> if nullcount == <NUM_LIT:3> : <EOL> if sample [ : <NUM_LIT:3> ] == _null3 : <EOL> return '<STR_LIT>' <EOL> if sample [ <NUM_LIT:1> : ] == _null3 : <EOL> return '<STR_LIT>' <EOL> return None <EOL> def prepend_scheme_if_needed ( url , new_scheme ) : <EOL> """<STR_LIT>""" <EOL> scheme , netloc , path , params , query , fragment = urlparse ( url , new_scheme ) <EOL> if not netloc : <EOL> netloc , path = path , netloc <EOL> return urlunparse ( ( scheme , netloc , path , params , query , fragment ) ) <EOL> def get_auth_from_url ( url ) : <EOL> """<STR_LIT>""" <EOL> parsed = urlparse ( url ) <EOL> try : <EOL> auth = ( unquote ( parsed . username ) , unquote ( parsed . password ) ) <EOL> except ( AttributeError , TypeError ) : <EOL> auth = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> return auth <EOL> def to_native_string ( string , encoding = '<STR_LIT:ascii>' ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( string , builtin_str ) : <EOL> out = string <EOL> else : <EOL> if is_py2 : <EOL> out = string . encode ( encoding ) <EOL> else : <EOL> out = string . decode ( encoding ) <EOL> return out <EOL> def urldefragauth ( url ) : <EOL> """<STR_LIT>""" <EOL> scheme , netloc , path , params , query , fragment = urlparse ( url ) <EOL> if not netloc : <EOL> netloc , path = path , netloc <EOL> netloc = netloc . rsplit ( '<STR_LIT:@>' , <NUM_LIT:1> ) [ - <NUM_LIT:1> ] <EOL> return urlunparse ( ( scheme , netloc , path , params , query , '<STR_LIT>' ) ) </s>
