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