Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Extract PDF text preserving the layout of the source PDF"""
3from collections.abc import Iterator
4from itertools import groupby
5from math import ceil
6from pathlib import Path
7from typing import Any, Literal, Optional, TypedDict
9from ..._font import Font
10from ..._utils import logger_warning
11from .. import LAYOUT_NEW_BT_GROUP_SPACE_WIDTHS
12from ._text_state_manager import TextStateManager
13from ._text_state_params import TextStateParams
15WHITESPACE_LIMIT = 10_000
16NEWLINE_LIMIT = 1_000
19class BTGroup(TypedDict):
20 """
21 Dict describing a line of text rendered within a BT/ET operator pair.
22 If multiple text show operations render text on the same line, the text
23 will be combined into a single BTGroup dict.
25 Keys:
26 tx: x coordinate of first character in BTGroup
27 ty: y coordinate of first character in BTGroup
28 font_size: nominal font size
29 font_height: effective font height
30 text: rendered text
31 displaced_tx: x coordinate of last character in BTGroup
32 flip_sort: -1 if page is upside down, else 1
33 """
35 tx: float
36 ty: float
37 font_size: float
38 font_height: float
39 text: str
40 displaced_tx: float
41 flip_sort: Literal[-1, 1]
44def resolve_font(fonts: dict[str, Font], name: str) -> Font:
45 """
46 Resolve a Tf font name to a layout mode Font.
48 A content stream may select a font name that is not declared in the page
49 resources. Fall back to an uninterpretable font so extraction degrades to
50 the existing incomplete-output path instead of raising KeyError.
52 Args:
53 fonts: font dictionary as returned by PageObject._layout_mode_fonts()
54 name: font name supplied by a Tf operator
56 Returns:
57 Font: the matching font, or an uninterpretable placeholder font.
59 """
60 if name in fonts:
61 return fonts[name]
62 logger_warning(
63 "Font %(name)s is not in the page resources.",
64 name=name, source=__name__
65 )
66 return Font("Unknown", encoding={}, interpretable=False)
69def bt_group(tj_op: TextStateParams, rendered_text: str, displaced_tx: float) -> BTGroup:
70 """
71 BTGroup constructed from a TextStateParams instance, rendered text, and
72 displaced tx value.
74 Args:
75 tj_op (TextStateParams): TextStateParams instance
76 rendered_text (str): rendered text
77 displaced_tx (float): x coordinate of last character in BTGroup
79 """
80 return BTGroup(
81 tx=tj_op.tx,
82 ty=tj_op.ty,
83 font_size=tj_op.font_size,
84 font_height=tj_op.font_height,
85 text=rendered_text,
86 displaced_tx=displaced_tx,
87 flip_sort=-1 if tj_op.flip_vertical else 1,
88 )
91def recurse_to_target_op(
92 ops: Iterator[tuple[list[Any], bytes]],
93 text_state_mgr: TextStateManager,
94 end_target: Literal[b"Q", b"ET"],
95 fonts: dict[str, Font],
96 strip_rotated: bool = True,
97) -> tuple[list[BTGroup], list[TextStateParams]]:
98 """
99 Recurse operators between BT/ET and/or q/Q operators managing the transform
100 stack and capturing text positioning and rendering data.
102 Args:
103 ops: iterator of operators in content stream
104 text_state_mgr: a TextStateManager instance
105 end_target: Either b"Q" (ends b"q" op) or b"ET" (ends b"BT" op)
106 fonts: font dictionary as returned by PageObject._layout_mode_fonts()
108 Returns:
109 tuple: list of BTGroup dicts + list of TextStateParams dataclass instances.
111 """
112 # 1 entry per line of text rendered within each BT/ET operation.
113 bt_groups: list[BTGroup] = []
115 # 1 entry per text show operator (Tj/TJ/'/")
116 tj_ops: list[TextStateParams] = []
118 if end_target == b"Q":
119 # add new q level. cm's added at this level will be popped at next b'Q'
120 text_state_mgr.add_q()
122 for operands, op in ops:
123 # The loop is broken by the end target, or exits normally when there are no more ops.
124 if op == end_target:
125 if op == b"Q":
126 text_state_mgr.remove_q()
127 if op == b"ET":
128 if not tj_ops:
129 return bt_groups, tj_ops
130 _text = ""
131 bt_idx = 0 # idx of first tj in this bt group
132 last_displaced_tx = tj_ops[bt_idx].displaced_tx
133 last_ty = tj_ops[bt_idx].ty
134 for _idx, _tj in enumerate(
135 tj_ops
136 ): # ... build text from new Tj operators
137 if strip_rotated and _tj.rotated:
138 continue
139 if not _tj.font.interpretable: # generates warning
140 continue
141 # if the y position of the text is greater than the font height, assume
142 # the text is on a new line and start a new group
143 if abs(_tj.ty - last_ty) > _tj.font_height:
144 if _text.strip():
145 bt_groups.append(
146 bt_group(tj_ops[bt_idx], _text, last_displaced_tx)
147 )
148 bt_idx = _idx
149 _text = ""
151 # if the x position of the text is less than the last x position by
152 # more than 5 spaces widths, assume the text order should be flipped
153 # and start a new group
154 if (
155 last_displaced_tx - _tj.tx
156 > _tj.space_tx * LAYOUT_NEW_BT_GROUP_SPACE_WIDTHS
157 ):
158 if _text.strip():
159 bt_groups.append(
160 bt_group(tj_ops[bt_idx], _text, last_displaced_tx)
161 )
162 bt_idx = _idx
163 last_displaced_tx = _tj.displaced_tx
164 _text = ""
166 # calculate excess x translation based on ending tx of previous Tj.
167 # multiply by bool (_idx != bt_idx) to ensure spaces aren't double
168 # applied to the first tj of a BTGroup in fixed_width_page().
169 excess_tx = round(_tj.tx - last_displaced_tx, 3) * (_idx != bt_idx)
170 # space_tx could be 0 if either Tz or font_size was 0 for this _tj.
171 spaces = int(excess_tx // _tj.space_tx) if _tj.space_tx else 0
172 if spaces > WHITESPACE_LIMIT:
173 logger_warning(
174 "Limiting excessive whitespace from %(actual)d to %(limit)d characters.",
175 actual=spaces, limit=WHITESPACE_LIMIT, source=__name__
176 )
177 spaces = WHITESPACE_LIMIT
178 new_text = f'{" " * spaces}{_tj.text}'
180 last_ty = _tj.ty
181 _text = f"{_text}{new_text}"
182 last_displaced_tx = _tj.displaced_tx
183 if _text:
184 bt_groups.append(bt_group(tj_ops[bt_idx], _text, last_displaced_tx))
185 text_state_mgr.reset_tm()
186 break
187 if op == b"q":
188 bts, tjs = recurse_to_target_op(
189 ops, text_state_mgr, b"Q", fonts, strip_rotated
190 )
191 bt_groups.extend(bts)
192 tj_ops.extend(tjs)
193 elif op == b"cm":
194 text_state_mgr.add_cm(*operands)
195 elif op == b"BT":
196 bts, tjs = recurse_to_target_op(
197 ops, text_state_mgr, b"ET", fonts, strip_rotated
198 )
199 bt_groups.extend(bts)
200 tj_ops.extend(tjs)
201 elif op == b"Tj":
202 tj_ops.append(text_state_mgr.text_state_params(operands[0]))
203 elif op == b"TJ":
204 _tj = text_state_mgr.text_state_params()
205 for tj_op in operands[0]:
206 if isinstance(tj_op, bytes):
207 _tj = text_state_mgr.text_state_params(tj_op)
208 tj_ops.append(_tj)
209 else:
210 text_state_mgr.add_trm(_tj.displacement_matrix(td_offset=tj_op))
211 elif op == b"'":
212 text_state_mgr.reset_trm()
213 text_state_mgr.add_tm([0, -text_state_mgr.TL])
214 tj_ops.append(text_state_mgr.text_state_params(operands[0]))
215 elif op == b'"':
216 text_state_mgr.reset_trm()
217 text_state_mgr.set_state_param(b"Tw", operands[0])
218 text_state_mgr.set_state_param(b"Tc", operands[1])
219 text_state_mgr.add_tm([0, -text_state_mgr.TL])
220 tj_ops.append(text_state_mgr.text_state_params(operands[2]))
221 elif op in (b"Td", b"Tm", b"TD", b"T*"):
222 text_state_mgr.reset_trm()
223 if op == b"Tm":
224 text_state_mgr.reset_tm()
225 elif op == b"TD":
226 text_state_mgr.set_state_param(b"TL", -operands[1])
227 elif op == b"T*":
228 operands = [0, -text_state_mgr.TL]
229 text_state_mgr.add_tm(operands)
230 elif op == b"Tf":
231 text_state_mgr.set_font(resolve_font(fonts, operands[0]), operands[1])
232 else: # handle Tc, Tw, Tz, TL, and Ts operators
233 text_state_mgr.set_state_param(op, operands)
234 else:
235 logger_warning(
236 "Unbalanced target operations, expected %(end_target)r.",
237 source=__name__,
238 end_target=end_target,
239 )
240 return bt_groups, tj_ops
243def y_coordinate_groups(
244 bt_groups: list[BTGroup], debug_path: Optional[Path] = None
245) -> dict[int, list[BTGroup]]:
246 """
247 Group text operations by rendered y coordinate, i.e. the line number.
249 Args:
250 bt_groups: list of dicts as returned by text_show_operations()
251 debug_path (Path, optional): Path to a directory for saving debug output.
253 Returns:
254 Dict[int, List[BTGroup]]: dict of lists of text rendered by each BT operator
255 keyed by y coordinate
257 """
258 ty_groups = {
259 ty: sorted(grp, key=lambda x: x["tx"])
260 for ty, grp in groupby(
261 bt_groups, key=lambda bt_grp: int(bt_grp["ty"] * bt_grp["flip_sort"])
262 )
263 }
264 # combine groups whose y coordinates differ by less than the effective font height
265 # (accounts for mixed fonts and other minor oddities)
266 last_ty = next(iter(ty_groups))
267 last_txs = {int(_t["tx"]) for _t in ty_groups[last_ty] if _t["text"].strip()}
268 for ty in list(ty_groups)[1:]:
269 fsz = min(ty_groups[_y][0]["font_height"] for _y in (ty, last_ty))
270 txs = {int(_t["tx"]) for _t in ty_groups[ty] if _t["text"].strip()}
271 # prevent merge if both groups are rendering in the same x position.
272 no_text_overlap = not (txs & last_txs)
273 offset_less_than_font_height = abs(ty - last_ty) < fsz
274 if no_text_overlap and offset_less_than_font_height:
275 ty_groups[last_ty] = sorted(
276 ty_groups.pop(ty) + ty_groups[last_ty], key=lambda x: x["tx"]
277 )
278 last_txs |= txs
279 else:
280 last_ty = ty
281 last_txs = txs
282 if debug_path: # pragma: no cover
283 import json # noqa: PLC0415
285 debug_path.joinpath("bt_groups.json").write_text(
286 json.dumps(ty_groups, indent=2, default=str), "utf-8"
287 )
288 return ty_groups
291def text_show_operations(
292 ops: Iterator[tuple[list[Any], bytes]],
293 fonts: dict[str, Font],
294 strip_rotated: bool = True,
295 debug_path: Optional[Path] = None,
296) -> list[BTGroup]:
297 """
298 Extract text from BT/ET operator pairs.
300 Args:
301 ops (Iterator[Tuple[List, bytes]]): iterator of operators in content stream
302 fonts (Dict[str, Font]): font dictionary
303 strip_rotated: Removes text if rotated w.r.t. to the page. Defaults to True.
304 debug_path (Path, optional): Path to a directory for saving debug output.
306 Returns:
307 List[BTGroup]: list of dicts of text rendered by each BT operator
309 """
310 state_mgr = TextStateManager() # transformation stack manager
311 bt_groups: list[BTGroup] = [] # BT operator dict
312 tj_ops: list[TextStateParams] = [] # Tj/TJ operator data
313 for operands, op in ops:
314 if op in (b"BT", b"q"):
315 bts, tjs = recurse_to_target_op(
316 ops, state_mgr, b"ET" if op == b"BT" else b"Q", fonts, strip_rotated
317 )
318 bt_groups.extend(bts)
319 tj_ops.extend(tjs)
320 elif op == b"Tf":
321 state_mgr.set_font(resolve_font(fonts, operands[0]), operands[1])
322 else: # set Tc, Tw, Tz, TL, and Ts if required. ignores all other ops
323 state_mgr.set_state_param(op, operands)
325 if any(tj.rotated for tj in tj_ops):
326 if strip_rotated:
327 logger_warning(
328 "Rotated text discovered. Output will be incomplete.", source=__name__
329 )
330 else:
331 logger_warning(
332 "Rotated text discovered. Layout will be degraded.", source=__name__
333 )
334 if not all(tj.font.interpretable for tj in tj_ops):
335 logger_warning(
336 "PDF contains an uninterpretable font. Output will be incomplete.", source=__name__
337 )
339 # left align the data, i.e. decrement all tx values by min(tx)
340 min_x = min((x["tx"] for x in bt_groups), default=0.0)
341 bt_groups = [
342 dict(ogrp, tx=ogrp["tx"] - min_x, displaced_tx=ogrp["displaced_tx"] - min_x) # type: ignore[misc]
343 for ogrp in sorted(
344 bt_groups, key=lambda x: (x["ty"] * x["flip_sort"], -x["tx"]), reverse=True
345 )
346 ]
348 if debug_path: # pragma: no cover
349 import json # noqa: PLC0415
351 debug_path.joinpath("bts.json").write_text(
352 json.dumps(bt_groups, indent=2, default=str), "utf-8"
353 )
354 debug_path.joinpath("tjs.json").write_text(
355 json.dumps(
356 tj_ops, indent=2, default=lambda x: getattr(x, "to_dict", str)(x)
357 ),
358 "utf-8",
359 )
360 return bt_groups
363def fixed_char_width(bt_groups: list[BTGroup], scale_weight: float = 1.25) -> float:
364 """
365 Calculate average character width weighted by the length of the rendered
366 text in each sample for conversion to fixed-width layout.
368 Args:
369 bt_groups (List[BTGroup]): List of dicts of text rendered by each
370 BT operator
372 Returns:
373 float: fixed character width
375 """
376 char_widths = []
377 for _bt in bt_groups:
378 _len = len(_bt["text"]) * scale_weight
379 char_widths.append(((_bt["displaced_tx"] - _bt["tx"]) / _len, _len))
380 return sum(_w * _l for _w, _l in char_widths) / sum(_l for _, _l in char_widths)
383def fixed_width_page(
384 ty_groups: dict[int, list[BTGroup]], char_width: float, space_vertically: bool, font_height_weight: float
385) -> str:
386 """
387 Generate page text from text operations grouped by rendered y coordinate.
389 Args:
390 ty_groups: dict of text show ops as returned by y_coordinate_groups()
391 char_width: fixed character width
392 space_vertically: include blank lines inferred from y distance + font height.
393 font_height_weight: multiplier for font height when calculating blank lines.
395 Returns:
396 str: page text in a fixed width format that closely adheres to the rendered
397 layout in the source pdf.
399 """
400 lines: list[str] = []
401 last_y_coord = 0
402 table = str.maketrans(dict.fromkeys(range(14, 32), " "))
403 for y_coord, line_data in ty_groups.items():
404 if space_vertically and lines:
405 fh = line_data[0]["font_height"]
406 blank_lines = 0 if fh == 0 else (
407 int(abs(y_coord - last_y_coord) / (fh * font_height_weight)) - 1
408 )
409 if blank_lines > NEWLINE_LIMIT:
410 logger_warning(
411 "Limiting excessive newlines from %(actual)d to %(limit)d.",
412 actual=blank_lines, limit=NEWLINE_LIMIT, source=__name__
413 )
414 blank_lines = NEWLINE_LIMIT
415 lines.extend([""] * blank_lines)
417 line_parts = [] # It uses a list to construct the line, avoiding string concatenation.
418 current_len = 0 # Track the size with int instead of len(str) overhead.
419 last_disp = 0.0
420 for bt_op in line_data:
421 tx = bt_op["tx"]
422 offset = int(tx // char_width)
423 needed_spaces = offset - current_len
424 if needed_spaces > 0 and ceil(last_disp) < int(tx):
425 if needed_spaces > WHITESPACE_LIMIT:
426 logger_warning(
427 "Limiting excessive whitespace from %(actual)d to %(limit)d characters.",
428 actual=needed_spaces, limit=WHITESPACE_LIMIT, source=__name__
429 )
430 needed_spaces = WHITESPACE_LIMIT
431 padding = " " * needed_spaces
432 line_parts.append(padding)
433 current_len += needed_spaces
435 raw_text = bt_op["text"]
436 text = raw_text.translate(table)
437 line_parts.append(text)
438 current_len += len(text)
439 last_disp = bt_op["displaced_tx"]
441 full_line = "".join(line_parts).rstrip()
442 if full_line.strip() or (space_vertically and lines):
443 lines.append(full_line)
445 last_y_coord = y_coord
447 return "\n".join(lines)