当前位置：首页 > news >正文

Qianfan-OCR代码实例：Python调用API实现批量PDF图像文字提取

news 2026/4/27 2:56:55

Qianfan-OCR代码实例：Python调用API实现批量PDF图像文字提取

1. 项目概述

Qianfan-OCR是百度千帆推出的开源文档智能多模态模型，基于4B参数的端到端架构设计。相比传统OCR技术，它不仅能识别文字，还能理解文档结构和语义信息。

核心优势：

单模型完成OCR+版面分析+文档理解
支持多语言文本识别
开源可商用（Apache 2.0协议）
基于Qwen3-4B语言模型，理解能力强

2. 环境准备

2.1 安装依赖库

pip install requests pillow pypdf2 python-multipart

2.2 服务访问配置

确保Qianfan-OCR服务已启动并运行在本地7860端口：

API_URL = "http://localhost:7860/api/ocr"

3. 基础OCR功能实现

3.1 单张图片文字提取

import requests from PIL import Image import io def ocr_single_image(image_path): with open(image_path, 'rb') as f: files = {'image': f} response = requests.post(API_URL, files=files) if response.status_code == 200: return response.json()['text'] else: raise Exception(f"OCR失败: {response.text}") # 使用示例 result = ocr_single_image('test.jpg') print("识别结果:", result)

3.2 PDF文档批量处理

from PyPDF2 import PdfReader import os def pdf_to_images(pdf_path, output_dir='temp_images'): os.makedirs(output_dir, exist_ok=True) reader = PdfReader(pdf_path) images = [] for i, page in enumerate(reader.pages): for img in page.images: img_path = f"{output_dir}/page_{i}_{img.name}" with open(img_path, "wb") as f: f.write(img.data) images.append(img_path) return images def ocr_pdf(pdf_path): image_paths = pdf_to_images(pdf_path) results = [] for img_path in image_paths: try: text = ocr_single_image(img_path) results.append(text) except Exception as e: print(f"处理{img_path}时出错:", str(e)) return "\n\n".join(results)

4. 进阶功能实现

4.1 带布局分析的OCR

def ocr_with_layout(image_path): data = { 'enable_layout': 'true' } with open(image_path, 'rb') as f: files = {'image': f} response = requests.post(API_URL, files=files, data=data) if response.status_code == 200: return response.json() else: raise Exception(f"带布局OCR失败: {response.text}") # 使用示例 layout_result = ocr_with_layout('document.jpg') print("标题:", layout_result['title']) print("正文:", layout_result['content']) print("表格:", layout_result['tables'])

4.2 定向信息提取

def extract_specific_info(image_path, prompt): data = { 'prompt': prompt } with open(image_path, 'rb') as f: files = {'image': f} response = requests.post(API_URL, files=files, data=data) if response.status_code == 200: return response.json() else: raise Exception(f"定向提取失败: {response.text}") # 使用示例：提取发票关键信息 invoice_info = extract_specific_info( 'invoice.jpg', '请提取发票中的以下信息：发票号码、开票日期、金额(大写)、金额(小写)。以JSON格式返回' ) print(invoice_info)

5. 批量处理实战案例

5.1 批量处理文件夹内PDF

import glob def batch_process_pdfs(folder_path, output_dir='results'): os.makedirs(output_dir, exist_ok=True) pdf_files = glob.glob(f"{folder_path}/*.pdf") for pdf in pdf_files: try: filename = os.path.basename(pdf).replace('.pdf', '.txt') output_path = f"{output_dir}/{filename}" text = ocr_pdf(pdf) with open(output_path, 'w', encoding='utf-8') as f: f.write(text) print(f"已处理: {pdf} -> {output_path}") except Exception as e: print(f"处理{pdf}时出错:", str(e))

5.2 结果后处理与保存

def save_structured_results(results, output_format='markdown'): if output_format == 'markdown': with open('output.md', 'w', encoding='utf-8') as f: for item in results: f.write(f"## {item['title']}\n\n") f.write(f"{item['content']}\n\n") if item['tables']: f.write("### 表格\n\n") f.write(item['tables'] + "\n\n") elif output_format == 'json': import json with open('output.json', 'w', encoding='utf-8') as f: json.dump(results, f, ensure_ascii=False, indent=2)

6. 性能优化建议

6.1 多线程处理

from concurrent.futures import ThreadPoolExecutor def parallel_ocr(image_paths, max_workers=4): with ThreadPoolExecutor(max_workers=max_workers) as executor: results = list(executor.map(ocr_single_image, image_paths)) return results

6.2 图像预处理

from PIL import Image, ImageEnhance def preprocess_image(image_path): img = Image.open(image_path) # 增强对比度 enhancer = ImageEnhance.Contrast(img) img = enhancer.enhance(1.5) # 转换为灰度图 img = img.convert('L') # 保存预处理后的图像 processed_path = f"processed_{os.path.basename(image_path)}" img.save(processed_path) return processed_path