程序员要最先成为AI的主人 AI高手速成 DeepSeek让你工作变轻松 deepseek从入门到精通实战指南人工智能 异步图书出品

---

一、基础操作

1. 文件复制

import shutil

# 复制文件内容（不保留元数据）
shutil.copy('source.txt', 'dest.txt')

# 复制文件内容+元数据（保留修改时间等）
shutil.copy2('source.txt', 'metadata_dest.txt')

# 复制权限（需配合其他操作使用）
shutil.copymode('source.txt', 'target.txt')

2. 目录操作

# 递归复制目录（目标目录必须不存在）
shutil.copytree('src_dir', 'dst_dir')

# 移动目录/文件（可跨文件系统）
shutil.move('old_location', 'new_location')

# 递归删除目录（谨慎使用！）
shutil.rmtree('directory_to_delete')

3. 文件打包

# 创建归档文件（支持 zip/tar/gztar 等格式）
shutil.make_archive('archive', 'zip', 'src_dir')

# 解压归档文件
shutil.unpack_archive('archive.zip', 'extract_dir')

---

二、进阶技巧

1. 过滤文件复制

# 使用 ignore 参数过滤文件
def ignore_func(dirname, filenames):
    return [f for f in filenames if f.endswith('.tmp')]

shutil.copytree('src', 'dst', ignore=ignore_func)

# 使用预定义忽略模式
shutil.copytree('src', 'dst', ignore=shutil.ignore_patterns('*.tmp', '*.log'))

2. 符号链接处理

# 保留符号链接（默认复制链接指向的内容）
shutil.copytree('src', 'dst', symlinks=True)

# 创建符号链接（需配合 os 模块）
import os
os.symlink('target', 'link_name')  # shutil 不直接提供此功能

3. 异常处理

try:
    shutil.copytree('src', 'dst')
except shutil.Error as e:
    print(f'Directory not copied. Error: {e}')
except OSError as e:
    print(f'Operation failed: {e.strerror}')

---

三、高级应用

1. 自定义复制函数

def custom_copy(src, dst):
    print(f'Copying {src} => {dst}')
    # 使用默认复制逻辑
    return shutil.copy2(src, dst)

shutil.copytree('src', 'dst', copy_function=custom_copy)

2. 文件系统监控

class ProgressTracker:
    def __init__(self):
        self.total = 0
    
    def __call__(self, path, names):
        self.total += len(names)
        print(f'Processed {self.total} items')
        return []

tracker = ProgressTracker()
shutil.copytree('src', 'dst', ignore=tracker)

3. 跨设备复制优化

def cross_device_copy(src, dst):
    # 处理不同文件系统间的复制
    try:
        shutil.copy2(src, dst)
        shutil.copystat(src, dst)
    except OSError as e:
        # 处理权限问题
        print(f'Skipped special handling: {e}')

shutil.copytree('/mnt/nfs/src', '/local/dst', copy_function=cross_device_copy)

4. 高级元数据处理

from pathlib import Path

def preserve_metadata(src, dst):
    st = os.stat(src)
    # 自定义元数据处理
    Path(dst).chmod(st.st_mode)
    os.utime(dst, (st.st_atime, st.st_mtime))
    # 保留扩展属性（需要xattr支持）
    try:
        xattrs = os.listxattr(src)
        for attr in xattrs:
            os.setxattr(dst, attr, os.getxattr(src, attr))
    except OSError:
        pass

shutil.copytree('src', 'dst', copy_function=preserve_metadata)

---

四、最佳实践

1. 路径处理：始终使用 os.path 或 pathlib 处理路径
2. 错误处理：对以下操作必须添加异常处理：
   • rmtree() 删除操作
   • 跨文件系统操作
   • 处理用户输入路径
3. 性能优化：

   # 使用更快的方法复制大文件
   def fast_copy(src, dst):
       with open(src, 'rb') as fsrc:
           with open(dst, 'wb') as fdst:
               shutil.copyfileobj(fsrc, fdst, length=16*1024*1024)  # 16MB 块
   

---

五、常见问题解决方案

1. 目标目录存在时的处理

def safe_copytree(src, dst):
    if os.path.exists(dst):
        shutil.rmtree(dst)
    shutil.copytree(src, dst)

2. 保留硬链接结构

from collections import defaultdict

link_map = defaultdict(list)

def preserve_hardlinks(src, dst):
    st = os.stat(src)
    ino = st.st_ino
    if ino in link_map:
        os.link(link_map[ino][0], dst)
        return dst
    link_map[ino].append(dst)
    return shutil.copy2(src, dst)

shutil.copytree('src', 'dst', copy_function=preserve_hardlinks)

