461bf6fd40
CI / lint (3.11) (push) Has been cancelled
CI / lint (3.12) (push) Has been cancelled
CI / lint (3.13) (push) Has been cancelled
CI / shellcheck (push) Has been cancelled
CI / shfmt (push) Has been cancelled
CI / setup (3.11) (push) Has been cancelled
CI / setup (3.12) (push) Has been cancelled
CI / setup (3.13) (push) Has been cancelled
CI / check-licenses (3.12) (push) Has been cancelled
CI / test_unit (3.11) (push) Has been cancelled
CI / test_unit (3.12) (push) Has been cancelled
CI / test_unit (3.13) (push) Has been cancelled
CI / test_unit_no_extras (3.11) (push) Has been cancelled
CI / test_unit_no_extras (3.12) (push) Has been cancelled
CI / test_json_to_html (3.12) (push) Has been cancelled
CI / test_unit_no_extras (3.13) (push) Has been cancelled
CI / test_unit_dependency_extras (csv, 3.12, --extra csv) (push) Has been cancelled
CI / test_unit_dependency_extras (xlsx, 3.11, --extra xlsx) (push) Has been cancelled
CI / test_unit_dependency_extras (xlsx, 3.12, --extra xlsx) (push) Has been cancelled
CI / test_unit_dependency_extras (csv, 3.11, --extra csv) (push) Has been cancelled
CI / test_unit_dependency_extras (csv, 3.13, --extra csv) (push) Has been cancelled
CI / test_unit_dependency_extras (docx, 3.11, --extra docx) (push) Has been cancelled
CI / test_unit_dependency_extras (docx, 3.12, --extra docx) (push) Has been cancelled
CI / test_unit_dependency_extras (docx, 3.13, --extra docx) (push) Has been cancelled
CI / test_unit_dependency_extras (markdown, 3.11, --extra md) (push) Has been cancelled
CI / test_unit_dependency_extras (markdown, 3.12, --extra md) (push) Has been cancelled
CI / test_unit_dependency_extras (markdown, 3.13, --extra md) (push) Has been cancelled
CI / test_unit_dependency_extras (odt, 3.11, --extra odt) (push) Has been cancelled
CI / test_unit_dependency_extras (odt, 3.12, --extra odt) (push) Has been cancelled
CI / test_unit_dependency_extras (odt, 3.13, --extra odt) (push) Has been cancelled
CI / test_unit_dependency_extras (pdf-image, 3.11, --extra pdf --extra image --extra paddleocr) (push) Has been cancelled
CI / test_unit_dependency_extras (pdf-image, 3.12, --extra pdf --extra image --extra paddleocr) (push) Has been cancelled
CI / test_unit_dependency_extras (pdf-image, 3.13, --extra pdf --extra image --extra paddleocr) (push) Has been cancelled
CI / test_unit_dependency_extras (pptx, 3.11, --extra pptx) (push) Has been cancelled
CI / test_unit_dependency_extras (pptx, 3.12, --extra pptx) (push) Has been cancelled
CI / test_unit_dependency_extras (pptx, 3.13, --extra pptx) (push) Has been cancelled
CI / test_unit_dependency_extras (pypandoc, 3.11, --extra epub --extra org --extra rtf --extra rst) (push) Has been cancelled
CI / test_unit_dependency_extras (pypandoc, 3.12, --extra epub --extra org --extra rtf --extra rst) (push) Has been cancelled
CI / test_unit_dependency_extras (pypandoc, 3.13, --extra epub --extra org --extra rtf --extra rst) (push) Has been cancelled
Build And Push Docker Image / set-short-sha (push) Has been cancelled
Partition Benchmark / setup (push) Has been cancelled
Partition Benchmark / Measure and compare partition() runtime (push) Has been cancelled
CI / test_unit_dependency_extras (xlsx, 3.13, --extra xlsx) (push) Has been cancelled
CI / test_ingest_src (3.12) (push) Has been cancelled
CI / test_json_to_markdown (3.12) (push) Has been cancelled
CI / changelog (push) Has been cancelled
CI / test_dockerfile (push) Has been cancelled
CodeQL / Analyze (python) (push) Has been cancelled
Build And Push Docker Image / build-images (linux/amd64, opensource-linux-8core) (push) Has been cancelled
Build And Push Docker Image / build-images (linux/arm64, ubuntu-24.04-arm) (push) Has been cancelled
Build And Push Docker Image / publish-images (push) Has been cancelled
1423 lines
57 KiB
Python
1423 lines
57 KiB
Python
from __future__ import annotations
|
|
|
|
import math
|
|
import os
|
|
from typing import TYPE_CHECKING, Any, BinaryIO, Iterable, List, Optional, Union, cast
|
|
|
|
import numpy as np
|
|
from pdfminer.layout import LAParams, LTChar, LTContainer, LTTextBox
|
|
from pdfminer.pdftypes import PDFObjRef
|
|
from pdfminer.utils import decode_text, open_filename
|
|
from unstructured_inference.config import inference_config
|
|
from unstructured_inference.constants import FULL_PAGE_REGION_THRESHOLD, IsExtracted
|
|
from unstructured_inference.inference.elements import Rectangle
|
|
|
|
from unstructured.documents.coordinates import PixelSpace, PointSpace
|
|
from unstructured.documents.elements import CoordinatesMetadata, ElementType
|
|
from unstructured.partition.pdf_image.pdf_image_utils import remove_control_characters
|
|
from unstructured.partition.pdf_image.pdfminer_utils import (
|
|
PDFMinerConfig,
|
|
_is_duplicate_char,
|
|
extract_image_objects,
|
|
extract_text_objects,
|
|
get_text_with_deduplication,
|
|
open_pdfminer_pages_generator,
|
|
rect_to_bbox,
|
|
)
|
|
from unstructured.partition.utils.config import env_config
|
|
from unstructured.partition.utils.constants import SORT_MODE_BASIC, Source
|
|
from unstructured.partition.utils.sorting import sort_text_regions
|
|
from unstructured.utils import requires_dependencies
|
|
|
|
if TYPE_CHECKING:
|
|
from unstructured_inference.inference.elements import TextRegion, TextRegions
|
|
from unstructured_inference.inference.layout import DocumentLayout
|
|
from unstructured_inference.inference.layoutelement import LayoutElements
|
|
|
|
|
|
EPSILON_AREA = 0.01
|
|
# rounding floating point to nearest machine precision
|
|
DEFAULT_ROUND = 15
|
|
|
|
|
|
def process_file_with_pdfminer(
|
|
filename: str = "",
|
|
dpi: int = env_config.PDF_RENDER_DPI,
|
|
password: Optional[str] = None,
|
|
pdfminer_config: Optional[PDFMinerConfig] = None,
|
|
rotation_corrections: Optional[List[int]] = None,
|
|
) -> tuple[List[List["TextRegion"]], List[List]]:
|
|
with open_filename(filename, "rb") as fp:
|
|
fp = cast(BinaryIO, fp)
|
|
extracted_layout, layouts_links = process_data_with_pdfminer(
|
|
file=fp,
|
|
dpi=dpi,
|
|
password=password,
|
|
pdfminer_config=pdfminer_config,
|
|
rotation_corrections=rotation_corrections,
|
|
)
|
|
return extracted_layout, layouts_links
|
|
|
|
|
|
def _rotate_bboxes(coords: np.ndarray, angle: int, width: float, height: float) -> np.ndarray:
|
|
"""Rotate bounding boxes to mirror a rendered page image that was rotated ``angle``
|
|
degrees counter-clockwise (PIL convention) with ``expand=True``.
|
|
|
|
``width``/``height`` are the page-image dimensions in the un-rotated (display) frame.
|
|
unstructured-inference may rotate a page image to make its dominant text upright;
|
|
applying the same rotation here keeps the pdfminer layer aligned with the
|
|
object-detection layer so the two merge correctly.
|
|
"""
|
|
angle %= 360
|
|
if angle == 0 or coords.size == 0:
|
|
return coords
|
|
x1, y1, x2, y2 = coords[:, 0], coords[:, 1], coords[:, 2], coords[:, 3]
|
|
if angle == 90:
|
|
return np.column_stack((y1, width - x2, y2, width - x1))
|
|
if angle == 180:
|
|
return np.column_stack((width - x2, height - y2, width - x1, height - y1))
|
|
if angle == 270:
|
|
return np.column_stack((height - y2, x1, height - y1, x2))
|
|
return coords
|
|
|
|
|
|
def _validate_bbox(bbox: list[int | float]) -> bool:
|
|
return all(x is not None for x in bbox) and (bbox[2] - bbox[0] > 0) and (bbox[3] - bbox[1] > 0)
|
|
|
|
|
|
def _minimum_containing_coords(*regions: TextRegions) -> np.ndarray:
|
|
# TODO: refactor to just use np array as input
|
|
# Optimization: Use np.stack and np.column_stack to build output in a single step
|
|
x1s = np.array([region.x1 for region in regions])
|
|
y1s = np.array([region.y1 for region in regions])
|
|
x2s = np.array([region.x2 for region in regions])
|
|
y2s = np.array([region.y2 for region in regions])
|
|
# Use np.min/max reduction rather than create matrix then operate.
|
|
return np.column_stack(
|
|
(
|
|
np.min(x1s, axis=0),
|
|
np.min(y1s, axis=0),
|
|
np.max(x2s, axis=0),
|
|
np.max(y2s, axis=0),
|
|
)
|
|
)
|
|
|
|
|
|
def _inferred_is_elementtype(
|
|
inferred_layout: LayoutElements, etypes: Iterable[ElementType]
|
|
) -> np.ndarry:
|
|
inferred_text_idx = [
|
|
idx
|
|
for idx, class_name in inferred_layout.element_class_id_map.items()
|
|
if class_name in etypes
|
|
]
|
|
inferred_is_etypes = np.zeros((len(inferred_layout),)).astype(bool)
|
|
for idx in inferred_text_idx:
|
|
inferred_is_etypes = np.logical_or(
|
|
inferred_is_etypes, inferred_layout.element_class_ids == idx
|
|
)
|
|
return inferred_is_etypes
|
|
|
|
|
|
def _inferred_is_text(inferred_layout: LayoutElements) -> np.ndarry:
|
|
"""return a boolean array masking for each element if it is non-image type (True) or image like
|
|
type (False); image types are ElementType.FIGURE/IMAGE/PAGE_BREAK/TABLE"""
|
|
return ~_inferred_is_elementtype(
|
|
inferred_layout,
|
|
etypes=(
|
|
ElementType.FIGURE,
|
|
ElementType.IMAGE,
|
|
# NOTE (yao): PICTURE is not in the loop version of the logic in inference library
|
|
# ElementType.PICTURE,
|
|
ElementType.PAGE_BREAK,
|
|
ElementType.TABLE,
|
|
),
|
|
)
|
|
|
|
|
|
def _merge_extracted_into_inferred_when_almost_the_same(
|
|
extracted_layout: LayoutElements,
|
|
inferred_layout: LayoutElements,
|
|
same_region_threshold: float,
|
|
) -> tuple[np.ndarray]:
|
|
"""merge exstracted elements that have almost the same bounding box as an inferrred element into
|
|
that inferred element: a) the inferred element bounding box is updated, if needed, to be able to
|
|
bound the merged extracted element; b) the inferred element uses the extracted element's text as
|
|
its text attribute. Return a boolean mask array indicating where (when True) an extracted
|
|
element is merged therefore should be excluded from later analysis"""
|
|
|
|
if len(inferred_layout) == 0:
|
|
return np.array([False] * len(extracted_layout))
|
|
if len(extracted_layout) == 0:
|
|
return np.array([])
|
|
|
|
boxes_almost_same = boxes_iou(
|
|
extracted_layout.element_coords,
|
|
inferred_layout.element_coords,
|
|
threshold=same_region_threshold,
|
|
)
|
|
extracted_almost_the_same_as_inferred = np.any(boxes_almost_same, axis=1)
|
|
# NOTE: if a row is full of False the argmax returns first index; we use the mask above to
|
|
# distinguish those (they would be False in the mask)
|
|
first_match = np.argmax(boxes_almost_same, axis=1)
|
|
inferred_indices_to_update = first_match[extracted_almost_the_same_as_inferred]
|
|
extracted_to_remove = extracted_layout.slice(extracted_almost_the_same_as_inferred)
|
|
# copy here in case we change the extracted layout later
|
|
inferred_layout.texts[inferred_indices_to_update] = extracted_to_remove.texts.copy()
|
|
inferred_layout.is_extracted_array[inferred_indices_to_update] = (
|
|
extracted_to_remove.is_extracted_array.copy()
|
|
)
|
|
# use coords that can bound BOTH the inferred and extracted region as final bounding box coords
|
|
inferred_layout.element_coords[inferred_indices_to_update] = _minimum_containing_coords(
|
|
inferred_layout.slice(inferred_indices_to_update),
|
|
extracted_to_remove,
|
|
)
|
|
return extracted_almost_the_same_as_inferred
|
|
|
|
|
|
def _merge_extracted_that_are_subregion_of_inferred_text(
|
|
extracted_layout: LayoutElements,
|
|
inferred_layout: LayoutElements,
|
|
extracted_is_subregion_of_inferred: np.ndarray,
|
|
extracted_to_proc: np.ndarray,
|
|
inferred_to_proc: np.ndarray,
|
|
) -> LayoutElements:
|
|
"""merged extracted elements that are subregions of inferrred elements into those inferred
|
|
elements: the inferred elements' bounding boxes expands, if needed, to include those subregion
|
|
elements. Returns the modified inferred layout where some of its elements' bounding boxes may
|
|
have expanded due to merging.
|
|
"""
|
|
# in theory one extracted __should__ only match at most one inferred region, given inferred
|
|
# region can not overlap; so first match here __should__ also be the only match
|
|
inferred_to_iter = inferred_to_proc[inferred_to_proc]
|
|
extracted_to_iter = extracted_to_proc[extracted_to_proc]
|
|
for inferred_index, inferred_row in enumerate(extracted_is_subregion_of_inferred.T):
|
|
matches = np.where(inferred_row)[0]
|
|
if not matches.size:
|
|
continue
|
|
# Technically those two lines below can be vectorized but this loop would still run anyway;
|
|
# it is not clear which one is overall faster so might worth profiling in the future
|
|
extracted_to_iter[matches] = False
|
|
inferred_to_iter[inferred_index] = False
|
|
# then expand inferred box by all the extracted boxes
|
|
# FIXME (yao): this part is broken at the moment
|
|
inferred_layout.element_coords[[inferred_index]] = _minimum_containing_coords(
|
|
inferred_layout.slice([inferred_index]),
|
|
*[extracted_layout.slice([match]) for match in matches],
|
|
)
|
|
inferred_to_proc[inferred_to_proc] = inferred_to_iter
|
|
extracted_to_proc[extracted_to_proc] = extracted_to_iter
|
|
return inferred_layout
|
|
|
|
|
|
def _mark_non_table_inferred_for_removal_if_has_subregion_relationship(
|
|
extracted_layout: LayoutElements,
|
|
inferred_layout: LayoutElements,
|
|
inferred_to_keep: np.ndarray,
|
|
subregion_threshold: float,
|
|
) -> np.ndaray:
|
|
"""
|
|
Marking elements in inferred layout to remove after merging when:
|
|
- if the inferred element is subregion of an extracted element
|
|
- and/or an extracted element is subregion of this inferred element
|
|
Return updated mask on which inferred indices to keep (when True)
|
|
"""
|
|
inferred_is_subregion_of_extracted = bboxes1_is_almost_subregion_of_bboxes2(
|
|
inferred_layout.element_coords,
|
|
extracted_layout.element_coords,
|
|
threshold=subregion_threshold,
|
|
)
|
|
extracted_is_subregion_of_inferred = bboxes1_is_almost_subregion_of_bboxes2(
|
|
extracted_layout.element_coords,
|
|
inferred_layout.element_coords,
|
|
threshold=subregion_threshold,
|
|
)
|
|
inferred_to_remove_mask = (
|
|
np.logical_or(
|
|
inferred_is_subregion_of_extracted,
|
|
extracted_is_subregion_of_inferred.T,
|
|
)
|
|
.sum(axis=1)
|
|
.astype(bool)
|
|
)
|
|
# NOTE (yao): maybe we should expand those matching extracted region to contain the inferred
|
|
# regions it has subregion relationship with? like we did for inferred regions
|
|
inferred_to_keep[inferred_to_remove_mask] = False
|
|
return inferred_to_keep
|
|
|
|
|
|
@requires_dependencies("unstructured_inference")
|
|
def array_merge_inferred_layout_with_extracted_layout(
|
|
inferred_layout: LayoutElements,
|
|
extracted_layout: LayoutElements,
|
|
page_image_size: tuple,
|
|
same_region_threshold: float = inference_config.LAYOUT_SAME_REGION_THRESHOLD,
|
|
subregion_threshold: float = inference_config.LAYOUT_SUBREGION_THRESHOLD,
|
|
max_rounds: int = 5,
|
|
) -> LayoutElements:
|
|
"""merge elements using array data structures; it also returns LayoutElements instead of
|
|
collection of LayoutElement"""
|
|
from unstructured_inference.inference.layoutelement import LayoutElements
|
|
|
|
if len(extracted_layout) == 0:
|
|
return inferred_layout
|
|
if len(inferred_layout) == 0:
|
|
return extracted_layout
|
|
|
|
w, h = page_image_size
|
|
full_page_region = Rectangle(0, 0, w, h)
|
|
# ==== RULE 0: Full page extracted images are ignored
|
|
# non full page extracted image regions are kept, except when they match a non-text inferred
|
|
# region then we use the common bounding boxes and keep just one of the two sets (see rules
|
|
# below)
|
|
image_indices_to_keep = np.where(extracted_layout.element_class_ids == 1)[0]
|
|
if len(image_indices_to_keep):
|
|
full_page_image_mask = (
|
|
boxes_iou(
|
|
extracted_layout.slice(image_indices_to_keep).element_coords,
|
|
[full_page_region],
|
|
threshold=FULL_PAGE_REGION_THRESHOLD,
|
|
)
|
|
.sum(axis=1)
|
|
.astype(bool)
|
|
)
|
|
image_indices_to_keep = image_indices_to_keep[~full_page_image_mask]
|
|
|
|
# ==== RULE 1: any inferred box that is almost the same as an extracted image box, inferred is
|
|
# removed
|
|
# NOTE (yao): what if od model detects table but pdfminer says image -> we would lose the table
|
|
boxes_almost_same = (
|
|
boxes_iou(
|
|
inferred_layout.element_coords,
|
|
extracted_layout.slice(image_indices_to_keep).element_coords,
|
|
threshold=same_region_threshold,
|
|
)
|
|
.sum(axis=1)
|
|
.astype(bool)
|
|
)
|
|
|
|
# drop off those matching inferred from processing
|
|
inferred_layout_to_proc = inferred_layout.slice(~boxes_almost_same)
|
|
inferred_to_keep = np.array([True] * len(inferred_layout_to_proc))
|
|
|
|
# TODO (yao): experiment with all regions, not just text region, being potential targets to be
|
|
# merged into inferred elements
|
|
text_element_indices = np.where(extracted_layout.element_class_ids == 0)[0]
|
|
|
|
if len(text_element_indices) == 0:
|
|
return LayoutElements.concatenate(
|
|
(
|
|
inferred_layout_to_proc,
|
|
extracted_layout.slice(image_indices_to_keep),
|
|
)
|
|
)
|
|
|
|
if len(inferred_layout_to_proc) == 0:
|
|
return extracted_layout.slice(np.concatenate((image_indices_to_keep, text_element_indices)))
|
|
|
|
extracted_text_layouts = extracted_layout.slice(text_element_indices)
|
|
# ==== RULE 2. if there is a inferred region almost the same as the extracted text-region ->
|
|
# keep inferred and removed extracted region; here we put more trust in OD model more than
|
|
# pdfminer for bounding box
|
|
extracted_to_remove = _merge_extracted_into_inferred_when_almost_the_same(
|
|
extracted_text_layouts,
|
|
inferred_layout_to_proc,
|
|
same_region_threshold,
|
|
)
|
|
|
|
# ==== RULE 3. if extracted is subregion of an inferrred text region:
|
|
# remove extracted and keep inferred;
|
|
# expand inferred bounding box if needed to encompass all subregion extracted boxes
|
|
# NOTE (yao):
|
|
# currently this rule can fail to capture almost overlaps of two text regions when the pdfminer
|
|
# has larger bounding boxes (in area). It might be worth it to use simpler IOU thresholding or
|
|
# use the minimum of the two areas when computing sub regions
|
|
inferred_to_proc = _inferred_is_text(inferred_layout_to_proc)
|
|
extracted_to_proc = ~extracted_to_remove
|
|
rounds = 0
|
|
|
|
# because inferred layout sizes can be increased after one pass we may need to run through
|
|
# multiple passes; the original looped version increases layout size when it is processed so
|
|
# order would matter in that version. Here we loop over multiple times to avoid order being a
|
|
# factor -> this is one big difference between the current refactor and the version in inference
|
|
# lib that uses loops
|
|
while rounds < max_rounds and any(inferred_to_proc) and any(extracted_to_proc):
|
|
rounds += 1
|
|
inferred_to_proc_at_start = inferred_to_proc.copy()
|
|
extracted_to_proc_start = extracted_to_proc.copy()
|
|
|
|
extracted_is_subregion_of_inferred = bboxes1_is_almost_subregion_of_bboxes2(
|
|
extracted_text_layouts.element_coords,
|
|
inferred_layout_to_proc.element_coords,
|
|
threshold=subregion_threshold,
|
|
)
|
|
|
|
updated_inferred = _merge_extracted_that_are_subregion_of_inferred_text(
|
|
extracted_text_layouts.slice(extracted_to_proc),
|
|
inferred_layout_to_proc.slice(inferred_to_proc),
|
|
extracted_is_subregion_of_inferred[extracted_to_proc][:, inferred_to_proc],
|
|
# both those following two are modified in place in the function
|
|
extracted_to_proc,
|
|
inferred_to_proc,
|
|
)
|
|
# unfortunately slice uses "fancy" indexing and it generates a copy instead of a view, which
|
|
# was intentional by design to avoid unintended modification of the original data
|
|
inferred_layout_to_proc.element_coords[inferred_to_proc_at_start] = (
|
|
updated_inferred.element_coords
|
|
)
|
|
|
|
if np.array_equal(extracted_to_proc_start, extracted_to_proc) and np.array_equal(
|
|
inferred_to_proc_at_start, inferred_to_proc
|
|
):
|
|
break
|
|
|
|
# ==== RULE 4. if extracted is subregion of an inferred or inferred is subregion of extracted,
|
|
# except for inferrred tables, remove inferred and chose extracted
|
|
extracted_to_keep = np.concatenate(
|
|
(image_indices_to_keep, text_element_indices[extracted_to_proc])
|
|
)
|
|
if extracted_to_keep.size:
|
|
inferred_to_proc = np.logical_or(
|
|
inferred_to_proc,
|
|
_inferred_is_elementtype(
|
|
inferred_layout_to_proc,
|
|
[
|
|
ElementType.FIGURE,
|
|
ElementType.IMAGE,
|
|
ElementType.PICTURE,
|
|
],
|
|
),
|
|
)
|
|
inferred_to_keep[inferred_to_proc] = (
|
|
_mark_non_table_inferred_for_removal_if_has_subregion_relationship(
|
|
extracted_layout.slice(extracted_to_keep),
|
|
inferred_layout_to_proc.slice(inferred_to_proc),
|
|
inferred_to_keep[inferred_to_proc],
|
|
subregion_threshold,
|
|
)
|
|
)
|
|
|
|
# ==== RULE 5. all else -> keep extracted region; note we also keep extracted image regions
|
|
# that is a subregion of an inferred text region
|
|
extracted_to_keep.sort()
|
|
|
|
final_layout = LayoutElements.concatenate(
|
|
(
|
|
extracted_layout.slice(extracted_to_keep),
|
|
inferred_layout_to_proc.slice(inferred_to_keep),
|
|
)
|
|
)
|
|
return final_layout
|
|
|
|
|
|
def _ltchar_is_rotated(char: LTChar) -> bool:
|
|
# Calculate rotation angle in degrees
|
|
# For standard text: a=1, b=0, c=0, d=1 (no rotation)
|
|
rotation_radians = math.atan2(char.matrix[1], char.matrix[0])
|
|
# 0.001 is the tolerance for nearly flat angles; mainly for handling numerical precision
|
|
return abs(rotation_radians) > 0.001
|
|
|
|
|
|
def text_is_embedded(obj, threshold=env_config.PDF_MAX_EMBED_LOW_FIDELITY_TEXT_RATIO):
|
|
"""Check if text object contains too many low_fidelity text: invisible or rotated
|
|
|
|
Low fidelity text means that even though the text is extracted from pdf data but its
|
|
representation in the partitioned elements may require post processing to make senmatic sense.
|
|
This includes:
|
|
- invisible text: text not rendered on the pdf are not present visually when reading the page
|
|
so those texts may not be high quality information for understanding the page
|
|
- rotated text: text rotated usually are extracted in the order they appear in the dominant
|
|
reading order of the page (e.g., left->right, top->down). But if a text is rotated so the
|
|
last character is at the top (y position) and first character is at the bottom the extracted
|
|
element would contain words written in reverse order. This makes the extraction low quality.
|
|
"""
|
|
low_fidelity_chars = 0
|
|
total_chars = 0
|
|
|
|
def extract_chars(layout_obj):
|
|
"""Recursively extract all LTChar objects from layout."""
|
|
nonlocal low_fidelity_chars, total_chars
|
|
|
|
if isinstance(layout_obj, LTChar):
|
|
total_chars += 1
|
|
|
|
# Check if text is low_fidelity:
|
|
# - rendering mode 3 (requires custom pdf interpreter comes with this library)
|
|
# - text is rotated
|
|
if (
|
|
hasattr(layout_obj, "rendermode") and layout_obj.rendermode == 3
|
|
) or _ltchar_is_rotated(layout_obj):
|
|
low_fidelity_chars += 1
|
|
elif isinstance(layout_obj, LTContainer):
|
|
# Recursively process container's children
|
|
for child in layout_obj:
|
|
extract_chars(child)
|
|
|
|
extract_chars(obj)
|
|
if total_chars > 0:
|
|
# when there are no-trivial amount of hidden characters in the object it means there are
|
|
# text that is not rendered -> most likely OCR'ed text for the image content overlying the
|
|
# text and not embedded text that also shows in the rendered pdf
|
|
low_fidelity_ratio = low_fidelity_chars / total_chars
|
|
return low_fidelity_ratio < threshold
|
|
return True
|
|
|
|
|
|
@requires_dependencies("unstructured_inference")
|
|
def process_page_layout_from_pdfminer(
|
|
annotation_list: list,
|
|
page_layout,
|
|
page_height: int | float,
|
|
page_number: int,
|
|
coord_coef: float,
|
|
pdfminer_config: Optional[PDFMinerConfig] = None,
|
|
widget_list: Optional[list[dict[str, Any]]] = None,
|
|
) -> tuple[LayoutElements, list]:
|
|
from unstructured_inference.inference.layoutelement import LayoutElements
|
|
|
|
urls_metadata: list[dict[str, Any]] = []
|
|
element_coords, texts, element_class = [], [], []
|
|
is_extracted = []
|
|
annotation_threshold = env_config.PDF_ANNOTATION_THRESHOLD
|
|
|
|
for obj in page_layout:
|
|
x1, y1, x2, y2 = rect_to_bbox(obj.bbox, page_height)
|
|
bbox = (x1, y1, x2, y2)
|
|
|
|
if len(annotation_list) > 0 and isinstance(obj, LTTextBox):
|
|
annotations_within_element = check_annotations_within_element(
|
|
annotation_list,
|
|
bbox,
|
|
page_number,
|
|
annotation_threshold,
|
|
)
|
|
_, words = get_words_from_obj(obj, page_height)
|
|
for annot in annotations_within_element:
|
|
urls_metadata.append(map_bbox_and_index(words, annot))
|
|
|
|
if hasattr(obj, "get_text"):
|
|
inner_text_objects = extract_text_objects(obj)
|
|
char_dedup_threshold = env_config.PDF_CHAR_DUPLICATE_THRESHOLD
|
|
for inner_obj in inner_text_objects:
|
|
inner_bbox = rect_to_bbox(inner_obj.bbox, page_height)
|
|
if not _validate_bbox(inner_bbox):
|
|
continue
|
|
# Use deduplication to handle fake bold text (characters rendered twice)
|
|
texts.append(get_text_with_deduplication(inner_obj, char_dedup_threshold))
|
|
element_coords.append(inner_bbox)
|
|
element_class.append(0)
|
|
is_extracted.append(IsExtracted.TRUE if text_is_embedded(inner_obj) else None)
|
|
else:
|
|
inner_image_objects = extract_image_objects(obj)
|
|
for img_obj in inner_image_objects:
|
|
inner_bbox = rect_to_bbox(img_obj.bbox, page_height)
|
|
if not _validate_bbox(inner_bbox):
|
|
continue
|
|
texts.append(None)
|
|
element_coords.append(inner_bbox)
|
|
element_class.append(1)
|
|
is_extracted.append(None)
|
|
# A container without a `get_text` method (e.g. an `LTFigure` overlay) can still hold
|
|
# real, rendered text as loose `LTChar`s -- for example text drawn into a figure/XObject
|
|
# overlay rather than the main content stream -- which `extract_text_objects`
|
|
# (LTTextLine only) misses. Re-run pdfminer layout analysis on the container, reusing
|
|
# the same LAParams settings as the main pass plus `all_texts=True`, so those characters
|
|
# are grouped into `LTTextLine`s, then extract them through the same path as the main
|
|
# text branch above.
|
|
if isinstance(obj, LTContainer):
|
|
laparams_kwargs = (
|
|
pdfminer_config.model_dump(exclude_none=True) if pdfminer_config else {}
|
|
)
|
|
laparams_kwargs["all_texts"] = True
|
|
obj.analyze(LAParams(**laparams_kwargs))
|
|
char_dedup_threshold = env_config.PDF_CHAR_DUPLICATE_THRESHOLD
|
|
for inner_obj in extract_text_objects(obj):
|
|
inner_bbox = rect_to_bbox(inner_obj.bbox, page_height)
|
|
if not _validate_bbox(inner_bbox):
|
|
continue
|
|
texts.append(get_text_with_deduplication(inner_obj, char_dedup_threshold))
|
|
element_coords.append(inner_bbox)
|
|
element_class.append(0)
|
|
is_extracted.append(IsExtracted.TRUE if text_is_embedded(inner_obj) else None)
|
|
|
|
# Filled AcroForm field values live in widget annotations rather than the content
|
|
# stream, so add them here as extracted text regions (see get_widget_text_from_annots).
|
|
for widget in widget_list or []:
|
|
widget_bbox = widget["bbox"]
|
|
if not _validate_bbox(widget_bbox):
|
|
continue
|
|
texts.append(widget["text"])
|
|
element_coords.append(widget_bbox)
|
|
element_class.append(0)
|
|
is_extracted.append(IsExtracted.TRUE)
|
|
|
|
return (
|
|
LayoutElements(
|
|
element_coords=coord_coef * np.array(element_coords),
|
|
texts=np.array(texts).astype(object),
|
|
element_class_ids=np.array(element_class),
|
|
element_class_id_map={0: ElementType.UNCATEGORIZED_TEXT, 1: ElementType.IMAGE},
|
|
sources=np.array([Source.PDFMINER] * len(element_class)),
|
|
is_extracted_array=np.array(is_extracted),
|
|
),
|
|
urls_metadata,
|
|
)
|
|
|
|
|
|
@requires_dependencies("unstructured_inference")
|
|
def process_data_with_pdfminer(
|
|
file: Optional[Union[bytes, BinaryIO]] = None,
|
|
dpi: int = env_config.PDF_RENDER_DPI,
|
|
password: Optional[str] = None,
|
|
pdfminer_config: Optional[PDFMinerConfig] = None,
|
|
rotation_corrections: Optional[List[int]] = None,
|
|
) -> tuple[List[LayoutElements], List[List]]:
|
|
"""Loads the image and word objects from a pdf using pdfplumber and the image renderings of the
|
|
pdf pages using pdf2image
|
|
|
|
``rotation_corrections`` is an optional per-page list of extra rotations (degrees,
|
|
counter-clockwise) that unstructured-inference applied to the rendered page images to
|
|
make their text upright. Mirroring those rotations onto the extracted coordinates keeps
|
|
the pdfminer layer aligned with the object-detection layer.
|
|
"""
|
|
|
|
from unstructured_inference.inference.layoutelement import LayoutElements
|
|
|
|
layouts = []
|
|
layouts_links = []
|
|
# Coefficient to rescale bounding box to be compatible with images
|
|
coef = dpi / 72
|
|
for page_number, (page, page_layout) in enumerate(
|
|
open_pdfminer_pages_generator(file, password=password, pdfminer_config=pdfminer_config)
|
|
):
|
|
width, height = page_layout.width, page_layout.height
|
|
|
|
annotation_list = []
|
|
widget_list = []
|
|
coordinate_system = PixelSpace(
|
|
width=width,
|
|
height=height,
|
|
)
|
|
if page.annots:
|
|
annotation_list = get_uris(page.annots, height, coordinate_system, page_number)
|
|
widget_list = get_widget_text_from_annots(page.annots, height)
|
|
|
|
layout, urls_metadata = process_page_layout_from_pdfminer(
|
|
annotation_list, page_layout, height, page_number, coef, pdfminer_config, widget_list
|
|
)
|
|
|
|
# Mirror any image rotation unstructured-inference applied for this page so the
|
|
# extracted coordinates share the object-detection layer's frame (see _rotate_bboxes).
|
|
angle = (
|
|
rotation_corrections[page_number]
|
|
if rotation_corrections is not None and page_number < len(rotation_corrections)
|
|
else 0
|
|
)
|
|
if angle:
|
|
layout.element_coords = _rotate_bboxes(
|
|
layout.element_coords, angle, width * coef, height * coef
|
|
)
|
|
|
|
links = []
|
|
for metadata in urls_metadata:
|
|
bbox = [x * coef for x in metadata["bbox"]]
|
|
if angle:
|
|
bbox = _rotate_bboxes(
|
|
np.array([bbox], dtype=float), angle, width * coef, height * coef
|
|
)[0].tolist()
|
|
links.append(
|
|
{
|
|
"bbox": bbox,
|
|
"text": metadata["text"],
|
|
"url": metadata["uri"],
|
|
"start_index": metadata["start_index"],
|
|
}
|
|
)
|
|
|
|
clean_layouts = []
|
|
for threshold, element_class in zip(
|
|
(
|
|
env_config.EMBEDDED_TEXT_SAME_REGION_THRESHOLD,
|
|
env_config.EMBEDDED_IMAGE_SAME_REGION_THRESHOLD,
|
|
),
|
|
(0, 1),
|
|
):
|
|
elements_to_sort = layout.slice(layout.element_class_ids == element_class)
|
|
clean_layouts.append(
|
|
remove_duplicate_elements(elements_to_sort, threshold)
|
|
if len(elements_to_sort)
|
|
else elements_to_sort
|
|
)
|
|
|
|
layout = LayoutElements.concatenate(clean_layouts)
|
|
# NOTE(christine): always do the basic sort first for deterministic order across
|
|
# python versions.
|
|
layout = sort_text_regions(layout, SORT_MODE_BASIC)
|
|
|
|
# apply the current default sorting to the layout elements extracted by pdfminer
|
|
layout = sort_text_regions(layout)
|
|
|
|
layouts.append(layout)
|
|
layouts_links.append(links)
|
|
return layouts, layouts_links
|
|
|
|
|
|
def _create_text_region(x1, y1, x2, y2, coef, text, source, region_class):
|
|
"""Creates a text region of the specified class with scaled coordinates."""
|
|
return region_class.from_coords(
|
|
x1 * coef,
|
|
y1 * coef,
|
|
x2 * coef,
|
|
y2 * coef,
|
|
text=text,
|
|
source=source,
|
|
)
|
|
|
|
|
|
def get_coords_from_bboxes(bboxes, round_to: int = DEFAULT_ROUND) -> np.ndarray:
|
|
"""convert a list of boxes's coords into np array"""
|
|
if isinstance(bboxes, np.ndarray):
|
|
return bboxes.round(round_to)
|
|
|
|
# preallocate memory
|
|
coords = np.zeros((len(bboxes), 4), dtype=np.float32)
|
|
|
|
for i, bbox in enumerate(bboxes):
|
|
coords[i, :] = [bbox.x1, bbox.y1, bbox.x2, bbox.y2]
|
|
|
|
return coords.round(round_to)
|
|
|
|
|
|
def areas_of_boxes_and_intersection_area(
|
|
coords1: np.ndarray, coords2: np.ndarray, round_to: int = DEFAULT_ROUND
|
|
):
|
|
"""compute intersection area and own areas for two groups of bounding boxes"""
|
|
x11, y11, x12, y12 = np.split(coords1, 4, axis=1)
|
|
x21, y21, x22, y22 = np.split(coords2, 4, axis=1)
|
|
|
|
inter_area = np.maximum(
|
|
(np.minimum(x12, np.transpose(x22)) - np.maximum(x11, np.transpose(x21)) + 1), 0
|
|
) * np.maximum((np.minimum(y12, np.transpose(y22)) - np.maximum(y11, np.transpose(y21)) + 1), 0)
|
|
boxa_area = (x12 - x11 + 1) * (y12 - y11 + 1)
|
|
boxb_area = (x22 - x21 + 1) * (y22 - y21 + 1)
|
|
|
|
return inter_area.round(round_to), boxa_area.round(round_to), boxb_area.round(round_to)
|
|
|
|
|
|
def bboxes1_is_almost_subregion_of_bboxes2(
|
|
bboxes1, bboxes2, threshold: float = 0.5, round_to: int = DEFAULT_ROUND
|
|
) -> np.ndarray:
|
|
"""compute if each element from bboxes1 is almost a subregion of one or more elements in
|
|
bboxes2"""
|
|
coords1 = get_coords_from_bboxes(bboxes1, round_to=round_to)
|
|
coords2 = get_coords_from_bboxes(bboxes2, round_to=round_to)
|
|
|
|
inter_area, boxa_area, boxb_area = areas_of_boxes_and_intersection_area(
|
|
coords1, coords2, round_to=round_to
|
|
)
|
|
|
|
return (inter_area / np.maximum(boxa_area, EPSILON_AREA) > threshold) & (
|
|
boxa_area <= boxb_area.T
|
|
)
|
|
|
|
|
|
def boxes_self_iou(bboxes, threshold: float = 0.5, round_to: int = DEFAULT_ROUND) -> np.ndarray:
|
|
"""compute iou for a group of elements"""
|
|
# only store one copy of coords in memory instead of calling get coords twice
|
|
coords = get_coords_from_bboxes(bboxes, round_to=round_to)
|
|
|
|
return boxes_iou(coords, coords, threshold, round_to)
|
|
|
|
|
|
# TODO (yao): move those vector math utils into a separated sub module to void import issues
|
|
def boxes_iou(
|
|
bboxes1, bboxes2, threshold: float = 0.75, round_to: int = DEFAULT_ROUND
|
|
) -> np.ndarray:
|
|
"""compute iou between two groups of elements"""
|
|
coords1 = get_coords_from_bboxes(bboxes1, round_to=round_to)
|
|
coords2 = get_coords_from_bboxes(bboxes2, round_to=round_to)
|
|
|
|
inter_area, boxa_area, boxb_area = areas_of_boxes_and_intersection_area(
|
|
coords1, coords2, round_to=round_to
|
|
)
|
|
denom = np.maximum(EPSILON_AREA, boxa_area + boxb_area.T - inter_area)
|
|
# Instead of (x/y) > t, use x > t*y for memory & speed with same result
|
|
return inter_area > (threshold * denom)
|
|
|
|
|
|
@requires_dependencies("unstructured_inference")
|
|
def pdfminer_elements_to_text_regions(layout_elements: LayoutElements) -> list[TextRegions]:
|
|
"""a temporary solution to convert layout elements to a list of either EmbeddedTextRegion or
|
|
ImageTextRegion; this should be made obsolete after we refactor the merging logic in inference
|
|
library"""
|
|
from unstructured_inference.inference.elements import (
|
|
EmbeddedTextRegion,
|
|
ImageTextRegion,
|
|
)
|
|
|
|
regions = []
|
|
for i, element_class in enumerate(layout_elements.element_class_ids):
|
|
region_class = EmbeddedTextRegion if element_class == 0 else ImageTextRegion
|
|
regions.append(
|
|
region_class.from_coords(
|
|
*layout_elements.element_coords[i],
|
|
text=layout_elements.texts[i],
|
|
source=Source.PDFMINER,
|
|
)
|
|
)
|
|
return regions
|
|
|
|
|
|
@requires_dependencies("unstructured_inference")
|
|
def merge_inferred_with_extracted_layout(
|
|
inferred_document_layout: "DocumentLayout",
|
|
extracted_layout: List[TextRegions],
|
|
hi_res_model_name: str,
|
|
) -> "DocumentLayout":
|
|
"""Merge an inferred layout with an extracted layout"""
|
|
|
|
from unstructured_inference.models.detectron2onnx import UnstructuredDetectronONNXModel
|
|
|
|
inferred_pages = inferred_document_layout.pages
|
|
for i, (inferred_page, extracted_page_layout) in enumerate(
|
|
zip(inferred_pages, extracted_layout)
|
|
):
|
|
image_metadata = inferred_page.image_metadata
|
|
w = image_metadata.get("width")
|
|
h = image_metadata.get("height")
|
|
image_size = (w, h)
|
|
|
|
threshold_kwargs = {}
|
|
# NOTE(Benjamin): With this the thresholds are only changed for detextron2_mask_rcnn
|
|
# In other case the default values for the functions are used
|
|
if (
|
|
isinstance(inferred_page.detection_model, UnstructuredDetectronONNXModel)
|
|
and "R_50" not in inferred_page.detection_model.model_path
|
|
):
|
|
threshold_kwargs = {"same_region_threshold": 0.5, "subregion_threshold": 0.5}
|
|
|
|
# NOTE (yao): after refactoring the algorithm to be vectorized we can then pass in the
|
|
# vectorized data structure into the merge function
|
|
|
|
merged_layout = array_merge_inferred_layout_with_extracted_layout(
|
|
inferred_page.elements_array,
|
|
extracted_page_layout,
|
|
page_image_size=image_size,
|
|
**threshold_kwargs,
|
|
)
|
|
|
|
merged_layout = sort_text_regions(merged_layout, SORT_MODE_BASIC)
|
|
# so that we can modify the text without worrying about hitting length limit
|
|
merged_layout.texts = merged_layout.texts.astype(object)
|
|
merged_layout.is_extracted_array = merged_layout.is_extracted_array.astype(object)
|
|
for i, text in enumerate(merged_layout.texts):
|
|
if text is None:
|
|
text, is_extracted = aggregate_embedded_text_by_block(
|
|
target_region=merged_layout.slice([i]),
|
|
source_regions=extracted_page_layout,
|
|
)
|
|
if merged_layout.element_class_id_map[merged_layout.element_class_ids[i]] not in (
|
|
"Image",
|
|
"Picture",
|
|
):
|
|
merged_layout.is_extracted_array[i] = is_extracted
|
|
merged_layout.texts[i] = remove_control_characters(text)
|
|
|
|
inferred_page.elements_array = merged_layout
|
|
|
|
return inferred_document_layout
|
|
|
|
|
|
def clean_pdfminer_inner_elements(document: "DocumentLayout") -> "DocumentLayout":
|
|
"""Clean pdfminer elements from inside tables.
|
|
|
|
This function removes elements sourced from PDFMiner that are subregions within table elements.
|
|
"""
|
|
|
|
for page in document.pages:
|
|
pdfminer_mask = page.elements_array.sources == Source.PDFMINER
|
|
non_pdfminer_element_boxes = page.elements_array.slice(~pdfminer_mask).element_coords
|
|
pdfminer_element_boxes = page.elements_array.slice(pdfminer_mask).element_coords
|
|
|
|
if len(pdfminer_element_boxes) == 0 or len(non_pdfminer_element_boxes) == 0:
|
|
continue
|
|
|
|
is_element_subregion_of_other_elements = (
|
|
bboxes1_is_almost_subregion_of_bboxes2(
|
|
pdfminer_element_boxes,
|
|
non_pdfminer_element_boxes,
|
|
env_config.EMBEDDED_TEXT_AGGREGATION_SUBREGION_THRESHOLD,
|
|
).sum(axis=1)
|
|
== 1
|
|
)
|
|
|
|
pdfminer_to_keep = np.where(pdfminer_mask)[0][~is_element_subregion_of_other_elements]
|
|
page.elements_array = page.elements_array.slice(
|
|
np.sort(np.concatenate((np.where(~pdfminer_mask)[0], pdfminer_to_keep)))
|
|
)
|
|
|
|
return document
|
|
|
|
|
|
@requires_dependencies("unstructured_inference")
|
|
def remove_duplicate_elements(
|
|
elements: TextRegions,
|
|
threshold: float = 0.5,
|
|
) -> TextRegions:
|
|
"""Removes duplicate text elements extracted by PDFMiner from a document layout."""
|
|
|
|
coords = elements.element_coords
|
|
# experiments show 2e3 is the block size that constrains the peak memory around 1Gb for this
|
|
# function; that accounts for all the intermediate matricies allocated and memory for storing
|
|
# final results
|
|
memory_cap_in_gb = float(os.getenv("UNST_MATMUL_MEMORY_CAP_IN_GB", 1))
|
|
if memory_cap_in_gb <= 0:
|
|
raise ValueError("UNST_MATMUL_MEMORY_CAP_IN_GB must be > 0")
|
|
n_split = np.ceil(coords.shape[0] / 2e3 / memory_cap_in_gb)
|
|
splits = np.array_split(coords, n_split, axis=0)
|
|
|
|
# A box is dropped only when it near-duplicates a *later* box (higher global index) -- the
|
|
# strict upper triangle of the full IoU matrix. Each split is a contiguous block of rows
|
|
# compared against all coords, so the triangle's diagonal must be offset by the split's
|
|
# global start index; otherwise rows in later splits match themselves (and earlier boxes)
|
|
# and get wrongly removed, decimating dense pages (> 2000 elements).
|
|
keep_masks = []
|
|
offset = 0
|
|
for split in splits:
|
|
iou = boxes_iou(split, coords, threshold)
|
|
keep_masks.append(~np.triu(iou, k=1 + offset).any(axis=1))
|
|
offset += split.shape[0]
|
|
return elements.slice(np.concatenate(keep_masks))
|
|
|
|
|
|
def _aggregated_iou(box1s, box2):
|
|
intersection = 0.0
|
|
sum_areas = calculate_bbox_area(box2)
|
|
|
|
for i in range(box1s.shape[0]):
|
|
intersection += calculate_intersection_area(box1s[i, :], box2)
|
|
sum_areas += calculate_bbox_area(box1s[i, :])
|
|
|
|
union = sum_areas - intersection
|
|
|
|
if union == 0:
|
|
return 1.0
|
|
return intersection / union
|
|
|
|
|
|
def aggregate_embedded_text_by_block(
|
|
target_region: TextRegions,
|
|
source_regions: TextRegions,
|
|
subregion_threshold: float = env_config.EMBEDDED_TEXT_AGGREGATION_SUBREGION_THRESHOLD,
|
|
text_coverage_threshold: float = env_config.TEXT_COVERAGE_THRESHOLD,
|
|
) -> tuple[str, IsExtracted | None]:
|
|
"""Extracts the text aggregated from the elements of the given layout that lie within the given
|
|
block."""
|
|
|
|
if len(source_regions) == 0 or len(target_region) == 0:
|
|
return "", None
|
|
|
|
mask = (
|
|
bboxes1_is_almost_subregion_of_bboxes2(
|
|
source_regions.element_coords,
|
|
target_region.element_coords,
|
|
subregion_threshold,
|
|
)
|
|
.sum(axis=1)
|
|
.astype(bool)
|
|
)
|
|
|
|
text = " ".join([text for text in source_regions.slice(mask).texts if text])
|
|
|
|
if sum(mask):
|
|
source_bboxes = source_regions.slice(mask).element_coords
|
|
target_bboxes = target_region.element_coords
|
|
|
|
iou = _aggregated_iou(source_bboxes, target_bboxes[0, :])
|
|
|
|
fully_filled = (
|
|
all(flag == IsExtracted.TRUE for flag in source_regions.slice(mask).is_extracted_array)
|
|
and iou > text_coverage_threshold
|
|
)
|
|
is_extracted = IsExtracted.TRUE if fully_filled else IsExtracted.PARTIAL
|
|
else:
|
|
# if nothing is sliced then it is not extracted
|
|
is_extracted = IsExtracted.FALSE
|
|
return text, is_extracted
|
|
|
|
|
|
def get_links_in_element(page_links: list, region: Rectangle) -> list:
|
|
links_bboxes = [Rectangle(*link.get("bbox")) for link in page_links]
|
|
results = bboxes1_is_almost_subregion_of_bboxes2(links_bboxes, [region])
|
|
links = [
|
|
{
|
|
"text": page_links[idx].get("text"),
|
|
"url": page_links[idx].get("url"),
|
|
"start_index": page_links[idx].get("start_index"),
|
|
}
|
|
for idx, result in enumerate(results)
|
|
if any(result)
|
|
]
|
|
|
|
return links
|
|
|
|
|
|
def get_uris(
|
|
annots: PDFObjRef | list[PDFObjRef],
|
|
height: float,
|
|
coordinate_system: PixelSpace | PointSpace,
|
|
page_number: int,
|
|
) -> list[dict[str, Any]]:
|
|
"""
|
|
Extracts URI annotations from a single or a list of PDF object references on a specific page.
|
|
The type of annots (list or not) depends on the pdf formatting. The function detectes the type
|
|
of annots and then pass on to get_uris_from_annots function as a list.
|
|
|
|
Args:
|
|
annots (PDFObjRef | list[PDFObjRef]): A single or a list of PDF object references
|
|
representing annotations on the page.
|
|
height (float): The height of the page in the specified coordinate system.
|
|
coordinate_system (PixelSpace | PointSpace): The coordinate system used to represent
|
|
the annotations' coordinates.
|
|
page_number (int): The page number from which to extract annotations.
|
|
|
|
Returns:
|
|
list[dict]: A list of dictionaries, each containing information about a URI annotation,
|
|
including its coordinates, bounding box, type, URI link, and page number.
|
|
"""
|
|
if isinstance(annots, list):
|
|
return get_uris_from_annots(annots, height, coordinate_system, page_number)
|
|
resolved_annots = annots.resolve()
|
|
if resolved_annots is None:
|
|
return []
|
|
return get_uris_from_annots(resolved_annots, height, coordinate_system, page_number)
|
|
|
|
|
|
def get_uris_from_annots(
|
|
annots: list[PDFObjRef],
|
|
height: int | float,
|
|
coordinate_system: PixelSpace | PointSpace,
|
|
page_number: int,
|
|
) -> list[dict[str, Any]]:
|
|
"""
|
|
Extracts URI annotations from a list of PDF object references.
|
|
|
|
Args:
|
|
annots (list[PDFObjRef]): A list of PDF object references representing annotations on
|
|
a page.
|
|
height (int | float): The height of the page in the specified coordinate system.
|
|
coordinate_system (PixelSpace | PointSpace): The coordinate system used to represent
|
|
the annotations' coordinates.
|
|
page_number (int): The page number from which to extract annotations.
|
|
|
|
Returns:
|
|
list[dict]: A list of dictionaries, each containing information about a URI annotation,
|
|
including its coordinates, bounding box, type, URI link, and page number.
|
|
"""
|
|
annotation_list = []
|
|
for annotation in annots:
|
|
# Check annotation is valid for extraction
|
|
annotation_dict = try_resolve(annotation)
|
|
if not isinstance(annotation_dict, dict):
|
|
continue
|
|
subtype = annotation_dict.get("Subtype", None)
|
|
if not subtype or isinstance(subtype, PDFObjRef) or str(subtype) != "/'Link'":
|
|
continue
|
|
# Extract bounding box and update coordinates
|
|
rect = annotation_dict.get("Rect", None)
|
|
if not rect or isinstance(rect, PDFObjRef) or len(rect) != 4:
|
|
continue
|
|
x1, y1, x2, y2 = rect_to_bbox(rect, height)
|
|
points = ((x1, y1), (x1, y2), (x2, y2), (x2, y1))
|
|
coordinates_metadata = CoordinatesMetadata(
|
|
points=points,
|
|
system=coordinate_system,
|
|
)
|
|
# Extract type
|
|
if "A" not in annotation_dict:
|
|
continue
|
|
uri_dict = try_resolve(annotation_dict["A"])
|
|
if not isinstance(uri_dict, dict):
|
|
continue
|
|
uri_type = None
|
|
if "S" in uri_dict and not isinstance(uri_dict["S"], PDFObjRef):
|
|
uri_type = str(uri_dict["S"])
|
|
# Extract URI link
|
|
uri = None
|
|
try:
|
|
if uri_type == "/'URI'":
|
|
uri = try_resolve(try_resolve(uri_dict["URI"])).decode("utf-8")
|
|
if uri_type == "/'GoTo'":
|
|
uri = try_resolve(try_resolve(uri_dict["D"])).decode("utf-8")
|
|
except Exception:
|
|
pass
|
|
|
|
annotation_list.append(
|
|
{
|
|
"coordinates": coordinates_metadata,
|
|
"bbox": (x1, y1, x2, y2),
|
|
"type": uri_type,
|
|
"uri": uri,
|
|
"page_number": page_number,
|
|
},
|
|
)
|
|
return annotation_list
|
|
|
|
|
|
def try_resolve(annot: PDFObjRef):
|
|
"""
|
|
Attempt to resolve a PDF object reference. If successful, returns the resolved object;
|
|
otherwise, returns the original reference.
|
|
"""
|
|
try:
|
|
return annot.resolve()
|
|
except Exception:
|
|
return annot
|
|
|
|
|
|
def _decode_scalar_field_value(value: Any) -> Optional[str]:
|
|
"""Decode a single AcroForm field value into text.
|
|
|
|
PDF text strings may be UTF-16 or PDFDocEncoded; choice-field export values can
|
|
arrive as name objects (PSLiteral).
|
|
"""
|
|
if isinstance(value, bytes):
|
|
return decode_text(value)
|
|
if isinstance(value, str):
|
|
return value
|
|
name = getattr(value, "name", None) # PSLiteral (e.g. choice export value)
|
|
if isinstance(name, bytes):
|
|
return name.decode("utf-8", "replace")
|
|
if isinstance(name, str):
|
|
return name
|
|
return None
|
|
|
|
|
|
def _decode_field_value(value: Any) -> Optional[str]:
|
|
"""Decode an AcroForm field value into text."""
|
|
value = try_resolve(value)
|
|
if isinstance(value, (list, tuple)):
|
|
decoded_values = [
|
|
text.strip()
|
|
for item in value
|
|
if (text := _decode_scalar_field_value(try_resolve(item))) and text.strip()
|
|
]
|
|
return "\n".join(decoded_values) if decoded_values else None
|
|
return _decode_scalar_field_value(value)
|
|
|
|
|
|
def get_widget_text_from_annots(
|
|
annots: PDFObjRef | list[PDFObjRef],
|
|
height: float,
|
|
) -> list[dict[str, Any]]:
|
|
"""Extract text from filled AcroForm widget annotations (fillable form fields).
|
|
|
|
pdfminer's page layout only covers the page content stream, so values typed into
|
|
fillable form fields are invisible to the normal text pass -- they live in widget
|
|
annotation objects (``/Annots``), not in the content stream. This recovers the value
|
|
text and bounding box for text (``/Tx``) and choice (``/Ch``) fields so they can be
|
|
emitted as elements alongside the content-stream text.
|
|
|
|
Returns a list of ``{"text", "bbox"}`` dicts, where ``bbox`` is ``(x1, y1, x2, y2)``
|
|
in the top-left page coordinate frame (same as ``rect_to_bbox``).
|
|
"""
|
|
resolved = annots if isinstance(annots, list) else try_resolve(annots)
|
|
if not isinstance(resolved, list):
|
|
return []
|
|
|
|
results: list[dict[str, Any]] = []
|
|
for annotation in resolved:
|
|
annotation_dict = try_resolve(annotation)
|
|
if not isinstance(annotation_dict, dict):
|
|
continue
|
|
if getattr(annotation_dict.get("Subtype"), "name", None) != "Widget":
|
|
continue
|
|
|
|
# Field type (FT) and value (V) may be inherited from a parent field node, so walk
|
|
# up the hierarchy until both are found (bounded to avoid cycles).
|
|
field_type = annotation_dict.get("FT")
|
|
value = annotation_dict.get("V")
|
|
parent = annotation_dict.get("Parent")
|
|
seen = 0
|
|
while (field_type is None or value is None) and parent is not None and seen < 32:
|
|
parent_dict = try_resolve(parent)
|
|
seen += 1
|
|
if not isinstance(parent_dict, dict):
|
|
break
|
|
field_type = field_type or parent_dict.get("FT")
|
|
value = value or parent_dict.get("V")
|
|
parent = parent_dict.get("Parent")
|
|
|
|
if getattr(field_type, "name", None) not in ("Tx", "Ch"):
|
|
continue
|
|
|
|
text = _decode_field_value(value)
|
|
if not text or not text.strip():
|
|
continue
|
|
|
|
rect = annotation_dict.get("Rect")
|
|
if not rect or isinstance(rect, PDFObjRef) or len(rect) != 4:
|
|
continue
|
|
try:
|
|
bbox = rect_to_bbox(tuple(float(v) for v in rect), height)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
|
|
results.append({"text": text.strip(), "bbox": bbox})
|
|
|
|
return results
|
|
|
|
|
|
def check_annotations_within_element(
|
|
annotation_list: list[dict[str, Any]],
|
|
element_bbox: tuple[float, float, float, float],
|
|
page_number: int,
|
|
annotation_threshold: float,
|
|
) -> list[dict[str, Any]]:
|
|
"""
|
|
Filter annotations that are within or highly overlap with a specified element on a page.
|
|
|
|
Args:
|
|
annotation_list (list[dict[str,Any]]): A list of dictionaries, each containing information
|
|
about an annotation.
|
|
element_bbox (tuple[float, float, float, float]): The bounding box coordinates of the
|
|
specified element in the bbox format (x1, y1, x2, y2).
|
|
page_number (int): The page number to which the annotations and element belong.
|
|
annotation_threshold (float, optional): The threshold value (between 0.0 and 1.0)
|
|
that determines the minimum overlap required for an annotation to be considered
|
|
within the element. Default is 0.9.
|
|
|
|
Returns:
|
|
list[dict[str,Any]]: A list of dictionaries containing information about annotations
|
|
that are within or highly overlap with the specified element on the given page, based on
|
|
the specified threshold.
|
|
"""
|
|
annotations_within_element = []
|
|
for annotation in annotation_list:
|
|
if annotation["page_number"] == page_number:
|
|
annotation_bbox_size = calculate_bbox_area(annotation["bbox"])
|
|
if annotation_bbox_size and (
|
|
calculate_intersection_area(element_bbox, annotation["bbox"]) / annotation_bbox_size
|
|
> annotation_threshold
|
|
):
|
|
annotations_within_element.append(annotation)
|
|
return annotations_within_element
|
|
|
|
|
|
def _deduplicate_ltchars(
|
|
chars: list[LTChar],
|
|
threshold: float,
|
|
) -> list[LTChar]:
|
|
"""Remove duplicate characters caused by fake bold rendering.
|
|
|
|
Some PDFs create bold text by rendering the same character twice at slightly offset
|
|
positions. This function removes such duplicates.
|
|
|
|
Args:
|
|
chars: List of LTChar objects to deduplicate.
|
|
threshold: Maximum pixel distance to consider characters as duplicates.
|
|
Set to 0 to disable deduplication.
|
|
|
|
Returns:
|
|
Deduplicated list of LTChar objects.
|
|
"""
|
|
if threshold <= 0 or not chars:
|
|
return chars
|
|
|
|
result = [chars[0]]
|
|
for char in chars[1:]:
|
|
if not _is_duplicate_char(result[-1], char, threshold):
|
|
result.append(char)
|
|
return result
|
|
|
|
|
|
def get_words_from_obj(
|
|
obj: LTTextBox,
|
|
height: float,
|
|
) -> tuple[list[LTChar], list[dict[str, Any]]]:
|
|
"""
|
|
Extracts characters and word bounding boxes from a PDF text element.
|
|
|
|
Args:
|
|
obj (LTTextBox): The PDF text element from which to extract characters and words.
|
|
height (float): The height of the page in the specified coordinate system.
|
|
|
|
Returns:
|
|
tuple[list[LTChar], list[dict[str,Any]]]: A tuple containing two lists:
|
|
- list[LTChar]: A list of LTChar objects representing individual characters.
|
|
- list[dict[str,Any]]]: A list of dictionaries, each containing information about
|
|
a word, including its text, bounding box, and start index in the element's text.
|
|
"""
|
|
characters = []
|
|
words = []
|
|
text_len = 0
|
|
char_dedup_threshold = env_config.PDF_CHAR_DUPLICATE_THRESHOLD
|
|
|
|
for text_line in obj:
|
|
word = ""
|
|
x1, y1, x2, y2 = None, None, None, None
|
|
start_index = 0
|
|
last_char: LTChar | None = None # Track last character for deduplication
|
|
|
|
for index, character in enumerate(text_line):
|
|
if isinstance(character, LTChar):
|
|
# Skip duplicate characters (fake bold fix)
|
|
if (
|
|
char_dedup_threshold > 0
|
|
and last_char is not None
|
|
and _is_duplicate_char(last_char, character, char_dedup_threshold)
|
|
):
|
|
continue
|
|
|
|
last_char = character
|
|
characters.append(character)
|
|
char = character.get_text()
|
|
|
|
if word and not char.strip():
|
|
words.append(
|
|
{"text": word, "bbox": (x1, y1, x2, y2), "start_index": start_index},
|
|
)
|
|
word = ""
|
|
continue
|
|
|
|
# TODO(klaijan) - isalnum() only works with A-Z, a-z and 0-9
|
|
# will need to switch to some pattern matching once we support more languages
|
|
if not word:
|
|
isalnum = char.isalnum()
|
|
if word and char.isalnum() != isalnum:
|
|
isalnum = char.isalnum()
|
|
words.append(
|
|
{"text": word, "bbox": (x1, y1, x2, y2), "start_index": start_index},
|
|
)
|
|
word = ""
|
|
|
|
if len(word) == 0:
|
|
start_index = text_len + index
|
|
x1 = character.x0
|
|
y2 = height - character.y0
|
|
x2 = character.x1
|
|
y1 = height - character.y1
|
|
else:
|
|
x2 = character.x1
|
|
y2 = height - character.y0
|
|
|
|
word += char
|
|
else:
|
|
# Non-LTChar items (e.g., LTAnno) act as word boundaries
|
|
words.append(
|
|
{"text": word, "bbox": (x1, y1, x2, y2), "start_index": start_index},
|
|
)
|
|
word = ""
|
|
text_len += len(text_line)
|
|
return characters, words
|
|
|
|
|
|
def map_bbox_and_index(words: list[dict[str, Any]], annot: dict[str, Any]):
|
|
"""
|
|
Maps a bounding box annotation to the corresponding text and start index within a list of words.
|
|
|
|
Args:
|
|
words (list[dict[str,Any]]): A list of dictionaries, each containing information about
|
|
a word, including its text, bounding box, and start index.
|
|
annot (dict[str,Any]): The annotation dictionary to be mapped, which will be updated with
|
|
"text" and "start_index" fields.
|
|
|
|
Returns:
|
|
dict: The updated annotation dictionary with "text" representing the mapped text and
|
|
"start_index" representing the start index of the mapped text in the list of words.
|
|
"""
|
|
if len(words) == 0:
|
|
annot["text"] = ""
|
|
annot["start_index"] = -1
|
|
return annot
|
|
distance_from_bbox_start = np.sqrt(
|
|
(annot["bbox"][0] - np.array([word["bbox"][0] for word in words])) ** 2
|
|
+ (annot["bbox"][1] - np.array([word["bbox"][1] for word in words])) ** 2,
|
|
)
|
|
distance_from_bbox_end = np.sqrt(
|
|
(annot["bbox"][2] - np.array([word["bbox"][2] for word in words])) ** 2
|
|
+ (annot["bbox"][3] - np.array([word["bbox"][3] for word in words])) ** 2,
|
|
)
|
|
closest_start = try_argmin(distance_from_bbox_start)
|
|
closest_end = try_argmin(distance_from_bbox_end)
|
|
|
|
# NOTE(klaijan) - get the word from closest start only if the end index comes after start index
|
|
text = ""
|
|
if closest_end >= closest_start:
|
|
for _ in range(closest_start, closest_end + 1):
|
|
text += " "
|
|
text += words[_]["text"]
|
|
else:
|
|
text = words[closest_start]["text"]
|
|
|
|
annot["text"] = text.strip()
|
|
annot["start_index"] = words[closest_start]["start_index"]
|
|
return annot
|
|
|
|
|
|
def calculate_intersection_area(
|
|
bbox1: tuple[float, float, float, float],
|
|
bbox2: tuple[float, float, float, float],
|
|
) -> float:
|
|
"""
|
|
Calculate the area of intersection between two bounding boxes.
|
|
|
|
Args:
|
|
bbox1 (tuple[float, float, float, float]): The coordinates of the first bounding box
|
|
in the format (x1, y1, x2, y2).
|
|
bbox2 (tuple[float, float, float, float]): The coordinates of the second bounding box
|
|
in the format (x1, y1, x2, y2).
|
|
|
|
Returns:
|
|
float: The area of intersection between the two bounding boxes. If there is no
|
|
intersection, the function returns 0.0.
|
|
"""
|
|
x1_1, y1_1, x2_1, y2_1 = bbox1
|
|
x1_2, y1_2, x2_2, y2_2 = bbox2
|
|
|
|
x_intersection = max(x1_1, x1_2)
|
|
y_intersection = max(y1_1, y1_2)
|
|
x2_intersection = min(x2_1, x2_2)
|
|
y2_intersection = min(y2_1, y2_2)
|
|
|
|
if x_intersection < x2_intersection and y_intersection < y2_intersection:
|
|
intersection_area = calculate_bbox_area(
|
|
(x_intersection, y_intersection, x2_intersection, y2_intersection),
|
|
)
|
|
return intersection_area
|
|
else:
|
|
return 0.0
|
|
|
|
|
|
def calculate_bbox_area(bbox: tuple[float, float, float, float]) -> float:
|
|
"""
|
|
Calculate the area of a bounding box.
|
|
|
|
Args:
|
|
bbox (tuple[float, float, float, float]): The coordinates of the bounding box
|
|
in the format (x1, y1, x2, y2).
|
|
|
|
Returns:
|
|
float: The area of the bounding box, computed as the product of its width and height.
|
|
"""
|
|
x1, y1, x2, y2 = bbox
|
|
area = (x2 - x1) * (y2 - y1)
|
|
return area
|
|
|
|
|
|
def try_argmin(array: np.ndarray) -> int:
|
|
"""
|
|
Attempt to find the index of the minimum value in a NumPy array.
|
|
|
|
Args:
|
|
array (np.ndarray): The NumPy array in which to find the minimum value's index.
|
|
|
|
Returns:
|
|
int: The index of the minimum value in the array. If the array is empty or an
|
|
IndexError occurs, it returns -1.
|
|
"""
|
|
try:
|
|
return int(np.argmin(array))
|
|
except IndexError:
|
|
return -1
|