如何用Python高效检测并分离目录中损坏/不可读及密码保护PDF文件
高效批量检测并分离异常PDF文件的Python方案
针对你需要处理10万+多页PDF、分离损坏/不可读及密码保护文件的需求,我整理了一套基于**PyMuPDF(fitz)**的高性能解决方案——PyMuPDF以其极快的解析速度和稳定的异常处理能力,非常适合大规模文件处理场景,同时搭配多进程并行处理,能最大化利用系统资源。
一、环境准备
首先安装核心依赖:
pip install pymupdf
如果需要更完善的日志和路径管理,也可以安装辅助库:
pip install python-dotenv logging
二、核心检测逻辑
我们需要针对两种异常场景做精准检测:
- 密码保护的PDF:PyMuPDF尝试打开时会抛出
fitz.PasswordError异常 - 损坏/不可读的PDF:会抛出
fitz.FileDataError或其他IO相关异常
同时,为了确保PDF确实可读,我们会尝试读取第一页的内容(避免那些能打开但无法解析内容的损坏文件)。
三、完整代码实现
下面是可直接运行的代码,包含多进程处理、日志记录和结果分类:
import fitz import os from pathlib import Path from concurrent.futures import ProcessPoolExecutor, as_completed import logging # 配置日志,记录检测过程和异常 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('pdf_detection.log'), logging.StreamHandler() ] ) def check_pdf(file_path): """检测单个PDF文件的状态""" file_path = Path(file_path) # 跳过非PDF文件 if file_path.suffix.lower() != '.pdf': return (file_path, 'non_pdf') try: # 尝试打开PDF with fitz.open(file_path) as doc: # 检查是否需要密码 if doc.needs_pass: return (file_path, 'password_protected') # 尝试读取第一页内容,确保文件可解析 first_page = doc.load_page(0) if not first_page.get_text(): # 极端情况:能打开但无内容,视为不可读 return (file_path, 'unreadable') return (file_path, 'normal') except fitz.PasswordError: return (file_path, 'password_protected') except fitz.FileDataError: return (file_path, 'corrupted') except Exception as e: # 捕获其他未知异常,标记为不可读 logging.warning(f"Unknown error with {file_path}: {str(e)}") return (file_path, 'unreadable') def main(pdf_directory, max_workers=None): """批量处理目录下的所有PDF文件""" pdf_dir = Path(pdf_directory) if not pdf_dir.is_dir(): logging.error(f"Directory {pdf_dir} does not exist!") return # 获取所有文件路径 all_files = [str(file) for file in pdf_dir.rglob('*') if file.is_file()] logging.info(f"Found {len(all_files)} files to process") # 初始化结果容器 results = { 'normal': [], 'password_protected': [], 'corrupted': [], 'unreadable': [], 'non_pdf': [] } # 用多进程并行处理,默认使用CPU核心数 worker_count = max_workers or os.cpu_count() logging.info(f"Using {worker_count} processes for parallel processing") with ProcessPoolExecutor(max_workers=worker_count) as executor: # 提交所有任务 futures = {executor.submit(check_pdf, file): file for file in all_files} # 处理完成的任务 for future in as_completed(futures): file_path, status = future.result() results[status].append(str(file_path)) logging.info(f"Processed {file_path} - Status: {status}") # 将结果保存到文件,方便后续查看 for status, files in results.items(): if files: with open(f"{status}_pdfs.txt", 'w', encoding='utf-8') as f: f.write('\n'.join(files)) logging.info(f"Saved {len(files)} {status} files to {status}_pdfs.txt") if __name__ == "__main__": # 替换为你的PDF目录路径 PDF_DIRECTORY = "/path/to/your/pdf/directory" main(PDF_DIRECTORY)
四、优化与扩展建议
- 控制进程数:如果你的系统内存有限,可以手动设置
max_workers参数(比如设为CPU核心数的一半),避免内存过载 - 增量处理:可以维护一个已处理文件的哈希表或日志,后续只处理新增的文件,节省时间
- 精细异常区分:如果需要更细致的异常分类,可以捕获更多PyMuPDF的特定异常类型
- 大文件优化:对于超大PDF,可以只检测文件头和第一页,无需加载整个文件,进一步提升速度
- 结果可视化:可以把结果导入Excel或生成统计报表,方便查看各类异常文件的占比
五、注意事项
- 部分PDF可能设置了权限密码而非打开密码,这类文件可以正常打开但无法修改,如果你需要区分这种情况,可以在代码中添加
doc.permissions的检测 - 确保你的Python版本在3.7及以上,
concurrent.futures和pathlib的兼容性更好 - 处理前建议先在小批量PDF上测试代码,确保逻辑符合你的预期
内容的提问来源于stack exchange,提问作者P.Natu




