""" 数据处理模块 负责处理 MinerU/PaddleOCR_VL/DotsOCR 数据，添加 bbox 信息 """ from typing import List, Dict, Tuple, Optional from bs4 import BeautifulSoup try: from .text_matcher import TextMatcher from .bbox_extractor import BBoxExtractor except ImportError: from text_matcher import TextMatcher from bbox_extractor import BBoxExtractor class DataProcessor: """数据处理器""" def __init__(self, text_matcher: TextMatcher, look_ahead_window: int = 10, x_tolerance: int = 3): """ Args: text_matcher: 文本匹配器 look_ahead_window: 向前查找窗口 x_tolerance: x轴容差 """ self.text_matcher = text_matcher self.look_ahead_window = look_ahead_window # X轴容差, 用于判断文本框是否在同一列 self.x_tolerance = x_tolerance def process_mineru_data(self, mineru_data: List[Dict], paddle_text_boxes: List[Dict]) -> List[Dict]: """ 处理 MinerU 数据，添加 bbox 信息 Args: mineru_data: MinerU 数据 paddle_text_boxes: PaddleOCR 文字框列表 Returns: 合并后的数据, table cell使用paddle的bbox，其他类型只是移动指针，bbox还是沿用minerU的bbox """ merged_data = [] paddle_pointer = 0 last_matched_index = 0 # 按 bbox 排序 mineru_data.sort( key=lambda x: (x['bbox'][1], x['bbox'][0]) if 'bbox' in x else (float('inf'), float('inf')) ) for item in mineru_data: item_type = item.get('type', '') if item_type == 'table': merged_item, paddle_pointer = self._process_table( item, paddle_text_boxes, paddle_pointer ) merged_data.append(merged_item) elif item_type in ['text', 'title']: merged_item, paddle_pointer, last_matched_index = self._process_text( item, paddle_text_boxes, paddle_pointer, last_matched_index ) merged_data.append(merged_item) elif item_type == 'list': merged_item, paddle_pointer, last_matched_index = self._process_list( item, paddle_text_boxes, paddle_pointer, last_matched_index ) merged_data.append(merged_item) else: merged_data.append(item.copy()) return merged_data def process_dotsocr_data(self, dotsocr_data: List[Dict], paddle_text_boxes: List[Dict]) -> List[Dict]: """ 🎯 处理 DotsOCR 数据，转换为 MinerU 格式并添加 bbox 信息 Args: dotsocr_data: DotsOCR 数据 paddle_text_boxes: PaddleOCR 文字框列表 Returns: MinerU 格式的合并数据 """ merged_data = [] paddle_pointer = 0 last_matched_index = 0 # 按 bbox 排序 dotsocr_data.sort( key=lambda x: (x['bbox'][1], x['bbox'][0]) if 'bbox' in x else (float('inf'), float('inf')) ) for item in dotsocr_data: # 🎯 转换为 MinerU 格式 mineru_item = self._convert_dotsocr_to_mineru(item) category = mineru_item.get('type', '') # 🎯 根据类型处理 if category.lower() == 'table': merged_item, paddle_pointer = self._process_table( mineru_item, paddle_text_boxes, paddle_pointer ) merged_data.append(merged_item) elif category.lower() in ['text', 'title', 'header', 'footer']: merged_item, paddle_pointer, last_matched_index = self._process_text( mineru_item, paddle_text_boxes, paddle_pointer, last_matched_index ) merged_data.append(merged_item) elif category.lower() == 'list': merged_item, paddle_pointer, last_matched_index = self._process_list( mineru_item, paddle_text_boxes, paddle_pointer, last_matched_index ) merged_data.append(merged_item) else: # Page-header, Page-footer, Picture 等 merged_data.append(mineru_item) return merged_data def _convert_dotsocr_to_mineru(self, dotsocr_item: Dict) -> Dict: """ 🎯 将 DotsOCR 格式转换为 MinerU 格式 DotsOCR: { "category": "Table", "bbox": [x1, y1, x2, y2], "text": "..." } MinerU: { "type": "table", "bbox": [x1, y1, x2, y2], "table_body": "...", "page_idx": 0 } """ category = dotsocr_item.get('category', '') # 🎯 Category 映射 category_map = { 'Page-header': 'header', 'Page-footer': 'footer', 'Picture': 'image', 'Figure': 'image', 'Section-header': 'title', 'Table': 'table', 'Text': 'text', 'Title': 'title', 'List': 'list', 'Caption': 'title' } mineru_type = category_map.get(category, 'text') # 🎯 基础转换 mineru_item = { 'type': mineru_type, 'bbox': dotsocr_item.get('bbox', []), 'page_idx': 0 # DotsOCR 默认单页 } # 🎯 处理文本内容 text = dotsocr_item.get('text', '') if mineru_type == 'table': # 表格：text -> table_body mineru_item['table_body'] = text else: # 其他类型：保持 text mineru_item['text'] = text # 标题级别 if category == 'Section-header': mineru_item['text_level'] = 1 return mineru_item def process_paddleocr_vl_data(self, paddleocr_vl_data: Dict, paddle_text_boxes: List[Dict]) -> List[Dict]: """ 处理 PaddleOCR_VL 数据，添加 bbox 信息 Args: paddleocr_vl_data: PaddleOCR_VL 数据 (JSON 对象) paddle_text_boxes: PaddleOCR 文字框列表 Returns: 🎯 MinerU 格式的合并数据（统一输出格式） """ merged_data = [] paddle_pointer = 0 last_matched_index = 0 # 🎯 获取旋转角度和原始图像尺寸 rotation_angle = self._get_rotation_angle_from_vl(paddleocr_vl_data) orig_image_size = None if rotation_angle != 0: orig_image_size = self._get_original_image_size_from_vl(paddleocr_vl_data) print(f"🔄 PaddleOCR_VL 检测到旋转角度: {rotation_angle}°") print(f"📐 原始图像尺寸: {orig_image_size[0]} x {orig_image_size[1]}") # 提取 parsing_res_list parsing_res_list = paddleocr_vl_data.get('parsing_res_list', []) # 按 bbox 排序 parsing_res_list.sort( key=lambda x: (x['block_bbox'][1], x['block_bbox'][0]) if 'block_bbox' in x else (float('inf'), float('inf')) ) for item in parsing_res_list: # 🎯 先转换 bbox 坐标（如果需要） if rotation_angle != 0 and orig_image_size: item = self._transform_vl_block_bbox(item, rotation_angle, orig_image_size) # 🎯 统一转换为 MinerU 格式 mineru_item = self._convert_paddleocr_vl_to_mineru(item) item_type = mineru_item.get('type', '') # 🎯 根据类型处理（复用 MinerU 的通用方法） if item_type == 'table': merged_item, paddle_pointer = self._process_table( mineru_item, paddle_text_boxes, paddle_pointer ) merged_data.append(merged_item) elif item_type in ['text', 'title', 'header', 'footer', 'equation']: merged_item, paddle_pointer, last_matched_index = self._process_text( mineru_item, paddle_text_boxes, paddle_pointer, last_matched_index ) merged_data.append(merged_item) elif item_type == 'list': merged_item, paddle_pointer, last_matched_index = self._process_list( mineru_item, paddle_text_boxes, paddle_pointer, last_matched_index ) merged_data.append(merged_item) else: # 其他类型（image 等）直接添加 merged_data.append(mineru_item) return merged_data def _get_rotation_angle_from_vl(self, paddleocr_vl_data: Dict) -> float: """从 PaddleOCR_VL 数据中获取旋转角度""" return BBoxExtractor._get_rotation_angle(paddleocr_vl_data) def _get_original_image_size_from_vl(self, paddleocr_vl_data: Dict) -> tuple: """从 PaddleOCR_VL 数据中获取原始图像尺寸""" return BBoxExtractor._get_original_image_size(paddleocr_vl_data) def _transform_vl_block_bbox(self, item: Dict, angle: float, orig_image_size: tuple) -> Dict: """ 转换 PaddleOCR_VL 的 block_bbox 坐标 Args: item: PaddleOCR_VL 的 block 数据 angle: 旋转角度 orig_image_size: 原始图像尺寸 Returns: 转换后的 block 数据 """ transformed_item = item.copy() if 'block_bbox' not in item: return transformed_item block_bbox = item['block_bbox'] if len(block_bbox) < 4: return transformed_item # block_bbox 格式: [x1, y1, x2, y2] # 转换为 poly 格式进行旋转 poly = [ [block_bbox[0], block_bbox[1]], # 左上 [block_bbox[2], block_bbox[1]], # 右上 [block_bbox[2], block_bbox[3]], # 右下 [block_bbox[0], block_bbox[3]] # 左下 ] # 🎯 使用 BBoxExtractor 的坐标转换方法 transformed_poly = BBoxExtractor._inverse_rotate_coordinates( poly, angle, orig_image_size ) # 转换回 bbox 格式 xs = [p[0] for p in transformed_poly] ys = [p[1] for p in transformed_poly] transformed_bbox = [min(xs), min(ys), max(xs), max(ys)] transformed_item['block_bbox'] = transformed_bbox return transformed_item def _convert_paddleocr_vl_to_mineru(self, paddleocr_vl_item: Dict) -> Dict: """ 🎯 将 PaddleOCR_VL 格式转换为 MinerU 格式 基于 PP-DocLayout_plus-L 的 20 种类别 """ block_label = paddleocr_vl_item.get('block_label', '') # 🎯 PP-DocLayout_plus-L 类别映射（共 20 种） label_map = { # 标题类（3种） 'paragraph_title': 'title', 'doc_title': 'title', 'figure_table_chart_title': 'title', # 文本类（9种） 'text': 'text', 'number': 'text', 'content': 'text', 'abstract': 'text', 'footnote': 'text', 'aside_text': 'text', 'algorithm': 'text', 'reference': 'text', 'reference_content': 'text', # 页眉页脚（2种） 'header': 'header', 'footer': 'footer', # 表格（1种） 'table': 'table', # 图片/图表（3种） 'image': 'image', 'chart': 'image', 'seal': 'image', # 公式（2种） 'formula': 'equation', 'formula_number': 'equation' } mineru_type = label_map.get(block_label, 'text') mineru_item = { 'type': mineru_type, 'bbox': paddleocr_vl_item.get('block_bbox', []), 'page_idx': 0 } content = paddleocr_vl_item.get('block_content', '') if mineru_type == 'table': mineru_item['table_body'] = content else: mineru_item['text'] = content # 标题级别 if block_label == 'doc_title': mineru_item['text_level'] = 1 elif block_label == 'paragraph_title': mineru_item['text_level'] = 2 elif block_label == 'figure_table_chart_title': mineru_item['text_level'] = 3 return mineru_item def _process_table(self, item: Dict, paddle_text_boxes: List[Dict], start_pointer: int) -> Tuple[Dict, int]: """ 处理表格类型（MinerU 格式） 策略： - 解析 HTML 表格 - 为每个单元格匹配 PaddleOCR 的 bbox - 返回处理后的表格和新指针位置 """ table_body = item.get('table_body', '') if not table_body: print(f"⚠️ 表格内容为空，跳过") return item, start_pointer try: # 🔑 传入 table_bbox 用于筛选 table_bbox = item.get('bbox') # MinerU 提供的表格边界 enhanced_html, cells, new_pointer = self._enhance_table_html_with_bbox( table_body, paddle_text_boxes, start_pointer, table_bbox # ✅ 传入边界框 ) # 更新 item item['table_body'] = enhanced_html item['table_cells'] = cells # 统计信息 matched_count = len(cells) total_cells = len(BeautifulSoup(table_body, 'html.parser').find_all(['td', 'th'])) print(f" 表格单元格: {matched_count}/{total_cells} 匹配") return item, new_pointer except Exception as e: print(f"⚠️ 表格处理失败: {e}") import traceback traceback.print_exc() return item, start_pointer def _process_text(self, item: Dict, paddle_text_boxes: List[Dict], paddle_pointer: int, last_matched_index: int) -> Tuple[Dict, int, int]: """处理文本""" merged_item = item.copy() text = item.get('text', '') matched_bbox, paddle_pointer, last_matched_index = \ self.text_matcher.find_matching_bbox( text, paddle_text_boxes, paddle_pointer, last_matched_index, self.look_ahead_window ) if matched_bbox: matched_bbox['used'] = True return merged_item, paddle_pointer, last_matched_index def _process_list(self, item: Dict, paddle_text_boxes: List[Dict], paddle_pointer: int, last_matched_index: int) -> Tuple[Dict, int, int]: """处理列表""" merged_item = item.copy() list_items = item.get('list_items', []) for list_item in list_items: matched_bbox, paddle_pointer, last_matched_index = \ self.text_matcher.find_matching_bbox( list_item, paddle_text_boxes, paddle_pointer, last_matched_index, self.look_ahead_window ) if matched_bbox: matched_bbox['used'] = True return merged_item, paddle_pointer, last_matched_index def _enhance_table_html_with_bbox(self, html: str, paddle_text_boxes: List[Dict], start_pointer: int, table_bbox: Optional[List[int]] = None) -> Tuple[str, List[Dict], int]: """ 为 HTML 表格添加 bbox 信息（优化版：先筛选表格区域） 策略： 1. 根据 table_bbox 筛选出表格区域内的 paddle_text_boxes 2. 将筛选后的 boxes 按行分组 3. 智能匹配 HTML 行与 paddle 行组 4. 在匹配的组内查找单元格 Args: html: HTML 表格 paddle_text_boxes: 全部 paddle OCR 结果 start_pointer: 开始位置 table_bbox: 表格边界框 [x1, y1, x2, y2] """ soup = BeautifulSoup(html, 'html.parser') cells = [] # 🔑 第一步：筛选表格区域内的 paddle boxes table_region_boxes, actual_table_bbox = self._filter_boxes_in_table_region( paddle_text_boxes[start_pointer:], table_bbox, html ) if not table_region_boxes: print(f"⚠️ 未在表格区域找到 paddle boxes") return str(soup), cells, start_pointer print(f"📊 表格区域: {len(table_region_boxes)} 个文本框") print(f" 边界: {actual_table_bbox}") # 🔑 第二步：将表格区域的 boxes 按行分组 grouped_boxes = self._group_paddle_boxes_by_rows( table_region_boxes, y_tolerance=20 ) # 🔑 第三步：在每组内按 x 坐标排序 for group in grouped_boxes: group['boxes'].sort(key=lambda x: x['bbox'][0]) grouped_boxes.sort(key=lambda g: g['y_center']) print(f" 分组: {len(grouped_boxes)} 行") # 🔑 第四步：智能匹配 HTML 行与 paddle 行组 html_rows = soup.find_all('tr') row_mapping = self._match_html_rows_to_paddle_groups(html_rows, grouped_boxes) print(f" HTML行: {len(html_rows)} 行") print(f" 映射: {len([v for v in row_mapping.values() if v])} 个有效映射") # 🔑 第五步：遍历 HTML 表格，使用映射关系查找 for row_idx, row in enumerate(html_rows): group_indices = row_mapping.get(row_idx, []) if not group_indices: continue # 合并多个组的 boxes current_boxes = [] for group_idx in group_indices: if group_idx < len(grouped_boxes): current_boxes.extend(grouped_boxes[group_idx]['boxes']) current_boxes.sort(key=lambda x: x['bbox'][0]) # 🎯 关键改进：提取 HTML 单元格并预先确定列边界 html_cells = row.find_all(['td', 'th']) if not html_cells: continue # 🔑 预估列边界（基于 x 坐标分布） col_boundaries = self._estimate_column_boundaries( current_boxes, len(html_cells) ) print(f" 行 {row_idx + 1}: {len(html_cells)} 列，边界: {col_boundaries}") # 🎯 关键改进：顺序指针匹配 box_pointer = 0 # 当前行的 boxes 指针 for col_idx, cell in enumerate(html_cells): cell_text = cell.get_text(strip=True) if not cell_text: continue # 🔑 从当前指针开始匹配 matched_result = self._match_cell_sequential( cell_text, current_boxes, col_boundaries, box_pointer ) if matched_result: merged_bbox = matched_result['bbox'] merged_text = matched_result['text'] cell['data-bbox'] = f"[{merged_bbox[0]},{merged_bbox[1]},{merged_bbox[2]},{merged_bbox[3]}]" cell['data-score'] = f"{matched_result['score']:.4f}" cell['data-paddle-indices'] = str(matched_result['paddle_indices']) cells.append({ 'type': 'table_cell', 'text': cell_text, 'matched_text': merged_text, 'bbox': merged_bbox, 'row': row_idx + 1, 'col': col_idx + 1, 'score': matched_result['score'], 'paddle_bbox_indices': matched_result['paddle_indices'] }) # 标记已使用 for box in matched_result['used_boxes']: box['used'] = True # 🎯 移动指针到最后使用的 box 之后 box_pointer = matched_result['last_used_index'] + 1 print(f" 列 {col_idx + 1}: '{cell_text[:20]}...' 匹配 {len(matched_result['used_boxes'])} 个box (指针: {box_pointer})") # 计算新的指针位置 used_count = sum(1 for box in table_region_boxes if box.get('used')) new_pointer = start_pointer + used_count print(f" 匹配: {len(cells)} 个单元格") return str(soup), cells, new_pointer def _estimate_column_boundaries(self, boxes: List[Dict], num_cols: int) -> List[Tuple[int, int]]: """ 估算列边界（改进版：处理同列多文本框） Args: boxes: 当前行的所有 boxes（已按 x 排序） num_cols: HTML 表格的列数 Returns: 列边界列表 [(x_start, x_end), ...] """ if not boxes: return [] # 🔑 关键改进：先按 x 坐标聚类（合并同列的多个文本框） x_clusters = self._cluster_boxes_by_x(boxes, x_tolerance=self.x_tolerance) print(f" X聚类: {len(boxes)} 个boxes -> {len(x_clusters)} 个列簇") # 获取所有 x 坐标范围 x_min = min(cluster['x_min'] for cluster in x_clusters) x_max = max(cluster['x_max'] for cluster in x_clusters) # 🎯 策略 1: 如果聚类数量<=列数接近 if len(x_clusters) <= num_cols: # 直接使用聚类边界 boundaries = [(cluster['x_min'], cluster['x_max']) for cluster in x_clusters] return boundaries # 🎯 策略 2: 聚类数多于列数（某些列有多个文本簇） if len(x_clusters) > num_cols: print(f" ℹ️ 聚类数 {len(x_clusters)} > 列数 {num_cols}，合并相近簇") # 合并相近的簇 merged_clusters = self._merge_close_clusters(x_clusters, num_cols) boundaries = [(cluster['x_min'], cluster['x_max']) for cluster in merged_clusters] return boundaries return [] def _cluster_boxes_by_x(self, boxes: List[Dict], x_tolerance: int = 3) -> List[Dict]: """ 按 x 坐标聚类（合并同列的多个文本框） Args: boxes: 文本框列表 x_tolerance: X坐标容忍度 Returns: 聚类列表 [{'x_min': int, 'x_max': int, 'boxes': List[Dict]}, ...] """ if not boxes: return [] # 按左边界 x 坐标排序 sorted_boxes = sorted(boxes, key=lambda b: b['bbox'][0]) clusters = [] current_cluster = None for box in sorted_boxes: bbox = box['bbox'] x_start = bbox[0] x_end = bbox[2] if current_cluster is None: # 开始新簇 current_cluster = { 'x_min': x_start, 'x_max': x_end, 'boxes': [box] } else: # 🔑 检查是否属于当前簇（修正后的逻辑） # 1. x 坐标有重叠：x_start <= current_x_max 且 x_end >= current_x_min # 2. 或者距离在容忍度内 has_overlap = (x_start <= current_cluster['x_max'] and x_end >= current_cluster['x_min']) is_close = abs(x_start - current_cluster['x_max']) <= x_tolerance if has_overlap or is_close: # 合并到当前簇 current_cluster['boxes'].append(box) current_cluster['x_min'] = min(current_cluster['x_min'], x_start) current_cluster['x_max'] = max(current_cluster['x_max'], x_end) else: # 保存当前簇，开始新簇 clusters.append(current_cluster) current_cluster = { 'x_min': x_start, 'x_max': x_end, 'boxes': [box] } # 添加最后一簇 if current_cluster: clusters.append(current_cluster) return clusters def _merge_close_clusters(self, clusters: List[Dict], target_count: int) -> List[Dict]: """ 合并相近的簇，直到数量等于目标列数 Args: clusters: 聚类列表 target_count: 目标列数 Returns: 合并后的聚类列表 """ if len(clusters) <= target_count: return clusters # 复制一份，避免修改原数据 working_clusters = [c.copy() for c in clusters] while len(working_clusters) > target_count: # 找到距离最近的两个簇 min_distance = float('inf') merge_idx = 0 for i in range(len(working_clusters) - 1): distance = working_clusters[i + 1]['x_min'] - working_clusters[i]['x_max'] if distance < min_distance: min_distance = distance merge_idx = i # 合并 cluster1 = working_clusters[merge_idx] cluster2 = working_clusters[merge_idx + 1] merged_cluster = { 'x_min': cluster1['x_min'], 'x_max': cluster2['x_max'], 'boxes': cluster1['boxes'] + cluster2['boxes'] } # 替换 working_clusters[merge_idx] = merged_cluster working_clusters.pop(merge_idx + 1) return working_clusters def _get_boxes_in_column(self, boxes: List[Dict], boundaries: List[Tuple[int, int]], col_idx: int) -> List[Dict]: """ 获取指定列范围内的 boxes（改进版：包含重叠） Args: boxes: 当前行的所有 boxes boundaries: 列边界 col_idx: 列索引 Returns: 该列的 boxes """ if col_idx >= len(boundaries): return [] x_start, x_end = boundaries[col_idx] col_boxes = [] for box in boxes: bbox = box['bbox'] box_x_start = bbox[0] box_x_end = bbox[2] # 🔑 改进：检查是否有重叠（不只是中心点） overlap = not (box_x_start > x_end or box_x_end < x_start) if overlap: col_boxes.append(box) return col_boxes def _filter_boxes_in_table_region(self, paddle_boxes: List[Dict], table_bbox: Optional[List[int]], html: str) -> Tuple[List[Dict], List[int]]: """ 筛选表格区域内的 paddle boxes 策略： 1. 如果有 table_bbox，使用边界框筛选（扩展边界） 2. 如果没有 table_bbox，通过内容匹配推断区域 Args: paddle_boxes: paddle OCR 结果 table_bbox: 表格边界框 [x1, y1, x2, y2] html: HTML 内容（用于内容验证） Returns: (筛选后的 boxes, 实际表格边界框) """ if not paddle_boxes: return [], [0, 0, 0, 0] # 🎯 策略 1: 使用提供的 table_bbox（扩展边界） if table_bbox and len(table_bbox) == 4: x1, y1, x2, y2 = table_bbox # 扩展边界（考虑边框外的文本） margin = 20 expanded_bbox = [ max(0, x1 - margin), max(0, y1 - margin), x2 + margin, y2 + margin ] filtered = [] for box in paddle_boxes: bbox = box['bbox'] box_center_x = (bbox[0] + bbox[2]) / 2 box_center_y = (bbox[1] + bbox[3]) / 2 # 中心点在扩展区域内 if (expanded_bbox[0] <= box_center_x <= expanded_bbox[2] and expanded_bbox[1] <= box_center_y <= expanded_bbox[3]): filtered.append(box) if filtered: # 计算实际边界框 actual_bbox = [ min(b['bbox'][0] for b in filtered), min(b['bbox'][1] for b in filtered), max(b['bbox'][2] for b in filtered), max(b['bbox'][3] for b in filtered) ] return filtered, actual_bbox # 🎯 策略 2: 通过内容匹配推断区域 print(" ℹ️ 无 table_bbox，使用内容匹配推断表格区域...") # 提取 HTML 中的所有文本 from bs4 import BeautifulSoup soup = BeautifulSoup(html, 'html.parser') html_texts = set() for cell in soup.find_all(['td', 'th']): text = cell.get_text(strip=True) if text: html_texts.add(self.text_matcher.normalize_text(text)) if not html_texts: return [], [0, 0, 0, 0] # 找出与 HTML 内容匹配的 boxes matched_boxes = [] for box in paddle_boxes: normalized_text = self.text_matcher.normalize_text(box['text']) # 检查是否匹配 if any(normalized_text in ht or ht in normalized_text for ht in html_texts): matched_boxes.append(box) if not matched_boxes: # 🔑 降级：如果精确匹配失败，使用模糊匹配 print(" ℹ️ 精确匹配失败，尝试模糊匹配...") from fuzzywuzzy import fuzz for box in paddle_boxes: normalized_text = self.text_matcher.normalize_text(box['text']) for ht in html_texts: similarity = fuzz.partial_ratio(normalized_text, ht) if similarity >= 70: # 降低阈值 matched_boxes.append(box) break if matched_boxes: # 计算边界框 actual_bbox = [ min(b['bbox'][0] for b in matched_boxes), min(b['bbox'][1] for b in matched_boxes), max(b['bbox'][2] for b in matched_boxes), max(b['bbox'][3] for b in matched_boxes) ] # 🔑 扩展边界，包含可能遗漏的文本 margin = 30 expanded_bbox = [ max(0, actual_bbox[0] - margin), max(0, actual_bbox[1] - margin), actual_bbox[2] + margin, actual_bbox[3] + margin ] # 重新筛选（包含边界上的文本） final_filtered = [] for box in paddle_boxes: bbox = box['bbox'] box_center_x = (bbox[0] + bbox[2]) / 2 box_center_y = (bbox[1] + bbox[3]) / 2 if (expanded_bbox[0] <= box_center_x <= expanded_bbox[2] and expanded_bbox[1] <= box_center_y <= expanded_bbox[3]): final_filtered.append(box) return final_filtered, actual_bbox # 🔑 最后的降级：返回所有 boxes print(" ⚠️ 无法确定表格区域，使用所有 paddle boxes") if paddle_boxes: actual_bbox = [ min(b['bbox'][0] for b in paddle_boxes), min(b['bbox'][1] for b in paddle_boxes), max(b['bbox'][2] for b in paddle_boxes), max(b['bbox'][3] for b in paddle_boxes) ] return paddle_boxes, actual_bbox return [], [0, 0, 0, 0] def _group_paddle_boxes_by_rows(self, paddle_boxes: List[Dict], y_tolerance: int = 20) -> List[Dict]: """ 将 paddle_text_boxes 按 y 坐标分组（聚类） Args: paddle_boxes: Paddle OCR 文字框列表 y_tolerance: Y 坐标容忍度（像素） Returns: 分组列表，每组包含 {'y_center': float, 'boxes': List[Dict]} """ if not paddle_boxes: return [] # 计算每个 box 的中心 y 坐标 boxes_with_y = [] for box in paddle_boxes: bbox = box['bbox'] y_center = (bbox[1] + bbox[3]) / 2 boxes_with_y.append({ 'y_center': y_center, 'box': box }) # 按 y 坐标排序 boxes_with_y.sort(key=lambda x: x['y_center']) # 聚类 groups = [] current_group = None for item in boxes_with_y: if current_group is None: # 开始新组 current_group = { 'y_center': item['y_center'], 'boxes': [item['box']] } else: # 检查是否属于当前组 if abs(item['y_center'] - current_group['y_center']) <= y_tolerance: current_group['boxes'].append(item['box']) # 更新组的中心（使用平均值） current_group['y_center'] = sum( b['bbox'][1] + b['bbox'][3] for b in current_group['boxes'] ) / (2 * len(current_group['boxes'])) else: # 保存当前组，开始新组 groups.append(current_group) current_group = { 'y_center': item['y_center'], 'boxes': [item['box']] } # 添加最后一组 if current_group: groups.append(current_group) return groups def _match_html_rows_to_paddle_groups(self, html_rows: List, grouped_boxes: List[Dict]) -> Dict[int, List[int]]: """ 智能匹配 HTML 行与 paddle 分组（改进版：处理跨行文本） 策略： 1. 第一遍：基于内容精确匹配 2. 第二遍：将未使用的组合并到相邻已匹配的行 """ if not html_rows or not grouped_boxes: return {} mapping = {} # 🎯 策略 1: 数量相等，简单 1:1 映射 if len(html_rows) == len(grouped_boxes): for i in range(len(html_rows)): mapping[i] = [i] return mapping # 🎯 策略 2: 第一遍 - 基于内容精确匹配 used_groups = set() for row_idx, row in enumerate(html_rows): row_texts = [cell.get_text(strip=True) for cell in row.find_all(['td', 'th'])] row_texts = [t for t in row_texts if t] if not row_texts: mapping[row_idx] = [] continue row_text_normalized = [self.text_matcher.normalize_text(t) for t in row_texts] # 查找最匹配的 paddle 组 best_groups = [] best_score = 0 # 尝试匹配单个组 for group_idx, group in enumerate(grouped_boxes): if group_idx in used_groups: continue group_texts = [self.text_matcher.normalize_text(b['text']) for b in group['boxes'] if not b.get('used')] match_count = sum(1 for rt in row_text_normalized if any(rt in gt or gt in rt for gt in group_texts)) coverage = match_count / len(row_texts) if row_texts else 0 if coverage > best_score: best_score = coverage best_groups = [group_idx] # 🔑 如果单组匹配度不高，尝试匹配多个连续组 if best_score < 0.5: # 从当前位置向后查找 start_group = min([g for g in range(len(grouped_boxes)) if g not in used_groups], default=0) combined_texts = [] combined_groups = [] for group_idx in range(start_group, min(start_group + 5, len(grouped_boxes))): if group_idx in used_groups: continue combined_groups.append(group_idx) combined_texts.extend([ self.text_matcher.normalize_text(b['text']) for b in grouped_boxes[group_idx]['boxes'] if not b.get('used') ]) match_count = sum(1 for rt in row_text_normalized if any(rt in gt or gt in rt for gt in combined_texts)) coverage = match_count / len(row_texts) if row_texts else 0 if coverage > best_score: best_score = coverage best_groups = combined_groups.copy() # 记录映射 if best_groups and best_score > 0.3: mapping[row_idx] = best_groups used_groups.update(best_groups) else: # 降级策略：位置推测 estimated_group = min(row_idx, len(grouped_boxes) - 1) if estimated_group not in used_groups: mapping[row_idx] = [estimated_group] used_groups.add(estimated_group) else: mapping[row_idx] = [] # 🎯 策略 3: 第二遍 - 处理未使用的组（关键！） unused_groups = [i for i in range(len(grouped_boxes)) if i not in used_groups] if unused_groups: print(f" ℹ️ 发现 {len(unused_groups)} 个未匹配的 paddle 组: {unused_groups}") # 🔑 将未使用的组合并到相邻的已匹配行 for unused_idx in unused_groups: # 🎯 关键改进：计算与相邻行的边界距离 unused_group = grouped_boxes[unused_idx] unused_y_min = min(b['bbox'][1] for b in unused_group['boxes']) unused_y_max = max(b['bbox'][3] for b in unused_group['boxes']) # 🔑 查找上方和下方最近的已使用组 above_idx = None below_idx = None above_distance = float('inf') below_distance = float('inf') # 向上查找 for i in range(unused_idx - 1, -1, -1): if i in used_groups: above_idx = i # 🎯 边界距离：unused 的最小 y - above 的最大 y above_group = grouped_boxes[i] max_y_box = max( above_group['boxes'], key=lambda b: b['bbox'][3] ) above_y_center = (max_y_box['bbox'][1] + max_y_box['bbox'][3]) / 2 above_distance = abs(unused_y_min - above_y_center) print(f" • 组 {unused_idx} 与上方组 {i} 距离: {above_distance:.1f}px") break # 向下查找 for i in range(unused_idx + 1, len(grouped_boxes)): if i in used_groups: below_idx = i # 🎯 边界距离：below 的最小 y - unused 的最大 y below_group = grouped_boxes[i] min_y_box = min( below_group['boxes'], key=lambda b: b['bbox'][1] ) below_y_center = (min_y_box['bbox'][1] + min_y_box['bbox'][3]) / 2 below_distance = abs(below_y_center - unused_y_max) print(f" • 组 {unused_idx} 与下方组 {i} 距离: {below_distance:.1f}px") break # 🎯 选择距离更近的一侧 if above_idx is not None and below_idx is not None: # 都存在，选择距离更近的 if above_distance < below_distance: closest_used_idx = above_idx merge_direction = "上方" else: closest_used_idx = below_idx merge_direction = "下方" print(f" ✓ 组 {unused_idx} 选择合并到{merge_direction}组 {closest_used_idx}") elif above_idx is not None: closest_used_idx = above_idx merge_direction = "上方" elif below_idx is not None: closest_used_idx = below_idx merge_direction = "下方" else: print(f" ⚠️ 组 {unused_idx} 无相邻已使用组，跳过") continue # 🔑 找到该组对应的 HTML 行 target_html_row = None for html_row_idx, group_indices in mapping.items(): if closest_used_idx in group_indices: target_html_row = html_row_idx break if target_html_row is not None: # 🎯 根据合并方向决定目标行 if merge_direction == "上方": # 合并到上方对应的 HTML 行 if target_html_row in mapping: if unused_idx not in mapping[target_html_row]: mapping[target_html_row].append(unused_idx) print(f" • 组 {unused_idx} 合并到 HTML 行 {target_html_row}（上方行）") else: # 合并到下方对应的 HTML 行 if target_html_row in mapping: if unused_idx not in mapping[target_html_row]: mapping[target_html_row].append(unused_idx) print(f" • 组 {unused_idx} 合并到 HTML 行 {target_html_row}（下方行）") used_groups.add(unused_idx) # 🔑 策略 4: 第三遍 - 按 y 坐标排序每行的组索引 for row_idx in mapping: if mapping[row_idx]: mapping[row_idx].sort(key=lambda idx: grouped_boxes[idx]['y_center']) return mapping def _match_cell_sequential(self, cell_text: str, boxes: List[Dict], col_boundaries: List[Tuple[int, int]], start_idx: int) -> Optional[Dict]: """ 🎯 顺序匹配单元格：从指定位置开始，逐步合并 boxes 直到匹配 策略： 1. 找到第一个未使用的 box 2. 尝试单个 box 精确匹配 3. 如果失败，尝试合并多个 boxes Args: cell_text: HTML 单元格文本 boxes: 候选 boxes（已按 x 坐标排序） col_boundaries: 列边界列表 start_idx: 起始索引 Returns: {'bbox': [x1,y1,x2,y2], 'text': str, 'score': float, 'paddle_indices': [idx1, idx2], 'used_boxes': [box1, box2], 'last_used_index': int} """ from fuzzywuzzy import fuzz cell_text_normalized = self.text_matcher.normalize_text(cell_text) if len(cell_text_normalized) < 2: return None # 🔑 找到第一个未使用的 box first_unused_idx = start_idx while first_unused_idx < len(boxes) and boxes[first_unused_idx].get('used'): first_unused_idx += 1 if first_unused_idx >= len(boxes): return None # 🔑 策略 1: 单个 box 精确匹配 for box in boxes[first_unused_idx:]: if box.get('used'): continue box_text = self.text_matcher.normalize_text(box['text']) if cell_text_normalized == box_text: return self._build_match_result([box], box['text'], 100.0, boxes.index(box)) # 🔑 策略 2: 多个 boxes 合并匹配 unused_boxes = [b for b in boxes if not b.get('used')] # 合并同列的 boxes 合并 merged_bboxes = [] for col_idx in range(len(col_boundaries)): combo_boxes = self._get_boxes_in_column(unused_boxes, col_boundaries, col_idx) if len(combo_boxes) > 0: sorted_combo = sorted(combo_boxes, key=lambda b: (b['bbox'][1], b['bbox'][0])) merged_text = ''.join([b['text'] for b in sorted_combo]) merged_bboxes.append({ 'text': merged_text, 'sorted_combo': sorted_combo }) for box in merged_bboxes: # 1. 精确匹配 merged_text_normalized = self.text_matcher.normalize_text(box['text']) if cell_text_normalized == merged_text_normalized: last_sort_idx = boxes.index(box['sorted_combo'][-1]) return self._build_match_result(box['sorted_combo'], box['text'], 100.0, last_sort_idx) # 2. 子串匹配 is_substring = (cell_text_normalized in merged_text_normalized or merged_text_normalized in cell_text_normalized) # 3. 模糊匹配 similarity = fuzz.partial_ratio(cell_text_normalized, merged_text_normalized) # 🎯 子串匹配加分 if is_substring: similarity = min(100, similarity + 10) if similarity >= self.text_matcher.similarity_threshold: print(f" ✓ 匹配成功: '{cell_text[:15]}' vs '{merged_text[:15]}' (相似度: {similarity})") return self._build_match_result(box['sorted_combo'], box['text'], similarity, start_idx) print(f" ✗ 匹配失败: '{cell_text[:15]}'") return None def _build_match_result(self, boxes: List[Dict], text: str, score: float, last_index: int) -> Dict: """构建匹配结果""" merged_bbox = [ min(b['bbox'][0] for b in boxes), min(b['bbox'][1] for b in boxes), max(b['bbox'][2] for b in boxes), max(b['bbox'][3] for b in boxes) ] return { 'bbox': merged_bbox, 'text': text, 'score': score, 'paddle_indices': [b['paddle_bbox_index'] for b in boxes], 'used_boxes': boxes, 'last_used_index': last_index }