Browse Source

feat: add core modules for document processing including coordinate utilities, element processors, layout utilities, and PDF handling, enhancing overall functionality and structure

zhch158_admin 20 hours ago
parent
commit
57c61906b5

+ 594 - 0
zhch/universal_doc_parser/core/coordinate_utils.py

@@ -0,0 +1,594 @@
+"""
+坐标转换工具模块
+
+提供各种坐标转换功能:
+- 相对坐标 → 绝对坐标转换
+- OCR 格式转换
+- 旋转坐标逆变换
+- HTML data-bbox 坐标转换
+"""
+import re
+import json
+from typing import Dict, List, Any, Optional, Tuple
+import numpy as np
+from loguru import logger
+
+# 导入 merger 组件
+try:
+    from merger import BBoxExtractor
+    MERGER_AVAILABLE = True
+except ImportError:
+    MERGER_AVAILABLE = False
+    BBoxExtractor = None
+
+
+class CoordinateUtils:
+    """坐标转换工具类"""
+    
+    @staticmethod
+    def crop_region(image: np.ndarray, bbox: List[float]) -> np.ndarray:
+        """
+        裁剪图像区域
+        
+        Args:
+            image: 原始图像
+            bbox: 裁剪区域 [x1, y1, x2, y2]
+            
+        Returns:
+            裁剪后的图像
+        """
+        if len(bbox) < 4:
+            return image
+        
+        x1, y1, x2, y2 = map(int, bbox[:4])
+        h, w = image.shape[:2]
+        
+        x1 = max(0, min(x1, w))
+        y1 = max(0, min(y1, h))
+        x2 = max(x1, min(x2, w))
+        y2 = max(y1, min(y2, h))
+        
+        return image[y1:y2, x1:x2]
+    
+    @staticmethod
+    def bbox_overlap(bbox1: List[float], bbox2: List[float]) -> bool:
+        """
+        检查两个 bbox 是否重叠
+        
+        Args:
+            bbox1: 第一个 bbox [x1, y1, x2, y2]
+            bbox2: 第二个 bbox [x1, y1, x2, y2]
+            
+        Returns:
+            是否重叠
+        """
+        if len(bbox1) < 4 or len(bbox2) < 4:
+            return False
+        
+        x1_1, y1_1, x2_1, y2_1 = bbox1[:4]
+        x1_2, y1_2, x2_2, y2_2 = bbox2[:4]
+        
+        if x2_1 < x1_2 or x2_2 < x1_1:
+            return False
+        if y2_1 < y1_2 or y2_2 < y1_1:
+            return False
+        
+        return True
+    
+    @staticmethod
+    def convert_to_absolute_coords(
+        relative_bbox: List, 
+        region_bbox: List[float]
+    ) -> List:
+        """
+        将相对坐标转换为绝对坐标
+        
+        Args:
+            relative_bbox: 相对坐标
+            region_bbox: 区域的绝对坐标 [x1, y1, x2, y2]
+            
+        Returns:
+            绝对坐标
+        """
+        if not relative_bbox or len(region_bbox) < 4:
+            return relative_bbox
+        
+        bx1, by1 = region_bbox[0], region_bbox[1]
+        
+        # 处理4点坐标格式 [[x1,y1], [x2,y2], [x3,y3], [x4,y4]]
+        if isinstance(relative_bbox[0], (list, tuple)):
+            return [
+                [p[0] + bx1, p[1] + by1] for p in relative_bbox
+            ]
+        
+        # 处理4值坐标格式 [x1, y1, x2, y2]
+        if len(relative_bbox) >= 4:
+            return [
+                relative_bbox[0] + bx1,
+                relative_bbox[1] + by1,
+                relative_bbox[2] + bx1,
+                relative_bbox[3] + by1
+            ]
+        
+        return relative_bbox
+    
+    @staticmethod
+    def convert_ocr_to_matcher_format(
+        ocr_poly: List,
+        text: str,
+        confidence: float,
+        idx: int,
+        table_bbox: Optional[List[float]] = None
+    ) -> Optional[Dict[str, Any]]:
+        """
+        将 OCR 结果转换为 TableCellMatcher 期望的格式
+        
+        OCR 返回格式:
+            bbox: [[x1,y1], [x2,y1], [x2,y2], [x1,y2]]  # 4点多边形
+            text: str
+            confidence: float
+        
+        TableCellMatcher 期望格式:
+            text: str
+            bbox: [x_min, y_min, x_max, y_max]  # 4值矩形
+            poly: [[x1,y1], [x2,y1], [x2,y2], [x1,y2]]  # 4点多边形
+            score: float
+            paddle_bbox_index: int
+            used: bool
+        
+        Args:
+            ocr_poly: OCR返回的4点多边形坐标
+            text: 识别文本
+            confidence: 置信度
+            idx: 索引
+            table_bbox: 表格的绝对坐标,用于转换相对坐标
+        
+        Returns:
+            转换后的字典,或 None(如果无效)
+        """
+        if not ocr_poly or not text:
+            return None
+        
+        poly = []
+        
+        # 格式1: [[x1,y1], [x2,y2], [x3,y3], [x4,y4]] - 4点多边形
+        if isinstance(ocr_poly[0], (list, tuple)) and len(ocr_poly) == 4:
+            poly = [[float(p[0]), float(p[1])] for p in ocr_poly]
+        
+        # 格式2: [x1, y1, x2, y1, x2, y2, x1, y2] - 8值展平格式
+        elif len(ocr_poly) == 8 and isinstance(ocr_poly[0], (int, float)):
+            poly = [
+                [float(ocr_poly[0]), float(ocr_poly[1])],
+                [float(ocr_poly[2]), float(ocr_poly[3])],
+                [float(ocr_poly[4]), float(ocr_poly[5])],
+                [float(ocr_poly[6]), float(ocr_poly[7])]
+            ]
+        
+        # 格式3: [x1, y1, x2, y2] - 4值矩形格式
+        elif len(ocr_poly) == 4 and isinstance(ocr_poly[0], (int, float)):
+            x1, y1, x2, y2 = ocr_poly
+            poly = [
+                [float(x1), float(y1)],
+                [float(x2), float(y1)],
+                [float(x2), float(y2)],
+                [float(x1), float(y2)]
+            ]
+        else:
+            logger.warning(f"Unknown OCR bbox format: {ocr_poly}")
+            return None
+        
+        # 转换为绝对坐标(相对于整页图片)
+        if table_bbox and len(table_bbox) >= 2:
+            offset_x, offset_y = table_bbox[0], table_bbox[1]
+            poly = [[p[0] + offset_x, p[1] + offset_y] for p in poly]
+        
+        # 从多边形计算 bbox [x_min, y_min, x_max, y_max]
+        xs = [p[0] for p in poly]
+        ys = [p[1] for p in poly]
+        bbox = [min(xs), min(ys), max(xs), max(ys)]
+        
+        return {
+            'text': text,
+            'bbox': bbox,
+            'poly': poly,
+            'score': confidence,
+            'paddle_bbox_index': idx,
+            'used': False
+        }
+    
+    @staticmethod
+    def inverse_rotate_table_coords(
+        cells: List[Dict],
+        html: str,
+        rotation_angle: float,
+        orig_table_size: Tuple[int, int],
+        table_bbox: List[float]
+    ) -> Tuple[List[Dict], str]:
+        """
+        将旋转后的坐标逆向转换回原图坐标
+        
+        Args:
+            cells: 单元格列表(坐标是旋转后的)
+            html: HTML字符串(data-bbox是旋转后的)
+            rotation_angle: 旋转角度
+            orig_table_size: 原始表格尺寸 (width, height)
+            table_bbox: 表格在整页图片中的位置 [x1, y1, x2, y2]
+        
+        Returns:
+            (转换后的cells, 转换后的html)
+        """
+        if not MERGER_AVAILABLE or BBoxExtractor is None:
+            # 如果 merger 不可用,只添加偏移量
+            converted_cells = CoordinateUtils.add_table_offset_to_cells(cells, table_bbox)
+            converted_html = CoordinateUtils.add_table_offset_to_html(html, table_bbox)
+            return converted_cells, converted_html
+        
+        table_offset_x, table_offset_y = table_bbox[0], table_bbox[1]
+        
+        # 转换 cells 中的 bbox
+        converted_cells = []
+        for cell in cells:
+            cell_copy = cell.copy()
+            cell_bbox = cell.get('bbox', [])
+            if cell_bbox and len(cell_bbox) == 4:
+                # 先逆向旋转,再加上表格偏移量
+                orig_bbox = BBoxExtractor.inverse_rotate_box_coordinates(
+                    cell_bbox, rotation_angle, orig_table_size
+                )
+                # 加上表格偏移量转换为整页坐标
+                cell_copy['bbox'] = [
+                    orig_bbox[0] + table_offset_x,
+                    orig_bbox[1] + table_offset_y,
+                    orig_bbox[2] + table_offset_x,
+                    orig_bbox[3] + table_offset_y
+                ]
+            converted_cells.append(cell_copy)
+        
+        # 转换 HTML 中的 data-bbox
+        def replace_bbox(match):
+            bbox_str = match.group(1)
+            try:
+                bbox = json.loads(bbox_str)
+                if len(bbox) == 4:
+                    orig_bbox = BBoxExtractor.inverse_rotate_box_coordinates(
+                        bbox, rotation_angle, orig_table_size
+                    )
+                    new_bbox = [
+                        orig_bbox[0] + table_offset_x,
+                        orig_bbox[1] + table_offset_y,
+                        orig_bbox[2] + table_offset_x,
+                        orig_bbox[3] + table_offset_y
+                    ]
+                    return f'data-bbox="[{new_bbox[0]},{new_bbox[1]},{new_bbox[2]},{new_bbox[3]}]"'
+            except:
+                pass
+            return match.group(0)
+        
+        converted_html = re.sub(
+            r'data-bbox="\[([^\]]+)\]"',
+            replace_bbox,
+            html
+        )
+        
+        return converted_cells, converted_html
+    
+    @staticmethod
+    def add_table_offset_to_cells(
+        cells: List[Dict],
+        table_bbox: List[float]
+    ) -> List[Dict]:
+        """
+        为单元格坐标添加表格偏移量(无旋转情况)
+        
+        Args:
+            cells: 单元格列表
+            table_bbox: 表格位置 [x1, y1, x2, y2]
+        
+        Returns:
+            转换后的 cells
+        """
+        offset_x, offset_y = table_bbox[0], table_bbox[1]
+        
+        converted_cells = []
+        for cell in cells:
+            cell_copy = cell.copy()
+            cell_bbox = cell.get('bbox', [])
+            if cell_bbox and len(cell_bbox) == 4:
+                cell_copy['bbox'] = [
+                    cell_bbox[0] + offset_x,
+                    cell_bbox[1] + offset_y,
+                    cell_bbox[2] + offset_x,
+                    cell_bbox[3] + offset_y
+                ]
+            converted_cells.append(cell_copy)
+        
+        return converted_cells
+    
+    @staticmethod
+    def add_table_offset_to_html(
+        html: str,
+        table_bbox: List[float]
+    ) -> str:
+        """
+        为HTML中的data-bbox添加表格偏移量(无旋转情况)
+        
+        Args:
+            html: HTML字符串
+            table_bbox: 表格位置 [x1, y1, x2, y2]
+        
+        Returns:
+            转换后的 HTML
+        """
+        offset_x, offset_y = table_bbox[0], table_bbox[1]
+        
+        def replace_bbox(match):
+            bbox_str = match.group(1)
+            try:
+                bbox = json.loads(f"[{bbox_str}]")
+                if len(bbox) == 4:
+                    new_bbox = [
+                        bbox[0] + offset_x,
+                        bbox[1] + offset_y,
+                        bbox[2] + offset_x,
+                        bbox[3] + offset_y
+                    ]
+                    return f'data-bbox="[{new_bbox[0]},{new_bbox[1]},{new_bbox[2]},{new_bbox[3]}]"'
+            except:
+                pass
+            return match.group(0)
+        
+        converted_html = re.sub(
+            r'data-bbox="\[([^\]]+)\]"',
+            replace_bbox,
+            html
+        )
+        
+        return converted_html
+    
+    @staticmethod
+    def add_table_offset_to_ocr_boxes(
+        ocr_boxes: List[Dict],
+        table_bbox: List[float]
+    ) -> List[Dict]:
+        """
+        为 ocr_boxes 添加表格偏移量,将相对坐标转换为页面绝对坐标
+        
+        Args:
+            ocr_boxes: OCR 框列表
+            table_bbox: 表格在页面中的位置 [x1, y1, x2, y2]
+            
+        Returns:
+            转换后的 ocr_boxes
+        """
+        if not ocr_boxes or not table_bbox:
+            return ocr_boxes
+        
+        offset_x = table_bbox[0]
+        offset_y = table_bbox[1]
+        
+        converted_boxes = []
+        for box in ocr_boxes:
+            new_box = box.copy()
+            
+            # 转换 bbox [x1, y1, x2, y2]
+            if 'bbox' in new_box and new_box['bbox']:
+                bbox = new_box['bbox']
+                if len(bbox) >= 4:
+                    new_box['bbox'] = [
+                        bbox[0] + offset_x,
+                        bbox[1] + offset_y,
+                        bbox[2] + offset_x,
+                        bbox[3] + offset_y
+                    ]
+            
+            # 转换 poly [[x1,y1], [x2,y2], [x3,y3], [x4,y4]]
+            if 'poly' in new_box and new_box['poly']:
+                poly = new_box['poly']
+                new_poly = []
+                for point in poly:
+                    if isinstance(point, (list, tuple)) and len(point) >= 2:
+                        new_poly.append([point[0] + offset_x, point[1] + offset_y])
+                    else:
+                        new_poly.append(point)
+                new_box['poly'] = new_poly
+            
+            converted_boxes.append(new_box)
+        
+        return converted_boxes
+    
+    @staticmethod
+    def inverse_rotate_ocr_boxes(
+        ocr_boxes: List[Dict],
+        rotation_angle: float,
+        orig_table_size: Tuple[int, int],
+        table_bbox: List[float]
+    ) -> List[Dict]:
+        """
+        对 ocr_boxes 进行逆向旋转并添加表格偏移量
+        
+        Args:
+            ocr_boxes: OCR 框列表
+            rotation_angle: 表格旋转角度
+            orig_table_size: 原始表格尺寸 (width, height)
+            table_bbox: 表格在页面中的位置
+            
+        Returns:
+            转换后的 ocr_boxes
+        """
+        if not ocr_boxes:
+            return ocr_boxes
+        
+        if not MERGER_AVAILABLE or BBoxExtractor is None:
+            return CoordinateUtils.add_table_offset_to_ocr_boxes(ocr_boxes, table_bbox)
+        
+        offset_x = table_bbox[0]
+        offset_y = table_bbox[1]
+        
+        converted_boxes = []
+        for box in ocr_boxes:
+            new_box = box.copy()
+            
+            # 逆向旋转 bbox
+            if 'bbox' in new_box and new_box['bbox']:
+                bbox = new_box['bbox']
+                if len(bbox) >= 4:
+                    try:
+                        rotated_bbox = BBoxExtractor.inverse_rotate_box_coordinates(
+                            bbox, rotation_angle, orig_table_size
+                        )
+                        new_box['bbox'] = [
+                            rotated_bbox[0] + offset_x,
+                            rotated_bbox[1] + offset_y,
+                            rotated_bbox[2] + offset_x,
+                            rotated_bbox[3] + offset_y
+                        ]
+                    except Exception as e:
+                        logger.debug(f"Failed to inverse rotate ocr_box bbox: {e}")
+            
+            # 逆向旋转 poly
+            if 'poly' in new_box and new_box['poly']:
+                poly = new_box['poly']
+                try:
+                    poly_list = [[float(p[0]), float(p[1])] for p in poly]
+                    rotated_poly = BBoxExtractor.inverse_rotate_coordinates(
+                        poly_list, rotation_angle, orig_table_size
+                    )
+                    new_poly = []
+                    for point in rotated_poly:
+                        new_poly.append([point[0] + offset_x, point[1] + offset_y])
+                    new_box['poly'] = new_poly
+                except Exception as e:
+                    logger.debug(f"Failed to inverse rotate ocr_box poly: {e}")
+            
+            converted_boxes.append(new_box)
+        
+        return converted_boxes
+    
+    @staticmethod
+    def is_poly_format(bbox: Any) -> bool:
+        """
+        检测 bbox 是否为四点多边形格式
+        
+        四点格式: [[x1,y1], [x2,y2], [x3,y3], [x4,y4]]
+        矩形格式: [x_min, y_min, x_max, y_max]
+        """
+        if not bbox or not isinstance(bbox, list):
+            return False
+        if len(bbox) != 4:
+            return False
+        return isinstance(bbox[0], (list, tuple))
+    
+    @staticmethod
+    def transform_coords_to_original(
+        element: Dict[str, Any],
+        rotate_angle: int,
+        rotated_shape: Tuple,
+        original_shape: Tuple
+    ) -> Dict[str, Any]:
+        """
+        将坐标从旋转后的图片坐标系转换回原始图片坐标系
+        
+        Args:
+            element: 元素字典(包含bbox等坐标信息)
+            rotate_angle: 旋转角度(0, 90, 180, 270)
+            rotated_shape: 旋转后图片的shape (h, w, c)
+            original_shape: 原始图片的shape (h, w, c)
+        
+        Returns:
+            坐标转换后的元素字典(深拷贝)
+        """
+        import copy
+        element = copy.deepcopy(element)
+        
+        if rotate_angle == 0 or not MERGER_AVAILABLE or BBoxExtractor is None:
+            return element
+        
+        # 原始图片尺寸 (width, height)
+        orig_h, orig_w = original_shape[:2]
+        orig_image_size = (orig_w, orig_h)
+        
+        # 转换主bbox
+        if 'bbox' in element and element['bbox']:
+            element['bbox'] = BBoxExtractor.inverse_rotate_box_coordinates(
+                element['bbox'], rotate_angle, orig_image_size
+            )
+        
+        # 转换表格相关坐标
+        if element.get('type') == 'table' and 'content' in element:
+            content = element['content']
+            
+            # 转换 OCR boxes
+            if 'ocr_boxes' in content and content['ocr_boxes']:
+                for box in content['ocr_boxes']:
+                    if 'bbox' in box and box['bbox']:
+                        box['bbox'] = BBoxExtractor.inverse_rotate_box_coordinates(
+                            box['bbox'], rotate_angle, orig_image_size
+                        )
+            
+            # 转换 cells
+            if 'cells' in content and content['cells']:
+                for cell in content['cells']:
+                    if 'bbox' in cell and cell['bbox']:
+                        cell['bbox'] = BBoxExtractor.inverse_rotate_box_coordinates(
+                            cell.get('bbox', []), rotate_angle, orig_image_size
+                        )
+            
+            # 转换 HTML 中的 data-bbox 属性
+            if 'html' in content and content['html']:
+                content['html'] = CoordinateUtils.transform_html_data_bbox(
+                    content['html'], rotate_angle, orig_image_size
+                )
+        
+        # 转换文本OCR details
+        if 'content' in element and 'ocr_details' in element.get('content', {}):
+            ocr_details = element['content'].get('ocr_details', [])
+            if ocr_details:
+                for detail in ocr_details:
+                    if 'bbox' in detail and detail['bbox']:
+                        if CoordinateUtils.is_poly_format(detail['bbox']):
+                            detail['bbox'] = BBoxExtractor.inverse_rotate_coordinates(
+                                detail['bbox'], rotate_angle, orig_image_size
+                            )
+                        else:
+                            detail['bbox'] = BBoxExtractor.inverse_rotate_box_coordinates(
+                                detail.get('bbox', []), rotate_angle, orig_image_size
+                            )
+        
+        return element
+    
+    @staticmethod
+    def transform_html_data_bbox(
+        html: str,
+        rotate_angle: int,
+        orig_image_size: Tuple[int, int]
+    ) -> str:
+        """
+        转换 HTML 中所有 data-bbox 属性的坐标
+        
+        Args:
+            html: 包含 data-bbox 属性的 HTML 字符串
+            rotate_angle: 旋转角度
+            orig_image_size: 原始图片尺寸 (width, height)
+        
+        Returns:
+            转换后的 HTML 字符串
+        """
+        if not MERGER_AVAILABLE or BBoxExtractor is None:
+            return html
+        
+        def replace_bbox(match):
+            try:
+                bbox_str = match.group(1)
+                bbox = json.loads(bbox_str)
+                if bbox and len(bbox) == 4:
+                    transformed = BBoxExtractor.inverse_rotate_box_coordinates(
+                        bbox, rotate_angle, orig_image_size
+                    )
+                    return f'data-bbox="{json.dumps(transformed)}"'
+            except (json.JSONDecodeError, ValueError):
+                pass
+            return match.group(0)
+        
+        pattern = r'data-bbox="(\[[^\]]+\])"'
+        return re.sub(pattern, replace_bbox, html)
+