---

六、版本注意事项

1. Python 3.8+ 新增：
   • shutil.copytree(dir_exist_ok=True) 允许目标目录存在
   • shutil.which() 增强
2. Python 3.10+：
   • 改进的异常信息
   • 更好的符号链接处理

---

---

七、深入扩展

1. 特殊文件类型处理

处理设备文件/命名管道

def copy_special_files(src, dst):
    """处理特殊文件（需要管理员权限）"""
    st = os.stat(src)
    if stat.S_ISFIFO(st.st_mode):
        os.mkfifo(dst, st.st_mode)
    elif stat.S_ISSOCK(st.st_mode):
        # 注意：无法直接复制套接字
        raise ValueError("Cannot copy sockets")
    elif stat.S_ISBLK(st.st_mode) or stat.S_ISCHR(st.st_mode):
        os.mknod(dst, st.st_mode, os.makedev(
            os.major(st.st_rdev), 
            os.minor(st.st_rdev)
        )
    else:
        return shutil.copy2(src, dst)
    shutil.copystat(src, dst)
    return dst

shutil.copytree('/dev', 'dev_backup', copy_function=copy_special_files)

2. 跨平台兼容性处理

Windows 长路径支持

# 在Windows启用长路径支持（需Python 3.6+）
if os.name == 'nt':
    import ctypes
    kernel32 = ctypes.windll.kernel32
    kernel32.SetFileAttributesW.argtypes = [ctypes.c_wchar_p, ctypes.c_uint32]
    kernel32.SetFileAttributesW.restype = ctypes.c_uint32
    FILE_ATTRIBUTE_NORMAL = 0x80
    kernel32.SetFileAttributesW(r'\\?\\C:\超长路径', FILE_ATTRIBUTE_NORMAL)

符号链接兼容性

def safe_symlink(src, dst):
    """跨平台符号链接处理"""
    if os.name == 'nt':
        # Windows需要特殊权限或开发者模式
        if os.path.isdir(src):
            shutil.copytree(src, dst, symlinks=True)
        else:
            shutil.copy(src, dst)
    else:
        os.symlink(src, dst)

3. 性能优化实战

多线程大文件复制

from concurrent.futures import ThreadPoolExecutor
import os

def parallel_copytree(src, dst, workers=4):
    os.makedirs(dst, exist_ok=True)
    with ThreadPoolExecutor(max_workers=workers) as executor:
        for root, dirs, files in os.walk(src):
            rel_path = os.path.relpath(root, src)
            dest_dir = os.path.join(dst, rel_path)
            os.makedirs(dest_dir, exist_ok=True)
            
            for file in files:
                src_file = os.path.join(root, file)
                dest_file = os.path.join(dest_dir, file)
                executor.submit(shutil.copy2, src_file, dest_file)

4. 安全增强

路径消毒处理

from pathlib import Path

def sanitized_copy(src, dst):
    """防止路径遍历攻击"""
    src_path = Path(src).resolve()
    dst_path = Path(dst).resolve()
    
    if not src_path.exists():
        raise ValueError("源路径不存在")
    
    # 禁止向上层目录操作
    if str(dst_path).startswith('../'):
        raise PermissionError("非法目标路径")
        
    return shutil.copy2(src_path, dst_path)

5. 高级监控集成

实时进度显示

class ProgressTracker:
    def __init__(self, total):
        self.total = total
        self.count = 0
        self.lock = threading.Lock()
    
    def update(self, processed):
        with self.lock:
            self.count += processed
            percent = self.count / self.total * 100
            print(f"\rProgress: {percent:.1f}%", end='')

def monitored_copy(src, dst, *, follow_symlinks=True):
    file_size = os.path.getsize(src)
    tracker = ProgressTracker(file_size)
    
    def callback(bytes_copied):
        tracker.update(bytes_copied)
    
    with open(src, 'rb') as fsrc:
        with open(dst, 'wb') as fdst:
            shutil.copyfileobj(fsrc, fdst, length=16*1024, callback=callback)
    
    shutil.copystat(src, dst)
    print("\nCopy complete!")

6. 文件系统特性保留

保留稀疏文件特性

def copy_sparse_file(src, dst):
    """高效复制稀疏文件（适用于虚拟磁盘文件等）"""
    with open(src, 'rb') as fsrc:
        with open(dst, 'wb') as fdst:
            # 检测稀疏区块
            blocksize = 4096
            while True:
                chunk = fsrc.read(blocksize)
                if not chunk:
                    break
                if all(b == b'\x00' for b in chunk):
                    fdst.seek(blocksize, os.SEEK_CUR)
                else:
                    fdst.write(chunk)
    shutil.copystat(src, dst)

---

八、调试与测试

1. 模拟文件操作错误

from unittest.mock import patch

def test_rmtree_failure():
    with patch('shutil.rmtree', side_effect=OSError('模拟错误')) as mock_rm:
        try:
            shutil.rmtree('/fake/path')
        except OSError as e:
            assert str(e) == '模拟错误'

2. 文件操作日志记录

import logging

logging.basicConfig(level=logging.INFO)

class LoggedCopyHandler:
    def __init__(self, original=shutil.copy2):
        self.original = original
    
    def __call__(self, src, dst):
        logging.info(f"Copying {src} -> {dst}")
        return self.original(src, dst)

shutil.copytree('src', 'dst', copy_function=LoggedCopyHandler())

---

九、底层机制解析

1. 文件复制过程分解

def manual_file_copy(src, dst):
    """手动实现文件复制流程"""
    # 1. 打开文件
    with open(src, 'rb') as fsrc:
        with open(dst, 'wb') as fdst:
            # 2. 读取缓冲区
            while True:
                buf = fsrc.read(16*1024)
                if not buf:
                    break
                # 3. 写入目标
                fdst.write(buf)
    # 4. 复制元数据
    shutil.copystat(src, dst)

2. shutil 模块架构

shutil
├── 文件操作核心
│   ├── copyfileobj (带缓冲区的底层复制)
│   ├── copyfile (基于copyfileobj)
│   └── copymode (权限复制)
├── 目录操作
│   ├── copytree (递归复制)
│   └── rmtree (递归删除)
└── 归档工具
    ├── make_archive (打包)
    └── unpack_archive (解包)

---

十、行业应用案例

1. 自动化备份系统

import datetime

def create_backup(source_dir, backup_root):
    timestamp = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
    backup_name = f"backup_{timestamp}.zip"
    backup_path = os.path.join(backup_root, backup_name)
    
    # 创建带时间戳的归档
    shutil.make_archive(
        base_name=os.path.join(backup_root, f"temp_{timestamp}"),
        format='zip',
        root_dir=source_dir
    )
    
    # 重命名避免中间文件
    os.rename(f"temp_{timestamp}.zip", backup_path)
    print(f"Backup created: {backup_path}")
    return backup_path

2. 分布式文件同步

import hashlib

def sync_directories(src, dst):
    # 1. 建立源文件索引
    src_files = {}
    for root, _, files in os.walk(src):
        for file in files:
            path = os.path.join(root, file)
            with open(path, 'rb') as f:
                src_files[path] = hashlib.md5(f.read()).hexdigest()
    
    # 2. 同步操作
    for src_path, checksum in src_files.items():
        rel_path = os.path.relpath(src_path, src)
        dst_path = os.path.join(dst, rel_path)
        
        if not os.path.exists(dst_path):
            shutil.copy2(src_path, dst_path)
        else:
            with open(dst_path, 'rb') as f:
                dst_checksum = hashlib.md5(f.read()).hexdigest()
            if checksum != dst_checksum:
                shutil.copy2(src_path, dst_path)

---

十一、版本新特性

Python 3.10+ 改进

# 更安全的目录删除
shutil.rmtree('path', ignore_errors=False, onerror=handle_error)

def handle_error(func, path, exc_info):
    """自定义错误处理"""
    import stat
    # 尝试解除只读属性
    os.chmod(path, stat.S_IWRITE)
    func(path)

Python 3.11+ 新增

# 高性能拷贝（使用平台特定优化）
shutil.copyfile(src, dst, follow_symlinks=False, 
               use_fastcopy=True)

# 增强的归档支持
shutil.register_archive_format(
    name='myformat',
    function=make_my_archive,
    description='My custom format',
    extra_args=[('level', 6)]
)

---

十二、故障排除指南

问题现象	解决方案
PermissionError 权限拒绝	使用copystat前检查文件属性
FileNotFoundError	使用os.path.exists预先验证路径
符号链接复制失败	设置symlinks=True参数
跨设备复制元数据丢失	手动调用copystat进行补充
文件名编码错误	使用surrogateescape错误处理
磁盘空间不足	添加disk_usage检查

# 磁盘空间检查示例
def check_disk_space(path, required):
    usage = shutil.disk_usage(path)
    if usage.free < required:
        raise RuntimeError(f"需要 {required} 字节，仅剩 {usage.free}")
    return True

---

以上补充内容涵盖了特殊场景处理、性能优化、安全增强、底层原理分析等深度主题，并提供了行业实际应用案例和最新版本特性说明。根据具体需求，可以组合这些技术构建健壮的文件管理系统。