import os import base64 import json import time from pathlib import Path from openai import OpenAI from dotenv import load_dotenv from typing import Any, Dict, List # 加载环境变量 load_dotenv() def verify_ocr_with_vlm(image_path, ocr_json_path, output_path="ocr_differences.json", api_key=None, api_base=None, model_id=None, temperature=0.1, max_tokens=4096, timeout=180): """ 使用VLM对比OCR识别结果和原图，找出差异部分 Args: image_path: 原图路径 ocr_json_path: OCR识别结果JSON文件路径 output_path: 差异分析输出文件路径 api_key: API密钥，如果为None则从环境变量获取 api_base: API基础URL，如果为None则从环境变量获取 model_id: 模型ID，如果为None则从环境变量获取 temperature: 生成温度，默认0.1 max_tokens: 最大输出token数，默认4096 timeout: 请求超时时间，默认180秒 """ # 从参数或环境变量获取API配置 api_key = api_key or os.getenv("YUSYS_MULTIMODAL_API_KEY") api_base = api_base or os.getenv("YUSYS_MULTIMODAL_API_BASE") model_id = model_id or os.getenv("YUSYS_MULTIMODAL_ID") if not api_key: raise ValueError("未找到API密钥，请通过参数传入或设置YUSYS_MULTIMODAL_API_KEY环境变量") if not api_base: raise ValueError("未找到API基础URL，请通过参数传入或设置YUSYS_MULTIMODAL_API_BASE环境变量") if not model_id: raise ValueError("未找到模型ID，请通过参数传入或设置YUSYS_MULTIMODAL_ID环境变量") # 去掉openai/前缀 model_name = model_id.replace("openai/", "") # 读取图片文件并转换为base64 try: with open(image_path, "rb") as image_file: image_data = base64.b64encode(image_file.read()).decode('utf-8') except FileNotFoundError: raise FileNotFoundError(f"找不到图片文件: {image_path}") # 读取OCR结果 try: with open(ocr_json_path, "r", encoding='utf-8') as f: ocr_results = json.load(f) except FileNotFoundError: raise FileNotFoundError(f"找不到OCR结果文件: {ocr_json_path}") # 获取图片的MIME类型 file_extension = Path(image_path).suffix.lower() mime_type_map = { '.jpg': 'image/jpeg', '.jpeg': 'image/jpeg', '.png': 'image/png', '.gif': 'image/gif', '.webp': 'image/webp' } mime_type = mime_type_map.get(file_extension, 'image/jpeg') # 构建详细的OCR结果文本，包含位置信息 ocr_text = "OCR识别结果：\n" for item in ocr_results: bbox = item.get('bbox', []) category = item.get('category', '') text = item.get('text', '') ocr_text += f"位置坐标[{bbox}] - 类别: {category} - 文本: {text}\n" # 构建分析提示词 prompt = f"""请仔细分析这张图片，并与以下OCR识别结果进行逐项详细对比： {ocr_text} 重要要求： 1. 对于表格中的每一个数据项（特别是数字、金额、项目名称），都必须逐一验证 2. 即使发现微小差异也要报告（如小数点位数、千分符、标点符号等） 3. 对于表格结构要仔细检查行列对应关系 4. 必须输出所有发现的问题，不要遗漏任何错误 请执行以下详细任务： 1. 逐行逐列验证表格中的每个数据项是否准确 2. 检查数字格式：小数点、千分符、负号等 3. 验证项目名称的完整性和准确性 4. 检查表格标题和表头信息 5. 验证行次编号的正确性 6. 识别任何遗漏的表格内容 7. 检查文本的位置坐标是否与实际位置匹配 请以JSON格式返回详细分析结果，对于表格中的每个识别项都要有明确的验证结果： {{ "table_verification": {{ "total_items_checked": 数字, "accuracy_rate": "百分比", "table_structure_correct": true/false }}, "accurate_items": [ {{ "bbox": [x1, y1, x2, y2], "original_text": "OCR识别的文本", "verified": true, "table_position": "行X列Y", "notes": "验证说明" }} ], "errors": [ {{ "bbox": [x1, y1, x2, y2], "original_text": "OCR识别的文本", "correct_text": "正确的文本", "error_type": "数字错误|格式错误|字符错误|缺失内容", "table_position": "行X列Y", "severity": "高|中|低", "confidence": "高|中|低", "detailed_description": "详细错误描述" }} ], "missing_items": [ {{ "estimated_bbox": [x1, y1, x2, y2], "missing_text": "遗漏的文本", "category": "数字|文本|标题|其他", "table_position": "行X列Y", "importance": "高|中|低", "impact": "对数据准确性的影响描述" }} ], "position_errors": [ {{ "text": "文本内容", "ocr_bbox": [x1, y1, x2, y2], "actual_bbox": [x1, y1, x2, y2], "table_position": "行X列Y", "deviation": "具体偏差描述" }} ], "format_issues": [ {{ "item": "受影响的项目", "issue_type": "千分符缺失|小数位错误|符号错误", "original_format": "OCR识别的格式", "correct_format": "正确格式", "table_position": "行X列Y" }} ] }} 特别注意： - 表格中的所有金额数据都必须精确验证 - 对于大额数字，要特别注意千分符和小数点 - 项目名称要检查是否完整，包括括号、冒号等标点符号 - 行次编号必须与对应项目正确匹配 - 任何发现的问题都要详细记录，包括位置和具体错误内容""" # 创建OpenAI客户端 client = OpenAI( api_key=api_key, base_url=api_base ) # 构建消息内容 messages: List[Dict[str, Any]] = [ { "role": "user", "content": [ { "type": "text", "text": prompt }, { "type": "image_url", "image_url": { "url": f"data:{mime_type};base64,{image_data}" } } ] } ] try: print(f"正在使用模型 {model_name} 进行OCR验证...") print(f"API地址: {api_base}") # 调用API response = client.chat.completions.create( model=model_name, messages=messages, # type: ignore temperature=temperature, max_tokens=max_tokens, timeout=timeout ) # 提取响应内容 generated_text = response.choices[0].message.content if not generated_text: raise Exception("模型没有生成文本内容") print(f"成功使用模型 {model_name} 完成OCR验证!") # 尝试解析JSON结果 verification_result: Dict[str, Any] = {} try: # 查找JSON部分 json_start = generated_text.find('{') json_end = generated_text.rfind('}') + 1 if json_start != -1 and json_end > json_start: json_content = generated_text[json_start:json_end] verification_result = json.loads(json_content) else: # 如果没有找到JSON，创建一个包含原始文本的结果 verification_result = { "raw_analysis": generated_text, "parsing_note": "无法解析为标准JSON格式，返回原始分析文本" } except json.JSONDecodeError: verification_result = { "raw_analysis": generated_text, "parsing_note": "JSON解析失败，返回原始分析文本" } # 添加元数据 verification_result["metadata"] = { "model_used": model_name, "model_id": model_id, "api_base": api_base, "temperature": temperature, "max_tokens": max_tokens, "timeout": timeout, "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), "original_image": image_path, "ocr_json": ocr_json_path } # 保存结果 with open(output_path, 'w', encoding='utf-8') as f: json.dump(verification_result, f, ensure_ascii=False, indent=2) print(f"OCR验证结果已保存到: {output_path}") # 打印详细统计 print("\n📊 验证结果统计:") if "table_verification" in verification_result: tv = verification_result["table_verification"] print(f" 📋 表格检查项目: {tv.get('total_items_checked', 'N/A')}") print(f" 📈 准确率: {tv.get('accuracy_rate', 'N/A')}") if "accurate_items" in verification_result: print(f" ✅ 准确项目数量: {len(verification_result['accurate_items'])}") if "errors" in verification_result: error_count = len(verification_result['errors']) print(f" ❌ 发现错误数量: {error_count}") if error_count > 0 and verification_result['errors']: high_errors = len([e for e in verification_result['errors'] if e.get('severity') == '高']) if high_errors > 0: print(f" 🔴 高严重程度错误: {high_errors}") if "missing_items" in verification_result: missing_count = len(verification_result['missing_items']) print(f" 📝 遗漏项目数量: {missing_count}") if missing_count > 0 and verification_result['missing_items']: high_missing = len([m for m in verification_result['missing_items'] if m.get('importance') == '高']) if high_missing > 0: print(f" 🔴 重要遗漏项目: {high_missing}") if "position_errors" in verification_result: print(f" 📍 位置错误数量: {len(verification_result['position_errors'])}") if "format_issues" in verification_result: print(f" 🎨 格式问题数量: {len(verification_result['format_issues'])}") return verification_result except Exception as e: print(f"OCR验证失败: {e}") raise Exception(f"OCR验证任务失败: {e}") def analyze_differences(verification_result_path): """ 分析OCR验证结果，生成详细的表格差异报告 Args: verification_result_path: 验证结果JSON文件路径 """ try: with open(verification_result_path, 'r', encoding='utf-8') as f: result = json.load(f) except FileNotFoundError: raise FileNotFoundError(f"找不到验证结果文件: {verification_result_path}") print("\n" + "="*60) print("OCR验证详细差异分析报告") print("="*60) if "metadata" in result: print(f"分析时间: {result['metadata']['timestamp']}") print(f"使用模型: {result['metadata']['model_used']}") print(f"原始图片: {result['metadata']['original_image']}") print(f"OCR结果: {result['metadata']['ocr_json']}") print(f"分析配置: 温度={result['metadata'].get('temperature', 'N/A')}, " f"最大tokens={result['metadata'].get('max_tokens', 'N/A')}") # 表格验证总览 if "table_verification" in result: tv = result["table_verification"] print(f"\n📊 表格验证总览:") print(f" 检查项目总数: {tv.get('total_items_checked', 'N/A')}") print(f" 准确率: {tv.get('accuracy_rate', 'N/A')}") print(f" 表格结构正确: {'✅' if tv.get('table_structure_correct') else '❌'}") print(f"\n1. ✅ 准确识别项目:") if "accurate_items" in result and result["accurate_items"]: for i, item in enumerate(result["accurate_items"], 1): pos = item.get('table_position', '未知位置') bbox = item.get('bbox', 'N/A') text = item.get('original_text', 'N/A') print(f" {i}. {pos} - 位置{bbox}") print(f" 内容: {text}") if item.get('notes'): print(f" 说明: {item['notes']}") print() else: print(" 无准确识别项目或未能解析") print(f"\n2. ❌ 识别错误 (需要重点关注):") if "errors" in result and result["errors"]: for i, error in enumerate(result["errors"], 1): pos = error.get('table_position', '未知位置') severity = error.get('severity', '未知') severity_icon = {"高": "🔴", "中": "🟡", "低": "🟢"}.get(severity, "⚪") print(f" {severity_icon} 错误 {i} - {pos}") print(f" 位置坐标: {error.get('bbox', 'N/A')}") print(f" OCR识别: 「{error.get('original_text', 'N/A')}」") print(f" 正确内容: 「{error.get('correct_text', 'N/A')}」") print(f" 错误类型: {error.get('error_type', 'N/A')}") print(f" 严重程度: {severity}") print(f" 置信度: {error.get('confidence', 'N/A')}") if error.get('detailed_description'): print(f" 详细说明: {error['detailed_description']}") print() else: print(" ✅ 未发现识别错误") print(f"\n3. 📝 遗漏项目:") if "missing_items" in result and result["missing_items"]: for i, missing in enumerate(result["missing_items"], 1): pos = missing.get('table_position', '未知位置') importance = missing.get('importance', '未知') importance_icon = {"高": "🔴", "中": "🟡", "低": "🟢"}.get(importance, "⚪") print(f" {importance_icon} 遗漏 {i} - {pos}") print(f" 预估位置: {missing.get('estimated_bbox', 'N/A')}") print(f" 遗漏内容: 「{missing.get('missing_text', 'N/A')}」") print(f" 内容类别: {missing.get('category', 'N/A')}") print(f" 重要程度: {importance}") if missing.get('impact'): print(f" 影响说明: {missing['impact']}") print() else: print(" ✅ 未发现遗漏项目") print(f"\n4. 📍 位置错误:") if "position_errors" in result and result["position_errors"]: for i, pos_error in enumerate(result["position_errors"], 1): table_pos = pos_error.get('table_position', '未知位置') print(f" 📍 位置错误 {i} - {table_pos}") print(f" 文本内容: 「{pos_error.get('text', 'N/A')}」") print(f" OCR位置: {pos_error.get('ocr_bbox', 'N/A')}") print(f" 实际位置: {pos_error.get('actual_bbox', 'N/A')}") print(f" 偏差描述: {pos_error.get('deviation', 'N/A')}") print() else: print(" ✅ 未发现位置错误") # 新增：格式问题 print(f"\n5. 🎨 格式问题:") if "format_issues" in result and result["format_issues"]: for i, format_issue in enumerate(result["format_issues"], 1): pos = format_issue.get('table_position', '未知位置') print(f" 🎨 格式问题 {i} - {pos}") print(f" 受影响项目: 「{format_issue.get('item', 'N/A')}」") print(f" 问题类型: {format_issue.get('issue_type', 'N/A')}") print(f" OCR格式: 「{format_issue.get('original_format', 'N/A')}」") print(f" 正确格式: 「{format_issue.get('correct_format', 'N/A')}」") print() else: print(" ✅ 未发现格式问题") # 生成错误摘要 total_errors = 0 if "errors" in result: total_errors += len(result["errors"]) if "missing_items" in result: total_errors += len(result["missing_items"]) if "position_errors" in result: total_errors += len(result["position_errors"]) if "format_issues" in result: total_errors += len(result["format_issues"]) print(f"\n" + "="*60) print(f"📈 验证摘要:") print(f" 总错误数量: {total_errors}") if "errors" in result: high_severity = len([e for e in result["errors"] if e.get('severity') == '高']) if high_severity > 0: print(f" 🔴 高严重程度错误: {high_severity} 个") if total_errors == 0: print(" 🎉 恭喜！未发现任何OCR错误") else: print(f" ⚠️ 建议：仔细检查所有标记为'高'严重程度的错误") # 如果有原始分析文本，也显示出来 if "raw_analysis" in result: print(f"\n6. 📄 VLM原始分析内容:") print("-" * 50) print(result["raw_analysis"]) print("-" * 50) if __name__ == "__main__": # 示例用法 image_path = "至远彩色印刷工业有限公司-2022年母公司_2.png" # 假设这是利润表图片 ocr_json_path = "demo_54fa7ad0_page_1.json" # OCR结果文件 try: # 进行OCR验证 result = verify_ocr_with_vlm(image_path, ocr_json_path) # 分析差异 analyze_differences("ocr_differences.json") except Exception as e: print(f"OCR验证失败: {e}")