@@ -38,6 +38,8 @@ class Config(object):

     force = False

     enable = None

     get_continue = False

+    put_continue = False

+    upload_id = None

     skip_existing = False

     recursive = False

     acl_public = None

@@ -3,10 +3,12 @@

 ## License: GPL Version 2

 import os

+import sys

 from stat import ST_SIZE

 from logging import debug, info, warning, error

-from Utils import getTextFromXml, formatSize, unicodise

+from Utils import getTextFromXml, getTreeFromXml, formatSize, unicodise, calculateChecksum, parseNodes

 from Exceptions import S3UploadError

+from collections import defaultdict

 class MultiPartUpload(object):

@@ -22,15 +24,55 @@ class MultiPartUpload(object):

         self.headers_baseline = headers_baseline

         self.upload_id = self.initiate_multipart_upload()

+    def get_parts_information(self, uri, upload_id):

+        multipart_response = self.s3.list_multipart(uri, upload_id)

+        tree = getTreeFromXml(multipart_response['data'])

+

+        parts = defaultdict(lambda: None)

+        for elem in parseNodes(tree):

+            try:

+                parts[int(elem['PartNumber'])] = {'checksum': elem['ETag'], 'size': elem['Size']}

+            except KeyError:

+                pass

+

+        return parts

+

+    def get_unique_upload_id(self, uri):

+        upload_id = None

+        multipart_response = self.s3.get_multipart(uri)

+        tree = getTreeFromXml(multipart_response['data'])

+        for mpupload in parseNodes(tree):

+            try:

+                mp_upload_id = mpupload['UploadId']

+                mp_path = mpupload['Key']

+                info("mp_path: %s, object: %s" % (mp_path, uri.object()))

+                if mp_path == uri.object():

+                    if upload_id is not None:

+                        raise ValueError("More than one UploadId for URI %s.  Disable multipart upload, or use\n %s multipart %s\nto list the Ids, then pass a unique --upload-id into the put command." % (uri, sys.argv[0], uri))

+                    upload_id = mp_upload_id

+            except KeyError:

+                pass

+

+        return upload_id

+

     def initiate_multipart_upload(self):

         """

         Begin a multipart upload

         http://docs.amazonwebservices.com/AmazonS3/latest/API/index.html?mpUploadInitiate.html

         """

-        request = self.s3.create_request("OBJECT_POST", uri = self.uri, headers = self.headers_baseline, extra = "?uploads")

-        response = self.s3.send_request(request)

-        data = response["data"]

-        self.upload_id = getTextFromXml(data, "UploadId")

+        if self.s3.config.upload_id is not None:

+            self.upload_id = self.s3.config.upload_id

+        elif self.s3.config.put_continue:

+            self.upload_id = self.get_unique_upload_id(self.uri)

+        else:

+            self.upload_id = None

+

+        if self.upload_id is None:

+            request = self.s3.create_request("OBJECT_POST", uri = self.uri, headers = self.headers_baseline, extra = "?uploads")

+            response = self.s3.send_request(request)

+            data = response["data"]

+            self.upload_id = getTextFromXml(data, "UploadId")

+

         return self.upload_id

     def upload_all_parts(self):

@@ -51,6 +93,10 @@ class MultiPartUpload(object):

         else:

             debug("MultiPart: Uploading from %s" % (self.file.name))

+        remote_statuses = defaultdict(lambda: None)

+        if self.s3.config.put_continue:

+            remote_statuses = self.get_parts_information(self.uri, self.upload_id)

+

         seq = 1

         if self.file.name != "<stdin>":

             while size_left > 0:

@@ -63,10 +109,10 @@ class MultiPartUpload(object):

                     'extra' : "[part %d of %d, %s]" % (seq, nr_parts, "%d%sB" % formatSize(current_chunk_size, human_readable = True))

                 }

                 try:

-                    self.upload_part(seq, offset, current_chunk_size, labels)

+                    self.upload_part(seq, offset, current_chunk_size, labels, remote_status = remote_statuses[seq])

                 except:

-                    error(u"Upload of '%s' part %d failed. Aborting multipart upload." % (self.file.name, seq))

-                    self.abort_upload()

+                    error(u"\nUpload of '%s' part %d failed. Use\n  %s abortmp %s %s\nto abort the upload, or\n  %s --upload-id %s put ...\nto continue the upload."

+                          % (self.file.name, seq, sys.argv[0], self.uri, self.upload_id, sys.argv[0], self.upload_id))

                     raise

                 seq += 1

         else:

@@ -82,22 +128,38 @@ class MultiPartUpload(object):

                 if len(buffer) == 0: # EOF

                     break

                 try:

-                    self.upload_part(seq, offset, current_chunk_size, labels, buffer)

