@@ -1,5 +1,13 @@

 2008-06-04  Michal Ludvig  <michal@logix.cz>

+	* s3cmd: Implemented cmd_sync_remote2local() for restoring

+	  backup from S3 to a local filesystem

+	* S3/S3.py: S3.object_get_uri() now requires writable stream 

+	  and not a path name.

+	* S3/Utils.py: Added mkdir_with_parents()

+

+2008-06-04  Michal Ludvig  <michal@logix.cz>

+

 	* s3cmd: Refactored cmd_sync() in preparation 

 	  for remote->local sync.

@@ -37,6 +37,9 @@ class S3Error (Exception):

 class S3UploadError(Exception):

 	pass

+class S3DownloadError(Exception):

+	pass

+

 class ParameterError(Exception):

 	pass

@@ -175,36 +175,25 @@ class S3(object):

 		response = self.send_file(request, file)

 		return response

-	def object_get_file(self, bucket, object, filename):

-		try:

-			stream = open(filename, "wb")

-		except IOError, e:

-			raise ParameterError("%s: %s" % (filename, e.strerror))

-		return self.object_get_stream(bucket, object, stream)

-

-	def object_get_stream(self, bucket, object, stream):

-		request = self.create_request("OBJECT_GET", bucket = bucket, object = object)

+	def object_get_uri(self, uri, stream):

+		if uri.type != "s3":

+			raise ValueError("Expected URI type 's3', got '%s'" % uri.type)

+		request = self.create_request("OBJECT_GET", bucket = uri.bucket(), object = uri.object())

 		response = self.recv_file(request, stream)

 		return response

-		

+

 	def object_delete(self, bucket, object):

 		request = self.create_request("OBJECT_DELETE", bucket = bucket, object = object)

 		response = self.send_request(request)

 		return response

 	def object_put_uri(self, filename, uri, extra_headers = None):

+		# TODO TODO

+		# Make it consistent with stream-oriented object_get_uri()

 		if uri.type != "s3":

 			raise ValueError("Expected URI type 's3', got '%s'" % uri.type)

 		return self.object_put(filename, uri.bucket(), uri.object(), extra_headers)

-	def object_get_uri(self, uri, filename):

-		if uri.type != "s3":

-			raise ValueError("Expected URI type 's3', got '%s'" % uri.type)

-		if filename == "-":

-			return self.object_get_stream(uri.bucket(), uri.object(), sys.stdout)

-		else:

-			return self.object_get_file(uri.bucket(), uri.object(), filename)

-

 	def object_delete_uri(self, uri):

 		if uri.type != "s3":

 			raise ValueError("Expected URI type 's3', got '%s'" % uri.type)

@@ -11,6 +11,8 @@ import random

 import md5

 import errno

+from logging import debug, info, warning, error

+

 try:

 	import xml.etree.ElementTree as ET

 except ImportError:

@@ -140,3 +142,29 @@ def hash_file_md5(filename):

 	h.update(f.read())

 	f.close()

 	return h.hexdigest()

+

+def mkdir_with_parents(dir_name, mode):

+	"""

+	mkdir_with_parents(dst_dir, mode)

+	

+	Create directory 'dir_name' with all parent directories

+

+	Returns True on success, False otherwise.

+	"""

+	pathmembers = dir_name.split(os.sep)

+	tmp_stack = []

+	while pathmembers and not os.path.isdir(os.sep.join(pathmembers)):

+		tmp_stack.append(pathmembers.pop())

+	while tmp_stack:

+		pathmembers.append(tmp_stack.pop())

+		cur_dir = os.sep.join(pathmembers)

+		try:

+			debug("mkdir(%s)" % cur_dir)

+			os.mkdir(cur_dir)

+		except IOError, e:

+			error("%s: can not make directory: %s" % (cur_dir, e.strerror))

+			return False

+		except Exception, e:

+			error("%s: %s" % (cur_dir, e))

+			return False

+	return True

@@ -243,10 +243,19 @@ def cmd_object_get(args):

 		else:

 			# By default the destination filename is the object name

 			destination = uri.object()

-

-		if not Config().force and os.path.exists(destination):

-			raise ParameterError("File %s already exists. Use --force to overwrite it" % destination)

-		response = s3.object_get_uri(uri, destination)

+		if destination == "-":

+			## stdout

+			dst_stream = sys.stdout

+		else:

+			## File

+			if not Config().force and os.path.exists(destination):

+				raise ParameterError("File %s already exists. Use --force to overwrite it" % destination)

+			try:

+				dst_stream = open(destination, "wb")

+			except IOError, e:

+				error("Skipping %s: %s" % (destination, e.strerror))

+				continue

+		response = s3.object_get_uri(uri, dst_stream)

 		if response["headers"].has_key("x-amz-meta-s3tools-gpgenc"):

 			gpg_decrypt(destination, response["headers"]["x-amz-meta-s3tools-gpgenc"])

 			response["size"] = os.stat(destination)[6]

@@ -383,7 +392,106 @@ def _compare_filelists(src_list, dst_list, src_is_local_and_dst_is_remote):

 	return src_list, dst_list, exists_list

 def cmd_sync_remote2local(src, dst):

