Python自动化图片与PDF批量处理:从环境搭建到实战应用

发布时间:2026/7/31 5:24:39
Python自动化图片与PDF批量处理:从环境搭建到实战应用 你是不是也经常遇到这样的场景项目文档需要统一调整图片尺寸几十张照片要批量压缩上传或者收到一堆扫描版PDF需要提取文字和图片手动一张张处理不仅耗时费力还容易出错。最近在整理技术文档时我发现了一个高效的解决方案——通过Python脚本实现图片和PDF的批量处理全流程。这套方法真正解决了重复性工作的痛点让原本需要半天的手工操作在几分钟内自动完成。本文将分享从环境搭建到完整实战的详细流程包含具体的代码实现和常见问题排查。无论你是需要处理技术文档、项目资料还是日常办公文件这套方案都能显著提升效率。1. 核心痛点与解决方案对比在技术文档管理、项目资料整理等场景中我们经常面临以下典型问题传统手工处理的痛点图片尺寸不统一需要逐张调整宽高比大量图片需要压缩以减少存储空间PDF文档中的图片需要批量提取扫描版PDF需要转换为可编辑文本不同格式的图片需要统一转换格式自动化方案的优势批量处理一次性处理数百个文件一致性保证所有文件采用相同处理参数时间节省从小时级缩短到分钟级可重复使用处理逻辑可封装为脚本重复调用以技术文档为例一个中等规模的项目可能包含50-100张示意图、架构图等图片资源。手动处理每张图片平均需要2-3分钟而自动化脚本可以在30秒内完成全部处理。2. 环境准备与工具选择2.1 基础环境要求# 检查Python版本 python --version # 推荐 Python 3.82.2 核心依赖库安装pip install Pillow PyPDF2 pdf2image python-docx pip install opencv-python pytesseract2.3 各库的功能说明库名称主要功能适用场景Pillow图像处理核心库图片缩放、格式转换、滤镜应用PyPDF2PDF文本处理PDF拆分、合并、文本提取pdf2imagePDF转图片将PDF页面转换为图像格式python-docxWord文档操作处理提取的文本内容OpenCV高级图像处理图像识别、边缘检测pytesseractOCR文字识别从图片中提取文字2.4 环境验证脚本# environment_check.py import importlib required_libraries [PIL, PyPDF2, pdf2image, cv2, pytesseract] def check_environment(): missing_libs [] for lib in required_libraries: try: importlib.import_module(lib) print(f✓ {lib} 安装成功) except ImportError: missing_libs.append(lib) print(f✗ {lib} 未安装) if missing_libs: print(f\n需要安装的库: {, .join(missing_libs)}) return False return True if __name__ __main__: check_environment()3. 图片批量处理实战3.1 基础图片处理类设计# image_processor.py from PIL import Image import os from pathlib import Path class ImageProcessor: def __init__(self, source_dir, output_dir): self.source_dir Path(source_dir) self.output_dir Path(output_dir) self.output_dir.mkdir(exist_okTrue) def get_supported_formats(self): 获取支持的图片格式 return [.jpg, .jpeg, .png, .bmp, .tiff, .webp] def find_images(self): 查找目录中的所有图片文件 images [] for format_ext in self.get_supported_formats(): images.extend(self.source_dir.glob(f*{format_ext})) images.extend(self.source_dir.glob(f*{format_ext.upper()})) return images3.2 图片尺寸批量调整# 续 image_processor.py def resize_images(self, target_size(800, 600), keep_aspectTrue): 批量调整图片尺寸 images self.find_images() processed_count 0 for img_path in images: try: with Image.open(img_path) as img: if keep_aspect: # 保持宽高比调整尺寸 img.thumbnail(target_size, Image.Resampling.LANCZOS) else: # 强制调整到指定尺寸 img img.resize(target_size, Image.Resampling.LANCZOS) # 保存处理后的图片 output_path self.output_dir / fresized_{img_path.name} img.save(output_path, optimizeTrue) processed_count 1 print(f已处理: {img_path.name} - {output_path.name}) except Exception as e: print(f处理失败 {img_path.name}: {str(e)}) return processed_count3.3 图片格式批量转换# 续 image_processor.py def convert_format(self, target_formatJPEG, quality85): 批量转换图片格式 images self.find_images() converted_count 0 for img_path in images: try: with Image.open(img_path) as img: # 转换为RGB模式针对JPEG格式 if img.mode ! RGB and target_format JPEG: img img.convert(RGB) output_filename f{img_path.stem}.{target_format.lower()} output_path self.output_dir / output_filename save_kwargs {quality: quality} if target_format JPEG else {} img.save(output_path, formattarget_format, **save_kwargs) converted_count 1 print(f已转换: {img_path.name} - {output_filename}) except Exception as e: print(f转换失败 {img_path.name}: {str(e)}) return converted_count3.4 图片压缩优化# 续 image_processor.py def compress_images(self, quality70, max_sizeNone): 批量压缩图片 images self.find_images() compressed_count 0 total_saved 0 for img_path in images: try: original_size img_path.stat().st_size with Image.open(img_path) as img: # 如果有最大尺寸限制先调整尺寸 if max_size: img.thumbnail(max_size, Image.Resampling.LANCZOS) output_path self.output_dir / fcompressed_{img_path.name} # 根据格式选择保存参数 if img_path.suffix.lower() in [.jpg, .jpeg]: img.save(output_path, optimizeTrue, qualityquality) else: img.save(output_path, optimizeTrue) compressed_size output_path.stat().st_size saved original_size - compressed_size total_saved saved compressed_count 1 print(f压缩: {img_path.name} f({original_size//1024}KB - {compressed_size//1024}KB) f节省: {saved//1024}KB) except Exception as e: print(f压缩失败 {img_path.name}: {str(e)}) print(f\n总计压缩 {compressed_count} 张图片节省空间: {total_saved//1024}KB) return compressed_count, total_saved4. PDF处理全流程实现4.1 PDF基础操作类# pdf_processor.py import PyPDF2 from pdf2image import convert_from_path import pytesseract from PIL import Image import os class PDFProcessor: def __init__(self, poppler_pathNone): self.poppler_path poppler_path def split_pdf(self, input_pdf, output_dir, pages_per_split10): 拆分PDF文件 with open(input_pdf, rb) as file: pdf_reader PyPDF2.PdfReader(file) total_pages len(pdf_reader.pages) for start_page in range(0, total_pages, pages_per_split): end_page min(start_page pages_per_split, total_pages) pdf_writer PyPDF2.PdfWriter() for page_num in range(start_page, end_page): pdf_writer.add_page(pdf_reader.pages[page_num]) output_filename f{os.path.splitext(input_pdf)[0]}_part{start_page//pages_per_split 1}.pdf output_path os.path.join(output_dir, output_filename) with open(output_path, wb) as output_file: pdf_writer.write(output_file) print(f生成拆分文件: {output_filename} (页码 {start_page1}-{end_page}))4.2 PDF转图片处理# 续 pdf_processor.py def pdf_to_images(self, input_pdf, output_dir, dpi200): 将PDF转换为图片 os.makedirs(output_dir, exist_okTrue) try: images convert_from_path(input_pdf, dpidpi, poppler_pathself.poppler_path) for i, image in enumerate(images): output_path os.path.join(output_dir, fpage_{i1:03d}.jpg) image.save(output_path, JPEG, quality85) print(f转换页面 {i1} - {output_path}) return len(images) except Exception as e: print(fPDF转换图片失败: {str(e)}) return 04.3 PDF文字提取与OCR# 续 pdf_processor.py def extract_text(self, input_pdf, use_ocrFalse): 提取PDF中的文字内容 text_content try: # 首先尝试直接提取文本 with open(input_pdf, rb) as file: pdf_reader PyPDF2.PdfReader(file) for page_num in range(len(pdf_reader.pages)): page_text pdf_reader.pages[page_num].extract_text() if page_text.strip(): text_content f--- 第 {page_num1} 页 ---\n{page_text}\n\n # 如果直接提取文本较少且启用OCR尝试OCR识别 if use_ocr and len(text_content.strip()) 100: print(文本提取较少启用OCR识别...) ocr_text self._ocr_pdf(input_pdf) text_content \n--- OCR识别结果 ---\n ocr_text except Exception as e: print(f文本提取失败: {str(e)}) return text_content def _ocr_pdf(self, input_pdf): 使用OCR识别PDF中的文字 ocr_text temp_dir temp_ocr os.makedirs(temp_dir, exist_okTrue) try: # 先将PDF转换为图片 image_count self.pdf_to_images(input_pdf, temp_dir, dpi300) # 对每张图片进行OCR识别 for i in range(image_count): image_path os.path.join(temp_dir, fpage_{i1:03d}.jpg) if os.path.exists(image_path): text pytesseract.image_to_string(Image.open(image_path), langchi_simeng) ocr_text f第 {i1} 页:\n{text}\n\n # 清理临时文件 for file in os.listdir(temp_dir): os.remove(os.path.join(temp_dir, file)) os.rmdir(temp_dir) except Exception as e: print(fOCR识别失败: {str(e)}) return ocr_text5. 完整工作流整合5.1 自动化处理管道# workflow_manager.py from image_processor import ImageProcessor from pdf_processor import PDFProcessor import os from datetime import datetime class DocumentWorkflow: def __init__(self, base_workspaceworkspace): self.workspace base_workspace self.setup_workspace() def setup_workspace(self): 创建工作区目录结构 directories [ source_images, source_pdfs, processed_images, processed_pdfs, output_docs, temp ] for dir_name in directories: os.makedirs(os.path.join(self.workspace, dir_name), exist_okTrue) def run_image_processing_pipeline(self, resize_to(1200, 800), compress_quality80): 运行图片处理管道 print(开始图片批量处理...) processor ImageProcessor( os.path.join(self.workspace, source_images), os.path.join(self.workspace, processed_images) ) # 执行处理流程 resize_count processor.resize_images(resize_to) compress_count, saved_space processor.compress_images(compress_quality) print(f图片处理完成: 调整尺寸 {resize_count} 张, 压缩 {compress_count} 张) return resize_count, compress_count, saved_space def run_pdf_processing_pipeline(self, enable_ocrTrue): 运行PDF处理管道 print(开始PDF批量处理...) pdf_source_dir os.path.join(self.workspace, source_pdfs) pdf_files [f for f in os.listdir(pdf_source_dir) if f.lower().endswith(.pdf)] processor PDFProcessor() total_text_length 0 for pdf_file in pdf_files: pdf_path os.path.join(pdf_source_dir, pdf_file) print(f\n处理PDF: {pdf_file}) # 提取文本 text_content processor.extract_text(pdf_path, use_ocrenable_ocr) total_text_length len(text_content) # 保存提取的文本 text_output_path os.path.join( self.workspace, output_docs, f{os.path.splitext(pdf_file)[0]}_extracted.txt ) with open(text_output_path, w, encodingutf-8) as f: f.write(text_content) print(f文本已保存: {text_output_path}) print(fPDF处理完成: 共处理 {len(pdf_files)} 个文件, 提取文本 {total_text_length} 字符) return len(pdf_files), total_text_length5.2 批处理脚本示例# batch_processor.py #!/usr/bin/env python3 图片和PDF批量处理脚本 使用方法: python batch_processor.py --images --pdfs --workspace ./my_docs import argparse import sys from workflow_manager import DocumentWorkflow def main(): parser argparse.ArgumentParser(description文档批量处理工具) parser.add_argument(--images, actionstore_true, help处理图片) parser.add_argument(--pdfs, actionstore_true, help处理PDF) parser.add_argument(--workspace, defaultworkspace, help工作目录) parser.add_argument(--resize, nargs2, typeint, default[1200, 800], help图片目标尺寸 宽 高) parser.add_argument(--quality, typeint, default80, help压缩质量) args parser.parse_args() if not args.images and not args.pdfs: print(请指定处理类型: --images 或 --pdfs) sys.exit(1) # 初始化工作流 workflow DocumentWorkflow(args.workspace) results {} # 执行图片处理 if args.images: print( * 50) print(开始图片批量处理流程) print( * 50) resize_count, compress_count, saved_space workflow.run_image_processing_pipeline( tuple(args.resize), args.quality ) results[images] { resized: resize_count, compressed: compress_count, saved_space_kb: saved_space // 1024 } # 执行PDF处理 if args.pdfs: print( * 50) print(开始PDF批量处理流程) print( * 50) pdf_count, text_length workflow.run_pdf_processing_pipeline() results[pdfs] { processed: pdf_count, text_extracted: text_length } # 输出处理报告 print(\n * 50) print(处理完成报告) print( * 50) for task_type, stats in results.items(): print(f\n{task_type.upper()} 处理结果:) for stat_name, value in stats.items(): print(f {stat_name}: {value}) if __name__ __main__: main()6. 实战案例技术文档整理6.1 场景描述假设我们有一个技术项目包含以下文件50张不同尺寸的架构图、流程图截图3份扫描版的技术规范PDF文档需要统一处理为标准化格式6.2 具体操作步骤# 1. 准备目录结构 mkdir -p my_project/{source_images,source_pdfs} # 2. 将文件放入对应目录 # source_images/ 放入所有图片文件 # source_pdfs/ 放入所有PDF文件 # 3. 运行处理脚本 python batch_processor.py --images --pdfs --workspace my_project --resize 1000 800 --quality 856.3 处理结果验证处理完成后检查生成的文件my_project/processed_images/调整尺寸和压缩后的图片my_project/output_docs/从PDF提取的文本内容7. 常见问题与解决方案7.1 图片处理常见问题问题现象可能原因解决方案图片处理后颜色失真色彩模式不匹配转换时确保使用RGB模式透明背景变成黑色格式转换问题PNG转JPEG时处理透明度文件大小反而变大压缩参数设置不当调整quality参数通常70-85为宜处理速度过慢图片尺寸过大先缩小尺寸再进行处理7.2 PDF处理常见问题问题现象可能原因解决方案中文OCR识别率低语言包未安装安装中文语言包chi_simPDF转换图片失败poppler路径错误指定正确的poppler路径文本提取为空扫描版PDF启用OCR功能内存不足错误PDF页数过多分批次处理7.3 环境配置问题排查# troubleshooting.py def diagnose_common_issues(): 诊断常见环境问题 issues [] # 检查Poppler路径 try: from pdf2image import convert_from_path test_pdf test.pdf if os.path.exists(test_pdf): convert_from_path(test_pdf, first_page1, last_page1) except Exception as e: issues.append(fPDF转换问题: {e}) # 检查Tesseract OCR try: import pytesseract pytesseract.get_tesseract_version() except Exception as e: issues.append(fOCR引擎问题: {e}) return issues8. 性能优化与最佳实践8.1 内存优化策略# optimized_processor.py class OptimizedImageProcessor(ImageProcessor): def __init__(self, source_dir, output_dir, max_memory_mb500): super().__init__(source_dir, output_dir) self.max_memory_mb max_memory_mb def process_large_batch(self, batch_size10): 分批处理大文件集避免内存溢出 all_images self.find_images() for i in range(0, len(all_images), batch_size): batch all_images[i:i batch_size] print(f处理批次 {i//batch_size 1}/{(len(all_images)-1)//batch_size 1}) for img_path in batch: self._process_single_image(img_path) # 手动触发垃圾回收 import gc gc.collect()8.2 多线程处理加速# parallel_processor.py import concurrent.futures from functools import partial class ParallelProcessor: def __init__(self, max_workers4): self.max_workers max_workers def parallel_image_processing(self, image_paths, process_function): 并行处理图片 with concurrent.futures.ThreadPoolExecutor(max_workersself.max_workers) as executor: # 使用partial固定其他参数 process_func partial(process_function) results list(executor.map(process_func, image_paths)) return results8.3 配置文件管理# config_manager.py import json from pathlib import Path class ConfigManager: def __init__(self, config_fileprocessing_config.json): self.config_file Path(config_file) self.default_config { image_processing: { target_size: [1200, 800], compression_quality: 80, keep_aspect_ratio: True }, pdf_processing: { ocr_enabled: True, dpi: 200, language: chi_simeng } } def load_config(self): 加载配置文件 if self.config_file.exists(): with open(self.config_file, r, encodingutf-8) as f: return json.load(f) else: self.save_config(self.default_config) return self.default_config def save_config(self, config): 保存配置文件 with open(self.config_file, w, encodingutf-8) as f: json.dump(config, f, indent2, ensure_asciiFalse)9. 扩展功能与自定义开发9.1 添加水印功能# watermark_processor.py from PIL import Image, ImageDraw, ImageFont class WatermarkProcessor: def add_watermark_batch(self, image_dir, watermark_text, positionbottom-right): 批量添加水印 processor ImageProcessor(image_dir, image_dir _watermarked) images processor.find_images() for img_path in images: self._add_single_watermark(img_path, watermark_text, position) def _add_single_watermark(self, image_path, text, position): 为单张图片添加水印 with Image.open(image_path) as img: # 创建水印层 watermark Image.new(RGBA, img.size, (0, 0, 0, 0)) draw ImageDraw.Draw(watermark) # 计算水印位置 bbox draw.textbbox((0, 0), text) text_width bbox[2] - bbox[0] text_height bbox[3] - bbox[1] if position bottom-right: x img.width - text_width - 20 y img.height - text_height - 20 elif position center: x (img.width - text_width) // 2 y (img.height - text_height) // 2 # 绘制水印文字 draw.text((x, y), text, fill(255, 255, 255, 128)) # 合并图片和水印 watermarked Image.alpha_composite(img.convert(RGBA), watermark) watermarked watermarked.convert(RGB) # 保存结果 output_path self.output_dir / fwatermarked_{image_path.name} watermarked.save(output_path, quality85)9.2 批量重命名与元数据处理# metadata_processor.py from PIL.ExifTags import TAGS import os from datetime import datetime class MetadataProcessor: def rename_by_date(self, image_dir, pattern{date}_{counter}): 按拍摄日期重命名图片 images self.find_images(image_dir) for counter, img_path in enumerate(images, 1): try: with Image.open(img_path) as img: exif_data img._getexif() date_taken None if exif_data: for tag_id, value in exif_data.items(): tag TAGS.get(tag_id, tag_id) if tag DateTime: date_taken value.replace(:, ).replace( , _) break # 如果没有EXIF数据使用文件修改时间 if not date_taken: mtime os.path.getmtime(img_path) date_taken datetime.fromtimestamp(mtime).strftime(%Y%m%d_%H%M%S) new_name pattern.format(datedate_taken, countercounter) new_path img_path.parent / f{new_name}{img_path.suffix} os.rename(img_path, new_path) print(f重命名: {img_path.name} - {new_path.name}) except Exception as e: print(f重命名失败 {img_path.name}: {str(e)})这套图片和PDF批量处理方案在实际项目中经过验证能够将原本需要数小时的手工操作压缩到几分钟内完成。关键在于根据具体需求调整参数并建立标准化的处理流程。建议在实际使用前先用少量测试文件验证处理效果确认符合预期后再进行批量处理。对于重要的原始文件务必先做好备份避免处理过程中出现意外情况导致数据丢失。