--- /dev/null
+"""Utility functions for copying and archiving files and directory trees.\r
+\r
+XXX The functions here don't copy the resource fork or other metadata on Mac.\r
+\r
+"""\r
+\r
+import os\r
+import sys\r
+import stat\r
+from os.path import abspath\r
+import fnmatch\r
+import collections\r
+import errno\r
+\r
+try:\r
+ from pwd import getpwnam\r
+except ImportError:\r
+ getpwnam = None\r
+\r
+try:\r
+ from grp import getgrnam\r
+except ImportError:\r
+ getgrnam = None\r
+\r
+__all__ = ["copyfileobj", "copyfile", "copymode", "copystat", "copy", "copy2",\r
+ "copytree", "move", "rmtree", "Error", "SpecialFileError",\r
+ "ExecError", "make_archive", "get_archive_formats",\r
+ "register_archive_format", "unregister_archive_format",\r
+ "ignore_patterns"]\r
+\r
+class Error(EnvironmentError):\r
+ pass\r
+\r
+class SpecialFileError(EnvironmentError):\r
+ """Raised when trying to do a kind of operation (e.g. copying) which is\r
+ not supported on a special file (e.g. a named pipe)"""\r
+\r
+class ExecError(EnvironmentError):\r
+ """Raised when a command could not be executed"""\r
+\r
+try:\r
+ WindowsError\r
+except NameError:\r
+ WindowsError = None\r
+\r
+def copyfileobj(fsrc, fdst, length=16*1024):\r
+ """copy data from file-like object fsrc to file-like object fdst"""\r
+ while 1:\r
+ buf = fsrc.read(length)\r
+ if not buf:\r
+ break\r
+ fdst.write(buf)\r
+\r
+def _samefile(src, dst):\r
+ # Macintosh, Unix.\r
+ if hasattr(os.path, 'samefile'):\r
+ try:\r
+ return os.path.samefile(src, dst)\r
+ except OSError:\r
+ return False\r
+\r
+ # All other platforms: check for same pathname.\r
+ return (os.path.normcase(os.path.abspath(src)) ==\r
+ os.path.normcase(os.path.abspath(dst)))\r
+\r
+def copyfile(src, dst):\r
+ """Copy data from src to dst"""\r
+ if _samefile(src, dst):\r
+ raise Error("`%s` and `%s` are the same file" % (src, dst))\r
+\r
+ for fn in [src, dst]:\r
+ try:\r
+ st = os.stat(fn)\r
+ except OSError:\r
+ # File most likely does not exist\r
+ pass\r
+ else:\r
+ # XXX What about other special files? (sockets, devices...)\r
+ if stat.S_ISFIFO(st.st_mode):\r
+ raise SpecialFileError("`%s` is a named pipe" % fn)\r
+\r
+ with open(src, 'rb') as fsrc:\r
+ with open(dst, 'wb') as fdst:\r
+ copyfileobj(fsrc, fdst)\r
+\r
+def copymode(src, dst):\r
+ """Copy mode bits from src to dst"""\r
+ if hasattr(os, 'chmod'):\r
+ st = os.stat(src)\r
+ mode = stat.S_IMODE(st.st_mode)\r
+ os.chmod(dst, mode)\r
+\r
+def copystat(src, dst):\r
+ """Copy all stat info (mode bits, atime, mtime, flags) from src to dst"""\r
+ st = os.stat(src)\r
+ mode = stat.S_IMODE(st.st_mode)\r
+ if hasattr(os, 'utime'):\r
+ os.utime(dst, (st.st_atime, st.st_mtime))\r
+ if hasattr(os, 'chmod'):\r
+ os.chmod(dst, mode)\r
+ if hasattr(os, 'chflags') and hasattr(st, 'st_flags'):\r
+ try:\r
+ os.chflags(dst, st.st_flags)\r
+ except OSError, why:\r
+ for err in 'EOPNOTSUPP', 'ENOTSUP':\r
+ if hasattr(errno, err) and why.errno == getattr(errno, err):\r
+ break\r
+ else:\r
+ raise\r
+\r
+def copy(src, dst):\r
+ """Copy data and mode bits ("cp src dst").\r
+\r
+ The destination may be a directory.\r
+\r
+ """\r
+ if os.path.isdir(dst):\r
+ dst = os.path.join(dst, os.path.basename(src))\r
+ copyfile(src, dst)\r
+ copymode(src, dst)\r
+\r
+def copy2(src, dst):\r
+ """Copy data and all stat info ("cp -p src dst").\r
+\r
+ The destination may be a directory.\r
+\r
+ """\r
+ if os.path.isdir(dst):\r
+ dst = os.path.join(dst, os.path.basename(src))\r
+ copyfile(src, dst)\r
+ copystat(src, dst)\r
+\r
+def ignore_patterns(*patterns):\r
+ """Function that can be used as copytree() ignore parameter.\r
+\r
+ Patterns is a sequence of glob-style patterns\r
+ that are used to exclude files"""\r
+ def _ignore_patterns(path, names):\r
+ ignored_names = []\r
+ for pattern in patterns:\r
+ ignored_names.extend(fnmatch.filter(names, pattern))\r
+ return set(ignored_names)\r
+ return _ignore_patterns\r
+\r
+def copytree(src, dst, symlinks=False, ignore=None):\r
+ """Recursively copy a directory tree using copy2().\r
+\r
+ The destination directory must not already exist.\r
+ If exception(s) occur, an Error is raised with a list of reasons.\r
+\r
+ If the optional symlinks flag is true, symbolic links in the\r
+ source tree result in symbolic links in the destination tree; if\r
+ it is false, the contents of the files pointed to by symbolic\r
+ links are copied.\r
+\r
+ The optional ignore argument is a callable. If given, it\r
+ is called with the `src` parameter, which is the directory\r
+ being visited by copytree(), and `names` which is the list of\r
+ `src` contents, as returned by os.listdir():\r
+\r
+ callable(src, names) -> ignored_names\r
+\r
+ Since copytree() is called recursively, the callable will be\r
+ called once for each directory that is copied. It returns a\r
+ list of names relative to the `src` directory that should\r
+ not be copied.\r
+\r
+ XXX Consider this example code rather than the ultimate tool.\r
+\r
+ """\r
+ names = os.listdir(src)\r
+ if ignore is not None:\r
+ ignored_names = ignore(src, names)\r
+ else:\r
+ ignored_names = set()\r
+\r
+ os.makedirs(dst)\r
+ errors = []\r
+ for name in names:\r
+ if name in ignored_names:\r
+ continue\r
+ srcname = os.path.join(src, name)\r
+ dstname = os.path.join(dst, name)\r
+ try:\r
+ if symlinks and os.path.islink(srcname):\r
+ linkto = os.readlink(srcname)\r
+ os.symlink(linkto, dstname)\r
+ elif os.path.isdir(srcname):\r
+ copytree(srcname, dstname, symlinks, ignore)\r
+ else:\r
+ # Will raise a SpecialFileError for unsupported file types\r
+ copy2(srcname, dstname)\r
+ # catch the Error from the recursive copytree so that we can\r
+ # continue with other files\r
+ except Error, err:\r
+ errors.extend(err.args[0])\r
+ except EnvironmentError, why:\r
+ errors.append((srcname, dstname, str(why)))\r
+ try:\r
+ copystat(src, dst)\r
+ except OSError, why:\r
+ if WindowsError is not None and isinstance(why, WindowsError):\r
+ # Copying file access times may fail on Windows\r
+ pass\r
+ else:\r
+ errors.append((src, dst, str(why)))\r
+ if errors:\r
+ raise Error, errors\r
+\r
+def rmtree(path, ignore_errors=False, onerror=None):\r
+ """Recursively delete a directory tree.\r
+\r
+ If ignore_errors is set, errors are ignored; otherwise, if onerror\r
+ is set, it is called to handle the error with arguments (func,\r
+ path, exc_info) where func is os.listdir, os.remove, or os.rmdir;\r
+ path is the argument to that function that caused it to fail; and\r
+ exc_info is a tuple returned by sys.exc_info(). If ignore_errors\r
+ is false and onerror is None, an exception is raised.\r
+\r
+ """\r
+ if ignore_errors:\r
+ def onerror(*args):\r
+ pass\r
+ elif onerror is None:\r
+ def onerror(*args):\r
+ raise\r
+ try:\r
+ if os.path.islink(path):\r
+ # symlinks to directories are forbidden, see bug #1669\r
+ raise OSError("Cannot call rmtree on a symbolic link")\r
+ except OSError:\r
+ onerror(os.path.islink, path, sys.exc_info())\r
+ # can't continue even if onerror hook returns\r
+ return\r
+ names = []\r
+ try:\r
+ names = os.listdir(path)\r
+ except os.error, err:\r
+ onerror(os.listdir, path, sys.exc_info())\r
+ for name in names:\r
+ fullname = os.path.join(path, name)\r
+ try:\r
+ mode = os.lstat(fullname).st_mode\r
+ except os.error:\r
+ mode = 0\r
+ if stat.S_ISDIR(mode):\r
+ rmtree(fullname, ignore_errors, onerror)\r
+ else:\r
+ try:\r
+ os.remove(fullname)\r
+ except os.error, err:\r
+ onerror(os.remove, fullname, sys.exc_info())\r
+ try:\r
+ os.rmdir(path)\r
+ except os.error:\r
+ onerror(os.rmdir, path, sys.exc_info())\r
+\r
+\r
+def _basename(path):\r
+ # A basename() variant which first strips the trailing slash, if present.\r
+ # Thus we always get the last component of the path, even for directories.\r
+ sep = os.path.sep + (os.path.altsep or '')\r
+ return os.path.basename(path.rstrip(sep))\r
+\r
+def move(src, dst):\r
+ """Recursively move a file or directory to another location. This is\r
+ similar to the Unix "mv" command.\r
+\r
+ If the destination is a directory or a symlink to a directory, the source\r
+ is moved inside the directory. The destination path must not already\r
+ exist.\r
+\r
+ If the destination already exists but is not a directory, it may be\r
+ overwritten depending on os.rename() semantics.\r
+\r
+ If the destination is on our current filesystem, then rename() is used.\r
+ Otherwise, src is copied to the destination and then removed.\r
+ A lot more could be done here... A look at a mv.c shows a lot of\r
+ the issues this implementation glosses over.\r
+\r
+ """\r
+ real_dst = dst\r
+ if os.path.isdir(dst):\r
+ if _samefile(src, dst):\r
+ # We might be on a case insensitive filesystem,\r
+ # perform the rename anyway.\r
+ os.rename(src, dst)\r
+ return\r
+\r
+ real_dst = os.path.join(dst, _basename(src))\r
+ if os.path.exists(real_dst):\r
+ raise Error, "Destination path '%s' already exists" % real_dst\r
+ try:\r
+ os.rename(src, real_dst)\r
+ except OSError:\r
+ if os.path.isdir(src):\r
+ if _destinsrc(src, dst):\r
+ raise Error, "Cannot move a directory '%s' into itself '%s'." % (src, dst)\r
+ copytree(src, real_dst, symlinks=True)\r
+ rmtree(src)\r
+ else:\r
+ copy2(src, real_dst)\r
+ os.unlink(src)\r
+\r
+def _destinsrc(src, dst):\r
+ src = abspath(src)\r
+ dst = abspath(dst)\r
+ if not src.endswith(os.path.sep):\r
+ src += os.path.sep\r
+ if not dst.endswith(os.path.sep):\r
+ dst += os.path.sep\r
+ return dst.startswith(src)\r
+\r
+def _get_gid(name):\r
+ """Returns a gid, given a group name."""\r
+ if getgrnam is None or name is None:\r
+ return None\r
+ try:\r
+ result = getgrnam(name)\r
+ except KeyError:\r
+ result = None\r
+ if result is not None:\r
+ return result[2]\r
+ return None\r
+\r
+def _get_uid(name):\r
+ """Returns an uid, given a user name."""\r
+ if getpwnam is None or name is None:\r
+ return None\r
+ try:\r
+ result = getpwnam(name)\r
+ except KeyError:\r
+ result = None\r
+ if result is not None:\r
+ return result[2]\r
+ return None\r
+\r
+def _make_tarball(base_name, base_dir, compress="gzip", verbose=0, dry_run=0,\r
+ owner=None, group=None, logger=None):\r
+ """Create a (possibly compressed) tar file from all the files under\r
+ 'base_dir'.\r
+\r
+ 'compress' must be "gzip" (the default), "bzip2", or None.\r
+\r
+ 'owner' and 'group' can be used to define an owner and a group for the\r
+ archive that is being built. If not provided, the current owner and group\r
+ will be used.\r
+\r
+ The output tar file will be named 'base_name' + ".tar", possibly plus\r
+ the appropriate compression extension (".gz", or ".bz2").\r
+\r
+ Returns the output filename.\r
+ """\r
+ tar_compression = {'gzip': 'gz', 'bzip2': 'bz2', None: ''}\r
+ compress_ext = {'gzip': '.gz', 'bzip2': '.bz2'}\r
+\r
+ # flags for compression program, each element of list will be an argument\r
+ if compress is not None and compress not in compress_ext.keys():\r
+ raise ValueError, \\r
+ ("bad value for 'compress': must be None, 'gzip' or 'bzip2'")\r
+\r
+ archive_name = base_name + '.tar' + compress_ext.get(compress, '')\r
+ archive_dir = os.path.dirname(archive_name)\r
+\r
+ if archive_dir and not os.path.exists(archive_dir):\r
+ if logger is not None:\r
+ logger.info("creating %s", archive_dir)\r
+ if not dry_run:\r
+ os.makedirs(archive_dir)\r
+\r
+\r
+ # creating the tarball\r
+ import tarfile # late import so Python build itself doesn't break\r
+\r
+ if logger is not None:\r
+ logger.info('Creating tar archive')\r
+\r
+ uid = _get_uid(owner)\r
+ gid = _get_gid(group)\r
+\r
+ def _set_uid_gid(tarinfo):\r
+ if gid is not None:\r
+ tarinfo.gid = gid\r
+ tarinfo.gname = group\r
+ if uid is not None:\r
+ tarinfo.uid = uid\r
+ tarinfo.uname = owner\r
+ return tarinfo\r
+\r
+ if not dry_run:\r
+ tar = tarfile.open(archive_name, 'w|%s' % tar_compression[compress])\r
+ try:\r
+ tar.add(base_dir, filter=_set_uid_gid)\r
+ finally:\r
+ tar.close()\r
+\r
+ return archive_name\r
+\r
+def _call_external_zip(base_dir, zip_filename, verbose=False, dry_run=False):\r
+ # XXX see if we want to keep an external call here\r
+ if verbose:\r
+ zipoptions = "-r"\r
+ else:\r
+ zipoptions = "-rq"\r
+ from distutils.errors import DistutilsExecError\r
+ from distutils.spawn import spawn\r
+ try:\r
+ spawn(["zip", zipoptions, zip_filename, base_dir], dry_run=dry_run)\r
+ except DistutilsExecError:\r
+ # XXX really should distinguish between "couldn't find\r
+ # external 'zip' command" and "zip failed".\r
+ raise ExecError, \\r
+ ("unable to create zip file '%s': "\r
+ "could neither import the 'zipfile' module nor "\r
+ "find a standalone zip utility") % zip_filename\r
+\r
+def _make_zipfile(base_name, base_dir, verbose=0, dry_run=0, logger=None):\r
+ """Create a zip file from all the files under 'base_dir'.\r
+\r
+ The output zip file will be named 'base_name' + ".zip". Uses either the\r
+ "zipfile" Python module (if available) or the InfoZIP "zip" utility\r
+ (if installed and found on the default search path). If neither tool is\r
+ available, raises ExecError. Returns the name of the output zip\r
+ file.\r
+ """\r
+ zip_filename = base_name + ".zip"\r
+ archive_dir = os.path.dirname(base_name)\r
+\r
+ if archive_dir and not os.path.exists(archive_dir):\r
+ if logger is not None:\r
+ logger.info("creating %s", archive_dir)\r
+ if not dry_run:\r
+ os.makedirs(archive_dir)\r
+\r
+ # If zipfile module is not available, try spawning an external 'zip'\r
+ # command.\r
+ try:\r
+ import zipfile\r
+ except ImportError:\r
+ zipfile = None\r
+\r
+ if zipfile is None:\r
+ _call_external_zip(base_dir, zip_filename, verbose, dry_run)\r
+ else:\r
+ if logger is not None:\r
+ logger.info("creating '%s' and adding '%s' to it",\r
+ zip_filename, base_dir)\r
+\r
+ if not dry_run:\r
+ with zipfile.ZipFile(zip_filename, "w",\r
+ compression=zipfile.ZIP_DEFLATED) as zf:\r
+ for dirpath, dirnames, filenames in os.walk(base_dir):\r
+ for name in filenames:\r
+ path = os.path.normpath(os.path.join(dirpath, name))\r
+ if os.path.isfile(path):\r
+ zf.write(path, path)\r
+ if logger is not None:\r
+ logger.info("adding '%s'", path)\r
+\r
+ return zip_filename\r
+\r
+_ARCHIVE_FORMATS = {\r
+ 'gztar': (_make_tarball, [('compress', 'gzip')], "gzip'ed tar-file"),\r
+ 'bztar': (_make_tarball, [('compress', 'bzip2')], "bzip2'ed tar-file"),\r
+ 'tar': (_make_tarball, [('compress', None)], "uncompressed tar file"),\r
+ 'zip': (_make_zipfile, [],"ZIP file")\r
+ }\r
+\r
+def get_archive_formats():\r
+ """Returns a list of supported formats for archiving and unarchiving.\r
+\r
+ Each element of the returned sequence is a tuple (name, description)\r
+ """\r
+ formats = [(name, registry[2]) for name, registry in\r
+ _ARCHIVE_FORMATS.items()]\r
+ formats.sort()\r
+ return formats\r
+\r
+def register_archive_format(name, function, extra_args=None, description=''):\r
+ """Registers an archive format.\r
+\r
+ name is the name of the format. function is the callable that will be\r
+ used to create archives. If provided, extra_args is a sequence of\r
+ (name, value) tuples that will be passed as arguments to the callable.\r
+ description can be provided to describe the format, and will be returned\r
+ by the get_archive_formats() function.\r
+ """\r
+ if extra_args is None:\r
+ extra_args = []\r
+ if not isinstance(function, collections.Callable):\r
+ raise TypeError('The %s object is not callable' % function)\r
+ if not isinstance(extra_args, (tuple, list)):\r
+ raise TypeError('extra_args needs to be a sequence')\r
+ for element in extra_args:\r
+ if not isinstance(element, (tuple, list)) or len(element) !=2 :\r
+ raise TypeError('extra_args elements are : (arg_name, value)')\r
+\r
+ _ARCHIVE_FORMATS[name] = (function, extra_args, description)\r
+\r
+def unregister_archive_format(name):\r
+ del _ARCHIVE_FORMATS[name]\r
+\r
+def make_archive(base_name, format, root_dir=None, base_dir=None, verbose=0,\r
+ dry_run=0, owner=None, group=None, logger=None):\r
+ """Create an archive file (eg. zip or tar).\r
+\r
+ 'base_name' is the name of the file to create, minus any format-specific\r
+ extension; 'format' is the archive format: one of "zip", "tar", "bztar"\r
+ or "gztar".\r
+\r
+ 'root_dir' is a directory that will be the root directory of the\r
+ archive; ie. we typically chdir into 'root_dir' before creating the\r
+ archive. 'base_dir' is the directory where we start archiving from;\r
+ ie. 'base_dir' will be the common prefix of all files and\r
+ directories in the archive. 'root_dir' and 'base_dir' both default\r
+ to the current directory. Returns the name of the archive file.\r
+\r
+ 'owner' and 'group' are used when creating a tar archive. By default,\r
+ uses the current owner and group.\r
+ """\r
+ save_cwd = os.getcwd()\r
+ if root_dir is not None:\r
+ if logger is not None:\r
+ logger.debug("changing into '%s'", root_dir)\r
+ base_name = os.path.abspath(base_name)\r
+ if not dry_run:\r
+ os.chdir(root_dir)\r
+\r
+ if base_dir is None:\r
+ base_dir = os.curdir\r
+\r
+ kwargs = {'dry_run': dry_run, 'logger': logger}\r
+\r
+ try:\r
+ format_info = _ARCHIVE_FORMATS[format]\r
+ except KeyError:\r
+ raise ValueError, "unknown archive format '%s'" % format\r
+\r
+ func = format_info[0]\r
+ for arg, val in format_info[1]:\r
+ kwargs[arg] = val\r
+\r
+ if format != 'zip':\r
+ kwargs['owner'] = owner\r
+ kwargs['group'] = group\r
+\r
+ try:\r
+ filename = func(base_name, base_dir, **kwargs)\r
+ finally:\r
+ if root_dir is not None:\r
+ if logger is not None:\r
+ logger.debug("changing back to '%s'", save_cwd)\r
+ os.chdir(save_cwd)\r
+\r
+ return filename\r
projects / mirror_edk2.git / blobdiff