在实际图像处理项目中我们经常需要处理批量图像比如调整尺寸、转换格式、添加水印或进行简单的滤镜处理。这类任务如果手动操作不仅效率低下还容易出错。Python 的 Pillow 库提供了强大的图像处理能力结合文件操作可以轻松实现批量自动化处理。本文将以一个实际的图像批量处理项目为例带你从环境准备开始逐步实现一个能够处理图像尺寸调整、格式转换和水印添加的自动化脚本。这个项目特别适合需要处理大量图片的开发者、设计师或内容创作者学完后你可以直接应用到自己的图片管理工作中。1. 理解 Pillow 库的核心能力与项目目标Pillow 是 Python 图像处理领域的事实标准库它支持多种图像格式的读写操作提供了丰富的图像处理功能。在开始编码前我们需要明确这个批量处理项目的具体目标批量读取能够自动遍历指定文件夹中的所有图像文件尺寸调整将图像统一调整为指定尺寸保持比例或强制拉伸格式转换将图像转换为统一的格式如 JPG、PNG水印添加在图像指定位置添加文字或图片水印输出管理处理后的图像保存到指定目录保持原有文件名或按规则重命名1.1 Pillow 基本概念与安装Pillow 是 PILPython Imaging Library的分支和升级版本提供了更友好的 API 和更好的 Python 3 支持。它核心的Image类封装了图像数据和各种操作方法。安装 Pillow 只需要一条命令pip install Pillow验证安装是否成功from PIL import Image print(Image.__version__)如果输出版本号如 9.5.0说明安装成功。2. 项目环境准备与目录结构设计一个清晰的目录结构能让批量处理逻辑更加明确。建议按以下方式组织image_batch_processor/ ├── src/ │ ├── main.py # 主程序入口 │ ├── image_processor.py # 图像处理核心类 │ └── config.py # 配置文件 ├── input/ # 原始图像目录 ├── output/ # 处理后的图像目录 ├── watermarks/ # 水印图片目录 └── requirements.txt # 项目依赖2.1 创建虚拟环境与依赖管理使用虚拟环境可以避免包版本冲突。创建并激活虚拟环境# 创建虚拟环境 python -m venv image_env # 激活虚拟环境Windows image_env\Scripts\activate # 激活虚拟环境macOS/Linux source image_env/bin/activate创建 requirements.txt 文件Pillow9.5.0安装依赖pip install -r requirements.txt2.2 配置文件设计config.py 用于集中管理处理参数避免硬编码# config.py class Config: # 输入输出路径 INPUT_DIR input OUTPUT_DIR output WATERMARK_DIR watermarks # 图像处理参数 TARGET_SIZE (800, 600) # 目标尺寸 (宽, 高) OUTPUT_FORMAT JPEG # 输出格式 QUALITY 85 # JPEG 质量 (1-100) # 水印参数 WATERMARK_TEXT Sample Watermark WATERMARK_POSITION bottom-right # top-left, top-right, bottom-left, bottom-right, center WATERMARK_OPACITY 0.6 # 水印透明度 (0-1) # 文件处理设置 OVERWRITE_EXISTING False # 是否覆盖已存在的输出文件 SUPPORTED_FORMATS {.jpg, .jpeg, .png, .bmp, .gif, .tiff}3. 实现图像处理核心功能图像处理的核心逻辑封装在 ImageProcessor 类中每个功能方法都保持单一职责。3.1 图像读取与格式检查首先实现基本的文件操作和格式验证# image_processor.py import os from PIL import Image from config import Config class ImageProcessor: def __init__(self, configNone): self.config config or Config() self.processed_count 0 self.error_files [] def get_image_files(self, directory): 获取目录中所有支持的图像文件 image_files [] for filename in os.listdir(directory): file_ext os.path.splitext(filename)[1].lower() if file_ext in self.config.SUPPORTED_FORMATS: image_files.append(os.path.join(directory, filename)) return image_files def open_image(self, image_path): 安全打开图像文件 try: return Image.open(image_path) except Exception as e: print(f无法打开图像 {image_path}: {e}) self.error_files.append((image_path, str(e))) return None3.2 尺寸调整实现尺寸调整需要考虑保持宽高比的问题def resize_image(self, image, target_sizeNone): 调整图像尺寸保持宽高比 if target_size is None: target_size self.config.TARGET_SIZE # 计算调整后的尺寸保持比例 original_width, original_height image.size target_width, target_height target_size # 计算缩放比例 width_ratio target_width / original_width height_ratio target_height / original_height # 选择较小的比例以确保图像完全包含在目标尺寸内 scale_ratio min(width_ratio, height_ratio) new_width int(original_width * scale_ratio) new_height int(original_height * scale_ratio) # 使用 LANCZOS 重采样算法高质量 resized_image image.resize((new_width, new_height), Image.LANCZOS) # 创建目标尺寸的画布白色背景 if self.config.OUTPUT_FORMAT.upper() in [JPEG, JPG]: background_color (255, 255, 255) # 白色 else: background_color (255, 255, 255, 0) # 透明背景 canvas Image.new(image.mode, target_size, background_color) # 将调整后的图像居中放置 x_offset (target_width - new_width) // 2 y_offset (target_height - new_height) // 2 canvas.paste(resized_image, (x_offset, y_offset)) return canvas3.3 水印添加功能水印功能支持文字和图片两种形式def add_watermark(self, image, watermark_typetext, **kwargs): 添加水印到图像 if watermark_type text: return self._add_text_watermark(image, **kwargs) elif watermark_type image: return self._add_image_watermark(image, **kwargs) else: return image def _add_text_watermark(self, image, textNone, positionNone, opacityNone): 添加文字水印 from PIL import ImageDraw, ImageFont text text or self.config.WATERMARK_TEXT position position or self.config.WATERMARK_POSITION opacity opacity or self.config.WATERMARK_OPACITY # 创建绘图对象 draw ImageDraw.Draw(image) # 尝试加载字体失败则使用默认字体 try: font ImageFont.truetype(arial.ttf, 36) except: font ImageFont.load_default() # 计算文字尺寸和位置 bbox draw.textbbox((0, 0), text, fontfont) text_width bbox[2] - bbox[0] text_height bbox[3] - bbox[1] img_width, img_height image.size # 根据位置参数计算坐标 margin 20 if position top-left: x margin y margin elif position top-right: x img_width - text_width - margin y margin elif position bottom-left: x margin y img_height - text_height - margin elif position bottom-right: x img_width - text_width - margin y img_height - text_height - margin else: # center x (img_width - text_width) // 2 y (img_height - text_height) // 2 # 添加文字阴影增强可读性 shadow_color (0, 0, 0, int(255 * opacity)) draw.text((x2, y2), text, fontfont, fillshadow_color) # 添加主文字 text_color (255, 255, 255, int(255 * opacity)) draw.text((x, y), text, fontfont, filltext_color) return image def _add_image_watermark(self, image, watermark_pathNone, positionNone, opacityNone): 添加图片水印 position position or self.config.WATERMARK_POSITION opacity opacity or self.config.WATERMARK_OPACITY # 查找水印图片 if watermark_path is None: watermark_files [f for f in os.listdir(self.config.WATERMARK_DIR) if f.lower().endswith((.png, .jpg, .jpeg))] if watermark_files: watermark_path os.path.join(self.config.WATERMARK_DIR, watermark_files[0]) else: print(未找到水印图片跳过水印添加) return image try: watermark Image.open(watermark_path) if watermark.mode ! RGBA: watermark watermark.convert(RGBA) # 调整水印大小最大为原图的1/4 img_width, img_height image.size max_watermark_size (img_width // 4, img_height // 4) watermark.thumbnail(max_watermark_size, Image.LANCZOS) # 设置透明度 alpha watermark.split()[3] alpha alpha.point(lambda p: p * opacity) watermark.putalpha(alpha) # 计算位置 wm_width, wm_height watermark.size margin 20 if position top-left: x margin y margin elif position top-right: x img_width - wm_width - margin y margin elif position bottom-left: x margin y img_height - wm_height - margin elif position bottom-right: x img_width - wm_width - margin y img_height - wm_height - margin else: # center x (img_width - wm_width) // 2 y (img_height - wm_height) // 2 # 如果原图不是 RGBA先转换 if image.mode ! RGBA: image image.convert(RGBA) # 合并图像 image.paste(watermark, (x, y), watermark) # 转换回原模式 if self.config.OUTPUT_FORMAT.upper() in [JPEG, JPG]: image image.convert(RGB) except Exception as e: print(f添加图片水印失败: {e}) return image4. 批量处理流程与主程序实现批量处理需要协调文件遍历、图像处理和结果保存的整个流程。4.1 完整的处理流水线def process_single_image(self, input_path, output_pathNone): 处理单张图像的完整流程 if output_path is None: filename os.path.basename(input_path) name, ext os.path.splitext(filename) output_filename f{name}_processed.{self.config.OUTPUT_FORMAT.lower()} output_path os.path.join(self.config.OUTPUT_DIR, output_filename) # 检查输出文件是否已存在 if not self.config.OVERWRITE_EXISTING and os.path.exists(output_path): print(f跳过已存在的文件: {output_path}) return True # 打开图像 image self.open_image(input_path) if image is None: return False try: # 调整尺寸 if self.config.TARGET_SIZE: image self.resize_image(image) # 添加水印 if self.config.WATERMARK_TEXT or os.path.exists(self.config.WATERMARK_DIR): image self.add_watermark(image) # 确保输出目录存在 os.makedirs(os.path.dirname(output_path), exist_okTrue) # 保存图像 save_kwargs {} if self.config.OUTPUT_FORMAT.upper() in [JPEG, JPG]: save_kwargs[quality] self.config.QUALITY if image.mode RGBA: image image.convert(RGB) image.save(output_path, formatself.config.OUTPUT_FORMAT, **save_kwargs) self.processed_count 1 print(f处理完成: {input_path} - {output_path}) return True except Exception as e: print(f处理图像失败 {input_path}: {e}) self.error_files.append((input_path, str(e))) return False def process_batch(self, input_dirNone): 批量处理目录中的所有图像 input_dir input_dir or self.config.INPUT_DIR if not os.path.exists(input_dir): print(f输入目录不存在: {input_dir}) return False image_files self.get_image_files(input_dir) if not image_files: print(f在 {input_dir} 中未找到支持的图像文件) return False print(f找到 {len(image_files)} 个图像文件开始处理...) success_count 0 for image_path in image_files: if self.process_single_image(image_path): success_count 1 print(f处理完成: 成功 {success_count}/{len(image_files)}) if self.error_files: print(处理失败的文件:) for file_path, error in self.error_files: print(f {file_path}: {error}) return success_count 04.2 主程序入口main.py 提供命令行接口和直接运行功能# main.py import argparse import sys from image_processor import ImageProcessor from config import Config def main(): parser argparse.ArgumentParser(description批量图像处理工具) parser.add_argument(--input, -i, help输入目录路径, defaultConfig.INPUT_DIR) parser.add_argument(--output, -o, help输出目录路径, defaultConfig.OUTPUT_DIR) parser.add_argument(--size, -s, help目标尺寸 (格式: 宽x高), defaultf{Config.TARGET_SIZE[0]}x{Config.TARGET_SIZE[1]}) parser.add_argument(--format, -f, help输出格式, choices[JPEG, PNG, BMP], defaultConfig.OUTPUT_FORMAT) parser.add_argument(--watermark, -w, help水印文字, defaultConfig.WATERMARK_TEXT) args parser.parse_args() # 更新配置 config Config() config.INPUT_DIR args.input config.OUTPUT_DIR args.output config.OUTPUT_FORMAT args.format.upper() config.WATERMARK_TEXT args.watermark # 解析尺寸参数 if x in args.size: width, height map(int, args.size.split(x)) config.TARGET_SIZE (width, height) # 执行处理 processor ImageProcessor(config) success processor.process_batch() sys.exit(0 if success else 1) if __name__ __main__: main()5. 运行验证与结果检查5.1 准备测试数据在 input 目录中放置一些测试图像创建基本的目录结构# 创建测试目录结构 mkdir -p input output watermarks # 复制一些测试图片到 input 目录 # 或者使用程序生成测试图片5.2 运行批量处理直接运行主程序python src/main.py或者使用命令行参数python src/main.py --input ./input --output ./output --size 800x600 --format JPEG --watermark Processed5.3 验证处理结果检查输出目录中的文件# 验证脚本 verify_results.py import os from PIL import Image def verify_processing_results(): input_dir input output_dir output input_files set(os.listdir(input_dir)) output_files set(os.listdir(output_dir)) print(f输入文件数: {len(input_files)}) print(f输出文件数: {len(output_files)}) # 检查文件尺寸 for filename in output_files: if filename.endswith((jpg, jpeg, png)): filepath os.path.join(output_dir, filename) with Image.open(filepath) as img: print(f{filename}: {img.size} - {img.format})6. 常见问题排查与解决方案在实际运行中可能会遇到各种问题下面是典型的排查路径。6.1 文件读取问题问题现象可能原因检查方式解决方案无法打开图像错误文件损坏/格式不支持检查文件扩展名和文件完整性使用其他工具验证文件或从备份恢复找不到输入文件路径错误/权限不足检查路径是否存在权限是否足够使用绝对路径检查文件权限内存错误处理大文件图像尺寸过大查看图像尺寸和系统内存增加处理尺寸限制使用流式处理6.2 处理结果异常问题现象可能原因检查方式解决方案图像变形尺寸计算错误检查原图尺寸和目标尺寸比例修改 resize_image 方法中的比例计算逻辑水印位置错误坐标计算错误打印水印位置坐标调试调整位置计算逻辑增加边界检查颜色异常色彩模式转换问题检查图像模式转换流程确保色彩模式转换的正确性输出质量差压缩参数设置不当检查质量参数和保存选项调整质量参数尝试不同的重采样算法6.3 性能优化建议当处理大量图像或大尺寸图像时性能可能成为瓶颈def optimize_performance(): 性能优化配置 # 1. 调整 Pillow 的缓存大小 Image.MAX_IMAGE_PIXELS None # 取消大图像限制谨慎使用 # 2. 使用更快的重采样算法质量稍差 # Image.LANCZOS - Image.BILINEAR # 3. 批量处理时限制并发数量 # 避免同时打开太多文件 # 4. 对于超大图像考虑分块处理 pass7. 生产环境最佳实践将脚本用于实际项目时还需要考虑更多生产环境因素。7.1 错误处理与日志记录增强的错误处理机制import logging import time def setup_logging(): 配置日志记录 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(image_processor.log), logging.StreamHandler() ] ) class ProductionImageProcessor(ImageProcessor): def process_batch(self, input_dirNone): 生产环境版本的批量处理 start_time time.time() # 记录开始信息 logging.info(f开始批量处理: {input_dir}) result super().process_batch(input_dir) # 记录统计信息 elapsed_time time.time() - start_time logging.info(f处理完成: 耗时 {elapsed_time:.2f}秒, 成功 {self.processed_count}个文件) if self.error_files: logging.warning(f失败 {len(self.error_files)}个文件) for file_path, error in self.error_files: logging.error(f处理失败: {file_path} - {error}) return result7.2 配置管理进阶使用环境变量或配置文件import os from dataclasses import dataclass dataclass class ProductionConfig: INPUT_DIR: str os.getenv(IMAGE_INPUT_DIR, input) OUTPUT_DIR: str os.getenv(IMAGE_OUTPUT_DIR, output) MAX_FILE_SIZE: int int(os.getenv(MAX_IMAGE_SIZE, 10485760)) # 10MB TIMEOUT: int int(os.getenv(PROCESS_TIMEOUT, 30))7.3 安全考虑验证输入文件确实是图像文件而不仅仅是扩展名限制处理文件的尺寸和数量对用户输入进行严格的验证和转义使用临时文件处理避免内存耗尽def safe_image_processing(self, input_path): 安全处理图像 # 检查文件大小 if os.path.getsize(input_path) self.config.MAX_FILE_SIZE: raise ValueError(文件大小超过限制) # 验证文件类型通过魔数 with open(input_path, rb) as f: header f.read(10) if not self.is_valid_image_header(header): raise ValueError(无效的图像文件) return self.process_single_image(input_path)这个图像批量处理项目展示了如何将 Pillow 库的强大功能与 Python 的文件操作结合构建一个实用的自动化工具。实际项目中你可以根据具体需求扩展更多功能如批量重命名、EXIF 信息处理、图像滤镜应用等。关键是要保持代码的模块化便于维护和扩展。