2
\$\begingroup\$

This is a class that downloads files in a smart way: it tries to download from different sessions. This do speed up things. How can I Improve it?

import os
import urllib2
import time
import multiprocessing
import string, random
import socket
from ctypes import c_int

import dummy
from useful_util import retry
import config
from logger import log

"Smart Downloading Module."

shared_bytes_var = multiprocessing.Value(c_int, 0) # a ctypes var that counts the bytes already downloaded

@retry(socket.timeout, tries=2, delay=1, backoff=1, logger=log)
def DownloadFile(url, path, startByte=0, endByte=None, ShowProgress=True):
    '''
    Function downloads file.
    @param url: File url address.
    @param path: Destination file path.
    @param startByte: Start byte.
    @param endByte: End byte. Will work only if server supports HTTPRange headers.
    @param ShowProgress: If true, shows textual progress bar. 

    @return path: Destination file path.
    '''
    url = url.replace(' ', '%20')
    headers = {}
    if endByte is not None:
        headers['Range'] = 'bytes=%d-%d' % (startByte,endByte)
    req = urllib2.Request(url, headers=headers)

    try:
        urlObj = urllib2.urlopen(req, timeout=4)
    except urllib2.HTTPError, e:
        if "HTTP Error 416" in str(e):
            # HTTP 416 Error: Requested Range Not Satisfiable. Happens when we ask
            # for a range that is not available on the server. It will happen when
            # the server will try to send us a .html page that means something like
            # "you opened too many connections to our server". If this happens, we
            # will wait for the other threads to finish their connections and try again.

            log.warning("Thread didn't got the file it was expecting. Retrying...")
            time.sleep(5)
            return DownloadFile(url, path, startByte, endByte, ShowProgress)
        else:
            raise Exception("urllib2.HTTPError: %s" % e)


    f = open(path, 'wb')
    meta = urlObj.info()
    try:
        filesize = int(meta.getheaders("Content-Length")[0])
    except IndexError:
        log.warning("Server did not send Content-Length.")
        ShowProgress=False

    filesize_dl = 0
    block_sz = 8192
    while True:
        try:
            buff = urlObj.read(block_sz)
        except socket.timeout, e:
            dummy.shared_bytes_var.value -= filesize_dl
            raise e

        if not buff:
            break

        filesize_dl += len(buff)
        try:
            dummy.shared_bytes_var.value += len(buff)
        except AttributeError:
            pass
        f.write(buff)

        if ShowProgress:
            status = r"%.2f MB / %.2f MB %s [%3.2f%%]" % (filesize_dl / 1024.0 / 1024.0,
                    filesize / 1024.0 / 1024.0, ProgressBar(1.0*filesize_dl/filesize),
                    filesize_dl * 100.0 / filesize)
            status += chr(8)*(len(status)+1)
            print status,
    if ShowProgress:
        print "\n"

    f.close()
    return path

