图片处理是日常工作的常见需求：压缩体积、格式转换、批量添加水印…每次手动操作繁琐又容易出错。今天教你用Python实现图片批量处理的自动化！

实战场景

• 批量压缩图片体积
• 图片格式互转（PNG/JPG/WebP）
• 批量添加水印或Logo
• 图片批量裁剪和缩放

核心实现

准备工作

pip install Pillow python-dotenv

基础版本：图片格式转换和压缩

from PIL import Image
import os
from pathlib import Path
from datetime import datetime

class ImageProcessor:
    """图片处理器"""
    
    def __init__(self, input_dir='./images_input', output_dir='./images_output'):
        """
        初始化处理器
        
        Args:
            input_dir: 输入目录
            output_dir: 输出目录
        """
        self.input_dir = input_dir
        self.output_dir = output_dir
        self._ensure_dir(output_dir)
    
    def _ensure_dir(self, directory):
        """确保目录存在"""
        if not os.path.exists(directory):
            os.makedirs(directory)
    
    def convert_format(self, input_file, output_format='JPEG', quality=85, 
                      output_name=None):
        """
        转换图片格式
        
        Args:
            input_file: 输入文件路径
            output_format: 输出格式（JPEG/PNG/WebP等）
            quality: 图片质量（1-100）
            output_name: 输出文件名
            
        Returns:
            输出文件路径
        """
        img = Image.open(input_file)
        
        # 转换RGBA为RGB（JPEG不支持透明通道）
        if output_format.upper() == 'JPEG' and img.mode == 'RGBA':
            img = img.convert('RGB')
        
        # 生成输出文件名
        if not output_name:
            name = Path(input_file).stem
            output_name = f"{name}.{output_format.lower()}"
        
        output_path = os.path.join(self.output_dir, output_name)
        
        # 保存
        save_kwargs = {'quality': quality}
        if output_format.upper() == 'PNG':
            save_kwargs.pop('quality', None)
            save_kwargs['optimize'] = True
        
        img.save(output_path, output_format.upper(), **save_kwargs)
        print(f"转换完成: {output_name}")
        
        return output_path
    
    def batch_convert_format(self, input_dir=None, output_format='JPEG', 
                            quality=85):
        """
        批量转换格式
        
        Args:
            input_dir: 输入目录
            output_format: 输出格式
            quality: 图片质量
            
        Returns:
            转换后的文件列表
        """
        input_dir = input_dir or self.input_dir
        converted_files = []
        
        extensions = ['.jpg', '.jpeg', '.png', '.bmp', '.tiff', '.webp']
        
        for filename in os.listdir(input_dir):
            ext = Path(filename).suffix.lower()
            if ext in extensions:
                input_path = os.path.join(input_dir, filename)
                try:
                    output_path = self.convert_format(
                        input_path, 
                        output_format, 
                        quality
                    )
                    converted_files.append(output_path)
                except Exception as e:
                    print(f"转换失败: {filename}, 错误: {e}")
        
        print(f"批量转换完成，共 {len(converted_files)} 个文件")
        return converted_files
    
    def compress_image(self, input_file, max_size_kb=500, quality=85,
                       output_name=None):
        """
        压缩图片
        
        Args:
            input_file: 输入文件路径
            max_size_kb: 最大文件大小（KB）
            quality: 初始质量
            output_name: 输出文件名
            
        Returns:
            输出文件路径
        """
        img = Image.open(input_file)
        
        # 生成输出文件名
        if not output_name:
            name = Path(input_file).stem
            output_name = f"{name}_compressed.jpg"
        
        output_path = os.path.join(self.output_dir, output_name)
        
        # 逐步降低质量直到满足大小要求
        current_quality = quality
        while current_quality > 10:
            img.save(output_path, 'JPEG', quality=current_quality)
            file_size = os.path.getsize(output_path) / 1024
            
            if file_size <= max_size_kb:
                print(f"压缩完成: {output_name} (质量={current_quality}, 大小={file_size:.1f}KB)")
                return output_path
            
            current_quality -= 10
        
        print(f"警告: 未能压缩到 {max_size_kb}KB，当前大小 {file_size:.1f}KB")
        return output_path
    
    def resize_image(self, input_file, width=None, height=None, 
                     keep_aspect_ratio=True, output_name=None):
        """
        调整图片尺寸
        
        Args:
            input_file: 输入文件路径
            width: 目标宽度
            height: 目标高度
            keep_aspect_ratio: 是否保持宽高比
            output_name: 输出文件名
            
        Returns:
            输出文件路径
        """
        img = Image.open(input_file)
        original_width, original_height = img.size
        
        # 计算新尺寸
        if keep_aspect_ratio:
            if width and not height:
                height = int(original_height * width / original_width)
            elif height and not width:
                width = int(original_width * height / original_height)
            elif width and height:
                ratio_w = width / original_width
                ratio_h = height / original_height
                ratio = min(ratio_w, ratio_h)
                width = int(original_width * ratio)
                height = int(original_height * ratio)
        else:
            width = width or original_width
            height = height or original_height
        
        # 调整尺寸
        resized_img = img.resize((width, height), Image.Resampling.LANCZOS)
        
        # 生成输出文件名
        if not output_name:
            name = Path(input_file).stem
            output_name = f"{name}_{width}x{height}.jpg"
        
        output_path = os.path.join(self.output_dir, output_name)
        resized_img.save(output_path, 'JPEG', quality=90)
        
        print(f"调整尺寸完成: {output_name} ({width}x{height})")
        return output_path

