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

PP-DocLayoutV3代码实例：批量处理图像目录并生成结构化JSON报告

news 2026/8/21 17:23:30

PP-DocLayoutV3代码实例：批量处理图像目录并生成结构化JSON报告

1. 引言：文档布局分析的实用价值

在日常工作中，我们经常需要处理大量的文档图像——可能是扫描的合同、报告、论文或者各种表格文件。手动从这些图像中提取结构化信息既耗时又容易出错。PP-DocLayoutV3正是为了解决这个问题而生的强大工具。

这个模型能够智能识别文档中的26种不同布局元素，包括标题、段落、表格、图片、公式等，并生成结构化的JSON报告。无论你是需要批量处理企业文档、学术论文还是历史档案，这个工具都能大幅提升工作效率。

本文将带你一步步实现一个完整的批量处理解决方案，让你能够轻松处理整个文件夹的文档图像，并获得详细的结构化分析结果。

2. 环境准备与快速部署

2.1 基础环境要求

在开始之前，确保你的系统满足以下要求：

Python 3.7+
至少4GB内存（处理大量图像建议8GB以上）
支持CUDA的GPU（可选，但能显著加速处理）

2.2 一键安装依赖

创建并安装所需的Python包：

# 创建requirements.txt文件 echo "gradio>=6.0.0 paddleocr>=3.3.0 paddlepaddle>=3.0.0 opencv-python>=4.8.0 pillow>=12.0.0 numpy>=1.24.0" > requirements.txt # 安装依赖 pip install -r requirements.txt

2.3 模型文件准备

确保模型文件位于正确路径，PP-DocLayoutV3会自动搜索以下位置：

/root/ai-models/PaddlePaddle/PP-DocLayoutV3/（推荐位置）
~/.cache/modelscope/hub/PaddlePaddle/PP-DocLayoutV3/
当前目录下的./inference.pdmodel

3. 批量处理核心代码实现

3.1 创建批量处理脚本

新建一个Python文件batch_process.py，包含以下完整代码：

import os import json import cv2 import numpy as np from PIL import Image import gradio as gr from paddleocr import PPStructure import time from datetime import datetime class BatchDocLayoutAnalyzer: def __init__(self, use_gpu=False): """初始化文档布局分析器""" self.use_gpu = use_gpu self.analyzer = PPStructure(use_gpu=use_gpu, layout=True) def process_single_image(self, image_path): """处理单张图像并返回结构化结果""" try: # 读取图像 img = cv2.imread(image_path) if img is None: raise ValueError(f"无法读取图像: {image_path}") # 分析文档布局 result = self.analyzer.analyze(img) # 提取关键信息 structured_data = self._extract_structured_data(result, image_path) return structured_data except Exception as e: print(f"处理图像 {image_path} 时出错: {str(e)}") return None def _extract_structured_data(self, result, image_path): """从分析结果中提取结构化数据""" elements = [] for item in result: element = { "type": item['type'], "bbox": item['bbox'].tolist() if hasattr(item['bbox'], 'tolist') else item['bbox'], "confidence": float(item['confidence']) if 'confidence' in item else 1.0, "text": item.get('text', ''), "image_width": item.get('img_size', [0, 0])[0], "image_height": item.get('img_size', [0, 0])[1] } elements.append(element) # 按位置排序（从上到下，从左到右） elements.sort(key=lambda x: (x['bbox'][1], x['bbox'][0])) return { "filename": os.path.basename(image_path), "processing_time": datetime.now().isoformat(), "total_elements": len(elements), "elements": elements, "element_summary": self._create_element_summary(elements) } def _create_element_summary(self, elements): """创建元素类型统计摘要""" summary = {} for element in elements: elem_type = element['type'] summary[elem_type] = summary.get(elem_type, 0) + 1 return summary def process_directory(self, input_dir, output_dir): """批量处理目录中的所有图像""" # 创建输出目录 os.makedirs(output_dir, exist_ok=True) # 支持的图像格式 supported_formats = ['.jpg', '.jpeg', '.png', '.bmp', '.tiff', '.tif'] # 查找所有图像文件 image_files = [] for file in os.listdir(input_dir): if any(file.lower().endswith(fmt) for fmt in supported_formats): image_files.append(os.path.join(input_dir, file)) print(f"找到 {len(image_files)} 个图像文件待处理") # 处理每个图像 all_results = [] processed_count = 0 for image_path in image_files: print(f"正在处理: {os.path.basename(image_path)}") start_time = time.time() result = self.process_single_image(image_path) if result: # 保存单个文件的JSON结果 output_filename = os.path.splitext(os.path.basename(image_path))[0] + '.json' output_path = os.path.join(output_dir, output_filename) with open(output_path, 'w', encoding='utf-8') as f: json.dump(result, f, ensure_ascii=False, indent=2) all_results.append(result) processed_count += 1 processing_time = time.time() - start_time print(f"完成处理: {os.path.basename(image_path)} (耗时: {processing_time:.2f}秒)") # 生成汇总报告 summary_report = self._generate_summary_report(all_results, input_dir) summary_path = os.path.join(output_dir, 'processing_summary.json') with open(summary_path, 'w', encoding='utf-8') as f: json.dump(summary_report, f, ensure_ascii=False, indent=2) print(f"处理完成! 成功处理 {processed_count}/{len(image_files)} 个文件") return summary_report def _generate_summary_report(self, all_results, input_dir): """生成处理汇总报告""" total_elements = sum(result['total_elements'] for result in all_results) # 统计所有元素类型 type_statistics = {} for result in all_results: for elem_type, count in result['element_summary'].items(): type_statistics[elem_type] = type_statistics.get(elem_type, 0) + count return { "processing_date": datetime.now().isoformat(), "input_directory": input_dir, "total_files_processed": len(all_results), "total_elements_detected": total_elements, "element_type_statistics": type_statistics, "file_details": all_results } # 使用示例 if __name__ == "__main__": # 初始化分析器（根据实际情况设置use_gpu） analyzer = BatchDocLayoutAnalyzer(use_gpu=False) # 设置输入输出目录 input_directory = "./input_images" # 你的图像目录 output_directory = "./output_results" # 结果输出目录 # 开始批量处理 summary = analyzer.process_directory(input_directory, output_directory) print("批量处理完成!") print(f"总共检测到 {summary['total_elements_detected']} 个布局元素") print("元素类型统计:", json.dumps(summary['element_type_statistics'], indent=2))