+                    self.upload_part(seq, offset, current_chunk_size, labels, buffer, remote_status = remote_statuses[seq])

                 except:

-                    error(u"Upload of '%s' part %d failed. Aborting multipart upload." % (self.file.name, seq))

-                    self.abort_upload()

+                    error(u"\nUpload of '%s' part %d failed. Use\n  %s abortmp %s %s\nto abort, or\n  %s --upload-id %s put ...\nto continue the upload."

+                          % (self.file.name, seq, self.uri, sys.argv[0], self.upload_id, sys.argv[0], self.upload_id))

                     raise

                 seq += 1

         debug("MultiPart: Upload finished: %d parts", seq - 1)

-    def upload_part(self, seq, offset, chunk_size, labels, buffer = ''):

+    def upload_part(self, seq, offset, chunk_size, labels, buffer = '', remote_status = None):

         """

         Upload a file chunk

         http://docs.amazonwebservices.com/AmazonS3/latest/API/index.html?mpUploadUploadPart.html

         """

         # TODO implement Content-MD5

         debug("Uploading part %i of %r (%s bytes)" % (seq, self.upload_id, chunk_size))

+

+        if remote_status is not None:

+            if int(remote_status['size']) == chunk_size:

+                checksum = calculateChecksum(buffer, self.file, offset, chunk_size, self.s3.config.send_chunk)

+                remote_checksum = remote_status['checksum'].strip('"')

+                if remote_checksum == checksum:

+                    warning("MultiPart: size and md5sum match for %s part %d, skipping." % (self.uri, seq))

+                    self.parts[seq] = remote_status['checksum']

+                    return

+                else:

+                    warning("MultiPart: checksum (%s vs %s) does not match for %s part %d, reuploading."

+                            % (remote_checksum, checksum, self.uri, seq))

+            else:

+                warning("MultiPart: size (%d vs %d) does not match for %s part %d, reuploading."

+                        % (int(remote_status['size']), chunk_size, self.uri, seq))

+

         headers = { "content-length": chunk_size }

         query_string = "?partNumber=%i&uploadId=%s" % (seq, self.upload_id)

         request = self.s3.create_request("OBJECT_PUT", uri = self.uri, headers = headers, extra = query_string)

