Python自动化批量处理图片与PDF：Pillow和PyMuPDF实战指南

在日常办公和学习中，我们经常会遇到大量图片需要统一调整尺寸、格式转换、添加水印，或者需要将多个图片合并成PDF、从PDF中提取图片等繁琐任务。手动一张张处理不仅效率低下，还容易出错。本文将围绕Python自动化批量处理图片与PDF的核心需求，手把手带你搭建一套完整的处理流程。无论你是需要快速处理项目文档的职场人士，还是希望提升编程实战能力的学生，都能从本文获得可直接复用的代码方案。

本文将重点介绍如何使用Python的PIL（Pillow）库进行图片批量处理，以及PyMuPDF（fitz）库进行PDF操作，涵盖从环境搭建、核心代码编写到异常处理的全流程。学完后，你将能够轻松实现图片尺寸调整、格式转换、水印添加、PDF合并拆分、图片提取等常见需求，彻底告别重复劳动。

1. 核心工具与库的选择

在开始实战之前，我们先了解几个核心的Python库及其适用场景。正确选择工具是高效完成任务的第一步。

1.1 Pillow：图像处理的瑞士军刀

Pillow是Python中最流行的图像处理库，它提供了广泛的文件格式支持、强大的图像处理能力和简单的API。无论是基本的尺寸调整、格式转换，还是复杂的滤镜应用、绘制操作，Pillow都能胜任。

主要功能包括：

• 支持数十种图像格式（JPEG、PNG、GIF、BMP等）
• 图像缩放、旋转、裁剪、翻转等几何变换
• 颜色空间转换、图像增强、滤镜效果
• 文本和图形绘制（可用于添加水印）
• 图像序列处理（GIF、WebP动画）

1.2 PyMuPDF：高性能PDF处理利器

PyMuPDF（fitz）是一个功能强大且速度极快的PDF处理库，相比其他PDF库，它在处理大型文档时表现尤为出色。

核心能力涵盖：

• PDF文档的读取、创建和编辑
• 页面提取、合并、旋转、删除
• 文本和图像提取
• 添加注释、书签和水印
• 文档转换（PDF转图像、HTML转PDF等）

1.3 其他备选方案对比

虽然本文以Pillow和PyMuPDF为主，但了解其他选项有助于你在不同场景下做出合适选择：

• OpenCV：更适合计算机视觉任务，如图像识别、视频处理
• ReportLab：专注于PDF生成，适合创建复杂的报表文档
• pdf2image：专门用于PDF到图像的转换，基于poppler后端
• Img2PDF：轻量级的图片转PDF工具，API极其简单

对于大多数办公自动化场景，Pillow + PyMuPDF的组合已经足够强大且易于上手。

2. 环境准备与安装指南

在编写代码之前，我们需要搭建合适的开发环境。以下是详细的安装步骤和版本说明。

2.1 Python版本要求

推荐使用Python 3.7及以上版本，本文示例基于Python 3.9开发测试。你可以通过以下命令检查当前Python版本：

python --version # 或 python3 --version

如果尚未安装Python，建议从Python官网下载最新稳定版，或使用Anaconda发行版。

2.2 安装必要的库

使用pip安装所需的库，建议创建虚拟环境以避免依赖冲突：

# 创建并激活虚拟环境（可选） python -m venv image_pdf_env source image_pdf_env/bin/activate # Linux/Mac image_pdf_env\Scripts\activate # Windows # 安装核心库 pip install Pillow PyMuPDF # 可选：安装其他辅助库 pip install opencv-python pdf2image img2pdf

2.3 验证安装

安装完成后，通过简单的导入测试验证库是否正常工作：

# 测试脚本：check_installation.py try: from PIL import Image, ImageDraw, ImageFont import fitz # PyMuPDF print("✓ Pillow和PyMuPDF安装成功！") except ImportError as e: print(f"✗ 导入失败: {e}")

2.4 开发工具推荐

虽然任何文本编辑器都可以编写Python代码，但使用合适的IDE能显著提升开发效率：

