Initial commit of Roster

This commit is contained in:
Media
2025-06-28 01:12:46 -05:00
commit c23d3a7676
4367 changed files with 890519 additions and 0 deletions
@@ -0,0 +1,18 @@
__all__ = [
"__version__",
"utils",
"pdfminer",
"open",
"repair",
"set_debug",
]
import pdfminer
import pdfminer.pdftypes
from . import utils
from ._version import __version__
from .pdf import PDF
from .repair import repair
open = PDF.open
@@ -0,0 +1,10 @@
from typing import Any, Dict, Iterable, List, Literal, Sequence, Tuple, Union
T_seq = Sequence
T_num = Union[int, float]
T_point = Tuple[T_num, T_num]
T_bbox = Tuple[T_num, T_num, T_num, T_num]
T_obj = Dict[str, Any]
T_obj_list = List[T_obj]
T_obj_iter = Iterable[T_obj]
T_dir = Union[Literal["ltr"], Literal["rtl"], Literal["ttb"], Literal["btt"]]
@@ -0,0 +1,2 @@
version_info = (0, 11, 7)
__version__ = ".".join(map(str, version_info))
@@ -0,0 +1,130 @@
#!/usr/bin/env python
import argparse
import json
import sys
from collections import defaultdict, deque
from itertools import chain
from typing import Any, DefaultDict, Dict, List
from .pdf import PDF
if len(sys.argv) == 1:
sys.argv.append("--help")
def parse_page_spec(p_str: str) -> List[int]:
if "-" in p_str:
start, end = map(int, p_str.split("-"))
return list(range(start, end + 1))
else:
return [int(p_str)]
def parse_args(args_raw: List[str]) -> argparse.Namespace:
parser = argparse.ArgumentParser("pdfplumber")
parser.add_argument(
"infile", nargs="?", type=argparse.FileType("rb"), default=sys.stdin.buffer
)
group = parser.add_mutually_exclusive_group()
group.add_argument(
"--structure",
help="Write the structure tree as JSON. "
"All other arguments except --pages, --laparams, and --indent will be ignored",
action="store_true",
)
group.add_argument(
"--structure-text",
help="Write the structure tree as JSON including text contents. "
"All other arguments except --pages, --laparams, and --indent will be ignored",
action="store_true",
)
parser.add_argument("--format", choices=["csv", "json", "text"], default="csv")
parser.add_argument("--types", nargs="+")
parser.add_argument(
"--include-attrs",
nargs="+",
help="Include *only* these object attributes in output.",
)
parser.add_argument(
"--exclude-attrs",
nargs="+",
help="Exclude these object attributes from output.",
)
parser.add_argument("--laparams", type=json.loads)
parser.add_argument("--precision", type=int)
parser.add_argument("--pages", nargs="+", type=parse_page_spec)
parser.add_argument(
"--indent", type=int, help="Indent level for JSON pretty-printing."
)
args = parser.parse_args(args_raw)
if args.pages is not None:
args.pages = list(chain(*args.pages))
return args
def add_text_to_mcids(pdf: PDF, data: List[Dict[str, Any]]) -> None:
page_contents: DefaultDict[int, Any] = defaultdict(lambda: defaultdict(str))
for page in pdf.pages:
text_contents = page_contents[page.page_number]
for c in page.chars:
mcid = c.get("mcid")
if mcid is None:
continue
text_contents[mcid] += c["text"]
d = deque(data)
while d:
el = d.popleft()
if "children" in el:
d.extend(el["children"])
pageno = el.get("page_number")
if pageno is None:
continue
text_contents = page_contents[pageno]
if "mcids" in el:
el["text"] = [text_contents[mcid] for mcid in el["mcids"]]
def main(args_raw: List[str] = sys.argv[1:]) -> None:
args = parse_args(args_raw)
with PDF.open(args.infile, pages=args.pages, laparams=args.laparams) as pdf:
if args.structure:
print(json.dumps(pdf.structure_tree, indent=args.indent))
elif args.structure_text:
tree = pdf.structure_tree
add_text_to_mcids(pdf, tree)
print(json.dumps(tree, indent=args.indent, ensure_ascii=False))
elif args.format == "csv":
pdf.to_csv(
sys.stdout,
args.types,
precision=args.precision,
include_attrs=args.include_attrs,
exclude_attrs=args.exclude_attrs,
)
elif args.format == "text":
for page in pdf.pages:
print(page.extract_text(layout=True))
else:
pdf.to_json(
sys.stdout,
args.types,
precision=args.precision,
include_attrs=args.include_attrs,
exclude_attrs=args.exclude_attrs,
indent=args.indent,
)
if __name__ == "__main__":
main()
@@ -0,0 +1,187 @@
import csv
import json
from io import StringIO
from itertools import chain
from typing import Any, Dict, List, Optional, Set, TextIO
from . import utils
from ._typing import T_obj, T_obj_list
from .convert import CSV_COLS_REQUIRED, CSV_COLS_TO_PREPEND, Serializer
class Container(object):
cached_properties = ["_rect_edges", "_curve_edges", "_edges", "_objects"]
@property
def pages(self) -> Optional[List[Any]]: # pragma: nocover
raise NotImplementedError
@property
def objects(self) -> Dict[str, T_obj_list]: # pragma: nocover
raise NotImplementedError
def to_dict(
self, object_types: Optional[List[str]] = None
) -> Dict[str, Any]: # pragma: nocover
raise NotImplementedError
def flush_cache(self, properties: Optional[List[str]] = None) -> None:
props = self.cached_properties if properties is None else properties
for p in props:
if hasattr(self, p):
delattr(self, p)
@property
def rects(self) -> T_obj_list:
return self.objects.get("rect", [])
@property
def lines(self) -> T_obj_list:
return self.objects.get("line", [])
@property
def curves(self) -> T_obj_list:
return self.objects.get("curve", [])
@property
def images(self) -> T_obj_list:
return self.objects.get("image", [])
@property
def chars(self) -> T_obj_list:
return self.objects.get("char", [])
@property
def textboxverticals(self) -> T_obj_list:
return self.objects.get("textboxvertical", [])
@property
def textboxhorizontals(self) -> T_obj_list:
return self.objects.get("textboxhorizontal", [])
@property
def textlineverticals(self) -> T_obj_list:
return self.objects.get("textlinevertical", [])
@property
def textlinehorizontals(self) -> T_obj_list:
return self.objects.get("textlinehorizontal", [])
@property
def rect_edges(self) -> T_obj_list:
if hasattr(self, "_rect_edges"):
return self._rect_edges
rect_edges_gen = (utils.rect_to_edges(r) for r in self.rects)
self._rect_edges: T_obj_list = list(chain(*rect_edges_gen))
return self._rect_edges
@property
def curve_edges(self) -> T_obj_list:
if hasattr(self, "_curve_edges"):
return self._curve_edges
curve_edges_gen = (utils.curve_to_edges(r) for r in self.curves)
self._curve_edges: T_obj_list = list(chain(*curve_edges_gen))
return self._curve_edges
@property
def edges(self) -> T_obj_list:
if hasattr(self, "_edges"):
return self._edges
line_edges = list(map(utils.line_to_edge, self.lines))
self._edges: T_obj_list = line_edges + self.rect_edges + self.curve_edges
return self._edges
@property
def horizontal_edges(self) -> T_obj_list:
def test(x: T_obj) -> bool:
return bool(x["orientation"] == "h")
return list(filter(test, self.edges))
@property
def vertical_edges(self) -> T_obj_list:
def test(x: T_obj) -> bool:
return bool(x["orientation"] == "v")
return list(filter(test, self.edges))
def to_json(
self,
stream: Optional[TextIO] = None,
object_types: Optional[List[str]] = None,
include_attrs: Optional[List[str]] = None,
exclude_attrs: Optional[List[str]] = None,
precision: Optional[int] = None,
indent: Optional[int] = None,
) -> Optional[str]:
data = self.to_dict(object_types)
serialized = Serializer(
precision=precision,
include_attrs=include_attrs,
exclude_attrs=exclude_attrs,
).serialize(data)
if stream is None:
return json.dumps(serialized, indent=indent)
else:
json.dump(serialized, stream, indent=indent)
return None
def to_csv(
self,
stream: Optional[TextIO] = None,
object_types: Optional[List[str]] = None,
precision: Optional[int] = None,
include_attrs: Optional[List[str]] = None,
exclude_attrs: Optional[List[str]] = None,
) -> Optional[str]:
if stream is None:
stream = StringIO()
to_string = True
else:
to_string = False
if object_types is None:
object_types = list(self.objects.keys()) + ["annot"]
serialized = []
fields: Set[str] = set()
pages = [self] if self.pages is None else self.pages
serializer = Serializer(
precision=precision,
include_attrs=include_attrs,
exclude_attrs=exclude_attrs,
)
for page in pages:
for t in object_types:
objs = getattr(page, t + "s")
if len(objs):
serialized += serializer.serialize(objs)
new_keys = [k for k, v in objs[0].items() if type(v) is not dict]
fields = fields.union(set(new_keys))
non_req_cols = CSV_COLS_TO_PREPEND + list(
sorted(set(fields) - set(CSV_COLS_REQUIRED + CSV_COLS_TO_PREPEND))
)
cols = CSV_COLS_REQUIRED + list(filter(serializer.attr_filter, non_req_cols))
w = csv.DictWriter(
stream,
fieldnames=cols,
extrasaction="ignore",
quoting=csv.QUOTE_MINIMAL,
escapechar="\\",
)
w.writeheader()
w.writerows(serialized)
if to_string:
stream.seek(0)
return stream.read()
else:
return None
@@ -0,0 +1,127 @@
import base64
from typing import Any, Callable, Dict, List, Optional, Tuple
from pdfminer.psparser import PSLiteral
from .utils import decode_text
ENCODINGS_TO_TRY = [
"utf-8",
"latin-1",
"utf-16",
"utf-16le",
]
CSV_COLS_REQUIRED = [
"object_type",
]
CSV_COLS_TO_PREPEND = [
"page_number",
"x0",
"x1",
"y0",
"y1",
"doctop",
"top",
"bottom",
"width",
"height",
]
def get_attr_filter(
include_attrs: Optional[List[str]] = None, exclude_attrs: Optional[List[str]] = None
) -> Callable[[str], bool]:
if include_attrs is not None and exclude_attrs is not None:
raise ValueError(
"Cannot specify `include_attrs` and `exclude_attrs` at the same time."
)
elif include_attrs is not None:
incl = set(CSV_COLS_REQUIRED + include_attrs)
return lambda attr: attr in incl
elif exclude_attrs is not None:
nonexcludable = set(exclude_attrs).intersection(set(CSV_COLS_REQUIRED))
if len(nonexcludable):
raise ValueError(
f"Cannot exclude these required properties: {list(nonexcludable)}"
)
excl = set(exclude_attrs)
return lambda attr: attr not in excl
else:
return lambda attr: True
def to_b64(data: bytes) -> str:
return base64.b64encode(data).decode("ascii")
class Serializer:
def __init__(
self,
precision: Optional[int] = None,
include_attrs: Optional[List[str]] = None,
exclude_attrs: Optional[List[str]] = None,
):
self.precision = precision
self.attr_filter = get_attr_filter(
include_attrs=include_attrs, exclude_attrs=exclude_attrs
)
def serialize(self, obj: Any) -> Any:
if obj is None:
return None
t = type(obj)
# Basic types don't need to be converted
if t in (int, str):
return obj
# Use one of the custom converters, if possible
fn = getattr(self, f"do_{t.__name__}", None)
if fn is not None:
return fn(obj)
# Otherwise, just use the string-representation
else:
return str(obj)
def do_float(self, x: float) -> float:
return x if self.precision is None else round(x, self.precision)
def do_bool(self, x: bool) -> int:
return int(x)
def do_list(self, obj: List[Any]) -> List[Any]:
return list(self.serialize(x) for x in obj)
def do_tuple(self, obj: Tuple[Any, ...]) -> Tuple[Any, ...]:
return tuple(self.serialize(x) for x in obj)
def do_dict(self, obj: Dict[str, Any]) -> Dict[str, Any]:
if "object_type" in obj.keys():
return {k: self.serialize(v) for k, v in obj.items() if self.attr_filter(k)}
else:
return {k: self.serialize(v) for k, v in obj.items()}
def do_PDFStream(self, obj: Any) -> Dict[str, Optional[str]]:
return {"rawdata": to_b64(obj.rawdata) if obj.rawdata else None}
def do_PSLiteral(self, obj: PSLiteral) -> str:
return decode_text(obj.name)
def do_bytes(self, obj: bytes) -> Optional[str]:
for e in ENCODINGS_TO_TRY:
try:
return obj.decode(e)
except UnicodeDecodeError: # pragma: no cover
return None
# If none of the decodings work, raise whatever error
# decoding with utf-8 causes
obj.decode(ENCODINGS_TO_TRY[0]) # pragma: no cover
return None # pragma: no cover
@@ -0,0 +1,38 @@
import math
from typing import NamedTuple
# For more details, see the PDF Reference, 6th Ed., Section 4.2.2 ("Common
# Transformations")
class CTM(NamedTuple):
a: float
b: float
c: float
d: float
e: float
f: float
@property
def scale_x(self) -> float:
return math.sqrt(pow(self.a, 2) + pow(self.b, 2))
@property
def scale_y(self) -> float:
return math.sqrt(pow(self.c, 2) + pow(self.d, 2))
@property
def skew_x(self) -> float:
return (math.atan2(self.d, self.c) * 180 / math.pi) - 90
@property
def skew_y(self) -> float:
return math.atan2(self.b, self.a) * 180 / math.pi
@property
def translation_x(self) -> float:
return self.e
@property
def translation_y(self) -> float:
return self.f
@@ -0,0 +1,403 @@
import pathlib
from io import BufferedReader, BytesIO
from typing import TYPE_CHECKING, Any, List, Optional, Tuple, Union
import PIL.Image
import PIL.ImageDraw
import pypdfium2 # type: ignore
from . import utils
from ._typing import T_bbox, T_num, T_obj, T_obj_list, T_point, T_seq
from .table import T_table_settings, Table, TableFinder, TableSettings
from .utils.exceptions import MalformedPDFException
if TYPE_CHECKING: # pragma: nocover
import pandas as pd
from .page import Page
class COLORS:
RED = (255, 0, 0)
GREEN = (0, 255, 0)
BLUE = (0, 0, 255)
TRANSPARENT = (0, 0, 0, 0)
DEFAULT_FILL = COLORS.BLUE + (50,)
DEFAULT_STROKE = COLORS.RED + (200,)
DEFAULT_STROKE_WIDTH = 1
DEFAULT_RESOLUTION = 72
T_color = Union[Tuple[int, int, int], Tuple[int, int, int, int], str]
T_contains_points = Union[Tuple[T_point, ...], List[T_point], T_obj]
def get_page_image(
stream: Union[BufferedReader, BytesIO],
path: Optional[pathlib.Path],
page_ix: int,
resolution: Union[int, float],
password: Optional[str],
antialias: bool = False,
) -> PIL.Image.Image:
src: Union[pathlib.Path, BufferedReader, BytesIO]
# If we are working with a file object saved to disk
if path:
src = path
# If we instead are working with a BytesIO stream
else:
stream.seek(0)
src = stream
try:
pdfium_doc = pypdfium2.PdfDocument(src, password=password)
except pypdfium2.PdfiumError as e:
raise MalformedPDFException(e)
pdfium_page = pdfium_doc.get_page(page_ix)
img: PIL.Image.Image = pdfium_page.render(
# Modifiable arguments
scale=resolution / 72,
no_smoothtext=not antialias,
no_smoothpath=not antialias,
no_smoothimage=not antialias,
# Non-modifiable arguments
prefer_bgrx=True,
).to_pil()
pdfium_doc.close()
return img.convert("RGB")
class PageImage:
def __init__(
self,
page: "Page",
original: Optional[PIL.Image.Image] = None,
resolution: Union[int, float] = DEFAULT_RESOLUTION,
antialias: bool = False,
force_mediabox: bool = False,
):
self.page = page
self.root = page if page.is_original else page.root_page
self.resolution = resolution
if original is None:
self.original = get_page_image(
stream=page.pdf.stream,
path=page.pdf.path,
page_ix=page.page_number - 1,
resolution=resolution,
antialias=antialias,
password=page.pdf.password,
)
else:
self.original = original
self.scale = self.original.size[0] / (page.cropbox[2] - page.cropbox[0])
# This value represents the coordinates of the page,
# in page-unit values, that will be displayed.
self.bbox = (
page.bbox
if page.bbox != page.mediabox
else (page.mediabox if force_mediabox else page.cropbox)
)
# If this value is different than the *Page*'s .cropbox
# (e.g., because the mediabox differs from the cropbox or
# or because we've used Page.crop(...)), then we'll need to
# crop the initially-converted image.
if page.bbox != page.cropbox:
crop_dims = self._reproject_bbox(page.cropbox)
bbox_dims = self._reproject_bbox(self.bbox)
self.original = self.original.crop(
(
bbox_dims[0] - crop_dims[0],
bbox_dims[1] - crop_dims[1],
bbox_dims[2] - crop_dims[0],
bbox_dims[3] - crop_dims[1],
)
)
self.reset()
def _reproject_bbox(self, bbox: T_bbox) -> Tuple[int, int, int, int]:
x0, top, x1, bottom = bbox
_x0, _top = self._reproject((x0, top))
_x1, _bottom = self._reproject((x1, bottom))
return (_x0, _top, _x1, _bottom)
def _reproject(self, coord: T_point) -> Tuple[int, int]:
"""
Given an (x0, top) tuple from the *root* coordinate system,
return an (x0, top) tuple in the *image* coordinate system.
"""
x0, top = coord
_x0 = (x0 - self.bbox[0]) * self.scale
_top = (top - self.bbox[1]) * self.scale
return (int(_x0), int(_top))
def reset(self) -> "PageImage":
self.annotated = PIL.Image.new("RGB", self.original.size)
self.annotated.paste(self.original)
self.draw = PIL.ImageDraw.Draw(self.annotated, "RGBA")
return self
def save(
self,
dest: Union[str, pathlib.Path, BytesIO],
format: str = "PNG",
quantize: bool = True,
colors: int = 256,
bits: int = 8,
**kwargs: Any,
) -> None:
if quantize:
out = self.annotated.quantize(colors, method=PIL.Image.FASTOCTREE).convert(
"P"
)
else:
out = self.annotated
out.save(
dest,
format=format,
bits=bits,
dpi=(self.resolution, self.resolution),
**kwargs,
)
def copy(self) -> "PageImage":
return self.__class__(self.page, self.original)
def draw_line(
self,
points_or_obj: T_contains_points,
stroke: T_color = DEFAULT_STROKE,
stroke_width: int = DEFAULT_STROKE_WIDTH,
) -> "PageImage":
# If passing a raw list of points, use those
if isinstance(points_or_obj, (tuple, list)):
points = points_or_obj
# Else, use the "pts" attribute if available
elif isinstance(points_or_obj, dict) and "pts" in points_or_obj:
points = [(x, y) for x, y in points_or_obj["pts"]]
# Otherwise, just use ((x0, top), (x1, bottom))
else:
obj = points_or_obj
points = ((obj["x0"], obj["top"]), (obj["x1"], obj["bottom"]))
self.draw.line(
list(map(self._reproject, points)), fill=stroke, width=stroke_width
)
return self
def draw_lines(
self,
list_of_lines: Union[T_seq[T_contains_points], "pd.DataFrame"],
stroke: T_color = DEFAULT_STROKE,
stroke_width: int = DEFAULT_STROKE_WIDTH,
) -> "PageImage":
for x in utils.to_list(list_of_lines):
self.draw_line(x, stroke=stroke, stroke_width=stroke_width)
return self
def draw_vline(
self,
location: T_num,
stroke: T_color = DEFAULT_STROKE,
stroke_width: int = DEFAULT_STROKE_WIDTH,
) -> "PageImage":
points = (location, self.bbox[1], location, self.bbox[3])
self.draw.line(self._reproject_bbox(points), fill=stroke, width=stroke_width)
return self
def draw_vlines(
self,
locations: Union[List[T_num], "pd.Series[float]"],
stroke: T_color = DEFAULT_STROKE,
stroke_width: int = DEFAULT_STROKE_WIDTH,
) -> "PageImage":
for x in list(locations):
self.draw_vline(x, stroke=stroke, stroke_width=stroke_width)
return self
def draw_hline(
self,
location: T_num,
stroke: T_color = DEFAULT_STROKE,
stroke_width: int = DEFAULT_STROKE_WIDTH,
) -> "PageImage":
points = (self.bbox[0], location, self.bbox[2], location)
self.draw.line(self._reproject_bbox(points), fill=stroke, width=stroke_width)
return self
def draw_hlines(
self,
locations: Union[List[T_num], "pd.Series[float]"],
stroke: T_color = DEFAULT_STROKE,
stroke_width: int = DEFAULT_STROKE_WIDTH,
) -> "PageImage":
for x in list(locations):
self.draw_hline(x, stroke=stroke, stroke_width=stroke_width)
return self
def draw_rect(
self,
bbox_or_obj: Union[T_bbox, T_obj],
fill: T_color = DEFAULT_FILL,
stroke: T_color = DEFAULT_STROKE,
stroke_width: int = DEFAULT_STROKE_WIDTH,
) -> "PageImage":
if isinstance(bbox_or_obj, (tuple, list)):
bbox = bbox_or_obj
else:
obj = bbox_or_obj
bbox = (obj["x0"], obj["top"], obj["x1"], obj["bottom"])
x0, top, x1, bottom = bbox
half = stroke_width / 2
x0 = min(x0 + half, (x0 + x1) / 2)
top = min(top + half, (top + bottom) / 2)
x1 = max(x1 - half, (x0 + x1) / 2)
bottom = max(bottom - half, (top + bottom) / 2)
fill_bbox = self._reproject_bbox((x0, top, x1, bottom))
self.draw.rectangle(fill_bbox, fill, COLORS.TRANSPARENT)
if stroke_width > 0:
segments = [
((x0, top), (x1, top)), # top
((x0, bottom), (x1, bottom)), # bottom
((x0, top), (x0, bottom)), # left
((x1, top), (x1, bottom)), # right
]
self.draw_lines(segments, stroke=stroke, stroke_width=stroke_width)
return self
def draw_rects(
self,
list_of_rects: Union[List[T_bbox], T_obj_list, "pd.DataFrame"],
fill: T_color = DEFAULT_FILL,
stroke: T_color = DEFAULT_STROKE,
stroke_width: int = DEFAULT_STROKE_WIDTH,
) -> "PageImage":
for x in utils.to_list(list_of_rects):
self.draw_rect(x, fill=fill, stroke=stroke, stroke_width=stroke_width)
return self
def draw_circle(
self,
center_or_obj: Union[T_point, T_obj],
radius: int = 5,
fill: T_color = DEFAULT_FILL,
stroke: T_color = DEFAULT_STROKE,
) -> "PageImage":
if isinstance(center_or_obj, tuple):
center = center_or_obj
else:
obj = center_or_obj
center = ((obj["x0"] + obj["x1"]) / 2, (obj["top"] + obj["bottom"]) / 2)
cx, cy = center
bbox = (cx - radius, cy - radius, cx + radius, cy + radius)
self.draw.ellipse(self._reproject_bbox(bbox), fill, stroke)
return self
def draw_circles(
self,
list_of_circles: Union[List[T_point], T_obj_list, "pd.DataFrame"],
radius: int = 5,
fill: T_color = DEFAULT_FILL,
stroke: T_color = DEFAULT_STROKE,
) -> "PageImage":
for x in utils.to_list(list_of_circles):
self.draw_circle(x, radius=radius, fill=fill, stroke=stroke)
return self
def debug_table(
self,
table: Table,
fill: T_color = DEFAULT_FILL,
stroke: T_color = DEFAULT_STROKE,
stroke_width: int = 1,
) -> "PageImage":
"""
Outline all found tables.
"""
self.draw_rects(
table.cells, fill=fill, stroke=stroke, stroke_width=stroke_width
)
return self
def debug_tablefinder(
self,
table_settings: Optional[
Union[TableFinder, TableSettings, T_table_settings]
] = None,
) -> "PageImage":
if isinstance(table_settings, TableFinder):
finder = table_settings
elif table_settings is None or isinstance(
table_settings, (TableSettings, dict)
):
finder = self.page.debug_tablefinder(table_settings)
else:
raise ValueError(
"Argument must be instance of TableFinder"
"or a TableFinder settings dict."
)
for table in finder.tables:
self.debug_table(table)
self.draw_lines(finder.edges, stroke_width=1)
self.draw_circles(
list(finder.intersections.keys()),
fill=COLORS.TRANSPARENT,
stroke=COLORS.BLUE + (200,),
radius=3,
)
return self
def outline_words(
self,
stroke: T_color = DEFAULT_STROKE,
fill: T_color = DEFAULT_FILL,
stroke_width: int = DEFAULT_STROKE_WIDTH,
x_tolerance: T_num = utils.DEFAULT_X_TOLERANCE,
y_tolerance: T_num = utils.DEFAULT_Y_TOLERANCE,
) -> "PageImage":
words = self.page.extract_words(
x_tolerance=x_tolerance, y_tolerance=y_tolerance
)
self.draw_rects(words, stroke=stroke, fill=fill, stroke_width=stroke_width)
return self
def outline_chars(
self,
stroke: T_color = (255, 0, 0, 255),
fill: T_color = (255, 0, 0, int(255 / 4)),
stroke_width: int = DEFAULT_STROKE_WIDTH,
) -> "PageImage":
self.draw_rects(
self.page.chars, stroke=stroke, fill=fill, stroke_width=stroke_width
)
return self
def _repr_png_(self) -> bytes:
b = BytesIO()
self.save(b, "PNG")
return b.getvalue()
def show(self) -> None: # pragma: no cover
self.annotated.show()
@@ -0,0 +1,731 @@
import numbers
import re
from functools import lru_cache
from typing import (
TYPE_CHECKING,
Any,
Callable,
Dict,
Generator,
List,
Optional,
Pattern,
Tuple,
Union,
)
from unicodedata import normalize as normalize_unicode
from warnings import warn
from pdfminer.converter import PDFPageAggregator
from pdfminer.layout import (
LTChar,
LTComponent,
LTContainer,
LTCurve,
LTItem,
LTPage,
LTTextContainer,
)
from pdfminer.pdfinterp import PDFPageInterpreter, PDFStackT
from pdfminer.pdfpage import PDFPage
from pdfminer.psparser import PSLiteral
from . import utils
from ._typing import T_bbox, T_num, T_obj, T_obj_list
from .container import Container
from .structure import PDFStructTree, StructTreeMissing
from .table import T_table_settings, Table, TableFinder, TableSettings
from .utils import decode_text, resolve_all, resolve_and_decode
from .utils.exceptions import MalformedPDFException, PdfminerException
from .utils.text import TextMap
lt_pat = re.compile(r"^LT")
ALL_ATTRS = set(
[
"adv",
"height",
"linewidth",
"pts",
"size",
"srcsize",
"width",
"x0",
"x1",
"y0",
"y1",
"bits",
"matrix",
"upright",
"fontname",
"text",
"imagemask",
"colorspace",
"evenodd",
"fill",
"non_stroking_color",
"stroke",
"stroking_color",
"stream",
"name",
"mcid",
"tag",
]
)
if TYPE_CHECKING: # pragma: nocover
from .display import PageImage
from .pdf import PDF
# via https://git.ghostscript.com/?p=mupdf.git;a=blob;f=source/pdf/pdf-font.c;h=6322cedf2c26cfb312c0c0878d7aff97b4c7470e;hb=HEAD#l774 # noqa
CP936_FONTNAMES = {
b"\xcb\xce\xcc\xe5": "SimSun,Regular",
b"\xba\xda\xcc\xe5": "SimHei,Regular",
b"\xbf\xac\xcc\xe5_GB2312": "SimKai,Regular",
b"\xb7\xc2\xcb\xce_GB2312": "SimFang,Regular",
b"\xc1\xa5\xca\xe9": "SimLi,Regular",
}
def fix_fontname_bytes(fontname: bytes) -> str:
if b"+" in fontname:
split_at = fontname.index(b"+") + 1
prefix, suffix = fontname[:split_at], fontname[split_at:]
else:
prefix, suffix = b"", fontname
suffix_new = CP936_FONTNAMES.get(suffix, str(suffix)[2:-1])
return str(prefix)[2:-1] + suffix_new
def tuplify_list_kwargs(kwargs: Dict[str, Any]) -> Dict[str, Any]:
return {
key: (tuple(value) if isinstance(value, list) else value)
for key, value in kwargs.items()
}
class PDFPageAggregatorWithMarkedContent(PDFPageAggregator):
"""Extract layout from a specific page, adding marked-content IDs to
objects where found."""
cur_mcid: Optional[int] = None
cur_tag: Optional[str] = None
def begin_tag(self, tag: PSLiteral, props: Optional[PDFStackT] = None) -> None:
"""Handle beginning of tag, setting current MCID if any."""
self.cur_tag = decode_text(tag.name)
if isinstance(props, dict) and "MCID" in props:
self.cur_mcid = props["MCID"]
else:
self.cur_mcid = None
def end_tag(self) -> None:
"""Handle beginning of tag, clearing current MCID."""
self.cur_tag = None
self.cur_mcid = None
def tag_cur_item(self) -> None:
"""Add current MCID to what we hope to be the most recent object created
by pdfminer.six."""
# This is somewhat hacky and would not be necessary if
# pdfminer.six supported MCIDs. In reading the code it's
# clear that the `render_*` methods methods will only ever
# create one object, but that is far from being guaranteed.
# Even if pdfminer.six's API would just return the objects it
# creates, we wouldn't have to do this.
if self.cur_item._objs:
cur_obj = self.cur_item._objs[-1]
cur_obj.mcid = self.cur_mcid # type: ignore
cur_obj.tag = self.cur_tag # type: ignore
def render_char(self, *args, **kwargs) -> float: # type: ignore
"""Hook for rendering characters, adding the `mcid` attribute."""
adv = super().render_char(*args, **kwargs)
self.tag_cur_item()
return adv
def render_image(self, *args, **kwargs) -> None: # type: ignore
"""Hook for rendering images, adding the `mcid` attribute."""
super().render_image(*args, **kwargs)
self.tag_cur_item()
def paint_path(self, *args, **kwargs) -> None: # type: ignore
"""Hook for rendering lines and curves, adding the `mcid` attribute."""
super().paint_path(*args, **kwargs)
self.tag_cur_item()
def _normalize_box(box_raw: T_bbox, rotation: T_num = 0) -> T_bbox:
# Per PDF Reference 3.8.4: "Note: Although rectangles are
# conventionally specified by their lower-left and upperright
# corners, it is acceptable to specify any two diagonally opposite
# corners."
if not all(isinstance(x, numbers.Number) for x in box_raw): # pragma: nocover
raise MalformedPDFException(
f"Bounding box contains non-number coordinate(s): {box_raw}"
)
x0, x1 = sorted((box_raw[0], box_raw[2]))
y0, y1 = sorted((box_raw[1], box_raw[3]))
if rotation in [90, 270]:
return (y0, x0, y1, x1)
else:
return (x0, y0, x1, y1)
# PDFs coordinate spaces refer to an origin in the bottom-left of the
# page; pdfplumber flips this vertically, so that the origin is in the
# top-left.
def _invert_box(box_raw: T_bbox, mb_height: T_num) -> T_bbox:
x0, y0, x1, y1 = box_raw
return (x0, mb_height - y1, x1, mb_height - y0)
class Page(Container):
cached_properties: List[str] = Container.cached_properties + ["_layout"]
is_original: bool = True
pages = None
def __init__(
self,
pdf: "PDF",
page_obj: PDFPage,
page_number: int,
initial_doctop: T_num = 0,
):
self.pdf = pdf
self.root_page = self
self.page_obj = page_obj
self.page_number = page_number
self.initial_doctop = initial_doctop
def get_attr(key: str, default: Any = None) -> Any:
value = resolve_all(page_obj.attrs.get(key))
return default if value is None else value
# Per PDF Reference Table 3.27: "The number of degrees by which the
# page should be rotated clockwise when displayed or printed. The value
# must be a multiple of 90. Default value: 0"
_rotation = get_attr("Rotate", 0)
self.rotation = _rotation % 360
mb_raw = _normalize_box(get_attr("MediaBox"), self.rotation)
mb_height = mb_raw[3] - mb_raw[1]
self.mediabox = _invert_box(mb_raw, mb_height)
for box_name in ["CropBox", "TrimBox", "BleedBox", "ArtBox"]:
if box_name in page_obj.attrs:
box_normalized = _invert_box(
_normalize_box(get_attr(box_name), self.rotation), mb_height
)
setattr(self, box_name.lower(), box_normalized)
if "CropBox" not in page_obj.attrs:
self.cropbox = self.mediabox
# Page.bbox defaults to self.mediabox, but can be altered by Page.crop(...)
self.bbox = self.mediabox
# See https://rednafi.com/python/lru_cache_on_methods/
self.get_textmap = lru_cache()(self._get_textmap)
def close(self) -> None:
self.flush_cache()
self.get_textmap.cache_clear()
@property
def width(self) -> T_num:
return self.bbox[2] - self.bbox[0]
@property
def height(self) -> T_num:
return self.bbox[3] - self.bbox[1]
@property
def structure_tree(self) -> List[Dict[str, Any]]:
"""Return the structure tree for a page, if any."""
try:
return [elem.to_dict() for elem in PDFStructTree(self.pdf, self)]
except StructTreeMissing:
return []
@property
def layout(self) -> LTPage:
if hasattr(self, "_layout"):
return self._layout
device = PDFPageAggregatorWithMarkedContent(
self.pdf.rsrcmgr,
pageno=self.page_number,
laparams=self.pdf.laparams,
)
interpreter = PDFPageInterpreter(self.pdf.rsrcmgr, device)
try:
interpreter.process_page(self.page_obj)
except Exception as e:
raise PdfminerException(e)
self._layout: LTPage = device.get_result()
return self._layout
@property
def annots(self) -> T_obj_list:
def rotate_point(pt: Tuple[float, float], r: int) -> Tuple[float, float]:
turns = r // 90
for i in range(turns):
x, y = pt
comp = self.width if i == turns % 2 else self.height
pt = (y, (comp - x))
return pt
def parse(annot: T_obj) -> T_obj:
_a, _b, _c, _d = annot["Rect"]
pt0 = rotate_point((_a, _b), self.rotation)
pt1 = rotate_point((_c, _d), self.rotation)
rh = self.root_page.height
x0, top, x1, bottom = _invert_box(_normalize_box((*pt0, *pt1)), rh)
a = annot.get("A", {})
extras = {
"uri": a.get("URI"),
"title": annot.get("T"),
"contents": annot.get("Contents"),
}
for k, v in extras.items():
if v is not None:
try:
extras[k] = v.decode("utf-8")
except UnicodeDecodeError:
try:
extras[k] = v.decode("utf-16")
except UnicodeDecodeError:
if self.pdf.raise_unicode_errors:
raise
warn(
f"Could not decode {k} of annotation."
f" {k} will be missing."
)
parsed = {
"page_number": self.page_number,
"object_type": "annot",
"x0": x0,
"y0": rh - bottom,
"x1": x1,
"y1": rh - top,
"doctop": self.initial_doctop + top,
"top": top,
"bottom": bottom,
"width": x1 - x0,
"height": bottom - top,
}
parsed.update(extras)
# Replace the indirect reference to the page dictionary
# with a pointer to our actual page
if "P" in annot:
annot["P"] = self
parsed["data"] = annot
return parsed
raw = resolve_all(self.page_obj.annots) or []
parsed = list(map(parse, raw))
if isinstance(self, CroppedPage):
return self._crop_fn(parsed)
else:
return parsed
@property
def hyperlinks(self) -> T_obj_list:
return [a for a in self.annots if a["uri"] is not None]
@property
def objects(self) -> Dict[str, T_obj_list]:
if hasattr(self, "_objects"):
return self._objects
self._objects: Dict[str, T_obj_list] = self.parse_objects()
return self._objects
def point2coord(self, pt: Tuple[T_num, T_num]) -> Tuple[T_num, T_num]:
# See note below re. #1181 and mediabox-adjustment reversions
return (self.mediabox[0] + pt[0], self.mediabox[1] + self.height - pt[1])
def process_object(self, obj: LTItem) -> T_obj:
kind = re.sub(lt_pat, "", obj.__class__.__name__).lower()
def process_attr(item: Tuple[str, Any]) -> Optional[Tuple[str, Any]]:
k, v = item
if k in ALL_ATTRS:
res = resolve_all(v)
return (k, res)
else:
return None
attr = dict(filter(None, map(process_attr, obj.__dict__.items())))
attr["object_type"] = kind
attr["page_number"] = self.page_number
for cs in ["ncs", "scs"]:
# Note: As of pdfminer.six v20221105, that library only
# exposes ncs for LTChars, and neither attribute for
# other objects. Keeping this code here, though,
# for ease of addition if color spaces become
# more available via pdfminer.six
if hasattr(obj, cs):
attr[cs] = resolve_and_decode(getattr(obj, cs).name)
if isinstance(obj, (LTChar, LTTextContainer)):
text = obj.get_text()
attr["text"] = (
normalize_unicode(self.pdf.unicode_norm, text)
if self.pdf.unicode_norm is not None
else text
)
if isinstance(obj, LTChar):
# pdfminer.six (at least as of v20221105) does not
# directly expose .stroking_color and .non_stroking_color
# for LTChar objects (unlike, e.g., LTRect objects).
gs = obj.graphicstate
attr["stroking_color"] = (
gs.scolor if isinstance(gs.scolor, tuple) else (gs.scolor,)
)
attr["non_stroking_color"] = (
gs.ncolor if isinstance(gs.ncolor, tuple) else (gs.ncolor,)
)
# Handle (rare) byte-encoded fontnames
if isinstance(attr["fontname"], bytes): # pragma: nocover
attr["fontname"] = fix_fontname_bytes(attr["fontname"])
elif isinstance(obj, (LTCurve,)):
attr["pts"] = list(map(self.point2coord, attr["pts"]))
# Ignoring typing because type signature for obj.original_path
# appears to be incorrect
attr["path"] = [(cmd, *map(self.point2coord, pts)) for cmd, *pts in obj.original_path] # type: ignore # noqa: E501
attr["dash"] = obj.dashing_style
# As noted in #1181, `pdfminer.six` adjusts objects'
# coordinates relative to the MediaBox:
# https://github.com/pdfminer/pdfminer.six/blob/1a8bd2f730295b31d6165e4d95fcb5a03793c978/pdfminer/converter.py#L79-L84
mb_x0, mb_top = self.mediabox[:2]
if "y0" in attr:
attr["top"] = (self.height - attr["y1"]) + mb_top
attr["bottom"] = (self.height - attr["y0"]) + mb_top
attr["doctop"] = self.initial_doctop + attr["top"]
if "x0" in attr and mb_x0 != 0:
attr["x0"] = attr["x0"] + mb_x0
attr["x1"] = attr["x1"] + mb_x0
return attr
def iter_layout_objects(
self, layout_objects: List[LTComponent]
) -> Generator[T_obj, None, None]:
for obj in layout_objects:
# If object is, like LTFigure, a higher-level object ...
if isinstance(obj, LTContainer):
# and LAParams is passed, process the object itself.
if self.pdf.laparams is not None:
yield self.process_object(obj)
# Regardless, iterate through its children
yield from self.iter_layout_objects(obj._objs)
else:
yield self.process_object(obj)
def parse_objects(self) -> Dict[str, T_obj_list]:
objects: Dict[str, T_obj_list] = {}
for obj in self.iter_layout_objects(self.layout._objs):
kind = obj["object_type"]
if kind in ["anno"]:
continue
if objects.get(kind) is None:
objects[kind] = []
objects[kind].append(obj)
return objects
def debug_tablefinder(
self, table_settings: Optional[T_table_settings] = None
) -> TableFinder:
tset = TableSettings.resolve(table_settings)
return TableFinder(self, tset)
def find_tables(
self, table_settings: Optional[T_table_settings] = None
) -> List[Table]:
tset = TableSettings.resolve(table_settings)
return TableFinder(self, tset).tables
def find_table(
self, table_settings: Optional[T_table_settings] = None
) -> Optional[Table]:
tset = TableSettings.resolve(table_settings)
tables = self.find_tables(tset)
if len(tables) == 0:
return None
# Return the largest table, as measured by number of cells.
def sorter(x: Table) -> Tuple[int, T_num, T_num]:
return (-len(x.cells), x.bbox[1], x.bbox[0])
largest = list(sorted(tables, key=sorter))[0]
return largest
def extract_tables(
self, table_settings: Optional[T_table_settings] = None
) -> List[List[List[Optional[str]]]]:
tset = TableSettings.resolve(table_settings)
tables = self.find_tables(tset)
return [table.extract(**(tset.text_settings or {})) for table in tables]
def extract_table(
self, table_settings: Optional[T_table_settings] = None
) -> Optional[List[List[Optional[str]]]]:
tset = TableSettings.resolve(table_settings)
table = self.find_table(tset)
if table is None:
return None
else:
return table.extract(**(tset.text_settings or {}))
def _get_textmap(self, **kwargs: Any) -> TextMap:
defaults: Dict[str, Any] = dict(
layout_bbox=self.bbox,
)
if "layout_width_chars" not in kwargs:
defaults.update({"layout_width": self.width})
if "layout_height_chars" not in kwargs:
defaults.update({"layout_height": self.height})
full_kwargs: Dict[str, Any] = {**defaults, **kwargs}
return utils.chars_to_textmap(self.chars, **full_kwargs)
def search(
self,
pattern: Union[str, Pattern[str]],
regex: bool = True,
case: bool = True,
main_group: int = 0,
return_chars: bool = True,
return_groups: bool = True,
**kwargs: Any,
) -> List[Dict[str, Any]]:
textmap = self.get_textmap(**tuplify_list_kwargs(kwargs))
return textmap.search(
pattern,
regex=regex,
case=case,
main_group=main_group,
return_chars=return_chars,
return_groups=return_groups,
)
def extract_text(self, **kwargs: Any) -> str:
return self.get_textmap(**tuplify_list_kwargs(kwargs)).as_string
def extract_text_simple(self, **kwargs: Any) -> str:
return utils.extract_text_simple(self.chars, **kwargs)
def extract_words(self, **kwargs: Any) -> T_obj_list:
return utils.extract_words(self.chars, **kwargs)
def extract_text_lines(
self, strip: bool = True, return_chars: bool = True, **kwargs: Any
) -> T_obj_list:
return self.get_textmap(**tuplify_list_kwargs(kwargs)).extract_text_lines(
strip=strip, return_chars=return_chars
)
def crop(
self, bbox: T_bbox, relative: bool = False, strict: bool = True
) -> "CroppedPage":
return CroppedPage(self, bbox, relative=relative, strict=strict)
def within_bbox(
self, bbox: T_bbox, relative: bool = False, strict: bool = True
) -> "CroppedPage":
"""
Same as .crop, except only includes objects fully within the bbox
"""
return CroppedPage(
self, bbox, relative=relative, strict=strict, crop_fn=utils.within_bbox
)
def outside_bbox(
self, bbox: T_bbox, relative: bool = False, strict: bool = True
) -> "CroppedPage":
"""
Same as .crop, except only includes objects fully within the bbox
"""
return CroppedPage(
self, bbox, relative=relative, strict=strict, crop_fn=utils.outside_bbox
)
def filter(self, test_function: Callable[[T_obj], bool]) -> "FilteredPage":
return FilteredPage(self, test_function)
def dedupe_chars(self, **kwargs: Any) -> "FilteredPage":
"""
Removes duplicate chars — those sharing the same text and positioning
(within `tolerance`) as other characters in the set. Adjust extra_args
to be more/less restrictive with the properties checked.
"""
p = FilteredPage(self, lambda x: True)
p._objects = {kind: objs for kind, objs in self.objects.items()}
p._objects["char"] = utils.dedupe_chars(self.chars, **kwargs)
return p
def to_image(
self,
resolution: Optional[Union[int, float]] = None,
width: Optional[Union[int, float]] = None,
height: Optional[Union[int, float]] = None,
antialias: bool = False,
force_mediabox: bool = False,
) -> "PageImage":
"""
You can pass a maximum of 1 of the following:
- resolution: The desired number pixels per inch. Defaults to 72.
- width: The desired image width in pixels.
- height: The desired image width in pixels.
"""
from .display import DEFAULT_RESOLUTION, PageImage
num_specs = sum(x is not None for x in [resolution, width, height])
if num_specs > 1:
raise ValueError(
f"Only one of these arguments can be provided: resolution, width, height. You provided {num_specs}" # noqa: E501
)
elif width is not None:
resolution = 72 * width / self.width
elif height is not None:
resolution = 72 * height / self.height
return PageImage(
self,
resolution=resolution or DEFAULT_RESOLUTION,
antialias=antialias,
force_mediabox=force_mediabox,
)
def to_dict(self, object_types: Optional[List[str]] = None) -> Dict[str, Any]:
if object_types is None:
_object_types = list(self.objects.keys()) + ["annot"]
else:
_object_types = object_types
d = {
"page_number": self.page_number,
"initial_doctop": self.initial_doctop,
"rotation": self.rotation,
"cropbox": self.cropbox,
"mediabox": self.mediabox,
"bbox": self.bbox,
"width": self.width,
"height": self.height,
}
for t in _object_types:
d[t + "s"] = getattr(self, t + "s")
return d
def __repr__(self) -> str:
return f"<Page:{self.page_number}>"
class DerivedPage(Page):
is_original: bool = False
def __init__(self, parent_page: Page):
self.parent_page = parent_page
self.root_page = parent_page.root_page
self.pdf = parent_page.pdf
self.page_obj = parent_page.page_obj
self.page_number = parent_page.page_number
self.initial_doctop = parent_page.initial_doctop
self.rotation = parent_page.rotation
self.mediabox = parent_page.mediabox
self.cropbox = parent_page.cropbox
self.flush_cache(Container.cached_properties)
self.get_textmap = lru_cache()(self._get_textmap)
def test_proposed_bbox(bbox: T_bbox, parent_bbox: T_bbox) -> None:
bbox_area = utils.calculate_area(bbox)
if bbox_area == 0:
raise ValueError(f"Bounding box {bbox} has an area of zero.")
overlap = utils.get_bbox_overlap(bbox, parent_bbox)
if overlap is None:
raise ValueError(
f"Bounding box {bbox} is entirely outside "
f"parent page bounding box {parent_bbox}"
)
overlap_area = utils.calculate_area(overlap)
if overlap_area < bbox_area:
raise ValueError(
f"Bounding box {bbox} is not fully within "
f"parent page bounding box {parent_bbox}"
)
class CroppedPage(DerivedPage):
def __init__(
self,
parent_page: Page,
crop_bbox: T_bbox,
crop_fn: Callable[[T_obj_list, T_bbox], T_obj_list] = utils.crop_to_bbox,
relative: bool = False,
strict: bool = True,
):
if relative:
o_x0, o_top, _, _ = parent_page.bbox
x0, top, x1, bottom = crop_bbox
crop_bbox = (x0 + o_x0, top + o_top, x1 + o_x0, bottom + o_top)
if strict:
test_proposed_bbox(crop_bbox, parent_page.bbox)
def _crop_fn(objs: T_obj_list) -> T_obj_list:
return crop_fn(objs, crop_bbox)
super().__init__(parent_page)
self._crop_fn = _crop_fn
# Note: testing for original function passed, not _crop_fn
if crop_fn is utils.outside_bbox:
self.bbox = parent_page.bbox
else:
self.bbox = crop_bbox
@property
def objects(self) -> Dict[str, T_obj_list]:
if hasattr(self, "_objects"):
return self._objects
self._objects: Dict[str, T_obj_list] = {
k: self._crop_fn(v) for k, v in self.parent_page.objects.items()
}
return self._objects
class FilteredPage(DerivedPage):
def __init__(self, parent_page: Page, filter_fn: Callable[[T_obj], bool]):
self.bbox = parent_page.bbox
self.filter_fn = filter_fn
super().__init__(parent_page)
@property
def objects(self) -> Dict[str, T_obj_list]:
if hasattr(self, "_objects"):
return self._objects
self._objects: Dict[str, T_obj_list] = {
k: list(filter(self.filter_fn, v))
for k, v in self.parent_page.objects.items()
}
return self._objects
@@ -0,0 +1,205 @@
import itertools
import logging
import pathlib
from io import BufferedReader, BytesIO
from types import TracebackType
from typing import Any, Dict, Generator, List, Literal, Optional, Tuple, Type, Union
from pdfminer.layout import LAParams
from pdfminer.pdfdocument import PDFDocument
from pdfminer.pdfinterp import PDFResourceManager
from pdfminer.pdfpage import PDFPage
from pdfminer.pdfparser import PDFParser
from ._typing import T_num, T_obj_list
from .container import Container
from .page import Page
from .repair import T_repair_setting, _repair
from .structure import PDFStructTree, StructTreeMissing
from .utils import resolve_and_decode
from .utils.exceptions import PdfminerException
logger = logging.getLogger(__name__)
class PDF(Container):
cached_properties: List[str] = Container.cached_properties + ["_pages"]
def __init__(
self,
stream: Union[BufferedReader, BytesIO],
stream_is_external: bool = False,
path: Optional[pathlib.Path] = None,
pages: Optional[Union[List[int], Tuple[int]]] = None,
laparams: Optional[Dict[str, Any]] = None,
password: Optional[str] = None,
strict_metadata: bool = False,
unicode_norm: Optional[Literal["NFC", "NFKC", "NFD", "NFKD"]] = None,
raise_unicode_errors: bool = True,
):
self.stream = stream
self.stream_is_external = stream_is_external
self.path = path
self.pages_to_parse = pages
self.laparams = None if laparams is None else LAParams(**laparams)
self.password = password
self.unicode_norm = unicode_norm
self.raise_unicode_errors = raise_unicode_errors
try:
self.doc = PDFDocument(PDFParser(stream), password=password or "")
except Exception as e:
raise PdfminerException(e)
self.rsrcmgr = PDFResourceManager()
self.metadata = {}
for info in self.doc.info:
self.metadata.update(info)
for k, v in self.metadata.items():
try:
self.metadata[k] = resolve_and_decode(v)
except Exception as e: # pragma: nocover
if strict_metadata:
# Raise an exception since unable to resolve the metadata value.
raise
# This metadata value could not be parsed. Instead of failing the PDF
# read, treat it as a warning only if `strict_metadata=False`.
logger.warning(
f'[WARNING] Metadata key "{k}" could not be parsed due to '
f"exception: {str(e)}"
)
@classmethod
def open(
cls,
path_or_fp: Union[str, pathlib.Path, BufferedReader, BytesIO],
pages: Optional[Union[List[int], Tuple[int]]] = None,
laparams: Optional[Dict[str, Any]] = None,
password: Optional[str] = None,
strict_metadata: bool = False,
unicode_norm: Optional[Literal["NFC", "NFKC", "NFD", "NFKD"]] = None,
repair: bool = False,
gs_path: Optional[Union[str, pathlib.Path]] = None,
repair_setting: T_repair_setting = "default",
raise_unicode_errors: bool = True,
) -> "PDF":
stream: Union[BufferedReader, BytesIO]
if repair:
stream = _repair(
path_or_fp, password=password, gs_path=gs_path, setting=repair_setting
)
stream_is_external = False
# Although the original file has a path,
# the repaired version does not
path = None
elif isinstance(path_or_fp, (str, pathlib.Path)):
stream = open(path_or_fp, "rb")
stream_is_external = False
path = pathlib.Path(path_or_fp)
else:
stream = path_or_fp
stream_is_external = True
path = None
try:
return cls(
stream,
path=path,
pages=pages,
laparams=laparams,
password=password,
strict_metadata=strict_metadata,
unicode_norm=unicode_norm,
stream_is_external=stream_is_external,
raise_unicode_errors=raise_unicode_errors,
)
except PdfminerException:
if not stream_is_external:
stream.close()
raise
def close(self) -> None:
self.flush_cache()
for page in self.pages:
page.close()
if not self.stream_is_external:
self.stream.close()
def __enter__(self) -> "PDF":
return self
def __exit__(
self,
t: Optional[Type[BaseException]],
value: Optional[BaseException],
traceback: Optional[TracebackType],
) -> None:
self.close()
@property
def pages(self) -> List[Page]:
if hasattr(self, "_pages"):
return self._pages
doctop: T_num = 0
pp = self.pages_to_parse
self._pages: List[Page] = []
def iter_pages() -> Generator[PDFPage, None, None]:
gen = PDFPage.create_pages(self.doc)
while True:
try:
yield next(gen)
except StopIteration:
break
except Exception as e:
raise PdfminerException(e)
for i, page in enumerate(iter_pages()):
page_number = i + 1
if pp is not None and page_number not in pp:
continue
p = Page(self, page, page_number=page_number, initial_doctop=doctop)
self._pages.append(p)
doctop += p.height
return self._pages
@property
def objects(self) -> Dict[str, T_obj_list]:
if hasattr(self, "_objects"):
return self._objects
all_objects: Dict[str, T_obj_list] = {}
for p in self.pages:
for kind in p.objects.keys():
all_objects[kind] = all_objects.get(kind, []) + p.objects[kind]
self._objects: Dict[str, T_obj_list] = all_objects
return self._objects
@property
def annots(self) -> List[Dict[str, Any]]:
gen = (p.annots for p in self.pages)
return list(itertools.chain(*gen))
@property
def hyperlinks(self) -> List[Dict[str, Any]]:
gen = (p.hyperlinks for p in self.pages)
return list(itertools.chain(*gen))
@property
def structure_tree(self) -> List[Dict[str, Any]]:
"""Return the structure tree for the document."""
try:
return [elem.to_dict() for elem in PDFStructTree(self)]
except StructTreeMissing:
return []
def to_dict(self, object_types: Optional[List[str]] = None) -> Dict[str, Any]:
return {
"metadata": self.metadata,
"pages": [page.to_dict(object_types) for page in self.pages],
}
@@ -0,0 +1,76 @@
import pathlib
import shutil
import subprocess
from io import BufferedReader, BytesIO
from typing import Literal, Optional, Union
T_repair_setting = Literal["default", "prepress", "printer", "ebook", "screen"]
def _repair(
path_or_fp: Union[str, pathlib.Path, BufferedReader, BytesIO],
password: Optional[str] = None,
gs_path: Optional[Union[str, pathlib.Path]] = None,
setting: T_repair_setting = "default",
) -> BytesIO:
executable = (
gs_path
or shutil.which("gs")
or shutil.which("gswin32c")
or shutil.which("gswin64c")
)
if executable is None: # pragma: nocover
raise Exception(
"Cannot find Ghostscript, which is required for repairs.\n"
"Visit https://www.ghostscript.com/ for installation instructions."
)
repair_args = [
executable,
"-sstdout=%stderr",
"-o",
"-",
"-sDEVICE=pdfwrite",
f"-dPDFSETTINGS=/{setting}",
]
if password:
repair_args += [f"-sPDFPassword={password}"]
if isinstance(path_or_fp, (str, pathlib.Path)):
stdin = None
repair_args += [str(pathlib.Path(path_or_fp).absolute())]
else:
stdin = path_or_fp
repair_args += ["-"]
proc = subprocess.Popen(
repair_args,
stdin=subprocess.PIPE if stdin else None,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
stdout, stderr = proc.communicate(stdin.read() if stdin else None)
if proc.returncode:
raise Exception(f"{stderr.decode('utf-8')}")
return BytesIO(stdout)
def repair(
path_or_fp: Union[str, pathlib.Path, BufferedReader, BytesIO],
outfile: Optional[Union[str, pathlib.Path]] = None,
password: Optional[str] = None,
gs_path: Optional[Union[str, pathlib.Path]] = None,
setting: T_repair_setting = "default",
) -> Optional[BytesIO]:
repaired = _repair(path_or_fp, password, gs_path=gs_path, setting=setting)
if outfile:
with open(outfile, "wb") as f:
f.write(repaired.read())
return None
else:
return repaired
@@ -0,0 +1,511 @@
import itertools
import logging
import re
from collections import deque
from dataclasses import asdict, dataclass, field
from typing import (
TYPE_CHECKING,
Any,
Callable,
Dict,
Iterable,
Iterator,
List,
Optional,
Pattern,
Tuple,
Union,
)
from pdfminer.data_structures import NumberTree
from pdfminer.pdfparser import PDFParser
from pdfminer.pdftypes import PDFObjRef, resolve1
from pdfminer.psparser import PSLiteral
from ._typing import T_bbox, T_obj
from .utils import decode_text, geometry
logger = logging.getLogger(__name__)
if TYPE_CHECKING: # pragma: nocover
from .page import Page
from .pdf import PDF
MatchFunc = Callable[["PDFStructElement"], bool]
def _find_all(
elements: Iterable["PDFStructElement"],
matcher: Union[str, Pattern[str], MatchFunc],
) -> Iterator["PDFStructElement"]:
"""
Common code for `find_all()` in trees and elements.
"""
def match_tag(x: "PDFStructElement") -> bool:
"""Match an element name."""
return x.type == matcher
def match_regex(x: "PDFStructElement") -> bool:
"""Match an element name by regular expression."""
return matcher.match(x.type) # type: ignore
if isinstance(matcher, str):
match_func = match_tag
elif isinstance(matcher, re.Pattern):
match_func = match_regex
else:
match_func = matcher # type: ignore
d = deque(elements)
while d:
el = d.popleft()
if match_func(el):
yield el
d.extendleft(reversed(el.children))
class Findable:
"""find() and find_all() methods that can be inherited to avoid
repeating oneself"""
children: List["PDFStructElement"]
def find_all(
self, matcher: Union[str, Pattern[str], MatchFunc]
) -> Iterator["PDFStructElement"]:
"""Iterate depth-first over matching elements in subtree.
The `matcher` argument is either an element name, a regular
expression, or a function taking a `PDFStructElement` and
returning `True` if the element matches.
"""
return _find_all(self.children, matcher)
def find(
self, matcher: Union[str, Pattern[str], MatchFunc]
) -> Optional["PDFStructElement"]:
"""Find the first matching element in subtree.
The `matcher` argument is either an element name, a regular
expression, or a function taking a `PDFStructElement` and
returning `True` if the element matches.
"""
try:
return next(_find_all(self.children, matcher))
except StopIteration:
return None
@dataclass
class PDFStructElement(Findable):
type: str
revision: Optional[int]
id: Optional[str]
lang: Optional[str]
alt_text: Optional[str]
actual_text: Optional[str]
title: Optional[str]
page_number: Optional[int]
attributes: Dict[str, Any] = field(default_factory=dict)
mcids: List[int] = field(default_factory=list)
children: List["PDFStructElement"] = field(default_factory=list)
def __iter__(self) -> Iterator["PDFStructElement"]:
return iter(self.children)
def all_mcids(self) -> Iterator[Tuple[Optional[int], int]]:
"""Collect all MCIDs (with their page numbers, if there are
multiple pages in the tree) inside a structure element.
"""
# Collect them depth-first to preserve ordering
for mcid in self.mcids:
yield self.page_number, mcid
d = deque(self.children)
while d:
el = d.popleft()
for mcid in el.mcids:
yield el.page_number, mcid
d.extendleft(reversed(el.children))
def to_dict(self) -> Dict[str, Any]:
"""Return a compacted dict representation."""
r = asdict(self)
# Prune empty values (does not matter in which order)
d = deque([r])
while d:
el = d.popleft()
for k in list(el.keys()):
if el[k] is None or el[k] == [] or el[k] == {}:
del el[k]
if "children" in el:
d.extend(el["children"])
return r
class StructTreeMissing(ValueError):
pass
class PDFStructTree(Findable):
"""Parse the structure tree of a PDF.
The constructor takes a `pdfplumber.PDF` and optionally a
`pdfplumber.Page`. To avoid creating the entire tree for a large
document it is recommended to provide a page.
This class creates a representation of the portion of the
structure tree that reaches marked content sections, either for a
single page, or for the whole document. Note that this is slightly
different from the behaviour of other PDF libraries which will
also include structure elements with no content.
If the PDF has no structure, the constructor will raise
`StructTreeMissing`.
"""
page: Optional["Page"]
def __init__(self, doc: "PDF", page: Optional["Page"] = None):
self.doc = doc.doc
if "StructTreeRoot" not in self.doc.catalog:
raise StructTreeMissing("PDF has no structure")
self.root = resolve1(self.doc.catalog["StructTreeRoot"])
self.role_map = resolve1(self.root.get("RoleMap", {}))
self.class_map = resolve1(self.root.get("ClassMap", {}))
self.children: List[PDFStructElement] = []
# If we have a specific page then we will work backwards from
# its ParentTree - this is because structure elements could
# span multiple pages, and the "Pg" attribute is *optional*,
# so this is the approved way to get a page's structure...
if page is not None:
self.page = page
self.pages = {page.page_number: page}
self.page_dict = None
# ...EXCEPT that the ParentTree is sometimes missing, in which
# case we fall back to the non-approved way.
parent_tree_obj = self.root.get("ParentTree")
if parent_tree_obj is None:
self._parse_struct_tree()
else:
parent_tree = NumberTree(parent_tree_obj)
# If there is no marked content in the structure tree for
# this page (which can happen even when there is a
# structure tree) then there is no `StructParents`.
# Note however that if there are XObjects in a page,
# *they* may have `StructParent` (not `StructParents`)
if "StructParents" not in self.page.page_obj.attrs:
return
parent_id = self.page.page_obj.attrs["StructParents"]
# NumberTree should have a `get` method like it does in pdf.js...
parent_array = resolve1(
next(array for num, array in parent_tree.values if num == parent_id)
)
self._parse_parent_tree(parent_array)
else:
self.page = None
# Overhead of creating pages shouldn't be too bad we hope!
self.pages = {page.page_number: page for page in doc.pages}
self.page_dict = {
page.page_obj.pageid: page.page_number for page in self.pages.values()
}
self._parse_struct_tree()
def _make_attributes(
self, obj: Dict[str, Any], revision: Optional[int]
) -> Dict[str, Any]:
attr_obj_list = []
for key in "C", "A":
if key not in obj:
continue
attr_obj = resolve1(obj[key])
# It could be a list of attribute objects (why?)
if isinstance(attr_obj, list):
attr_obj_list.extend(attr_obj)
else:
attr_obj_list.append(attr_obj)
attr_objs = []
prev_obj = None
for aref in attr_obj_list:
# If we find a revision number, which might "follow the
# revision object" (the spec is not clear about what this
# should look like but it implies they are simply adjacent
# in a flat array), then use it to decide whether to take
# the previous object...
if isinstance(aref, int):
if aref == revision and prev_obj is not None:
attr_objs.append(prev_obj)
prev_obj = None
else:
if prev_obj is not None:
attr_objs.append(prev_obj)
prev_obj = resolve1(aref)
if prev_obj is not None:
attr_objs.append(prev_obj)
# Now merge all the attribute objects in the collected to a
# single set (again, the spec doesn't really explain this but
# does say that attributes in /A supersede those in /C)
attr = {}
for obj in attr_objs:
if isinstance(obj, PSLiteral):
key = decode_text(obj.name)
if key not in self.class_map:
logger.warning("Unknown attribute class %s", key)
continue
obj = self.class_map[key]
for k, v in obj.items():
if isinstance(v, PSLiteral):
attr[k] = decode_text(v.name)
else:
attr[k] = obj[k]
return attr
def _make_element(self, obj: Any) -> Tuple[Optional[PDFStructElement], List[Any]]:
# We hopefully caught these earlier
assert "MCID" not in obj, "Uncaught MCR: %s" % obj
assert "Obj" not in obj, "Uncaught OBJR: %s" % obj
# Get page number if necessary
page_number = None
if self.page_dict is not None and "Pg" in obj:
page_objid = obj["Pg"].objid
assert page_objid in self.page_dict, "Object on unparsed page: %s" % obj
page_number = self.page_dict[page_objid]
obj_tag = ""
if "S" in obj:
obj_tag = decode_text(obj["S"].name)
if obj_tag in self.role_map:
obj_tag = decode_text(self.role_map[obj_tag].name)
children = resolve1(obj["K"]) if "K" in obj else []
if isinstance(children, int): # ugh... isinstance...
children = [children]
elif isinstance(children, dict): # a single object.. ugh...
children = [obj["K"]]
revision = obj.get("R")
attributes = self._make_attributes(obj, revision)
element_id = decode_text(resolve1(obj["ID"])) if "ID" in obj else None
title = decode_text(resolve1(obj["T"])) if "T" in obj else None
lang = decode_text(resolve1(obj["Lang"])) if "Lang" in obj else None
alt_text = decode_text(resolve1(obj["Alt"])) if "Alt" in obj else None
actual_text = (
decode_text(resolve1(obj["ActualText"])) if "ActualText" in obj else None
)
element = PDFStructElement(
type=obj_tag,
id=element_id,
page_number=page_number,
revision=revision,
lang=lang,
title=title,
alt_text=alt_text,
actual_text=actual_text,
attributes=attributes,
)
return element, children
def _parse_parent_tree(self, parent_array: List[Any]) -> None:
"""Populate the structure tree using the leaves of the parent tree for
a given page."""
# First walk backwards from the leaves to the root, tracking references
d = deque(parent_array)
s = {}
found_root = False
while d:
ref = d.popleft()
# In the case where an MCID is not associated with any
# structure, there will be a "null" in the parent tree.
if ref == PDFParser.KEYWORD_NULL:
continue
if repr(ref) in s:
continue
obj = resolve1(ref)
# This is required! It's in the spec!
if "Type" in obj and decode_text(obj["Type"].name) == "StructTreeRoot":
found_root = True
else:
# We hope that these are actual elements and not
# references or marked-content sections...
element, children = self._make_element(obj)
# We have no page tree so we assume this page was parsed
assert element is not None
s[repr(ref)] = element, children
d.append(obj["P"])
# If we didn't reach the root something is quite wrong!
assert found_root
self._resolve_children(s)
def on_parsed_page(self, obj: Dict[str, Any]) -> bool:
if "Pg" not in obj:
return True
page_objid = obj["Pg"].objid
if self.page_dict is not None:
return page_objid in self.page_dict
if self.page is not None:
# We have to do this to satisfy mypy
if page_objid != self.page.page_obj.pageid:
return False
return True
def _parse_struct_tree(self) -> None:
"""Populate the structure tree starting from the root, skipping
unparsed pages and empty elements."""
root = resolve1(self.root["K"])
# It could just be a single object ... it's in the spec (argh)
if isinstance(root, dict):
root = [self.root["K"]]
d = deque(root)
s = {}
while d:
ref = d.popleft()
# In case the tree is actually a DAG and not a tree...
if repr(ref) in s: # pragma: nocover (shouldn't happen)
continue
obj = resolve1(ref)
# Deref top-level OBJR skipping refs to unparsed pages
if isinstance(obj, dict) and "Obj" in obj:
if not self.on_parsed_page(obj):
continue
ref = obj["Obj"]
obj = resolve1(ref)
element, children = self._make_element(obj)
# Similar to above, delay resolving the children to avoid
# tree-recursion.
s[repr(ref)] = element, children
for child in children:
obj = resolve1(child)
if isinstance(obj, dict):
if not self.on_parsed_page(obj):
continue
if "Obj" in obj:
child = obj["Obj"]
elif "MCID" in obj:
continue
if isinstance(child, PDFObjRef):
d.append(child)
# Traverse depth-first, removing empty elements (unsure how to
# do this non-recursively)
def prune(elements: List[Any]) -> List[Any]:
next_elements = []
for ref in elements:
obj = resolve1(ref)
if isinstance(ref, int):
next_elements.append(ref)
continue
elif isinstance(obj, dict):
if not self.on_parsed_page(obj):
continue
if "MCID" in obj:
next_elements.append(obj["MCID"])
continue
elif "Obj" in obj:
ref = obj["Obj"]
element, children = s[repr(ref)]
children = prune(children)
# See assertions below
if element is None or not children:
del s[repr(ref)]
else:
s[repr(ref)] = element, children
next_elements.append(ref)
return next_elements
prune(root)
self._resolve_children(s)
def _resolve_children(self, seen: Dict[str, Any]) -> None:
"""Resolve children starting from the tree root based on references we
saw when traversing the structure tree.
"""
root = resolve1(self.root["K"])
# It could just be a single object ... it's in the spec (argh)
if isinstance(root, dict):
root = [self.root["K"]]
self.children = []
# Create top-level self.children
parsed_root = []
for ref in root:
obj = resolve1(ref)
if isinstance(obj, dict) and "Obj" in obj:
if not self.on_parsed_page(obj):
continue
ref = obj["Obj"]
if repr(ref) in seen:
parsed_root.append(ref)
d = deque(parsed_root)
while d:
ref = d.popleft()
element, children = seen[repr(ref)]
assert element is not None, "Unparsed element"
for child in children:
obj = resolve1(child)
if isinstance(obj, int):
element.mcids.append(obj)
elif isinstance(obj, dict):
# Skip out-of-page MCIDS and OBJRs
if not self.on_parsed_page(obj):
continue
if "MCID" in obj:
element.mcids.append(obj["MCID"])
elif "Obj" in obj:
child = obj["Obj"]
# NOTE: if, not elif, in case of OBJR above
if isinstance(child, PDFObjRef):
child_element, _ = seen.get(repr(child), (None, None))
if child_element is not None:
element.children.append(child_element)
d.append(child)
self.children = [seen[repr(ref)][0] for ref in parsed_root]
def __iter__(self) -> Iterator[PDFStructElement]:
return iter(self.children)
def element_bbox(self, el: PDFStructElement) -> T_bbox:
"""Get the bounding box for an element for visual debugging."""
page = None
if self.page is not None:
page = self.page
elif el.page_number is not None:
page = self.pages[el.page_number]
bbox = el.attributes.get("BBox", None)
if page is not None and bbox is not None:
from .page import CroppedPage, _invert_box, _normalize_box
# Use secret knowledge of CroppedPage (cannot use
# page.height because it is the *cropped* dimension, but
# cropping does not actually translate coordinates)
bbox = _invert_box(
_normalize_box(bbox), page.mediabox[3] - page.mediabox[1]
)
# Use more secret knowledge of CroppedPage
if isinstance(page, CroppedPage):
rect = geometry.bbox_to_rect(bbox)
rects = page._crop_fn([rect])
if not rects:
raise IndexError("Element no longer on page")
return geometry.obj_to_bbox(rects[0])
else:
# Not sure why mypy complains here
return bbox # type: ignore
else:
mcid_objs = []
for page_number, mcid in el.all_mcids():
objects: Iterable[T_obj]
if page_number is None:
if page is not None:
objects = itertools.chain.from_iterable(page.objects.values())
else:
objects = [] # pragma: nocover
else:
objects = itertools.chain.from_iterable(
self.pages[page_number].objects.values()
)
for c in objects:
if c["mcid"] == mcid:
mcid_objs.append(c)
if not mcid_objs:
raise IndexError("No objects found") # pragma: nocover
return geometry.objects_to_bbox(mcid_objs)
@@ -0,0 +1,691 @@
import itertools
from dataclasses import dataclass
from operator import itemgetter
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Tuple, Type, Union
from . import utils
from ._typing import T_bbox, T_num, T_obj, T_obj_iter, T_obj_list, T_point
DEFAULT_SNAP_TOLERANCE = 3
DEFAULT_JOIN_TOLERANCE = 3
DEFAULT_MIN_WORDS_VERTICAL = 3
DEFAULT_MIN_WORDS_HORIZONTAL = 1
T_intersections = Dict[T_point, Dict[str, T_obj_list]]
T_table_settings = Union["TableSettings", Dict[str, Any]]
if TYPE_CHECKING: # pragma: nocover
from .page import Page
def snap_edges(
edges: T_obj_list,
x_tolerance: T_num = DEFAULT_SNAP_TOLERANCE,
y_tolerance: T_num = DEFAULT_SNAP_TOLERANCE,
) -> T_obj_list:
"""
Given a list of edges, snap any within `tolerance` pixels of one another
to their positional average.
"""
by_orientation: Dict[str, T_obj_list] = {"v": [], "h": []}
for e in edges:
by_orientation[e["orientation"]].append(e)
snapped_v = utils.snap_objects(by_orientation["v"], "x0", x_tolerance)
snapped_h = utils.snap_objects(by_orientation["h"], "top", y_tolerance)
return snapped_v + snapped_h
def join_edge_group(
edges: T_obj_iter, orientation: str, tolerance: T_num = DEFAULT_JOIN_TOLERANCE
) -> T_obj_list:
"""
Given a list of edges along the same infinite line, join those that
are within `tolerance` pixels of one another.
"""
if orientation == "h":
min_prop, max_prop = "x0", "x1"
elif orientation == "v":
min_prop, max_prop = "top", "bottom"
else:
raise ValueError("Orientation must be 'v' or 'h'")
sorted_edges = list(sorted(edges, key=itemgetter(min_prop)))
joined = [sorted_edges[0]]
for e in sorted_edges[1:]:
last = joined[-1]
if e[min_prop] <= (last[max_prop] + tolerance):
if e[max_prop] > last[max_prop]:
# Extend current edge to new extremity
joined[-1] = utils.resize_object(last, max_prop, e[max_prop])
else:
# Edge is separate from previous edges
joined.append(e)
return joined
def merge_edges(
edges: T_obj_list,
snap_x_tolerance: T_num,
snap_y_tolerance: T_num,
join_x_tolerance: T_num,
join_y_tolerance: T_num,
) -> T_obj_list:
"""
Using the `snap_edges` and `join_edge_group` methods above,
merge a list of edges into a more "seamless" list.
"""
def get_group(edge: T_obj) -> Tuple[str, T_num]:
if edge["orientation"] == "h":
return ("h", edge["top"])
else:
return ("v", edge["x0"])
if snap_x_tolerance > 0 or snap_y_tolerance > 0:
edges = snap_edges(edges, snap_x_tolerance, snap_y_tolerance)
_sorted = sorted(edges, key=get_group)
edge_groups = itertools.groupby(_sorted, key=get_group)
edge_gen = (
join_edge_group(
items, k[0], (join_x_tolerance if k[0] == "h" else join_y_tolerance)
)
for k, items in edge_groups
)
edges = list(itertools.chain(*edge_gen))
return edges
def words_to_edges_h(
words: T_obj_list, word_threshold: int = DEFAULT_MIN_WORDS_HORIZONTAL
) -> T_obj_list:
"""
Find (imaginary) horizontal lines that connect the tops
of at least `word_threshold` words.
"""
by_top = utils.cluster_objects(words, itemgetter("top"), 1)
large_clusters = filter(lambda x: len(x) >= word_threshold, by_top)
rects = list(map(utils.objects_to_rect, large_clusters))
if len(rects) == 0:
return []
min_x0 = min(map(itemgetter("x0"), rects))
max_x1 = max(map(itemgetter("x1"), rects))
edges = []
for r in rects:
edges += [
# Top of text
{
"x0": min_x0,
"x1": max_x1,
"top": r["top"],
"bottom": r["top"],
"width": max_x1 - min_x0,
"orientation": "h",
},
# For each detected row, we also add the 'bottom' line. This will
# generate extra edges, (some will be redundant with the next row
# 'top' line), but this catches the last row of every table.
{
"x0": min_x0,
"x1": max_x1,
"top": r["bottom"],
"bottom": r["bottom"],
"width": max_x1 - min_x0,
"orientation": "h",
},
]
return edges
def words_to_edges_v(
words: T_obj_list, word_threshold: int = DEFAULT_MIN_WORDS_VERTICAL
) -> T_obj_list:
"""
Find (imaginary) vertical lines that connect the left, right, or
center of at least `word_threshold` words.
"""
# Find words that share the same left, right, or centerpoints
by_x0 = utils.cluster_objects(words, itemgetter("x0"), 1)
by_x1 = utils.cluster_objects(words, itemgetter("x1"), 1)
def get_center(word: T_obj) -> T_num:
return float(word["x0"] + word["x1"]) / 2
by_center = utils.cluster_objects(words, get_center, 1)
clusters = by_x0 + by_x1 + by_center
# Find the points that align with the most words
sorted_clusters = sorted(clusters, key=lambda x: -len(x))
large_clusters = filter(lambda x: len(x) >= word_threshold, sorted_clusters)
# For each of those points, find the bboxes fitting all matching words
bboxes = list(map(utils.objects_to_bbox, large_clusters))
# Iterate through those bboxes, condensing overlapping bboxes
condensed_bboxes: List[T_bbox] = []
for bbox in bboxes:
overlap = any(utils.get_bbox_overlap(bbox, c) for c in condensed_bboxes)
if not overlap:
condensed_bboxes.append(bbox)
if len(condensed_bboxes) == 0:
return []
condensed_rects = map(utils.bbox_to_rect, condensed_bboxes)
sorted_rects = list(sorted(condensed_rects, key=itemgetter("x0")))
max_x1 = max(map(itemgetter("x1"), sorted_rects))
min_top = min(map(itemgetter("top"), sorted_rects))
max_bottom = max(map(itemgetter("bottom"), sorted_rects))
return [
{
"x0": b["x0"],
"x1": b["x0"],
"top": min_top,
"bottom": max_bottom,
"height": max_bottom - min_top,
"orientation": "v",
}
for b in sorted_rects
] + [
{
"x0": max_x1,
"x1": max_x1,
"top": min_top,
"bottom": max_bottom,
"height": max_bottom - min_top,
"orientation": "v",
}
]
def edges_to_intersections(
edges: T_obj_list, x_tolerance: T_num = 1, y_tolerance: T_num = 1
) -> T_intersections:
"""
Given a list of edges, return the points at which they intersect
within `tolerance` pixels.
"""
intersections: T_intersections = {}
v_edges, h_edges = [
list(filter(lambda x: x["orientation"] == o, edges)) for o in ("v", "h")
]
for v in sorted(v_edges, key=itemgetter("x0", "top")):
for h in sorted(h_edges, key=itemgetter("top", "x0")):
if (
(v["top"] <= (h["top"] + y_tolerance))
and (v["bottom"] >= (h["top"] - y_tolerance))
and (v["x0"] >= (h["x0"] - x_tolerance))
and (v["x0"] <= (h["x1"] + x_tolerance))
):
vertex = (v["x0"], h["top"])
if vertex not in intersections:
intersections[vertex] = {"v": [], "h": []}
intersections[vertex]["v"].append(v)
intersections[vertex]["h"].append(h)
return intersections
def intersections_to_cells(intersections: T_intersections) -> List[T_bbox]:
"""
Given a list of points (`intersections`), return all rectangular "cells"
that those points describe.
`intersections` should be a dictionary with (x0, top) tuples as keys,
and a list of edge objects as values. The edge objects should correspond
to the edges that touch the intersection.
"""
def edge_connects(p1: T_point, p2: T_point) -> bool:
def edges_to_set(edges: T_obj_list) -> Set[T_bbox]:
return set(map(utils.obj_to_bbox, edges))
if p1[0] == p2[0]:
common = edges_to_set(intersections[p1]["v"]).intersection(
edges_to_set(intersections[p2]["v"])
)
if len(common):
return True
if p1[1] == p2[1]:
common = edges_to_set(intersections[p1]["h"]).intersection(
edges_to_set(intersections[p2]["h"])
)
if len(common):
return True
return False
points = list(sorted(intersections.keys()))
n_points = len(points)
def find_smallest_cell(points: List[T_point], i: int) -> Optional[T_bbox]:
if i == n_points - 1:
return None
pt = points[i]
rest = points[i + 1 :]
# Get all the points directly below and directly right
below = [x for x in rest if x[0] == pt[0]]
right = [x for x in rest if x[1] == pt[1]]
for below_pt in below:
if not edge_connects(pt, below_pt):
continue
for right_pt in right:
if not edge_connects(pt, right_pt):
continue
bottom_right = (right_pt[0], below_pt[1])
if (
(bottom_right in intersections)
and edge_connects(bottom_right, right_pt)
and edge_connects(bottom_right, below_pt)
):
return (pt[0], pt[1], bottom_right[0], bottom_right[1])
return None
cell_gen = (find_smallest_cell(points, i) for i in range(len(points)))
return list(filter(None, cell_gen))
def cells_to_tables(cells: List[T_bbox]) -> List[List[T_bbox]]:
"""
Given a list of bounding boxes (`cells`), return a list of tables that
hold those cells most simply (and contiguously).
"""
def bbox_to_corners(bbox: T_bbox) -> Tuple[T_point, T_point, T_point, T_point]:
x0, top, x1, bottom = bbox
return ((x0, top), (x0, bottom), (x1, top), (x1, bottom))
remaining_cells = list(cells)
# Iterate through the cells found above, and assign them
# to contiguous tables
current_corners: Set[T_point] = set()
current_cells: List[T_bbox] = []
tables = []
while len(remaining_cells):
initial_cell_count = len(current_cells)
for cell in list(remaining_cells):
cell_corners = bbox_to_corners(cell)
# If we're just starting a table ...
if len(current_cells) == 0:
# ... immediately assign it to the empty group
current_corners |= set(cell_corners)
current_cells.append(cell)
remaining_cells.remove(cell)
else:
# How many corners does this table share with the current group?
corner_count = sum(c in current_corners for c in cell_corners)
# If touching on at least one corner...
if corner_count > 0:
# ... assign it to the current group
current_corners |= set(cell_corners)
current_cells.append(cell)
remaining_cells.remove(cell)
# If this iteration did not find any more cells to append...
if len(current_cells) == initial_cell_count:
# ... start a new cell group
tables.append(list(current_cells))
current_corners.clear()
current_cells.clear()
# Once we have exhausting the list of cells ...
# ... and we have a cell group that has not been stored
if len(current_cells):
# ... store it.
tables.append(list(current_cells))
# Sort the tables top-to-bottom-left-to-right based on the value of the
# topmost-and-then-leftmost coordinate of a table.
_sorted = sorted(tables, key=lambda t: min((c[1], c[0]) for c in t))
filtered = [t for t in _sorted if len(t) > 1]
return filtered
class CellGroup(object):
def __init__(self, cells: List[Optional[T_bbox]]):
self.cells = cells
self.bbox = (
min(map(itemgetter(0), filter(None, cells))),
min(map(itemgetter(1), filter(None, cells))),
max(map(itemgetter(2), filter(None, cells))),
max(map(itemgetter(3), filter(None, cells))),
)
class Row(CellGroup):
pass
class Column(CellGroup):
pass
class Table(object):
def __init__(self, page: "Page", cells: List[T_bbox]):
self.page = page
self.cells = cells
@property
def bbox(self) -> T_bbox:
c = self.cells
return (
min(map(itemgetter(0), c)),
min(map(itemgetter(1), c)),
max(map(itemgetter(2), c)),
max(map(itemgetter(3), c)),
)
def _get_rows_or_cols(self, kind: Type[CellGroup]) -> List[CellGroup]:
axis = 0 if kind is Row else 1
antiaxis = int(not axis)
# Sort first by top/x0, then by x0/top
_sorted = sorted(self.cells, key=itemgetter(antiaxis, axis))
# Sort get all x0s/tops
xs = list(sorted(set(map(itemgetter(axis), self.cells))))
# Group by top/x0
grouped = itertools.groupby(_sorted, itemgetter(antiaxis))
rows = []
# for y/x, row/column-cells ...
for y, row_cells in grouped:
xdict = {cell[axis]: cell for cell in row_cells}
row = kind([xdict.get(x) for x in xs])
rows.append(row)
return rows
@property
def rows(self) -> List[CellGroup]:
return self._get_rows_or_cols(Row)
@property
def columns(self) -> List[CellGroup]:
return self._get_rows_or_cols(Column)
def extract(self, **kwargs: Any) -> List[List[Optional[str]]]:
chars = self.page.chars
table_arr = []
def char_in_bbox(char: T_obj, bbox: T_bbox) -> bool:
v_mid = (char["top"] + char["bottom"]) / 2
h_mid = (char["x0"] + char["x1"]) / 2
x0, top, x1, bottom = bbox
return bool(
(h_mid >= x0) and (h_mid < x1) and (v_mid >= top) and (v_mid < bottom)
)
for row in self.rows:
arr = []
row_chars = [char for char in chars if char_in_bbox(char, row.bbox)]
for cell in row.cells:
if cell is None:
cell_text = None
else:
cell_chars = [
char for char in row_chars if char_in_bbox(char, cell)
]
if len(cell_chars):
if "layout" in kwargs:
kwargs["layout_width"] = cell[2] - cell[0]
kwargs["layout_height"] = cell[3] - cell[1]
kwargs["layout_bbox"] = cell
cell_text = utils.extract_text(cell_chars, **kwargs)
else:
cell_text = ""
arr.append(cell_text)
table_arr.append(arr)
return table_arr
TABLE_STRATEGIES = ["lines", "lines_strict", "text", "explicit"]
NON_NEGATIVE_SETTINGS = [
"snap_tolerance",
"snap_x_tolerance",
"snap_y_tolerance",
"join_tolerance",
"join_x_tolerance",
"join_y_tolerance",
"edge_min_length",
"min_words_vertical",
"min_words_horizontal",
"intersection_tolerance",
"intersection_x_tolerance",
"intersection_y_tolerance",
]
class UnsetFloat(float):
pass
UNSET = UnsetFloat(0)
@dataclass
class TableSettings:
vertical_strategy: str = "lines"
horizontal_strategy: str = "lines"
explicit_vertical_lines: Optional[List[Union[T_obj, T_num]]] = None
explicit_horizontal_lines: Optional[List[Union[T_obj, T_num]]] = None
snap_tolerance: T_num = DEFAULT_SNAP_TOLERANCE
snap_x_tolerance: T_num = UNSET
snap_y_tolerance: T_num = UNSET
join_tolerance: T_num = DEFAULT_JOIN_TOLERANCE
join_x_tolerance: T_num = UNSET
join_y_tolerance: T_num = UNSET
edge_min_length: T_num = 3
min_words_vertical: int = DEFAULT_MIN_WORDS_VERTICAL
min_words_horizontal: int = DEFAULT_MIN_WORDS_HORIZONTAL
intersection_tolerance: T_num = 3
intersection_x_tolerance: T_num = UNSET
intersection_y_tolerance: T_num = UNSET
text_settings: Optional[Dict[str, Any]] = None
def __post_init__(self) -> None:
"""Clean up user-provided table settings.
Validates that the table settings provided consists of acceptable values and
returns a cleaned up version. The cleaned up version fills out the missing
values with the default values in the provided settings.
TODO: Can be further used to validate that the values are of the correct
type. For example, raising a value error when a non-boolean input is
provided for the key ``keep_blank_chars``.
:param table_settings: User-provided table settings.
:returns: A cleaned up version of the user-provided table settings.
:raises ValueError: When an unrecognised key is provided.
"""
for setting in NON_NEGATIVE_SETTINGS:
if (getattr(self, setting) or 0) < 0:
raise ValueError(f"Table setting '{setting}' cannot be negative")
for orientation in ["horizontal", "vertical"]:
strategy = getattr(self, orientation + "_strategy")
if strategy not in TABLE_STRATEGIES:
raise ValueError(
f"{orientation}_strategy must be one of"
f'{{{",".join(TABLE_STRATEGIES)}}}'
)
if self.text_settings is None:
self.text_settings = {}
# This next section is for backwards compatibility
for attr in ["x_tolerance", "y_tolerance"]:
if attr not in self.text_settings:
self.text_settings[attr] = self.text_settings.get("tolerance", 3)
if "tolerance" in self.text_settings:
del self.text_settings["tolerance"]
# End of that section
for attr, fallback in [
("snap_x_tolerance", "snap_tolerance"),
("snap_y_tolerance", "snap_tolerance"),
("join_x_tolerance", "join_tolerance"),
("join_y_tolerance", "join_tolerance"),
("intersection_x_tolerance", "intersection_tolerance"),
("intersection_y_tolerance", "intersection_tolerance"),
]:
if getattr(self, attr) is UNSET:
setattr(self, attr, getattr(self, fallback))
@classmethod
def resolve(cls, settings: Optional[T_table_settings]) -> "TableSettings":
if settings is None:
return cls()
elif isinstance(settings, cls):
return settings
elif isinstance(settings, dict):
core_settings = {}
text_settings = {}
for k, v in settings.items():
if k[:5] == "text_":
text_settings[k[5:]] = v
else:
core_settings[k] = v
core_settings["text_settings"] = text_settings
return cls(**core_settings)
else:
raise ValueError(f"Cannot resolve settings: {settings}")
class TableFinder(object):
"""
Given a PDF page, find plausible table structures.
Largely borrowed from Anssi Nurminen's master's thesis:
http://dspace.cc.tut.fi/dpub/bitstream/handle/123456789/21520/Nurminen.pdf?sequence=3
... and inspired by Tabula:
https://github.com/tabulapdf/tabula-extractor/issues/16
"""
def __init__(self, page: "Page", settings: Optional[T_table_settings] = None):
self.page = page
self.settings = TableSettings.resolve(settings)
self.edges = self.get_edges()
self.intersections = edges_to_intersections(
self.edges,
self.settings.intersection_x_tolerance,
self.settings.intersection_y_tolerance,
)
self.cells = intersections_to_cells(self.intersections)
self.tables = [
Table(self.page, cell_group) for cell_group in cells_to_tables(self.cells)
]
def get_edges(self) -> T_obj_list:
settings = self.settings
for orientation in ["vertical", "horizontal"]:
strategy = getattr(settings, orientation + "_strategy")
if strategy == "explicit":
lines = getattr(settings, "explicit_" + orientation + "_lines")
if len(lines) < 2:
raise ValueError(
f"If {orientation}_strategy == 'explicit', "
f"explicit_{orientation}_lines "
f"must be specified as a list/tuple of two or more "
f"floats/ints."
)
v_strat = settings.vertical_strategy
h_strat = settings.horizontal_strategy
if v_strat == "text" or h_strat == "text":
words = self.page.extract_words(**(settings.text_settings or {}))
v_explicit = []
for desc in settings.explicit_vertical_lines or []:
if isinstance(desc, dict):
for e in utils.obj_to_edges(desc):
if e["orientation"] == "v":
v_explicit.append(e)
else:
v_explicit.append(
{
"x0": desc,
"x1": desc,
"top": self.page.bbox[1],
"bottom": self.page.bbox[3],
"height": self.page.bbox[3] - self.page.bbox[1],
"orientation": "v",
}
)
if v_strat == "lines":
v_base = utils.filter_edges(self.page.edges, "v")
elif v_strat == "lines_strict":
v_base = utils.filter_edges(self.page.edges, "v", edge_type="line")
elif v_strat == "text":
v_base = words_to_edges_v(words, word_threshold=settings.min_words_vertical)
elif v_strat == "explicit":
v_base = []
v = v_base + v_explicit
h_explicit = []
for desc in settings.explicit_horizontal_lines or []:
if isinstance(desc, dict):
for e in utils.obj_to_edges(desc):
if e["orientation"] == "h":
h_explicit.append(e)
else:
h_explicit.append(
{
"x0": self.page.bbox[0],
"x1": self.page.bbox[2],
"width": self.page.bbox[2] - self.page.bbox[0],
"top": desc,
"bottom": desc,
"orientation": "h",
}
)
if h_strat == "lines":
h_base = utils.filter_edges(self.page.edges, "h")
elif h_strat == "lines_strict":
h_base = utils.filter_edges(self.page.edges, "h", edge_type="line")
elif h_strat == "text":
h_base = words_to_edges_h(
words, word_threshold=settings.min_words_horizontal
)
elif h_strat == "explicit":
h_base = []
h = h_base + h_explicit
edges = list(v) + list(h)
edges = merge_edges(
edges,
snap_x_tolerance=settings.snap_x_tolerance,
snap_y_tolerance=settings.snap_y_tolerance,
join_x_tolerance=settings.join_x_tolerance,
join_y_tolerance=settings.join_y_tolerance,
)
return utils.filter_edges(edges, min_length=settings.edge_min_length)
@@ -0,0 +1,43 @@
from .clustering import cluster_list, cluster_objects, make_cluster_dict # noqa: F401
from .generic import to_list # noqa: F401
from .geometry import ( # noqa: F401
bbox_to_rect,
calculate_area,
clip_obj,
crop_to_bbox,
curve_to_edges,
filter_edges,
get_bbox_overlap,
intersects_bbox,
line_to_edge,
merge_bboxes,
move_object,
obj_to_bbox,
obj_to_edges,
objects_to_bbox,
objects_to_rect,
outside_bbox,
rect_to_edges,
resize_object,
snap_objects,
within_bbox,
)
from .pdfinternals import ( # noqa: F401
decode_psl_list,
decode_text,
resolve,
resolve_all,
resolve_and_decode,
)
from .text import ( # noqa: F401
DEFAULT_X_DENSITY,
DEFAULT_X_TOLERANCE,
DEFAULT_Y_DENSITY,
DEFAULT_Y_TOLERANCE,
chars_to_textmap,
collate_line,
dedupe_chars,
extract_text,
extract_text_simple,
extract_words,
)
@@ -0,0 +1,66 @@
import itertools
from collections.abc import Hashable
from operator import itemgetter
from typing import Any, Callable, Dict, Iterable, List, Tuple, TypeVar, Union
from .._typing import T_num, T_obj
def cluster_list(xs: List[T_num], tolerance: T_num = 0) -> List[List[T_num]]:
if tolerance == 0:
return [[x] for x in sorted(xs)]
if len(xs) < 2:
return [[x] for x in sorted(xs)]
groups = []
xs = list(sorted(xs))
current_group = [xs[0]]
last = xs[0]
for x in xs[1:]:
if x <= (last + tolerance):
current_group.append(x)
else:
groups.append(current_group)
current_group = [x]
last = x
groups.append(current_group)
return groups
def make_cluster_dict(values: Iterable[T_num], tolerance: T_num) -> Dict[T_num, int]:
clusters = cluster_list(list(set(values)), tolerance)
nested_tuples = [
[(val, i) for val in value_cluster] for i, value_cluster in enumerate(clusters)
]
return dict(itertools.chain(*nested_tuples))
Clusterable = TypeVar("Clusterable", T_obj, Tuple[Any, ...])
def cluster_objects(
xs: List[Clusterable],
key_fn: Union[Hashable, Callable[[Clusterable], T_num]],
tolerance: T_num,
preserve_order: bool = False,
) -> List[List[Clusterable]]:
if not callable(key_fn):
key_fn = itemgetter(key_fn)
values = map(key_fn, xs)
cluster_dict = make_cluster_dict(values, tolerance)
get_0, get_1 = itemgetter(0), itemgetter(1)
if preserve_order:
cluster_tuples = [(x, cluster_dict.get(key_fn(x))) for x in xs]
else:
cluster_tuples = sorted(
((x, cluster_dict.get(key_fn(x))) for x in xs), key=get_1
)
grouped = itertools.groupby(cluster_tuples, key=get_1)
return [list(map(get_0, v)) for k, v in grouped]
@@ -0,0 +1,6 @@
class MalformedPDFException(Exception):
pass
class PdfminerException(Exception):
pass
@@ -0,0 +1,21 @@
from collections.abc import Sequence
from typing import TYPE_CHECKING, Any, Dict, Hashable, List, Union
from .._typing import T_seq
if TYPE_CHECKING: # pragma: nocover
from pandas.core.frame import DataFrame
def to_list(collection: Union[T_seq[Any], "DataFrame"]) -> List[Any]:
if isinstance(collection, list):
return collection
elif isinstance(collection, Sequence):
return list(collection)
elif hasattr(collection, "to_dict"):
res: List[Dict[Hashable, Any]] = collection.to_dict(
"records"
) # pragma: nocover
return res
else:
return list(collection)
@@ -0,0 +1,279 @@
import itertools
from operator import itemgetter
from typing import Dict, Iterable, Optional
from .._typing import T_bbox, T_num, T_obj, T_obj_list
from .clustering import cluster_objects
def objects_to_rect(objects: Iterable[T_obj]) -> Dict[str, T_num]:
"""
Given an iterable of objects, return the smallest rectangle (i.e. a
dict with "x0", "top", "x1", and "bottom" keys) that contains them
all.
"""
return bbox_to_rect(objects_to_bbox(objects))
def objects_to_bbox(objects: Iterable[T_obj]) -> T_bbox:
"""
Given an iterable of objects, return the smallest bounding box that
contains them all.
"""
return merge_bboxes(map(bbox_getter, objects))
bbox_getter = itemgetter("x0", "top", "x1", "bottom")
def obj_to_bbox(obj: T_obj) -> T_bbox:
"""
Return the bounding box for an object.
"""
bbox: T_bbox = bbox_getter(obj)
return bbox
def bbox_to_rect(bbox: T_bbox) -> Dict[str, T_num]:
"""
Return the rectangle (i.e a dict with keys "x0", "top", "x1",
"bottom") for an object.
"""
return {"x0": bbox[0], "top": bbox[1], "x1": bbox[2], "bottom": bbox[3]}
def merge_bboxes(bboxes: Iterable[T_bbox]) -> T_bbox:
"""
Given an iterable of bounding boxes, return the smallest bounding box
that contains them all.
"""
x0, top, x1, bottom = zip(*bboxes)
return (min(x0), min(top), max(x1), max(bottom))
def get_bbox_overlap(a: T_bbox, b: T_bbox) -> Optional[T_bbox]:
a_left, a_top, a_right, a_bottom = a
b_left, b_top, b_right, b_bottom = b
o_left = max(a_left, b_left)
o_right = min(a_right, b_right)
o_bottom = min(a_bottom, b_bottom)
o_top = max(a_top, b_top)
o_width = o_right - o_left
o_height = o_bottom - o_top
if o_height >= 0 and o_width >= 0 and o_height + o_width > 0:
return (o_left, o_top, o_right, o_bottom)
else:
return None
def calculate_area(bbox: T_bbox) -> T_num:
left, top, right, bottom = bbox
if left > right or top > bottom:
raise ValueError(f"{bbox} has a negative width or height.")
return (right - left) * (bottom - top)
def clip_obj(obj: T_obj, bbox: T_bbox) -> Optional[T_obj]:
overlap = get_bbox_overlap(obj_to_bbox(obj), bbox)
if overlap is None:
return None
dims = bbox_to_rect(overlap)
copy = dict(obj)
for attr in ["x0", "top", "x1", "bottom"]:
copy[attr] = dims[attr]
diff = dims["top"] - obj["top"]
if "doctop" in copy:
copy["doctop"] = obj["doctop"] + diff
copy["width"] = copy["x1"] - copy["x0"]
copy["height"] = copy["bottom"] - copy["top"]
return copy
def intersects_bbox(objs: Iterable[T_obj], bbox: T_bbox) -> T_obj_list:
"""
Filters objs to only those intersecting the bbox
"""
return [obj for obj in objs if get_bbox_overlap(obj_to_bbox(obj), bbox) is not None]
def within_bbox(objs: Iterable[T_obj], bbox: T_bbox) -> T_obj_list:
"""
Filters objs to only those fully within the bbox
"""
return [
obj
for obj in objs
if get_bbox_overlap(obj_to_bbox(obj), bbox) == obj_to_bbox(obj)
]
def outside_bbox(objs: Iterable[T_obj], bbox: T_bbox) -> T_obj_list:
"""
Filters objs to only those fully outside the bbox
"""
return [obj for obj in objs if get_bbox_overlap(obj_to_bbox(obj), bbox) is None]
def crop_to_bbox(objs: Iterable[T_obj], bbox: T_bbox) -> T_obj_list:
"""
Filters objs to only those intersecting the bbox,
and crops the extent of the objects to the bbox.
"""
return list(filter(None, (clip_obj(obj, bbox) for obj in objs)))
def move_object(obj: T_obj, axis: str, value: T_num) -> T_obj:
assert axis in ("h", "v")
if axis == "h":
new_items = [
("x0", obj["x0"] + value),
("x1", obj["x1"] + value),
]
if axis == "v":
new_items = [
("top", obj["top"] + value),
("bottom", obj["bottom"] + value),
]
if "doctop" in obj:
new_items += [("doctop", obj["doctop"] + value)]
if "y0" in obj:
new_items += [
("y0", obj["y0"] - value),
("y1", obj["y1"] - value),
]
return obj.__class__(tuple(obj.items()) + tuple(new_items))
def snap_objects(objs: Iterable[T_obj], attr: str, tolerance: T_num) -> T_obj_list:
axis = {"x0": "h", "x1": "h", "top": "v", "bottom": "v"}[attr]
list_objs = list(objs)
clusters = cluster_objects(list_objs, itemgetter(attr), tolerance)
avgs = [sum(map(itemgetter(attr), cluster)) / len(cluster) for cluster in clusters]
snapped_clusters = [
[move_object(obj, axis, avg - obj[attr]) for obj in cluster]
for cluster, avg in zip(clusters, avgs)
]
return list(itertools.chain(*snapped_clusters))
def resize_object(obj: T_obj, key: str, value: T_num) -> T_obj:
assert key in ("x0", "x1", "top", "bottom")
old_value = obj[key]
diff = value - old_value
new_items = [
(key, value),
]
if key == "x0":
assert value <= obj["x1"]
new_items.append(("width", obj["x1"] - value))
elif key == "x1":
assert value >= obj["x0"]
new_items.append(("width", value - obj["x0"]))
elif key == "top":
assert value <= obj["bottom"]
new_items.append(("doctop", obj["doctop"] + diff))
new_items.append(("height", obj["height"] - diff))
if "y1" in obj:
new_items.append(("y1", obj["y1"] - diff))
elif key == "bottom":
assert value >= obj["top"]
new_items.append(("height", obj["height"] + diff))
if "y0" in obj:
new_items.append(("y0", obj["y0"] - diff))
return obj.__class__(tuple(obj.items()) + tuple(new_items))
def curve_to_edges(curve: T_obj) -> T_obj_list:
point_pairs = zip(curve["pts"], curve["pts"][1:])
return [
{
"object_type": "curve_edge",
"x0": min(p0[0], p1[0]),
"x1": max(p0[0], p1[0]),
"top": min(p0[1], p1[1]),
"doctop": min(p0[1], p1[1]) + (curve["doctop"] - curve["top"]),
"bottom": max(p0[1], p1[1]),
"width": abs(p0[0] - p1[0]),
"height": abs(p0[1] - p1[1]),
"orientation": "v" if p0[0] == p1[0] else ("h" if p0[1] == p1[1] else None),
}
for p0, p1 in point_pairs
]
def rect_to_edges(rect: T_obj) -> T_obj_list:
top, bottom, left, right = [dict(rect) for x in range(4)]
top.update(
{
"object_type": "rect_edge",
"height": 0,
"y0": rect["y1"],
"bottom": rect["top"],
"orientation": "h",
}
)
bottom.update(
{
"object_type": "rect_edge",
"height": 0,
"y1": rect["y0"],
"top": rect["top"] + rect["height"],
"doctop": rect["doctop"] + rect["height"],
"orientation": "h",
}
)
left.update(
{
"object_type": "rect_edge",
"width": 0,
"x1": rect["x0"],
"orientation": "v",
}
)
right.update(
{
"object_type": "rect_edge",
"width": 0,
"x0": rect["x1"],
"orientation": "v",
}
)
return [top, bottom, left, right]
def line_to_edge(line: T_obj) -> T_obj:
edge = dict(line)
edge["orientation"] = "h" if (line["top"] == line["bottom"]) else "v"
return edge
def obj_to_edges(obj: T_obj) -> T_obj_list:
t = obj["object_type"]
if "_edge" in t:
return [obj]
elif t == "line":
return [line_to_edge(obj)]
else:
return {"rect": rect_to_edges, "curve": curve_to_edges}[t](obj)
def filter_edges(
edges: Iterable[T_obj],
orientation: Optional[str] = None,
edge_type: Optional[str] = None,
min_length: T_num = 1,
) -> T_obj_list:
if orientation not in ("v", "h", None):
raise ValueError("Orientation must be 'v' or 'h'")
def test(e: T_obj) -> bool:
dim = "height" if e["orientation"] == "v" else "width"
et_correct = e["object_type"] == edge_type if edge_type is not None else True
orient_correct = orientation is None or e["orientation"] == orientation
return bool(et_correct and orient_correct and (e[dim] >= min_length))
return list(filter(test, edges))
@@ -0,0 +1,87 @@
from typing import Any, List, Optional, Union
from pdfminer.pdftypes import PDFObjRef
from pdfminer.psparser import PSLiteral
from pdfminer.utils import PDFDocEncoding
from .exceptions import MalformedPDFException
def decode_text(s: Union[bytes, str]) -> str:
"""
Decodes a PDFDocEncoding string to Unicode.
Adds py3 compatibility to pdfminer's version.
"""
if isinstance(s, bytes) and s.startswith(b"\xfe\xff"):
return str(s[2:], "utf-16be", "ignore")
try:
ords = (ord(c) if isinstance(c, str) else c for c in s)
return "".join(PDFDocEncoding[o] for o in ords)
except IndexError:
return str(s)
def resolve_and_decode(obj: Any) -> Any:
"""Recursively resolve the metadata values."""
if hasattr(obj, "resolve"):
obj = obj.resolve()
if isinstance(obj, list):
return list(map(resolve_and_decode, obj))
elif isinstance(obj, PSLiteral):
return decode_text(obj.name)
elif isinstance(obj, (str, bytes)):
return decode_text(obj)
elif isinstance(obj, dict):
for k, v in obj.items():
obj[k] = resolve_and_decode(v)
return obj
return obj
def decode_psl_list(_list: List[Union[PSLiteral, str]]) -> List[str]:
return [
decode_text(value.name) if isinstance(value, PSLiteral) else value
for value in _list
]
def resolve(x: Any) -> Any:
if isinstance(x, PDFObjRef):
return x.resolve()
else:
return x
def get_dict_type(d: Any) -> Optional[str]:
if not isinstance(d, dict):
return None
t = d.get("Type")
if isinstance(t, PSLiteral):
return decode_text(t.name)
else:
return t
def resolve_all(x: Any) -> Any:
"""
Recursively resolves the given object and all the internals.
"""
if isinstance(x, PDFObjRef):
resolved = x.resolve()
# Avoid infinite recursion
if get_dict_type(resolved) == "Page":
return x
try:
return resolve_all(resolved)
except RecursionError as e:
raise MalformedPDFException(e)
elif isinstance(x, (list, tuple)):
return type(x)(resolve_all(v) for v in x)
elif isinstance(x, dict):
exceptions = ["Parent"] if get_dict_type(x) == "Annot" else []
return {k: v if k in exceptions else resolve_all(v) for k, v in x.items()}
else:
return x
@@ -0,0 +1,819 @@
import inspect
import itertools
import logging
import re
import string
from operator import itemgetter
from typing import (
Any,
Callable,
Dict,
Generator,
List,
Match,
Optional,
Pattern,
Tuple,
Union,
)
from .._typing import T_bbox, T_dir, T_num, T_obj, T_obj_iter, T_obj_list
from .clustering import cluster_objects
from .generic import to_list
from .geometry import objects_to_bbox
logger = logging.getLogger(__name__)
DEFAULT_X_TOLERANCE = 3
DEFAULT_Y_TOLERANCE = 3
DEFAULT_X_DENSITY = 7.25
DEFAULT_Y_DENSITY = 13
DEFAULT_LINE_DIR: T_dir = "ttb"
DEFAULT_CHAR_DIR: T_dir = "ltr"
LIGATURES = {
"ff": "ff",
"ffi": "ffi",
"ffl": "ffl",
"fi": "fi",
"fl": "fl",
"st": "st",
"ſt": "st",
}
def get_line_cluster_key(line_dir: T_dir) -> Callable[[T_obj], T_num]:
return {
"ttb": lambda x: x["top"],
"btt": lambda x: -x["bottom"],
"ltr": lambda x: x["x0"],
"rtl": lambda x: -x["x1"],
}[line_dir]
def get_char_sort_key(char_dir: T_dir) -> Callable[[T_obj], Tuple[T_num, T_num]]:
return {
"ttb": lambda x: (x["top"], x["bottom"]),
"btt": lambda x: (-(x["top"] + x["height"]), -x["top"]),
"ltr": lambda x: (x["x0"], x["x0"]),
"rtl": lambda x: (-x["x1"], -x["x0"]),
}[char_dir]
BBOX_ORIGIN_KEYS = {
"ttb": itemgetter(1),
"btt": itemgetter(3),
"ltr": itemgetter(0),
"rtl": itemgetter(2),
}
POSITION_KEYS = {
"ttb": itemgetter("top"),
"btt": itemgetter("bottom"),
"ltr": itemgetter("x0"),
"rtl": itemgetter("x1"),
}
def validate_directions(line_dir: T_dir, char_dir: T_dir, suffix: str = "") -> None:
valid_dirs = set(POSITION_KEYS.keys())
if line_dir not in valid_dirs:
raise ValueError(
f"line_dir{suffix} must be one of {valid_dirs}, not {line_dir}"
)
if char_dir not in valid_dirs:
raise ValueError(
f"char_dir{suffix} must be one of {valid_dirs}, not {char_dir}"
)
if set(line_dir) == set(char_dir):
raise ValueError(
f"line_dir{suffix}={line_dir} is incompatible "
f"with char_dir{suffix}={char_dir}"
)
class TextMap:
"""
A TextMap maps each unicode character in the text to an individual `char`
object (or, in the case of layout-implied whitespace, `None`).
"""
def __init__(
self,
tuples: List[Tuple[str, Optional[T_obj]]],
line_dir_render: T_dir,
char_dir_render: T_dir,
) -> None:
validate_directions(line_dir_render, char_dir_render, "_render")
self.tuples = tuples
self.line_dir_render = line_dir_render
self.char_dir_render = char_dir_render
self.as_string = self.to_string()
def to_string(self) -> str:
cd = self.char_dir_render
ld = self.line_dir_render
base = "".join(map(itemgetter(0), self.tuples))
if cd == "ltr" and ld == "ttb":
return base
else:
lines = base.split("\n")
if ld in ("btt", "rtl"):
lines = list(reversed(lines))
if cd == "rtl":
lines = ["".join(reversed(line)) for line in lines]
if ld in ("rtl", "ltr"):
max_line_length = max(map(len, lines))
if cd == "btt":
lines = [
(" " * (max_line_length - len(line))) + line for line in lines
]
else:
lines = [
line + (" " * (max_line_length - len(line))) for line in lines
]
return "\n".join(
"".join(line[i] for line in lines) for i in range(max_line_length)
)
else:
return "\n".join(lines)
def match_to_dict(
self,
m: Match[str],
main_group: int = 0,
return_groups: bool = True,
return_chars: bool = True,
) -> Dict[str, Any]:
subset = self.tuples[m.start(main_group) : m.end(main_group)]
chars = [c for (text, c) in subset if c is not None]
x0, top, x1, bottom = objects_to_bbox(chars)
result = {
"text": m.group(main_group),
"x0": x0,
"top": top,
"x1": x1,
"bottom": bottom,
}
if return_groups:
result["groups"] = m.groups()
if return_chars:
result["chars"] = chars
return result
def search(
self,
pattern: Union[str, Pattern[str]],
regex: bool = True,
case: bool = True,
return_groups: bool = True,
return_chars: bool = True,
main_group: int = 0,
) -> List[Dict[str, Any]]:
if isinstance(pattern, Pattern):
if regex is False:
raise ValueError(
"Cannot pass a compiled search pattern *and* regex=False together."
)
if case is False:
raise ValueError(
"Cannot pass a compiled search pattern *and* case=False together."
)
compiled = pattern
else:
if regex is False:
pattern = re.escape(pattern)
flags = re.I if case is False else 0
compiled = re.compile(pattern, flags)
gen = re.finditer(compiled, self.as_string)
# Remove zero-length matches (can happen, e.g., with optional
# patterns in regexes) and whitespace-only matches
filtered = filter(lambda m: bool(m.group(main_group).strip()), gen)
return [
self.match_to_dict(
m,
return_groups=return_groups,
return_chars=return_chars,
main_group=main_group,
)
for m in filtered
]
def extract_text_lines(
self, strip: bool = True, return_chars: bool = True
) -> List[Dict[str, Any]]:
"""
`strip` is analogous to Python's `str.strip()` method, and returns
`text` attributes without their surrounding whitespace. Only
relevant when the relevant TextMap is created with `layout` = True
Setting `return_chars` to False will exclude the individual
character objects from the returned text-line dicts.
"""
if strip:
pat = r" *([^\n]+?) *(\n|$)"
else:
pat = r"([^\n]+)"
return self.search(
pat, main_group=1, return_chars=return_chars, return_groups=False
)
class WordMap:
"""
A WordMap maps words->chars.
"""
def __init__(self, tuples: List[Tuple[T_obj, T_obj_list]]) -> None:
self.tuples = tuples
def to_textmap(
self,
layout: bool = False,
layout_width: T_num = 0,
layout_height: T_num = 0,
layout_width_chars: int = 0,
layout_height_chars: int = 0,
layout_bbox: T_bbox = (0, 0, 0, 0),
x_density: T_num = DEFAULT_X_DENSITY,
y_density: T_num = DEFAULT_Y_DENSITY,
x_shift: T_num = 0,
y_shift: T_num = 0,
y_tolerance: T_num = DEFAULT_Y_TOLERANCE,
line_dir: T_dir = DEFAULT_LINE_DIR,
char_dir: T_dir = DEFAULT_CHAR_DIR,
line_dir_rotated: Optional[T_dir] = None,
char_dir_rotated: Optional[T_dir] = None,
char_dir_render: Optional[T_dir] = None,
line_dir_render: Optional[T_dir] = None,
use_text_flow: bool = False,
presorted: bool = False,
expand_ligatures: bool = True,
) -> TextMap:
"""
Given a list of (word, chars) tuples (i.e., a WordMap), return a list of
(char-text, char) tuples (i.e., a TextMap) that can be used to mimic
the structural layout of the text on the page(s), using the following
approach for top-to-bottom, left-to-right text:
- Sort the words by (top, x0) if not already sorted.
- Cluster the words by top (taking `y_tolerance` into account), and
iterate through them.
- For each cluster, divide (top - y_shift) by `y_density` to calculate
the minimum number of newlines that should come before this cluster.
Append that number of newlines *minus* the number of newlines already
appended, with a minimum of one.
- Then for each cluster, iterate through each word in it. Divide each
word's x0, minus `x_shift`, by `x_density` to calculate the minimum
number of characters that should come before this cluster. Append that
number of spaces *minus* the number of characters and spaces already
appended, with a minimum of one. Then append the word's text.
- At the termination of each line, add more spaces if necessary to
mimic `layout_width`.
- Finally, add newlines to the end if necessary to mimic to
`layout_height`.
For other line/character directions (e.g., bottom-to-top,
right-to-left), these steps are adjusted.
"""
_textmap: List[Tuple[str, Optional[T_obj]]] = []
if not len(self.tuples):
return TextMap(
_textmap,
line_dir_render=line_dir_render or line_dir,
char_dir_render=char_dir_render or char_dir,
)
expansions = LIGATURES if expand_ligatures else {}
if layout:
if layout_width_chars:
if layout_width:
raise ValueError(
"`layout_width` and `layout_width_chars` cannot both be set."
)
else:
layout_width_chars = int(round(layout_width / x_density))
if layout_height_chars:
if layout_height:
raise ValueError(
"`layout_height` and `layout_height_chars` cannot both be set."
)
else:
layout_height_chars = int(round(layout_height / y_density))
blank_line = [(" ", None)] * layout_width_chars
else:
blank_line = []
num_newlines = 0
line_cluster_key = get_line_cluster_key(line_dir)
char_sort_key = get_char_sort_key(char_dir)
line_position_key = POSITION_KEYS[line_dir]
char_position_key = POSITION_KEYS[char_dir]
y_origin = BBOX_ORIGIN_KEYS[line_dir](layout_bbox)
x_origin = BBOX_ORIGIN_KEYS[char_dir](layout_bbox)
words_sorted_line_dir = (
self.tuples
if presorted or use_text_flow
else sorted(self.tuples, key=lambda x: line_cluster_key(x[0]))
)
tuples_by_line = cluster_objects(
words_sorted_line_dir,
lambda x: line_cluster_key(x[0]),
y_tolerance,
preserve_order=presorted or use_text_flow,
)
for i, line_tuples in enumerate(tuples_by_line):
if layout:
line_position = line_position_key(line_tuples[0][0])
y_dist_raw = line_position - (y_origin + y_shift)
adj = -1 if line_dir in ["btt", "rtl"] else 1
y_dist = y_dist_raw * adj / y_density
else:
y_dist = 0
num_newlines_prepend = max(
# At least one newline, unless this iis the first line
int(i > 0),
# ... or as many as needed to get the imputed "distance" from the top
round(y_dist) - num_newlines,
)
for i in range(num_newlines_prepend):
if not len(_textmap) or _textmap[-1][0] == "\n":
_textmap += blank_line
_textmap.append(("\n", None))
num_newlines += num_newlines_prepend
line_len = 0
line_tuples_sorted = (
line_tuples
if presorted or use_text_flow
else sorted(line_tuples, key=lambda x: char_sort_key(x[0]))
)
for word, chars in line_tuples_sorted:
if layout:
char_position = char_position_key(word)
x_dist_raw = char_position - (x_origin + x_shift)
adj = -1 if char_dir in ["btt", "rtl"] else 1
x_dist = x_dist_raw * adj / x_density
else:
x_dist = 0
num_spaces_prepend = max(min(1, line_len), round(x_dist) - line_len)
_textmap += [(" ", None)] * num_spaces_prepend
line_len += num_spaces_prepend
for c in chars:
letters = expansions.get(c["text"], c["text"])
for letter in letters:
_textmap.append((letter, c))
line_len += 1
# Append spaces at end of line
if layout:
_textmap += [(" ", None)] * (layout_width_chars - line_len)
# Append blank lines at end of text
if layout:
num_newlines_append = layout_height_chars - (num_newlines + 1)
for i in range(num_newlines_append):
if i > 0:
_textmap += blank_line
_textmap.append(("\n", None))
# Remove terminal newline
if _textmap[-1] == ("\n", None):
_textmap = _textmap[:-1]
return TextMap(
_textmap,
line_dir_render=line_dir_render or line_dir,
char_dir_render=char_dir_render or char_dir,
)
class WordExtractor:
def __init__(
self,
x_tolerance: T_num = DEFAULT_X_TOLERANCE,
y_tolerance: T_num = DEFAULT_Y_TOLERANCE,
x_tolerance_ratio: Union[int, float, None] = None,
y_tolerance_ratio: Union[int, float, None] = None,
keep_blank_chars: bool = False,
use_text_flow: bool = False,
vertical_ttb: bool = True, # Should vertical words be read top-to-bottom?
horizontal_ltr: bool = True, # Should words be read left-to-right?
line_dir: T_dir = DEFAULT_LINE_DIR,
char_dir: T_dir = DEFAULT_CHAR_DIR,
line_dir_rotated: Optional[T_dir] = None,
char_dir_rotated: Optional[T_dir] = None,
extra_attrs: Optional[List[str]] = None,
split_at_punctuation: Union[bool, str] = False,
expand_ligatures: bool = True,
):
self.x_tolerance = x_tolerance
self.y_tolerance = y_tolerance
self.x_tolerance_ratio = x_tolerance_ratio
self.y_tolerance_ratio = y_tolerance_ratio
self.keep_blank_chars = keep_blank_chars
self.use_text_flow = use_text_flow
self.horizontal_ltr = horizontal_ltr
self.vertical_ttb = vertical_ttb
if vertical_ttb is False:
logger.warning(
"vertical_ttb is deprecated and will be removed;"
" use line_dir/char_dir instead."
)
if horizontal_ltr is False:
logger.warning(
"horizontal_ltr is deprecated and will be removed;"
" use line_dir/char_dir instead."
)
self.line_dir = line_dir
self.char_dir = char_dir
# Default is to "flip" the directions for rotated text
self.line_dir_rotated = line_dir_rotated or char_dir
self.char_dir_rotated = char_dir_rotated or line_dir
validate_directions(self.line_dir, self.char_dir)
validate_directions(self.line_dir_rotated, self.char_dir_rotated, "_rotated")
self.extra_attrs = [] if extra_attrs is None else extra_attrs
# Note: string.punctuation = '!"#$%&\'()*+,-./:;<=>?@[\\]^_`{|}~'
self.split_at_punctuation = (
string.punctuation
if split_at_punctuation is True
else (split_at_punctuation or "")
)
self.expansions = LIGATURES if expand_ligatures else {}
def get_char_dir(self, upright: int) -> T_dir:
# Note: This can be simplified and reincorporated into .merge_chars and
# .iter_chars_to_lines once .vertical_ttb and .horizontal_ltr
# deprecation is complete.
if not upright and not self.vertical_ttb:
return "btt"
elif upright and not self.horizontal_ltr:
return "rtl"
return self.char_dir if upright else self.char_dir_rotated
def merge_chars(self, ordered_chars: T_obj_list) -> T_obj:
x0, top, x1, bottom = objects_to_bbox(ordered_chars)
doctop_adj = ordered_chars[0]["doctop"] - ordered_chars[0]["top"]
upright = ordered_chars[0]["upright"]
char_dir = self.get_char_dir(upright)
word = {
"text": "".join(
self.expansions.get(c["text"], c["text"]) for c in ordered_chars
),
"x0": x0,
"x1": x1,
"top": top,
"doctop": top + doctop_adj,
"bottom": bottom,
"upright": upright,
"height": bottom - top,
"width": x1 - x0,
"direction": char_dir,
}
for key in self.extra_attrs:
word[key] = ordered_chars[0][key]
return word
def char_begins_new_word(
self,
prev_char: T_obj,
curr_char: T_obj,
direction: T_dir,
x_tolerance: T_num,
y_tolerance: T_num,
) -> bool:
"""This method takes several factors into account to determine if
`curr_char` represents the beginning of a new word:
- Whether the text is "upright" (i.e., non-rotated)
- Whether the user has specified that horizontal text runs
left-to-right (default) or right-to-left, as represented by
self.horizontal_ltr
- Whether the user has specified that vertical text the text runs
top-to-bottom (default) or bottom-to-top, as represented by
self.vertical_ttb
- The x0, top, x1, and bottom attributes of prev_char and
curr_char
- The self.x_tolerance and self.y_tolerance settings. Note: In
this case, x/y refer to those directions for non-rotated text.
For vertical text, they are flipped. A more accurate terminology
might be "*intra*line character distance tolerance" and
"*inter*line character distance tolerance"
An important note: The *intra*line distance is measured from the
*end* of the previous character to the *beginning* of the current
character, while the *inter*line distance is measured from the
*top* of the previous character to the *top* of the next
character. The reasons for this are partly repository-historical,
and partly logical, as successive text lines' bounding boxes often
overlap slightly (and we don't want that overlap to be interpreted
as the two lines being the same line).
The upright-ness of the character determines the attributes to
compare, while horizontal_ltr/vertical_ttb determine the direction
of the comparison.
"""
# Note: Due to the grouping step earlier in the process,
# curr_char["upright"] will always equal prev_char["upright"].
if direction in ("ltr", "rtl"):
x = x_tolerance
y = y_tolerance
ay = prev_char["top"]
cy = curr_char["top"]
if direction == "ltr":
ax = prev_char["x0"]
bx = prev_char["x1"]
cx = curr_char["x0"]
else:
ax = -prev_char["x1"]
bx = -prev_char["x0"]
cx = -curr_char["x1"]
else:
x = y_tolerance
y = x_tolerance
ay = prev_char["x0"]
cy = curr_char["x0"]
if direction == "ttb":
ax = prev_char["top"]
bx = prev_char["bottom"]
cx = curr_char["top"]
else:
ax = -prev_char["bottom"]
bx = -prev_char["top"]
cx = -curr_char["bottom"]
return bool(
# Intraline test
(cx < ax)
or (cx > bx + x)
# Interline test
or abs(cy - ay) > y
)
def iter_chars_to_words(
self,
ordered_chars: T_obj_iter,
direction: T_dir,
) -> Generator[T_obj_list, None, None]:
current_word: T_obj_list = []
def start_next_word(
new_char: Optional[T_obj],
) -> Generator[T_obj_list, None, None]:
nonlocal current_word
if current_word:
yield current_word
current_word = [] if new_char is None else [new_char]
xt = self.x_tolerance
xtr = self.x_tolerance_ratio
yt = self.y_tolerance
ytr = self.y_tolerance_ratio
for char in ordered_chars:
text = char["text"]
if not self.keep_blank_chars and text.isspace():
yield from start_next_word(None)
elif text in self.split_at_punctuation:
yield from start_next_word(char)
yield from start_next_word(None)
elif current_word and self.char_begins_new_word(
current_word[-1],
char,
direction,
x_tolerance=(xt if xtr is None else xtr * current_word[-1]["size"]),
y_tolerance=(yt if ytr is None else ytr * current_word[-1]["size"]),
):
yield from start_next_word(char)
else:
current_word.append(char)
# Finally, after all chars processed
if current_word:
yield current_word
def iter_chars_to_lines(
self, chars: T_obj_iter
) -> Generator[Tuple[T_obj_list, T_dir], None, None]:
chars = list(chars)
upright = chars[0]["upright"]
line_dir = self.line_dir if upright else self.line_dir_rotated
char_dir = self.get_char_dir(upright)
line_cluster_key = get_line_cluster_key(line_dir)
char_sort_key = get_char_sort_key(char_dir)
# Cluster by line
subclusters = cluster_objects(
chars,
line_cluster_key,
(self.y_tolerance if line_dir in ("ttb", "btt") else self.x_tolerance),
)
for sc in subclusters:
# Sort within line
chars_sorted = sorted(sc, key=char_sort_key)
yield (chars_sorted, char_dir)
def iter_extract_tuples(
self, chars: T_obj_iter
) -> Generator[Tuple[T_obj, T_obj_list], None, None]:
grouping_key = itemgetter("upright", *self.extra_attrs)
grouped_chars = itertools.groupby(chars, grouping_key)
for keyvals, char_group in grouped_chars:
line_groups = (
[(char_group, self.char_dir)]
if self.use_text_flow
else self.iter_chars_to_lines(char_group)
)
for line_chars, direction in line_groups:
for word_chars in self.iter_chars_to_words(line_chars, direction):
yield (self.merge_chars(word_chars), word_chars)
def extract_wordmap(self, chars: T_obj_iter) -> WordMap:
return WordMap(list(self.iter_extract_tuples(chars)))
def extract_words(
self, chars: T_obj_list, return_chars: bool = False
) -> T_obj_list:
if return_chars:
return list(
{**word, "chars": word_chars}
for word, word_chars in self.iter_extract_tuples(chars)
)
else:
return list(word for word, word_chars in self.iter_extract_tuples(chars))
def extract_words(
chars: T_obj_list, return_chars: bool = False, **kwargs: Any
) -> T_obj_list:
return WordExtractor(**kwargs).extract_words(chars, return_chars)
TEXTMAP_KWARGS = inspect.signature(WordMap.to_textmap).parameters.keys()
WORD_EXTRACTOR_KWARGS = inspect.signature(WordExtractor).parameters.keys()
def chars_to_textmap(chars: T_obj_list, **kwargs: Any) -> TextMap:
kwargs.update(
{
"presorted": True,
"layout_bbox": kwargs.get("layout_bbox") or objects_to_bbox(chars),
}
)
extractor = WordExtractor(
**{k: kwargs[k] for k in WORD_EXTRACTOR_KWARGS if k in kwargs}
)
wordmap = extractor.extract_wordmap(chars)
textmap = wordmap.to_textmap(
**{k: kwargs[k] for k in TEXTMAP_KWARGS if k in kwargs}
)
return textmap
def extract_text(
chars: T_obj_list,
line_dir_render: Optional[T_dir] = None,
char_dir_render: Optional[T_dir] = None,
**kwargs: Any,
) -> str:
chars = to_list(chars)
if len(chars) == 0:
return ""
if kwargs.get("layout"):
textmap_kwargs = {
**kwargs,
**{"line_dir_render": line_dir_render, "char_dir_render": char_dir_render},
}
return chars_to_textmap(chars, **textmap_kwargs).as_string
else:
extractor = WordExtractor(
**{k: kwargs[k] for k in WORD_EXTRACTOR_KWARGS if k in kwargs}
)
words = extractor.extract_words(chars)
line_dir_render = line_dir_render or extractor.line_dir
char_dir_render = char_dir_render or extractor.char_dir
line_cluster_key = get_line_cluster_key(extractor.line_dir)
x_tolerance = kwargs.get("x_tolerance", DEFAULT_X_TOLERANCE)
y_tolerance = kwargs.get("y_tolerance", DEFAULT_Y_TOLERANCE)
lines = cluster_objects(
words,
line_cluster_key,
y_tolerance if line_dir_render in ("ttb", "btt") else x_tolerance,
)
return TextMap(
[
(char, None)
for char in (
"\n".join(" ".join(word["text"] for word in line) for line in lines)
)
],
line_dir_render=line_dir_render,
char_dir_render=char_dir_render,
).as_string
def collate_line(
line_chars: T_obj_list,
tolerance: T_num = DEFAULT_X_TOLERANCE,
) -> str:
coll = ""
last_x1 = None
for char in sorted(line_chars, key=itemgetter("x0")):
if (last_x1 is not None) and (char["x0"] > (last_x1 + tolerance)):
coll += " "
last_x1 = char["x1"]
coll += char["text"]
return coll
def extract_text_simple(
chars: T_obj_list,
x_tolerance: T_num = DEFAULT_X_TOLERANCE,
y_tolerance: T_num = DEFAULT_Y_TOLERANCE,
) -> str:
clustered = cluster_objects(chars, itemgetter("doctop"), y_tolerance)
return "\n".join(collate_line(c, x_tolerance) for c in clustered)
def dedupe_chars(
chars: T_obj_list,
tolerance: T_num = 1,
extra_attrs: Optional[Tuple[str, ...]] = ("fontname", "size"),
) -> T_obj_list:
"""
Removes duplicate chars — those sharing the same text and positioning
(within `tolerance`) as other characters in the set. Use extra_args to
be more restrictive with the properties shared by the matching chars.
"""
key = itemgetter(*("upright", "text"), *(extra_attrs or tuple()))
pos_key = itemgetter("doctop", "x0")
def yield_unique_chars(chars: T_obj_list) -> Generator[T_obj, None, None]:
sorted_chars = sorted(chars, key=key)
for grp, grp_chars in itertools.groupby(sorted_chars, key=key):
for y_cluster in cluster_objects(
list(grp_chars), itemgetter("doctop"), tolerance
):
for x_cluster in cluster_objects(
y_cluster, itemgetter("x0"), tolerance
):
yield sorted(x_cluster, key=pos_key)[0]
deduped = yield_unique_chars(chars)
return sorted(deduped, key=chars.index)