基于Python+Dify实现批量OCR识别的自动化解决方案

目录

• 一、方案整体架构
• 二、前置准备工作
• 1. 开发环境配置
• 2. Dify平台配置
• 三、核心功能实现
• 1. 工具函数：批量读取PDF文件
• 2. PDF转PNG：按页拆分并独立存储
• 3. Dify交互：图片上传与OCR工作流调用
• 4. 文本合并：按原顺序生成结构化文件
• 5. 主函数：串联全流程执行
• 四、关键问题解决方案
• 1. PDF转PNG失败：Poppler未找到
• 2. Dify API调用400错误
• 3. OCR识别结果为空或乱码
• 4. 批量处理速度慢
• 五、方案优势与扩展方向
• 方案优势
• 扩展方向
• 六、总结

在日常工作中，我们经常会遇到大量图片格式的PDF文件（如扫描件、截图生成的PDF），这类文件无法直接复制文本，手动逐页OCR识别效率极低。本文将介绍一套基于Python+Dify的全自动化解决方案，实现从PDF批量转图片、图片上传OCR识别到文本合并的完整流程，适用于文档数字化、数据提取等场景。

本方案围绕"批量处理、自动化执行、精准识别"三大核心需求设计，整体架构分为4个关键环节，全程无需人工干预：

graph TD
    A[PDF文件目录] –> B[Python批量读取PDF]
    B –> C[PDF转PNG图片（按页拆分）]
    C –> D[图片上传Dify平台]
    D –> E[Dify OCR工作流识别文本]
    E –> F[按PDF原顺序合并文本]
    F –> G[生成同名结构化文本文件]

核心技术栈

• 后端语言：Python 3.8+（跨平台兼容，生态丰富）
• PDF转图片：pdf2image（封装Poppler引擎，转换精度高）
• 图像处理：Pillow（图片格式标准化）
• API调用：requests（与Dify平台交互）
• OCR识别：Dify工作流（可视化配置，支持第三方OCR工具集成）

在开始实现前，需完成环境配置和平台准备，确保全流程顺畅运行。

（1）Python依赖安装

# 核心依赖库
pip install pdf2image pillow requests glob3 pathlib

（2）Poppler引擎安装（PDF转图片核心依赖）

• Windows：下载二进制包（https://github.com/oschwartz10612/poppler-wheels/releases），解压后将bin目录添加到系统环境变量
• Mac：通过Homebrew安装 brew install poppler
• Linux（Ubuntu/Debian）：sudo apt-get install poppler-utils

Dify作为低代码平台，提供了便捷的工作流配置能力，无需手动开发OCR算法：

• 登录Dify平台（https://dify.ai/），创建「工作流」
• 添加「图片上传」输入节点，字段名设为photo（类型选择「图片」，允许PNG格式）
• 集成OCR识别节点（支持内置OCR或第三方工具如阿里云OCR、百度OCR、谷歌LLM模型）
• 配置输出节点，将识别结果映射为text字段
• 发布工作流，获取「工作流ID」和「API密钥」（个人设置→API密钥）

下面按流程拆解核心功能代码，每个模块均提供完整实现和关键注释。

实现指定目录下所有PDF文件的自动扫描和排序，确保处理顺序可预期：

import glob
import os
from typing import List

def get_all_pdf_in_dir(pdf_dir: str) -> List[str]:
    """
    获取指定目录下所有PDF文件的绝对路径（按文件名排序）
    """
    if not os.path.exists(pdf_dir):
        raise FileNotFoundError(f"PDF目录不存在：{pdf_dir}")
    
    # 匹配所有.pdf文件（不区分大小写）
    pdf_pattern = os.path.join(pdf_dir, "*.pdf")
    pdf_files = glob.glob(pdf_pattern, recursive=False)
    
    # 按文件名排序，确保处理顺序一致
    pdf_files.sort()
    
    if not pdf_files:
        print(f"警告：目录 {pdf_dir} 中未找到PDF文件")
        return []
    
    print(f"成功找到 {len(pdf_files)} 个PDF文件：")
    for idx, pdf_file in enumerate(pdf_files, start=1):
        print(f"  {idx}. {os.path.basename(pdf_file)}")
    return pdf_files