• VS Code：轻量级，插件丰富，适合初学者
• PyCharm：专业Python IDE，调试功能强大
• Jupyter Notebook：适合交互式开发和数据探索

本文示例使用VS Code编写，但代码在任何环境中都能正常运行。

3. 图片批量处理实战

现在进入实战环节，我们先从图片处理开始。假设你有一个包含数百张图片的文件夹，需要统一进行处理。

3.1 项目结构设计

在开始编码前，先规划好项目目录结构：

image_pdf_processor/ ├── src/ │ ├── image_processor.py # 图片处理核心模块 │ ├── pdf_processor.py # PDF处理核心模块 │ └── utils.py # 工具函数 ├── input/ # 输入文件目录 │ ├── images/ # 待处理的图片 │ └── pdfs/ # 待处理的PDF ├── output/ # 输出文件目录 │ ├── processed_images/ # 处理后的图片 │ └── processed_pdfs/ # 处理后的PDF ├── config/ # 配置文件 └── requirements.txt # 依赖列表

3.2 图片批量缩放与格式转换

最常见的需求之一是将大量图片统一调整为指定尺寸并转换格式。以下是完整的实现代码：

# 文件路径：src/image_processor.py import os from PIL import Image from pathlib import Path class ImageProcessor: def __init__(self, input_dir, output_dir): self.input_dir = Path(input_dir) self.output_dir = Path(output_dir) self.output_dir.mkdir(parents=True, exist_ok=True) def batch_resize_and_convert(self, target_size=(800, 600), output_format='JPEG', quality=85): """ 批量调整图片尺寸并转换格式 Args: target_size: 目标尺寸 (宽, 高) output_format: 输出格式 ('JPEG', 'PNG', 'WEBP') quality: 输出质量 (1-100)，仅对JPEG有效 """ supported_formats = {'.jpg', '.jpeg', '.png', '.bmp', '.gif', '.webp'} for img_path in self.input_dir.iterdir(): if img_path.suffix.lower() in supported_formats: try: with Image.open(img_path) as img: # 调整尺寸，保持宽高比 img.thumbnail(target_size, Image.Resampling.LANCZOS) # 转换模式（如PNG的RGBA转JPEG的RGB） if output_format == 'JPEG' and img.mode in ('RGBA', 'P'): img = img.convert('RGB') # 生成输出路径 output_path = self.output_dir / f"{img_path.stem}_resized.{output_format.lower()}" # 保存图片 save_kwargs = {'quality': quality} if output_format == 'JPEG' else {} img.save(output_path, format=output_format, **save_kwargs) print(f"✓ 处理完成: {img_path.name} -> {output_path.name}") except Exception as e: print(f"✗ 处理失败 {img_path.name}: {e}") # 使用示例 if __name__ == "__main__": processor = ImageProcessor('input/images', 'output/processed_images') processor.batch_resize_and_convert(target_size=(1024, 768), output_format='JPEG')

关键参数说明：

• target_size：使用thumbnail方法会保持原图宽高比，确保图片不变形
• output_format：根据需求选择格式，JPEG适合照片，PNG适合需要透明度的图片
• quality：JPEG质量参数，85是较好的平衡点，数值越大文件越大

3.3 批量添加水印功能

为图片添加水印可以保护版权或标注来源。以下是带文字水印的实现：

