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[str, bytes, None] = 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[str, list[str], dict[int, str], None] = 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[str, bytes, None] = 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[str, list[str], dict[int, str], None] = 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 key not in data:
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[IndirectObject, "TreeObject", None], 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[IndirectObject, "TreeObject", None], 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 visited: set[int] = set()
821 while prev.indirect_reference != before:
822 prev_id = id(prev)
823 if prev_id in visited:
824 raise LimitReachedError("Detected cycle in tree structure.")
825 visited.add(prev_id)
826 if "/Next" in prev:
827 prev = cast("TreeObject", prev["/Next"])
828 continue
830 # append at the end
831 prev[NameObject("/Next")] = cast("TreeObject", child_reference)
832 child_obj[NameObject("/Prev")] = prev.indirect_reference
833 child_obj[NameObject("/Parent")] = self.indirect_reference
834 if "/Next" in child_obj:
835 del child_obj["/Next"]
836 self[NameObject("/Last")] = child_reference
837 inc_parent_counter(self, child_obj.get("/Count", 1))
838 return child_reference
839 try: # insert as first or in the middle
840 assert isinstance(prev["/Prev"], DictionaryObject)
841 prev["/Prev"][NameObject("/Next")] = child_reference
842 child_obj[NameObject("/Prev")] = prev["/Prev"]
843 except Exception: # it means we are inserting in first position
844 child_obj.pop("/Next", None)
845 child_obj[NameObject("/Next")] = prev
846 prev[NameObject("/Prev")] = child_reference
847 child_obj[NameObject("/Parent")] = self.indirect_reference
848 inc_parent_counter(self, child_obj.get("/Count", 1))
849 return child_reference
851 def _remove_node_from_tree(
852 self, prev: Any, prev_ref: Any, cur: Any, last: Any
853 ) -> None:
854 """
855 Adjust the pointers of the linked list and tree node count.
857 Args:
858 prev:
859 prev_ref:
860 cur:
861 last:
863 """
864 next_ref = cur.get(NameObject("/Next"), None)
865 if prev is None:
866 if next_ref:
867 # Removing first tree node
868 next_obj = next_ref.get_object()
869 del next_obj[NameObject("/Prev")]
870 self[NameObject("/First")] = next_ref
871 self[NameObject("/Count")] = NumberObject(
872 self[NameObject("/Count")] - 1 # type: ignore[operator]
873 )
875 else:
876 # Removing only tree node
877 self[NameObject("/Count")] = NumberObject(0)
878 del self[NameObject("/First")]
879 if NameObject("/Last") in self:
880 del self[NameObject("/Last")]
881 else:
882 if next_ref:
883 # Removing middle tree node
884 next_obj = next_ref.get_object()
885 next_obj[NameObject("/Prev")] = prev_ref
886 prev[NameObject("/Next")] = next_ref
887 else:
888 # Removing last tree node
889 assert cur == last
890 del prev[NameObject("/Next")]
891 self[NameObject("/Last")] = prev_ref
892 self[NameObject("/Count")] = NumberObject(self[NameObject("/Count")] - 1) # type: ignore[operator]
894 def remove_child(self, child: Any) -> None:
895 child_obj = child.get_object()
896 child = child_obj.indirect_reference
898 if NameObject("/Parent") not in child_obj:
899 raise ValueError("Removed child does not appear to be a tree item")
900 if child_obj[NameObject("/Parent")] != self:
901 raise ValueError("Removed child is not a member of this tree")
903 found = False
904 prev_ref = None
905 prev = None
906 cur_ref: Optional[Any] = self[NameObject("/First")]
907 cur: Optional[dict[str, Any]] = cur_ref.get_object() # type: ignore[union-attr]
908 last_ref = self[NameObject("/Last")]
909 last = last_ref.get_object()
910 while cur is not None:
911 if cur == child_obj:
912 TreeObject._remove_node_from_tree(self, prev, prev_ref, cur, last)
913 found = True
914 break
916 # Go to the next node
917 prev_ref = cur_ref
918 prev = cur
919 if NameObject("/Next") in cur:
920 cur_ref = cur[NameObject("/Next")]
921 cur = cur_ref.get_object()
922 else:
923 cur_ref = None
924 cur = None
926 if not found:
927 raise ValueError("Removal couldn't find item in tree")
929 _reset_node_tree_relationship(child_obj)
931 def remove_from_tree(self) -> None:
932 """Remove the object from the tree it is in."""
933 if "/Parent" not in self:
934 raise ValueError("Removed child does not appear to be a tree item")
935 cast("TreeObject", self["/Parent"]).remove_child(self)
937 def empty_tree(self) -> None:
938 for child in self:
939 child_obj = child.get_object()
940 _reset_node_tree_relationship(child_obj)
942 if NameObject("/Count") in self:
943 del self[NameObject("/Count")]
944 if NameObject("/First") in self:
945 del self[NameObject("/First")]
946 if NameObject("/Last") in self:
947 del self[NameObject("/Last")]
950def _reset_node_tree_relationship(child_obj: Any) -> None:
951 """
952 Call this after a node has been removed from a tree.
954 This resets the nodes attributes in respect to that tree.
956 Args:
957 child_obj:
959 """
960 del child_obj[NameObject("/Parent")]
961 if NameObject("/Next") in child_obj:
962 del child_obj[NameObject("/Next")]
963 if NameObject("/Prev") in child_obj:
964 del child_obj[NameObject("/Prev")]
967class StreamObject(DictionaryObject):
968 def __init__(self) -> None:
969 self._data: bytes = b""
970 self.decoded_self: Optional[DecodedStreamObject] = None
972 def replicate(
973 self,
974 pdf_dest: PdfWriterProtocol,
975 ) -> "StreamObject":
976 d__ = cast(
977 "StreamObject",
978 self._reference_clone(self.__class__(), pdf_dest, False),
979 )
980 d__._data = self._data
981 try:
982 decoded_self = self.decoded_self
983 if decoded_self is None:
984 self.decoded_self = None
985 else:
986 self.decoded_self = cast(
987 "DecodedStreamObject", decoded_self.replicate(pdf_dest)
988 )
989 except Exception:
990 pass
991 for k, v in self.items():
992 d__[k.replicate(pdf_dest)] = (
993 v.replicate(pdf_dest) if hasattr(v, "replicate") else v
994 )
995 return d__
997 def _clone(
998 self,
999 src: DictionaryObject,
1000 pdf_dest: PdfWriterProtocol,
1001 force_duplicate: bool,
1002 ignore_fields: Optional[Sequence[Union[str, int]]],
1003 visited: set[tuple[int, int]],
1004 ) -> None:
1005 """
1006 Update the object from src.
1008 Args:
1009 src:
1010 pdf_dest:
1011 force_duplicate:
1012 ignore_fields:
1014 """
1015 self._data = cast("StreamObject", src)._data
1016 try:
1017 decoded_self = cast("StreamObject", src).decoded_self
1018 if decoded_self is None:
1019 self.decoded_self = None
1020 else:
1021 self.decoded_self = cast(
1022 "DecodedStreamObject",
1023 decoded_self.clone(pdf_dest, force_duplicate, ignore_fields),
1024 )
1025 except Exception:
1026 pass
1027 super()._clone(src, pdf_dest, force_duplicate, ignore_fields, visited)
1029 def hash_bin(self) -> int:
1030 """
1031 Used to detect modified object.
1033 Returns:
1034 Hash considering type and value.
1036 """
1037 # Use _data to prevent errors on non-decoded streams.
1038 return hash((super().hash_bin(), self._data))
1040 def get_data(self) -> bytes:
1041 return self._data
1043 def set_data(self, data: bytes) -> None:
1044 self._data = data
1046 def hash_value_data(self) -> bytes:
1047 data = super().hash_value_data()
1048 data += self.get_data()
1049 return data
1051 def write_to_stream(
1052 self, stream: StreamType, encryption_key: Union[str, bytes, None] = None
1053 ) -> None:
1054 if encryption_key is not None: # deprecated
1055 deprecation_no_replacement(
1056 "the encryption_key parameter of write_to_stream", "5.0.0"
1057 )
1058 self[NameObject(StreamAttributes.LENGTH)] = NumberObject(len(self._data))
1059 DictionaryObject.write_to_stream(self, stream)
1060 del self[StreamAttributes.LENGTH]
1061 stream.write(b"\nstream\n")
1062 stream.write(self._data)
1063 stream.write(b"\nendstream")
1065 @staticmethod
1066 def initialize_from_dictionary(
1067 data: dict[str, Any]
1068 ) -> Union["EncodedStreamObject", "DecodedStreamObject"]:
1069 retval: Union[EncodedStreamObject, DecodedStreamObject]
1070 if StreamAttributes.FILTER in data:
1071 retval = EncodedStreamObject()
1072 else:
1073 retval = DecodedStreamObject()
1074 retval._data = data["__streamdata__"]
1075 del data["__streamdata__"]
1076 if StreamAttributes.LENGTH in data:
1077 del data[StreamAttributes.LENGTH]
1078 retval.update(data)
1079 return retval
1081 def flate_encode(self, level: int = -1) -> "EncodedStreamObject":
1082 from ..filters import FlateDecode # noqa: PLC0415
1084 if StreamAttributes.FILTER in self:
1085 f = self[StreamAttributes.FILTER]
1086 if isinstance(f, ArrayObject):
1087 f = ArrayObject([NameObject(FT.FLATE_DECODE), *f])
1088 try:
1089 params = ArrayObject(
1090 [NullObject(), *self.get(StreamAttributes.DECODE_PARMS, ArrayObject())]
1091 )
1092 except TypeError:
1093 # case of error where the * operator is not working (not an array
1094 params = ArrayObject(
1095 [NullObject(), self.get(StreamAttributes.DECODE_PARMS, ArrayObject())]
1096 )
1097 else:
1098 f = ArrayObject([NameObject(FT.FLATE_DECODE), f])
1099 params = ArrayObject(
1100 [NullObject(), self.get(StreamAttributes.DECODE_PARMS, NullObject())]
1101 )
1102 else:
1103 f = NameObject(FT.FLATE_DECODE)
1104 params = None
1105 retval = EncodedStreamObject()
1106 retval.update(self)
1107 retval[NameObject(StreamAttributes.FILTER)] = f
1108 if params is not None:
1109 retval[NameObject(StreamAttributes.DECODE_PARMS)] = params
1110 retval._data = FlateDecode.encode(self._data, level)
1111 return retval
1113 def decode_as_image(self, pillow_parameters: Union[dict[str, Any], None] = None) -> Any:
1114 """
1115 Try to decode the stream object as an image
1117 Args:
1118 pillow_parameters: parameters provided to Pillow Image.save() method,
1119 cf. <https://pillow.readthedocs.io/en/stable/reference/Image.html#PIL.Image.Image.save>
1121 Returns:
1122 a PIL image if proper decoding has been found
1123 Raises:
1124 Exception: Errors during decoding will be reported.
1125 It is recommended to catch exceptions to prevent
1126 stops in your program.
1128 """
1129 from ._image_xobject import _xobj_to_image # noqa: PLC0415
1131 if self.get("/Subtype", "") != "/Image":
1132 try:
1133 logger_warning( # pragma: no cover
1134 "%(indirect_reference)s does not seem to be an Image",
1135 source=__name__,
1136 indirect_reference=self.indirect_reference,
1137 )
1138 except AttributeError:
1139 logger_warning( # pragma: no cover
1140 "%(obj)r object does not seem to be an Image",
1141 source=__name__,
1142 obj=self,
1143 )
1144 extension, _, img = _xobj_to_image(self, pillow_parameters)
1145 if extension is None:
1146 return None # pragma: no cover
1147 return img
1150class DecodedStreamObject(StreamObject):
1151 pass
1154class EncodedStreamObject(StreamObject):
1155 def __init__(self) -> None:
1156 self.decoded_self: Optional[DecodedStreamObject] = None
1158 # This overrides the parent method
1159 def get_data(self) -> bytes:
1160 from ..filters import decode_stream_data # noqa: PLC0415
1162 if self.decoded_self is not None:
1163 # Cached version of decoded object
1164 return self.decoded_self.get_data()
1166 # Create decoded object
1167 decoded = DecodedStreamObject()
1168 decoded.set_data(decode_stream_data(self))
1169 for key, value in self.items():
1170 if key not in (StreamAttributes.LENGTH, StreamAttributes.FILTER, StreamAttributes.DECODE_PARMS):
1171 decoded[key] = value
1172 self.decoded_self = decoded
1173 return decoded.get_data()
1175 # This overrides the parent method:
1176 def set_data(self, data: bytes) -> None:
1177 from ..filters import FlateDecode # noqa: PLC0415
1179 if self.get(StreamAttributes.FILTER, "") in (FT.FLATE_DECODE, [FT.FLATE_DECODE]):
1180 if not isinstance(data, bytes):
1181 raise TypeError("Data must be bytes")
1182 if self.decoded_self is None:
1183 self.get_data() # to create self.decoded_self
1184 assert self.decoded_self is not None, "mypy"
1185 self.decoded_self.set_data(data)
1186 super().set_data(FlateDecode.encode(data))
1187 else:
1188 raise PdfReadError(
1189 "Streams encoded with a filter different from FlateDecode are not supported"
1190 )
1193CONTENT_STREAM_ARRAY_MAX_LENGTH = 10_000
1196class ContentStream(DecodedStreamObject):
1197 """
1198 In order to be fast, this data structure can contain either:
1200 * raw data in ._data
1201 * parsed stream operations in ._operations.
1203 At any time, ContentStream object can either have both of those fields defined,
1204 or one field defined and the other set to None.
1206 These fields are "rebuilt" lazily, when accessed:
1208 * when .get_data() is called, if ._data is None, it is rebuilt from ._operations.
1209 * when .operations is called, if ._operations is None, it is rebuilt from ._data.
1211 Conversely, these fields can be invalidated:
1213 * when .set_data() is called, ._operations is set to None.
1214 * when .operations is set, ._data is set to None.
1215 """
1216 _OPERATOR_LENGTH_LIMIT = 128
1218 def __init__(
1219 self,
1220 stream: Any,
1221 pdf: Any,
1222 forced_encoding: Union[str, list[str], dict[int, str], None] = None,
1223 ) -> None:
1224 self.pdf = pdf
1225 self._operations: list[tuple[Any, bytes]] = []
1227 # stream may be a StreamObject or an ArrayObject containing
1228 # StreamObjects to be concatenated together.
1229 if stream is None:
1230 super().set_data(b"")
1231 else:
1232 stream = stream.get_object()
1233 if isinstance(stream, ArrayObject):
1234 from pypdf.filters import MAX_ARRAY_BASED_STREAM_OUTPUT_LENGTH # noqa: PLC0415
1236 if (stream_length := len(stream)) > CONTENT_STREAM_ARRAY_MAX_LENGTH:
1237 raise LimitReachedError(
1238 f"Array-based stream has {stream_length} > {CONTENT_STREAM_ARRAY_MAX_LENGTH} elements."
1239 )
1240 data = bytearray()
1241 length = 0
1242 for s in stream:
1243 s_resolved = s.get_object()
1244 if isinstance(s_resolved, NullObject):
1245 continue
1246 if not isinstance(s_resolved, StreamObject):
1247 # No need to emit an exception here for now - the PDF structure
1248 # seems to already be broken beforehand in these cases.
1249 logger_warning(
1250 "Expected StreamObject, got %(type_name)s instead. Data might be wrong.",
1251 source=__name__,
1252 type_name=type(s_resolved).__name__,
1253 )
1254 else:
1255 new_data = s_resolved.get_data()
1256 length += len(new_data)
1257 if length > MAX_ARRAY_BASED_STREAM_OUTPUT_LENGTH:
1258 raise LimitReachedError(
1259 f"Array-based stream has at least {length} > "
1260 f"{MAX_ARRAY_BASED_STREAM_OUTPUT_LENGTH} output bytes."
1261 )
1262 data += new_data
1263 if len(data) == 0 or data[-1:] != b"\n":
1264 # There should be no direct need to check for a change of one byte.
1265 length += 1
1266 data += b"\n"
1267 super().set_data(bytes(data))
1268 else:
1269 stream_data = stream.get_data()
1270 assert stream_data is not None
1271 super().set_data(stream_data)
1272 self.forced_encoding = forced_encoding
1274 def replicate(
1275 self,
1276 pdf_dest: PdfWriterProtocol,
1277 ) -> "ContentStream":
1278 d__ = cast(
1279 "ContentStream",
1280 self._reference_clone(self.__class__(None, None), pdf_dest, False),
1281 )
1282 d__._data = self._data
1283 try:
1284 decoded_self = self.decoded_self
1285 if decoded_self is None:
1286 self.decoded_self = None
1287 else:
1288 self.decoded_self = cast(
1289 "DecodedStreamObject", decoded_self.replicate(pdf_dest)
1290 )
1291 except Exception:
1292 pass
1293 for k, v in self.items():
1294 d__[k.replicate(pdf_dest)] = (
1295 v.replicate(pdf_dest) if hasattr(v, "replicate") else v
1296 )
1297 return d__
1298 d__.set_data(self._data)
1299 d__.pdf = pdf_dest
1300 d__._operations = list(self._operations)
1301 d__.forced_encoding = self.forced_encoding
1302 return d__
1304 def clone(
1305 self,
1306 pdf_dest: Any,
1307 force_duplicate: bool = False,
1308 ignore_fields: Optional[Sequence[Union[str, int]]] = (),
1309 ) -> "ContentStream":
1310 """
1311 Clone object into pdf_dest.
1313 Args:
1314 pdf_dest:
1315 force_duplicate:
1316 ignore_fields:
1318 Returns:
1319 The cloned ContentStream
1321 """
1322 try:
1323 if self.indirect_reference.pdf == pdf_dest and not force_duplicate: # type: ignore[union-attr]
1324 return self
1325 except Exception:
1326 pass
1328 visited: set[tuple[int, int]] = set()
1329 d__ = cast(
1330 "ContentStream",
1331 self._reference_clone(
1332 self.__class__(None, None), pdf_dest, force_duplicate
1333 ),
1334 )
1335 if ignore_fields is None:
1336 ignore_fields = []
1337 d__._clone(self, pdf_dest, force_duplicate, ignore_fields, visited)
1338 return d__
1340 def _clone(
1341 self,
1342 src: DictionaryObject,
1343 pdf_dest: PdfWriterProtocol,
1344 force_duplicate: bool,
1345 ignore_fields: Optional[Sequence[Union[str, int]]],
1346 visited: set[tuple[int, int]],
1347 ) -> None:
1348 """
1349 Update the object from src.
1351 Args:
1352 src:
1353 pdf_dest:
1354 force_duplicate:
1355 ignore_fields:
1357 """
1358 src_cs = cast("ContentStream", src)
1359 super().set_data(src_cs._data)
1360 self.pdf = pdf_dest
1361 self._operations = list(src_cs._operations)
1362 self.forced_encoding = src_cs.forced_encoding
1363 # no need to call DictionaryObjection or anything
1364 # like super(DictionaryObject,self)._clone(src, pdf_dest, force_duplicate, ignore_fields, visited)
1366 def _parse_content_stream(self, stream: StreamType) -> None:
1367 # 7.8.2 Content Streams
1368 stream.seek(0, 0)
1369 operands: list[Union[int, str, PdfObject]] = []
1370 while True:
1371 peek = read_non_whitespace(stream)
1372 if peek in (b"", 0):
1373 break
1374 stream.seek(-1, 1)
1375 if peek.isalpha() or peek in (b"'", b'"'):
1376 operator = read_until_regex(
1377 stream=stream, regex=NameObject.delimiter_pattern, length=self._OPERATOR_LENGTH_LIMIT
1378 )
1379 if operator == b"BI":
1380 # begin inline image - a completely different parsing
1381 # mechanism is required, of course... thanks buddy...
1382 assert operands == []
1383 ii = self._read_inline_image(stream)
1384 self._operations.append((ii, b"INLINE IMAGE"))
1385 else:
1386 self._operations.append((operands, operator))
1387 operands = []
1388 elif peek == b"%":
1389 # If we encounter a comment in the content stream, we have to
1390 # handle it here. Typically, read_object will handle
1391 # encountering a comment -- but read_object assumes that
1392 # following the comment must be the object we're trying to
1393 # read. In this case, it could be an operator instead.
1394 while peek not in (b"\r", b"\n", b""):
1395 peek = stream.read(1)
1396 else:
1397 operands.append(read_object(stream, None, self.forced_encoding))
1399 def _read_inline_image(self, stream: StreamType) -> dict[str, Any]:
1400 # begin reading just after the "BI" - begin image
1401 # first read the dictionary of settings.
1402 settings = DictionaryObject()
1403 while True:
1404 tok = read_non_whitespace(stream)
1405 if not tok:
1406 raise PdfReadError("Unexpected end of stream.")
1407 stream.seek(-1, 1)
1408 if tok == b"I":
1409 # "ID" - begin of image data
1410 break
1411 key = read_object(stream, self.pdf)
1412 tok = read_non_whitespace(stream)
1413 stream.seek(-1, 1)
1414 value = read_object(stream, self.pdf)
1415 settings[key] = value
1416 # left at beginning of ID
1417 tmp = stream.read(3)
1418 assert tmp[:2] == b"ID"
1419 filtr = settings.get("/F", settings.get("/Filter", "not set"))
1420 savpos = stream.tell()
1421 if isinstance(filtr, list):
1422 filtr = filtr[0] if filtr else "not set" # used forencoding
1423 if not isinstance(filtr, str):
1424 # A valid filter is a name or an array of names; anything else
1425 # cannot match the abbreviations below, so treat it as unfiltered.
1426 filtr = "not set"
1427 if "AHx" in filtr or "ASCIIHexDecode" in filtr:
1428 data = extract_inline__ascii_hex_decode(stream)
1429 elif "A85" in filtr or "ASCII85Decode" in filtr:
1430 data = extract_inline__ascii85_decode(stream)
1431 elif "RL" in filtr or "RunLengthDecode" in filtr:
1432 data = extract_inline__run_length_decode(stream)
1433 elif "DCT" in filtr or "DCTDecode" in filtr:
1434 data = extract_inline__dct_decode(stream)
1435 elif filtr == "not set":
1436 cs = settings.get("/CS", "")
1437 if isinstance(cs, list):
1438 cs = cs[0] if cs else ""
1439 if not isinstance(cs, str):
1440 cs = ""
1441 if "RGB" in cs:
1442 lcs: float = 3
1443 elif "CMYK" in cs:
1444 lcs = 4
1445 else:
1446 bits = settings.get(
1447 "/BPC",
1448 8 if cs in {"/I", "/G", "/Indexed", "/DeviceGray"} else -1,
1449 )
1450 if isinstance(bits, (int, float)) and bits > 0:
1451 lcs = bits / 8.0
1452 else:
1453 data = extract_inline_default(stream)
1454 lcs = -1
1455 width = settings.get("/W")
1456 height = settings.get("/H")
1457 if lcs > 0 and isinstance(width, int) and isinstance(height, int):
1458 data = stream.read(ceil(width * lcs) * height)
1459 elif lcs > 0:
1460 # Without usable dimensions the raw sample length is unknown;
1461 # fall back to scanning for the `EI` marker.
1462 data = extract_inline_default(stream)
1463 # Move to the `EI` if possible.
1464 ei = read_non_whitespace(stream)
1465 stream.seek(-1, 1)
1466 else:
1467 data = extract_inline_default(stream)
1469 ei = stream.read(3)
1470 # An `EI` at the very end of the stream yields only two bytes; rewinding
1471 # unconditionally would step back into the marker (#3468).
1472 if len(ei) == 3:
1473 stream.seek(-1, 1)
1474 ei_trailing = ei[2:3]
1475 if ei[:2] != b"EI" or (ei_trailing != b"" and ei_trailing not in WHITESPACES):
1476 # Deal with wrong/missing `EI` tags. Example: Wrong dimensions specified above.
1477 stream.seek(savpos, 0)
1478 data = extract_inline_default(stream)
1479 ei = stream.read(3)
1480 if len(ei) == 3:
1481 stream.seek(-1, 1)
1482 ei_trailing = ei[2:3]
1483 if ei[:2] != b"EI" or (
1484 ei_trailing != b"" and ei_trailing not in WHITESPACES
1485 ): # pragma: no cover
1486 # Check the same condition again. This should never fail as
1487 # edge cases are covered by `extract_inline_default` above,
1488 # but check this ot make sure that we are behind the `EI` afterwards.
1489 raise PdfStreamError(
1490 f"Could not extract inline image, even using fallback. Expected 'EI', got {ei!r}"
1491 )
1492 return {"settings": settings, "data": data}
1494 # This overrides the parent method
1495 def get_data(self) -> bytes:
1496 if not self._data:
1497 new_data = BytesIO()
1498 for operands, operator in self._operations:
1499 if operator == b"INLINE IMAGE":
1500 new_data.write(b"BI")
1501 dict_text = BytesIO()
1502 operands["settings"].write_to_stream(dict_text)
1503 new_data.write(dict_text.getvalue()[2:-2])
1504 new_data.write(b"ID ")
1505 new_data.write(operands["data"])
1506 new_data.write(b"EI")
1507 else:
1508 for op in operands:
1509 op.write_to_stream(new_data)
1510 new_data.write(b" ")
1511 new_data.write(operator)
1512 new_data.write(b"\n")
1513 self._data = new_data.getvalue()
1514 return self._data
1516 # This overrides the parent method
1517 def set_data(self, data: bytes) -> None:
1518 super().set_data(data)
1519 self._operations = []
1521 @property
1522 def operations(self) -> list[tuple[Any, bytes]]:
1523 if not self._operations and self._data:
1524 self._parse_content_stream(BytesIO(self._data))
1525 self._data = b""
1526 return self._operations
1528 @operations.setter
1529 def operations(self, operations: list[tuple[Any, bytes]]) -> None:
1530 self._operations = operations
1531 self._data = b""
1533 def isolate_graphics_state(self) -> None:
1534 if self._operations:
1535 self._operations.insert(0, ([], b"q"))
1536 self._operations.append(([], b"Q"))
1537 elif self._data:
1538 self._data = b"q\n" + self._data + b"\nQ\n"
1540 # This overrides the parent method
1541 def write_to_stream(
1542 self, stream: StreamType, encryption_key: Union[str, bytes, None] = None
1543 ) -> None:
1544 if not self._data and self._operations:
1545 self.get_data() # this ensures ._data is rebuilt
1546 super().write_to_stream(stream, encryption_key)
1549def read_object(
1550 stream: StreamType,
1551 pdf: Optional[PdfReaderProtocol],
1552 forced_encoding: Union[str, list[str], dict[int, str], None] = None,
1553) -> Union[PdfObject, int, str, ContentStream]:
1554 tok = stream.read(1)
1555 stream.seek(-1, 1) # reset to start
1556 if tok == b"/":
1557 return NameObject.read_from_stream(stream, pdf)
1558 if tok == b"<":
1559 # hexadecimal string OR dictionary
1560 peek = stream.read(2)
1561 stream.seek(-2, 1) # reset to start
1562 if peek == b"<<":
1563 return DictionaryObject.read_from_stream(stream, pdf, forced_encoding)
1564 return read_hex_string_from_stream(stream, forced_encoding)
1565 if tok == b"[":
1566 return ArrayObject.read_from_stream(stream, pdf, forced_encoding)
1567 if tok in (b"t", b"f"):
1568 return BooleanObject.read_from_stream(stream)
1569 if tok == b"(":
1570 return read_string_from_stream(stream, forced_encoding)
1571 if tok == b"e" and stream.read(6) == b"endobj":
1572 return NullObject()
1573 if tok == b"n":
1574 return NullObject.read_from_stream(stream)
1575 if tok == b"%":
1576 # comment
1577 skip_over_comment(stream)
1578 tok = read_non_whitespace(stream)
1579 stream.seek(-1, 1)
1580 return read_object(stream, pdf, forced_encoding)
1581 if tok in b"0123456789+-.":
1582 # number object OR indirect reference
1583 peek = stream.read(20)
1584 stream.seek(-len(peek), 1) # reset to start
1585 if IndirectPattern.match(peek) is not None:
1586 assert pdf is not None, "mypy"
1587 return IndirectObject.read_from_stream(stream, pdf)
1588 return NumberObject.read_from_stream(stream)
1589 pos = stream.tell()
1590 stream.seek(-20, 1)
1591 stream_extract = stream.read(80)
1592 stream.seek(pos)
1593 read_until_whitespace(stream)
1594 raise PdfReadError(
1595 f"Invalid Elementary Object starting with {tok!r} @{pos}: {stream_extract!r}"
1596 )
1599class Field(TreeObject):
1600 """
1601 A class representing a field dictionary.
1603 This class is accessed through
1604 :meth:`get_fields()<pypdf.PdfReader.get_fields>`
1605 """
1607 def __init__(self, data: DictionaryObject) -> None:
1608 DictionaryObject.__init__(self)
1609 field_attributes = (
1610 FieldDictionaryAttributes.attributes()
1611 + CheckboxRadioButtonAttributes.attributes()
1612 )
1613 self.indirect_reference = data.indirect_reference
1614 for attr in field_attributes:
1615 try:
1616 self[NameObject(attr)] = data[attr]
1617 except KeyError:
1618 pass
1619 if isinstance(self.get("/V"), EncodedStreamObject):
1620 d = cast(EncodedStreamObject, self[NameObject("/V")]).get_data()
1621 if isinstance(d, bytes):
1622 d_str = d.decode()
1623 elif d is None:
1624 d_str = ""
1625 else:
1626 raise Exception("Should never happen")
1627 self[NameObject("/V")] = TextStringObject(d_str)
1629 # TABLE 8.69 Entries common to all field dictionaries
1630 @property
1631 def field_type(self) -> Optional[NameObject]:
1632 """Read-only property accessing the type of this field."""
1633 return self.get(FieldDictionaryAttributes.FT)
1635 @property
1636 def parent(self) -> Optional[DictionaryObject]:
1637 """Read-only property accessing the parent of this field."""
1638 return self.get(FieldDictionaryAttributes.Parent)
1640 @property
1641 def kids(self) -> Optional["ArrayObject"]:
1642 """Read-only property accessing the kids of this field."""
1643 return self.get(FieldDictionaryAttributes.Kids)
1645 @property
1646 def name(self) -> Optional[str]:
1647 """Read-only property accessing the name of this field."""
1648 return self.get(FieldDictionaryAttributes.T)
1650 @property
1651 def alternate_name(self) -> Optional[str]:
1652 """Read-only property accessing the alternate name of this field."""
1653 return self.get(FieldDictionaryAttributes.TU)
1655 @property
1656 def mapping_name(self) -> Optional[str]:
1657 """
1658 Read-only property accessing the mapping name of this field.
1660 This name is used by pypdf as a key in the dictionary returned by
1661 :meth:`get_fields()<pypdf.PdfReader.get_fields>`
1662 """
1663 return self.get(FieldDictionaryAttributes.TM)
1665 @property
1666 def flags(self) -> Optional[int]:
1667 """
1668 Read-only property accessing the field flags, specifying various
1669 characteristics of the field (see Table 8.70 of the PDF 1.7 reference).
1670 """
1671 return self.get(FieldDictionaryAttributes.Ff)
1673 @property
1674 def value(self) -> Optional[Any]:
1675 """
1676 Read-only property accessing the value of this field.
1678 Format varies based on field type.
1679 """
1680 return self.get(FieldDictionaryAttributes.V)
1682 @property
1683 def default_value(self) -> Optional[Any]:
1684 """Read-only property accessing the default value of this field."""
1685 return self.get(FieldDictionaryAttributes.DV)
1687 @property
1688 def additional_actions(self) -> Optional[DictionaryObject]:
1689 """
1690 Read-only property accessing the additional actions dictionary.
1692 This dictionary defines the field's behavior in response to trigger
1693 events. See Section 8.5.2 of the PDF 1.7 reference.
1694 """
1695 return self.get(FieldDictionaryAttributes.AA)
1698class Destination(TreeObject):
1699 """
1700 A class representing a destination within a PDF file.
1702 See section 12.3.2 of the PDF 2.0 reference.
1704 Args:
1705 title: Title of this destination.
1706 page: Reference to the page of this destination. Should
1707 be an instance of :class:`IndirectObject<pypdf.generic.IndirectObject>`.
1708 fit: How the destination is displayed.
1710 Raises:
1711 PdfReadError: If destination type is invalid.
1713 """
1715 node: Optional[
1716 DictionaryObject
1717 ] = None # node provide access to the original Object
1719 def remove_from_tree(self) -> None:
1720 """
1721 Remove the outline item this destination was built from.
1723 `reader.outline` and `writer.outline` yield detached copies rather than
1724 the nodes themselves, so the copy never has a `/Parent` and removing it
1725 would be a no-op. `node` is the dictionary in the document.
1726 """
1727 if self.node is None:
1728 super().remove_from_tree()
1729 elif "/Parent" not in self.node:
1730 raise ValueError("Removed child does not appear to be a tree item")
1731 else:
1732 TreeObject.remove_child(
1733 cast("TreeObject", self.node["/Parent"]), self.node
1734 )
1736 def __init__(
1737 self,
1738 title: Union[str, bytes],
1739 page: Union[NumberObject, IndirectObject, NullObject, DictionaryObject],
1740 fit: Fit,
1741 ) -> None:
1742 self._filtered_children: list[Any] = [] # used in PdfWriter
1744 typ = fit.fit_type
1745 args = fit.fit_args
1747 DictionaryObject.__init__(self)
1748 self[NameObject("/Title")] = TextStringObject(title)
1749 self[NameObject("/Page")] = page
1750 self[NameObject("/Type")] = typ
1752 # from table 8.2 of the PDF 1.7 reference.
1753 if typ == "/XYZ":
1754 if len(args) < 1: # left is missing : should never occur
1755 args.append(NumberObject(0.0))
1756 if len(args) < 2: # top is missing
1757 args.append(NumberObject(0.0))
1758 if len(args) < 3: # zoom is missing
1759 args.append(NumberObject(0.0))
1760 # surplus arguments are ignored rather than failing the unpacking
1761 (
1762 self[NameObject(TA.LEFT)],
1763 self[NameObject(TA.TOP)],
1764 self[NameObject("/Zoom")],
1765 ) = args[:3]
1766 elif len(args) == 0:
1767 pass
1768 elif typ == TF.FIT_R:
1769 if len(args) == 4: # a wrong number of arguments degrades to null
1770 (
1771 self[NameObject(TA.LEFT)],
1772 self[NameObject(TA.BOTTOM)],
1773 self[NameObject(TA.RIGHT)],
1774 self[NameObject(TA.TOP)],
1775 ) = args
1776 else:
1777 (
1778 self[NameObject(TA.LEFT)],
1779 self[NameObject(TA.BOTTOM)],
1780 self[NameObject(TA.RIGHT)],
1781 self[NameObject(TA.TOP)],
1782 ) = (NullObject(), NullObject(), NullObject(), NullObject())
1783 elif typ in [TF.FIT_H, TF.FIT_BH]:
1784 try: # Prefer to be more robust not only to null parameters
1785 (self[NameObject(TA.TOP)],) = args
1786 except Exception:
1787 (self[NameObject(TA.TOP)],) = (NullObject(),)
1788 elif typ in [TF.FIT_V, TF.FIT_BV]:
1789 try: # Prefer to be more robust not only to null parameters
1790 (self[NameObject(TA.LEFT)],) = args
1791 except Exception:
1792 (self[NameObject(TA.LEFT)],) = (NullObject(),)
1793 elif typ in [TF.FIT, TF.FIT_B]:
1794 pass
1795 else:
1796 raise PdfReadError(f"Unknown Destination Type: {typ!r}")
1798 @property
1799 def dest_array(self) -> "ArrayObject":
1800 return ArrayObject(
1801 [self.raw_get("/Page"), self["/Type"]]
1802 + [
1803 self[x]
1804 for x in ["/Left", "/Bottom", "/Right", "/Top", "/Zoom"]
1805 if x in self
1806 ]
1807 )
1809 def write_to_stream(
1810 self, stream: StreamType, encryption_key: Union[str, bytes, None] = None
1811 ) -> None:
1812 if encryption_key is not None: # deprecated
1813 deprecation_no_replacement(
1814 "the encryption_key parameter of write_to_stream", "5.0.0"
1815 )
1816 stream.write(b"<<\n")
1817 key = NameObject("/D")
1818 key.write_to_stream(stream)
1819 stream.write(b" ")
1820 value = self.dest_array
1821 value.write_to_stream(stream)
1823 key = NameObject("/S")
1824 key.write_to_stream(stream)
1825 stream.write(b" ")
1826 value_s = NameObject("/GoTo")
1827 value_s.write_to_stream(stream)
1829 stream.write(b"\n")
1830 stream.write(b">>")
1832 @property
1833 def title(self) -> Optional[str]:
1834 """Read-only property accessing the destination title."""
1835 return self.get("/Title")
1837 @property
1838 def page(self) -> Optional[IndirectObject]:
1839 """Read-only property accessing the IndirectObject of the destination page."""
1840 return self.get("/Page")
1842 @property
1843 def typ(self) -> Optional[str]:
1844 """Read-only property accessing the destination type."""
1845 return self.get("/Type")
1847 @property
1848 def zoom(self) -> Optional[int]:
1849 """Read-only property accessing the zoom factor."""
1850 return self.get("/Zoom", None)
1852 @property
1853 def left(self) -> Optional[FloatObject]:
1854 """Read-only property accessing the left horizontal coordinate."""
1855 return self.get("/Left", None)
1857 @property
1858 def right(self) -> Optional[FloatObject]:
1859 """Read-only property accessing the right horizontal coordinate."""
1860 return self.get("/Right", None)
1862 @property
1863 def top(self) -> Optional[FloatObject]:
1864 """Read-only property accessing the top vertical coordinate."""
1865 return self.get("/Top", None)
1867 @property
1868 def bottom(self) -> Optional[FloatObject]:
1869 """Read-only property accessing the bottom vertical coordinate."""
1870 return self.get("/Bottom", None)
1872 @property
1873 def color(self) -> Optional["ArrayObject"]:
1874 """Read-only property accessing the color in (R, G, B) with values 0.0-1.0."""
1875 return cast(
1876 "ArrayObject",
1877 self.get("/C", ArrayObject([FloatObject(0), FloatObject(0), FloatObject(0)])),
1878 )
1880 @property
1881 def font_format(self) -> Optional[OutlineFontFlag]:
1882 """
1883 Read-only property accessing the font type.
1885 1=italic, 2=bold, 3=both
1886 """
1887 return OutlineFontFlag(self.get("/F", 0))
1889 @property
1890 def outline_count(self) -> Optional[int]:
1891 """
1892 Read-only property accessing the outline count.
1894 positive = expanded
1895 negative = collapsed
1896 absolute value = number of visible descendants at all levels
1897 """
1898 return self.get("/Count", None)