3.2 配置文件设置

创建配置文件config.json来管理处理参数：

{ "input_directory": "./input_images", "output_directory": "./output_results", "supported_formats": [".jpg", ".jpeg", ".png", ".bmp", ".tiff", ".tif"], "use_gpu": false, "max_image_size": 2000, "confidence_threshold": 0.5, "output_format": "json", "generate_visualization": true, "save_individual_results": true }

3.3 可视化结果生成

添加可视化功能，生成带标注的图像：

def visualize_results(image_path, result, output_path): """生成可视化结果图像""" img = cv2.imread(image_path) if img is None: return False # 为不同元素类型设置不同颜色 colors = { 'text': (0, 255, 0), # 绿色 - 文本 'title': (255, 0, 0), # 蓝色 - 标题 'table': (0, 0, 255), # 红色 - 表格 'figure': (255, 255, 0), # 青色 - 图片 'formula': (255, 0, 255), # 紫色 - 公式 'default': (0, 255, 255) # 黄色 - 其他 } for element in result['elements']: elem_type = element['type'] color = colors.get(elem_type, colors['default']) bbox = element['bbox'] # 绘制边界框 cv2.rectangle(img, (bbox[0], bbox[1]), (bbox[2], bbox[3]), color, 2) # 添加类型标签 label = f"{elem_type}: {element.get('confidence', 0):.2f}" cv2.putText(img, label, (bbox[0], bbox[1]-10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, 1) # 保存结果图像 cv2.imwrite(output_path, img) return True

4. 实际应用案例演示

4.1 学术论文处理案例

假设我们有一批学术论文的扫描件，需要提取结构化信息：