进阶版本：图片编辑和批处理

from PIL import Image, ImageDraw, ImageFont, ImageFilter

class ImageEditor:
    """图片编辑器"""
    
    def __init__(self, output_dir='./images_output'):
        """
        初始化编辑器
        
        Args:
            output_dir: 输出目录
        """
        self.output_dir = output_dir
        os.makedirs(output_dir, exist_ok=True)
    
    def add_watermark_text(self, input_file, text, position='bottom_right',
                           font_size=36, color=(255, 255, 255, 128)):
        """
        添加文字水印
        
        Args:
            input_file: 输入文件路径
            text: 水印文字
            position: 位置（bottom_right/bottom_left/top_right/top_left/center）
            font_size: 字体大小
            color: 颜色
            
        Returns:
            输出文件路径
        """
        img = Image.open(input_file).convert('RGBA')
        
        # 创建水印层
        txt_layer = Image.new('RGBA', img.size, (255, 255, 255, 0))
        draw = ImageDraw.Draw(txt_layer)
        
        # 尝试加载字体
        try:
            font = ImageFont.truetype("arial.ttf", font_size)
        except:
            font = ImageFont.load_default()
        
        # 计算文字位置
        bbox = draw.textbbox((0, 0), text, font=font)
        text_width = bbox[2] - bbox[0]
        text_height = bbox[3] - bbox[1]
        padding = 20
        
        img_width, img_height = img.size
        
        positions = {
            'bottom_right': (img_width - text_width - padding, 
                           img_height - text_height - padding),
            'bottom_left': (padding, img_height - text_height - padding),
            'top_right': (img_width - text_width - padding, padding),
            'top_left': (padding, padding),
            'center': ((img_width - text_width) // 2, 
                      (img_height - text_height) // 2)
        }
        
        pos = positions.get(position, positions['bottom_right'])
        
        # 绘制阴影
        shadow_offset = 2
        draw.text((pos[0] + shadow_offset, pos[1] + shadow_offset), 
                 text, font=font, fill=(0, 0, 0, 64))
        # 绘制文字
        draw.text(pos, text, font=font, fill=color)
        
        # 合并图层
        watermarked = Image.alpha_composite(img, txt_layer)
        
        output_name = f"{Path(input_file).stem}_watermarked.png"
        output_path = os.path.join(self.output_dir, output_name)
        watermarked.save(output_path)
        
        print(f"添加水印完成: {output_name}")
        return output_path
    
    def add_watermark_image(self, input_file, watermark_file, 
                           position='bottom_right', opacity=0.5):
        """
        添加图片水印
        
        Args:
            input_file: 输入文件路径
            watermark_file: 水印图片路径
            position: 位置
            opacity: 透明度
            
        Returns:
            输出文件路径
        """
        img = Image.open(input_file).convert('RGBA')
        watermark = Image.open(watermark_file).convert('RGBA')
        
        # 调整水印大小（为原图的20%）
        wm_scale = min(img.width, img.height) * 0.2 / max(watermark.size)
        new_size = (int(watermark.width * wm_scale), 
                   int(watermark.height * wm_scale))
        watermark = watermark.resize(new_size, Image.Resampling.LANCZOS)
        
        # 调整透明度
        watermark = Image.new('RGBA', watermark.size)
        watermark.putalpha(int(255 * opacity))
        
        # 计算位置
        padding = 20
        positions = {
            'bottom_right': (img.width - watermark.width - padding, 
                           img.height - watermark.height - padding),
            'bottom_left': (padding, img.height - watermark.height - padding),
            'top_right': (img.width - watermark.width - padding, padding),
            'top_left': (padding, padding),
            'center': ((img.width - watermark.width) // 2, 
                      (img.height - watermark.height) // 2)
        }
        
        pos = positions.get(position, positions['bottom_right'])
        
        # 创建带水印的图片
        result = Image.new('RGBA', img.size)
        result.paste(img, (0, 0))
        result.paste(watermark, pos, watermark)
        
        output_name = f"{Path(input_file).stem}_logo.png"
        output_path = os.path.join(self.output_dir, output_name)
        result.save(output_path)
        
        print(f"添加Logo水印完成: {output_name}")
        return output_path
    
    def create_thumbnail(self, input_file, size=(200, 200)):
        """
        创建缩略图
        
        Args:
            input_file: 输入文件路径
            size: 缩略图尺寸
            
        Returns:
            输出文件路径
        """
        img = Image.open(input_file)
        
        # 使用thumbnail保持比例
        img.thumbnail(size, Image.Resampling.LANCZOS)
        
        output_name = f"{Path(input_file).stem}_thumb.jpg"
        output_path = os.path.join(self.output_dir, output_name)
        img.save(output_path, 'JPEG', quality=85)
        
        print(f"缩略图创建完成: {output_name}")
        return output_path
    
    def apply_filter(self, input_file, filter_type='blur'):
        """
        应用滤镜
        
        Args:
            input_file: 输入文件路径
            filter_type: 滤镜类型（blur/sharpen/edge/emboss）
            
        Returns:
            输出文件路径
        """
        img = Image.open(input_file)
        
        filters = {
            'blur': ImageFilter.BLUR,
            'sharpen': ImageFilter.SHARPEN,
            'edge': ImageFilter.FIND_EDGES,
            'emboss': ImageFilter.EMBOSS,
            'smooth': ImageFilter.SMOOTH,
        }
        
        filtered_img = img.filter(filters.get(filter_type, ImageFilter.BLUR))
        
        output_name = f"{Path(input_file).stem}_{filter_type}.jpg"
        output_path = os.path.join(self.output_dir, output_name)
        filtered_img.save(output_path)
        
        print(f"滤镜应用完成: {output_name}")
        return output_path

实战：自动化图片处理工作流

class BatchImageProcessor:
    """批量图片处理工作流"""
    
    def __init__(self, input_dir, output_dir):
        """
        初始化处理器
        
        Args:
            input_dir: 输入目录
            output_dir: 输出目录
        """
        self.input_dir = input_dir
        self.output_dir = output_dir
        self.processor = ImageProcessor(input_dir, output_dir)
        self.editor = ImageEditor(output_dir)
    
    def process_product_images(self, compress=True, resize_to=(800, 800),
                               add_watermark=True, watermark_text="@店铺名"):
        """
        处理商品图片工作流
        
        Args:
            compress: 是否压缩
            resize_to: 调整尺寸
            add_watermark: 是否添加水印
            watermark_text: 水印文字
            
        Returns:
            处理完成的文件列表
        """
        processed_files = []
        extensions = ['.jpg', '.jpeg', '.png', '.webp']
        
        for filename in os.listdir(self.input_dir):
            ext = Path(filename).suffix.lower()
            if ext not in extensions:
                continue
            
            input_path = os.path.join(self.input_dir, filename)
            temp_path = input_path
            
            try:
                # 1. 调整尺寸
                if resize_to:
                    temp_path = self.processor.resize_image(
                        input_path, 
                        width=resize_to[0], 
                        height=resize_to[1],
                        output_name=f"_temp_{filename}"
                    )
                
                # 2. 压缩
                if compress:
                    temp_path = self.processor.compress_image(
                        temp_path,
                        max_size_kb=500,
                        output_name=f"_temp_comp_{filename}"
                    )
                
                # 3. 添加水印
                if add_watermark:
                    temp_path = self.editor.add_watermark_text(
                        temp_path,
                        watermark_text,
                        position='bottom_right',
                        font_size=24
                    )
                
                processed_files.append(temp_path)
                print(f"处理完成: {filename}")
                
            except Exception as e:
                print(f"处理失败: {filename}, 错误: {e}")
        
        print(f"\n工作流处理完成，共 {len(processed_files)} 个文件")
        return processed_files
    
    def generate_image_report(self, processed_files):
        """
        生成处理报告
        
        Args:
            processed_files: 处理后的文件列表
        """
        report_path = os.path.join(self.output_dir, '处理报告.txt')
        
        total_size = sum(os.path.getsize(f) for f in processed_files)
        
        with open(report_path, 'w', encoding='utf-8') as f:
            f.write(f"图片处理报告\n")
            f.write(f"=" * 50 + "\n")
            f.write(f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
            f.write(f"处理文件数: {len(processed_files)}\n")
            f.write(f"总大小: {total_size / 1024:.1f} KB\n")
            f.write(f"平均大小: {total_size / len(processed_files) / 1024:.1f} KB\n")
            f.write(f"\n文件列表:\n")
            
            for idx, filepath in enumerate(processed_files, 1):
                size = os.path.getsize(filepath)
                f.write(f"{idx}. {Path(filepath).name} ({size / 1024:.1f} KB)\n")
        
        print(f"报告已生成: {report_path}")
        return report_path


# 使用示例
def main():
    # 格式转换
    processor = ImageProcessor()
    processor.convert_format('photo.png', 'JPEG', quality=85)
    
    # 批量压缩
    processor.batch_convert_format(
        input_dir='./photos',
        output_format='JPEG',
        quality=80
    )
    
    # 批量处理工作流
    workflow = BatchImageProcessor(
        input_dir='./商品图片',
        output_dir='./处理后图片'
    )
    
    processed = workflow.process_product_images(
        compress=True,
        resize_to=(1200, 1200),
        add_watermark=True,
        watermark_text="@扣扣优品"
    )
    
    # 生成报告
    workflow.generate_image_report(processed)

if __name__ == '__main__':
    main()

注意事项

1. 保持原图：重要图片先备份再处理
2. 质量平衡：压缩率太高会影响画质
3. 透明通道：PNG转JPEG需要处理透明通道
4. 内存占用：大图片处理注意内存管理

总结

本文介绍了Python图片处理的完整方案，从基础格式转换、压缩、调整尺寸，到进阶的添加水印、滤镜应用，再到自动化的批量处理工作流。掌握这些技能后，你可以轻松实现各种图片处理需求！