我们从Python开源项目中，提取了以下9个代码示例，用于说明如何使用six.moves.urllib()。
def query(self, method, values={}): if method in self.public: url = 'https://bittrex.com/api/v1.1/public/' elif method in self.market: url = 'https://bittrex.com/api/v1.1/market/' elif method in self.account: url = 'https://bittrex.com/api/v1.1/account/' else: return 'Something went wrong, sorry.' url += method + '?' + urllib.parse.urlencode(values) if method not in self.public: url += '&apikey=' + self.key url += '&nonce=' + str(int(time.time())) signature = hmac.new(self.secret.encode('utf-8'), url.encode('utf-8'), hashlib.sha512).hexdigest() headers = {'apisign': signature} else: headers = {} req = urllib.request.Request(url, headers=headers) response = json.loads(urlopen( req, context=ssl._create_unverified_context()).read()) if response["result"]: return response["result"] else: return response["message"]
def download(path): """Use urllib to download a file. Parameters ---------- path : str Url to download Returns ------- path : str Location of downloaded file. """ from six.moves import urllib from IPython.core.display import clear_output print('Downloading ' + path) def progress(count, block_size, total_size): print('Downloaded %02.02f/%02.02f MB' % ( count * block_size / 1024.0 / 1024.0, total_size / 1024.0 / 1024.0)) clear_output(wait=True) filepath, _ = urllib.request.urlretrieve(path, reporthook=progress) return filepath
def get_celeb_files(dst='img_align_celeba', max_images=100): """Download the first 100 images of the celeb dataset. Files will be placed in a directory 'img_align_celeba' if one doesn't exist. Returns ------- files : list of strings Locations to the first 100 images of the celeb net dataset. """ # Create a directory if not os.path.exists(dst): os.mkdir(dst) # Now perform the following 100 times: for img_i in range(1, max_images + 1): # create a string using the current loop counter f = '000%03d.jpg' % img_i if not os.path.exists(os.path.join(dst, f)): # and get the url with that string appended the end url = 'https://s3.amazonaws.com/cadl/celeb-align/' + f # We'll print this out to the console so we can see how far we've gone print(url, end='\r') # And now download the url to a location inside our new directory urllib.request.urlretrieve(url, os.path.join(dst, f)) files = [os.path.join(dst, file_i) for file_i in os.listdir(dst) if '.jpg' in file_i][:max_images] return files
def download(path): """Use urllib to download a file. Parameters ---------- path : str Url to download Returns ------- path : str Location of downloaded file. """ import os from six.moves import urllib fname = path.split('/')[-1] if os.path.exists(fname): return fname print('Downloading ' + path) def progress(count, block_size, total_size): if count % 20 == 0: print('Downloaded %02.02f/%02.02f MB' % ( count * block_size / 1024.0 / 1024.0, total_size / 1024.0 / 1024.0), end='\r') filepath, _ = urllib.request.urlretrieve( path, filename=fname, reporthook=progress) return filepath
def query(self, method, req={}): if method in self.public: url = 'https://poloniex.com/public?command=' + method + '&' + \ urllib.parse.urlencode(req) headers = {} post_data = None elif method in self.trading: url = 'https://poloniex.com/tradingApi' req['command'] = method req['nonce'] = int(time.time() * 1000) post_data = urllib.parse.urlencode(req) signature = hmac.new(self.secret.encode('utf-8'), post_data.encode('utf-8'), hashlib.sha512).hexdigest() headers = {'Sign': signature, 'Key': self.key} post_data = post_data.encode('utf-8') else: raise ValueError( 'Method "' + method + '" not found in neither the Public API ' 'or Trading API endpoints' ) self.ask_request() req = urllib.request.Request( url, data=post_data, headers=headers, ) resource = urlopen(req, context=ssl._create_unverified_context()) content = resource.read().decode('utf-8') return json.loads(content)
def create_url(hostname, port=None, isSecure=False, path=None, params=None): """ Create a WebSocket URL from components. :param hostname: WebSocket server hostname. :type hostname: str :param port: WebSocket service port or None (to select default ports 80/443 depending on isSecure). :type port: int :param isSecure: Set True for secure WebSocket ("wss" scheme). :type isSecure: bool :param path: Path component of addressed resource (will be properly URL escaped). :type path: str :param params: A dictionary of key-values to construct the query component of the addressed resource (will be properly URL escaped). :type params: dict :returns: str -- Constructed WebSocket URL. """ if port is not None: netloc = "%s:%d" % (hostname, port) else: if isSecure: netloc = "%s:443" % hostname else: netloc = "%s:80" % hostname if isSecure: scheme = "wss" else: scheme = "ws" if path is not None: ppath = urllib.parse.quote(path) else: ppath = "/" if params is not None: query = urllib.parse.urlencode(params) else: query = None return urllib.parse.urlunparse((scheme, netloc, ppath, None, query, None))
def parse_url(url): """ Parses as WebSocket URL into it's components and returns a tuple (isSecure, host, port, resource, path, params). - ``isSecure`` is a flag which is True for wss URLs. - ``host`` is the hostname or IP from the URL. - ``port`` is the port from the URL or standard port derived from scheme (ws = 80, wss = 443). - ``resource`` is the /resource name/ from the URL, the /path/ together with the (optional) /query/ component. - ``path`` is the /path/ component properly unescaped. - ``params`` is the /query/ component properly unescaped and returned as dictionary. :param url: A valid WebSocket URL, i.e. ``ws://localhost:9000/myresource?param1=23¶m2=456`` :type url: str :returns: tuple -- A tuple (isSecure, host, port, resource, path, params) """ parsed = urlparse.urlparse(url) if parsed.scheme not in ["ws", "wss"]: raise Exception("invalid WebSocket URL: protocol scheme '{}' is not for WebSocket".format(parsed.scheme)) if not parsed.hostname or parsed.hostname == "": raise Exception("invalid WebSocket URL: missing hostname") if parsed.port is None or parsed.port == "": if parsed.scheme == "ws": port = 80 else: port = 443 else: port = int(parsed.port) if parsed.fragment is not None and parsed.fragment != "": raise Exception("invalid WebSocket URL: non-empty fragment '%s" % parsed.fragment) if parsed.path is not None and parsed.path != "": ppath = parsed.path path = urllib.parse.unquote(ppath) else: ppath = "/" path = ppath if parsed.query is not None and parsed.query != "": resource = ppath + "?" + parsed.query params = urlparse.parse_qs(parsed.query) else: resource = ppath params = {} return parsed.scheme == "wss", parsed.hostname, port, resource, path, params
