Qianfan-OCR实操手册批量处理脚本编写与OCR结果去重/合并/校验逻辑1. 项目概述Qianfan-OCR是百度千帆推出的开源文档智能多模态模型基于4B参数的端到端架构设计。相比传统OCR方案它集成了文字识别、版面分析和文档理解三大核心功能大幅简化了文档处理流程。核心优势一体化处理单模型完成传统OCR流水线多个步骤智能分析支持布局识别和结构化信息提取开源商用Apache 2.0协议可自由商用和二次开发多语言支持覆盖主流语言的文字识别需求2. 环境准备2.1 基础环境配置确保已安装以下组件Conda环境推荐使用torch28Python 3.11CUDA 11.7GPU加速至少16GB显存推荐24GB以上2.2 服务启动# 进入项目目录 cd /root/Qianfan-OCR # 启动服务 supervisorctl start qianfan-ocr服务启动后可通过http://localhost:7860访问Web界面。3. 批量处理脚本编写3.1 基础批量处理脚本以下Python脚本实现了图片目录的批量OCR处理import os import requests from tqdm import tqdm def batch_ocr(image_dir, output_file, endpointhttp://localhost:7860/api/ocr): 批量OCR处理脚本 :param image_dir: 图片目录路径 :param output_file: 结果输出文件 :param endpoint: API端点地址 results {} image_files [f for f in os.listdir(image_dir) if f.lower().endswith((.png, .jpg, .jpeg))] for img_file in tqdm(image_files, descProcessing images): img_path os.path.join(image_dir, img_file) try: with open(img_path, rb) as f: response requests.post(endpoint, files{image: f}) if response.status_code 200: results[img_file] response.json()[text] else: results[img_file] fError: {response.text} except Exception as e: results[img_file] fException: {str(e)} with open(output_file, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2)3.2 带布局分析的增强脚本启用Layout-as-Thought模式获取结构化结果def batch_ocr_with_layout(image_dir, output_file): 带布局分析的批量OCR处理 results {} for img_file in os.listdir(image_dir): if not img_file.lower().endswith((.png, .jpg, .jpeg)): continue img_path os.path.join(image_dir, img_file) try: with open(img_path, rb) as f: response requests.post( http://localhost:7860/api/ocr, files{image: f}, data{layout: true} # 启用布局分析 ) results[img_file] response.json() except Exception as e: results[img_file] {error: str(e)} with open(output_file, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2)4. OCR结果后处理4.1 文本去重算法针对重复出现的文本内容实现智能去重from collections import defaultdict import re def deduplicate_texts(text_list, threshold0.9): 文本内容去重 :param text_list: 待处理文本列表 :param threshold: 相似度阈值(0-1) :return: 去重后的文本列表 unique_texts [] text_groups defaultdict(list) # 基于文本指纹的粗筛 for text in text_list: if not text.strip(): continue # 生成简化指纹去除空格和标点 fingerprint re.sub(r[\s\W], , text.lower()) text_groups[fingerprint[:20]].append(text) # 精细相似度计算 for group in text_groups.values(): if len(group) 1: unique_texts.append(group[0]) continue # 使用编辑距离计算相似度 representative group[0] unique_texts.append(representative) for text in group[1:]: distance levenshtein_distance(representative, text) similarity 1 - (distance / max(len(representative), len(text))) if similarity threshold: unique_texts.append(text) return unique_texts def levenshtein_distance(s1, s2): 计算两个字符串的编辑距离 if len(s1) len(s2): return levenshtein_distance(s2, s1) if len(s2) 0: return len(s1) previous_row range(len(s2) 1) for i, c1 in enumerate(s1): current_row [i 1] for j, c2 in enumerate(s2): insertions previous_row[j 1] 1 deletions current_row[j] 1 substitutions previous_row[j] (c1 ! c2) current_row.append(min(insertions, deletions, substitutions)) previous_row current_row return previous_row[-1]4.2 多页文档合并处理扫描版多页文档的合并逻辑def merge_document_pages(page_results, merge_strategysequential): 合并多页OCR结果 :param page_results: 按页码排序的OCR结果列表 :param merge_strategy: 合并策略(sequential/smart) :return: 合并后的完整文档 merged_text if merge_strategy sequential: # 简单顺序合并 for page in sorted(page_results.keys()): merged_text f\n\n--- Page {page} ---\n{page_results[page]} elif merge_strategy smart: # 智能合并识别页眉页脚 header_footer_threshold 0.7 header_candidates [] footer_candidates [] # 分析页眉页脚模式 sample_pages list(page_results.values())[:3] for text in sample_pages: lines text.split(\n) if len(lines) 1: header_candidates.append(lines[0]) footer_candidates.append(lines[-1]) # 确定公共页眉页脚 common_header find_common_text(header_candidates, thresholdheader_footer_threshold) common_footer find_common_text(footer_candidates, thresholdheader_footer_threshold) # 应用合并 for page_num, text in sorted(page_results.items()): lines text.split(\n) if len(lines) 1: # 移除页眉页脚 if lines[0] common_header: lines lines[1:] if lines[-1] common_footer: lines lines[:-1] merged_text \n.join(lines) \n\n return merged_text.strip()4.3 结果校验机制实现OCR结果的自动校验def validate_ocr_result(text, validation_rulesNone): OCR结果校验 :param text: 待校验文本 :param validation_rules: 校验规则字典 :return: (是否有效, 错误信息) if not validation_rules: validation_rules { min_length: 3, # 最小文本长度 max_repeat: 5, # 最大连续重复字符数 invalid_chars: [, ], # 非法字符 expected_patterns: None # 预期正则模式 } # 检查最小长度 if len(text.strip()) validation_rules[min_length]: return False, Text too short # 检查重复字符 if any(len(list(g)) validation_rules[max_repeat] for k, g in groupby(text)): return False, Excessive character repetition # 检查非法字符 if any(invalid_char in text for invalid_char in validation_rules[invalid_chars]): return False, Contains invalid characters # 检查正则模式 if validation_rules[expected_patterns]: if not any(re.search(pattern, text) for pattern in validation_rules[expected_patterns]): return False, Does not match expected patterns return True, 5. 实战案例发票处理系统5.1 发票信息提取流程def process_invoice_images(image_dir): 完整发票处理流程 # 1. 批量OCR处理 raw_results batch_ocr_with_layout(image_dir, raw_results.json) # 2. 关键信息提取 extracted_data [] for img_file, ocr_result in raw_results.items(): if error in ocr_result: continue # 使用提示词提取关键字段 prompt 请从发票中提取以下信息 - 发票号码 - 开票日期 - 销售方名称 - 购买方名称 - 金额合计(大写) - 金额合计(小写) 以JSON格式返回 response requests.post( http://localhost:7860/api/ocr, json{ text: ocr_result[text], prompt: prompt } ) if response.status_code 200: extracted_data.append({ image: img_file, data: response.json() }) # 3. 数据校验 valid_data [] for item in extracted_data: is_valid, error validate_invoice_data(item[data]) if is_valid: valid_data.append(item) else: print(fInvalid data in {item[image]}: {error}) return valid_data def validate_invoice_data(data): 发票数据校验 required_fields [ 发票号码, 开票日期, 销售方名称, 金额合计(小写) ] for field in required_fields: if field not in data or not data[field].strip(): return False, fMissing {field} # 检查金额格式 if not re.match(r^\d(\.\d{1,2})?$, data[金额合计(小写)]): return False, Invalid amount format return True, 5.2 处理结果可视化import pandas as pd def generate_invoice_report(data, output_file): 生成发票处理报告 df pd.DataFrame([item[data] for item in data]) df[源文件] [item[image] for item in data] # 保存Excel报告 with pd.ExcelWriter(output_file) as writer: df.to_excel(writer, sheet_name发票汇总, indexFalse) # 添加统计信息 stats pd.DataFrame({ 统计项: [发票总数, 总金额, 平均金额], 值: [len(df), df[金额合计(小写)].astype(float).sum(), df[金额合计(小写)].astype(float).mean()] }) stats.to_excel(writer, sheet_name统计信息, indexFalse) print(f报告已生成: {output_file})6. 总结与最佳实践6.1 性能优化建议批量处理优化使用多线程/协程提高处理速度实现失败重试机制设置合理的请求间隔建议200-500ms质量提升技巧对模糊图片进行预处理锐化、二值化组合使用通用OCR和带提示OCR对关键字段实施多重校验资源管理监控GPU显存使用情况定期清理缓存文件对大型任务实施分批处理6.2 扩展应用场景合同处理关键条款提取签名位置识别版本差异比对报表分析表格数据提取趋势图表识别数据交叉验证档案数字化多页文档合并元数据提取智能分类归档获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。