# 在ImageProcessor类中添加方法 def add_watermark(self, watermark_text, font_size=30, opacity=128, position='center'): """ 为图片批量添加文字水印 Args: watermark_text: 水印文字 font_size: 字体大小 opacity: 透明度 (0-255) position: 水印位置 ('center', 'bottom-right', 'top-left'等) """ try: # 尝试加载字体，失败则使用默认字体 try: font = ImageFont.truetype("arial.ttf", font_size) except IOError: font = ImageFont.load_default() for img_path in self.input_dir.iterdir(): if img_path.suffix.lower() in {'.jpg', '.jpeg', '.png', '.bmp'}: with Image.open(img_path).convert('RGBA') as base_img: # 创建水印层 watermark = Image.new('RGBA', base_img.size, (0, 0, 0, 0)) draw = ImageDraw.Draw(watermark) # 计算文字尺寸和位置 bbox = draw.textbbox((0, 0), watermark_text, font=font) text_width = bbox[2] - bbox[0] text_height = bbox[3] - bbox[1] # 根据位置参数计算坐标 if position == 'center': x = (base_img.width - text_width) // 2 y = (base_img.height - text_height) // 2 elif position == 'bottom-right': x = base_img.width - text_width - 20 y = base_img.height - text_height - 20 else: # top-left x = 20 y = 20 # 绘制水印文字（带透明度） draw.text((x, y), watermark_text, font=font, fill=(255, 255, 255, opacity)) # 合并图层 combined = Image.alpha_composite(base_img, watermark) # 转换回RGB模式（如果需要） if base_img.mode == 'RGB': combined = combined.convert('RGB') output_path = self.output_dir / f"{img_path.stem}_watermarked.png" combined.save(output_path) print(f"✓ 水印添加完成: {output_path.name}") except Exception as e: print(f"水印处理异常: {e}")

3.4 高级功能：批量重命名与EXIF信息处理

对于摄影爱好者或需要管理大量图片的用户，重命名和EXIF信息处理也很重要：

import exifread from datetime import datetime def batch_rename_by_date(self, pattern="{date}_{counter:04d}"): """ 根据EXIF信息中的拍摄日期批量重命名图片 """ for img_path in self.input_dir.iterdir(): if img_path.suffix.lower() in {'.jpg', '.jpeg'}: try: # 读取EXIF信息 with open(img_path, 'rb') as f: tags = exifread.process_file(f) # 获取拍摄日期 date_tag = tags.get('EXIF DateTimeOriginal') if date_tag: date_str = str(date_tag).replace(':', '').replace(' ', '_') date_obj = datetime.strptime(str(date_tag), '%Y:%m:%d %H:%M:%S') date_formatted = date_obj.strftime('%Y%m%d_%H%M%S') else: # 如果没有EXIF日期，使用文件修改时间 date_formatted = datetime.fromtimestamp( img_path.stat().st_mtime).strftime('%Y%m%d_%H%M%S') # 生成新文件名 new_name = pattern.format(date=date_formatted, counter=1) new_path = self.output_dir / f"{new_name}{img_path.suffix}" # 处理重名文件 counter = 1 while new_path.exists(): new_name = pattern.format(date=date_formatted, counter=counter) new_path = self.output_dir / f"{new_name}{img_path.suffix}" counter += 1 # 复制文件 import shutil shutil.copy2(img_path, new_path) print(f"✓ 重命名: {img_path.name} -> {new_path.name}") except Exception as e: print(f"✗ 重命名失败 {img_path.name}: {e}")

4. PDF处理全流程实战

处理完图片，我们转向PDF文档的自动化处理。PDF处理的需求同样多样且频繁。

4.1 PDF与图片互转

将图片合并为PDF：

# 文件路径：src/pdf_processor.py import fitz # PyMuPDF from PIL import Image import os from pathlib import Path class PDFProcessor: def __init__(self, input_dir, output_dir): self.input_dir = Path(input_dir) self.output_dir = Path(output_dir) self.output_dir.mkdir(parents=True, exist_ok=True) def images_to_pdf(self, pdf_name="output.pdf", page_size=(595, 842)): """ 将多张图片合并为一个PDF文件 Args: pdf_name: 输出的PDF文件名 page_size: 页面尺寸 (A4默认: 595x842 points) """ try: # 创建新PDF文档 doc = fitz.open() # 获取所有图片文件 image_files = sorted([f for f in self.input_dir.iterdir() if f.suffix.lower() in ['.jpg', '.jpeg', '.png', '.bmp']]) if not image_files: print("未找到图片文件") return for img_path in image_files: try: # 创建新页面 page = doc.new_page(width=page_size[0], height=page_size[1]) # 计算图片在页面中的位置和尺寸 img = Image.open(img_path) img_width, img_height = img.size # 保持宽高比缩放图片以适应页面 page_width, page_height = page_size ratio = min(page_width/img_width, page_height/img_height) * 0.9 # 留边距 new_width = int(img_width * ratio) new_height = int(img_height * ratio) # 居中放置图片 x = (page_width - new_width) / 2 y = (page_height - new_height) / 2 # 插入图片到PDF页面 page.insert_image(fitz.Rect(x, y, x + new_width, y + new_height), filename=str(img_path)) print(f"✓ 添加页面: {img_path.name}") except Exception as e: print(f"✗ 处理图片失败 {img_path.name}: {e}") # 保存PDF output_path = self.output_dir / pdf_name doc.save(output_path) doc.close() print(f"PDF生成完成: {output_path}") except Exception as e: print(f"PDF创建失败: {e}") # 使用示例 if __name__ == "__main__": processor = PDFProcessor('input/images', 'output/processed_pdfs') processor.images_to_pdf("我的相册.pdf")

