Python批量压缩案例如何批量压缩文件

本文目录导读：

1. 使用zipfile模块压缩文件夹
2. 批量压缩指定文件列表
3. 高级版本：支持多格式压缩
4. 并行压缩（提高效率）
5. 实用案例：压缩指定类型的文件
6. 注意事项

我来为您介绍几种Python批量压缩文件的方法：

使用zipfile模块压缩文件夹

import zipfile
import os
from pathlib import Path
def batch_zip_folders(source_dir, output_dir=None):
    """
    批量压缩文件夹中的每个子文件夹
    Args:
        source_dir: 源目录路径
        output_dir: 输出目录路径（默认与源目录相同）
    """
    source_path = Path(source_dir)
    if output_dir is None:
        output_dir = source_dir
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)
    # 遍历所有子文件夹
    for folder_path in source_path.iterdir():
        if folder_path.is_dir():
            # 创建zip文件名
            zip_name = f"{folder_path.name}.zip"
            zip_path = output_path / zip_name
            with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
                # 遍历文件夹中的所有文件
                for file_path in folder_path.rglob('*'):
                    if file_path.is_file():
                        # 计算相对路径，保持目录结构
                        arcname = file_path.relative_to(folder_path)
                        zipf.write(file_path, arcname)
            print(f"已压缩: {folder_path.name} -> {zip_name}")
# 使用示例
batch_zip_folders("/path/to/source/folder")

批量压缩指定文件列表

import zipfile
from pathlib import Path
def batch_zip_files(file_list, output_zip_name="batch.zip"):
    """
    将多个文件压缩到同一个zip文件中
    Args:
        file_list: 文件路径列表
        output_zip_name: 输出zip文件名
    """
    with zipfile.ZipFile(output_zip_name, 'w', zipfile.ZIP_DEFLATED) as zipf:
        for file_path in file_list:
            file_path = Path(file_path)
            if file_path.exists():
                # 使用文件名作为存档名称
                zipf.write(file_path, file_path.name)
                print(f"已添加: {file_path.name}")
            else:
                print(f"文件不存在: {file_path}")
# 使用示例
files_to_compress = [
    "/path/to/file1.txt",
    "/path/to/file2.jpg", 
    "/path/to/file3.pdf"
]
# 压缩到同一个文件
batch_zip_files(files_to_compress, "my_files.zip")

高级版本：支持多格式压缩

