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# Copyright (c) 2006, Mathieu Fenniak
2# All rights reserved.
3#
4# Redistribution and use in source and binary forms, with or without
5# modification, are permitted provided that the following conditions are
6# met:
7#
8# * Redistributions of source code must retain the above copyright notice,
9# this list of conditions and the following disclaimer.
10# * Redistributions in binary form must reproduce the above copyright notice,
11# this list of conditions and the following disclaimer in the documentation
12# and/or other materials provided with the distribution.
13# * The name of the author may not be used to endorse or promote products
14# derived from this software without specific prior written permission.
15#
16# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
17# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
18# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
19# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
20# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
21# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
22# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
23# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
24# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
25# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
26# POSSIBILITY OF SUCH DAMAGE.
29__author__ = "Mathieu Fenniak"
30__author_email__ = "biziqe@mathieu.fenniak.net"
32import logging
33import os
34import re
35import sys
36from collections.abc import Iterable, Sequence
37from io import BytesIO
38from math import ceil
39from typing import (
40 Any,
41 Callable,
42 Optional,
43 Union,
44 cast,
45)
47from .._protocols import PdfReaderProtocol, PdfWriterProtocol, XmpInformationProtocol
48from .._utils import (
49 WHITESPACES,
50 BinaryStreamType,
51 StreamType,
52 deprecation_no_replacement,
53 logger_warning,
54 read_non_whitespace,
55 read_until_regex,
56 read_until_whitespace,
57 skip_over_comment,
58)
59from ..constants import (
60 CheckboxRadioButtonAttributes,
61 FieldDictionaryAttributes,
62 OutlineFontFlag,
63 StreamAttributes,
64)
65from ..constants import FilterTypes as FT
66from ..constants import TypArguments as TA
67from ..constants import TypFitArguments as TF
68from ..errors import STREAM_TRUNCATED_PREMATURELY, LimitReachedError, PdfReadError, PdfStreamError
69from ._base import (
70 BooleanObject,
71 ByteStringObject,
72 FloatObject,
73 IndirectObject,
74 NameObject,
75 NullObject,
76 NumberObject,
77 PdfObject,
78 TextStringObject,
79 is_null_or_none,
80)
81from ._fit import Fit
82from ._image_inline import (
83 extract_inline__ascii85_decode,
84 extract_inline__ascii_hex_decode,
85 extract_inline__dct_decode,
86 extract_inline__run_length_decode,
87 extract_inline_default,
88)
89from ._utils import read_hex_string_from_stream, read_string_from_stream
91if sys.version_info >= (3, 11):
92 from typing import Self
93else:
94 from typing_extensions import Self
96logger = logging.getLogger(__name__)
98IndirectPattern = re.compile(rb"[+-]?(\d+)\s+(\d+)\s+R[^a-zA-Z]")
101class ArrayObject(list[Any], PdfObject):
102 def replicate(
103 self,
104 pdf_dest: PdfWriterProtocol,
105 ) -> "ArrayObject":
106 arr = cast(
107 "ArrayObject",
108 self._reference_clone(ArrayObject(), pdf_dest, False),
109 )
110 for data in self:
111 if hasattr(data, "replicate"):
112 arr.append(data.replicate(pdf_dest))
113 else:
114 arr.append(data)
115 return arr
117 def clone(
118 self,
119 pdf_dest: PdfWriterProtocol,
120 force_duplicate: bool = False,
121 ignore_fields: Optional[Sequence[Union[str, int]]] = (),
122 ) -> "ArrayObject":
123 """Clone object into pdf_dest."""
124 try:
125 if self.indirect_reference.pdf == pdf_dest and not force_duplicate: # type: ignore[union-attr]
126 return self
127 except Exception:
128 pass
129 arr = cast(
130 "ArrayObject",
131 self._reference_clone(ArrayObject(), pdf_dest, force_duplicate=True),
132 )
133 for data in self:
134 if isinstance(data, StreamObject):
135 dup = data._reference_clone(
136 data.clone(pdf_dest, force_duplicate, ignore_fields),
137 pdf_dest,
138 force_duplicate,
139 )
140 arr.append(dup.indirect_reference)
141 elif isinstance(data, IndirectObject) and isinstance(resolved := data.get_object(), StreamObject):
142 dup = data._reference_clone(
143 resolved.clone(pdf_dest, force_duplicate=True, ignore_fields=ignore_fields),
144 pdf_dest,
145 force_duplicate,
146 )
147 arr.append(dup.indirect_reference)
148 elif hasattr(data, "clone"):
149 arr.append(data.clone(pdf_dest, force_duplicate, ignore_fields))
150 else:
151 arr.append(data)
152 return arr
154 def hash_bin(self) -> int:
155 """
156 Used to detect modified object.
158 Returns:
159 Hash considering type and value.
161 """
162 return hash((self.__class__, tuple(x.hash_bin() for x in self)))
164 def items(self) -> Iterable[Any]:
165 """Emulate DictionaryObject.items for a list (index, object)."""
166 return enumerate(self)
168 def _to_lst(self, lst: Any) -> list[Any]:
169 # Convert to list, internal
170 result: list[Any]
171 if isinstance(lst, (list, tuple, set)):
172 result = list(lst)
173 elif isinstance(lst, PdfObject):
174 result = [lst]
175 elif isinstance(lst, str):
176 if lst[0] == "/":
177 result = [NameObject(lst)]
178 else:
179 result = [TextStringObject(lst)]
180 elif isinstance(lst, bytes):
181 result = [ByteStringObject(lst)]
182 else: # for numbers,...
183 result = [lst]
184 return result
186 def __add__(self, lst: Any) -> "ArrayObject":
187 """
188 Allow extension by adding list or add one element only
190 Args:
191 lst: any list, tuples are extended the list.
192 other types(numbers,...) will be appended.
193 if str is passed it will be converted into TextStringObject
194 or NameObject (if starting with "/")
195 if bytes is passed it will be converted into ByteStringObject
197 Returns:
198 ArrayObject with all elements
200 """
201 temp = ArrayObject(self)
202 temp.extend(self._to_lst(lst))
203 return temp
205 def __iadd__(self, lst: Any) -> Self:
206 """
207 Allow extension by adding list or add one element only
209 Args:
210 lst: any list, tuples are extended the list.
211 other types(numbers,...) will be appended.
212 if str is passed it will be converted into TextStringObject
213 or NameObject (if starting with "/")
214 if bytes is passed it will be converted into ByteStringObject
216 """
217 self.extend(self._to_lst(lst))
218 return self
220 def __isub__(self, lst: Any) -> Self:
221 """Allow to remove items"""
222 for x in self._to_lst(lst):
223 try:
224 index = self.index(x)
225 del self[index]
226 except ValueError:
227 pass
228 return self
230 def write_to_stream(
231 self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
232 ) -> None:
233 if encryption_key is not None: # deprecated
234 deprecation_no_replacement(
235 "the encryption_key parameter of write_to_stream", "5.0.0"
236 )
237 stream.write(b"[")
238 for data in self:
239 stream.write(b" ")
240 data.write_to_stream(stream)
241 stream.write(b" ]")
243 @staticmethod
244 def read_from_stream(
245 stream: StreamType,
246 pdf: Optional[PdfReaderProtocol],
247 forced_encoding: Union[None, str, list[str], dict[int, str]] = None,
248 ) -> "ArrayObject":
249 arr = ArrayObject()
250 tmp = stream.read(1)
251 if tmp != b"[":
252 raise PdfReadError("Could not read array")
253 while True:
254 # skip leading whitespace
255 tok = stream.read(1)
256 while tok.isspace():
257 tok = stream.read(1)
258 if tok == b"":
259 break
260 if tok == b"%":
261 stream.seek(-1, 1)
262 skip_over_comment(stream)
263 continue
264 stream.seek(-1, 1)
265 # check for array ending
266 peek_ahead = stream.read(1)
267 if peek_ahead == b"]":
268 break
269 stream.seek(-1, 1)
270 # read and append object
271 arr.append(read_object(stream, pdf, forced_encoding))
272 return arr
275class DictionaryObject(dict[Any, Any], PdfObject):
276 def replicate(
277 self,
278 pdf_dest: PdfWriterProtocol,
279 ) -> "DictionaryObject":
280 d__ = cast(
281 "DictionaryObject",
282 self._reference_clone(self.__class__(), pdf_dest, False),
283 )
284 for k, v in self.items():
285 d__[k.replicate(pdf_dest)] = (
286 v.replicate(pdf_dest) if hasattr(v, "replicate") else v
287 )
288 return d__
290 def clone(
291 self,
292 pdf_dest: PdfWriterProtocol,
293 force_duplicate: bool = False,
294 ignore_fields: Optional[Sequence[Union[str, int]]] = (),
295 ) -> "DictionaryObject":
296 """Clone object into pdf_dest."""
297 try:
298 if self.indirect_reference.pdf == pdf_dest and not force_duplicate: # type: ignore[union-attr]
299 return self
300 except Exception:
301 pass
303 visited: set[tuple[int, int]] = set() # (idnum, generation)
304 d__ = cast(
305 "DictionaryObject",
306 self._reference_clone(self.__class__(), pdf_dest, force_duplicate),
307 )
308 if ignore_fields is None:
309 ignore_fields = []
310 if len(d__.keys()) == 0:
311 d__._clone(self, pdf_dest, force_duplicate, ignore_fields, visited)
312 return d__
314 def _clone(
315 self,
316 src: "DictionaryObject",
317 pdf_dest: PdfWriterProtocol,
318 force_duplicate: bool,
319 ignore_fields: Optional[Sequence[Union[str, int]]],
320 visited: set[tuple[int, int]], # (idnum, generation)
321 ) -> None:
322 """
323 Update the object from src.
325 Args:
326 src: "DictionaryObject":
327 pdf_dest:
328 force_duplicate:
329 ignore_fields:
331 """
332 # First we remove the ignore_fields
333 # that are for a limited number of levels
334 assert ignore_fields is not None
335 ignore_fields = list(ignore_fields)
336 x = 0
337 while x < len(ignore_fields):
338 if isinstance(ignore_fields[x], int):
339 if cast(int, ignore_fields[x]) <= 0:
340 del ignore_fields[x]
341 del ignore_fields[x]
342 continue
343 ignore_fields[x] -= 1 # type:ignore
344 x += 1
345 # Check if this is a chain list, we need to loop to prevent recur
346 if any(
347 field not in ignore_fields
348 and field in src
349 and isinstance(src.raw_get(field), IndirectObject)
350 and isinstance(src[field], DictionaryObject)
351 and (
352 src.get("/Type", None) is None
353 or cast(DictionaryObject, src[field]).get("/Type", None) is None
354 or src.get("/Type", None)
355 == cast(DictionaryObject, src[field]).get("/Type", None)
356 )
357 for field in ["/Next", "/Prev", "/N", "/V"]
358 ):
359 ignore_fields = list(ignore_fields)
360 for lst in (("/Next", "/Prev"), ("/N", "/V")):
361 for k in lst:
362 objs = []
363 if (
364 k in src
365 and k not in self
366 and isinstance(src.raw_get(k), IndirectObject)
367 and isinstance(src[k], DictionaryObject)
368 # If need to go further the idea is to check
369 # that the types are the same
370 and (
371 src.get("/Type", None) is None
372 or cast(DictionaryObject, src[k]).get("/Type", None) is None
373 or src.get("/Type", None)
374 == cast(DictionaryObject, src[k]).get("/Type", None)
375 )
376 ):
377 cur_obj: Optional[DictionaryObject] = cast(
378 "DictionaryObject", src[k]
379 )
380 prev_obj: Optional[DictionaryObject] = self
381 while cur_obj is not None:
382 clon = cast(
383 "DictionaryObject",
384 cur_obj._reference_clone(
385 cur_obj.__class__(), pdf_dest, force_duplicate
386 ),
387 )
388 # Check to see if we've previously processed our item
389 if clon.indirect_reference is not None:
390 idnum = clon.indirect_reference.idnum
391 generation = clon.indirect_reference.generation
392 if (idnum, generation) in visited:
393 cur_obj = None
394 break
395 visited.add((idnum, generation))
396 objs.append((cur_obj, clon))
397 assert prev_obj is not None
398 prev_obj[NameObject(k)] = clon.indirect_reference
399 prev_obj = clon
400 try:
401 if cur_obj == src:
402 cur_obj = None
403 else:
404 cur_obj = cast("DictionaryObject", cur_obj[k])
405 except Exception:
406 cur_obj = None
407 for s, c in objs:
408 c._clone(
409 s, pdf_dest, force_duplicate, ignore_fields, visited
410 )
412 for k, v in src.items():
413 if k not in ignore_fields:
414 if isinstance(v, StreamObject):
415 if not hasattr(v, "indirect_reference"):
416 v.indirect_reference = None
417 vv = v.clone(pdf_dest, force_duplicate, ignore_fields)
418 assert vv.indirect_reference is not None
419 self[k.clone(pdf_dest)] = vv.indirect_reference
420 elif k not in self:
421 self[NameObject(k)] = (
422 v.clone(pdf_dest, force_duplicate, ignore_fields)
423 if hasattr(v, "clone")
424 else v
425 )
427 def hash_bin(self) -> int:
428 """
429 Used to detect modified object.
431 Returns:
432 Hash considering type and value.
434 """
435 return hash(
436 (self.__class__, tuple(((k, v.hash_bin()) for k, v in self.items())))
437 )
439 def raw_get(self, key: Any) -> Any:
440 return dict.__getitem__(self, key)
442 def get_inherited(self, key: str, default: Any = None) -> Any:
443 """
444 Returns the value of a key or from the parent if not found.
445 If not found returns default.
447 Args:
448 key: string identifying the field to return
450 default: default value to return
452 Returns:
453 Current key or inherited one, otherwise default value.
455 """
456 current = self
457 visited: set[int] = set()
459 while True:
460 # Detect cyclic parent references
461 obj_id = id(current)
462 if obj_id in visited:
463 raise LimitReachedError(f"Detected cycle in /Parent hierarchy when retrieving value for key {key!r}.")
464 visited.add(obj_id)
466 if key in current:
467 return current[key]
469 if "/Parent" not in current:
470 return default
472 # Walk upward
473 current = cast(
474 "DictionaryObject",
475 current["/Parent"].get_object(),
476 )
478 def __setitem__(self, key: Any, value: Any) -> Any:
479 if not isinstance(key, PdfObject):
480 raise ValueError("Key must be a PdfObject")
481 if not isinstance(value, PdfObject):
482 raise ValueError("Value must be a PdfObject")
483 return dict.__setitem__(self, key, value)
485 def setdefault(self, key: Any, value: Optional[Any] = None) -> Any:
486 if not isinstance(key, PdfObject):
487 raise ValueError("Key must be a PdfObject")
488 if not isinstance(value, PdfObject):
489 raise ValueError("Value must be a PdfObject")
490 return dict.setdefault(self, key, value)
492 def __getitem__(self, key: Any) -> PdfObject:
493 return cast(PdfObject, dict.__getitem__(self, key).get_object())
495 @property
496 def xmp_metadata(self) -> Optional[XmpInformationProtocol]:
497 """
498 Retrieve XMP (Extensible Metadata Platform) data relevant to this
499 object, if available.
501 See Table 347 — Additional entries in a metadata stream dictionary.
503 Returns:
504 Returns a :class:`~pypdf.xmp.XmpInformation` instance
505 that can be used to access XMP metadata from the document. Can also
506 return None if no metadata was found on the document root.
508 """
509 from ..xmp import XmpInformation # noqa: PLC0415
511 metadata = self.get("/Metadata", None)
512 if is_null_or_none(metadata):
513 return None
514 assert metadata is not None, "mypy"
515 metadata = metadata.get_object()
516 return XmpInformation(metadata)
518 def write_to_stream(
519 self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
520 ) -> None:
521 if encryption_key is not None: # deprecated
522 deprecation_no_replacement(
523 "the encryption_key parameter of write_to_stream", "5.0.0"
524 )
525 stream.write(b"<<\n")
526 for key, value in self.items():
527 if len(key) > 2 and key[1] == "%" and key[-1] == "%":
528 continue
529 key.write_to_stream(stream, encryption_key)
530 stream.write(b" ")
531 value.write_to_stream(stream)
532 stream.write(b"\n")
533 stream.write(b">>")
535 @classmethod
536 def _get_next_object_position(
537 cls, position_before: int, position_end: int, generations: list[int], pdf: PdfReaderProtocol
538 ) -> int:
539 out = position_end
540 for generation in generations:
541 for x in pdf.xref[generation].values():
542 if position_before < x <= position_end:
543 out = min(out, x)
544 return out
546 @classmethod
547 def _read_unsized_from_stream(
548 cls, *, stream: BinaryStreamType, pdf: PdfReaderProtocol, length: int,
549 ) -> bytes:
550 current_position = stream.tell()
552 # Determine stream size.
553 try:
554 stream.seek(0, os.SEEK_END)
555 stream_length = stream.tell()
556 finally:
557 stream.seek(current_position)
559 object_position = cls._get_next_object_position(
560 position_before=current_position, position_end=stream_length, generations=list(pdf.xref), pdf=pdf
561 )
563 bytes_to_read = object_position - current_position
564 if bytes_to_read >= length:
565 raise LimitReachedError(f"Requested length of {bytes_to_read} exceeds maximum allowed length.")
567 # Read until the next object position.
568 read_value = stream.read(bytes_to_read)
569 endstream_position = read_value.find(b"endstream")
570 if endstream_position < 0:
571 raise PdfReadError(
572 f"Unable to find 'endstream' marker for obj starting at {current_position}."
573 )
574 # 9 = len(b"endstream")
575 stream.seek(current_position + endstream_position + 9)
576 return read_value[: endstream_position - 1]
578 @staticmethod
579 def read_from_stream(
580 stream: StreamType,
581 pdf: Optional[PdfReaderProtocol],
582 forced_encoding: Union[None, str, list[str], dict[int, str]] = None,
583 ) -> "DictionaryObject":
584 tmp = stream.read(2)
585 if tmp != b"<<":
586 raise PdfReadError(
587 f"Dictionary read error at byte {hex(stream.tell())}: "
588 "stream must begin with '<<'"
589 )
590 data: dict[Any, Any] = {}
591 while True:
592 tok = read_non_whitespace(stream)
593 if tok == b"\x00":
594 continue
595 if tok == b"%":
596 stream.seek(-1, 1)
597 skip_over_comment(stream)
598 continue
599 if not tok:
600 raise PdfStreamError(STREAM_TRUNCATED_PREMATURELY)
602 if tok == b">":
603 stream.read(1)
604 break
605 stream.seek(-1, 1)
606 try:
607 try:
608 key = read_object(stream, pdf)
609 if isinstance(key, NullObject):
610 break
611 if not isinstance(key, NameObject):
612 raise PdfReadError(
613 f"Expecting a NameObject for key but found {key!r}"
614 )
615 except PdfReadError as exc:
616 if pdf is not None and pdf.strict:
617 raise
618 logger_warning("%(exception)r", source=__name__, exception=exc)
619 continue
620 tok = read_non_whitespace(stream)
621 stream.seek(-1, 1)
622 value = read_object(stream, pdf, forced_encoding)
623 except (RecursionError, LimitReachedError) as exc:
624 raise PdfReadError(exc.__repr__())
625 except Exception as exc:
626 if pdf is not None and pdf.strict:
627 raise PdfReadError(exc.__repr__())
628 logger_warning("%(exception)r", source=__name__, exception=exc)
629 retval = DictionaryObject()
630 retval.update(data)
631 return retval # return partial data
633 if not data.get(key):
634 data[key] = value
635 else:
636 # multiple definitions of key not permitted
637 msg = (
638 "Multiple definitions in dictionary at byte "
639 "%(position)s for key %(key)s"
640 )
641 values = {"position": hex(stream.tell()), "key": key}
642 if pdf is not None and pdf.strict:
643 raise PdfReadError(msg % values)
644 logger_warning(msg, source=__name__, **values)
646 pos = stream.tell()
647 s = read_non_whitespace(stream)
648 if s == b"s" and stream.read(5) == b"tream":
649 eol = stream.read(1)
650 # Occasional PDF file output has spaces after 'stream' keyword but before EOL.
651 # patch provided by Danial Sandler
652 while eol == b" ":
653 eol = stream.read(1)
654 if eol not in (b"\n", b"\r"):
655 raise PdfStreamError("Stream data must be followed by a newline")
656 if eol == b"\r" and stream.read(1) != b"\n":
657 stream.seek(-1, 1)
658 # this is a stream object, not a dictionary
659 if StreamAttributes.LENGTH not in data:
660 if pdf is not None and pdf.strict:
661 raise PdfStreamError("Stream length not defined")
662 logger_warning(
663 "Stream length not defined @pos=%(position)d",
664 source=__name__,
665 position=stream.tell(),
666 )
667 data[NameObject(StreamAttributes.LENGTH)] = NumberObject(-1)
668 length = data[StreamAttributes.LENGTH]
669 if isinstance(length, IndirectObject):
670 t = stream.tell()
671 assert pdf is not None, "mypy"
672 length = pdf.get_object(length)
673 stream.seek(t, 0)
674 if length is None: # if the PDF is damaged
675 length = -1
676 pstart = stream.tell()
678 from ..filters import MAX_DECLARED_STREAM_LENGTH # noqa: PLC0415
679 if length >= 0:
680 if length > MAX_DECLARED_STREAM_LENGTH:
681 raise LimitReachedError(f"Declared stream length of {length} exceeds maximum allowed length.")
683 data["__streamdata__"] = stream.read(length)
684 else:
685 data["__streamdata__"] = read_until_regex(
686 stream=stream, regex=re.compile(b"endstream"), length=MAX_DECLARED_STREAM_LENGTH,
687 )
688 e = read_non_whitespace(stream)
689 ndstream = stream.read(8)
690 if (e + ndstream) != b"endstream":
691 # the odd PDF file has a length that is too long, so
692 # we need to read backwards to find the "endstream" ending.
693 # ReportLab (unknown version) generates files with this bug,
694 # and Python users into PDF files tend to be our audience.
695 # we need to do this to correct the streamdata and chop off
696 # an extra character.
697 pos = stream.tell()
698 stream.seek(-10, 1)
699 end = stream.read(9)
700 if end == b"endstream":
701 # we found it by looking back one character further.
702 data["__streamdata__"] = data["__streamdata__"][:-1]
703 elif pdf is not None and not pdf.strict:
704 stream.seek(pstart, 0)
705 data["__streamdata__"] = DictionaryObject._read_unsized_from_stream(
706 stream=stream, pdf=pdf, length=MAX_DECLARED_STREAM_LENGTH
707 )
708 pos = stream.tell()
709 else:
710 stream.seek(pos, 0)
711 raise PdfReadError(
712 "Unable to find 'endstream' marker after stream at byte "
713 f"{hex(stream.tell())} (nd='{ndstream!r}', end='{end!r}')."
714 )
715 else:
716 stream.seek(pos, 0)
717 if "__streamdata__" in data:
718 return StreamObject.initialize_from_dictionary(data)
719 retval = DictionaryObject()
720 retval.update(data)
721 return retval
724class TreeObject(DictionaryObject):
725 def __init__(self, dct: Optional[DictionaryObject] = None) -> None:
726 DictionaryObject.__init__(self)
727 if dct:
728 self.update(dct)
730 def has_children(self) -> bool:
731 return "/First" in self
733 def __iter__(self) -> Any:
734 return self.children()
736 def children(self) -> Iterable[Any]:
737 if not self.has_children():
738 return
740 child_ref = self[NameObject("/First")]
741 last = self[NameObject("/Last")]
742 child = child_ref.get_object()
743 visited: set[int] = set()
744 while True:
745 child_id = id(child)
746 if child_id in visited:
747 logger_warning("Detected cycle in outline structure for %(child)s", source=__name__, child=child)
748 return
749 visited.add(child_id)
751 yield child
753 if child == last:
754 return
755 child_ref = child.get(NameObject("/Next")) # type: ignore[union-attr]
756 if is_null_or_none(child_ref):
757 return
758 child = child_ref.get_object()
760 def add_child(self, child: Any, pdf: PdfWriterProtocol) -> None:
761 self.insert_child(child, None, pdf)
763 def inc_parent_counter_default(
764 self, parent: Union[None, IndirectObject, "TreeObject"], n: int
765 ) -> None:
766 if is_null_or_none(parent):
767 return
768 assert parent is not None, "mypy"
769 parent = cast("TreeObject", parent.get_object())
770 if "/Count" in parent:
771 parent[NameObject("/Count")] = NumberObject(
772 max(0, cast(int, parent[NameObject("/Count")]) + n)
773 )
774 self.inc_parent_counter_default(parent.get("/Parent", None), n)
776 def inc_parent_counter_outline(
777 self, parent: Union[None, IndirectObject, "TreeObject"], n: int
778 ) -> None:
779 if is_null_or_none(parent):
780 return
781 assert parent is not None, "mypy"
782 parent = cast("TreeObject", parent.get_object())
783 # BooleanObject requires comparison with == not is
784 opn = parent.get("/%is_open%", True) == True # noqa: E712
785 c = cast(int, parent.get("/Count", 0))
786 if c < 0:
787 c = abs(c)
788 parent[NameObject("/Count")] = NumberObject((c + n) * (1 if opn else -1))
789 if not opn:
790 return
791 self.inc_parent_counter_outline(parent.get("/Parent", None), n)
793 def insert_child(
794 self,
795 child: Any,
796 before: Any,
797 pdf: PdfWriterProtocol,
798 inc_parent_counter: Optional[Callable[..., Any]] = None,
799 ) -> IndirectObject:
800 if inc_parent_counter is None:
801 inc_parent_counter = self.inc_parent_counter_default
802 child_obj = child.get_object()
803 assert child.indirect_reference is not None, "mypy"
804 child_reference: IndirectObject = child.indirect_reference
806 prev: Optional[DictionaryObject]
807 if "/First" not in self: # no child yet
808 self[NameObject("/First")] = child_reference
809 self[NameObject("/Count")] = NumberObject(0)
810 self[NameObject("/Last")] = child_reference
811 child_obj[NameObject("/Parent")] = self.indirect_reference
812 inc_parent_counter(self, child_obj.get("/Count", 1))
813 if "/Next" in child_obj:
814 del child_obj["/Next"]
815 if "/Prev" in child_obj:
816 del child_obj["/Prev"]
817 return child_reference
818 prev = cast("DictionaryObject", self["/Last"])
820 while prev.indirect_reference != before:
821 if "/Next" in prev:
822 prev = cast("TreeObject", prev["/Next"])
823 else: # append at the end
824 prev[NameObject("/Next")] = cast("TreeObject", child_reference)
825 child_obj[NameObject("/Prev")] = prev.indirect_reference
826 child_obj[NameObject("/Parent")] = self.indirect_reference
827 if "/Next" in child_obj:
828 del child_obj["/Next"]
829 self[NameObject("/Last")] = child_reference
830 inc_parent_counter(self, child_obj.get("/Count", 1))
831 return child_reference
832 try: # insert as first or in the middle
833 assert isinstance(prev["/Prev"], DictionaryObject)
834 prev["/Prev"][NameObject("/Next")] = child_reference
835 child_obj[NameObject("/Prev")] = prev["/Prev"]
836 except Exception: # it means we are inserting in first position
837 child_obj.pop("/Next", None)
838 child_obj[NameObject("/Next")] = prev
839 prev[NameObject("/Prev")] = child_reference
840 child_obj[NameObject("/Parent")] = self.indirect_reference
841 inc_parent_counter(self, child_obj.get("/Count", 1))
842 return child_reference
844 def _remove_node_from_tree(
845 self, prev: Any, prev_ref: Any, cur: Any, last: Any
846 ) -> None:
847 """
848 Adjust the pointers of the linked list and tree node count.
850 Args:
851 prev:
852 prev_ref:
853 cur:
854 last:
856 """
857 next_ref = cur.get(NameObject("/Next"), None)
858 if prev is None:
859 if next_ref:
860 # Removing first tree node
861 next_obj = next_ref.get_object()
862 del next_obj[NameObject("/Prev")]
863 self[NameObject("/First")] = next_ref
864 self[NameObject("/Count")] = NumberObject(
865 self[NameObject("/Count")] - 1 # type: ignore[operator]
866 )
868 else:
869 # Removing only tree node
870 self[NameObject("/Count")] = NumberObject(0)
871 del self[NameObject("/First")]
872 if NameObject("/Last") in self:
873 del self[NameObject("/Last")]
874 else:
875 if next_ref:
876 # Removing middle tree node
877 next_obj = next_ref.get_object()
878 next_obj[NameObject("/Prev")] = prev_ref
879 prev[NameObject("/Next")] = next_ref
880 else:
881 # Removing last tree node
882 assert cur == last
883 del prev[NameObject("/Next")]
884 self[NameObject("/Last")] = prev_ref
885 self[NameObject("/Count")] = NumberObject(self[NameObject("/Count")] - 1) # type: ignore[operator]
887 def remove_child(self, child: Any) -> None:
888 child_obj = child.get_object()
889 child = child_obj.indirect_reference
891 if NameObject("/Parent") not in child_obj:
892 raise ValueError("Removed child does not appear to be a tree item")
893 if child_obj[NameObject("/Parent")] != self:
894 raise ValueError("Removed child is not a member of this tree")
896 found = False
897 prev_ref = None
898 prev = None
899 cur_ref: Optional[Any] = self[NameObject("/First")]
900 cur: Optional[dict[str, Any]] = cur_ref.get_object() # type: ignore[union-attr]
901 last_ref = self[NameObject("/Last")]
902 last = last_ref.get_object()
903 while cur is not None:
904 if cur == child_obj:
905 self._remove_node_from_tree(prev, prev_ref, cur, last)
906 found = True
907 break
909 # Go to the next node
910 prev_ref = cur_ref
911 prev = cur
912 if NameObject("/Next") in cur:
913 cur_ref = cur[NameObject("/Next")]
914 cur = cur_ref.get_object()
915 else:
916 cur_ref = None
917 cur = None
919 if not found:
920 raise ValueError("Removal couldn't find item in tree")
922 _reset_node_tree_relationship(child_obj)
924 def remove_from_tree(self) -> None:
925 """Remove the object from the tree it is in."""
926 if NameObject("/Parent") not in self:
927 raise ValueError("Removed child does not appear to be a tree item")
928 cast("TreeObject", self["/Parent"]).remove_child(self)
930 def empty_tree(self) -> None:
931 for child in self:
932 child_obj = child.get_object()
933 _reset_node_tree_relationship(child_obj)
935 if NameObject("/Count") in self:
936 del self[NameObject("/Count")]
937 if NameObject("/First") in self:
938 del self[NameObject("/First")]
939 if NameObject("/Last") in self:
940 del self[NameObject("/Last")]
943def _reset_node_tree_relationship(child_obj: Any) -> None:
944 """
945 Call this after a node has been removed from a tree.
947 This resets the nodes attributes in respect to that tree.
949 Args:
950 child_obj:
952 """
953 del child_obj[NameObject("/Parent")]
954 if NameObject("/Next") in child_obj:
955 del child_obj[NameObject("/Next")]
956 if NameObject("/Prev") in child_obj:
957 del child_obj[NameObject("/Prev")]
960class StreamObject(DictionaryObject):
961 def __init__(self) -> None:
962 self._data: bytes = b""
963 self.decoded_self: Optional[DecodedStreamObject] = None
965 def replicate(
966 self,
967 pdf_dest: PdfWriterProtocol,
968 ) -> "StreamObject":
969 d__ = cast(
970 "StreamObject",
971 self._reference_clone(self.__class__(), pdf_dest, False),
972 )
973 d__._data = self._data
974 try:
975 decoded_self = self.decoded_self
976 if decoded_self is None:
977 self.decoded_self = None
978 else:
979 self.decoded_self = cast(
980 "DecodedStreamObject", decoded_self.replicate(pdf_dest)
981 )
982 except Exception:
983 pass
984 for k, v in self.items():
985 d__[k.replicate(pdf_dest)] = (
986 v.replicate(pdf_dest) if hasattr(v, "replicate") else v
987 )
988 return d__
990 def _clone(
991 self,
992 src: DictionaryObject,
993 pdf_dest: PdfWriterProtocol,
994 force_duplicate: bool,
995 ignore_fields: Optional[Sequence[Union[str, int]]],
996 visited: set[tuple[int, int]],
997 ) -> None:
998 """
999 Update the object from src.
1001 Args:
1002 src:
1003 pdf_dest:
1004 force_duplicate:
1005 ignore_fields:
1007 """
1008 self._data = cast("StreamObject", src)._data
1009 try:
1010 decoded_self = cast("StreamObject", src).decoded_self
1011 if decoded_self is None:
1012 self.decoded_self = None
1013 else:
1014 self.decoded_self = cast(
1015 "DecodedStreamObject",
1016 decoded_self.clone(pdf_dest, force_duplicate, ignore_fields),
1017 )
1018 except Exception:
1019 pass
1020 super()._clone(src, pdf_dest, force_duplicate, ignore_fields, visited)
1022 def hash_bin(self) -> int:
1023 """
1024 Used to detect modified object.
1026 Returns:
1027 Hash considering type and value.
1029 """
1030 # Use _data to prevent errors on non-decoded streams.
1031 return hash((super().hash_bin(), self._data))
1033 def get_data(self) -> bytes:
1034 return self._data
1036 def set_data(self, data: bytes) -> None:
1037 self._data = data
1039 def hash_value_data(self) -> bytes:
1040 data = super().hash_value_data()
1041 data += self.get_data()
1042 return data
1044 def write_to_stream(
1045 self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
1046 ) -> None:
1047 if encryption_key is not None: # deprecated
1048 deprecation_no_replacement(
1049 "the encryption_key parameter of write_to_stream", "5.0.0"
1050 )
1051 self[NameObject(StreamAttributes.LENGTH)] = NumberObject(len(self._data))
1052 DictionaryObject.write_to_stream(self, stream)
1053 del self[StreamAttributes.LENGTH]
1054 stream.write(b"\nstream\n")
1055 stream.write(self._data)
1056 stream.write(b"\nendstream")
1058 @staticmethod
1059 def initialize_from_dictionary(
1060 data: dict[str, Any]
1061 ) -> Union["EncodedStreamObject", "DecodedStreamObject"]:
1062 retval: Union[EncodedStreamObject, DecodedStreamObject]
1063 if StreamAttributes.FILTER in data:
1064 retval = EncodedStreamObject()
1065 else:
1066 retval = DecodedStreamObject()
1067 retval._data = data["__streamdata__"]
1068 del data["__streamdata__"]
1069 if StreamAttributes.LENGTH in data:
1070 del data[StreamAttributes.LENGTH]
1071 retval.update(data)
1072 return retval
1074 def flate_encode(self, level: int = -1) -> "EncodedStreamObject":
1075 from ..filters import FlateDecode # noqa: PLC0415
1077 if StreamAttributes.FILTER in self:
1078 f = self[StreamAttributes.FILTER]
1079 if isinstance(f, ArrayObject):
1080 f = ArrayObject([NameObject(FT.FLATE_DECODE), *f])
1081 try:
1082 params = ArrayObject(
1083 [NullObject(), *self.get(StreamAttributes.DECODE_PARMS, ArrayObject())]
1084 )
1085 except TypeError:
1086 # case of error where the * operator is not working (not an array
1087 params = ArrayObject(
1088 [NullObject(), self.get(StreamAttributes.DECODE_PARMS, ArrayObject())]
1089 )
1090 else:
1091 f = ArrayObject([NameObject(FT.FLATE_DECODE), f])
1092 params = ArrayObject(
1093 [NullObject(), self.get(StreamAttributes.DECODE_PARMS, NullObject())]
1094 )
1095 else:
1096 f = NameObject(FT.FLATE_DECODE)
1097 params = None
1098 retval = EncodedStreamObject()
1099 retval.update(self)
1100 retval[NameObject(StreamAttributes.FILTER)] = f
1101 if params is not None:
1102 retval[NameObject(StreamAttributes.DECODE_PARMS)] = params
1103 retval._data = FlateDecode.encode(self._data, level)
1104 return retval
1106 def decode_as_image(self, pillow_parameters: Union[dict[str, Any], None] = None) -> Any:
1107 """
1108 Try to decode the stream object as an image
1110 Args:
1111 pillow_parameters: parameters provided to Pillow Image.save() method,
1112 cf. <https://pillow.readthedocs.io/en/stable/reference/Image.html#PIL.Image.Image.save>
1114 Returns:
1115 a PIL image if proper decoding has been found
1116 Raises:
1117 Exception: Errors during decoding will be reported.
1118 It is recommended to catch exceptions to prevent
1119 stops in your program.
1121 """
1122 from ._image_xobject import _xobj_to_image # noqa: PLC0415
1124 if self.get("/Subtype", "") != "/Image":
1125 try:
1126 logger_warning( # pragma: no cover
1127 "%(indirect_reference)s does not seem to be an Image",
1128 source=__name__,
1129 indirect_reference=self.indirect_reference,
1130 )
1131 except AttributeError:
1132 logger_warning( # pragma: no cover
1133 "%(obj)r object does not seem to be an Image",
1134 source=__name__,
1135 obj=self,
1136 )
1137 extension, _, img = _xobj_to_image(self, pillow_parameters)
1138 if extension is None:
1139 return None # pragma: no cover
1140 return img
1143class DecodedStreamObject(StreamObject):
1144 pass
1147class EncodedStreamObject(StreamObject):
1148 def __init__(self) -> None:
1149 self.decoded_self: Optional[DecodedStreamObject] = None
1151 # This overrides the parent method
1152 def get_data(self) -> bytes:
1153 from ..filters import decode_stream_data # noqa: PLC0415
1155 if self.decoded_self is not None:
1156 # Cached version of decoded object
1157 return self.decoded_self.get_data()
1159 # Create decoded object
1160 decoded = DecodedStreamObject()
1161 decoded.set_data(decode_stream_data(self))
1162 for key, value in self.items():
1163 if key not in (StreamAttributes.LENGTH, StreamAttributes.FILTER, StreamAttributes.DECODE_PARMS):
1164 decoded[key] = value
1165 self.decoded_self = decoded
1166 return decoded.get_data()
1168 # This overrides the parent method:
1169 def set_data(self, data: bytes) -> None:
1170 from ..filters import FlateDecode # noqa: PLC0415
1172 if self.get(StreamAttributes.FILTER, "") in (FT.FLATE_DECODE, [FT.FLATE_DECODE]):
1173 if not isinstance(data, bytes):
1174 raise TypeError("Data must be bytes")
1175 if self.decoded_self is None:
1176 self.get_data() # to create self.decoded_self
1177 assert self.decoded_self is not None, "mypy"
1178 self.decoded_self.set_data(data)
1179 super().set_data(FlateDecode.encode(data))
1180 else:
1181 raise PdfReadError(
1182 "Streams encoded with a filter different from FlateDecode are not supported"
1183 )
1186CONTENT_STREAM_ARRAY_MAX_LENGTH = 10_000
1189class ContentStream(DecodedStreamObject):
1190 """
1191 In order to be fast, this data structure can contain either:
1193 * raw data in ._data
1194 * parsed stream operations in ._operations.
1196 At any time, ContentStream object can either have both of those fields defined,
1197 or one field defined and the other set to None.
1199 These fields are "rebuilt" lazily, when accessed:
1201 * when .get_data() is called, if ._data is None, it is rebuilt from ._operations.
1202 * when .operations is called, if ._operations is None, it is rebuilt from ._data.
1204 Conversely, these fields can be invalidated:
1206 * when .set_data() is called, ._operations is set to None.
1207 * when .operations is set, ._data is set to None.
1208 """
1209 _OPERATOR_LENGTH_LIMIT = 128
1211 def __init__(
1212 self,
1213 stream: Any,
1214 pdf: Any,
1215 forced_encoding: Union[None, str, list[str], dict[int, str]] = None,
1216 ) -> None:
1217 self.pdf = pdf
1218 self._operations: list[tuple[Any, bytes]] = []
1220 # stream may be a StreamObject or an ArrayObject containing
1221 # StreamObjects to be concatenated together.
1222 if stream is None:
1223 super().set_data(b"")
1224 else:
1225 stream = stream.get_object()
1226 if isinstance(stream, ArrayObject):
1227 from pypdf.filters import MAX_ARRAY_BASED_STREAM_OUTPUT_LENGTH # noqa: PLC0415
1229 if (stream_length := len(stream)) > CONTENT_STREAM_ARRAY_MAX_LENGTH:
1230 raise LimitReachedError(
1231 f"Array-based stream has {stream_length} > {CONTENT_STREAM_ARRAY_MAX_LENGTH} elements."
1232 )
1233 data = bytearray()
1234 length = 0
1235 for s in stream:
1236 s_resolved = s.get_object()
1237 if isinstance(s_resolved, NullObject):
1238 continue
1239 if not isinstance(s_resolved, StreamObject):
1240 # No need to emit an exception here for now - the PDF structure
1241 # seems to already be broken beforehand in these cases.
1242 logger_warning(
1243 "Expected StreamObject, got %(type_name)s instead. Data might be wrong.",
1244 source=__name__,
1245 type_name=type(s_resolved).__name__,
1246 )
1247 else:
1248 new_data = s_resolved.get_data()
1249 length += len(new_data)
1250 if length > MAX_ARRAY_BASED_STREAM_OUTPUT_LENGTH:
1251 raise LimitReachedError(
1252 f"Array-based stream has at least {length} > "
1253 f"{MAX_ARRAY_BASED_STREAM_OUTPUT_LENGTH} output bytes."
1254 )
1255 data += new_data
1256 if len(data) == 0 or data[-1:] != b"\n":
1257 # There should be no direct need to check for a change of one byte.
1258 length += 1
1259 data += b"\n"
1260 super().set_data(bytes(data))
1261 else:
1262 stream_data = stream.get_data()
1263 assert stream_data is not None
1264 super().set_data(stream_data)
1265 self.forced_encoding = forced_encoding
1267 def replicate(
1268 self,
1269 pdf_dest: PdfWriterProtocol,
1270 ) -> "ContentStream":
1271 d__ = cast(
1272 "ContentStream",
1273 self._reference_clone(self.__class__(None, None), pdf_dest, False),
1274 )
1275 d__._data = self._data
1276 try:
1277 decoded_self = self.decoded_self
1278 if decoded_self is None:
1279 self.decoded_self = None
1280 else:
1281 self.decoded_self = cast(
1282 "DecodedStreamObject", decoded_self.replicate(pdf_dest)
1283 )
1284 except Exception:
1285 pass
1286 for k, v in self.items():
1287 d__[k.replicate(pdf_dest)] = (
1288 v.replicate(pdf_dest) if hasattr(v, "replicate") else v
1289 )
1290 return d__
1291 d__.set_data(self._data)
1292 d__.pdf = pdf_dest
1293 d__._operations = list(self._operations)
1294 d__.forced_encoding = self.forced_encoding
1295 return d__
1297 def clone(
1298 self,
1299 pdf_dest: Any,
1300 force_duplicate: bool = False,
1301 ignore_fields: Optional[Sequence[Union[str, int]]] = (),
1302 ) -> "ContentStream":
1303 """
1304 Clone object into pdf_dest.
1306 Args:
1307 pdf_dest:
1308 force_duplicate:
1309 ignore_fields:
1311 Returns:
1312 The cloned ContentStream
1314 """
1315 try:
1316 if self.indirect_reference.pdf == pdf_dest and not force_duplicate: # type: ignore[union-attr]
1317 return self
1318 except Exception:
1319 pass
1321 visited: set[tuple[int, int]] = set()
1322 d__ = cast(
1323 "ContentStream",
1324 self._reference_clone(
1325 self.__class__(None, None), pdf_dest, force_duplicate
1326 ),
1327 )
1328 if ignore_fields is None:
1329 ignore_fields = []
1330 d__._clone(self, pdf_dest, force_duplicate, ignore_fields, visited)
1331 return d__
1333 def _clone(
1334 self,
1335 src: DictionaryObject,
1336 pdf_dest: PdfWriterProtocol,
1337 force_duplicate: bool,
1338 ignore_fields: Optional[Sequence[Union[str, int]]],
1339 visited: set[tuple[int, int]],
1340 ) -> None:
1341 """
1342 Update the object from src.
1344 Args:
1345 src:
1346 pdf_dest:
1347 force_duplicate:
1348 ignore_fields:
1350 """
1351 src_cs = cast("ContentStream", src)
1352 super().set_data(src_cs._data)
1353 self.pdf = pdf_dest
1354 self._operations = list(src_cs._operations)
1355 self.forced_encoding = src_cs.forced_encoding
1356 # no need to call DictionaryObjection or anything
1357 # like super(DictionaryObject,self)._clone(src, pdf_dest, force_duplicate, ignore_fields, visited)
1359 def _parse_content_stream(self, stream: StreamType) -> None:
1360 # 7.8.2 Content Streams
1361 stream.seek(0, 0)
1362 operands: list[Union[int, str, PdfObject]] = []
1363 while True:
1364 peek = read_non_whitespace(stream)
1365 if peek in (b"", 0):
1366 break
1367 stream.seek(-1, 1)
1368 if peek.isalpha() or peek in (b"'", b'"'):
1369 operator = read_until_regex(
1370 stream=stream, regex=NameObject.delimiter_pattern, length=self._OPERATOR_LENGTH_LIMIT
1371 )
1372 if operator == b"BI":
1373 # begin inline image - a completely different parsing
1374 # mechanism is required, of course... thanks buddy...
1375 assert operands == []
1376 ii = self._read_inline_image(stream)
1377 self._operations.append((ii, b"INLINE IMAGE"))
1378 else:
1379 self._operations.append((operands, operator))
1380 operands = []
1381 elif peek == b"%":
1382 # If we encounter a comment in the content stream, we have to
1383 # handle it here. Typically, read_object will handle
1384 # encountering a comment -- but read_object assumes that
1385 # following the comment must be the object we're trying to
1386 # read. In this case, it could be an operator instead.
1387 while peek not in (b"\r", b"\n", b""):
1388 peek = stream.read(1)
1389 else:
1390 operands.append(read_object(stream, None, self.forced_encoding))
1392 def _read_inline_image(self, stream: StreamType) -> dict[str, Any]:
1393 # begin reading just after the "BI" - begin image
1394 # first read the dictionary of settings.
1395 settings = DictionaryObject()
1396 while True:
1397 tok = read_non_whitespace(stream)
1398 if not tok:
1399 raise PdfReadError("Unexpected end of stream.")
1400 stream.seek(-1, 1)
1401 if tok == b"I":
1402 # "ID" - begin of image data
1403 break
1404 key = read_object(stream, self.pdf)
1405 tok = read_non_whitespace(stream)
1406 stream.seek(-1, 1)
1407 value = read_object(stream, self.pdf)
1408 settings[key] = value
1409 # left at beginning of ID
1410 tmp = stream.read(3)
1411 assert tmp[:2] == b"ID"
1412 filtr = settings.get("/F", settings.get("/Filter", "not set"))
1413 savpos = stream.tell()
1414 if isinstance(filtr, list):
1415 filtr = filtr[0] if filtr else "not set" # used forencoding
1416 if not isinstance(filtr, str):
1417 # A valid filter is a name or an array of names; anything else
1418 # cannot match the abbreviations below, so treat it as unfiltered.
1419 filtr = "not set"
1420 if "AHx" in filtr or "ASCIIHexDecode" in filtr:
1421 data = extract_inline__ascii_hex_decode(stream)
1422 elif "A85" in filtr or "ASCII85Decode" in filtr:
1423 data = extract_inline__ascii85_decode(stream)
1424 elif "RL" in filtr or "RunLengthDecode" in filtr:
1425 data = extract_inline__run_length_decode(stream)
1426 elif "DCT" in filtr or "DCTDecode" in filtr:
1427 data = extract_inline__dct_decode(stream)
1428 elif filtr == "not set":
1429 cs = settings.get("/CS", "")
1430 if isinstance(cs, list):
1431 cs = cs[0] if cs else ""
1432 if not isinstance(cs, str):
1433 cs = ""
1434 if "RGB" in cs:
1435 lcs: float = 3
1436 elif "CMYK" in cs:
1437 lcs = 4
1438 else:
1439 bits = settings.get(
1440 "/BPC",
1441 8 if cs in {"/I", "/G", "/Indexed", "/DeviceGray"} else -1,
1442 )
1443 if isinstance(bits, (int, float)) and bits > 0:
1444 lcs = bits / 8.0
1445 else:
1446 data = extract_inline_default(stream)
1447 lcs = -1
1448 width = settings.get("/W")
1449 height = settings.get("/H")
1450 if lcs > 0 and isinstance(width, int) and isinstance(height, int):
1451 data = stream.read(ceil(width * lcs) * height)
1452 elif lcs > 0:
1453 # Without usable dimensions the raw sample length is unknown;
1454 # fall back to scanning for the `EI` marker.
1455 data = extract_inline_default(stream)
1456 # Move to the `EI` if possible.
1457 ei = read_non_whitespace(stream)
1458 stream.seek(-1, 1)
1459 else:
1460 data = extract_inline_default(stream)
1462 ei = stream.read(3)
1463 # An `EI` at the very end of the stream yields only two bytes; rewinding
1464 # unconditionally would step back into the marker (#3468).
1465 if len(ei) == 3:
1466 stream.seek(-1, 1)
1467 ei_trailing = ei[2:3]
1468 if ei[:2] != b"EI" or (ei_trailing != b"" and ei_trailing not in WHITESPACES):
1469 # Deal with wrong/missing `EI` tags. Example: Wrong dimensions specified above.
1470 stream.seek(savpos, 0)
1471 data = extract_inline_default(stream)
1472 ei = stream.read(3)
1473 if len(ei) == 3:
1474 stream.seek(-1, 1)
1475 ei_trailing = ei[2:3]
1476 if ei[:2] != b"EI" or (
1477 ei_trailing != b"" and ei_trailing not in WHITESPACES
1478 ): # pragma: no cover
1479 # Check the same condition again. This should never fail as
1480 # edge cases are covered by `extract_inline_default` above,
1481 # but check this ot make sure that we are behind the `EI` afterwards.
1482 raise PdfStreamError(
1483 f"Could not extract inline image, even using fallback. Expected 'EI', got {ei!r}"
1484 )
1485 return {"settings": settings, "data": data}
1487 # This overrides the parent method
1488 def get_data(self) -> bytes:
1489 if not self._data:
1490 new_data = BytesIO()
1491 for operands, operator in self._operations:
1492 if operator == b"INLINE IMAGE":
1493 new_data.write(b"BI")
1494 dict_text = BytesIO()
1495 operands["settings"].write_to_stream(dict_text)
1496 new_data.write(dict_text.getvalue()[2:-2])
1497 new_data.write(b"ID ")
1498 new_data.write(operands["data"])
1499 new_data.write(b"EI")
1500 else:
1501 for op in operands:
1502 op.write_to_stream(new_data)
1503 new_data.write(b" ")
1504 new_data.write(operator)
1505 new_data.write(b"\n")
1506 self._data = new_data.getvalue()
1507 return self._data
1509 # This overrides the parent method
1510 def set_data(self, data: bytes) -> None:
1511 super().set_data(data)
1512 self._operations = []
1514 @property
1515 def operations(self) -> list[tuple[Any, bytes]]:
1516 if not self._operations and self._data:
1517 self._parse_content_stream(BytesIO(self._data))
1518 self._data = b""
1519 return self._operations
1521 @operations.setter
1522 def operations(self, operations: list[tuple[Any, bytes]]) -> None:
1523 self._operations = operations
1524 self._data = b""
1526 def isolate_graphics_state(self) -> None:
1527 if self._operations:
1528 self._operations.insert(0, ([], b"q"))
1529 self._operations.append(([], b"Q"))
1530 elif self._data:
1531 self._data = b"q\n" + self._data + b"\nQ\n"
1533 # This overrides the parent method
1534 def write_to_stream(
1535 self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
1536 ) -> None:
1537 if not self._data and self._operations:
1538 self.get_data() # this ensures ._data is rebuilt
1539 super().write_to_stream(stream, encryption_key)
1542def read_object(
1543 stream: StreamType,
1544 pdf: Optional[PdfReaderProtocol],
1545 forced_encoding: Union[None, str, list[str], dict[int, str]] = None,
1546) -> Union[PdfObject, int, str, ContentStream]:
1547 tok = stream.read(1)
1548 stream.seek(-1, 1) # reset to start
1549 if tok == b"/":
1550 return NameObject.read_from_stream(stream, pdf)
1551 if tok == b"<":
1552 # hexadecimal string OR dictionary
1553 peek = stream.read(2)
1554 stream.seek(-2, 1) # reset to start
1555 if peek == b"<<":
1556 return DictionaryObject.read_from_stream(stream, pdf, forced_encoding)
1557 return read_hex_string_from_stream(stream, forced_encoding)
1558 if tok == b"[":
1559 return ArrayObject.read_from_stream(stream, pdf, forced_encoding)
1560 if tok in (b"t", b"f"):
1561 return BooleanObject.read_from_stream(stream)
1562 if tok == b"(":
1563 return read_string_from_stream(stream, forced_encoding)
1564 if tok == b"e" and stream.read(6) == b"endobj":
1565 return NullObject()
1566 if tok == b"n":
1567 return NullObject.read_from_stream(stream)
1568 if tok == b"%":
1569 # comment
1570 skip_over_comment(stream)
1571 tok = read_non_whitespace(stream)
1572 stream.seek(-1, 1)
1573 return read_object(stream, pdf, forced_encoding)
1574 if tok in b"0123456789+-.":
1575 # number object OR indirect reference
1576 peek = stream.read(20)
1577 stream.seek(-len(peek), 1) # reset to start
1578 if IndirectPattern.match(peek) is not None:
1579 assert pdf is not None, "mypy"
1580 return IndirectObject.read_from_stream(stream, pdf)
1581 return NumberObject.read_from_stream(stream)
1582 pos = stream.tell()
1583 stream.seek(-20, 1)
1584 stream_extract = stream.read(80)
1585 stream.seek(pos)
1586 read_until_whitespace(stream)
1587 raise PdfReadError(
1588 f"Invalid Elementary Object starting with {tok!r} @{pos}: {stream_extract!r}"
1589 )
1592class Field(TreeObject):
1593 """
1594 A class representing a field dictionary.
1596 This class is accessed through
1597 :meth:`get_fields()<pypdf.PdfReader.get_fields>`
1598 """
1600 def __init__(self, data: DictionaryObject) -> None:
1601 DictionaryObject.__init__(self)
1602 field_attributes = (
1603 FieldDictionaryAttributes.attributes()
1604 + CheckboxRadioButtonAttributes.attributes()
1605 )
1606 self.indirect_reference = data.indirect_reference
1607 for attr in field_attributes:
1608 try:
1609 self[NameObject(attr)] = data[attr]
1610 except KeyError:
1611 pass
1612 if isinstance(self.get("/V"), EncodedStreamObject):
1613 d = cast(EncodedStreamObject, self[NameObject("/V")]).get_data()
1614 if isinstance(d, bytes):
1615 d_str = d.decode()
1616 elif d is None:
1617 d_str = ""
1618 else:
1619 raise Exception("Should never happen")
1620 self[NameObject("/V")] = TextStringObject(d_str)
1622 # TABLE 8.69 Entries common to all field dictionaries
1623 @property
1624 def field_type(self) -> Optional[NameObject]:
1625 """Read-only property accessing the type of this field."""
1626 return self.get(FieldDictionaryAttributes.FT)
1628 @property
1629 def parent(self) -> Optional[DictionaryObject]:
1630 """Read-only property accessing the parent of this field."""
1631 return self.get(FieldDictionaryAttributes.Parent)
1633 @property
1634 def kids(self) -> Optional["ArrayObject"]:
1635 """Read-only property accessing the kids of this field."""
1636 return self.get(FieldDictionaryAttributes.Kids)
1638 @property
1639 def name(self) -> Optional[str]:
1640 """Read-only property accessing the name of this field."""
1641 return self.get(FieldDictionaryAttributes.T)
1643 @property
1644 def alternate_name(self) -> Optional[str]:
1645 """Read-only property accessing the alternate name of this field."""
1646 return self.get(FieldDictionaryAttributes.TU)
1648 @property
1649 def mapping_name(self) -> Optional[str]:
1650 """
1651 Read-only property accessing the mapping name of this field.
1653 This name is used by pypdf as a key in the dictionary returned by
1654 :meth:`get_fields()<pypdf.PdfReader.get_fields>`
1655 """
1656 return self.get(FieldDictionaryAttributes.TM)
1658 @property
1659 def flags(self) -> Optional[int]:
1660 """
1661 Read-only property accessing the field flags, specifying various
1662 characteristics of the field (see Table 8.70 of the PDF 1.7 reference).
1663 """
1664 return self.get(FieldDictionaryAttributes.Ff)
1666 @property
1667 def value(self) -> Optional[Any]:
1668 """
1669 Read-only property accessing the value of this field.
1671 Format varies based on field type.
1672 """
1673 return self.get(FieldDictionaryAttributes.V)
1675 @property
1676 def default_value(self) -> Optional[Any]:
1677 """Read-only property accessing the default value of this field."""
1678 return self.get(FieldDictionaryAttributes.DV)
1680 @property
1681 def additional_actions(self) -> Optional[DictionaryObject]:
1682 """
1683 Read-only property accessing the additional actions dictionary.
1685 This dictionary defines the field's behavior in response to trigger
1686 events. See Section 8.5.2 of the PDF 1.7 reference.
1687 """
1688 return self.get(FieldDictionaryAttributes.AA)
1691class Destination(TreeObject):
1692 """
1693 A class representing a destination within a PDF file.
1695 See section 12.3.2 of the PDF 2.0 reference.
1697 Args:
1698 title: Title of this destination.
1699 page: Reference to the page of this destination. Should
1700 be an instance of :class:`IndirectObject<pypdf.generic.IndirectObject>`.
1701 fit: How the destination is displayed.
1703 Raises:
1704 PdfReadError: If destination type is invalid.
1706 """
1708 node: Optional[
1709 DictionaryObject
1710 ] = None # node provide access to the original Object
1712 def __init__(
1713 self,
1714 title: Union[str, bytes],
1715 page: Union[NumberObject, IndirectObject, NullObject, DictionaryObject],
1716 fit: Fit,
1717 ) -> None:
1718 self._filtered_children: list[Any] = [] # used in PdfWriter
1720 typ = fit.fit_type
1721 args = fit.fit_args
1723 DictionaryObject.__init__(self)
1724 self[NameObject("/Title")] = TextStringObject(title)
1725 self[NameObject("/Page")] = page
1726 self[NameObject("/Type")] = typ
1728 # from table 8.2 of the PDF 1.7 reference.
1729 if typ == "/XYZ":
1730 if len(args) < 1: # left is missing : should never occur
1731 args.append(NumberObject(0.0))
1732 if len(args) < 2: # top is missing
1733 args.append(NumberObject(0.0))
1734 if len(args) < 3: # zoom is missing
1735 args.append(NumberObject(0.0))
1736 # surplus arguments are ignored rather than failing the unpacking
1737 (
1738 self[NameObject(TA.LEFT)],
1739 self[NameObject(TA.TOP)],
1740 self[NameObject("/Zoom")],
1741 ) = args[:3]
1742 elif len(args) == 0:
1743 pass
1744 elif typ == TF.FIT_R:
1745 if len(args) == 4: # a wrong number of arguments degrades to null
1746 (
1747 self[NameObject(TA.LEFT)],
1748 self[NameObject(TA.BOTTOM)],
1749 self[NameObject(TA.RIGHT)],
1750 self[NameObject(TA.TOP)],
1751 ) = args
1752 else:
1753 (
1754 self[NameObject(TA.LEFT)],
1755 self[NameObject(TA.BOTTOM)],
1756 self[NameObject(TA.RIGHT)],
1757 self[NameObject(TA.TOP)],
1758 ) = (NullObject(), NullObject(), NullObject(), NullObject())
1759 elif typ in [TF.FIT_H, TF.FIT_BH]:
1760 try: # Prefer to be more robust not only to null parameters
1761 (self[NameObject(TA.TOP)],) = args
1762 except Exception:
1763 (self[NameObject(TA.TOP)],) = (NullObject(),)
1764 elif typ in [TF.FIT_V, TF.FIT_BV]:
1765 try: # Prefer to be more robust not only to null parameters
1766 (self[NameObject(TA.LEFT)],) = args
1767 except Exception:
1768 (self[NameObject(TA.LEFT)],) = (NullObject(),)
1769 elif typ in [TF.FIT, TF.FIT_B]:
1770 pass
1771 else:
1772 raise PdfReadError(f"Unknown Destination Type: {typ!r}")
1774 @property
1775 def dest_array(self) -> "ArrayObject":
1776 return ArrayObject(
1777 [self.raw_get("/Page"), self["/Type"]]
1778 + [
1779 self[x]
1780 for x in ["/Left", "/Bottom", "/Right", "/Top", "/Zoom"]
1781 if x in self
1782 ]
1783 )
1785 def write_to_stream(
1786 self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
1787 ) -> None:
1788 if encryption_key is not None: # deprecated
1789 deprecation_no_replacement(
1790 "the encryption_key parameter of write_to_stream", "5.0.0"
1791 )
1792 stream.write(b"<<\n")
1793 key = NameObject("/D")
1794 key.write_to_stream(stream)
1795 stream.write(b" ")
1796 value = self.dest_array
1797 value.write_to_stream(stream)
1799 key = NameObject("/S")
1800 key.write_to_stream(stream)
1801 stream.write(b" ")
1802 value_s = NameObject("/GoTo")
1803 value_s.write_to_stream(stream)
1805 stream.write(b"\n")
1806 stream.write(b">>")
1808 @property
1809 def title(self) -> Optional[str]:
1810 """Read-only property accessing the destination title."""
1811 return self.get("/Title")
1813 @property
1814 def page(self) -> Optional[IndirectObject]:
1815 """Read-only property accessing the IndirectObject of the destination page."""
1816 return self.get("/Page")
1818 @property
1819 def typ(self) -> Optional[str]:
1820 """Read-only property accessing the destination type."""
1821 return self.get("/Type")
1823 @property
1824 def zoom(self) -> Optional[int]:
1825 """Read-only property accessing the zoom factor."""
1826 return self.get("/Zoom", None)
1828 @property
1829 def left(self) -> Optional[FloatObject]:
1830 """Read-only property accessing the left horizontal coordinate."""
1831 return self.get("/Left", None)
1833 @property
1834 def right(self) -> Optional[FloatObject]:
1835 """Read-only property accessing the right horizontal coordinate."""
1836 return self.get("/Right", None)
1838 @property
1839 def top(self) -> Optional[FloatObject]:
1840 """Read-only property accessing the top vertical coordinate."""
1841 return self.get("/Top", None)
1843 @property
1844 def bottom(self) -> Optional[FloatObject]:
1845 """Read-only property accessing the bottom vertical coordinate."""
1846 return self.get("/Bottom", None)
1848 @property
1849 def color(self) -> Optional["ArrayObject"]:
1850 """Read-only property accessing the color in (R, G, B) with values 0.0-1.0."""
1851 return cast(
1852 "ArrayObject",
1853 self.get("/C", ArrayObject([FloatObject(0), FloatObject(0), FloatObject(0)])),
1854 )
1856 @property
1857 def font_format(self) -> Optional[OutlineFontFlag]:
1858 """
1859 Read-only property accessing the font type.
1861 1=italic, 2=bold, 3=both
1862 """
1863 return OutlineFontFlag(self.get("/F", 0))
1865 @property
1866 def outline_count(self) -> Optional[int]:
1867 """
1868 Read-only property accessing the outline count.
1870 positive = expanded
1871 negative = collapsed
1872 absolute value = number of visible descendants at all levels
1873 """
1874 return self.get("/Count", None)