从PDF中提取图片：

def extract_images_from_pdf(self, pdf_name, output_dir="extracted_images", dpi=150): """ 从PDF中提取所有图片 Args: pdf_name: PDF文件名 output_dir: 图片输出目录 dpi: 输出图片分辨率 """ pdf_path = self.input_dir / pdf_name extract_dir = self.output_dir / output_dir extract_dir.mkdir(parents=True, exist_ok=True) try: doc = fitz.open(pdf_path) image_counter = 0 for page_num in range(len(doc)): page = doc[page_num] # 获取页面中的所有图片 image_list = page.get_images() for img_index, img in enumerate(image_list): # 提取图片 xref = img[0] pix = fitz.Pixmap(doc, xref) if pix.n - pix.alpha < 4: # 非CMYK图片 # 保存为PNG output_path = extract_dir / f"page_{page_num+1}_img_{img_index+1}.png" pix.save(output_path) image_counter += 1 print(f"✓ 提取图片: {output_path.name}") pix = None # 释放内存 doc.close() print(f"共提取 {image_counter} 张图片") except Exception as e: print(f"图片提取失败: {e}")

4.2 PDF页面操作与合并拆分

合并多个PDF文件：

def merge_pdfs(self, pdf_list, output_name="merged.pdf"): """ 合并多个PDF文件 Args: pdf_list: 要合并的PDF文件名列表 output_name: 合并后的文件名 """ try: merged_doc = fitz.open() for pdf_name in pdf_list: pdf_path = self.input_dir / pdf_name if pdf_path.exists(): doc = fitz.open(pdf_path) merged_doc.insert_pdf(doc) # 插入整个文档 doc.close() print(f"✓ 已添加: {pdf_name}") else: print(f"✗ 文件不存在: {pdf_name}") output_path = self.output_dir / output_name merged_doc.save(output_path) merged_doc.close() print(f"合并完成: {output_path}") except Exception as e: print(f"PDF合并失败: {e}") # 使用示例 pdf_list = ["文档1.pdf", "文档2.pdf", "文档3.pdf"] processor.merge_pdfs(pdf_list, "合并文档.pdf")

拆分PDF文件：

def split_pdf(self, pdf_name, split_ranges=None): """ 按页面范围拆分PDF文件 Args: pdf_name: 要拆分的PDF文件名 split_ranges: 拆分范围列表，如 [(1,3), (4,6)] 表示拆分为1-3页和4-6页 """ pdf_path = self.input_dir / pdf_name if not split_ranges: # 如果没有指定范围，默认每页拆分为一个文件 doc = fitz.open(pdf_path) split_ranges = [(i+1, i+1) for i in range(len(doc))] doc.close() try: for i, (start_page, end_page) in enumerate(split_ranges): doc = fitz.open(pdf_path) new_doc = fitz.open() # 调整页码（从0开始） start_idx = max(0, start_page - 1) end_idx = min(len(doc) - 1, end_page - 1) # 插入指定页面范围 new_doc.insert_pdf(doc, from_page=start_idx, to_page=end_idx) output_name = f"{pdf_path.stem}_part_{i+1}.pdf" output_path = self.output_dir / output_name new_doc.save(output_path) new_doc.close() doc.close() print(f"✓ 生成拆分文件: {output_name} (页码 {start_page}-{end_page})") except Exception as e: print(f"PDF拆分失败: {e}")

