DeepSeek-OCR-2实战教程:Python调用Gradio API实现批量PDF OCR脚本

1. 项目介绍与准备工作

DeepSeek-OCR-2是DeepSeek团队在2026年1月发布的开源OCR模型,它采用了创新的DeepEncoder V2技术,能够智能理解图像内容并动态重排识别区域,而不是传统的从左到右机械扫描。这个模型只需要256到1120个视觉token就能处理复杂的文档页面,在OmniDocBench v1.5评测中获得了91.09%的高分。

传统的OCR工具往往需要复杂的配置和手动操作,而DeepSeek-OCR-2通过Gradio提供了友好的Web界面。但如果我们想要批量处理大量PDF文件,或者将OCR功能集成到自己的系统中,就需要通过API方式来调用。本教程将教你如何用Python编写脚本,通过Gradio API批量处理PDF文件。

1.1 环境准备

在开始之前,确保你的系统已经安装了Python 3.8或更高版本。我们需要安装几个必要的Python库:

pip install requests gradio_client pdf2image python-dotenv

这些库的作用分别是:

  • requests:用于发送HTTP请求到Gradio API
  • gradio_client:Gradio的官方客户端库,简化API调用
  • pdf2image:将PDF文件转换为图像,因为OCR模型通常处理图像格式
  • python-dotenv:管理环境变量,保护你的API密钥等敏感信息

1.2 获取Gradio API地址

首先,你需要知道DeepSeek-OCR-2的Gradio应用地址。这个地址通常是公开的,但可能会变化。你可以在DeepSeek的官方文档或社区中找到最新的地址。

2. 理解Gradio API的工作原理

Gradio为每个Web应用自动生成了API端点,我们可以通过HTTP请求或者使用gradio_client库来调用这些API。DeepSeek-OCR-2的Gradio应用通常有两个主要功能:单张图片OCR和PDF文件OCR。

2.1 API端点分析

通过查看Gradio应用的界面,我们可以推断出API的参数。通常,OCR应用的输入是一个文件(图片或PDF),输出是识别后的文本。

使用gradio_client库,我们可以这样连接到一个Gradio应用:

from gradio_client import Client

client = Client("https://your-gradio-app-url.com/")

然后通过查看client.view_api()来了解可用的端点和参数。

3. 编写批量PDF OCR脚本

现在我们来编写完整的批量处理脚本。这个脚本会遍历指定文件夹中的所有PDF文件,逐个发送到DeepSeek-OCR-2进行处理,并保存识别结果。

3.1 基础脚本结构

import os
import requests
from pdf2image import convert_from_path
import time
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

class DeepSeekOCRProcessor:
    def __init__(self, api_url):
        self.api_url = api_url
        self.session = requests.Session()
    
    def process_pdf(self, pdf_path):
        """处理单个PDF文件"""
        try:
            # 将PDF转换为图像
            images = convert_from_path(pdf_path)
            
            results = []
            for i, image in enumerate(images):
                # 临时保存图像
                image_path = f"temp_page_{i}.jpg"
                image.save(image_path, "JPEG")
                
                # 调用OCR API
                text = self._call_ocr_api(image_path)
                results.append(text)
                
                # 清理临时文件
                os.remove(image_path)
                
                # 避免频繁请求,添加延迟
                time.sleep(1)
            
            return "\n\n".join(results)
            
        except Exception as e:
            print(f"处理文件 {pdf_path} 时出错: {str(e)}")
            return None
    
    def _call_ocr_api(self, image_path):
        """调用Gradio OCR API"""
        with open(image_path, "rb") as f:
            files = {"file": f}
            response = self.session.post(
                f"{self.api_url}/api/predict",
                files=files,
                timeout=60
            )
        
        if response.status_code == 200:
            return response.json().get("data", [""])[0]
        else:
            raise Exception(f"API调用失败: {response.status_code}")

3.2 使用gradio_client的优化版本

使用官方的gradio_client库可以更稳定地调用API:

from gradio_client import Client
import os
from pdf2image import convert_from_path

class DeepSeekOCRProcessor:
    def __init__(self, api_url):
        self.client = Client(api_url)
    
    def process_pdf(self, pdf_path, output_dir="output"):
        """处理PDF并保存结果"""
        os.makedirs(output_dir, exist_ok=True)
        
        # 转换PDF为图像
        images = convert_from_path(pdf_path)
        
        all_text = []
        for i, image in enumerate(images):
            temp_image = f"temp_page_{i}.jpg"
            image.save(temp_image, "JPEG")
            
            try:
                # 调用OCR功能
                result = self.client.predict(
                    temp_image,  # 通常是第一个参数:图像文件
                    api_name="/predict"  # 根据实际API名称调整
                )
                all_text.append(result)
            except Exception as e:
                print(f"处理第 {i+1} 页时出错: {str(e)}")
                all_text.append(f"[第 {i+1} 页识别失败]")
            
            # 清理临时文件
            os.remove(temp_image)
        
        # 保存结果
        base_name = os.path.splitext(os.path.basename(pdf_path))[0]
        output_file = os.path.join(output_dir, f"{base_name}_ocr.txt")
        
        with open(output_file, "w", encoding="utf-8") as f:
            f.write("\n\n".join(all_text))
        
        return output_file

