PDF-Parser-1.0批量处理：写个Python脚本，自动解析文件夹里所有PDF

1. 为什么需要批量处理PDF？

在日常工作和学习中，我们经常会遇到需要处理大量PDF文档的情况。比如：

• 学术研究人员需要从几十篇论文中提取关键数据和结论
• 财务人员要分析季度报表中的数百页数据
• 法律从业者需要审阅多份合同文档
• 内容创作者要从参考资料中整理有用信息

传统的手动处理方式效率极低，而且容易出错。想象一下，你要从100个PDF文件中提取表格数据，每个文件平均20页，手动复制粘贴不仅耗时数天，还难免会出现遗漏或错误。

这就是为什么我们需要自动化批量处理PDF的解决方案。本文将教你如何使用PDF-Parser-1.0文档理解模型，编写一个Python脚本，自动解析指定文件夹中的所有PDF文件，高效完成这项繁琐任务。

2. 准备工作：了解PDF-Parser-1.0

2.1 PDF-Parser-1.0的核心能力

PDF-Parser-1.0是一个基于深度学习的文档理解模型，主要提供以下功能：

• 文本提取：准确识别PDF中的文字内容，保持原有段落结构
• 表格识别：将PDF中的表格转换为结构化数据，保留表头和单元格关系
• 公式识别：提取数学公式并转换为LaTeX格式
• 布局分析：识别文档中的标题、正文、图片等元素的排版位置

2.2 两种使用模式

PDF-Parser-1.0支持两种使用方式：

1. Web界面：通过浏览器上传单个PDF文件进行解析
2. API接口：通过HTTP请求批量处理PDF文件，适合编程集成

对于批量处理需求，我们将重点使用API接口方式。

3. 编写Python批量处理脚本

3.1 脚本功能设计

我们的批量处理脚本需要实现以下功能：

1. 遍历指定文件夹中的所有PDF文件
2. 逐个调用PDF-Parser-1.0的API接口进行处理
3. 将解析结果保存为结构化数据（JSON格式）
4. 记录处理日志，便于排查问题

3.2 完整脚本代码

以下是实现上述功能的Python脚本：

import os
import requests
import json
import time
from tqdm import tqdm  # 进度条库