4.3 PDF内容处理与优化

为PDF添加水印：

def add_watermark_to_pdf(self, pdf_name, watermark_text, output_name=None): """ 为PDF每一页添加文字水印 """ if output_name is None: output_name = f"{Path(pdf_name).stem}_watermarked.pdf" pdf_path = self.input_dir / pdf_name output_path = self.output_dir / output_name try: doc = fitz.open(pdf_path) for page_num in range(len(doc)): page = doc[page_num] # 创建水印文本 watermark = page.insert_text((page.rect.width/2, page.rect.height/2), watermark_text, fontsize=40, color=(0.8, 0.8, 0.8), # 浅灰色 rotate=45, # 45度倾斜 opacity=0.3) # 透明度 doc.save(output_path) doc.close() print(f"水印添加完成: {output_path}") except Exception as e: print(f"添加水印失败: {e}")

压缩PDF文件大小：

def compress_pdf(self, pdf_name, output_name=None, quality="normal"): """ 压缩PDF文件大小 Args: quality: 压缩质量 ('high', 'normal', 'low') """ if output_name is None: output_name = f"{Path(pdf_name).stem}_compressed.pdf" quality_map = { 'high': 0.8, # 高质量压缩 'normal': 0.6, # 正常压缩 'low': 0.4 # 高压缩率 } compression = quality_map.get(quality, 0.6) try: doc = fitz.open(self.input_dir / pdf_name) doc.save(self.output_dir / output_name, garbage=4, # 清理无用对象 deflate=True, # 压缩流 clean=True, # 清理文档结构 compress=compression) # 图片压缩率 # 比较文件大小 original_size = (self.input_dir / pdf_name).stat().st_size compressed_size = (self.output_dir / output_name).stat().st_size reduction = (1 - compressed_size/original_size) * 100 print(f"压缩完成: {original_size/1024:.1f}KB -> {compressed_size/1024:.1f}KB " f"(减少{reduction:.1f}%)") doc.close() except Exception as e: print(f"PDF压缩失败: {e}")

5. 完整工作流整合

现在我们将图片处理和PDF处理功能整合成一个完整的工作流，实现真正的"一条龙"处理。

5.1 自动化流水线设计

# 文件路径：src/workflow_manager.py from pathlib import Path from image_processor import ImageProcessor from pdf_processor import PDFProcessor import shutil class WorkflowManager: def __init__(self, base_dir="."): self.base_dir = Path(base_dir) self.setup_directories() def setup_directories(self): """创建标准目录结构""" directories = [ 'input/images', 'input/pdfs', 'output/processed_images', 'output/processed_pdfs', 'output/final_results', 'temp' ] for dir_path in directories: (self.base_dir / dir_path).mkdir(parents=True, exist_ok=True) def process_photo_collection(self, image_patterns=None): """ 完整的图片处理流水线：重命名 → 调整尺寸 → 添加水印 → 生成PDF相册 """ if image_patterns is None: image_patterns = ['*.jpg', '*.jpeg', '*.png'] # 初始化处理器 img_processor = ImageProcessor( self.base_dir / 'input/images', self.base_dir / 'temp/resized_images' ) pdf_processor = PDFProcessor( self.base_dir / 'temp/resized_images', self.base_dir / 'output/final_results' ) try: print("=== 开始图片处理流水线 ===") # 步骤1: 按日期重命名 print("步骤1: 批量重命名图片") img_processor.input_dir = self.base_dir / 'input/images' img_processor.output_dir = self.base_dir / 'temp/renamed_images' img_processor.output_dir.mkdir(parents=True, exist_ok=True) # 这里调用重命名方法（需在实际代码中实现） # 步骤2: 调整尺寸和格式 print("步骤2: 调整图片尺寸和格式") img_processor.input_dir = self.base_dir / 'temp/renamed_images' img_processor.output_dir = self.base_dir / 'temp/resized_images' img_processor.batch_resize_and_convert( target_size=(1200, 800), output_format='JPEG', quality=90 ) # 步骤3: 添加水印 print("步骤3: 添加水印") img_processor.input_dir = self.base_dir / 'temp/resized_images' img_processor.output_dir = self.base_dir / 'temp/watermarked_images' img_processor.add_watermark( "© My Collection 2024", font_size=24, position='bottom-right' ) # 步骤4: 生成PDF相册 print("步骤4: 生成PDF相册") pdf_processor.input_dir = self.base_dir / 'temp/watermarked_images' pdf_processor.images_to_pdf("我的照片集.pdf") print("=== 处理完成 ===") except Exception as e: print(f"流水线执行失败: {e}") finally: # 清理临时文件（可选） # shutil.rmtree(self.base_dir / 'temp') pass # 使用示例 if __name__ == "__main__": workflow = WorkflowManager() workflow.process_photo_collection()

5.2 配置文件管理

为了提升代码的灵活性，我们可以添加配置文件支持：

# 文件路径：config/settings.yaml image_processing: default_size: [1200, 800] output_format: "JPEG" quality: 90 watermark: text: "© Confidential" font_size: 24 position: "bottom-right" pdf_processing: page_size: [595, 842] # A4 compression_quality: "normal" workflow: auto_clean_temp: true keep_original: true

# 配置文件读取工具 import yaml from pathlib import Path def load_config(config_path="config/settings.yaml"): """加载配置文件""" try: with open(config_path, 'r', encoding='utf-8') as f: return yaml.safe_load(f) except FileNotFoundError: print("配置文件不存在，使用默认配置") return get_default_config() def get_default_config(): """返回默认配置""" return { 'image_processing': { 'default_size': [1200, 800], 'output_format': 'JPEG', 'quality': 85 }, 'pdf_processing': { 'page_size': [595, 842], 'compression_quality': 'normal' } }

6. 常见问题与解决方案

在实际使用过程中，你可能会遇到各种问题。以下是常见问题的排查指南。

6.1 图片处理常见问题

问题1：内存不足错误

• 现象：处理大图片时出现"MemoryError"
• 原因：高分辨率图片占用内存过大
• 解决方案：
  • 在处理前先调整图片尺寸
  • 使用流式处理，不要同时加载所有图片
  • 增加系统虚拟内存

# 内存优化版本的处理函数 def memory_efficient_resize(self, target_size): """内存友好的图片处理""" for img_path in self.input_dir.iterdir(): try: # 逐步处理，及时释放内存 with Image.open(img_path) as img: img.thumbnail(target_size) # 立即保存并释放内存 output_path = self.output_dir / f"{img_path.stem}_resized.jpg" img.save(output_path, optimize=True) except Exception as e: print(f"处理失败 {img_path.name}: {e}")

问题2：格式兼容性问题

• 现象：某些图片无法打开或保存
• 原因：格式不支持或文件损坏
• 解决方案：
  • 使用Pillow的格式检测功能
  • 添加异常处理和日志记录

def safe_image_open(self, img_path): """安全的图片打开方式""" try: with Image.open(img_path) as img: # 验证图片完整性 img.verify() img = Image.open(img_path) # 重新打开已验证的图片 return img except Exception as e: print(f"图片损坏或格式不支持: {img_path.name} - {e}") return None

6.2 PDF处理常见问题

问题1：中文显示乱码

• 现象：PDF中的中文显示为乱码
• 原因：字体缺失或编码问题
• 解决方案：
  • 确保系统安装中文字体
  • 在代码中指定中文字体路径

# 使用中文字体添加水印 def add_chinese_watermark(self, watermark_text): """支持中文的水印添加""" try: # 尝试加载系统中文字体 font_paths = [ "C:/Windows/Fonts/simhei.ttf", # Windows黑体 "/System/Library/Fonts/PingFang.ttc", # Mac字体 "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf" # Linux ] font = None for font_path in font_paths: if Path(font_path).exists(): font = ImageFont.truetype(font_path, 30) break if font is None: raise Exception("未找到合适的中文字体") # 使用找到的字体添加水印 # ... 其余水印代码 ... except Exception as e: print(f"中文字体处理失败: {e}")

问题2：PDF权限限制

• 现象：无法编辑或提取受保护的PDF
• 原因：PDF设置了密码保护或权限限制
• 解决方案：
  • 合法的密码解除保护（需有权限）
  • 使用专门的PDF解锁工具
  • 联系文档所有者获取权限

6.3 性能优化建议

当处理大量文件时，性能成为关键因素。以下是一些优化技巧：

批量处理优化：

from concurrent.futures import ThreadPoolExecutor import multiprocessing def parallel_image_processing(self, max_workers=None): """使用多线程并行处理图片""" if max_workers is None: max_workers = multiprocessing.cpu_count() image_files = [f for f in self.input_dir.iterdir() if f.suffix.lower() in {'.jpg', '.jpeg', '.png'}] def process_single_image(img_path): try: with Image.open(img_path) as img: img.thumbnail((800, 600)) output_path = self.output_dir / f"{img_path.stem}_processed.jpg" img.save(output_path) return True except Exception as e: print(f"处理失败 {img_path.name}: {e}") return False with ThreadPoolExecutor(max_workers=max_workers) as executor: results = list(executor.map(process_single_image, image_files)) success_count = sum(results) print(f"处理完成: {success_count}/{len(image_files)} 成功")

内存使用监控：

import psutil import os def memory_usage(): """监控内存使用情况""" process = psutil.Process(os.getpid()) return process.memory_info().rss / 1024 / 1024 # MB def process_with_memory_check(self): """带内存检查的处理函数""" for img_path in self.input_dir.iterdir(): if memory_usage() > 500: # 如果内存使用超过500MB print("内存使用过高，建议分批处理") break # 正常处理逻辑 self.process_image(img_path)

7. 最佳实践与工程化建议

将脚本升级为可维护的工程代码，需要遵循一些最佳实践。

7.1 错误处理与日志记录

完善的错误处理：

import logging from datetime import datetime def setup_logging(): """配置日志系统""" logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(f'processing_{datetime.now().strftime("%Y%m%d_%H%M%S")}.log'), logging.StreamHandler() ] ) class RobustImageProcessor(ImageProcessor): """增强的错误处理版本""" def safe_batch_process(self): """带完整错误处理的批量处理""" success_count = 0 error_count = 0 error_details = [] for img_path in self.input_dir.iterdir(): try: self._validate_image(img_path) self._process_single_image(img_path) success_count += 1 except Exception as e: error_count += 1 error_details.append(f"{img_path.name}: {str(e)}") logging.error(f"处理失败 {img_path.name}: {e}") # 根据错误类型采取不同措施 if "corrupt" in str(e).lower(): self._handle_corrupt_file(img_path) elif "memory" in str(e).lower(): self._handle_memory_issue() # 生成处理报告 self._generate_report(success_count, error_count, error_details) def _validate_image(self, img_path): """验证图片文件""" if not img_path.exists(): raise FileNotFoundError(f"文件不存在: {img_path}") if img_path.stat().st_size == 0: raise ValueError("文件大小为0") # 简单的格式验证 valid_extensions = {'.jpg', '.jpeg', '.png', '.bmp'} if img_path.suffix.lower() not in valid_extensions: raise ValueError(f"不支持的格式: {img_path.suffix}")

7.2 配置管理与环境隔离

使用环境变量管理敏感信息：

import os from dotenv import load_dotenv load_dotenv() # 加载.env文件 class Config: """配置管理类""" @property def input_dir(self): return os.getenv('INPUT_DIR', 'input') @property def output_dir(self): return os.getenv('OUTPUT_DIR', 'output') @property def max_file_size(self): return int(os.getenv('MAX_FILE_SIZE', '104857600')) # 100MB默认值 @property def allowed_formats(self): formats = os.getenv('ALLOWED_FORMATS', 'jpg,jpeg,png,pdf') return set(format.strip().lower() for format in formats.split(','))

7.3 单元测试与质量保证

编写测试用例：

# tests/test