(url: str, cache_dir: str = None, num_attempts: int = 10, verbose: bool = True, return_filename: bool = False, cache: bool = True)
return io.BytesIO(url_data)
Download the given URL and return a binary-mode file object to access the data.
Download the given URL and return a binary-mode file object to access the data.
Download the given URL and return a binary - mode file object to access the data .
def open_url(url: str, cache_dir: str = None, num_attempts: int = 10, verbose: bool = True, return_filename: bool = False, cache: bool = True) -> Any:
    """Download the given URL and return a binary-mode file object to access the data."""
    assert num_attempts >= 1
    assert not (return_filename and (not cache))

    # Doesn't look like an URL scheme so interpret it as a local filename.
    if not re.match('^[a-z]+://', url):
        return url if return_filename else open(url, "rb")

    # Handle file URLs.  This code handles unusual file:// patterns that
    # arise on Windows:
    #
    # file:///c:/foo.txt
    #
    # which would translate to a local '/c:/foo.txt' filename that's
    # invalid.  Drop the forward slash for such pathnames.
    #
    # If you touch this code path, you should test it on both Linux and
    # Windows.
    #
    # Some internet resources suggest using urllib.request.url2pathname() but
    # but that converts forward slashes to backslashes and this causes
    # its own set of problems.
    if url.startswith('file://'):
        filename = urllib.parse.urlparse(url).path
        if re.match(r'^/[a-zA-Z]:', filename):
            filename = filename[1:]
        return filename if return_filename else open(filename, "rb")

    assert is_url(url)

    # Lookup from cache.
    if cache_dir is None:
        cache_dir = make_cache_dir_path('downloads')

    url_md5 = hashlib.md5(url.encode("utf-8")).hexdigest()
    if cache:
        cache_dir = os.path.join(cache_dir, url_md5)
        cache_files = glob.glob(os.path.join(cache_dir, "*"))
        if len(cache_files) == 1:
            filename = cache_files[0]
            return filename if return_filename else open(filename, "rb")

    # Download.
    url_name = None
    url_data = None
    with requests.Session() as session:
        if verbose:
            print("Downloading %s ..." % url, end="", flush=True)
        for attempts_left in reversed(range(num_attempts)):
            try:
                with session.get(url) as res:
                    res.raise_for_status()
                    if len(res.content) == 0:
                        raise IOError("No data received")

                    if len(res.content) < 8192:
                        content_str = res.content.decode("utf-8")
                        if "download_warning" in res.headers.get("Set-Cookie", ""):
                            links = [html.unescape(link) for link in content_str.split('"') if "export=download" in link]
                            if len(links) == 1:
                                url = requests.compat.urljoin(url, links[0])
                                raise IOError("Google Drive virus checker nag")
                        if "Google Drive - Quota exceeded" in content_str:
                            raise IOError("Google Drive download quota exceeded -- please try again later")

                    match = re.search(r'filename="([^"]*)"', res.headers.get("Content-Disposition", ""))
                    url_name = match[1] if match else url
                    url_data = res.content
                    if verbose:
                        print(" done")
                    break
            except KeyboardInterrupt:
                raise
            except:
                if not attempts_left:
                    if verbose:
                        print(" failed")
                    raise
                if verbose:
                    print(".", end="", flush=True)

    # Save to cache.
    if cache:
        filename = os.path.basename(url_name)
        cache_file = os.path.join(cache_dir, filename)
        temp_file = os.path.join(cache_dir, "tmp_" + uuid.uuid4().hex + "_" + filename)
        os.makedirs(cache_dir, exist_ok=True)
        with open(temp_file, "wb") as f:
            f.write(url_data)
        os.replace(temp_file, cache_file) # atomic
        if return_filename:
            return cache_file

    # Return data as file object.
    assert not return_filename
    return io.BytesIO(url_data)
def open_url ( url : str , cache_dir : str = None , num_attempts : int = 10 , verbose : bool = True , return_filename : bool = False , cache : bool = True ) -> Any : assert num_attempts >= 1 assert not ( return_filename and ( not cache ) ) # Doesn't look like an URL scheme so interpret it as a local filename. if not re . match ( '^[a-z]+://' , url ) : return url if return_filename else open ( url , "rb" ) # Handle file URLs.  This code handles unusual file:// patterns that # arise on Windows: # # file:///c:/foo.txt # # which would translate to a local '/c:/foo.txt' filename that's # invalid.  Drop the forward slash for such pathnames. # # If you touch this code path, you should test it on both Linux and # Windows. # # Some internet resources suggest using urllib.request.url2pathname() but # but that converts forward slashes to backslashes and this causes # its own set of problems. if url . startswith ( 'file://' ) : filename = urllib . parse . urlparse ( url ) . path if re . match ( r'^/[a-zA-Z]:' , filename ) : filename = filename [ 1 : ] return filename if return_filename else open ( filename , "rb" ) assert is_url ( url ) # Lookup from cache. if cache_dir is None : cache_dir = make_cache_dir_path ( 'downloads' ) url_md5 = hashlib . md5 ( url . encode ( "utf-8" ) ) . hexdigest ( ) if cache : cache_dir = os . path . join ( cache_dir , url_md5 ) cache_files = glob . glob ( os . path . join ( cache_dir , "*" ) ) if len ( cache_files ) == 1 : filename = cache_files [ 0 ] return filename if return_filename else open ( filename , "rb" ) # Download. url_name = None url_data = None with requests . Session ( ) as session : if verbose : print ( "Downloading %s ..." % url , end = "" , flush = True ) for attempts_left in reversed ( range ( num_attempts ) ) : try : with session . get ( url ) as res : res . raise_for_status ( ) if len ( res . content ) == 0 : raise IOError ( "No data received" ) if len ( res . content ) < 8192 : content_str = res . content . decode ( "utf-8" ) if "download_warning" in res . headers . get ( "Set-Cookie" , "" ) : links = [ html . unescape ( link ) for link in content_str . split ( '"' ) if "export=download" in link ] if len ( links ) == 1 : url = requests . compat . urljoin ( url , links [ 0 ] ) raise IOError ( "Google Drive virus checker nag" ) if "Google Drive - Quota exceeded" in content_str : raise IOError ( "Google Drive download quota exceeded -- please try again later" ) match = re . search ( r'filename="([^"]*)"' , res . headers . get ( "Content-Disposition" , "" ) ) url_name = match [ 1 ] if match else url url_data = res . content if verbose : print ( " done" ) break except KeyboardInterrupt : raise except : if not attempts_left : if verbose : print ( " failed" ) raise if verbose : print ( "." , end = "" , flush = True ) # Save to cache. if cache : filename = os . path . basename ( url_name ) cache_file = os . path . join ( cache_dir , filename ) temp_file = os . path . join ( cache_dir , "tmp_" + uuid . uuid4 ( ) . hex + "_" + filename ) os . makedirs ( cache_dir , exist_ok = True ) with open ( temp_file , "wb" ) as f : f . write ( url_data ) os . replace ( temp_file , cache_file ) # atomic if return_filename : return cache_file # Return data as file object. assert not return_filename return io . BytesIO ( url_data )