You need to enable JavaScript to run this app.
最新活动
大模型
产品
解决方案
定价
生态与合作
支持与服务
开发者
了解我们

如何用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)

四、优化与扩展建议

  1. 控制进程数:如果你的系统内存有限,可以手动设置max_workers参数(比如设为CPU核心数的一半),避免内存过载
  2. 增量处理:可以维护一个已处理文件的哈希表或日志,后续只处理新增的文件,节省时间
  3. 精细异常区分:如果需要更细致的异常分类,可以捕获更多PyMuPDF的特定异常类型
  4. 大文件优化:对于超大PDF,可以只检测文件头和第一页,无需加载整个文件,进一步提升速度
  5. 结果可视化:可以把结果导入Excel或生成统计报表,方便查看各类异常文件的占比

五、注意事项

  • 部分PDF可能设置了权限密码而非打开密码,这类文件可以正常打开但无法修改,如果你需要区分这种情况,可以在代码中添加doc.permissions的检测
  • 确保你的Python版本在3.7及以上,concurrent.futurespathlib的兼容性更好
  • 处理前建议先在小批量PDF上测试代码,确保逻辑符合你的预期

内容的提问来源于stack exchange,提问作者P.Natu

火山引擎 最新活动