Python文件管理之批量压缩指定目录中的图片

import os
import sys
import argparse
from PIL import Image
import shutil
from datetime import datetime

class BatchImageCompressor:
    def __init__(self, input_dir, output_dir=None, quality=85, resize_ratio=1.0):
        self.input_dir = input_dir
        self.output_dir = output_dir or input_dir
        self.quality = quality
        self.resize_ratio = resize_ratio
        self.supported_formats = ('.jpg', '.jpeg', '.png', '.bmp', '.tiff', '.webp')
        self.compression_log = []
        
    def check_directories(self):
        """检查输入和输出目录"""
        if not os.path.exists(self.input_dir):
            raise FileNotFoundError(f"输入目录 '{self.input_dir}' 不存在")
        
        if not os.path.exists(self.output_dir):
            try:
                os.makedirs(self.output_dir)
                print(f"创建输出目录: {self.output_dir}")
            except Exception as e:
                raise OSError(f"无法创建输出目录 '{self.output_dir}': {e}")
                
    def is_supported_image(self, filename):
        """检查文件是否为支持的图片格式"""
        return filename.lower().endswith(self.supported_formats)
        
    def get_image_files(self):
        """获取目录中的所有图片文件"""
        try:
            files = os.listdir(self.input_dir)
            image_files = [f for f in files if self.is_supported_image(f)]
            return sorted(image_files)
        except Exception as e:
            print(f"读取目录时出错: {e}")
            return []
            
    def get_file_size(self, filepath):
        """获取文件大小（MB）"""
        size_bytes = os.path.getsize(filepath)
        return size_bytes / (1024 * 1024)
        
    def compress_image(self, filename):
        """压缩单个图片文件"""
        input_path = os.path.join(self.input_dir, filename)
        output_path = os.path.join(self.output_dir, filename)
        
        try:
            # 获取原始文件大小
            original_size = self.get_file_size(input_path)
            
            # 打开图片
            with Image.open(input_path) as img:
                # 如果需要调整尺寸
                if self.resize_ratio < 1.0:
                    width, height = img.size
                    new_width = int(width * self.resize_ratio)
                    new_height = int(height * self.resize_ratio)
                    img = img.resize((new_width, new_height), Image.LANCZOS)
                
                # 处理RGBA模式的图片（如PNG）
                if img.mode in ('RGBA', 'LA', 'P'):
                    # 转换为RGB模式以支持JPEG格式
                    if filename.lower().endswith(('.jpg', '.jpeg')):
                        background = Image.new('RGB', img.size, (255, 255, 255))
                        background.paste(img, mask=img.split()[-1] if img.mode == 'RGBA' else None)
                        img = background
                
                # 保存压缩后的图片
                if filename.lower().endswith(('.jpg', '.jpeg')):
                    img.save(output_path, 'JPEG', quality=self.quality, optimize=True)
                elif filename.lower().endswith('.png'):
                    img.save(output_path, 'PNG', optimize=True)
                else:
                    img.save(output_path, optimize=True)
                    
            # 获取压缩后文件大小
            compressed_size = self.get_file_size(output_path)
            compression_ratio = (1 - compressed_size / original_size) * 100
            
            return {
                'filename': filename,
                'original_size': original_size,
                'compressed_size': compressed_size,
                'compression_ratio': compression_ratio,
                'status': 'success'
            }
            
        except Exception as e:
            return {
                'filename': filename,
                'error': str(e),
                'status': 'failed'
            }
            
    def process_images(self, backup_original=False):
        """批量处理图片"""
        try:
            self.check_directories()
            image_files = self.get_image_files()
            
            if not image_files:
                print("未找到支持的图片文件")
                return
                
            print(f"找到 {len(image_files)} 个图片文件")
            print(f"压缩质量: {self.quality}%")
            if self.resize_ratio < 1.0:
                print(f"尺寸调整比例: {self.resize_ratio}")
                
            # 如果需要备份且输入输出目录相同，则创建备份目录
            if backup_original and self.input_dir == self.output_dir:
                backup_dir = self.input_dir + "_backup"
                if not os.path.exists(backup_dir):
                    os.makedirs(backup_dir)
                print(f"原始文件将备份到: {backup_dir}")
                
            processed_count = 0
            failed_count = 0
            
            for i, filename in enumerate(image_files, 1):
                print(f"\r处理进度: {i}/{len(image_files)} ({i/len(image_files)*100:.1f}%)", end='')
                
                # 如果需要备份，先复制原始文件
                if backup_original and self.input_dir == self.output_dir:
                    backup_path = os.path.join(self.input_dir + "_backup", filename)
                    if not os.path.exists(backup_path):
                        shutil.copy2(os.path.join(self.input_dir, filename), backup_path)
                
                result = self.compress_image(filename)
                
                if result['status'] == 'success':
                    self.compression_log.append(result)
                    processed_count += 1
                    print(f"\n{filename}: {result['original_size']:.2f}MB -> {result['compressed_size']:.2f}MB "
                          f"(压缩率: {result['compression_ratio']:.1f}%)")
                else:
                    failed_count += 1
                    print(f"\n{filename}: 处理失败 - {result['error']}")
                    
            print(f"\n\n处理完成!")
            print(f"成功处理: {processed_count}")
            print(f"处理失败: {failed_count}")
            
            if processed_count > 0:
                total_original = sum(item['original_size'] for item in self.compression_log)
                total_compressed = sum(item['compressed_size'] for item in self.compression_log)
                avg_compression = (1 - total_compressed / total_original) * 100
                print(f"总压缩率: {avg_compression:.1f}%")
                print(f"节省空间: {total_original - total_compressed:.2f}MB")
                
        except KeyboardInterrupt:
            print("\n\n用户中断操作")
        except Exception as e:
            print(f"\n处理过程中发生错误: {e}")
            
    def save_log(self, log_file="compression_log.txt"):
        """保存压缩日志"""
        try:
            log_path = os.path.join(self.output_dir, log_file)
            with open(log_path, "w", encoding="utf-8") as f:
                f.write("批量图片压缩操作日志\n")
                f.write("=" * 50 + "\n")
                f.write(f"操作时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
                f.write(f"输入目录: {self.input_dir}\n")
                f.write(f"输出目录: {self.output_dir}\n")
                f.write(f"压缩质量: {self.quality}%\n")
                f.write(f"尺寸调整比例: {self.resize_ratio}\n\n")
                
                if self.compression_log:
                    f.write("压缩详情:\n")
                    f.write("-" * 50 + "\n")
                    for item in self.compression_log:
                        f.write(f"{item['filename']}:\n")
                        f.write(f"  原始大小: {item['original_size']:.2f}MB\n")
                        f.write(f"  压缩后大小: {item['compressed_size']:.2f}MB\n")
                        f.write(f"  压缩率: {item['compression_ratio']:.1f}%\n\n")
                        
            print(f"操作日志已保存到: {log_file}")
        except Exception as e:
            print(f"保存日志时出错: {e}")

def main():
    parser = argparse.ArgumentParser(description="批量图片压缩工具")
    parser.add_argument("input_dir", help="输入目录路径")
    parser.add_argument("-o", "--output_dir", help="输出目录路径（默认与输入目录相同）")
    parser.add_argument("-q", "--quality", type=int, default=85, 
                       help="压缩质量 (1-100, 默认: 85)")
    parser.add_argument("-r", "--resize", type=float, default=1.0, 
                       help="尺寸调整比例 (0.1-1.0, 默认: 1.0)")
    parser.add_argument("-b", "--backup", action="store_true", 
                       help="备份原始文件")
    
    args = parser.parse_args()
    
    # 验证参数
    if not 1 <= args.quality <= 100:
        print("错误: 压缩质量必须在1-100之间")
        sys.exit(1)
        
    if not 0.1 <= args.resize <= 1.0:
        print("错误: 尺寸调整比例必须在0.1-1.0之间")
        sys.exit(1)
    
    try:
        compressor = BatchImageCompressor(
            input_dir=args.input_dir,
            output_dir=args.output_dir,
            quality=args.quality,
            resize_ratio=args.resize
        )
        
        compressor.process_images(backup_original=args.backup)
        compressor.save_log()
        
    except Exception as e:
        print(f"程序执行出错: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

pip install Pillow

# 基本用法，使用默认压缩质量
python image_compressor.py /path/to/images

# 指定压缩质量
python image_compressor.py /path/to/images -q 70

# 指定输出目录
python image_compressor.py /path/to/images -o /path/to/compressed

# 调整图片尺寸（缩小到原来的80%）
python image_compressor.py /path/to/images -r 0.8

# 备份原始文件
python image_compressor.py /path/to/images -b

python image_compressor.py ./images -q 70 -r 0.9