@retry(urllib2.URLError, tries=2, delay=1, backoff=1, logger=log)
def DownloadFile_Parall(url, path=None, processes=config.DownloadFile_Parall_processes,
                            minChunkFile=1024**2, nonBlocking=False):
    '''
    Function downloads file parally.
    @param url: File url address.
    @param path: Destination file path.
    @param processes: Number of processes to use in the pool.
    @param minChunkFile: Minimum chunk file in bytes.
    @param nonBlocking: If true, returns (mapObj, pool). A list of file parts will be returned
                        from the mapObj results, and the developer must join them himself.
                        Developer also must close and join the pool.

    @return mapObj: Only if nonBlocking is True. A multiprocessing.pool.AsyncResult object.
    @return pool: Only if nonBlocking is True. A multiprocessing.pool object.
    '''
    from HTTPQuery import Is_ServerSupportHTTPRange
    global shared_bytes_var
    shared_bytes_var.value = 0
    url = url.replace(' ', '%20')

    if not path:
        path = r"%s\%s.TMP" % (config.temp_dir, "".join(random.choice(string.letters + string.digits) for i in range(16)))
        if not os.path.exists(os.path.dirname(path)):
            os.makedirs(os.path.dirname(path))
    log.debug("Downloading to %s..." % path)

    urlObj = urllib2.urlopen(url)
    meta = urlObj.info()
    filesize = int(meta.getheaders("Content-Length")[0])

    if filesize/processes > minChunkFile and Is_ServerSupportHTTPRange(url):
        args = []
        pos = 0
        chunk = filesize/processes
        for i in range(processes):
            startByte = pos
            endByte = pos + chunk
            if endByte > filesize-1:
                endByte = filesize-1
            args.append([url, path+".%.3d" % i, startByte, endByte, False])
            pos += chunk+1
    else:
        args = [[url, path+".000", None, None, False]]

    log.debug("Running %d processes..." % processes)
    pool = multiprocessing.Pool(processes, initializer=_initProcess,initargs=(shared_bytes_var,))
    mapObj = pool.map_async(_DownloadFile_ArgsList, args)
    if nonBlocking:
        return mapObj, pool
    while not mapObj.ready():
        status = r"%.2f MB / %.2f MB %s [%3.2f%%]" % (shared_bytes_var.value / 1024.0 / 1024.0,
                filesize / 1024.0 / 1024.0, ProgressBar(1.0*shared_bytes_var.value/filesize),
                shared_bytes_var.value * 100.0 / filesize)
        status = status + chr(8)*(len(status)+1)
        print status,
        time.sleep(0.1)

    file_parts = mapObj.get()
    pool.terminate()
    pool.join()

    combine_files(file_parts, path)

def combine_files(parts, path):
    '''
    Function combines file parts.
    @param parts: List of file paths.
    @param path: Destination path.
    '''
    with open(path,'wb') as output:
        for part in parts:
            with open(part,'rb') as f:
                output.writelines(f.readlines())
            os.remove(part)

def ProgressBar(progress, length=20):
    '''
    Function creates a textual progress bar.
    @param progress: Float number between 0 and 1 describes the progress.
    @param length: The length of the progress bar in chars. Default is 20.
    '''
    length -= 2 # The bracket are 2 chars long.
    return "[" + "#"*int(progress*length) + "-"*(length-int(progress*length)) + "]"

def _DownloadFile_ArgsList(x):
    "Function gets args for DownloadFile as a list and executes it"
    return DownloadFile(*x)

def _initProcess(x):
  dummy.shared_bytes_var = x
  • dummy class is a blank file.
  • from useful_util import retry is this function: http://www.saltycrane.com/blog/2009/11/trying-out-retry-decorator-python/
  • config has config about the project. config.DownloadFile_Parall_processes is 6.
  • The Is_ServerSupportHTTPRange from HTTPQuery is:

    @retry(urllib2.URLError, tries=4, delay=3, backoff=2, logger=log) def Is_ServerSupportHTTPRange(url, timeout=config.Is_ServerSupportHTTPRange_timeout): " Function checks if a server allows HTTP Range " url = url.replace(' ', '%20')

    fullsize = GetFileSize(url)
    
    headers = {'Range': 'bytes=0-4'}
    req = urllib2.Request(url, headers=headers)
    try:
        urlObj = urllib2.urlopen(req, timeout=timeout)
    except urllib2.HTTPError as e:
        log.error(e)
        return False
    except urllib2.URLError as e:
        log.error(e)
        return False
    
    meta = urlObj.info()
    filesize = int(meta.getheaders("Content-Length")[0])
    
    return (filesize != fullsize)
    

Thanks!

\$\endgroup\$

1 Answer 1

3
\$\begingroup\$
    url = url.replace(' ', '%20')

has codesmell. You probably want to use the general-case URL encoding method here.

    f = open(path, 'wb')

should probably be using the 'with' context manager to make sure the filehandle gets closed automatically in the case of exceptions or the like.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.