ocr_platform/ocr_comparator/ocr_comparator.py

import os
import sys
from typing import Dict, List, Tuple
from datetime import datetime
from pathlib import Path

# 添加 ocr_platform 根目录到 Python 路径
# 使用 resolve() 确保路径是绝对路径，避免相对路径导致的 IndexError
_file_path = Path(__file__).resolve()
ocr_platform_root = _file_path.parents[1]  # ocr_comparator.py -> ocr_comparator -> ocr_platform
if str(ocr_platform_root) not in sys.path:
    sys.path.insert(0, str(ocr_platform_root))

try:
    from .content_extractor import ContentExtractor
    from .table_comparator import TableComparator
    from .paragraph_comparator import ParagraphComparator
except ImportError:
    from content_extractor import ContentExtractor
    from table_comparator import TableComparator
    from paragraph_comparator import ParagraphComparator


class OCRResultComparator:
    """OCR结果比较器主类"""
    
    def __init__(self):
        self.content_extractor = ContentExtractor()
        self.table_comparator = TableComparator()
        self.paragraph_comparator = ParagraphComparator()
        
        self.differences = []
        self.paragraph_match_threshold = 80
        self.content_similarity_threshold = 95
        self.max_paragraph_window = 6
        self.table_comparison_mode = 'standard'
        self.header_similarity_threshold = 90
    
    def compare_files(self, file1_path: str, file2_path: str) -> Dict:
        """比较两个OCR结果文件"""
        print(f"\n📖 读取文件...")
        
        # 读取文件内容
        with open(file1_path, 'r', encoding='utf-8') as f:
            content1 = f.read()
        
        with open(file2_path, 'r', encoding='utf-8') as f:
            content2 = f.read()
        
        print(f"✅ 文件读取完成")
        print(f"   文件1大小: {len(content1)} 字符")
        print(f"   文件2大小: {len(content2)} 字符")
        
        # ✅ 提取结构化内容（包含位置信息）
        print(f"\n📊 提取结构化内容...")
        structured_content1 = self.content_extractor.extract_structured_content(content1)
        structured_content2 = self.content_extractor.extract_structured_content(content2)
        
        print(f"   文件1: {len(structured_content1['tables'])}个表格, {len(structured_content1['paragraph_blocks'])}个段落块")
        print(f"   文件2: {len(structured_content2['tables'])}个表格, {len(structured_content2['paragraph_blocks'])}个段落块")
        
        # 初始化差异列表
        all_differences = []
        
        # ✅ 智能表格匹配与比较
        print(f"\n🔍 开始表格智能匹配...")
        
        tables1 = structured_content1['tables']
        tables2 = structured_content2['tables']
        
        # 记录匹配的表格对
        table_matches = []
        
        if tables1 and tables2:
            # 找到匹配的表格对
            table_matches = self.table_comparator.find_matching_tables(
                [t['data'] for t in tables1],
                [t['data'] for t in tables2]
            )
            
            if not table_matches:
                print(f"   ⚠️  未找到匹配的表格")
                all_differences.append({
                    'type': 'table_structure',
                    'position': '表格匹配',
                    'file1_value': f'{len(tables1)}个表格',
                    'file2_value': f'{len(tables2)}个表格',
                    'description': '未找到可匹配的表格',
                    'severity': 'high'
                })
            else:
                # 比较每对匹配的表格
                for idx1, idx2, similarity in table_matches:
                    print(f"\n   📋 对比匹配的表格: 表格{idx1+1} vs 表格{idx2+1}")
                    
                    if self.table_comparison_mode == 'flow_list':
                        table_diffs = self.table_comparator.compare_table_flow_list(
                            tables1[idx1]['data'], tables2[idx2]['data']
                        )
                    else:
                        table_diffs = self.table_comparator.compare_tables(
                            tables1[idx1]['data'], tables2[idx2]['data']
                        )
                    
                    # 为每个差异添加表格标识
                    for diff in table_diffs:
                        diff['table_pair'] = f'表格{idx1+1}↔表格{idx2+1}'
                        diff['table_similarity'] = similarity
                    
                    all_differences.extend(table_diffs)
                    print(f"      发现 {len(table_diffs)} 个差异")
                
                # 检查未匹配的表格
                matched_tables1 = {m[0] for m in table_matches}
                matched_tables2 = {m[1] for m in table_matches}
                
                for i in range(len(tables1)):
                    if i not in matched_tables1:
                        all_differences.append({
                            'type': 'table_unmatched',
                            'position': f'文件1表格{i+1}',
                            'file1_value': f'表格{i+1} (无匹配)',
                            'file2_value': '',
                            'description': f'文件1的表格{i+1}在文件2中无匹配表格',
                            'severity': 'medium'
                        })
                
                for j in range(len(tables2)):
                    if j not in matched_tables2:
                        all_differences.append({
                            'type': 'table_unmatched',
                            'position': f'文件2表格{j+1}',
                            'file1_value': '',
                            'file2_value': f'表格{j+1} (无匹配)',
                            'description': f'文件2的表格{j+1}在文件1中无匹配表格',
                            'severity': 'medium'
                        })
        
        elif tables1 and not tables2:
            all_differences.append({
                'type': 'table_structure',
                'position': '表格结构',
                'file1_value': f'包含{len(tables1)}个表格',
                'file2_value': '无表格',
                'description': '文件1包含表格但文件2无表格',
                'severity': 'high'
            })
        elif not tables1 and tables2:
            all_differences.append({
                'type': 'table_structure',
                'position': '表格结构',
                'file1_value': '无表格',
                'file2_value': f'包含{len(tables2)}个表格',
                'description': '文件2包含表格但文件1无表格',
                'severity': 'high'
            })
        
        # ✅ 根据表格匹配结果对齐段落块
        print(f"\n🔍 开始段落对比（基于表格位置对齐）...")
        
        paragraph_blocks1 = structured_content1['paragraph_blocks']
        paragraph_blocks2 = structured_content2['paragraph_blocks']
        
        # ✅ 构建段落块对应关系
        aligned_blocks = self._align_paragraph_blocks(
            paragraph_blocks1, paragraph_blocks2, 
            tables1, tables2, table_matches
        )
        
        for block_pair in aligned_blocks:
            block1 = block_pair['block1']
            block2 = block_pair['block2']
            position_desc = block_pair['position']
            
            paragraphs1 = block1['paragraphs'] if block1 else []
            paragraphs2 = block2['paragraphs'] if block2 else []
            
            if not paragraphs1 and not paragraphs2:
                continue
            
            print(f"   📦 {position_desc}: 文件1有{len(paragraphs1)}个段落, 文件2有{len(paragraphs2)}个段落")
            
            # 每个段落块独立对比，指针重新初始化
            block_diffs = self.paragraph_comparator.compare_paragraphs(
                paragraphs1, paragraphs2
            )
            
            # 为每个差异添加段落块标识
            for diff in block_diffs:
                diff['paragraph_block'] = position_desc
            
            all_differences.extend(block_diffs)
        
        total_paragraph_diffs = len([d for d in all_differences if d['type'] == 'paragraph'])
        print(f"✅ 段落对比完成，共发现 {total_paragraph_diffs} 个差异")
        
        print(f"\n✅ 对比完成")
        
        # 统计差异
        stats = {
            'total_differences': len(all_differences),
            'table_differences': len([d for d in all_differences if d['type'].startswith('table')]),
            'paragraph_differences': len([d for d in all_differences if d['type'] == 'paragraph']),
            'amount_differences': len([d for d in all_differences if d['type'] == 'table_amount']),
            'datetime_differences': len([d for d in all_differences if d['type'] == 'table_datetime']),
            'text_differences': len([d for d in all_differences if d['type'] == 'table_text']),
            'table_pre_header': len([d for d in all_differences if d['type'] == 'table_pre_header']),
            'table_header_mismatch': len([d for d in all_differences if d['type'] == 'table_header_mismatch']),
            'table_header_critical': len([d for d in all_differences if d['type'] == 'table_header_critical']),
            'table_header_position': len([d for d in all_differences if d['type'] == 'table_header_position']),
            'table_row_missing': len([d for d in all_differences if d['type'] == 'table_row_missing']),
            'high_severity': len([d for d in all_differences if d.get('severity') in ['critical', 'high']]),
            'medium_severity': len([d for d in all_differences if d.get('severity') == 'medium']),
            'low_severity': len([d for d in all_differences if d.get('severity') == 'low'])
        }
        
        # ✅ 构建返回结果
        result = {
            'differences': all_differences,
            'statistics': stats,
            'file1_tables': len(tables1),
            'file2_tables': len(tables2),
            'file1_paragraphs': sum(len(b['paragraphs']) for b in paragraph_blocks1),
            'file2_paragraphs': sum(len(b['paragraphs']) for b in paragraph_blocks2),
            'file1_path': file1_path,
            'file2_path': file2_path,
            'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        }
        
        print(f"\n" + "="*60)
        print(f"📊 对比结果汇总")
        print(f"="*60)
        print(f"总差异数: {result['statistics']['total_differences']}")
        print(f"  - 段落差异: {result['statistics']['paragraph_differences']}")
        print(f"  - 表格差异: {result['statistics']['table_differences']}")
        print(f"    - 金额: {result['statistics']['amount_differences']}")
        print(f"    - 日期: {result['statistics']['datetime_differences']}")
        print(f"    - 文本: {result['statistics']['text_differences']}")
        print(f"\n严重级别分布:")
        print(f"  🔴 高: {result['statistics']['high_severity']}")
        print(f"  🟡 中: {result['statistics']['medium_severity']}")
        print(f"  🟢 低: {result['statistics']['low_severity']}")
        print(f"="*60)
        
        return result
    
    def _align_paragraph_blocks(self, blocks1: List[Dict], blocks2: List[Dict],
                               tables1: List[Dict], tables2: List[Dict],
                               table_matches: List[Tuple[int, int, float]]) -> List[Dict]:
        """
        根据表格位置对齐段落块
        
        Returns:
            [
                {'block1': dict, 'block2': dict, 'position': str},
                ...
            ]
        """
        aligned = []
        
        # 如果没有表格，直接对比所有段落块
        if not tables1 and not tables2:
            max_blocks = max(len(blocks1), len(blocks2))
            for i in range(max_blocks):
                aligned.append({
                    'block1': blocks1[i] if i < len(blocks1) else None,
                    'block2': blocks2[i] if i < len(blocks2) else None,
                    'position': f'段落块{i+1}'
                })
            return aligned
        
        # 构建表格索引映射
        table_map = {idx1: idx2 for idx1, idx2, _ in table_matches}
        
        # ✅ 策略：根据表格位置划分段落块
        # 1. 第一个表格前的段落块
        # 2. 每对匹配表格之间的段落块
        # 3. 最后一个表格后的段落块
        
        # 第一个表格前的段落块
        if blocks1 or blocks2:
            first_table_idx1 = min(table_map.keys()) if table_map else len(blocks1)
            first_table_idx2 = min(table_map.values()) if table_map else len(blocks2)
            
            # 找到第一个表格前的所有段落块
            pre_blocks1 = [b for b in blocks1 if b['end_pos'] <= (tables1[first_table_idx1]['start_pos'] if first_table_idx1 < len(tables1) else float('inf'))]
            pre_blocks2 = [b for b in blocks2 if b['end_pos'] <= (tables2[first_table_idx2]['start_pos'] if first_table_idx2 < len(tables2) else float('inf'))]
            
            if pre_blocks1 or pre_blocks2:
                # 合并所有表格前的段落
                merged_block1 = self._merge_paragraph_blocks(pre_blocks1) if pre_blocks1 else None
                merged_block2 = self._merge_paragraph_blocks(pre_blocks2) if pre_blocks2 else None
                
                aligned.append({
                    'block1': merged_block1,
                    'block2': merged_block2,
                    'position': '文档开头（表格前）'
                })
        
        # 每对匹配表格之间的段落块
        sorted_matches = sorted(table_matches, key=lambda x: x[0])
        
        for i, (idx1, idx2, _) in enumerate(sorted_matches):
            # 当前表格后、下一个表格前的段落块
            table1_end = tables1[idx1]['end_pos']
            table2_end = tables2[idx2]['end_pos']
            
            # 下一个表格的开始位置
            if i + 1 < len(sorted_matches):
                next_idx1 = sorted_matches[i + 1][0]
                next_idx2 = sorted_matches[i + 1][1]
                next_table1_start = tables1[next_idx1]['start_pos']
                next_table2_start = tables2[next_idx2]['start_pos']
            else:
                next_table1_start = float('inf')
                next_table2_start = float('inf')
            
            # 找到这个范围内的段落块
            between_blocks1 = [b for b in blocks1 
                             if b['start_pos'] >= table1_end and b['end_pos'] <= next_table1_start]
            between_blocks2 = [b for b in blocks2 
                             if b['start_pos'] >= table2_end and b['end_pos'] <= next_table2_start]
            
            if between_blocks1 or between_blocks2:
                merged_block1 = self._merge_paragraph_blocks(between_blocks1) if between_blocks1 else None
                merged_block2 = self._merge_paragraph_blocks(between_blocks2) if between_blocks2 else None
                
                aligned.append({
                    'block1': merged_block1,
                    'block2': merged_block2,
                    'position': f'表格{idx1+1}↔表格{idx2+1} 之后'
                })
        
        return aligned
    
    def _merge_paragraph_blocks(self, blocks: List[Dict]) -> Dict:
        """合并多个段落块为一个"""
        if not blocks:
            return None
        
        if len(blocks) == 1:
            return blocks[0]
        
        all_paragraphs = []
        for block in blocks:
            all_paragraphs.extend(block['paragraphs'])
        
        return {
            'start_pos': blocks[0]['start_pos'],
            'end_pos': blocks[-1]['end_pos'],
            'paragraphs': all_paragraphs
        }