+ 485 - 0
zhch/universal_doc_parser/core/element_processors.py

@@ -0,0 +1,485 @@
+"""
+元素处理器模块
+
+提供各类文档元素的处理功能:
+- 文本元素处理
+- 表格元素处理
+- 图片元素处理
+- 公式元素处理
+- 代码元素处理
+- 丢弃元素处理
+"""
+from typing import Dict, List, Any, Optional, Tuple
+import numpy as np
+from loguru import logger
+
+from .coordinate_utils import CoordinateUtils
+from .pdf_utils import PDFUtils
+
+# 导入 merger 组件
+try:
+    from merger import TableCellMatcher, TextMatcher, BBoxExtractor
+    MERGER_AVAILABLE = True
+except ImportError:
+    MERGER_AVAILABLE = False
+    TableCellMatcher = None
+    TextMatcher = None
+    BBoxExtractor = None
+
+
+class ElementProcessors:
+    """元素处理器类"""
+    
+    def __init__(
+        self,
+        preprocessor: Any,
+        ocr_recognizer: Any,
+        vl_recognizer: Any,
+        table_cell_matcher: Optional[Any] = None
+    ):
+        """
+        初始化元素处理器
+        
+        Args:
+            preprocessor: 预处理器(方向检测)
+            ocr_recognizer: OCR识别器
+            vl_recognizer: VL识别器(表格、公式)
+            table_cell_matcher: 表格单元格匹配器
+        """
+        self.preprocessor = preprocessor
+        self.ocr_recognizer = ocr_recognizer
+        self.vl_recognizer = vl_recognizer
+        self.table_cell_matcher = table_cell_matcher
+    
+    def _convert_ocr_details_to_absolute(
+        self,
+        ocr_details: List[Dict[str, Any]],
+        region_bbox: List[float]
+    ) -> List[Dict[str, Any]]:
+        """
+        将 OCR 详情中的相对坐标转换为绝对坐标
+        
+        Args:
+            ocr_details: OCR 结果列表,每项包含 'bbox' 字段
+            region_bbox: 裁剪区域的绝对坐标 [x1, y1, x2, y2]
+            
+        Returns:
+            坐标转换后的 OCR 详情列表
+        """
+        if not ocr_details or not region_bbox or len(region_bbox) < 2:
+            return ocr_details
+        
+        converted_details = []
+        for item in ocr_details:
+            new_item = item.copy()
+            ocr_bbox = item.get('bbox', [])
+            if ocr_bbox:
+                new_item['bbox'] = CoordinateUtils.convert_to_absolute_coords(
+                    ocr_bbox, region_bbox
+                )
+            converted_details.append(new_item)
+        
+        return converted_details
+    
+    def process_text_element(
+        self,
+        image: np.ndarray,
+        layout_item: Dict[str, Any],
+        pdf_type: str,
+        pdf_doc: Optional[Any],
+        page_idx: int,
+        scale: float
+    ) -> Dict[str, Any]:
+        """
+        处理文本元素
+        
+        - 扫描件:OCR检测+识别
+        - 数字PDF:尝试PDF字符提取,失败则OCR补充
+        
+        Args:
+            image: 页面图像
+            layout_item: 布局检测项
+            pdf_type: PDF类型 ('ocr' 或 'txt')
+            pdf_doc: PDF文档对象
+            page_idx: 页码索引
+            scale: 缩放比例
+            
+        Returns:
+            处理后的元素字典
+        """
+        bbox = layout_item.get('bbox', [0, 0, 0, 0])
+        cropped_image = CoordinateUtils.crop_region(image, bbox)
+        
+        text_content = ""
+        ocr_details = []
+        extraction_method = "ocr"
+        
+        # 数字原生PDF:尝试直接提取文本
+        if pdf_type == 'txt' and pdf_doc is not None:
+            try:
+                text_content, extraction_success = PDFUtils.extract_text_from_pdf(
+                    pdf_doc, page_idx, bbox, scale
+                )
+                if extraction_success and text_content.strip():
+                    extraction_method = "pdf_extract"
+                    logger.debug(f"📝 Text extracted from PDF: '{text_content[:30]}...'")
+            except Exception as e:
+                logger.debug(f"PDF text extraction failed: {e}")
+        
+        # OCR识别(扫描件或PDF提取失败)
+        if extraction_method == "ocr" or not text_content.strip():
+            try:
+                ocr_results = self.ocr_recognizer.recognize_text(cropped_image)
+                if ocr_results:
+                    text_parts = [
+                        item['text'] for item in ocr_results 
+                        if item.get('confidence', 0) > 0.5
+                    ]
+                    text_content = " ".join(text_parts)
+                    # 将 OCR 坐标转换为绝对坐标
+                    ocr_details = self._convert_ocr_details_to_absolute(ocr_results, bbox)
+                    extraction_method = "ocr"
+            except Exception as e:
+                logger.warning(f"OCR recognition failed: {e}")
+        
+        return {
+            'type': layout_item.get('category', 'text'),
+            'bbox': bbox,
+            'confidence': layout_item.get('confidence', 0.0),
+            'content': {
+                'text': text_content,
+                'ocr_details': ocr_details,
+                'extraction_method': extraction_method
+            }
+        }
+    
+    def process_table_element(
+        self,
+        image: np.ndarray,
+        layout_item: Dict[str, Any],
+        scale: float
+    ) -> Dict[str, Any]:
+        """
+        处理表格元素
+        
+        流程:
+        1. 表格方向检测
+        2. OCR检测获取文本框坐标(旋转后的图片)
+        3. VLM识别获取表格结构HTML
+        4. 匹配OCR坐标与VLM结构
+        5. 将坐标逆向转换回原图坐标
+        
+        Args:
+            image: 页面图像
+            layout_item: 布局检测项
+            scale: 缩放比例
+            
+        Returns:
+            处理后的元素字典
+        """
+        bbox = layout_item.get('bbox', [0, 0, 0, 0])
+        cropped_table = CoordinateUtils.crop_region(image, bbox)
+        
+        # 获取裁剪后表格图片的尺寸
+        orig_table_h, orig_table_w = cropped_table.shape[:2]
+        orig_table_size = (orig_table_w, orig_table_h)
+        
+        # 1. 表格方向检测
+        table_angle = 0
+        try:
+            rotated_table, table_angle = self.preprocessor.process(cropped_table)
+            if table_angle != 0:
+                logger.info(f"📐 Table rotated {table_angle}°")
+                cropped_table = rotated_table
+        except Exception as e:
+            logger.debug(f"Table orientation detection skipped: {e}")
+            table_angle = 0
+        
+        # 2. OCR检测获取文本框坐标
+        ocr_boxes = []
+        try:
+            ocr_results = self.ocr_recognizer.recognize_text(cropped_table)
+            if ocr_results:
+                for idx, item in enumerate(ocr_results):
+                    ocr_poly = item.get('bbox', [])
+                    if ocr_poly:
+                        formatted_box = CoordinateUtils.convert_ocr_to_matcher_format(
+                            ocr_poly, 
+                            item.get('text', ''),
+                            item.get('confidence', 0.0),
+                            idx,
+                            table_bbox=None
+                        )
+                        if formatted_box:
+                            ocr_boxes.append(formatted_box)
+            logger.info(f"📊 OCR detected {len(ocr_boxes)} text boxes in table")
+        except Exception as e:
+            logger.warning(f"Table OCR detection failed: {e}")
+        
+        # 3. VLM识别获取表格结构HTML
+        table_html = ""
+        try:
+            vl_result = self.vl_recognizer.recognize_table(
+                cropped_table,
+                return_cells_coordinate=True
+            )
+            table_html = vl_result.get('html', '')
+            logger.info(f"📊 VLM recognized table structure")
+        except Exception as e:
+            logger.warning(f"VLM table recognition failed: {e}")
+        
+        # 4. 匹配OCR坐标与VLM结构
+        cells = []
+        enhanced_html = table_html
+        skew_angle = 0.0
+        
+        if table_html and ocr_boxes and self.table_cell_matcher:
+            try:
+                enhanced_html, cells, _, skew_angle = self.table_cell_matcher.enhance_table_html_with_bbox(
+                    html=table_html,
+                    paddle_text_boxes=ocr_boxes,
+                    start_pointer=0,
+                    table_bbox=None
+                )
+                logger.info(f"📊 Matched {len(cells)} cells with coordinates (skew: {skew_angle:.2f}°)")
+            except Exception as e:
+                logger.warning(f"Cell coordinate matching failed: {e}")
+        
+        # 5. 坐标转换:将旋转后的坐标转换回原图坐标
+        if table_angle != 0 and MERGER_AVAILABLE:
+            cells, enhanced_html = CoordinateUtils.inverse_rotate_table_coords(
+                cells=cells,
+                html=enhanced_html,
+                rotation_angle=table_angle,
+                orig_table_size=orig_table_size,
+                table_bbox=bbox
+            )
+            ocr_boxes = CoordinateUtils.inverse_rotate_ocr_boxes(
+                ocr_boxes=ocr_boxes,
+                rotation_angle=table_angle,
+                orig_table_size=orig_table_size,
+                table_bbox=bbox
+            )
+            logger.info(f"📐 Coordinates transformed back to original image")
+        else:
+            # 没有旋转,只需要加上表格偏移量
+            cells = CoordinateUtils.add_table_offset_to_cells(cells, bbox)
+            enhanced_html = CoordinateUtils.add_table_offset_to_html(enhanced_html, bbox)
+            ocr_boxes = CoordinateUtils.add_table_offset_to_ocr_boxes(ocr_boxes, bbox)
+        
+        return {
+            'type': 'table',
+            'bbox': bbox,
+            'confidence': layout_item.get('confidence', 0.0),
+            'content': {
+                'html': enhanced_html,
+                'original_html': table_html,
+                'cells': cells,
+                'ocr_boxes': ocr_boxes,
+                'table_angle': table_angle,
+                'skew_angle': skew_angle
+            },
+        }
+    
+    def process_equation_element(
+        self,
+        image: np.ndarray,
+        layout_item: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """
+        处理公式元素
+        
+        Args:
+            image: 页面图像
+            layout_item: 布局检测项
+            
+        Returns:
+            处理后的元素字典
+        """
+        bbox = layout_item.get('bbox', [0, 0, 0, 0])
+        category = layout_item.get('category', '')
+        cropped_region = CoordinateUtils.crop_region(image, bbox)
+        
+        content = {'latex': '', 'confidence': 0.0}
+        
+        try:
+            formula_result = self.vl_recognizer.recognize_formula(cropped_region)
+            content = {
+                'latex': formula_result.get('latex', ''),
+                'confidence': formula_result.get('confidence', 0.0)
+            }
+        except Exception as e:
+            logger.warning(f"Formula recognition failed: {e}")
+        
+        return {
+            'type': category,
+            'bbox': bbox,
+            'confidence': layout_item.get('confidence', 0.0),
+            'content': content
+        }
+    
+    def process_image_element(
+        self,
+        image: np.ndarray,
+        layout_item: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """
+        处理图片元素
+        
+        Args:
+            image: 页面图像
+            layout_item: 布局检测项
+            
+        Returns:
+            处理后的元素字典
+        """
+        bbox = layout_item.get('bbox', [0, 0, 0, 0])
+        category = layout_item.get('category', '')
+        
+        cropped_image = CoordinateUtils.crop_region(image, bbox)
+        
+        return {
+            'type': category,
+            'bbox': bbox,
+            'confidence': layout_item.get('confidence', 0.0),
+            'content': {
+                'type': 'image',
+                'description': '',
+                'image_data': cropped_image,
+                'image_path': ''
+            }
+        }
+    
+    def process_code_element(
+        self,
+        image: np.ndarray,
+        layout_item: Dict[str, Any],
+        pdf_type: str,
+        pdf_doc: Optional[Any],
+        page_idx: int,
+        scale: float
+    ) -> Dict[str, Any]:
+        """
+        处理代码元素
+        
+        Args:
+            image: 页面图像
+            layout_item: 布局检测项
+            pdf_type: PDF类型
+            pdf_doc: PDF文档对象
+            page_idx: 页码索引
+            scale: 缩放比例
+            
+        Returns:
+            处理后的元素字典
+        """
+        bbox = layout_item.get('bbox', [0, 0, 0, 0])
+        category = layout_item.get('category', '')
+        cropped_image = CoordinateUtils.crop_region(image, bbox)
+        
+        code_content = ""
+        ocr_details = []
+        
+        # 优先从PDF提取
+        if pdf_type == 'txt' and pdf_doc is not None:
+            code_content, success = PDFUtils.extract_text_from_pdf(
+                pdf_doc, page_idx, bbox, scale
+            )
+            if not success:
+                code_content = ""
+        
+        # PDF提取失败或扫描件,使用OCR
+        if not code_content:
+            try:
+                ocr_results = self.ocr_recognizer.recognize_text(cropped_image)
+                if ocr_results:
+                    lines = []
+                    for item in ocr_results:
+                        lines.append(item.get('text', ''))
+                    code_content = '\n'.join(lines)
+                    # 将 OCR 坐标转换为绝对坐标
+                    ocr_details = self._convert_ocr_details_to_absolute(ocr_results, bbox)
+            except Exception as e:
+                logger.warning(f"Code OCR failed: {e}")
+        
+        return {
+            'type': category,
+            'bbox': bbox,
+            'confidence': layout_item.get('confidence', 0.0),
+            'content': {
+                'code': code_content,
+                'language': '',
+                'ocr_details': ocr_details
+            }
+        }
+    
+    def process_discard_element(
+        self,
+        image: np.ndarray,
+        layout_item: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """
+        处理丢弃元素(水印、装饰等)
+        
+        提取文字块用于后续分析,但标记为丢弃类型
+        
+        Args:
+            image: 页面图像
+            layout_item: 布局检测项
+            
+        Returns:
+            处理后的元素字典
+        """
+        bbox = layout_item.get('bbox', [0, 0, 0, 0])
+        category = layout_item.get('category', 'abandon')
+        cropped_image = CoordinateUtils.crop_region(image, bbox)
+        
+        text_content = ""
+        ocr_details: List[Dict[str, Any]] = []
+        try:
+            ocr_results = self.ocr_recognizer.recognize_text(cropped_image)
+            if ocr_results:
+                text_parts = [
+                    item['text'] for item in ocr_results 
+                    if item.get('confidence', 0) > 0.5
+                ]
+                text_content = " ".join(text_parts)
+                # 将 OCR 坐标转换为绝对坐标
+                ocr_details = self._convert_ocr_details_to_absolute(ocr_results, bbox)
+        except Exception as e:
+            logger.debug(f"Discard element OCR failed: {e}")
+        
+        return {
+            'type': 'discarded',
+            'bbox': bbox,
+            'confidence': layout_item.get('confidence', 0.0),
+            'original_category': category,
+            'content': {
+                'text': text_content,
+                'ocr_details': ocr_details
+            }
+        }
+    
+    @staticmethod
+    def create_error_element(
+        layout_item: Dict[str, Any], 
+        error_msg: str
+    ) -> Dict[str, Any]:
+        """
+        创建错误元素
+        
+        Args:
+            layout_item: 原始布局项
+            error_msg: 错误信息
+            
+        Returns:
+            错误元素字典
+        """
+        return {
+            'type': 'error',
+            'bbox': layout_item.get('bbox', [0, 0, 0, 0]),
+            'confidence': 0.0,
+            'content': {'error': error_msg},
+            'original_category': layout_item.get('category', 'unknown')
+        }
+

+ 255 - 0
zhch/universal_doc_parser/core/layout_utils.py

@@ -0,0 +1,255 @@
+"""
+布局处理工具模块
+
+提供布局相关处理功能:
+- 重叠框检测与去重
+- 阅读顺序排序
+- IoU/重叠比例计算
+"""
+from typing import Dict, List, Any
+from loguru import logger
+
+# 导入 MinerU 组件
+try:
+    from mineru.utils.boxbase import calculate_iou, calculate_overlap_area_2_minbox_area_ratio
+    MINERU_AVAILABLE = True
+except ImportError:
+    MINERU_AVAILABLE = False
+    calculate_iou = None
+    calculate_overlap_area_2_minbox_area_ratio = None
+
+
+class LayoutUtils:
+    """布局处理工具类"""
+    
+    @staticmethod
+    def calculate_iou(bbox1: List[float], bbox2: List[float]) -> float:
+        """
+        计算两个 bbox 的 IoU(交并比)
+        
+        Args:
+            bbox1: 第一个 bbox [x1, y1, x2, y2]
+            bbox2: 第二个 bbox [x1, y1, x2, y2]
+            
+        Returns:
+            IoU 值
+        """
+        if MINERU_AVAILABLE and calculate_iou is not None:
+            return calculate_iou(bbox1, bbox2)
+        
+        # 备用实现
+        x_left = max(bbox1[0], bbox2[0])
+        y_top = max(bbox1[1], bbox2[1])
+        x_right = min(bbox1[2], bbox2[2])
+        y_bottom = min(bbox1[3], bbox2[3])
+        
+        if x_right < x_left or y_bottom < y_top:
+            return 0.0
+        
+        intersection_area = (x_right - x_left) * (y_bottom - y_top)
+        bbox1_area = (bbox1[2] - bbox1[0]) * (bbox1[3] - bbox1[1])
+        bbox2_area = (bbox2[2] - bbox2[0]) * (bbox2[3] - bbox2[1])
+        
+        if bbox1_area == 0 or bbox2_area == 0:
+            return 0.0
+        
+        return intersection_area / float(bbox1_area + bbox2_area - intersection_area)
+    
+    @staticmethod
+    def calculate_overlap_ratio(bbox1: List[float], bbox2: List[float]) -> float:
+        """
+        计算重叠面积占小框面积的比例
+        
+        Args:
+            bbox1: 第一个 bbox [x1, y1, x2, y2]
+            bbox2: 第二个 bbox [x1, y1, x2, y2]
+            
+        Returns:
+            重叠比例
+        """
+        if MINERU_AVAILABLE and calculate_overlap_area_2_minbox_area_ratio is not None:
+            return calculate_overlap_area_2_minbox_area_ratio(bbox1, bbox2)
+        
+        # 备用实现
+        x_left = max(bbox1[0], bbox2[0])
+        y_top = max(bbox1[1], bbox2[1])
+        x_right = min(bbox1[2], bbox2[2])
+        y_bottom = min(bbox1[3], bbox2[3])
+        
+        if x_right < x_left or y_bottom < y_top:
+            return 0.0
+        
+        intersection_area = (x_right - x_left) * (y_bottom - y_top)
+        area1 = (bbox1[2] - bbox1[0]) * (bbox1[3] - bbox1[1])
+        area2 = (bbox2[2] - bbox2[0]) * (bbox2[3] - bbox2[1])
+        min_area = min(area1, area2)
+        
+        if min_area == 0:
+            return 0.0
+        
+        return intersection_area / min_area
+    
+    @staticmethod
+    def remove_overlapping_boxes(
+        layout_results: List[Dict[str, Any]],
+        iou_threshold: float = 0.8,
+        overlap_ratio_threshold: float = 0.8
+    ) -> List[Dict[str, Any]]:
+        """
+        处理重叠的布局框(参考 MinerU 的去重策略)
+        
+        策略:
+        1. 高 IoU 重叠:保留置信度高的框
+        2. 包含关系:小框被大框高度包含时,保留大框并扩展边界
+        3. 同类型优先合并
+        
+        Args:
+            layout_results: Layout 检测结果列表
+            iou_threshold: IoU 阈值,超过此值认为高度重叠
+            overlap_ratio_threshold: 重叠面积占小框面积的比例阈值
+            
+        Returns:
+            去重后的布局结果列表
+        """
+        if not layout_results or len(layout_results) <= 1:
+            return layout_results
+        
+        # 复制列表避免修改原数据
+        results = [item.copy() for item in layout_results]
+        need_remove = set()
+        
+        for i in range(len(results)):
+            if i in need_remove:
+                continue
+                
+            for j in range(i + 1, len(results)):
+                if j in need_remove:
+                    continue
+                
+                bbox1 = results[i].get('bbox', [0, 0, 0, 0])
+                bbox2 = results[j].get('bbox', [0, 0, 0, 0])
+                
+                if len(bbox1) < 4 or len(bbox2) < 4:
+                    continue
+                
+                # 计算 IoU
+                iou = LayoutUtils.calculate_iou(bbox1, bbox2)
+                
+                if iou > iou_threshold:
+                    # 高度重叠,保留置信度高的
+                    score1 = results[i].get('confidence', results[i].get('score', 0))
+                    score2 = results[j].get('confidence', results[j].get('score', 0))
+                    
+                    if score1 >= score2:
+                        need_remove.add(j)
+                    else:
+                        need_remove.add(i)
+                        break  # i 被移除,跳出内层循环
+                else:
+                    # 检查包含关系
+                    overlap_ratio = LayoutUtils.calculate_overlap_ratio(bbox1, bbox2)
+                    
+                    if overlap_ratio > overlap_ratio_threshold:
+                        # 小框被大框高度包含
+                        area1 = (bbox1[2] - bbox1[0]) * (bbox1[3] - bbox1[1])
+                        area2 = (bbox2[2] - bbox2[0]) * (bbox2[3] - bbox2[1])
+                        
+                        if area1 <= area2:
+                            small_idx, large_idx = i, j
+                        else:
+                            small_idx, large_idx = j, i
+                        
+                        # 扩展大框的边界
+                        small_bbox = results[small_idx]['bbox']
+                        large_bbox = results[large_idx]['bbox']
+                        results[large_idx]['bbox'] = [
+                            min(small_bbox[0], large_bbox[0]),
+                            min(small_bbox[1], large_bbox[1]),
+                            max(small_bbox[2], large_bbox[2]),
+                            max(small_bbox[3], large_bbox[3])
+                        ]
+                        need_remove.add(small_idx)
+                        
+                        if small_idx == i:
+                            break  # i 被移除,跳出内层循环
+        
+        # 返回去重后的结果
+        return [results[i] for i in range(len(results)) if i not in need_remove]
+    
+    @staticmethod
+    def sort_elements_by_reading_order(
+        elements: List[Dict[str, Any]],
+        y_tolerance: float = 15.0
+    ) -> List[Dict[str, Any]]:
+        """
+        根据阅读顺序对元素进行排序,并添加 reading_order 字段
+        
+        排序规则:
+        1. 按Y坐标分行(考虑容差,Y坐标相近的元素视为同一行)
+        2. 同一行内按X坐标从左到右排序
+        3. 行与行之间按Y坐标从上到下排序
+        
+        Args:
+            elements: 元素列表(坐标已转换为原始图片坐标系)
+            y_tolerance: Y坐标容差,在此范围内的元素被视为同一行
+            
+        Returns:
+            排序后的元素列表,每个元素都添加了 reading_order 字段
+        """
+        if not elements:
+            return elements
+        
+        # 为每个元素提取排序用的坐标
+        elements_with_coords = []
+        for elem in elements:
+            bbox = elem.get('bbox', [0, 0, 0, 0])
+            if len(bbox) >= 4:
+                y_top = bbox[1]  # 上边界
+                x_left = bbox[0]  # 左边界
+            else:
+                y_top = 0
+                x_left = 0
+            elements_with_coords.append((elem, y_top, x_left))
+        
+        # 先按Y坐标排序
+        elements_with_coords.sort(key=lambda x: (x[1], x[2]))
+        
+        # 按Y坐标分行
+        rows = []
+        current_row = []
+        current_row_y = None
+        
+        for elem, y_top, x_left in elements_with_coords:
+            if current_row_y is None:
+                # 第一个元素
+                current_row.append((elem, x_left))
+                current_row_y = y_top
+            elif abs(y_top - current_row_y) <= y_tolerance:
+                # 同一行
+                current_row.append((elem, x_left))
+            else:
+                # 新的一行
+                rows.append(current_row)
+                current_row = [(elem, x_left)]
+                current_row_y = y_top
+        
+        # 添加最后一行
+        if current_row:
+            rows.append(current_row)
+        
+        # 每行内按X坐标排序,然后展平
+        sorted_elements = []
+        reading_order = 0
+        
+        for row in rows:
+            # 行内按X坐标排序
+            row.sort(key=lambda x: x[1])
+            for elem, _ in row:
+                # 添加 reading_order 字段
+                elem['reading_order'] = reading_order
+                sorted_elements.append(elem)
+                reading_order += 1
+        
+        logger.debug(f"📖 Elements sorted by reading order: {len(sorted_elements)} elements")
+        return sorted_elements
+

+ 214 - 0
zhch/universal_doc_parser/core/pdf_utils.py

@@ -0,0 +1,214 @@
+"""
+PDF处理工具模块
+
+提供PDF相关处理功能:
+- PDF加载与分类
+- PDF文本提取
+- 跨页表格合并
+"""
+from typing import Dict, List, Any, Optional, Tuple
+from pathlib import Path
+from PIL import Image
+from loguru import logger
+
+# 导入 MinerU 组件
+try:
+    from mineru.utils.pdf_classify import classify as pdf_classify
+    from mineru.utils.pdf_image_tools import load_images_from_pdf
+    from mineru.utils.enum_class import ImageType
+    from mineru.utils.pdf_text_tool import get_page as pdf_get_page_text
+    MINERU_AVAILABLE = True
+except ImportError:
+    MINERU_AVAILABLE = False
+    pdf_classify = None
+    load_images_from_pdf = None
+    ImageType = None
+    pdf_get_page_text = None
+
+
+class PDFUtils:
+    """PDF处理工具类"""
+    
+    @staticmethod
+    def load_and_classify_document(
+        document_path: Path,
+        dpi: int = 200
+    ) -> Tuple[List[Dict], str, Optional[Any]]:
+        """
+        加载文档并分类
+        
+        Args:
+            document_path: 文档路径
+            dpi: PDF渲染DPI
+            
+        Returns:
+            (images_list, pdf_type, pdf_doc)
+            - images_list: 图像列表,每个元素包含 {'img_pil': PIL.Image, 'scale': float}
+            - pdf_type: 'ocr' 或 'txt'
+            - pdf_doc: PDF文档对象(如果是PDF)
+        """
+        pdf_doc = None
+        pdf_type = 'ocr'  # 默认使用OCR模式
+        images = []
+        
+        if document_path.is_dir():
+            # 处理目录:遍历所有图片
+            image_extensions = {'.png', '.jpg', '.jpeg', '.bmp', '.tiff', '.tif'}
+            image_files = sorted([
+                f for f in document_path.iterdir() 
+                if f.suffix.lower() in image_extensions
+            ])
+            
+            for img_file in image_files:
+                img = Image.open(img_file)
+                if img.mode != 'RGB':
+                    img = img.convert('RGB')
+                images.append({
+                    'img_pil': img,
+                    'scale': 1.0,
+                    'source_path': str(img_file)
+                })
+            
+            pdf_type = 'ocr'  # 图片目录始终使用OCR模式
+            
+        elif document_path.suffix.lower() == '.pdf':
+            # 处理PDF文件
+            if not MINERU_AVAILABLE:
+                raise RuntimeError("MinerU components not available for PDF processing")
+            
+            with open(document_path, 'rb') as f:
+                pdf_bytes = f.read()
+            
+            # PDF分类
+            pdf_type = pdf_classify(pdf_bytes)
+            logger.info(f"📋 PDF classified as: {pdf_type}")
+            
+            # 加载图像
+            images_list, pdf_doc = load_images_from_pdf(
+                pdf_bytes, 
+                dpi=dpi,
+                image_type=ImageType.PIL
+            )
+            
+            for img_dict in images_list:
+                images.append({
+                    'img_pil': img_dict['img_pil'],
+                    'scale': img_dict.get('scale', dpi / 72),
+                    'source_path': str(document_path)
+                })
+                
+        elif document_path.suffix.lower() in ['.png', '.jpg', '.jpeg', '.bmp', '.tiff', '.tif']:
+            # 处理单个图片
+            img = Image.open(document_path)
+            if img.mode != 'RGB':
+                img = img.convert('RGB')
+            images.append({
+                'img_pil': img,
+                'scale': 1.0,
+                'source_path': str(document_path)
+            })
+            pdf_type = 'ocr'
+            
+        else:
+            raise ValueError(f"Unsupported file format: {document_path.suffix}")
+        
+        return images, pdf_type, pdf_doc
+    
+    @staticmethod
+    def extract_text_from_pdf(
+        pdf_doc: Any,
+        page_idx: int,
+        bbox: List[float],
+        scale: float
+    ) -> Tuple[str, bool]:
+        """
+        从PDF直接提取文本(使用 MinerU 的 pypdfium2 方式)
+        
+        Args:
+            pdf_doc: pypdfium2 的 PdfDocument 对象
+            page_idx: 页码索引
+            bbox: 目标区域的bbox(图像坐标)
+            scale: 图像与PDF的缩放比例
+            
+        Returns:
+            (text, success)
+        """
+        if not MINERU_AVAILABLE or pdf_get_page_text is None:
+            logger.debug("MinerU pdf_text_tool not available")
+            return "", False
+            
+        try:
+            page = pdf_doc[page_idx]
+            
+            # 将图像坐标转换为PDF坐标
+            pdf_bbox = [
+                bbox[0] / scale,
+                bbox[1] / scale,
+                bbox[2] / scale,
+                bbox[3] / scale
+            ]
+            
+            # 使用 MinerU 的方式获取页面文本信息
+            page_dict = pdf_get_page_text(page)
+            
+            # 从 blocks 中提取与 bbox 重叠的文本
+            text_parts = []
+            for block in page_dict.get('blocks', []):
+                for line in block.get('lines', []):
+                    line_bbox = line.get('bbox')
+                    if line_bbox and hasattr(line_bbox, 'bbox'):
+                        line_bbox = line_bbox.bbox  # pdftext 的 BBox 对象
+                    elif isinstance(line_bbox, (list, tuple)) and len(line_bbox) >= 4:
+                        line_bbox = list(line_bbox)
+                    else:
+                        continue
+                    
+                    # 检查 line 是否与目标 bbox 重叠
+                    if PDFUtils._bbox_overlap(pdf_bbox, line_bbox):
+                        for span in line.get('spans', []):
+                            span_text = span.get('text', '')
+                            if span_text:
+                                text_parts.append(span_text)
+            
+            text = ' '.join(text_parts)
+            return text.strip(), bool(text.strip())
+            
+        except Exception as e:
+            import traceback
+            logger.debug(f"PDF text extraction error: {e}")
+            logger.debug(traceback.format_exc())
+            return "", False
+    
+    @staticmethod
+    def _bbox_overlap(bbox1: List[float], bbox2: List[float]) -> bool:
+        """检查两个 bbox 是否重叠"""
+        if len(bbox1) < 4 or len(bbox2) < 4:
+            return False
+        
+        x1_1, y1_1, x2_1, y2_1 = bbox1[:4]
+        x1_2, y1_2, x2_2, y2_2 = bbox2[:4]
+        
+        if x2_1 < x1_2 or x2_2 < x1_1:
+            return False
+        if y2_1 < y1_2 or y2_2 < y1_1:
+            return False
+        
+        return True
+    
+    @staticmethod
+    def merge_cross_page_tables(results: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        合并跨页表格
+        
+        TODO: 实现跨页表格合并逻辑
+        可以参考 MinerU 的 cross_page_table_merge 实现
+        
+        Args:
+            results: 处理结果字典
+            
+        Returns:
+            合并后的结果
+        """
+        # TODO: 实现跨页表格合并逻辑
+        return results
+

+ 0 - 419
zhch/universal_doc_parser/core/pipeline_manager.py

@@ -1,419 +0,0 @@
-from typing import Dict, List, Any, Optional, Union
-from pathlib import Path
-import numpy as np
-from PIL import Image
-import fitz  # PyMuPDF
-from loguru import logger
-
-from .model_factory import ModelFactory
-from .config_manager import ConfigManager
-from models.adapters import BaseAdapter
-
-class FinancialDocPipeline:
-    """金融文档处理统一流水线"""
-    
-    def __init__(self, config_path: str):
-        self.config = ConfigManager.load_config(config_path)
-        self.scene_name = self.config.get('scene_name', 'unknown')
-        
-        # 初始化各个组件
-        self._init_components()
-        
-    def _init_components(self):
-        """初始化处理组件"""
-        try:
-            # 1. 预处理器(方向分类、图像矫正等)
-            self.preprocessor = ModelFactory.create_preprocessor(
-                self.config['preprocessor']
-            )
-            
-            # 2. 版式检测器
-            self.layout_detector = ModelFactory.create_layout_detector(
-                self.config['layout_detection']
-            )
-            
-            # 3. VL识别器(表格、公式等)
-            self.vl_recognizer = ModelFactory.create_vl_recognizer(
-                self.config['vl_recognition']
-            )
-            
-            # 4. OCR识别器
-            self.ocr_recognizer = ModelFactory.create_ocr_recognizer(
-                self.config['ocr_recognition']
-            )
-            
-            logger.info(f"✅ Initialized pipeline for scene: {self.scene_name}")
-            
-        except Exception as e:
-            logger.error(f"❌ Failed to initialize pipeline components: {e}")
-            raise
-    
-    def process_document(self, document_path: str) -> Dict[str, Any]:
-        """
-        处理文档的主流程
-        
-        Args:
-            document_path: 文档路径
-            
-        Returns:
-            处理结果,包含所有元素的坐标和内容信息
-        """
-        results = {
-            'scene': self.scene_name,
-            'document_path': document_path,
-            'pages': [],
-            'metadata': self._extract_metadata(document_path)
-        }
-        
-        try:
-            # 加载文档图像
-            images = self._load_document_images(document_path)
-            logger.info(f"📄 Loaded {len(images)} pages from document")
-            
-            for page_idx, image in enumerate(images):
-                logger.info(f"🔍 Processing page {page_idx + 1}/{len(images)}")
-                page_result = self._process_single_page(image, page_idx)
-                results['pages'].append(page_result)
-                
-        except Exception as e:
-            logger.error(f"❌ Failed to process document: {e}")
-            raise
-            
-        return results
-    
-    def _load_document_images(self, document_path: str) -> List[np.ndarray]:
-        """加载文档图像"""
-        document_path = Path(document_path)
-        
-        if not document_path.exists():
-            raise FileNotFoundError(f"Document not found: {document_path}")
-        
-        images = []
-        
-        if document_path.suffix.lower() == '.pdf':
-            # 处理PDF文件
-            doc = fitz.open(document_path)
-            try:
-                for page_num in range(len(doc)):
-                    page = doc.load_page(page_num)
-                    # 设置合适的DPI
-                    dpi = self.config.get('input', {}).get('dpi', 200)
-                    mat = fitz.Matrix(dpi/72, dpi/72)
-                    pix = page.get_pixmap(matrix=mat)
-                    img_data = pix.tobytes("ppm")
-                    
-                    # 转换为numpy数组
-                    from io import BytesIO
-                    img = Image.open(BytesIO(img_data))
-                    img_array = np.array(img)
-                    images.append(img_array)
-            finally:
-                doc.close()
-                
-        elif document_path.suffix.lower() in ['.png', '.jpg', '.jpeg', '.bmp', '.tiff']:
-            # 处理图像文件
-            img = Image.open(document_path)
-            if img.mode != 'RGB':
-                img = img.convert('RGB')
-            img_array = np.array(img)
-            images.append(img_array)
-            
-        else:
-            raise ValueError(f"Unsupported file format: {document_path.suffix}")
-        
-        return images
-    
-    def _extract_metadata(self, document_path: str) -> Dict[str, Any]:
-        """提取文档元数据"""
-        document_path = Path(document_path)
-        
-        metadata = {
-            'filename': document_path.name,
-            'size': document_path.stat().st_size,
-            'format': document_path.suffix.lower()
-        }
-        
-        # 如果是PDF,提取更多元数据
-        if document_path.suffix.lower() == '.pdf':
-            try:
-                doc = fitz.open(document_path)
-                metadata.update({
-                    'page_count': len(doc),
-                    'title': doc.metadata.get('title', ''),
-                    'author': doc.metadata.get('author', ''),
-                    'subject': doc.metadata.get('subject', ''),
-                    'creator': doc.metadata.get('creator', '')
-                })
-                doc.close()
-            except Exception:
-                pass
-        
-        return metadata
-    
-    def _process_single_page(self, image: np.ndarray, page_idx: int) -> Dict[str, Any]:
-        """处理单页文档"""
-        # 1. 预处理(方向校正等)
-        try:
-            preprocessed_image, rotate_angle = self.preprocessor.process(image)
-        except Exception as e:
-            logger.warning(f"⚠️ Preprocessing failed for page {page_idx}: {e}")
-            preprocessed_image = image
-        
-        # 2. 版式检测
-        try:
-            layout_results = self.layout_detector.detect(preprocessed_image)
-            logger.info(f"📋 Detected {len(layout_results)} layout elements on page {page_idx}")
-        except Exception as e:
-            logger.error(f"❌ Layout detection failed for page {page_idx}: {e}")
-            layout_results = []
-        
-        # 3. 根据场景类型分别处理不同元素
-        page_elements = []
-        
-        for layout_item in layout_results:
-            try:
-                element_type = layout_item['category']
-                
-                if element_type in ['table_body', 'table']:
-                    # 表格使用VL模型处理
-                    element_result = self._process_table_element(
-                        preprocessed_image, layout_item
-                    )
-                elif element_type in ['text', 'title', 'ocr_text']:
-                    # 文本使用OCR处理
-                    element_result = self._process_text_element(
-                        preprocessed_image, layout_item
-                    )
-                elif element_type in ['interline_equation', 'inline_equation']:
-                    # 公式使用VL模型处理
-                    element_result = self._process_formula_element(
-                        preprocessed_image, layout_item
-                    )
-                else:
-                    # 其他元素保持原样
-                    element_result = layout_item.copy()
-                    element_result['type'] = element_type
-                    
-                page_elements.append(element_result)
-                
-            except Exception as e:
-                logger.warning(f"⚠️ Failed to process element {element_type}: {e}")
-                # 添加失败的元素,标记为错误
-                error_element = layout_item.copy()
-                error_element['type'] = 'error'
-                error_element['error'] = str(e)
-                page_elements.append(error_element)
-        
-        return {
-            'page_idx': page_idx,
-            'elements': page_elements,
-            'layout_raw': layout_results,
-            'image_shape': preprocessed_image.shape,
-            'processed_image': preprocessed_image,
-            'angle': rotate_angle
-        }
-    
-    def _process_table_element(self, image: np.ndarray, layout_item: Dict[str, Any]) -> Dict[str, Any]:
-        """处理表格元素"""
-        try:
-            # 裁剪表格区域
-            cropped_table = self._crop_region(image, layout_item['bbox'])
-            
-            # 使用VL模型识别表格
-            table_result = self.vl_recognizer.recognize_table(
-                cropped_table,
-                return_cells_coordinate=True  # 关键:返回单元格坐标
-            )
-            
-            # 转换坐标到原图坐标系
-            if 'cells' in table_result:
-                for cell in table_result['cells']:
-                    cell['absolute_bbox'] = self._convert_to_absolute_coords(
-                        cell['bbox'], layout_item['bbox']
-                    )
-            
-            result = {
-                'type': 'table',
-                'bbox': layout_item['bbox'],
-                'confidence': layout_item.get('confidence', 0.0),
-                'content': table_result,
-                'scene_specific': self._add_scene_specific_info(table_result)
-            }
-            
-            logger.info(f"✅ Table processed with {len(table_result.get('cells', []))} cells")
-            return result
-            
-        except Exception as e:
-            logger.error(f"❌ Table processing failed: {e}")
-            return {
-                'type': 'table',
-                'bbox': layout_item['bbox'],
-                'content': {'html': '', 'markdown': '', 'cells': []},
-                'error': str(e)
-            }
-    
-    def _process_text_element(self, image: np.ndarray, layout_item: Dict[str, Any]) -> Dict[str, Any]:
-        """处理文本元素"""
-        try:
-            # 裁剪文本区域
-            cropped_text = self._crop_region(image, layout_item['bbox'])
-            
-            # 使用OCR识别文本
-            text_results = self.ocr_recognizer.recognize_text(cropped_text)
-            
-            # 合并识别结果
-            combined_text = ""
-            if text_results:
-                text_parts = [item['text'] for item in text_results if item['confidence'] > 0.5]
-                combined_text = " ".join(text_parts)
-            
-            result = {
-                'type': layout_item['category'],
-                'bbox': layout_item['bbox'],
-                'confidence': layout_item.get('confidence', 0.0),
-                'content': {
-                    'text': combined_text,
-                    'ocr_details': text_results
-                }
-            }
-            
-            logger.info(f"✅ Text processed: '{combined_text[:50]}...'")
-            return result
-            
-        except Exception as e:
-            logger.error(f"❌ Text processing failed: {e}")
-            return {
-                'type': layout_item['category'],
-                'bbox': layout_item['bbox'],
-                'content': {'text': '', 'ocr_details': []},
-                'error': str(e)
-            }
-    
-    def _process_formula_element(self, image: np.ndarray, layout_item: Dict[str, Any]) -> Dict[str, Any]:
-        """处理公式元素"""
-        try:
-            # 裁剪公式区域
-            cropped_formula = self._crop_region(image, layout_item['bbox'])
-            
-            # 使用VL模型识别公式
-            formula_result = self.vl_recognizer.recognize_formula(cropped_formula)
-            
-            result = {
-                'type': 'formula',
-                'bbox': layout_item['bbox'],
-                'confidence': layout_item.get('confidence', 0.0),
-                'content': formula_result
-            }
-            
-            logger.info(f"✅ Formula processed: {formula_result.get('latex', '')[:50]}...")
-            return result
-            
-        except Exception as e:
-            logger.error(f"❌ Formula processing failed: {e}")
-            return {
-                'type': 'formula',
-                'bbox': layout_item['bbox'],
-                'content': {'latex': '', 'confidence': 0.0},
-                'error': str(e)
-            }
-    
-    def _crop_region(self, image: np.ndarray, bbox: List[float]) -> np.ndarray:
-        """裁剪图像区域"""
-        if len(bbox) < 4:
-            return image
-            
-        x1, y1, x2, y2 = map(int, bbox)
-        
-        # 边界检查
-        h, w = image.shape[:2]
-        x1 = max(0, min(x1, w))
-        y1 = max(0, min(y1, h))
-        x2 = max(x1, min(x2, w))
-        y2 = max(y1, min(y2, h))
-        
-        return image[y1:y2, x1:x2]
-    
-    def _convert_to_absolute_coords(self, relative_bbox: List[float], region_bbox: List[float]) -> List[float]:
-        """将相对坐标转换为绝对坐标"""
-        if len(relative_bbox) < 4 or len(region_bbox) < 4:
-            return relative_bbox
-            
-        rx1, ry1, rx2, ry2 = relative_bbox
-        bx1, by1, bx2, by2 = region_bbox
-        
-        # 计算绝对坐标
-        abs_x1 = bx1 + rx1
-        abs_y1 = by1 + ry1
-        abs_x2 = bx1 + rx2
-        abs_y2 = by1 + ry2
-        
-        return [abs_x1, abs_y1, abs_x2, abs_y2]
-    
-    def _add_scene_specific_info(self, content: Dict[str, Any]) -> Dict[str, Any]:
-        """根据场景添加特定信息"""
-        if self.scene_name == 'bank_statement':
-            return self._process_bank_statement_table(content)
-        elif self.scene_name == 'financial_report':
-            return self._process_financial_report_table(content)
-        return {}
-    
-    def _process_bank_statement_table(self, content: Dict[str, Any]) -> Dict[str, Any]:
-        """处理银行流水表格特定逻辑"""
-        scene_info = {
-            'table_type': 'bank_statement',
-            'expected_columns': ['日期', '摘要', '收入', '支出', '余额'],
-            'validation_rules': {
-                'amount_format': True,
-                'date_format': True,
-                'balance_consistency': True
-            }
-        }
-        
-        # 进行银行流水特定的验证和处理
-        if 'html' in content and content['html']:
-            # 这里可以添加银行流水特定的HTML后处理逻辑
-            pass
-            
-        return scene_info
-    
-    def _process_financial_report_table(self, content: Dict[str, Any]) -> Dict[str, Any]:
-        """处理财务报表特定逻辑"""
-        scene_info = {
-            'table_type': 'financial_report',
-            'complex_headers': True,
-            'merged_cells': True,
-            'validation_rules': {
-                'accounting_format': True,
-                'sum_validation': True
-            }
-        }
-        
-        # 进行财务报表特定的验证和处理
-        if 'html' in content and content['html']:
-            # 这里可以添加财务报表特定的HTML后处理逻辑
-            pass
-            
-        return scene_info
-    
-    def cleanup(self):
-        """清理资源"""
-        try:
-            if hasattr(self, 'preprocessor'):
-                self.preprocessor.cleanup()
-            if hasattr(self, 'layout_detector'):
-                self.layout_detector.cleanup()
-            if hasattr(self, 'vl_recognizer'):
-                self.vl_recognizer.cleanup()
-            if hasattr(self, 'ocr_recognizer'):
-                self.ocr_recognizer.cleanup()
-                
-            logger.info("✅ Pipeline cleanup completed")
-            
-        except Exception as e:
-            logger.warning(f"⚠️ Cleanup failed: {e}")
-    
-    def __enter__(self):
-        return self
-    
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        self.cleanup()

File diff suppressed because it is too large
+ 229 - 1138
zhch/universal_doc_parser/core/pipeline_manager_v2.py


Some files were not shown because too many files changed in this diff