为每个PDF创建独立目录存储图片，避免文件混淆，同时保证页码顺序：

from pdf2image import convert_from_path
from pathlib import Path

def pdf_to_png(
    pdf_path: str,
    root_image_dir: str = "pdf_images",
    dpi: int = 300,
    fmt: str = "png"
) -> List[str]:
    """
    PDF按页转PNG，每个PDF生成独立目录，返回图片路径列表（按页码排序）
    """
    pdf_filename = os.path.splitext(os.path.basename(pdf_path))[0]
    pdf_image_dir = os.path.join(root_image_dir, pdf_filename)
    
    # 创建独立目录（支持多级目录自动创建）
    Path(pdf_image_dir).mkdir(parents=True, exist_ok=True)
    print(f"  图片输出目录：{os.path.abspath(pdf_image_dir)}")
    
    # Mac/Linux可自动识别Poppler，Windows需指定路径
    poppler_path = "/opt/homebrew/Cellar/poppler/25.12.0/bin" if os.name != "nt" else None
    if os.name == "nt" and not poppler_path:
        raise ValueError("Windows系统需指定Poppler的bin目录路径")
    
    try:
        # 按页转换PDF为图片（保持原顺序）
        images = convert_from_path(
            pdf_path=pdf_path,
            dpi=dpi,  # 分辨率越高识别越准，建议≥200
            thread_count=4,  # 多线程提速
            fmt=fmt,
            grayscale=False,
            poppler_path=poppler_path
        )
        
        image_paths = []
        for page_num, image in enumerate(images, start=1):
            output_filename = f"{pdf_filename}_page_{page_num:02d}.{fmt}"
            output_path = os.path.join(pdf_image_dir, output_filename)
            image.save(output_path, fmt.upper())
            image_paths.append(output_path)
        
        print(f"  PDF转PNG完成：共生成 {len(image_paths)} 张图片")
        return image_paths
    except Exception as e:
        print(f"  PDF转PNG失败：{str(e)}")
        raise

基于Dify API实现图片上传和工作流调用，处理网络异常和重试逻辑：

import requests
import time
from typing import Optional

# Dify全局配置（需替换为实际信息）
DIFY_API_KEY = "app-xxxxxxxx"
DIFY_API_BASE_URL = "https://api.dify.ai/v1"
DIFY_USER = "difyuser"
RETRY_TIMES = 3
RETRY_DELAY = 2

def upload_image_to_dify(image_path: str) -> Optional[str]:
    """上传图片到Dify，返回文件ID"""
    upload_url = f"{DIFY_API_BASE_URL}/files/upload"
    headers = {"Authorization": f"Bearer {DIFY_API_KEY}"}
    
    try:
        with open(image_path, 'rb') as file:
            files = {'file': (os.path.basename(image_path), file, 'image/png')}
            data = {"user": DIFY_USER, "type": "IMAGE"}
            
            response = requests.post(upload_url, headers=headers, files=files, data=data)
            if response.status_code == 201:
                file_id = response.json().get("id")
                print(f"    图片上传成功，file_id：{file_id[:10]}...")
                return file_id
            else:
                print(f"    图片上传失败，状态码：{response.status_code}，响应：{response.text}")
                return None
    except Exception as e:
        print(f"    图片上传异常：{str(e)}")
        return None

def run_ocr_workflow(file_id: str) -> Optional[str]:
    """调用Dify OCR工作流，返回识别文本"""
    workflow_url = f"{DIFY_API_BASE_URL}/workflows/run"
    headers = {
        "Authorization": f"Bearer {DIFY_API_KEY}",
        "Content-Type": "application/json"
    }
    
    data = {
        "inputs": {
            "photo": {
                "transfer_method": "local_file",
                "upload_file_id": file_id,
                "type": "image"
            }
        },
        "response_mode": "blocking",
        "user": DIFY_USER
    }
    
    try:
        response = requests.post(workflow_url, headers=headers, json=data)
        if response.status_code == 200:
            result = response.json()
            # 解析Dify响应结构（data→outputs→text）
            if "data" in result and result["data"]["status"] == "succeeded":
                outputs = result["data"].get("outputs", {})
                ocr_text = outputs.get("text", "").strip()
                if ocr_text:
                    print(f"    OCR识别成功，文本长度：{len(ocr_text)}字")
                    return ocr_text
            print(f"    OCR结果解析失败：{result}")
            return None
        else:
            print(f"    工作流调用失败，状态码：{response.status_code}，响应：{response.text}")
            return None
    except Exception as e:
        print(f"    工作流调用异常：{str(e)}")
        return None