class PDFBatchProcessor:
    def __init__(self, api_url="http://localhost:7860/api/predict"):
        self.api_url = api_url
        self.session = requests.Session()
        
    def process_single_pdf(self, pdf_path, mode="full"):
        """处理单个PDF文件"""
        try:
            with open(pdf_path, "rb") as f:
                files = {"file": f}
                data = {"mode": mode}
                
                # 设置超时时间为10分钟
                response = self.session.post(
                    self.api_url,
                    files=files,
                    data=data,
                    timeout=600
                )
                
            if response.status_code == 200:
                return True, response.json()
            else:
                return False, f"API请求失败，状态码：{response.status_code}"
                
        except Exception as e:
            return False, f"处理文件时出错：{str(e)}"
    
    def batch_process(self, input_folder, output_folder, mode="full"):
        """批量处理文件夹中的所有PDF"""
        # 确保输出文件夹存在
        os.makedirs(output_folder, exist_ok=True)
        
        # 获取所有PDF文件
        pdf_files = [
            f for f in os.listdir(input_folder) 
            if f.lower().endswith(".pdf")
        ]
        
        if not pdf_files:
            print(f"在文件夹 {input_folder} 中未找到PDF文件")
            return
        
        print(f"找到 {len(pdf_files)} 个PDF文件需要处理")
        
        # 准备日志文件
        log_file = os.path.join(output_folder, "processing_log.txt")
        with open(log_file, "w", encoding="utf-8") as log:
            log.write(f"批量处理日志 - {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
            log.write(f"输入文件夹: {input_folder}\n")
            log.write(f"输出文件夹: {output_folder}\n")
            log.write(f"处理模式: {mode}\n\n")
            
            # 使用进度条显示处理进度
            for pdf_file in tqdm(pdf_files, desc="处理进度"):
                file_path = os.path.join(input_folder, pdf_file)
                output_file = os.path.join(
                    output_folder, 
                    f"{os.path.splitext(pdf_file)[0]}.json"
                )
                
                log.write(f"\n处理文件: {pdf_file}\n")
                
                # 处理PDF文件
                success, result = self.process_single_pdf(file_path, mode)
                
                if success:
                    # 保存结果到JSON文件
                    with open(output_file, "w", encoding="utf-8") as f:
                        json.dump(result, f, ensure_ascii=False, indent=2)
                    log.write(f"状态: 成功\n")
                    log.write(f"结果已保存到: {output_file}\n")
                    
                    # 记录基本信息
                    log.write(f"提取文本长度: {len(result.get('text', ''))}\n")
                    log.write(f"识别表格数量: {len(result.get('tables', []))}\n")
                    log.write(f"识别公式数量: {len(result.get('formulas', []))}\n")
                else:
                    log.write(f"状态: 失败\n")
                    log.write(f"错误信息: {result}\n")
                
                # 避免服务器压力过大，处理间隔1秒
                time.sleep(1)
            
            log.write("\n批量处理完成!\n")
        
        print(f"\n所有文件处理完成! 日志已保存到 {log_file}")

if __name__ == "__main__":
    # 使用示例
    processor = PDFBatchProcessor()
    
    # 设置输入输出文件夹
    input_folder = "/path/to/your/pdf/folder"  # 替换为你的PDF文件夹路径
    output_folder = "/path/to/output/results"  # 替换为输出结果文件夹路径
    
    # 开始批量处理
    processor.batch_process(input_folder, output_folder)

3.3 脚本使用说明

1. 安装依赖： 在运行脚本前，需要安装必要的Python库：

   pip install requests tqdm
   

2. 修改路径： 将脚本中的input_folder和output_folder修改为你实际的文件夹路径。

3. 运行脚本：

   python batch_pdf_parser.py
   

4. 查看结果：

   • 每个PDF的解析结果会保存为单独的JSON文件
   • 处理日志保存在processing_log.txt中

4. 脚本功能详解

4.1 核心类：PDFBatchProcessor

这个类封装了批量处理的主要逻辑：

• __init__：初始化API地址和HTTP会话
• process_single_pdf：处理单个PDF文件
• batch_process：批量处理整个文件夹

4.2 处理模式选择

脚本支持两种处理模式：

• full：完整分析模式，提取文本、表格、公式等所有内容
• text：快速提取模式，只提取文本内容

在batch_process方法中可以通过mode参数指定。

4.3 错误处理与日志

脚本包含完善的错误处理机制：

1. 每个文件的处理结果都会记录到日志文件
2. 处理失败时会记录详细错误信息
3. 使用try-except捕获各种异常情况

4.4 进度显示

使用tqdm库显示处理进度条，直观展示：

• 已处理文件数
• 剩余文件数
• 预计剩余时间

5. 解析结果的使用

5.1 结果数据结构

每个PDF的解析结果保存为JSON格式，主要包含以下字段：

{
  "text": "提取的全部文本内容...",
  "tables": [
    {
      "html": "<table>...</table>",
      "markdown": "| Header1 | Header2 |...",
      "cells": [...]
    }
  ],
  "formulas": [
    {
      "latex": "E=mc^2",
      "position": [x1, y1, x2, y2]
    }
  ],
  "layout": {
    "pages": [
      {
        "width": 595,
        "height": 842,
        "blocks": [...]
      }
    ]
  }
}

5.2 结果后处理示例

以下是一些常见的结果后处理代码示例：

提取所有表格数据并保存为CSV：

import pandas as pd

def tables_to_csv(result_json, output_folder):
    """将JSON结果中的表格保存为CSV文件"""
    with open(result_json, "r", encoding="utf-8") as f:
        result = json.load(f)
    
    for i, table in enumerate(result.get("tables", [])):
        # 使用pandas读取HTML表格
        df = pd.read_html(table["html"])[0]
        
        # 生成输出文件名
        base_name = os.path.splitext(os.path.basename(result_json))[0]
        csv_file = os.path.join(output_folder, f"{base_name}_table_{i+1}.csv")
        
        # 保存为CSV
        df.to_csv(csv_file, index=False, encoding="utf-8-sig")

提取所有公式并保存为LaTeX文件：

def formulas_to_latex(result_json, output_folder):
    """提取公式并保存为LaTeX文件"""
    with open(result_json, "r", encoding="utf-8") as f:
        result = json.load(f)
    
    formulas = result.get("formulas", [])
    if not formulas:
        return
    
    # 生成输出文件名
    base_name = os.path.splitext(os.path.basename(result_json))[0]
    latex_file = os.path.join(output_folder, f"{base_name}_formulas.tex")
    
    # 保存所有公式
    with open(latex_file, "w", encoding="utf-8") as f:
        f.write("% 从PDF中提取的数学公式\n\n")
        for i, formula in enumerate(formulas, 1):
            f.write(f"% 公式 {i}\n")
            f.write(f"{formula['latex']}\n\n")

6. 高级技巧与优化建议

6.1 处理大型PDF文件

对于超过50页的大型PDF文件，建议：

1. 增加超时时间：修改process_single_pdf方法中的timeout参数
2. 分批处理：将大PDF拆分为多个小文件
3. 监控内存：处理过程中注意系统内存使用情况

6.2 性能优化

1. 多线程处理：可以使用concurrent.futures实现并行处理

   from concurrent.futures import ThreadPoolExecutor
   
   def batch_process_parallel(self, input_folder, output_folder, workers=4):
       """多线程批量处理"""
       pdf_files = [f for f in os.listdir(input_folder) if f.lower().endswith(".pdf")]
       
       with ThreadPoolExecutor(max_workers=workers) as executor:
           futures = []
           for pdf_file in pdf_files:
               file_path = os.path.join(input_folder, pdf_file)
               future = executor.submit(self.process_single_pdf, file_path)
               futures.append((pdf_file, future))
           
           for pdf_file, future in futures:
               success, result = future.result()
               # 保存结果...
   

2. 断点续处理：记录已处理文件，避免重复处理

   def get_processed_files(output_folder):
       """获取已处理的文件列表"""
       return [f.replace(".json", ".pdf") 
               for f in os.listdir(output_folder) 
               if f.endswith(".json")]
   
   # 在batch_process中过滤已处理文件
   processed = get_processed_files(output_folder)
   pdf_files = [f for f in os.listdir(input_folder) 
                if f.lower().endswith(".pdf") and f not in processed]
   

6.3 错误处理增强

1. 重试机制：对失败的请求自动重试

   def process_single_pdf(self, pdf_path, mode="full", max_retries=3):
       """带重试机制的单个文件处理"""
       for attempt in range(max_retries):
           try:
               success, result = self._process(pdf_path, mode)
               if success:
                   return True, result
           except Exception as e:
               if attempt == max_retries - 1:
                   return False, str(e)
               time.sleep(2 ** attempt)  # 指数退避
       return False, "Max retries exceeded"
   

2. 结果验证：检查返回结果是否完整

   def validate_result(self, result):
       """验证结果是否完整有效"""
       if not isinstance(result, dict):
           return False
       if "text" not in result:
           return False
       return True
   

7. 总结

通过本文介绍的Python脚本，你可以轻松实现：

1. 批量解析文件夹中的所有PDF文件
2. 自动提取文本、表格、公式等结构化数据
3. 将结果保存为易于处理的JSON格式
4. 通过日志监控处理进度和结果

这个解决方案特别适合需要处理大量PDF文档的场景，能够将原本需要数天的手工工作缩短到几小时内完成，效率提升数十倍。

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。