|
|
@@ -39,19 +39,7 @@ class TableCellMatcher:
|
|
|
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]
|
|
|
+ 为 HTML 表格添加 bbox 信息（优化版：使用行级动态规划）
|
|
|
"""
|
|
|
soup = BeautifulSoup(html, 'html.parser')
|
|
|
cells = []
|
|
|
@@ -68,7 +56,6 @@ class TableCellMatcher:
|
|
|
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(
|
|
|
@@ -84,16 +71,13 @@ class TableCellMatcher:
|
|
|
|
|
|
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])} 个有效映射")
|
|
|
+ print(f" HTML行: {len(html_rows)} 行, 映射: {len([v for v in row_mapping.values() if v])} 个有效映射")
|
|
|
|
|
|
- # 🔑 第五步：遍历 HTML 表格，使用映射关系查找
|
|
|
+ # 🔑 第五步：遍历 HTML 表格，使用 DP 进行行内匹配
|
|
|
for row_idx, row in enumerate(html_rows):
|
|
|
group_indices = row_mapping.get(row_idx, [])
|
|
|
|
|
|
@@ -106,263 +90,244 @@ class TableCellMatcher:
|
|
|
if group_idx < len(grouped_boxes):
|
|
|
current_boxes.extend(grouped_boxes[group_idx]['boxes'])
|
|
|
|
|
|
+ # 再次按 x 排序确保顺序
|
|
|
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 指针
|
|
|
+ # 🎯 核心变更：使用行级 DP 替代原来的顺序匹配
|
|
|
+ # 输入：HTML 单元格列表, OCR Box 列表
|
|
|
+ # 输出：匹配结果列表
|
|
|
+ dp_results = self._match_cells_in_row_dp(html_cells, current_boxes)
|
|
|
|
|
|
- for col_idx, cell in enumerate(html_cells):
|
|
|
- cell_text = cell.get_text(strip=True)
|
|
|
+ print(f" 行 {row_idx + 1}: {len(html_cells)} 列, 匹配到 {len(dp_results)} 个单元格")
|
|
|
+
|
|
|
+ # 解析 DP 结果并填充 cells 列表
|
|
|
+ for res in dp_results:
|
|
|
+ cell_idx = res['cell_idx']
|
|
|
+ match_info = res['match_info']
|
|
|
|
|
|
- if not cell_text:
|
|
|
- continue
|
|
|
+ cell_element = html_cells[cell_idx]
|
|
|
+ cell_text = cell_element.get_text(strip=True)
|
|
|
|
|
|
- # 🔑 从当前指针开始匹配
|
|
|
- matched_result = self._match_cell_sequential(
|
|
|
- cell_text,
|
|
|
- current_boxes,
|
|
|
- col_boundaries,
|
|
|
- box_pointer
|
|
|
- )
|
|
|
+ matched_boxes = match_info['boxes']
|
|
|
+ matched_text = match_info['text']
|
|
|
+ score = match_info['score']
|
|
|
|
|
|
- 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})")
|
|
|
-
|
|
|
- # 计算新的指针位置
|
|
|
+ # 标记 box 为已使用
|
|
|
+ paddle_indices = []
|
|
|
+ for box in matched_boxes:
|
|
|
+ box['used'] = True
|
|
|
+ paddle_indices.append(box.get('paddle_bbox_index', -1))
|
|
|
+
|
|
|
+ # 计算合并后的 bbox (使用原始坐标 original_bbox 优先)
|
|
|
+ merged_bbox = self._merge_boxes_bbox(matched_boxes)
|
|
|
+
|
|
|
+ # 注入 HTML 属性
|
|
|
+ cell_element['data-bbox'] = f"[{merged_bbox[0]},{merged_bbox[1]},{merged_bbox[2]},{merged_bbox[3]}]"
|
|
|
+ cell_element['data-score'] = f"{score:.4f}"
|
|
|
+ cell_element['data-paddle-indices'] = str(paddle_indices)
|
|
|
+
|
|
|
+ # 构建返回结构 (保持与原函数一致)
|
|
|
+ cells.append({
|
|
|
+ 'type': 'table_cell',
|
|
|
+ 'text': cell_text,
|
|
|
+ 'matched_text': matched_text,
|
|
|
+ 'bbox': merged_bbox,
|
|
|
+ 'row': row_idx + 1,
|
|
|
+ 'col': cell_idx + 1,
|
|
|
+ 'score': score,
|
|
|
+ 'paddle_bbox_indices': paddle_indices
|
|
|
+ })
|
|
|
+
|
|
|
+ print(f" 列 {cell_idx + 1}: '{cell_text[:15]}...' 匹配 {len(matched_boxes)} 个box (分值: {score:.1f})")
|
|
|
+
|
|
|
+ # 计算新的指针位置 (逻辑保持不变：基于 used 标记)
|
|
|
used_count = sum(1 for box in table_region_boxes if box.get('used'))
|
|
|
new_pointer = start_pointer + used_count
|
|
|
|
|
|
- print(f" 匹配: {len(cells)} 个单元格")
|
|
|
+ print(f" 总计匹配: {len(cells)} 个单元格")
|
|
|
|
|
|
return str(soup), cells, new_pointer
|
|
|
|
|
|
+ def _merge_boxes_bbox(self, boxes: List[Dict]) -> List[int]:
|
|
|
+ """辅助函数：合并多个 box 的坐标"""
|
|
|
+ if not boxes:
|
|
|
+ return [0, 0, 0, 0]
|
|
|
+
|
|
|
+ # 优先使用 original_bbox，如果没有则使用 bbox
|
|
|
+ def get_coords(b):
|
|
|
+ return b.get('original_bbox', b['bbox'])
|
|
|
+
|
|
|
+ x1 = min(get_coords(b)[0] for b in boxes)
|
|
|
+ y1 = min(get_coords(b)[1] for b in boxes)
|
|
|
+ x2 = max(get_coords(b)[2] for b in boxes)
|
|
|
+ y2 = max(get_coords(b)[3] for b in boxes)
|
|
|
+ return [x1, y1, x2, y2]
|
|
|
|
|
|
- def _estimate_column_boundaries(self, boxes: List[Dict],
|
|
|
- num_cols: int) -> List[Tuple[int, int]]:
|
|
|
+ def _match_cells_in_row_dp(self, html_cells: List, row_boxes: List[Dict]) -> List[Dict]:
|
|
|
"""
|
|
|
- 估算列边界（改进版：处理同列多文本框）
|
|
|
-
|
|
|
- Args:
|
|
|
- boxes: 当前行的所有 boxes（已按 x 排序）
|
|
|
- num_cols: HTML 表格的列数
|
|
|
-
|
|
|
- Returns:
|
|
|
- 列边界列表 [(x_start, x_end), ...]
|
|
|
+ 使用动态规划进行行内单元格匹配
|
|
|
+ 目标：找到一种分配方案，使得整行的匹配总分最高
|
|
|
"""
|
|
|
- if not boxes:
|
|
|
- return []
|
|
|
+ n_cells = len(html_cells)
|
|
|
+ n_boxes = len(row_boxes)
|
|
|
|
|
|
- # 🔑 关键改进：先按 x 坐标聚类（合并同列的多个文本框）
|
|
|
- x_clusters = self._cluster_boxes_by_x(boxes, x_tolerance=self.x_tolerance)
|
|
|
+ # dp[i][j] 表示：前 i 个单元格 消耗了 前 j 个 boxes 的最大得分
|
|
|
+ dp = np.full((n_cells + 1, n_boxes + 1), -np.inf)
|
|
|
+ dp[0][0] = 0
|
|
|
|
|
|
- print(f" X聚类: {len(boxes)} 个boxes -> {len(x_clusters)} 个列簇")
|
|
|
+ # path[i][j] = (prev_j, matched_info) 用于回溯
|
|
|
+ path = {}
|
|
|
|
|
|
- # 获取所有 x 坐标范围
|
|
|
- x_min = min(cluster['x_min'] for cluster in x_clusters)
|
|
|
- x_max = max(cluster['x_max'] for cluster in x_clusters)
|
|
|
+ # 允许合并的最大 box 数量
|
|
|
+ MAX_MERGE = 5
|
|
|
|
|
|
- # 🎯 策略 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)
|
|
|
+ for i in range(1, n_cells + 1):
|
|
|
+ cell = html_cells[i-1]
|
|
|
+ cell_text = cell.get_text(strip=True)
|
|
|
|
|
|
- boundaries = [(cluster['x_min'], cluster['x_max'])
|
|
|
- for cluster in merged_clusters]
|
|
|
- return boundaries
|
|
|
-
|
|
|
- return []
|
|
|
+ # 如果单元格为空，允许继承状态（相当于跳过该单元格）
|
|
|
+ if not cell_text:
|
|
|
+ for j in range(n_boxes + 1):
|
|
|
+ if dp[i-1][j] > -np.inf:
|
|
|
+ dp[i][j] = dp[i-1][j]
|
|
|
+ path[(i, j)] = (j, None)
|
|
|
+ continue
|
|
|
|
|
|
+ # 遍历当前 box 指针 j
|
|
|
+ for j in range(n_boxes + 1):
|
|
|
+ # 策略 A: 当前单元格不匹配任何 box (Cell Missing / OCR漏检)
|
|
|
+ if dp[i-1][j] > dp[i][j]:
|
|
|
+ dp[i][j] = dp[i-1][j]
|
|
|
+ path[(i, j)] = (j, None)
|
|
|
|
|
|
- def _cluster_boxes_by_x(self, boxes: List[Dict],
|
|
|
- x_tolerance: int = 3) -> List[Dict]:
|
|
|
+ # 策略 B: 当前单元格匹配了 k 个 boxes (从 prev_j 到 j)
|
|
|
+ # 限制搜索范围：最多往前看 MAX_MERGE 个 box
|
|
|
+ search_limit = max(0, j - MAX_MERGE)
|
|
|
+
|
|
|
+ # 允许中间跳过少量噪音 box (例如 prev_j 到 j 之间跨度大，但只取了部分)
|
|
|
+ # 但为了简化，这里假设是连续取用 row_boxes[prev_j:j]
|
|
|
+ for prev_j in range(j - 1, search_limit - 1, -1):
|
|
|
+ if dp[i-1][prev_j] == -np.inf:
|
|
|
+ continue
|
|
|
+
|
|
|
+ candidate_boxes = row_boxes[prev_j:j]
|
|
|
+
|
|
|
+ # 组合文本 (使用空格连接)
|
|
|
+ merged_text = " ".join([b['text'] for b in candidate_boxes])
|
|
|
+
|
|
|
+ # 计算得分
|
|
|
+ score = self._compute_match_score(cell_text, merged_text)
|
|
|
+
|
|
|
+ # 只有及格的匹配才考虑
|
|
|
+ if score > 50:
|
|
|
+ new_score = dp[i-1][prev_j] + score
|
|
|
+ if new_score > dp[i][j]:
|
|
|
+ dp[i][j] = new_score
|
|
|
+ path[(i, j)] = (prev_j, {
|
|
|
+ 'text': merged_text,
|
|
|
+ 'boxes': candidate_boxes,
|
|
|
+ 'score': score
|
|
|
+ })
|
|
|
+
|
|
|
+ # --- 回溯找最优解 ---
|
|
|
+ best_j = np.argmax(dp[n_cells])
|
|
|
+ if dp[n_cells][best_j] == -np.inf:
|
|
|
+ return []
|
|
|
+
|
|
|
+ results = []
|
|
|
+ curr_i, curr_j = n_cells, best_j
|
|
|
+
|
|
|
+ while curr_i > 0:
|
|
|
+ step_info = path.get((curr_i, curr_j))
|
|
|
+ if step_info:
|
|
|
+ prev_j, match_info = step_info
|
|
|
+ if match_info:
|
|
|
+ results.append({
|
|
|
+ 'cell_idx': curr_i - 1,
|
|
|
+ 'match_info': match_info
|
|
|
+ })
|
|
|
+ curr_j = prev_j
|
|
|
+ curr_i -= 1
|
|
|
+
|
|
|
+ return results[::-1]
|
|
|
+
|
|
|
+ def _compute_match_score(self, cell_text: str, box_text: str) -> float:
|
|
|
"""
|
|
|
- 按 x 坐标聚类（合并同列的多个文本框）
|
|
|
-
|
|
|
- Args:
|
|
|
- boxes: 文本框列表
|
|
|
- x_tolerance: X坐标容忍度
|
|
|
-
|
|
|
- Returns:
|
|
|
- 聚类列表 [{'x_min': int, 'x_max': int, 'boxes': List[Dict]}, ...]
|
|
|
+ 纯粹的评分函数：计算单元格文本与候选 Box 文本的匹配得分
|
|
|
+ 包含所有防御逻辑
|
|
|
"""
|
|
|
- if not boxes:
|
|
|
- return []
|
|
|
-
|
|
|
- # 按左边界 x 坐标排序
|
|
|
- sorted_boxes = sorted(boxes, key=lambda b: b['bbox'][0])
|
|
|
-
|
|
|
- clusters = []
|
|
|
- current_cluster = None
|
|
|
+ # 1. 预处理
|
|
|
+ cell_norm = self.text_matcher.normalize_text(cell_text)
|
|
|
+ box_norm = self.text_matcher.normalize_text(box_text)
|
|
|
|
|
|
- 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 not cell_norm or not box_norm:
|
|
|
+ return 0.0
|
|
|
|
|
|
- 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)
|
|
|
+ # --- ⚡️ 快速防御 ---
|
|
|
+ len_cell = len(cell_norm)
|
|
|
+ len_box = len(box_norm)
|
|
|
|
|
|
- return clusters
|
|
|
+ # 长度差异过大直接 0 分 (除非是包含关系且特征明显)
|
|
|
+ if len_box > len_cell * 3 + 5:
|
|
|
+ if len_cell < 5: return 0.0
|
|
|
|
|
|
-
|
|
|
- def _merge_close_clusters(self, clusters: List[Dict],
|
|
|
- target_count: int) -> List[Dict]:
|
|
|
- """
|
|
|
- 合并相近的簇，直到数量等于目标列数
|
|
|
+ # --- 🔍 核心相似度计算 ---
|
|
|
+ cell_proc = self._preprocess_text_for_matching(cell_text)
|
|
|
+ box_proc = self._preprocess_text_for_matching(box_text)
|
|
|
|
|
|
- Args:
|
|
|
- clusters: 聚类列表
|
|
|
- target_count: 目标列数
|
|
|
+ # A. Token Sort (解决乱序)
|
|
|
+ score_sort = fuzz.token_sort_ratio(cell_proc, box_proc)
|
|
|
|
|
|
- Returns:
|
|
|
- 合并后的聚类列表
|
|
|
- """
|
|
|
- if len(clusters) <= target_count:
|
|
|
- return clusters
|
|
|
+ # B. Partial (解决截断/包含)
|
|
|
+ score_partial = fuzz.partial_ratio(cell_norm, box_norm)
|
|
|
|
|
|
- # 复制一份，避免修改原数据
|
|
|
- working_clusters = [c.copy() for c in clusters]
|
|
|
+ # C. Subsequence (解决噪音插入)
|
|
|
+ score_subseq = 0.0
|
|
|
+ if len_cell > 5:
|
|
|
+ score_subseq = self._calculate_subsequence_score(cell_norm, box_norm)
|
|
|
+
|
|
|
+ # --- 🛡️ 深度防御逻辑 ---
|
|
|
|
|
|
- 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
|
|
|
+ # 1. 短文本防御
|
|
|
+ if score_partial > 80:
|
|
|
+ import re
|
|
|
+ has_content = lambda t: bool(re.search(r'[a-zA-Z0-9\u4e00-\u9fa5]', t))
|
|
|
|
|
|
- # 合并
|
|
|
- cluster1 = working_clusters[merge_idx]
|
|
|
- cluster2 = working_clusters[merge_idx + 1]
|
|
|
+ # 纯符号防御
|
|
|
+ if not has_content(cell_norm) and has_content(box_norm):
|
|
|
+ if len_box > len_cell + 2: score_partial = 0.0
|
|
|
|
|
|
- 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
|
|
|
+ # 微小碎片防御
|
|
|
+ elif len_cell <= 2 and len_box > 8:
|
|
|
+ score_partial = 0.0
|
|
|
+
|
|
|
+ # 覆盖率防御
|
|
|
+ else:
|
|
|
+ coverage = len_cell / len_box if len_box > 0 else 0
|
|
|
+ if coverage < 0.3 and score_sort < 45:
|
|
|
+ score_partial = 0.0
|
|
|
|
|
|
+ # 2. 子序列防御
|
|
|
+ if score_subseq > 80:
|
|
|
+ if len_box > len_cell * 1.5:
|
|
|
+ import re
|
|
|
+ if re.match(r'^[\d\-\:\.\s]+$', cell_norm) and len_cell < 12:
|
|
|
+ score_subseq = 0.0
|
|
|
|
|
|
- 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 []
|
|
|
+ # --- 📊 综合评分 ---
|
|
|
+ final_score = max(score_sort, score_partial, score_subseq)
|
|
|
|
|
|
- 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]
|
|
|
+ # 精确匹配奖励
|
|
|
+ if cell_norm == box_norm:
|
|
|
+ final_score = 100.0
|
|
|
+ elif cell_norm in box_norm:
|
|
|
+ final_score = min(100, final_score + 5)
|
|
|
|
|
|
- # 🔑 改进：检查是否有重叠（不只是中心点）
|
|
|
- overlap = not (box_x_start > x_end or box_x_end < x_start)
|
|
|
-
|
|
|
- if overlap:
|
|
|
- col_boxes.append(box)
|
|
|
-
|
|
|
- return col_boxes
|
|
|
+ return final_score
|
|
|
|
|
|
|
|
|
def _filter_boxes_in_table_region(self, paddle_boxes: List[Dict],
|
|
|
@@ -942,232 +907,3 @@ class TableCellMatcher:
|
|
|
|
|
|
final_score = (match_rate * 100) - penalty
|
|
|
return max(0, final_score)
|
|
|
-
|
|
|
- def _match_cell_sequential(self, cell_text: str,
|
|
|
- boxes: List[Dict],
|
|
|
- col_boundaries: List[Tuple[int, int]],
|
|
|
- start_idx: int) -> Optional[Dict]:
|
|
|
- """
|
|
|
- 🎯 顺序匹配单元格：从指定位置开始，逐步合并 boxes 直到匹配
|
|
|
- """
|
|
|
- cell_text_normalized = self.text_matcher.normalize_text(cell_text)
|
|
|
- cell_text_processed = self._preprocess_text_for_matching(cell_text)
|
|
|
-
|
|
|
- if len(cell_text_normalized) < 1:
|
|
|
- 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:]:
|
|
|
- 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[first_unused_idx:] 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]))
|
|
|
- # 🎯 改进：使用空格连接，以便于 token_sort_ratio 进行乱序匹配
|
|
|
- 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. 模糊匹配
|
|
|
- # 🎯 改进：使用预处理后的文本进行 token_sort_ratio 计算
|
|
|
- box_text_processed = self._preprocess_text_for_matching(box['text'])
|
|
|
-
|
|
|
- # token_sort_ratio: 自动分词并排序比较，解决 OCR 结果顺序与 HTML 不一致的问题
|
|
|
- token_sort_sim = fuzz.token_sort_ratio(cell_text_processed, box_text_processed)
|
|
|
-
|
|
|
- # partial_ratio: 子串模糊匹配，解决 OCR 识别错误
|
|
|
- partial_sim = fuzz.partial_ratio(cell_text_normalized, merged_text_normalized)
|
|
|
-
|
|
|
- # 🛡️ 增强版防御：防止“短文本”误匹配“长文本”
|
|
|
- if partial_sim > 80:
|
|
|
- len_cell = len(cell_text_normalized)
|
|
|
- len_box = len(merged_text_normalized)
|
|
|
-
|
|
|
- # 确定短方和长方
|
|
|
- if len_cell < len_box:
|
|
|
- len_short, len_long = len_cell, len_box
|
|
|
- text_short = cell_text_normalized
|
|
|
- text_long = merged_text_normalized
|
|
|
- else:
|
|
|
- len_short, len_long = len_box, len_cell
|
|
|
- text_short = merged_text_normalized
|
|
|
- text_long = cell_text_normalized
|
|
|
-
|
|
|
- # 🎯 修正：检测有效内容 (字母、数字、汉字)
|
|
|
- # 使用 Unicode 范围匹配汉字: \u4e00-\u9fa5
|
|
|
- import re
|
|
|
- def has_valid_content(text):
|
|
|
- return bool(re.search(r'[a-zA-Z0-9\u4e00-\u9fa5]', text))
|
|
|
-
|
|
|
- short_has_content = has_valid_content(text_short)
|
|
|
- long_has_content = has_valid_content(text_long)
|
|
|
-
|
|
|
- # 🛑 拒绝条件 1: 短方是纯符号 (无有效内容)，且长方有内容
|
|
|
- # 例如: Cell="-" vs Box="-200" (拦截)
|
|
|
- # 例如: Cell="中国银行" vs Box="中国银行储蓄卡" (不拦截，因为都有汉字)
|
|
|
- if not short_has_content and long_has_content:
|
|
|
- # 允许例外：如果长方也很短 (比如 Cell="-" Box="- ")，可能只是多了个空格，不拦截
|
|
|
- if len_long > len_short + 2:
|
|
|
- print(f" ⚠️ 拒绝纯符号部分匹配: '{cell_text}' vs '{merged_text_normalized}'")
|
|
|
- partial_sim = 0.0
|
|
|
-
|
|
|
- # 🛑 拒绝条件 2: 短方虽然有内容，但太短了 (信息量不足)
|
|
|
- elif short_has_content:
|
|
|
- # 如果短方只有 1 个字符，且长方超过 3 个字符 -> 拒绝
|
|
|
- if len_short == 1 and len_long > 3:
|
|
|
- print(f" ⚠️ 拒绝单字符部分匹配: '{cell_text}' vs '{merged_text_normalized}'")
|
|
|
- partial_sim = 0.0
|
|
|
- # 如果短方只有 2 个字符，且长方超过 8 个字符 -> 拒绝
|
|
|
- elif len_short == 2 and len_long > 8:
|
|
|
- print(f" ⚠️ 拒绝微小碎片部分匹配: '{cell_text}' vs '{merged_text_normalized}'")
|
|
|
- partial_sim = 0.0
|
|
|
-
|
|
|
- # 🆕 新增条件 3: 覆盖率过低 (防止 "2024" 匹配 "ID2024...")
|
|
|
- # 场景: Cell 是长文本, Box 是短文本, 恰好包含在 Cell 中
|
|
|
- # 逻辑: 如果覆盖率 < 30% 且 整体相似度(token_sort) < 45，说明 Box 缺失了 Cell 的绝大部分内容
|
|
|
- else:
|
|
|
- coverage = len_short / len_long if len_long > 0 else 0
|
|
|
- if coverage < 0.3 and token_sort_sim < 45:
|
|
|
- print(f" ⚠️ 拒绝低覆盖率部分匹配: '{text_short}' in '{text_long}' (cov={coverage:.2f})")
|
|
|
- partial_sim = 0.0
|
|
|
-
|
|
|
- # 🎯 新增：token_set_ratio (集合匹配)
|
|
|
- # 专门解决：目标文本被 OCR 文本中的噪音隔开的情况
|
|
|
- # 例如 Target="A B", OCR="A noise B" -> token_set_ratio 会很高
|
|
|
- token_set_sim = fuzz.token_set_ratio(cell_text_processed, box_text_processed)
|
|
|
-
|
|
|
- # 🎯 策略 4: 重构匹配 (Reconstruction Match) - 解决 ID 被噪音打断的问题
|
|
|
- # 逻辑：提取 OCR 中所有属于 Target 子串的 token，拼起来再比
|
|
|
- reconstruct_sim = 0.0
|
|
|
- if len(cell_text_normalized) > 10: # 仅对长文本启用，防止短文本误判
|
|
|
- # 使用预处理后的文本分词 (已处理中文/数字间隔)
|
|
|
- box_tokens = box_text_processed.split()
|
|
|
- # 筛选出所有是目标文本子串的 token
|
|
|
- valid_tokens = []
|
|
|
- for token in box_tokens:
|
|
|
- # 忽略太短的 token (除非目标也很短)，防止 "1" 这种误匹配
|
|
|
- if len(token) < 2 and len(cell_text_normalized) > 5:
|
|
|
- continue
|
|
|
- if token in cell_text_normalized:
|
|
|
- valid_tokens.append(token)
|
|
|
-
|
|
|
- if valid_tokens:
|
|
|
- # 拼接回原始形态
|
|
|
- reconstructed_text = "".join(valid_tokens)
|
|
|
- reconstruct_sim = fuzz.ratio(cell_text_normalized, reconstructed_text)
|
|
|
- if reconstruct_sim > 90:
|
|
|
- print(f" 🧩 重构匹配生效: '{reconstructed_text}' (sim={reconstruct_sim})")
|
|
|
-
|
|
|
- # 🎯 策略 5: 子序列匹配 (Subsequence Match) - 解决粘连噪音问题
|
|
|
- # 专门针对: '1544...1050' + '2024-08-10' + '0433...' 这种场景
|
|
|
- subseq_sim = 0.0
|
|
|
- if len(cell_text_normalized) > 8: # 仅对较长文本启用
|
|
|
- subseq_sim = self._calculate_subsequence_score(cell_text_normalized, merged_text_normalized)
|
|
|
- # 🛡️ 关键修复：长度和类型防御
|
|
|
- if subseq_sim > 80:
|
|
|
- len_cell = len(cell_text_normalized)
|
|
|
- len_box = len(merged_text_normalized)
|
|
|
-
|
|
|
- # 1. 长度差异过大 (Box 比 Cell 长很多)
|
|
|
- if len_box > len_cell * 1.5:
|
|
|
- # 2. 且 Cell 是数字/日期/时间类型
|
|
|
- import re
|
|
|
- if re.match(r'^[\d\-\:\.\s]+$', cell_text_normalized):
|
|
|
- # 🧠 智能豁免：如果 Cell 本身很长 (例如 > 12字符)，说明是长ID
|
|
|
- # 长ID即使夹杂了噪音 (如 "ID...日期...文字")，只要子序列匹配高，通常也是对的
|
|
|
- # 只有短文本 (如 "2024") 才需要严格防御
|
|
|
- if len_cell < 12:
|
|
|
- print(f" ⚠️ 拒绝子序列匹配: 长度差异大且为短数字类型 (sim={subseq_sim})")
|
|
|
- subseq_sim = 0.0
|
|
|
- else:
|
|
|
- print(f" ✅ 接受长ID子序列匹配: 尽管长度差异大，但特征显著 (len={len_cell})")
|
|
|
-
|
|
|
- if subseq_sim > 90:
|
|
|
- print(f" 🔗 子序列匹配生效: '{cell_text[:10]}...' (sim={subseq_sim:.1f})")
|
|
|
-
|
|
|
- # 综合得分：取五者最大值
|
|
|
- similarity = max(token_sort_sim, partial_sim, token_set_sim, reconstruct_sim, subseq_sim)
|
|
|
-
|
|
|
- # 🎯 子串匹配加分
|
|
|
- if is_substring:
|
|
|
- similarity = min(100, similarity + 10)
|
|
|
-
|
|
|
- # 🎯 长度惩罚：如果 box 内容比 cell 多太多（例如吞了下一个单元格），扣分
|
|
|
- # 注意：token_set_ratio 对长度不敏感，所以这里必须严格检查长度，防止误判
|
|
|
- # 只有当 similarity 很高时才检查，防止误杀
|
|
|
- if similarity > 80:
|
|
|
- len_cell = len(cell_text_normalized)
|
|
|
- len_box = len(merged_text_normalized)
|
|
|
-
|
|
|
- # 如果是 token_set_sim 贡献的高分，说明 OCR 里包含了很多噪音
|
|
|
- # 我们需要确保这些噪音不是“下一个单元格的内容”
|
|
|
- # 这里可以加一个更严格的长度检查，或者检查是否包含换行符等
|
|
|
- if len_box > len_cell * 2.0 + 10: # 放宽一点，因为 token_set 本来就是处理噪音的
|
|
|
- similarity -= 10 # 稍微扣一点分，表示虽然全找到了，但噪音太多不太完美
|
|
|
-
|
|
|
- if similarity >= self.text_matcher.similarity_threshold:
|
|
|
- print(f" ✓ 匹配成功: '{cell_text[:15]}' vs '{box['text'][:15]}' (相似度: {similarity})")
|
|
|
- # 由于是模糊匹配，返回第一个未使用的 box 作为 last_index
|
|
|
- for b in boxes:
|
|
|
- if not b.get('used'):
|
|
|
- last_idx = max(boxes.index(b)-1, 0)
|
|
|
- break
|
|
|
- return self._build_match_result(box['sorted_combo'], box['text'], similarity, max(start_idx, last_idx))
|
|
|
-
|
|
|
- print(f" ✗ 匹配失败: '{cell_text[:15]}'")
|
|
|
- return None
|
|
|
-
|
|
|
- def _build_match_result(self, boxes: List[Dict], text: str,
|
|
|
- score: float, last_index: int) -> Dict:
|
|
|
- """构建匹配结果（使用原始坐标）"""
|
|
|
-
|
|
|
- # 🔑 关键修复：使用 original_bbox（如果存在）
|
|
|
- def get_original_bbox(box: Dict) -> List[int]:
|
|
|
- return box.get('original_bbox', box['bbox'])
|
|
|
-
|
|
|
- original_bboxes = [get_original_bbox(b) for b in boxes]
|
|
|
-
|
|
|
- merged_bbox = [
|
|
|
- min(b[0] for b in original_bboxes),
|
|
|
- min(b[1] for b in original_bboxes),
|
|
|
- max(b[2] for b in original_bboxes),
|
|
|
- max(b[3] for b in original_bboxes)
|
|
|
- ]
|
|
|
-
|
|
|
- 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
|
|
|
- }
|