-	raise NotImplementedError("Remote->Local sync is not yet implemented.") 

+	def _parse_attrs_header(attrs_header):

+		attrs = {}

+		for attr in attrs_header.split("/"):

+			key, val = attr.split(":")

+			attrs[key] = val

+		return attrs

+		

+	s3 = S3(Config())

+

+	src_uri = S3Uri(src)

+	dst_uri = S3Uri(dst)

+

+	rem_list = _get_filelist_remote(src_uri)

+	rem_count = len(rem_list)

+

+	loc_list = _get_filelist_local(dst_uri)

+	loc_count = len(loc_list)

+	

+	output("Found %d remote files, %d local files" % (rem_count, loc_count))

+

+	_compare_filelists(rem_list, loc_list, False)

+

+	output("Summary: %d remote files to download, %d local files to delete" % (len(rem_list), len(loc_list)))

+

+	for file in loc_list:

+		if cfg.delete_removed:

+			# os.unlink(file)

+			output("deleted '%s'" % file)

+		else:

+			output("not-deleted '%s'" % file)

+

+	total_size = 0

+	total_count = len(rem_list)

+	total_elapsed = 0.0

+	timestamp_start = time.time()

+	seq = 0

+	dir_cache = {}

+	src_base = src_uri.uri()

+	dst_base = dst_uri.path()

+	if not src_base[-1] == "/": src_base += "/"

+	file_list = rem_list.keys()

+	file_list.sort()

+	for file in file_list:

+		seq += 1

+		uri = S3Uri(src_base + file)

+		dst_file = dst_base + file

+		try:

+			dst_dir = os.path.dirname(dst_file)

+			if not dir_cache.has_key(dst_dir):

+				dir_cache[dst_dir] = Utils.mkdir_with_parents(dst_dir, mode = 022)

+			if dir_cache[dst_dir] == False:

+				warning("%s: destination directory not writable: %s" % (file, dst_dir))

+				continue

+			try:

+				open_flags = os.O_CREAT

+				if cfg.force:

+					open_flags |= os.O_TRUNC

+				else:

+					open_flags |= os.O_EXCL

+

+				debug("dst_file=%s" % dst_file)

+				# This will have failed should the file exist

+				os.open(dst_file, open_flags)

+				# Yeah I know there is a race condition here. Sadly I don't know how to open() in exclusive mode.

+				dst_stream = open(dst_file, "wb")

+				response = s3.object_get_uri(uri, dst_stream)

+				dst_stream.close()

+				if response['headers'].has_key('x-amz-meta-s3cmd-attrs') and cfg.preserve_attrs:

+					attrs = _parse_attrs_header(response['headers']['x-amz-meta-s3cmd-attrs'])

+					if attrs.has_key('mode'):

+						os.chmod(dst_file, int(attrs['mode']))

+					## FIXME: uid/gid and mtime/ctime handling comes here! TODO

+			except OSError, e:

+				if e.errno == errno.EEXIST:

+					warning("%s exists - not overwriting" % (dst_file))

+					continue

+				raise

+			except IOError, e:

+				## See if it's missing path and try again

+				error("%s: %s" % (file, e))

+				continue

+			finally:

+				## Close the file if still open. Don't care if not.

+				try:

+					dst_stream.close()

+				except:

+					pass

+		except S3DownloadError, e:

+			error("%s: download failed too many times. Skipping that file." % file)

+			continue

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

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

+			(uri, dst_file, response["size"], response["elapsed"], speed_fmt[0], speed_fmt[1],

+			seq, total_count))

+		total_size += response["size"]

+

+	total_elapsed = time.time() - timestamp_start

+	speed_fmt = formatSize(total_size/total_elapsed, human_readable = True, floating_point = True)

+	output("Done. Downloaded %d bytes in %0.1f seconds, %0.2f %sB/s" % 

+	       (total_size, total_elapsed, speed_fmt[0], speed_fmt[1]))

 def cmd_sync_local2remote(src, dst):

 	def _build_attr_header(src):

@@ -427,6 +535,7 @@ def cmd_sync_local2remote(src, dst):

 	_compare_filelists(loc_list, rem_list, True)

 	output("Summary: %d local files to upload, %d remote files to delete" % (len(loc_list), len(rem_list)))

+

 	for file in rem_list:

 		uri = S3Uri("s3://" + dst_uri.bucket()+"/"+rem_list[file]['object_key'])

 		if cfg.delete_removed:

@@ -454,7 +563,7 @@ def cmd_sync_local2remote(src, dst):

 		try:

 			response = s3.object_put_uri(src, uri, attr_header)

 		except S3UploadError, e:

-			error("Upload of '%s' failed too many times. Skipping that file." % src)

+			error("%s: upload failed too many times. Skipping that file." % src)

 			continue

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

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

@@ -473,6 +582,9 @@ def cmd_sync(args):

 	if (len(args)):

 		raise ParameterError("Too many parameters! Expected: %s" % commands['sync']['param'])

+	if S3Uri(src).type == "s3" and not src.endswith('/'):

+		src += "/"

+

 	if not dst.endswith('/'):

 		dst += "/"