3.3 批量处理功能

def batch_process_pdfs(pdf_directory, api_url, output_dir="batch_output"):
    """批量处理文件夹中的所有PDF文件"""
    processor = DeepSeekOCRProcessor(api_url)
    
    # 确保输出目录存在
    os.makedirs(output_dir, exist_ok=True)
    
    # 查找所有PDF文件
    pdf_files = [f for f in os.listdir(pdf_directory) 
                if f.lower().endswith('.pdf')]
    
    results = []
    for pdf_file in pdf_files:
        pdf_path = os.path.join(pdf_directory, pdf_file)
        print(f"正在处理: {pdf_file}")
        
        output_file = processor.process_pdf(pdf_path, output_dir)
        results.append({
            "input": pdf_file,
            "output": output_file,
            "status": "成功" if output_file else "失败"
        })
    
    # 生成处理报告
    with open(os.path.join(output_dir, "processing_report.txt"), "w") as f:
        for result in results:
            f.write(f"{result['input']} -> {result['output']} [{result['status']}]\n")
    
    return results

4. 完整可运行脚本

下面是一个完整的、可以直接使用的脚本:

#!/usr/bin/env python3
"""
DeepSeek-OCR-2 批量PDF处理脚本
作者: [你的名字]
日期: 2024年
"""

import os
import argparse
from gradio_client import Client
from pdf2image import convert_from_path
import time
from typing import List

class DeepSeekOCRBatchProcessor:
    def __init__(self, api_url: str):
        """
        初始化OCR处理器
        
        Args:
            api_url: Gradio应用的URL
        """
        self.client = Client(api_url)
        self.api_url = api_url
    
    def process_single_pdf(self, pdf_path: str, output_dir: str = "output") -> str:
        """
        处理单个PDF文件
        
        Args:
            pdf_path: PDF文件路径
            output_dir: 输出目录
            
        Returns:
            输出文件路径
        """
        # 创建输出目录
        os.makedirs(output_dir, exist_ok=True)
        
        print(f"开始处理: {os.path.basename(pdf_path)}")
        
        try:
            # 转换PDF为图像
            images = convert_from_path(pdf_path)
            print(f"PDF共 {len(images)} 页")
            
            all_text = []
            for page_num, image in enumerate(images, 1):
                print(f"正在处理第 {page_num} 页...")
                
                # 保存临时图像文件
                temp_image = f"temp_page_{page_num}.jpg"
                image.save(temp_image, "JPEG", quality=95)
                
                try:
                    # 调用OCR API - 参数可能需要根据实际API调整
                    result = self.client.predict(
                        temp_image,
                        api_name="/predict"
                    )
                    all_text.append(f"--- 第 {page_num} 页 ---\n{result}")
                    
                except Exception as e:
                    error_msg = f"第 {page_num} 页处理失败: {str(e)}"
                    print(error_msg)
                    all_text.append(f"--- 第 {page_num} 页 [识别失败] ---\n{error_msg}")
                
                finally:
                    # 清理临时文件
                    if os.path.exists(temp_image):
                        os.remove(temp_image)
                
                # 添加延迟避免频繁请求
                time.sleep(0.5)
            
            # 生成输出文件名
            base_name = os.path.splitext(os.path.basename(pdf_path))[0]
            output_file = os.path.join(output_dir, f"{base_name}_ocr.txt")
            
            # 保存结果
            with open(output_file, "w", encoding="utf-8") as f:
                f.write("\n\n".join(all_text))
            
            print(f"处理完成: {output_file}")
            return output_file
            
        except Exception as e:
            print(f"处理文件 {pdf_path} 时发生错误: {str(e)}")
            return None
    
    def process_batch(self, input_dir: str, output_dir: str = "batch_output") -> List[dict]:
        """
        批量处理目录中的所有PDF文件
        
        Args:
            input_dir: 输入目录路径
            output_dir: 输出目录路径
            
        Returns:
            处理结果列表
        """
        # 确保目录存在
        os.makedirs(output_dir, exist_ok=True)
        
        # 查找所有PDF文件
        pdf_files = []
        for file in os.listdir(input_dir):
            if file.lower().endswith('.pdf'):
                pdf_files.append(os.path.join(input_dir, file))
        
        print(f"找到 {len(pdf_files)} 个PDF文件")
        
        results = []
        for pdf_file in pdf_files:
            result_file = self.process_single_pdf(pdf_file, output_dir)
            
            results.append({
                "input_file": pdf_file,
                "output_file": result_file,
                "status": "成功" if result_file else "失败"
            })
        
        # 生成处理报告
        report_file = os.path.join(output_dir, "processing_report.txt")
        with open(report_file, "w", encoding="utf-8") as f:
            f.write("DeepSeek-OCR-2 批量处理报告\n")
            f.write("=" * 50 + "\n\n")
            
            success_count = sum(1 for r in results if r["status"] == "成功")
            f.write(f"处理总结: 成功 {success_count}/{len(results)} 个文件\n\n")
            
            for result in results:
                f.write(f"输入: {os.path.basename(result['input_file'])}\n")
                f.write(f"输出: {os.path.basename(result['output_file']) if result['output_file'] else '无'}\n")
                f.write(f"状态: {result['status']}\n")
                f.write("-" * 30 + "\n")
        
        print(f"批量处理完成!报告已保存至: {report_file}")
        return results