将每页OCR结果按PDF原页码顺序合并，生成与原PDF同名的文本文件：

def batch_ocr_and_merge(
    image_paths: List[str],
    pdf_path: str,
    output_dir: str = "ocr_results",
    output_suffix: str = "txt"
) -> None:
    """批量OCR识别并按页码顺序合并文本"""
    total_images = len(image_paths)
    merged_text = []
    
    # 创建文本输出目录
    Path(output_dir).mkdir(parents=True, exist_ok=True)
    pdf_filename = os.path.splitext(os.path.basename(pdf_path))[0]
    output_text_path = os.path.join(output_dir, f"{pdf_filename}.{output_suffix}")
    
    # 跳过已处理文件，避免重复劳动
    if os.path.exists(output_text_path):
        print(f"  跳过已处理文件：{os.path.basename(output_text_path)}")
        return
    
    print(f"  开始OCR处理（共{total_images}张图片）...")
    for idx, image_path in enumerate(image_paths, start=1):
        print(f"n    处理第{idx}/{total_images}张：{os.path.basename(image_path)}")
        
        # 图片上传（带重试机制）
        file_id = None
        for retry in range(RETRY_TIMES):
            file_id = upload_image_to_dify(image_path)
            if file_id:
                break
            print(f"    第{retry+1}次上传重试...")
            time.sleep(RETRY_DELAY)
        
        if not file_id:
            merged_text.extend([f"===== 第{idx}页（图片上传失败） =====", "n"])
            continue
        
        # OCR识别（带重试机制）
        ocr_text = None
        for retry in range(RETRY_TIMES):
            ocr_text = run_ocr_workflow(file_id)
            if ocr_text:
                break
            print(f"    第{retry+1}次OCR重试...")
            time.sleep(RETRY_DELAY)
        
        if ocr_text:
            merged_text.extend([f"===== 第{idx}页 =====", ocr_text, "n"])
        else:
            merged_text.extend([f"===== 第{idx}页（OCR识别失败） =====", "n"])
    
    # 保存合并文本（UTF-8编码避免中文乱码）
    final_text = "n".join(merged_text)
    with open(output_text_path, "w", encoding="utf-8") as f:
        f.write(final_text)
    
    print(f"n  文本合并完成！保存路径：{os.path.abspath(output_text_path)}")
    print(f"  合并后总文本长度：{len(final_text)}字")

def main():
    # 配置参数（可根据实际需求调整）
    PDF_DIR = "./pdf_files"  # PDF文件存放目录
    ROOT_IMAGE_DIR = "pdf_images"  # 图片存储根目录
    OCR_OUTPUT_DIR = "ocr_results"  # 文本输出目录
    OUTPUT_SUFFIX = "txt"  # 输出文件后缀（支持txt/md）
    PDF_DPI = 300  # 图片分辨率（建议200-300dpi）
    
    try:
        # 步骤1：获取所有PDF文件
        pdf_files = get_all_pdf_in_dir(PDF_DIR)
        if not pdf_files:
            return
        
        # 步骤2：批量处理每个PDF
        total_pdfs = len(pdf_files)
        for pdf_idx, pdf_path in enumerate(pdf_files, start=1):
            print(f"n=== 处理第{pdf_idx}/{total_pdfs}个PDF：{os.path.basename(pdf_path)} ===")
            try:
                # 子步骤1：PDF转PNG
                image_paths = pdf_to_png(pdf_path, ROOT_IMAGE_DIR, PDF_DPI)
                # 子步骤2：OCR识别+文本合并
                batch_ocr_and_merge(image_paths, pdf_path, OCR_OUTPUT_DIR, OUTPUT_SUFFIX)
                print(f"=== 第{pdf_idx}/{total_pdfs}个PDF处理完成！ ===n" + "-"*50)
            except Exception as e:
                print(f"=== 第{pdf_idx}/{total_pdfs}个PDF处理失败：{str(e)} ===n" + "-"*50)
                continue
        
        print(f"n所有PDF文件处理完毕！")
        print(f"📁 图片文件存放于：{os.path.abspath(ROOT_IMAGE_DIR)}")
        print(f"📄 OCR文本文件存放于：{os.path.abspath(OCR_OUTPUT_DIR)}")
    except Exception as e:
        print(f"n=== 批量处理流程异常终止：{str(e)} ===")

