python - 使用 Python Dropbox Api 的进度条

Question

我正在使用 Dropbox API 为我的 python Dropbox 应用程序制作进度条。问题是我不知道如何获取到目前为止写入的字节数以便我可以构建它。有没有办法使用 python Dropbox API 来做到这一点？如果不是，我可以使用 os 模块来测量从系统发出的字节吗？

我正在尝试让它与 get_chunked_loader 一起工作，但是如果我能够获得为 put_file() 以及 file_copy() 和 file_move() 写入的字节，那就太好了。到目前为止我的代码是这样的:

if (file_size >= 4194304):                    
    big_file = open(path_to_file, 'rb')
    uploader = client.get_chunked_uploader(big_file)
    print "uploading " + path_to_file
    while uploader.offset < file_size:  
        percent_complete = bytes_written / file_size * 100
        clearscreen()
        print "%.2f" % percent_complete + "%"

谢谢!

Answer 1

我重新创建 ChunkedUploader

f = open(filetoupload, 'rb')
uploader = MMChunkedUploader(self.client, f, file_size, 1024*200)  
uploader.upload_chunked()
uploader.finish(dropboxfilename)  


class MMChunkedUploader(object):
  """Contains the logic around a chunked upload, which uploads a
   large file to Dropbox via the /chunked_upload endpoint.
   """

  def __init__(self, client, file_obj, length, chunk_size = 4 * 1024 * 1024):
    self.client = client
    self.offset = 0
    self.upload_id = None

    self.last_block = None
    self.file_obj = file_obj
    self.target_length = length
    self.chunk_size=chunk_size
    self.clocknumber=0
    dec=float(self.target_length)/chunk_size - self.target_length//chunk_size
    if dec >0:
      self.totalblock=self.target_length/chunk_size +1
    else:  
      self.totalblock=self.target_length/chunk_size
def upload_chunked(self, chunk_size = 0):
    """Uploads data from this ChunkedUploader's file_obj in chunks, until
    an error occurs. Throws an exception when an error occurs, and can
    be called again to resume the upload.

    Parameters
        chunk_size
          The number of bytes to put in each chunk. (Default 4 MB.)
    """
    if chunk_size ==0:
        chunk_size=self.chunk_size

    self.clocknumber=0
    while self.offset < self.target_length:
        self.clocknumber+=1
        print "Block n.", repr(self.clocknumber) , " of " , repr(self.totalblock), " %",  round((float(self.clocknumber)  * 100) / self.totalblock, 0) 

        next_chunk_size = min(chunk_size, self.target_length - self.offset)  #sceglie tra min e chuck size
        if self.last_block == None:
            self.last_block = self.file_obj.read(next_chunk_size)  
            print "Leggo blocco file"

        try:
            (self.offset, self.upload_id) = self.client.upload_chunk(
                StringIO(self.last_block), next_chunk_size, self.offset, self.upload_id)
            self.last_block = None
        except dropbox.rest.ErrorResponse as e:
            # Handle the case where the server tells us our offset is wrong.
            must_reraise = True
            if e.status == 400:
                reply = e.body
                if "offset" in reply and reply['offset'] != 0 and reply['offset'] > self.offset:
                    self.last_block = None
                    self.offset = reply['offset']
                    must_reraise = False
            if must_reraise:
                raise        

def finish(self, path, overwrite=False, parent_rev=None):


        path = "/commit_chunked_upload/%s%s" % (self.client.session.root, dropbox.client.format_path(path))

        params = dict(
            overwrite = bool(overwrite),
            upload_id = self.upload_id
        )

        if parent_rev is not None:
            params['parent_rev'] = parent_rev

        url, params, headers = self.client.request(path, params, content_server=True)

        return self.client.rest_client.POST(url, params, headers)