def main():
    """主函数"""
    parser = argparse.ArgumentParser(description="DeepSeek-OCR-2 批量PDF处理工具")
    parser.add_argument("--api-url", required=True, help="Gradio应用URL")
    parser.add_argument("--input", required=True, help="输入PDF文件或目录")
    parser.add_argument("--output", default="output", help="输出目录")
    parser.add_argument("--batch", action="store_true", help批量处理模式")
    
    args = parser.parse_args()
    
    # 初始化处理器
    processor = DeepSeekOCRBatchProcessor(args.api_url)
    
    if args.batch:
        # 批量处理模式
        if not os.path.isdir(args.input):
            print("错误: 批量处理模式需要输入目录")
            return
        
        processor.process_batch(args.input, args.output)
    else:
        # 单文件模式
        if not os.path.isfile(args.input):
            print("错误: 请输入有效的PDF文件")
            return
        
        processor.process_single_pdf(args.input, args.output)

if __name__ == "__main__":
    main()

5. 使用方法和示例

5.1 安装依赖

首先安装所需的Python包:

pip install requests gradio_client pdf2image python-dotenv

5.2 处理单个PDF文件

python deepseek_ocr_batch.py \
    --api-url "https://your-gradio-app-url.com/" \
    --input "document.pdf" \
    --output "results"

5.3 批量处理多个PDF

python deepseek_ocr_batch.py \
    --api-url "https://your-gradio-app-url.com/" \
    --input "pdf_folder/" \
    --output "batch_results" \
    --batch

5.4 在Python中直接使用

你也可以在自己的Python项目中直接使用这个类:

from deepseek_ocr_batch import DeepSeekOCRBatchProcessor

# 初始化处理器
processor = DeepSeekOCRBatchProcessor("https://your-gradio-app-url.com/")

# 处理单个文件
result = processor.process_single_pdf("document.pdf", "output")

# 或者批量处理
results = processor.process_batch("pdf_folder/", "batch_output")

6. 常见问题与解决方案

6.1 API连接问题

如果遇到连接问题,首先检查API URL是否正确,以及网络是否通畅:

# 测试连接
import requests
try:
    response = requests.get("https://your-gradio-app-url.com/", timeout=10)
    print("连接成功")
except Exception as e:
    print(f"连接失败: {e}")

6.2 处理速度优化

对于大量PDF处理,可以考虑以下优化措施:

# 调整PDF转换质量以提高速度
images = convert_from_path(pdf_path, dpi=200)  # 降低DPI提高速度

# 使用多线程处理(注意API的并发限制)
from concurrent.futures import ThreadPoolExecutor, as_completed

def process_image(image_data):
    image, page_num = image_data
    # 处理单张图像
    # ...

# 使用线程池
with ThreadPoolExecutor(max_workers=2) as executor:  # 限制并发数
    futures = [executor.submit(process_image, (img, i)) 
               for i, img in enumerate(images)]
    
    for future in as_completed(futures):
        result = future.result()
        # 处理结果

6.3 处理大文件

对于特别大的PDF文件,可以考虑分块处理:

def process_large_pdf(pdf_path, output_dir, chunk_size=50):
    """分块处理大PDF文件"""
    total_pages = get_pdf_page_count(pdf_path)
    
    for start_page in range(0, total_pages, chunk_size):
        end_page = min(start_page + chunk_size, total_pages)
        print(f"处理页面 {start_page + 1}-{end_page}")
        
        # 转换指定范围的页面
        images = convert_from_path(pdf_path, first_page=start_page + 1, 
                                  last_page=end_page)
        
        # 处理这些页面...

7. 总结

通过本教程,你学会了如何使用Python调用DeepSeek-OCR-2的Gradio API来批量处理PDF文件。这个方法不仅适用于DeepSeek-OCR-2,也适用于其他基于Gradio的AI模型。

主要收获

  • 理解了Gradio API的工作原理和调用方式
  • 掌握了PDF到图像的转换技术
  • 学会了编写健壮的批量处理脚本
  • 了解了常见问题的解决方法

下一步建议

  1. 尝试将识别结果保存为更结构化的格式(如JSON、Word文档)
  2. 添加进度条和更详细的日志记录
  3. 探索其他OCR模型的API集成
  4. 考虑添加自动重试机制和错误恢复功能

这个脚本可以作为一个基础框架,你可以根据实际需求进行扩展和优化,比如添加数据库支持、Web界面、或者与其他系统集成。


获取更多AI镜像

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

Logo

欢迎加入DeepSeek 技术社区。在这里,你可以找到志同道合的朋友,共同探索AI技术的奥秘。

更多推荐