import zipfile
import tarfile
import gzip
import shutil
from pathlib import Path
from typing import Union, List
class BatchCompressor:
    """批量压缩工具类"""
    def __init__(self, compression_type='zip'):
        """
        初始化压缩器
        Args:
            compression_type: 'zip', 'tar', 'gz', 'targz'
        """
        self.compression_type = compression_type
    def compress_single_folder(self, folder_path: Path, output_path: Path):
        """压缩单个文件夹"""
        folder_path = Path(folder_path)
        output_path = Path(output_path)
        # 确保输出目录存在
        output_path.parent.mkdir(parents=True, exist_ok=True)
        if self.compression_type == 'zip':
            self._compress_to_zip(folder_path, output_path)
        elif self.compression_type == 'tar':
            self._compress_to_tar(folder_path, output_path)
        elif self.compression_type == 'targz':
            self._compress_to_targz(folder_path, output_path)
        elif self.compression_type == 'gz':
            self._compress_to_gz(folder_path, output_path)
    def _compress_to_zip(self, folder_path: Path, output_path: Path):
        """压缩为zip格式"""
        with zipfile.ZipFile(output_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
            for file_path in folder_path.rglob('*'):
                if file_path.is_file():
                    arcname = file_path.relative_to(folder_path)
                    zipf.write(file_path, arcname)
    def _compress_to_tar(self, folder_path: Path, output_path: Path):
        """压缩为tar格式"""
        with tarfile.open(output_path, 'w') as tar:
            tar.add(folder_path, arcname=folder_path.name)
    def _compress_to_targz(self, folder_path: Path, output_path: Path):
        """压缩为tar.gz格式"""
        with tarfile.open(output_path, 'w:gz') as tar:
            tar.add(folder_path, arcname=folder_path.name)
    def _compress_to_gz(self, folder_path: Path, output_path: Path):
        """压缩为gz格式（仅压缩文件，不保留目录结构）"""
        with open(folder_path, 'rb') as f_in:
            with gzip.open(output_path, 'wb') as f_out:
                shutil.copyfileobj(f_in, f_out)
    def batch_compress(self, source_dir: Union[str, Path], 
                      output_dir: Union[str, Path] = None,
                      pattern: str = "*"):
        """
        批量压缩文件夹
        Args:
            source_dir: 源目录
            output_dir: 输出目录
            pattern: 文件匹配模式（如 "*.txt"）
        """
        source_path = Path(source_dir)
        if output_dir is None:
            output_dir = source_path
        output_path = Path(output_dir)
        output_path.mkdir(parents=True, exist_ok=True)
        compressed_count = 0
        for folder_path in source_path.iterdir():
            if folder_path.is_dir():
                # 生成输出文件名
                ext = {
                    'zip': '.zip',
                    'tar': '.tar',
                    'targz': '.tar.gz',
                    'gz': '.gz'
                }.get(self.compression_type, '.zip')
                output_file = output_path / f"{folder_path.name}{ext}"
                try:
                    self.compress_single_folder(folder_path, output_file)
                    print(f"✅ 已压缩: {folder_path.name} -> {output_file.name}")
                    compressed_count += 1
                except Exception as e:
                    print(f"❌ 压缩失败: {folder_path.name} - {str(e)}")
        print(f"\n总计压缩: {compressed_count} 个文件夹")
# 使用示例
if __name__ == "__main__":
    # 创建压缩器实例
    compressor = BatchCompressor(compression_type='zip')
    # 批量压缩所有子文件夹
    compressor.batch_compress(
        source_dir="/path/to/source",
        output_dir="/path/to/output"
    )
    # 添加进度显示
    compressor.batch_compress(
        source_dir="/path/to/source",
        output_dir="/path/to/output",
        verbose=True  # 如果需要详细输出
    )

并行压缩（提高效率）

import zipfile
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed
import threading
class ParallelBatchCompressor:
    """并行批量压缩器"""
    def __init__(self, max_workers=4):
        self.max_workers = max_workers
        self.lock = threading.Lock()
    def compress_folder(self, folder_path: Path, output_dir: Path):
        """压缩单个文件夹"""
        folder_path = Path(folder_path)
        output_dir = Path(output_dir)
        zip_name = f"{folder_path.name}.zip"
        zip_path = output_dir / zip_name
        try:
            with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
                for file_path in folder_path.rglob('*'):
                    if file_path.is_file():
                        arcname = file_path.relative_to(folder_path)
                        zipf.write(file_path, arcname)
            with self.lock:
                print(f"✅ 已完成: {folder_path.name}")
            return True, folder_path.name
        except Exception as e:
            with self.lock:
                print(f"❌ 失败: {folder_path.name} - {str(e)}")
            return False, folder_path.name
    def batch_compress_parallel(self, source_dir: Union[str, Path], 
                               output_dir: Union[str, Path] = None):
        """并行批量压缩"""
        source_path = Path(source_dir)
        if output_dir is None:
            output_dir = source_path
        output_path = Path(output_dir)
        output_path.mkdir(parents=True, exist_ok=True)
        # 获取所有子文件夹
        folders = [f for f in source_path.iterdir() if f.is_dir()]
        print(f"找到 {len(folders)} 个文件夹需要压缩")
        # 使用线程池并行压缩
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            # 提交所有压缩任务
            future_to_folder = {
                executor.submit(self.compress_folder, folder, output_path): folder
                for folder in folders
            }
            # 收集结果
            success_count = 0
            for future in as_completed(future_to_folder):
                success, name = future.result()
                if success:
                    success_count += 1
        print(f"\n压缩完成! 成功: {success_count}/{len(folders)}")
# 使用示例
compressor = ParallelBatchCompressor(max_workers=4)
compressor.batch_compress_parallel("/path/to/source")

实用案例：压缩指定类型的文件

import zipfile
from pathlib import Path
def compress_by_pattern(source_dir, patterns=["*.txt", "*.jpg"], output_name="selected.zip"):
    """
    压缩目录中指定模式的文件
    Args:
        source_dir: 源目录
        patterns: 文件模式列表
        output_name: 输出文件名
    """
    source_path = Path(source_dir)
    with zipfile.ZipFile(output_name, 'w', zipfile.ZIP_DEFLATED) as zipf:
        for pattern in patterns:
            for file_path in source_path.rglob(pattern):
                if file_path.is_file():
                    # 保持相对路径
                    arcname = file_path.relative_to(source_path)
                    zipf.write(file_path, arcname)
                    print(f"已添加: {arcname}")
    print(f"\n压缩完成: {output_name}")
# 使用示例
compress_by_pattern(
    source_dir="/path/to/documents",
    patterns=["*.pdf", "*.docx", "*.txt"],
    output_name="documents_backup.zip"
)

注意事项

1. 路径处理：使用 Path 对象处理路径，避免跨平台问题
2. 大文件处理：对于大文件，考虑使用分块压缩或并行压缩
3. 内存管理：处理大量文件时，注意内存使用
4. 错误处理：添加异常处理，避免程序崩溃
5. 进度显示：对大量文件压缩时，添加进度显示

这些示例涵盖了大多数批量压缩的场景,您可以根据具体需求选择合适的方案。