本文共 16434 字，大约阅读时间需要 54 分钟。
shutil 模块是 Python 标准库中用于文件操作的强大工具，提供了丰富的文件拷贝、目录管理和其他文件操作功能。以下是 shutil 模块的主要功能及其使用方法。
copyfileobj 用于将文件对象（如文件流）逐次拷贝到目标文件中。每次拷贝的默认大小为 16KB。
示例：
def copyfileobj(fsrc, fdst, length=16*1024): """copy data from file-like object fsrc to file-like object fdst""" while True: buf = fsrc.read(length) if not buf: break fdst.write(buf)
copyfile 用于拷贝文件到目标路径。如果 follow_symlinks 未设置且源文件是符号链接，拷贝时会拷贝链接本身而非指向的文件。
参数：
follow_symlinks：控制是否跟随符号链接，默认为 True。示例：
def copyfile(src, dst, *, follow_symlinks=True): """Copy data from src to dst. If follow_symlinks is not set and src is a symbolic link, a new symlink will be created instead of copying the file it points to.""" if _samefile(src, dst): raise SameFileError("{!r} and {!r} are the same file".format(src, dst)) for fn in [src, dst]: try: st = os.stat(fn) except OSError: pass else: if stat.S_ISFIFO(st.st_mode): raise SpecialFileError("`%s` is a named pipe".format(fn)) if not follow_symlinks and os.path.islink(src): os.symlink(os.readlink(src), dst) else: with open(src, 'rb') as fsrc: with open(dst, 'wb') as fdst: copyfileobj(fsrc, fdst) return dst copymode 用于拷贝文件的权限信息（如权限位、用户、组等）。如果 follow_symlinks 未设置且源和目标都是符号链接且系统支持 lchmod，则创建新的符号链接。
示例：
def copymode(src, dst, *, follow_symlinks=True): """Copy mode bits from src to dst. If follow_symlinks is not set, symlinks aren't followed if and only if both `src` and `dst` are symlinks. If `lchmod` isn't available (e.g. Linux) this method does nothing.""" if not follow_symlinks and os.path.islink(src) and os.path.islink(dst): if hasattr(os, 'lchmod'): stat_func, chmod_func = os.lstat, os.lchmod else: return elif hasattr(os, 'chmod'): stat_func, chmod_func = os.stat, os.chmod else: return st = stat_func(src) chmod_func(dst, stat.S_IMODE(st.st_mode))
copystat 用于拷贝文件的状态信息（如权限位、访问时间、修改时间等）。如果 follow_symlinks 未设置且源和目标都是符号链接且系统支持 lchown，则创建新的符号链接。
示例：
def copystat(src, dst, *, follow_symlinks=True): """Copy all stat info (mode bits, atime, mtime, flags) from src to dst. If the optional flag `follow_symlinks` is not set, symlinks aren't followed if and only if both `src` and `dst` are symlinks.""" def _nop(*args, ns=None, follow_symlinks=None): pass follow = follow_symlinks or not (os.path.islink(src) and os.path.islink(dst)) if follow: def lookup(name): return getattr(os, name, _nop) else: def lookup(name): fn = getattr(os, name, _nop) if fn in os.supports_follow_symlinks: return fn return _nop st = lookup("stat")(src, follow_symlinks=follow) mode = stat.S_IMODE(st.st_mode) lookup("utime")(dst, ns=(st.st_atime_ns, st.st_mtime_ns), follow_symlinks=follow) try: lookup("chmod")(dst, mode, follow_symlinks=follow) except NotImplementedError: pass if hasattr(st, 'st_flags'): try: lookup("chflags")(dst, st.st_flags, follow_symlinks=follow) except OSError as why: for err in 'EOPNOTSUPP', 'ENOTSUP': if hasattr(errno, err) and why.errno == getattr(errno, err): break else: raise _copyxattr(src, dst, follow_symlinks=follow) copy 用于完整拷贝文件和状态信息，默认使用 copy2 作为拷贝函数。
示例：
def copy(src, dst, *, follow_symlinks=True): """Copy data and mode bits ("cp src dst"). Return the file's destination. The destination may be a directory.""" if os.path.isdir(dst): dst = os.path.join(dst, os.path.basename(src)) copyfile(src, dst, follow_symlinks=follow_symlinks) copymode(src, dst, follow_symlinks=follow_symlinks) return dst copy2 用于完整拷贝文件和状态信息，类似于 cp -p 的行为。
示例：
def copy2(src, dst, *, follow_symlinks=True): """Copy data and all stat info ("cp -p src dst"). Return the file's destination.""" if os.path.isdir(dst): dst = os.path.join(dst, os.path.basename(src)) copyfile(src, dst, follow_symlinks=follow_symlinks) copystat(src, dst, follow_symlinks=follow_symlinks) return dst ignore_patterns 用于忽略特定文件或目录，可用于递归拷贝时排除不需要处理的文件。
示例：
def ignore_patterns(*patterns): """Function that can be used as copytree() ignore parameter. Patterns is a sequence of glob-style patterns that are used to exclude files.""" def _ignore_patterns(path, names): ignored_names = [] for pattern in patterns: ignored_names.extend(fnmatch.filter(names, pattern)) return set(ignored_names) return _ignore_patterns
copytree 是递归拷贝目录树的核心函数，支持符号链接和忽略特定文件。
示例：
def copytree(src, dst, symlinks=False, ignore=None, copy_function=copy2, ignore_dangling_symlinks=False): """Recursively copy a directory tree. The destination directory must not already exist.""" names = os.listdir(src) if ignore is not None: ignored_names = ignore(src, names) else: ignored_names = set() os.makedirs(dst) errors = [] for name in names: if name in ignored_names: continue srcname = os.path.join(src, name) dstname = os.path.join(dst, name) try: if os.path.islink(srcname): linkto = os.readlink(srcname) if symlinks: os.symlink(linkto, dstname) copystat(srcname, dstname, follow_symlinks=not symlinks) else: if not os.path.exists(linkto) and ignore_dangling_symlinks: continue if os.path.isdir(srcname): copytree(srcname, dstname, symlinks, ignore, copy_function) else: copy_function(srcname, dstname) elif os.path.isdir(srcname): copytree(srcname, dstname, symlinks, ignore, copy_function) else: copy_function(srcname, dstname) except Error as err: errors.extend(err.args[0]) except OSError as why: errors.append((srcname, dstname, str(why))) try: copystat(src, dst) except OSError as why: if getattr(why, 'winerror', None) is None: errors.append((src, dst, str(why))) if errors: raise Error(errors) return dst
rmtree 递归删除目录树，支持忽略错误和自定义错误处理函数。
示例：
def rmtree(path, ignore_errors=False, onerror=None): """Recursively delete a directory tree. If ignore_errors is set, errors are ignored; otherwise, if onerror is set, it is called to handle the error.""" if ignore_errors: def onerror(*args): pass elif onerror is None: def onerror(*args): raise try: orig_st = os.lstat(path) except Exception: onerror(os.lstat, path, sys.exc_info()) return try: fd = os.open(path, os.O_RDONLY) except Exception: onerror(os.lstat, path, sys.exc_info()) return try: if os.path.samestat(orig_st, os.fstat(fd)): _rmtree_safe_fd(fd, path, onerror) try: os.rmdir(path) except OSError: onerror(os.rmdir, path, sys.exc_info()) except OSError: if os.path.islink(path): raise OSError("Cannot call rmtree on a symbolic link") onerror(os.path.islink, path, sys.exc_info()) finally: os.close(fd) move 递归移动文件或目录，类似于 mv 命令。
示例：
def move(src, dst, copy_function=copy2): """Recursively move a file or directory to another location. This is similar to the Unix "mv" command.""" real_dst = dst if os.path.isdir(dst): real_dst = os.path.join(dst, _basename(src)) if os.path.exists(real_dst): raise Error("Destination path '%s' already exists" % real_dst) try: os.rename(src, real_dst) except OSError: if os.path.islink(src): linkto = os.readlink(src) os.symlink(linkto, real_dst) os.unlink(src) elif os.path.isdir(src): if _destinsrc(src, dst): raise Error("Cannot move a directory '%s' into itself '%s'." % (src, dst)) copytree(src, real_dst, copy_function=copy_function, symlinks=True) rmtree(src) else: copy_function(src, real_dst) os.unlink(src) return real_dst get_archive_formats 返回支持存档的格式列表。
示例：
def get_archive_formats(): """Returns a list of supported formats for archiving and unarchiving. Each element of the returned sequence is a tuple (name, description).""" formats = [(name, registry[2]) for name, registry in _ARCHIVE_FORMATS.items()] formats.sort() return formats
注册自定义存档格式。
示例：
def register_archive_format(name, function, extra_args=None, description=''): """Registers an archive format. name is the name of the format. function is the callable that will be used to create archives. If provided, extra_args is a sequence of (name, value) tuples that will be passed as arguments to the callable. description can be provided to describe the format, and will be returned by the get_archive_formats() function.""" if extra_args is None: extra_args = [] if not callable(function): raise TypeError('The %s object is not callable' % function) if not isinstance(extra_args, (tuple, list)): raise TypeError('extra_args needs to be a sequence') for element in extra_args: if not isinstance(element, (tuple, list)) or len(element) != 2: raise TypeError('extra_args elements are : (arg_name, value)') _ARCHIVE_FORMATS[name] = (function, extra_args, description) 取消注册的存档格式。
示例：
def unregister_archive_format(name): """Delete the archive format registration for the specified name.""" if name in _ARCHIVE_FORMATS: del _ARCHIVE_FORMATS[name]
创建压缩包。
示例：
def make_archive(base_name, format, root_dir=None, base_dir=None, verbose=0, dry_run=0, owner=None, group=None, logger=None): """Create an archive file (eg. zip or tar). 'base_name' is the name of the file to create, minus any format-specific extension; 'format' is the archive format: one of "zip", "tar", "bztar" or "gztar". Returns the name of the archive file.""" save_cwd = os.getcwd() if root_dir is not None: if logger is not None: logger.debug("changing into '%s'", root_dir) base_name = os.path.abspath(base_name) if not dry_run: os.chdir(root_dir) if base_dir is None: base_dir = os.curdir kwargs = {} if dry_run: kwargs['dry_run'] = dry_run kwargs['logger'] = logger try: format_info = _ARCHIVE_FORMATS[format] except KeyError: raise ValueError("unknown archive format '%s'" % format) func = format_info[0] for arg, val in format_info[1]: kwargs[arg] = val if format != 'zip': kwargs['owner'] = owner kwargs['group'] = group try: filename = func(base_name, base_dir, **kwargs) finally: if root_dir is not None: if logger is not None: logger.debug("changing back to '%s'", save_cwd) os.chdir(save_cwd) return filename 返回支持解压的格式列表。
示例：
def get_unpack_formats(): """Returns a list of supported formats for unarchiving.""" formats = [(name, registry[1]) for name, registry in _UNPACK_FORMATS.items()] formats.sort() return formats
注册自定义解压格式。
示例：
def register_unpack_format(name, extensions, function, extra_args=None, description=''): """Register an unpack format. name is the format name, extensions is a list of file extensions that identify this format, function is the unpack function, extra_args is a list of (name, value) tuples for the function's arguments, and description is the format description.""" if not isinstance(extensions, (list, tuple)): raise TypeError('extensions must be a list or tuple') if not callable(function): raise TypeError('The %s object is not callable' % function) _UNPACK_FORMATS[name] = (extensions, function, extra_args, description) 取消注册的解压格式。
示例：
def unregister_unpack_format(name): """Unregister the unpack format for the specified name.""" if name in _UNPACK_FORMATS: del _UNPACK_FORMATS[name]
解压指定的存档文件。
示例：
def unpack_archive(filename, extract_dir=None, format=None): """Unpack an archive. `filename` is the name of the archive. `extract_dir` is the name of the target directory, where the archive is unpacked. If not provided, the current working directory is used. `format` is the archive format: one of "zip", "tar", or "gztar". Or any other registered format. If not provided, unpack_archive will use the filename extension and see if an unpacker was registered for that extension. If none is found, a ValueError is raised.""" if extract_dir is None: extract_dir = os.getcwd() if format is not None: try: format_info = _UNPACK_FORMATS[format] except KeyError: raise ValueError("Unknown unpack format '{0}'".format(format)) func = format_info[1] func(filename, extract_dir, **dict(format_info[2])) else: format = _find_unpack_format(filename) if format is None: raise ReadError("Unknown archive format '{0}'".format(filename)) func = _UNPACK_FORMATS[format][1] kwargs = dict(_UNPACK_FORMATS[format][2]) func(filename, extract_dir, **kwargs) chown 用于改变文件或目录的所有者和所属组。
示例：
def chown(path, user=None, group=None): """Change owner user and group of the given path. user and group can be the uid/gid or the user/group names, and in that case, they are converted to their respective uid/gid.""" if user is None and group is None: raise ValueError("user and/or group must be set") _user = user _group = group if user is None: _user = -1 elif isinstance(user, str): _user = _get_uid(user) if _user is None: raise LookupError("no such user: {!r}".format(user)) if group is None: _group = -1 elif not isinstance(group, int): _group = _get_gid(group) if _group is None: raise LookupError("no such group: {!r}".format(group)) os.chown(path, _user, _group) 获取终端窗口大小。
示例：
def get_terminal_size(fallback=(80, 24)): """Get the terminal size. If the terminal cannot be queried, use the fallback value.""" try: import termios import os size = termios_terminal_size() return (size.ws_colnum, size.ws_rows) except ImportError: return fallback
查找命令路径。
示例：
def which(cmd, mode=os.F_OK | os.X_OK, path=None): """Return the path of the command. If path is None, use os.environ.get('PATH'). If no such command exists, return None.""" if path is None: path = os.environ.get('PATH', '') if not path: return None for p in os.pathsep.split(path): cmd_path = os.path.join(p, cmd) if not os.path.isabs(cmd_path): cmd_path = os.path.abspath(cmd_path) if os.path.isfile(cmd_path): if (os.access(cmd_path, mode) if mode else os.path.exists(cmd_path)): return cmd_path return None 通过这些函数，开发者可以高效地处理文件和目录的拷贝、移动、解压、管理权限等操作，shutil 模块为 Python 生态系统提供了强大的文件操作工具。
转载地址：http://sulfk.baihongyu.com/