ocr_verify/comparator/table_comparator.py

import re
from typing import Dict, List
# ✅ 兼容相对导入和绝对导入
try:
    from .data_type_detector import DataTypeDetector
    from .similarity_calculator import SimilarityCalculator
    from .text_processor import TextProcessor
except ImportError:
    from data_type_detector import DataTypeDetector
    from similarity_calculator import SimilarityCalculator
    from text_processor import TextProcessor

class TableComparator:
    """表格数据比较"""
    
    def __init__(self):
        self.detector = DataTypeDetector()
        self.calculator = SimilarityCalculator()
        self.text_processor = TextProcessor()
        self.header_similarity_threshold = 90
        self.content_similarity_threshold = 95
        self.max_paragraph_window = 6
    
    def normalize_header_text(self, text: str) -> str:
        """标准化表头文本"""
        text = re.sub(r'[（(].*?[)）]', '', text)
        text = re.sub(r'\s+', '', text)
        text = re.sub(r'[^\w\u4e00-\u9fff]', '', text)
        return text.lower().strip()
    
    def compare_table_headers(self, headers1: List[str], headers2: List[str]) -> Dict:
        """比较表格表头"""
        result = {
            'match': True,
            'differences': [],
            'column_mapping': {},
            'similarity_scores': []
        }
        
        if len(headers1) != len(headers2):
            result['match'] = False
            result['differences'].append({
                'type': 'table_header_critical',
                'description': f'表头列数不一致: {len(headers1)} vs {len(headers2)}',
                'severity': 'critical'
            })
            return result
        
        for i, (h1, h2) in enumerate(zip(headers1, headers2)):
            norm_h1 = self.normalize_header_text(h1)
            norm_h2 = self.normalize_header_text(h2)
            
            similarity = self.calculator.calculate_text_similarity(norm_h1, norm_h2)
            result['similarity_scores'].append({
                'column_index': i,
                'header1': h1,
                'header2': h2,
                'similarity': similarity
            })
            
            if similarity < self.header_similarity_threshold:
                result['match'] = False
                result['differences'].append({
                    'type': 'table_header_mismatch',
                    'column_index': i,
                    'header1': h1,
                    'header2': h2,
                    'similarity': similarity,
                    'description': f'第{i+1}列表头不匹配: "{h1}" vs "{h2}" (相似度: {similarity:.1f}%)',
                    'severity': 'medium' if similarity < 50 else 'high'
                })
            else:
                result['column_mapping'][i] = i
        
        return result
    
    def detect_table_header_row(self, table: List[List[str]]) -> int:
        """智能检测表格的表头行索引"""
        header_keywords = [
            '序号', '编号', '时间', '日期', '名称', '类型', '金额', '数量', '单价',
            '备注', '说明', '状态', '类别', '方式', '账号', '单号', '订单',
            '交易单号', '交易时间', '交易类型', '收/支', '支出', '收入', 
            '交易方式', '交易对方', '商户单号', '付款方式', '收款方',
            'no', 'id', 'time', 'date', 'name', 'type', 'amount', 'status'
        ]
        
        for row_idx, row in enumerate(table):
            if not row:
                continue
            
            keyword_count = 0
            for cell in row:
                cell_lower = cell.lower().strip()
                for keyword in header_keywords:
                    if keyword in cell_lower:
                        keyword_count += 1
                        break
            
            if keyword_count >= len(row) * 0.4 and keyword_count >= 2:
                if row_idx + 1 < len(table):
                    next_row = table[row_idx + 1]
                    if self._is_data_row(next_row):
                        print(f"   📍 检测到表头在第 {row_idx + 1} 行")
                        return row_idx
        
        print(f"   ⚠️  未检测到明确表头，默认使用第1行")
        return 0
    
    def _is_data_row(self, row: List[str]) -> bool:
        """判断是否为数据行"""
        data_pattern_count = 0
        
        for cell in row:
            if not cell:
                continue
            
            if re.search(r'\d', cell):
                data_pattern_count += 1
            
            if re.search(r'\d{4}[-/年]\d{1,2}[-/月]\d{1,2}', cell):
                data_pattern_count += 1
        
        return data_pattern_count >= len(row) * 0.5
    
    def compare_cell_value(self, value1: str, value2: str, column_type: str, 
                          column_name: str = '') -> Dict:
        """比较单元格值"""
        result = {
            'match': True,
            'difference': None
        }
        
        v1 = self.text_processor.normalize_text(value1)
        v2 = self.text_processor.normalize_text(value2)
        
        if v1 == v2:
            return result
        
        if column_type == 'text_number':
            norm_v1 = self.detector.normalize_text_number(v1)
            norm_v2 = self.detector.normalize_text_number(v2)
            
            if norm_v1 == norm_v2:
                result['match'] = False
                result['difference'] = {
                    'type': 'table_text',
                    'value1': value1,
                    'value2': value2,
                    'description': f'文本型数字格式差异: "{value1}" vs "{value2}" (内容相同，空格不同)',
                    'severity': 'low'
                }
            else:
                result['match'] = False
                result['difference'] = {
                    'type': 'table_text',
                    'value1': value1,
                    'value2': value2,
                    'description': f'文本型数字不一致: {value1} vs {value2}',
                    'severity': 'high'
                }
            return result
        
        if column_type == 'numeric':
            if self.detector.is_numeric(v1) and self.detector.is_numeric(v2):
                num1 = self.detector.parse_number(v1)
                num2 = self.detector.parse_number(v2)
                if abs(num1 - num2) > 0.01:
                    result['match'] = False
                    result['difference'] = {
                        'type': 'table_amount',
                        'value1': value1,
                        'value2': value2,
                        'diff_amount': abs(num1 - num2),
                        'description': f'金额不一致: {value1} vs {value2}'
                    }
            else:
                result['match'] = False
                result['difference'] = {
                    'type': 'table_text',
                    'value1': value1,
                    'value2': value2,
                    'description': f'长数字字符串不一致: {value1} vs {value2}'
                }
        elif column_type == 'datetime':
            datetime1 = self.detector.extract_datetime(v1)
            datetime2 = self.detector.extract_datetime(v2)
            
            if datetime1 != datetime2:
                result['match'] = False
                result['difference'] = {
                    'type': 'table_datetime',
                    'value1': value1,
                    'value2': value2,
                    'description': f'日期时间不一致: {value1} vs {value2}'
                }
        else:
            similarity = self.calculator.calculate_text_similarity(v1, v2)
            if similarity < self.content_similarity_threshold:
                result['match'] = False
                result['difference'] = {
                    'type': 'table_text',
                    'value1': value1,
                    'value2': value2,
                    'similarity': similarity,
                    'description': f'文本不一致: {value1} vs {value2} (相似度: {similarity:.1f}%)'
                }
        
        return result
    
    def compare_tables(self, table1: List[List[str]], table2: List[List[str]]) -> List[Dict]:
        """标准表格比较"""
        differences = []
        max_rows = max(len(table1), len(table2))
        
        for i in range(max_rows):
            row1 = table1[i] if i < len(table1) else []
            row2 = table2[i] if i < len(table2) else []
            
            max_cols = max(len(row1), len(row2))
            
            for j in range(max_cols):
                cell1 = row1[j] if j < len(row1) else ""
                cell2 = row2[j] if j < len(row2) else ""
                
                if "[图片内容-忽略]" in cell1 or "[图片内容-忽略]" in cell2:
                    continue
                
                if cell1 != cell2:
                    if self.detector.is_numeric(cell1) and self.detector.is_numeric(cell2):
                        num1 = self.detector.parse_number(cell1)
                        num2 = self.detector.parse_number(cell2)
                        if abs(num1 - num2) > 0.001:
                            differences.append({
                                'type': 'table_amount',
                                'position': f'行{i+1}列{j+1}',
                                'file1_value': cell1,
                                'file2_value': cell2,
                                'description': f'金额不一致: {cell1} vs {cell2}',
                                'row_index': i,
                                'col_index': j
                            })
                    else:
                        differences.append({
                            'type': 'table_text',
                            'position': f'行{i+1}列{j+1}',
                            'file1_value': cell1,
                            'file2_value': cell2,
                            'description': f'文本不一致: {cell1} vs {cell2}',
                            'row_index': i,
                            'col_index': j
                        })
        
        return differences
    
    def compare_table_flow_list(self, table1: List[List[str]], table2: List[List[str]]) -> List[Dict]:
        """流水列表表格比较算法"""
        differences = []
        
        if not table1 or not table2:
            return [{
                'type': 'table_empty',
                'description': '表格为空',
                'severity': 'critical'
            }]
        
        print(f"\n📋 开始流水表格对比...")
        
        # 检测表头位置
        header_row_idx1 = self.detect_table_header_row(table1)
        header_row_idx2 = self.detect_table_header_row(table2)
        
        if header_row_idx1 != header_row_idx2:
            differences.append({
                'type': 'table_header_position',
                'position': '表头位置',
                'file1_value': f'第{header_row_idx1 + 1}行',
                'file2_value': f'第{header_row_idx2 + 1}行',
                'description': f'表头位置不一致: 文件1在第{header_row_idx1 + 1}行，文件2在第{header_row_idx2 + 1}行',
                'severity': 'high'
            })
        
        # 比对表头前的内容
        if header_row_idx1 > 0 or header_row_idx2 > 0:
            print(f"\n📝 对比表头前的内容...")
            pre_header_table1 = table1[:header_row_idx1] if header_row_idx1 > 0 else []
            pre_header_table2 = table2[:header_row_idx2] if header_row_idx2 > 0 else []
            
            if pre_header_table1 or pre_header_table2:
                pre_header_diffs = self.compare_tables(pre_header_table1, pre_header_table2)
                for diff in pre_header_diffs:
                    diff['type'] = 'table_pre_header'
                    diff['position'] = f"表头前{diff['position']}"
                    diff['severity'] = 'medium'
                differences.extend(pre_header_diffs)
        
        # 比较表头
        headers1 = table1[header_row_idx1]
        headers2 = table2[header_row_idx2]
        
        print(f"\n📋 对比表头...")
        header_result = self.compare_table_headers(headers1, headers2)
        
        if not header_result['match']:
            print(f"\n⚠️  表头文字存在差异")
            for diff in header_result['differences']:
                differences.append({
                    'type': diff.get('type', 'table_header_mismatch'),
                    'position': '表头',
                    'file1_value': diff.get('header1', ''),
                    'file2_value': diff.get('header2', ''),
                    'description': diff['description'],
                    'severity': diff.get('severity', 'high'),
                })
                if diff.get('severity') == 'critical':
                    return differences
        
        # 检测列类型并比较数据行
        column_types1 = self._detect_column_types(table1, header_row_idx1, headers1)
        column_types2 = self._detect_column_types(table2, header_row_idx2, headers2)
        
        # 处理列类型不匹配
        mismatched_columns = self._check_column_type_mismatch(
            column_types1, column_types2, headers1, headers2, differences
        )
        
        # 合并列类型
        column_types = self._merge_column_types(column_types1, column_types2, mismatched_columns)
        
        # 逐行比较数据
        data_diffs = self._compare_data_rows(
            table1, table2, header_row_idx1, header_row_idx2,
            headers1, column_types, mismatched_columns, header_result['match']
        )
        differences.extend(data_diffs)
        
        print(f"\n✅ 流水表格对比完成，发现 {len(differences)} 个差异")
        return differences
    
    def _detect_column_types(self, table: List[List[str]], header_row_idx: int, 
                            headers: List[str]) -> List[str]:
        """检测列类型"""
        column_types = []
        for col_idx in range(len(headers)):
            col_values = [
                row[col_idx] 
                for row in table[header_row_idx + 1:] 
                if col_idx < len(row)
            ]
            col_type = self.detector.detect_column_type(col_values)
            column_types.append(col_type)
        return column_types
    
    def _check_column_type_mismatch(self, column_types1: List[str], column_types2: List[str],
                                   headers1: List[str], headers2: List[str],
                                   differences: List[Dict]) -> List[int]:
        """检查列类型不匹配"""
        mismatched_columns = []
        for col_idx in range(min(len(column_types1), len(column_types2))):
            if column_types1[col_idx] != column_types2[col_idx]:
                mismatched_columns.append(col_idx)
                differences.append({
                    'type': 'table_column_type_mismatch',
                    'position': f'第{col_idx + 1}列',
                    'file1_value': f'{headers1[col_idx]} ({column_types1[col_idx]})',
                    'file2_value': f'{headers2[col_idx]} ({column_types2[col_idx]})',
                    'description': f'列类型不一致: {column_types1[col_idx]} vs {column_types2[col_idx]}',
                    'severity': 'high',
                    'column_index': col_idx
                })
        
        total_columns = min(len(column_types1), len(column_types2))
        mismatch_ratio = len(mismatched_columns) / total_columns if total_columns > 0 else 0
        
        if mismatch_ratio > 0.5:
            differences.append({
                'type': 'table_header_critical',
                'position': '表格列类型',
                'file1_value': f'{len(mismatched_columns)}列类型不一致',
                'file2_value': f'共{total_columns}列',
                'description': f'列类型差异过大: {len(mismatched_columns)}/{total_columns}列不匹配 ({mismatch_ratio:.1%})',
                'severity': 'critical'
            })
        
        return mismatched_columns
    
    def _merge_column_types(self, column_types1: List[str], column_types2: List[str],
                           mismatched_columns: List[int]) -> List[str]:
        """合并列类型"""
        column_types = []
        for col_idx in range(max(len(column_types1), len(column_types2))):
            if col_idx >= len(column_types1):
                column_types.append(column_types2[col_idx])
            elif col_idx >= len(column_types2):
                column_types.append(column_types1[col_idx])
            elif col_idx in mismatched_columns:
                type1 = column_types1[col_idx]
                type2 = column_types2[col_idx]
                
                if type1 == 'text' or type2 == 'text':
                    column_types.append('text')
                elif type1 == 'text_number' or type2 == 'text_number':
                    column_types.append('text_number')
                else:
                    column_types.append(type1)
            else:
                column_types.append(column_types1[col_idx])
        
        return column_types
    
    def _compare_data_rows(self, table1: List[List[str]], table2: List[List[str]],
                          header_row_idx1: int, header_row_idx2: int,
                          headers1: List[str], column_types: List[str],
                          mismatched_columns: List[int], header_match: bool) -> List[Dict]:
        """逐行比较数据"""
        differences = []
        data_rows1 = table1[header_row_idx1 + 1:]
        data_rows2 = table2[header_row_idx2 + 1:]
        max_rows = max(len(data_rows1), len(data_rows2))
        
        for row_idx in range(max_rows):
            row1 = data_rows1[row_idx] if row_idx < len(data_rows1) else []
            row2 = data_rows2[row_idx] if row_idx < len(data_rows2) else []
            actual_row_num = header_row_idx1 + row_idx + 2
            
            if not row1:
                differences.append({
                    'type': 'table_row_missing',
                    'position': f'第{actual_row_num}行',
                    'file1_value': '',
                    'file2_value': ', '.join(row2),
                    'description': f'文件1缺少第{actual_row_num}行',
                    'severity': 'high',
                    'row_index': actual_row_num
                })
                continue
            
            if not row2:
                differences.append({
                    'type': 'table_row_missing',
                    'position': f'第{actual_row_num}行',
                    'file1_value': ', '.join(row1),
                    'file2_value': '',
                    'description': f'文件2缺少第{actual_row_num}行',
                    'severity': 'high',
                    'row_index': actual_row_num
                })
                continue
            
            # 逐列比较
            max_cols = max(len(row1), len(row2))
            for col_idx in range(max_cols):
                cell1 = row1[col_idx] if col_idx < len(row1) else ''
                cell2 = row2[col_idx] if col_idx < len(row2) else ''
                
                if "[图片内容-忽略]" in cell1 or "[图片内容-忽略]" in cell2:
                    continue
                
                column_type = column_types[col_idx] if col_idx < len(column_types) else 'text'
                column_name = headers1[col_idx] if col_idx < len(headers1) else f'列{col_idx + 1}'
                
                compare_result = self.compare_cell_value(cell1, cell2, column_type, column_name)
                
                if not compare_result['match']:
                    diff_info = compare_result['difference']
                    type_mismatch_note = ""
                    if col_idx in mismatched_columns:
                        type_mismatch_note = " [列类型冲突]"
                    
                    differences.append({
                        'type': diff_info['type'],
                        'position': f'第{actual_row_num}行第{col_idx + 1}列',
                        'file1_value': diff_info['value1'],
                        'file2_value': diff_info['value2'],
                        'description': diff_info['description'] + type_mismatch_note,
                        'severity': 'high' if col_idx in mismatched_columns else 'medium',
                        'row_index': actual_row_num,
                        'col_index': col_idx,
                        'column_name': column_name,
                        'column_type': column_type,
                        'column_type_mismatch': col_idx in mismatched_columns,
                    })
        
        return differences