你是不是也经常遇到这样的场景:项目文档需要统一调整图片尺寸,几十张照片要批量压缩上传,或者收到一堆扫描版PDF需要提取文字和图片?手动一张张处理不仅耗时费力,还容易出错。
最近在整理技术文档时,我发现了一个高效的解决方案——通过Python脚本实现图片和PDF的批量处理全流程。这套方法真正解决了重复性工作的痛点,让原本需要半天的手工操作在几分钟内自动完成。
本文将分享从环境搭建到完整实战的详细流程,包含具体的代码实现和常见问题排查。无论你是需要处理技术文档、项目资料还是日常办公文件,这套方案都能显著提升效率。
1. 核心痛点与解决方案对比
在技术文档管理、项目资料整理等场景中,我们经常面临以下典型问题:
传统手工处理的痛点:
- 图片尺寸不统一,需要逐张调整宽高比
- 大量图片需要压缩以减少存储空间
- PDF文档中的图片需要批量提取
- 扫描版PDF需要转换为可编辑文本
- 不同格式的图片需要统一转换格式
自动化方案的优势:
- 批量处理:一次性处理数百个文件
- 一致性保证:所有文件采用相同处理参数
- 时间节省:从小时级缩短到分钟级
- 可重复使用:处理逻辑可封装为脚本重复调用
以技术文档为例,一个中等规模的项目可能包含50-100张示意图、架构图等图片资源。手动处理每张图片平均需要2-3分钟,而自动化脚本可以在30秒内完成全部处理。
2. 环境准备与工具选择
2.1 基础环境要求
# 检查Python版本 python --version # 推荐 Python 3.8+2.2 核心依赖库安装
pip install Pillow PyPDF2 pdf2image python-docx pip install opencv-python pytesseract2.3 各库的功能说明
| 库名称 | 主要功能 | 适用场景 |
|---|---|---|
| Pillow | 图像处理核心库 | 图片缩放、格式转换、滤镜应用 |
| PyPDF2 | PDF文本处理 | PDF拆分、合并、文本提取 |
| pdf2image | PDF转图片 | 将PDF页面转换为图像格式 |
| python-docx | Word文档操作 | 处理提取的文本内容 |
| OpenCV | 高级图像处理 | 图像识别、边缘检测 |
| pytesseract | OCR文字识别 | 从图片中提取文字 |
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_ok=True) 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_aspect=True): """批量调整图片尺寸""" 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 / f"resized_{img_path.name}" img.save(output_path, optimize=True) 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_format='JPEG', quality=85): """批量转换图片格式""" 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, format=target_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, quality=70, max_size=None): """批量压缩图片""" 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 / f"compressed_{img_path.name}" # 根据格式选择保存参数 if img_path.suffix.lower() in ['.jpg', '.jpeg']: img.save(output_path, optimize=True, quality=quality) else: img.save(output_path, optimize=True) 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_path=None): self.poppler_path = poppler_path def split_pdf(self, input_pdf, output_dir, pages_per_split=10): """拆分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_page+1}-{end_page})")4.2 PDF转图片处理
# 续 pdf_processor.py def pdf_to_images(self, input_pdf, output_dir, dpi=200): """将PDF转换为图片""" os.makedirs(output_dir, exist_ok=True) try: images = convert_from_path(input_pdf, dpi=dpi, poppler_path=self.poppler_path) for i, image in enumerate(images): output_path = os.path.join(output_dir, f"page_{i+1:03d}.jpg") image.save(output_path, 'JPEG', quality=85) print(f"转换页面 {i+1} -> {output_path}") return len(images) except Exception as e: print(f"PDF转换图片失败: {str(e)}") return 04.3 PDF文字提取与OCR
# 续 pdf_processor.py def extract_text(self, input_pdf, use_ocr=False): """提取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_num+1} 页 ---\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_ok=True) try: # 先将PDF转换为图片 image_count = self.pdf_to_images(input_pdf, temp_dir, dpi=300) # 对每张图片进行OCR识别 for i in range(image_count): image_path = os.path.join(temp_dir, f"page_{i+1:03d}.jpg") if os.path.exists(image_path): text = pytesseract.image_to_string(Image.open(image_path), lang='chi_sim+eng') ocr_text += f"第 {i+1} 页:\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(f"OCR识别失败: {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_workspace="workspace"): 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_ok=True) def run_image_processing_pipeline(self, resize_to=(1200, 800), compress_quality=80): """运行图片处理管道""" 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_ocr=True): """运行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_ocr=enable_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', encoding='utf-8') as f: f.write(text_content) print(f"文本已保存: {text_output_path}") print(f"PDF处理完成: 共处理 {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', action='store_true', help='处理图片') parser.add_argument('--pdfs', action='store_true', help='处理PDF') parser.add_argument('--workspace', default='workspace', help='工作目录') parser.add_argument('--resize', nargs=2, type=int, default=[1200, 800], help='图片目标尺寸 宽 高') parser.add_argument('--quality', type=int, default=80, 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_sim |
| PDF转换图片失败 | 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_page=1, last_page=1) except Exception as e: issues.append(f"PDF转换问题: {e}") # 检查Tesseract OCR try: import pytesseract pytesseract.get_tesseract_version() except Exception as e: issues.append(f"OCR引擎问题: {e}") return issues8. 性能优化与最佳实践
8.1 内存优化策略
# optimized_processor.py class OptimizedImageProcessor(ImageProcessor): def __init__(self, source_dir, output_dir, max_memory_mb=500): super().__init__(source_dir, output_dir) self.max_memory_mb = max_memory_mb def process_large_batch(self, batch_size=10): """分批处理大文件集,避免内存溢出""" 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_workers=4): self.max_workers = max_workers def parallel_image_processing(self, image_paths, process_function): """并行处理图片""" with concurrent.futures.ThreadPoolExecutor(max_workers=self.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_file="processing_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_sim+eng" } } def load_config(self): """加载配置文件""" if self.config_file.exists(): with open(self.config_file, 'r', encoding='utf-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', encoding='utf-8') as f: json.dump(config, f, indent=2, ensure_ascii=False)9. 扩展功能与自定义开发
9.1 添加水印功能
# watermark_processor.py from PIL import Image, ImageDraw, ImageFont class WatermarkProcessor: def add_watermark_batch(self, image_dir, watermark_text, position='bottom-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 / f"watermarked_{image_path.name}" watermarked.save(output_path, quality=85)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(date=date_taken, counter=counter) 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批量处理方案在实际项目中经过验证,能够将原本需要数小时的手工操作压缩到几分钟内完成。关键在于根据具体需求调整参数,并建立标准化的处理流程。
建议在实际使用前,先用少量测试文件验证处理效果,确认符合预期后再进行批量处理。对于重要的原始文件,务必先做好备份,避免处理过程中出现意外情况导致数据丢失。