if __name__ == "__main__":
    main()

在实际部署和运行过程中，可能会遇到各类问题，以下是常见问题的解决方案：

问题原因：系统未识别到Poppler引擎

解决方案：

• Windows：手动指定Poppler路径，在pdf_to_png函数中添加poppler_path=r"C:popplerbin"
• Mac/Linux：重新安装Poppler，确保环境变量配置正确

错误类型1：photo in input form must be a file

解决方案：确保run_ocr_workflow函数中photo字段是对象格式（而非列表），即"photo": {}而非"photo": [{}]

错误类型2：Detected file type does not match

解决方案：上传图片时MIME类型设为image/png，Dify工作流输入字段类型设为「图片」，参数type统一为image

解决方案：

• 提高PDF转图片的DPI（建议300dpi），确保图片清晰度
• 检查Dify工作流中OCR节点配置，选择支持中文识别的模型
• 合并文本时使用UTF-8编码，避免中文乱码

解决方案：

• 增加thread_count参数（如CPU为8核可设为6-7）
• 优化Dify工作流响应模式（使用blocking同步模式确保顺序）
• 对超大PDF文件可分批次处理，避免内存占用过高

• 全自动化：从PDF读取到文本生成全程无需人工干预，支持批量处理
• 顺序保障：严格保持原PDF页码顺序，文本合并时添加分页标识，便于核对
• 高兼容性：支持Windows/Mac/Linux跨平台运行，适配不同格式的图片PDF
• 健壮性强：包含重试机制、异常捕获、已处理文件跳过等功能，稳定性高
• 低代码门槛：借助Dify平台快速配置OCR工作流，无需手动开发复杂识别算法

• 支持加密PDF：集成PyPDF2库解析加密PDF文件
• 文本格式优化：添加多余空行去除、表格结构还原等后处理逻辑
• 多线程并发：使用concurrent.futures实现多PDF并行处理，提升效率
• 结果校验：集成人工审核环节，标记识别失败或模糊的页面
• 云原生部署：打包为Docker镜像，部署到云服务器或K8s集群，支持定时任务

本文提出的基于Python+Dify的批量OCR解决方案，完美解决了图片PDF的文本提取难题。通过将PDF转图片、Dify OCR识别、文本合并等环节自动化串联，大幅提升了文档处理效率，适用于企业办公、教育培训、政务处理等多个场景。

该方案兼顾了易用性和扩展性，既可以直接部署使用，也可以根据实际需求进行二次开发。随着AI识别技术的不断进步，结合Dify平台的灵活配置能力，后续还可以进一步优化识别精度和处理效率，为文档数字化转型提供更强大的支持。

以上就是基于Python+Dify实现批量OCR识别的自动化解决方案的详细内容，更多关于Python OCR识别的资料请关注风君子博客其它相关文章！

您可能感兴趣的文章:

• Python使用PaddleOCR实现PDF/图片文字识别与版面还原
• Python使用Tesseract OCR实现识别图片中的文字
• 从原理到实战详解Python实现PDF图片OCR识别全流程
• Python中图片与PDF识别文本(OCR)的全面指南
• Python中图片与pdf识别文本的方法总结(OCR)
• 使用Python实现pdf转图片再进行OCR识别