text: # Licensed to Cloudera, Inc. under one
  # or more contributor license agreements.  See the NOTICE file
  # distributed with this work for additional information
  # regarding copyright ownership.  Cloudera, Inc. licenses this file
  # to you under the Apache License, Version 2.0 (the
  # "License"); you may not use this file except in compliance
  # with the License.  You may obtain a copy of the License at
  #
  #     http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  from __future__ import absolute_import
  
  import sys
  
  import errno
  import itertools
  import logging
  import os
  import posixpath
  
  from boto.exception import S3ResponseError
  from boto.s3.key import Key
  from boto.s3.prefix import Prefix
  
  from aws import s3
  from aws.s3 import translate_s3_error, s3file
  from aws.s3.s3stat import S3Stat
  
  from hadoop.fs import normpath
  
  
  DEFAULT_READ_SIZE = 1024 * 1024  # 1MB
  LOG = logging.getLogger(__name__)
  
  
  class S3FileSystem(object):
    def __init__(self, s3_connection):
      self._s3_connection = s3_connection
      self._bucket_cache = None
  
    def _init_bucket_cache(self):
      if self._bucket_cache is None:
        buckets = self._s3_connection.get_all_buckets()
        self._bucket_cache = {}
        for bucket in buckets:
          self._bucket_cache[bucket.name] = bucket
  
    def _get_bucket(self, name):
      self._init_bucket_cache()
      if name not in self._bucket_cache:
        self._bucket_cache[name] = self._s3_connection.get_bucket(name)
      return self._bucket_cache[name]
  
    def _get_key(self, path, validate=True):
      bucket_name, key_name = s3.parse_uri(path)[:2]
      bucket = self._get_bucket(bucket_name)
      try:
        return bucket.get_key(key_name, validate=validate)
      except:
        e, exc, tb = sys.exc_info()
        raise ValueError(e)
  
    def _stats(self, path):
      if s3.is_root(path):
        return S3Stat.for_s3_root()
  
      try:
        key = self._get_key(path, validate=True)
      except S3ResponseError as e:
        if e.status == 404:
          return None
        else:
          exc_class, exc, tb = sys.exc_info()
          raise exc_class, exc, tb
  
      if key is None:
        key = self._get_key(path, validate=False)
      return self._stats_key(key)
  
    @staticmethod
    def _stats_key(key):
      if key.size is not None:
        is_directory_name = not key.name or key.name[-1] == '/'
        return S3Stat.from_key(key, is_dir=is_directory_name)
      else:
        key.name = S3FileSystem._append_separator(key.name)
        ls = key.bucket.get_all_keys(prefix=key.name, max_keys=1)
        if len(ls) > 0:
          return S3Stat.from_key(key, is_dir=True)
      return None
  
    @staticmethod
    def _append_separator(path):
      if path and not path.endswith('/'):
        path += '/'
      return path
  
    @staticmethod
    def _cut_separator(path):
      return path.endswith('/') and path[:-1] or path
  
    @staticmethod
    def isroot(path):
      return s3.is_root(path)
  
    @staticmethod
    def join(*comp_list):
      return s3.join(*comp_list)
  
    @staticmethod
    def normpath(path):
      return normpath(path)
  
    @translate_s3_error
    def open(self, path, mode='r'):
      key = self._get_key(path, validate=True)
      if key is None:
        raise IOError(errno.ENOENT, "No such file or directory: '%s'" % path)
      return s3file.open(key, mode=mode)
  
    @translate_s3_error
    def read(self, path, offset, length):
      fh = self.open(path, 'r')
      fh.seek(offset, os.SEEK_SET)
      return fh.read(length)
  
    @translate_s3_error
    def isfile(self, path):
      stat = self._stats(path)
      if stat is None:
        return False
      return not stat.isDir
  
    @translate_s3_error
    def isdir(self, path):
      stat = self._stats(path)
      if stat is None:
        return False
      return stat.isDir
  
    @translate_s3_error
    def exists(self, path):
      return self._stats(path) is not None
  
    @translate_s3_error
    def stats(self, path):
      path = normpath(path)
      stats = self._stats(path)
      if stats:
        return stats
      raise IOError(errno.ENOENT, "No such file or directory: '%s'" % path)
  
    @translate_s3_error
    def listdir_stats(self, path, glob=None):
      if glob is not None:
        raise NotImplementedError("Option `glob` is not implemented")
  
      if s3.is_root(path):
        self._init_bucket_cache()
        return [S3Stat.from_bucket(b) for b in self._bucket_cache.values()]
  
      bucket_name, prefix = s3.parse_uri(path)[:2]
      bucket = self._get_bucket(bucket_name)
      prefix = self._append_separator(prefix)
      res = []
      for item in bucket.list(prefix=prefix, delimiter='/'):
        if isinstance(item, Prefix):
          res.append(S3Stat.from_key(Key(item.bucket, item.name), is_dir=True))
        else:
          if item.name == prefix:
            continue
          res.append(self._stats_key(item))
      return res
  
    def listdir(self, path, glob=None):
      return [s3.parse_uri(x.path)[2] for x in self.listdir_stats(path, glob)]
  
    @translate_s3_error
    def rmtree(self, path, skipTrash=False):
      if not skipTrash:
        raise NotImplementedError('Moving to trash is not implemented for S3')
      key = self._get_key(path, validate=False)
  
      if key.exists():
        to_delete = iter([key])
      else:
        to_delete = iter([])
  
      if self.isdir(path):
        # add `/` to prevent removing of `s3://b/a_new` trying to remove `s3://b/a`
        prefix = self._append_separator(key.name)
        keys = key.bucket.list(prefix=prefix)
        to_delete = itertools.chain(keys, to_delete)
      result = key.bucket.delete_keys(to_delete)
      if result.errors:
        msg = "%d errors occurred during deleting '%s':\n%s" % (
          len(result.errors),
          '\n'.join(map(repr, result.errors)))
        LOG.error(msg)
        raise IOError(msg)
  
    @translate_s3_error
    def remove(self, path, skip_trash=False):
      if not skip_trash:
        raise NotImplementedError('Moving to trash is not implemented for S3')
      key = self._get_key(path, validate=False)
      key.bucket.delete_key(key.name)
  
    def restore(self, *args, **kwargs):
      raise NotImplementedError('Moving to trash is not implemented for S3')
  
    @translate_s3_error
    def mkdir(self, path, *args, **kwargs):
      """
      Creates a directory and any parent directory if necessary.
  
      Actually it creates an empty object: s3://[bucket]/[path]/
      """
      stats = self._stats(path)
      if stats:
        if stats.isDir:
          return None
        else:
          raise IOError(errno.ENOTDIR, "'%s' already exists and is not a directory" % path)
      path = self._append_separator(path)  # folder-key should ends by /
      self.create(path)  # create empty object
  
    @translate_s3_error
    def copy(self, src, dst, recursive=False, *args, **kwargs):
      self._copy(src, dst, recursive=recursive, use_src_basename=True)
  
    @translate_s3_error
    def copyfile(self, src, dst, *args, **kwargs):
      if self.isdir(dst):
        raise IOError(errno.EINVAL, "Copy dst '%s' is a directory" % dst)
      self._copy(src, dst, recursive=False, use_src_basename=False)
  
    @translate_s3_error
    def copy_remote_dir(self, src, dst, *args, **kwargs):
      self._copy(src, dst, recursive=True, use_src_basename=False)
  
    def _copy(self, src, dst, recursive, use_src_basename):
      src_st = self.stats(src)
      if src_st.isDir and not recursive:
        return # omitting directory
  
      dst = s3.abspath(src, dst)
      dst_st = self._stats(dst)
      if src_st.isDir and dst_st and not dst_st.isDir:
        raise IOError(errno.EEXIST, "Cannot overwrite non-directory '%s' with directory '%s'" % (dst, src))
  
      src_bucket, src_key = s3.parse_uri(src)[:2]
      dst_bucket, dst_key = s3.parse_uri(dst)[:2]
  
      keep_src_basename = use_src_basename and dst_st and dst_st.isDir
      src_bucket = self._get_bucket(src_bucket)
      dst_bucket = self._get_bucket(dst_bucket)
  
      if keep_src_basename:
        cut = len(posixpath.dirname(src_key))  # cut of an parent directory name
        if cut:
          cut += 1
      else:
        cut = len(src_key)
        if not src_key.endswith('/'):
          cut += 1
  
      for key in src_bucket.list(prefix=src_key):
        if not key.name.startswith(src_key):
          raise RuntimeError("Invalid key to transform: %s" % key.name)
        dst_name = posixpath.normpath(s3.join(dst_key, key.name[cut:]))
        key.copy(dst_bucket, dst_name)
  
    @translate_s3_error
    def rename(self, old, new):
      new = s3.abspath(old, new)
      self.copy(old, new, recursive=True)
      self.rmtree(old, skipTrash=True)
  
    @translate_s3_error
    def rename_star(self, old_dir, new_dir):
      if not self.isdir(old_dir):
        raise IOError(errno.ENOTDIR, "'%s' is not a directory" % old_dir)
      if self.isfile(new_dir):
        raise IOError(errno.ENOTDIR, "'%s' is not a directory" % new_dir)
      ls = self.listdir(old_dir)
      for entry in ls:
        self.rename(s3.join(old_dir, entry), s3.join(new_dir, entry))
  
    @translate_s3_error
    def create(self, path, overwrite=False, data=None):
      key = self._get_key(path, validate=False)
      key.set_contents_from_string(data or '', replace=overwrite)
  
    @translate_s3_error
    def copyFromLocal(self, local_src, remote_dst, *args, **kwargs):
      local_src = self._cut_separator(local_src)
      remote_dst = self._cut_separator(remote_dst)
  
      def _copy_file(src, dst):
        key = self._get_key(dst, validate=False)
        fp = open(src, 'r')
        key.set_contents_from_file(fp)
  
      if os.path.isdir(local_src):
        for (local_dir, sub_dirs, files) in os.walk(local_src, followlinks=False):
          remote_dir = local_dir.replace(local_src, remote_dst)
  
          if not sub_dirs and not files:
            self.mkdir(remote_dir)
          else:
            for file_name in files:
              _copy_file(os.path.join(local_dir, file_name), os.path.join(remote_dir, file_name))
      else:
        file_name = os.path.split(local_src)[1]
        if self.isdir(remote_dst):
          remote_file = os.path.join(remote_dst, file_name)
        else:
          remote_file = remote_dst
        _copy_file(local_src, remote_file)
  
    def setuser(self, user):
      pass  # user-concept doesn't have sense for this implementation
  