# 专门处理学术论文的配置 academic_config = { "focus_categories": ["abstract", "algorithm", "display_formula", "figure_title", "paragraph_title", "reference", "table", "text"], "expected_structure": ["doc_title", "abstract", "section_title", "text", "figure", "table", "reference"], "output_template": { "paper_title": "", "authors": "", "abstract": "", "sections": [], "references": [], "figures_count": 0, "tables_count": 0 } } def extract_academic_paper_structure(result): """从结果中提取学术论文结构""" paper_data = { "title": "", "authors": "", "abstract": "", "sections": [], "references": [], "figures": [], "tables": [] } current_section = None for element in result['elements']: if element['type'] == 'doc_title': paper_data['title'] = element.get('text', '') elif element['type'] == 'abstract': paper_data['abstract'] = element.get('text', '') elif element['type'] == 'paragraph_title': current_section = { "title": element.get('text', ''), "content": [] } paper_data['sections'].append(current_section) elif element['type'] == 'text' and current_section: current_section['content'].append(element.get('text', '')) elif element['type'] == 'reference': paper_data['references'].append(element.get('text', '')) elif element['type'] == 'figure_title': paper_data['figures'].append({ "title": element.get('text', ''), "position": element['bbox'] }) return paper_data

4.2 商业文档处理案例

对于商业合同和报告的处理：

def extract_business_document_structure(result): """提取商业文档结构""" doc_data = { "document_type": "unknown", "parties": [], "dates": [], "key_terms": [], "signature_blocks": [], "tables": [] } # 根据内容特征判断文档类型 text_content = " ".join([elem.get('text', '') for elem in result['elements'] if elem['type'] == 'text']) if any(term in text_content.lower() for term in ['合同', '协议', 'contract', 'agreement']): doc_data['document_type'] = 'contract' elif any(term in text_content.lower() for term in ['报告', 'report', '分析']): doc_data['document_type'] = 'report' elif any(term in text_content.lower() for term in ['发票', 'invoice', '账单']): doc_data['document_type'] = 'invoice' # 提取关键信息 for element in result['elements']: text = element.get('text', '').lower() if any(term in text for term in ['甲方', '乙方', 'party a', 'party b']): doc_data['parties'].append({ "text": element.get('text', ''), "position": element['bbox'] }) elif any(term in text for term in ['日期', 'date', '生效日']): doc_data['dates'].append({ "text": element.get('text', ''), "position": element['bbox'] }) elif element['type'] == 'table': doc_data['tables'].append({ "content": element.get('text', ''), "position": element['bbox'] }) return doc_data

5. 高级功能与优化建议

5.1 并行处理加速

对于大量图像，可以使用并行处理：

from concurrent.futures import ThreadPoolExecutor, as_completed def parallel_process_directory(analyzer, input_dir, output_dir, max_workers=4): """并行处理目录中的图像""" supported_formats = ['.jpg', '.jpeg', '.png', '.bmp', '.tiff', '.tif'] image_files = [os.path.join(input_dir, f) for f in os.listdir(input_dir) if any(f.lower().endswith(fmt) for fmt in supported_formats)] os.makedirs(output_dir, exist_ok=True) with ThreadPoolExecutor(max_workers=max_workers) as executor: # 提交所有任务 future_to_file = { executor.submit(analyzer.process_single_image, file_path): file_path for file_path in image_files } results = [] for future in as_completed(future_to_file): file_path = future_to_file[future] try: result = future.result() if result: # 保存结果 output_filename = os.path.splitext(os.path.basename(file_path))[0] + '.json' output_path = os.path.join(output_dir, output_filename) with open(output_path, 'w', encoding='utf-8') as f: json.dump(result, f, ensure_ascii=False, indent=2) results.append(result) print(f"完成处理: {os.path.basename(file_path)}") except Exception as e: print(f"处理 {file_path} 时出错: {str(e)}") return results

5.2 结果验证与质量检查

添加结果质量检查功能：

def validate_results(result, min_confidence=0.5): """验证处理结果的质量""" validation = { "is_valid": True, "issues": [], "element_count": len(result['elements']), "low_confidence_elements": 0 } # 检查元素数量 if validation['element_count'] == 0: validation['is_valid'] = False validation['issues'].append("未检测到任何布局元素") # 检查置信度 for element in result['elements']: if element.get('confidence', 1.0) < min_confidence: validation['low_confidence_elements'] += 1 if validation['low_confidence_elements'] > 0: validation['issues'].append( f"发现 {validation['low_confidence_elements']} 个低置信度元素" ) # 检查必要的元素类型（根据文档类型） required_types = ['text'] # 至少应该有文本内容 present_types = set(elem['type'] for elem in result['elements']) missing_types = [t for t in required_types if t not in present_types] if missing_types: validation['is_valid'] = False validation['issues'].append(f"缺少必要的元素类型: {missing_types}") return validation