@@ -130,8 +192,19 @@ class MultiPartUpload(object):

         http://docs.amazonwebservices.com/AmazonS3/latest/API/index.html?mpUploadAbort.html

         """

         debug("MultiPart: Aborting upload: %s" % self.upload_id)

-        request = self.s3.create_request("OBJECT_DELETE", uri = self.uri, extra = "?uploadId=%s" % (self.upload_id))

-        response = self.s3.send_request(request)

+        #request = self.s3.create_request("OBJECT_DELETE", uri = self.uri, extra = "?uploadId=%s" % (self.upload_id))

+        #response = self.s3.send_request(request)

+        response = None

         return response

 # vim:et:ts=4:sts=4:ai

+

+

+

+

+

+

+

+

+

+

@@ -438,6 +438,31 @@ class S3(object):

             return self.send_file_multipart(file, headers, uri, size)

         ## Not multipart...

+        if self.config.put_continue:

+            # Note, if input was stdin, we would be performing multipart upload.

+            # So this will always work as long as the file already uploaded was

+            # not uploaded via MultiUpload, in which case its ETag will not be

+            # an md5.

+            try:

+                info = self.object_info(uri)

+            except:

+                info = None

+

+            if info is not None:

+                remote_size = int(info['headers']['content-length'])

+                remote_checksum = info['headers']['etag'].strip('"')

+                if size == remote_size:

+                    checksum = calculateChecksum('', file, 0, size, self.config.send_chunk)

+                    if remote_checksum == checksum:

+                        warning("Put: size and md5sum match for %s, skipping." % uri)

+                        return

+                    else:

+                        warning("MultiPart: checksum (%s vs %s) does not match for %s, reuploading."

+                                % (remote_checksum, checksum, uri))

+                else:

+                    warning("MultiPart: size (%d vs %d) does not match for %s, reuploading."

+                            % (remote_size, size, uri))

+

         headers["content-length"] = size

         request = self.create_request("OBJECT_PUT", uri = uri, headers = headers)

         labels = { 'source' : unicodise(filename), 'destination' : unicodise(uri.uri()), 'extra' : extra_label }

@@ -535,6 +560,23 @@ class S3(object):

         response = self.send_request(request)

         return response

+    def get_multipart(self, uri):

+        request = self.create_request("BUCKET_LIST", bucket = uri.bucket(), extra = "?uploads")

+        response = self.send_request(request)

+        return response

+

+    def abort_multipart(self, uri, id):

+        request = self.create_request("OBJECT_DELETE", uri=uri,

+                                      extra = ("?uploadId=%s" % id))

+        response = self.send_request(request)

+        return response

+

+    def list_multipart(self, uri, id):

+        request = self.create_request("OBJECT_GET", uri=uri,

+                                      extra = ("?uploadId=%s" % id))

+        response = self.send_request(request)

+        return response

+

     def get_accesslog(self, uri):

         request = self.create_request("BUCKET_LIST", bucket = uri.bucket(), extra = "?logging")

         response = self.send_request(request)

@@ -739,6 +781,7 @@ class S3(object):

         if buffer == '':

             file.seek(offset)

         md5_hash = md5()

+

         try:

             while (size_left > 0):

                 #debug("SendFile: Reading up to %d bytes from '%s' - remaining bytes: %s" % (self.config.send_chunk, file.name, size_left))

@@ -746,6 +789,7 @@ class S3(object):

                     data = file.read(min(self.config.send_chunk, size_left))

                 else:

                     data = buffer

+

                 md5_hash.update(data)

                 conn.c.send(data)

                 if self.config.progress_meter:

@@ -754,6 +798,7 @@ class S3(object):

                 if throttle:

                     time.sleep(throttle)

             md5_computed = md5_hash.hexdigest()

+

             response = {}

             http_response = conn.c.getresponse()

             response["status"] = http_response.status

@@ -459,4 +459,22 @@ def getHostnameFromBucket(bucket):

     return Config.Config().host_bucket % { 'bucket' : bucket }

 __all__.append("getHostnameFromBucket")

+

+def calculateChecksum(buffer, mfile, offset, chunk_size, send_chunk):

+    md5_hash = md5()

+    size_left = chunk_size

+    if buffer == '':

+        mfile.seek(offset)

+        while size_left > 0:

+            data = mfile.read(min(send_chunk, size_left))

+            md5_hash.update(data)

+            size_left -= len(data)

+    else:

+        md5_hash.update(buffer)

+

+    return md5_hash.hexdigest()

+

+

+__all__.append("calculateChecksum")

+

 # vim:et:ts=4:sts=4:ai

@@ -339,14 +339,15 @@ def cmd_object_put(args):

         except InvalidFileError, e:

             warning(u"File can not be uploaded: %s" % e)

             continue

-        speed_fmt = formatSize(response["speed"], human_readable = True, floating_point = True)

-        if not Config().progress_meter:

-            output(u"File '%s' stored as '%s' (%d bytes in %0.1f seconds, %0.2f %sB/s) %s" %

-                (unicodise(full_name_orig), uri_final, response["size"], response["elapsed"],

-                speed_fmt[0], speed_fmt[1], seq_label))

+        if response is not None:

+            speed_fmt = formatSize(response["speed"], human_readable = True, floating_point = True)

+            if not Config().progress_meter:

+                output(u"File '%s' stored as '%s' (%d bytes in %0.1f seconds, %0.2f %sB/s) %s" %

+                       (unicodise(full_name_orig), uri_final, response["size"], response["elapsed"],

+                        speed_fmt[0], speed_fmt[1], seq_label))

         if Config().acl_public:

             output(u"Public URL of the object is: %s" %

-                (uri_final.public_url()))

+                   (uri_final.public_url()))

         if Config().encrypt and full_name != full_name_orig:

             debug(u"Removing temporary encrypted file: %s" % unicodise(full_name))

             os.remove(full_name)

@@ -1340,6 +1341,50 @@ def cmd_delpolicy(args):

     output(u"%s: Policy deleted" % uri)

+def cmd_multipart(args):

+    s3 = S3(cfg)

+    uri = S3Uri(args[0])

+

+    #id = ''

+    #if(len(args) > 1): id = args[1]

+

+    response = s3.get_multipart(uri)

+    debug(u"response - %s" % response['status'])

+    output(u"%s" % uri)

+    tree = getTreeFromXml(response['data'])

+    debug(parseNodes(tree))

+    output(u"Initiated\tPath\tId")

+    for mpupload in parseNodes(tree):

+        try:

+            output("%s\t%s\t%s" % (mpupload['Initiated'], "s3://" + uri.bucket() + "/" + mpupload['Key'], mpupload['UploadId']))

+        except KeyError:

+            pass

+

+def cmd_abort_multipart(args):

+    '''{"cmd":"abortmp",   "label":"abort a multipart upload", "param":"s3://BUCKET Id", "func":cmd_abort_multipart, "argc":2},'''

+    s3 = S3(cfg)

+    uri = S3Uri(args[0])

+    id = args[1]

+    response = s3.abort_multipart(uri, id)

+    debug(u"response - %s" % response['status'])

+    output(u"%s" % uri)

+

+def cmd_list_multipart(args):

+    '''{"cmd":"abortmp",   "label":"list a multipart upload", "param":"s3://BUCKET Id", "func":cmd_list_multipart, "argc":2},'''

+    s3 = S3(cfg)

+    uri = S3Uri(args[0])

+    id = args[1]

+

+    response = s3.list_multipart(uri, id)

+    debug(u"response - %s" % response['status'])

+    tree = getTreeFromXml(response['data'])

+    output(u"LastModified\t\t\tPartNumber\tETag\tSize")

+    for mpupload in parseNodes(tree):

+        try:

+            output("%s\t%s\t%s\t%s" % (mpupload['LastModified'], mpupload['PartNumber'], mpupload['ETag'], mpupload['Size']))

+        except:

+            pass

+

 def cmd_accesslog(args):

     s3 = S3(cfg)

     bucket_uri = S3Uri(args.pop())

@@ -1694,6 +1739,11 @@ def get_commands_list():

     {"cmd":"setpolicy", "label":"Modify Bucket Policy", "param":"FILE s3://BUCKET", "func":cmd_setpolicy, "argc":2},

     {"cmd":"delpolicy", "label":"Delete Bucket Policy", "param":"s3://BUCKET", "func":cmd_delpolicy, "argc":1},

+    {"cmd":"multipart", "label":"show multipart uploads", "param":"s3://BUCKET [Id]", "func":cmd_multipart, "argc":1},

+    {"cmd":"abortmp",   "label":"abort a multipart upload", "param":"s3://BUCKET/OBJECT Id", "func":cmd_abort_multipart, "argc":2},

+

+    {"cmd":"listmp",    "label":"list parts of a multipart upload", "param":"s3://BUCKET/OBJECT Id", "func":cmd_list_multipart, "argc":2},

+

     {"cmd":"accesslog", "label":"Enable/disable bucket access logging", "param":"s3://BUCKET", "func":cmd_accesslog, "argc":1},

     {"cmd":"sign", "label":"Sign arbitrary string using the secret key", "param":"STRING-TO-SIGN", "func":cmd_sign, "argc":1},

     {"cmd":"signurl", "label":"Sign an S3 URL to provide limited public access with expiry", "param":"s3://BUCKET/OBJECT expiry_epoch", "func":cmd_signurl, "argc":2},

@@ -1834,6 +1884,8 @@ def main():

     optparser.add_option(      "--no-encrypt", dest="encrypt", action="store_false", help="Don't encrypt files.")

     optparser.add_option("-f", "--force", dest="force", action="store_true", help="Force overwrite and other dangerous operations.")

     optparser.add_option(      "--continue", dest="get_continue", action="store_true", help="Continue getting a partially downloaded file (only for [get] command).")

+    optparser.add_option(      "--continue-put", dest="put_continue", action="store_true", help="Continue uploading partially uploaded files or multipart upload parts.  Restarts/parts files that don't have matching size and md5.  Skips files/parts that do.  Note: md5sum checks are not always sufficient to check (part) file equality.  Enable this at your own risk.")

+    optparser.add_option(      "--upload-id", dest="upload_id", help="UploadId for Multipart Upload, in case you want continue an existing upload (equivalent to --continue-put) and there are multiple partial uploads.  Use s3cmd multipart [URI] to see what UploadIds are associated with the given URI.")

     optparser.add_option(      "--skip-existing", dest="skip_existing", action="store_true", help="Skip over files that exist at the destination (only for [get] and [sync] commands).")

     optparser.add_option("-r", "--recursive", dest="recursive", action="store_true", help="Recursive upload, download or removal.")

     optparser.add_option(      "--check-md5", dest="check_md5", action="store_true", help="Check MD5 sums when comparing files for [sync]. (default)")

@@ -2027,6 +2079,14 @@ def main():

     if cfg.multipart_chunk_size_mb > MultiPartUpload.MAX_CHUNK_SIZE_MB:

         raise ParameterError("Chunk size %d MB is too large, must be <= %d MB. Please adjust --multipart-chunk-size-mb" % (cfg.multipart_chunk_size_mb, MultiPartUpload.MAX_CHUNK_SIZE_MB))

+    ## If an UploadId was provided, set put_continue True

+    if options.upload_id is not None:

+        cfg.upload_id = options.upload_id

+        cfg.put_continue = True

+

+    if cfg.upload_id and not cfg.multipart_chunk_size_mb:

+        raise ParameterError("Must have --multipart-chunk-size-mb if using --put-continue or --upload-id")

+

     ## CloudFront's cf_enable and Config's enable share the same --enable switch

     options.cf_enable = options.enable