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