|
|
@@ -0,0 +1,440 @@
|
|
|
+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}")
|
