Python Pillow图像批处理实战：尺寸调整、格式转换与水印添加

在实际图像处理项目中经常需要批量处理图片例如调整尺寸、转换格式、添加水印或批量重命名。这类任务如果手动操作不仅效率低下还容易出错。本文将基于常见的Python图像处理库Pillow演示如何构建一个实用的批量图像处理工具涵盖从环境搭建到生产级优化的完整流程。这个工具将实现以下核心功能读取指定目录下的所有图片统一调整尺寸为800x600像素转换为JPEG格式添加文字水印并保存到输出目录。整个过程将采用可配置的方式方便根据实际需求调整参数。1. 理解图像批处理的核心需求与技术选型图像批处理在Web开发、摄影后期、电商平台等场景中非常常见。例如用户上传的图片需要生成多种尺寸的缩略图或者批量添加版权信息。手动用Photoshop处理几十上百张图片既不现实也不专业。1.1 为什么选择Python和Pillow库Python的Pillow库PIL Fork是处理图像任务的首选原因包括安装简单依赖少API设计直观学习成本低支持JPEG、PNG、GIF等常见格式丰富的图像操作功能裁剪、旋转、滤镜等活跃的社区支持和良好的文档相比OpenCVPillow更轻量适合简单的批处理任务相比ImageMagick的命令行方式Pillow提供了更友好的编程接口。1.2 批处理工具的设计要点一个实用的批处理工具需要考虑支持递归遍历子目录处理不同格式的图片文件避免修改原始文件读写分离提供进度反馈良好的错误处理跳过损坏文件可配置的处理参数2. 环境准备与项目结构在开始编码前需要准备好开发环境。建议使用Python 3.7及以上版本这些版本对Pillow的支持最稳定。2.1 安装Pillow库使用pip安装Pillowpip install Pillow验证安装是否成功from PIL import Image print(Image.__version__)如果输出版本号如9.0.0说明安装正确。2.2 创建项目目录结构建议按以下结构组织项目文件image_batch_processor/ ├── src/ │ ├── processor.py # 核心处理逻辑 │ └── config.py # 配置文件 ├── input/ # 输入图片目录 ├── output/ # 输出图片目录 ├── requirements.txt # 依赖列表 └── README.md # 项目说明这种结构将代码、配置、输入输出分离符合生产项目的规范。2.3 准备测试图片在input目录中放置一些测试图片建议包含不同格式JPG、PNG等和尺寸的图片。这样可以在开发过程中验证工具的兼容性。3. 实现核心批处理功能现在开始实现批处理工具的核心功能。我们将采用模块化设计便于后续扩展和维护。3.1 定义配置参数首先在config.py中定义可配置的参数# 图像处理配置 IMAGE_CONFIG { target_size: (800, 600), # 目标尺寸宽, 高 output_format: JPEG, # 输出格式 quality: 85, # JPEG质量1-100 watermark_text: Sample, # 水印文字 watermark_position: (10, 10), # 水印位置x, y watermark_color: (255, 255, 255, 128) # 水印颜色RGBA } # 文件处理配置 FILE_CONFIG { supported_formats: [.jpg, .jpeg, .png, .bmp, .gif], recursive: True, # 是否递归处理子目录 overwrite: False # 是否覆盖已存在文件 }使用配置字典的好处是参数集中管理修改时不需要深入代码逻辑。3.2 实现图像处理类在processor.py中创建主要的处理类import os from PIL import Image, ImageDraw, ImageFont import time class ImageBatchProcessor: def __init__(self, input_dir, output_dir, image_config, file_config): self.input_dir input_dir self.output_dir output_dir self.image_config image_config self.file_config file_config self.processed_count 0 self.error_count 0 # 创建输出目录 os.makedirs(output_dir, exist_okTrue) def get_image_files(self): 获取所有图片文件路径 image_files [] if self.file_config[recursive]: # 递归遍历所有子目录 for root, dirs, files in os.walk(self.input_dir): for file in files: if self.is_supported_format(file): full_path os.path.join(root, file) image_files.append(full_path) else: # 仅处理当前目录 for file in os.listdir(self.input_dir): if self.is_supported_format(file): full_path os.path.join(self.input_dir, file) image_files.append(full_path) return image_files def is_supported_format(self, filename): 检查文件格式是否支持 ext os.path.splitext(filename)[1].lower() return ext in self.file_config[supported_formats] def process_single_image(self, input_path): 处理单张图片 try: # 打开图片 with Image.open(input_path) as img: # 转换RGB模式处理PNG透明背景 if img.mode ! RGB: img img.convert(RGB) # 调整尺寸 img self.resize_image(img) # 添加水印 img self.add_watermark(img) # 生成输出路径 output_path self.get_output_path(input_path) # 保存图片 img.save(output_path, formatself.image_config[output_format], qualityself.image_config[quality]) self.processed_count 1 return True except Exception as e: print(f处理图片失败: {input_path}, 错误: {str(e)}) self.error_count 1 return False def resize_image(self, img): 调整图片尺寸 target_size self.image_config[target_size] return img.resize(target_size, Image.Resampling.LANCZOS) def add_watermark(self, img): 添加文字水印 draw ImageDraw.Draw(img) # 尝试使用系统字体失败时使用默认字体 try: font ImageFont.truetype(arial.ttf, 20) except: font ImageFont.load_default() text self.image_config[watermark_text] position self.image_config[watermark_position] color self.image_config[watermark_color] draw.text(position, text, fillcolor, fontfont) return img def get_output_path(self, input_path): 生成输出文件路径 # 获取相对路径保持目录结构 rel_path os.path.relpath(input_path, self.input_dir) base_name os.path.splitext(rel_path)[0] # 构建输出路径 output_path os.path.join(self.output_dir, f{base_name}.{self.image_config[output_format].lower()}) # 处理文件重名 if not self.file_config[overwrite]: counter 1 original_path output_path while os.path.exists(output_path): name, ext os.path.splitext(original_path) output_path f{name}_{counter}{ext} counter 1 # 创建输出目录 os.makedirs(os.path.dirname(output_path), exist_okTrue) return output_path def process_all(self): 批量处理所有图片 start_time time.time() image_files self.get_image_files() if not image_files: print(未找到支持的图片文件) return print(f找到 {len(image_files)} 个图片文件开始处理...) for i, image_path in enumerate(image_files, 1): print(f处理进度: {i}/{len(image_files)} - {image_path}) self.process_single_image(image_path) end_time time.time() print(f处理完成成功: {self.processed_count}, 失败: {self.error_count}) print(f总耗时: {end_time - start_time:.2f} 秒)这个类封装了完整的处理逻辑包括文件遍历、图像处理、错误处理和进度反馈。3.3 创建使用示例创建main.py作为入口文件from src.processor import ImageBatchProcessor from src.config import IMAGE_CONFIG, FILE_CONFIG def main(): # 配置参数 input_dir input output_dir output # 创建处理器实例 processor ImageBatchProcessor( input_dirinput_dir, output_diroutput_dir, image_configIMAGE_CONFIG, file_configFILE_CONFIG ) # 执行批处理 processor.process_all() if __name__ __main__: main()4. 运行验证与结果分析完成代码编写后需要验证工具是否能正确工作。4.1 准备测试环境在项目根目录创建input文件夹放入几张测试图片。图片应该包含不同格式和尺寸例如photo1.jpg1920x1080photo2.png800x600带透明背景photo3.bmp640x4804.2 执行批处理运行main.pypython main.py正常输出应该类似找到 3 个图片文件开始处理... 处理进度: 1/3 - input/photo1.jpg 处理进度: 2/3 - input/photo2.png 处理进度: 3/3 - input/photo3.bmp 处理完成成功: 3, 失败: 0 总耗时: 2.34 秒4.3 检查输出结果查看output目录应该生成处理后的图片photo1.jpeg800x600带水印photo2.jpeg800x600透明背景转白色带水印photo3.jpeg800x600带水印打开图片验证尺寸是否统一为800x600格式是否均为JPEG水印位置和内容是否正确图片质量是否正常5. 常见问题排查在实际使用中可能会遇到各种问题下面是典型的排查流程。5.1 图片处理失败常见原因问题现象可能原因检查方式解决方案处理图片失败错误文件损坏或格式不支持检查文件是否能正常打开跳过损坏文件确认格式在支持列表中输出图片空白颜色模式转换问题检查原始图片模式在convert(RGB)前打印img.mode调试水印不显示字体文件缺失检查字体路径使用绝对路径或改用默认字体内存错误图片尺寸过大检查图片文件大小增加内存或优化处理流程权限错误输出目录不可写检查目录权限修改输出目录权限或换位置5.2 调试技巧遇到问题时可以添加调试信息def process_single_image(self, input_path): try: print(f调试: 开始处理 {input_path}) with Image.open(input_path) as img: print(f调试: 图片模式 {img.mode}, 尺寸 {img.size}) # ... 其余处理逻辑 except Exception as e: print(f调试: 处理失败 {input_path}, 错误详情: {traceback.format_exc()}) return False5.3 性能优化建议处理大量图片时可以考虑以下优化# 使用多进程处理适用于CPU密集型任务 from multiprocessing import Pool def process_parallel(self): image_files self.get_image_files() with Pool(processes4) as pool: # 4个进程 results pool.map(self.process_single_image_wrapper, image_files) def process_single_image_wrapper(self, args): 包装函数用于多进程调用 return self.process_single_image(args)6. 生产环境最佳实践将工具用于实际项目时还需要考虑更多生产级需求。6.1 配置管理改进生产环境建议使用JSON或YAML配置文件// config.json { image_processing: { target_size: [800, 600], output_format: JPEG, quality: 85 }, file_handling: { supported_formats: [.jpg, .png], recursive: true } }加载配置import json with open(config.json, r) as f: config json.load(f) image_config config[image_processing] file_config config[file_handling]6.2 日志记录使用logging模块替代print语句import logging logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(batch_processor.log), logging.StreamHandler() ] ) logger logging.getLogger(__name__) # 在代码中使用 logger.info(f开始处理图片: {image_path}) logger.error(f处理失败: {str(e)})6.3 异常处理增强针对不同异常类型分别处理def process_single_image(self, input_path): try: with Image.open(input_path) as img: # 处理逻辑 pass except FileNotFoundError: logger.error(f文件不存在: {input_path}) except PIL.UnidentifiedImageError: logger.error(f无法识别的图片格式: {input_path}) except Exception as e: logger.error(f处理图片时发生未知错误: {input_path}, {str(e)})6.4 进度反馈优化添加更详细的进度信息def process_all(self): image_files self.get_image_files() total_files len(image_files) for i, image_path in enumerate(image_files, 1): # 计算进度百分比 progress (i / total_files) * 100 print(f[{progress:.1f}%] 处理中: {os.path.basename(image_path)}) self.process_single_image(image_path)6.5 内存管理处理大图片时注意内存使用def resize_image(self, img): 优化内存使用的尺寸调整 original_size img.size target_size self.image_config[target_size] # 如果目标尺寸比原图大使用高质量缩放 if target_size[0] original_size[0] or target_size[1] original_size[1]: return img.resize(target_size, Image.Resampling.LANCZOS) else: # 缩小图片时使用更节省内存的算法 return img.resize(target_size, Image.Resampling.BILINEAR)7. 功能扩展方向基础版本完成后可以根据实际需求添加更多功能。7.1 添加图片滤镜def apply_filter(self, img, filter_name): 应用图片滤镜 filters { blur: ImageFilter.BLUR, contour: ImageFilter.CONTOUR, detail: ImageFilter.DETAIL, sharpen: ImageFilter.SHARPEN } if filter_name in filters: return img.filter(filters[filter_name]) return img7.2 批量重命名def rename_files(self, patternimage_{counter:04d}): 批量重命名输出文件 image_files self.get_image_files() for i, image_path in enumerate(image_files, 1): new_filename f{pattern.format(counteri)}.jpg output_path os.path.join(self.output_dir, new_filename) # ... 处理逻辑7.3 支持更多输出格式def save_image(self, img, output_path): 根据格式优化保存参数 format self.image_config[output_format].upper() save_args {format: format} if format JPEG: save_args[quality] self.image_config[quality] elif format PNG: save_args[compress_level] 6 # PNG压缩级别 img.save(output_path, **save_args)7.4 添加配置文件验证def validate_config(self): 验证配置参数有效性 errors [] # 检查尺寸参数 width, height self.image_config[target_size] if width 0 or height 0: errors.append(图片尺寸必须大于0) # 检查质量参数 quality self.image_config.get(quality, 85) if not 1 quality 100: errors.append(图片质量必须在1-100之间) if errors: raise ValueError(配置验证失败: ; .join(errors))这个图像批处理工具展示了从需求分析到生产优化的完整开发流程。实际项目中还可以根据具体需求添加更多功能如EXIF信息保留、批量格式转换、图片压缩优化等。关键是要保持代码的可维护性和可扩展性便于后续迭代开发。