Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/pypdf/_utils.py: 52%
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.
28"""Utility functions for PDF library."""
29__author__ = "Mathieu Fenniak"
30__author_email__ = "biziqe@mathieu.fenniak.net"
32import functools
33import logging
34import re
35import sys
36import warnings
37from dataclasses import dataclass
38from datetime import datetime, timezone
39from io import DEFAULT_BUFFER_SIZE
40from os import SEEK_CUR
41from re import Pattern
42from typing import (
43 IO,
44 Any,
45 NoReturn,
46 Optional,
47 Union,
48)
50if sys.version_info[:2] >= (3, 10):
51 # Python 3.10+: https://www.python.org/dev/peps/pep-0484/
52 from typing import TypeAlias
53else:
54 from typing_extensions import TypeAlias
56if sys.version_info >= (3, 11):
57 from typing import Self
58else:
59 from typing_extensions import Self
61from .errors import (
62 STREAM_TRUNCATED_PREMATURELY,
63 DeprecationError,
64 LimitReachedError,
65 PdfStreamError,
66)
68TransformationMatrixType: TypeAlias = tuple[
69 tuple[float, float, float], tuple[float, float, float], tuple[float, float, float]
70]
71CompressedTransformationMatrix: TypeAlias = tuple[
72 float, float, float, float, float, float
73]
75StreamType = IO[Any]
76BinaryStreamType = IO[bytes]
77StrByteType = Union[str, StreamType]
80def parse_iso8824_date(text: Optional[str]) -> Optional[datetime]:
81 orgtext = text
82 if not text:
83 return None
84 if text[0].isdigit():
85 text = "D:" + text
86 if text.endswith(("Z", "z")):
87 text += "0000"
88 text = text.replace("z", "+").replace("Z", "+").replace("'", "")
89 i = max(text.find("+"), text.find("-"))
90 if i > 0 and i != len(text) - 5:
91 text += "00"
92 for f in (
93 "D:%Y",
94 "D:%Y%m",
95 "D:%Y%m%d",
96 "D:%Y%m%d%H",
97 "D:%Y%m%d%H%M",
98 "D:%Y%m%d%H%M%S",
99 "D:%Y%m%d%H%M%S%z",
100 ):
101 try:
102 d = datetime.strptime(text, f) # noqa: DTZ007
103 except ValueError:
104 continue
105 else:
106 if text.endswith("+0000"):
107 d = d.replace(tzinfo=timezone.utc)
108 return d
109 raise ValueError(f"Can not convert date: {orgtext}")
112def format_iso8824_date(dt: datetime) -> str:
113 """
114 Convert a datetime object to PDF date string format.
116 Converts datetime to the PDF date format D:YYYYMMDDHHmmSSOHH'mm
117 as specified in the PDF Reference.
119 Args:
120 dt: A datetime object to convert.
122 Returns:
123 A date string in PDF format.
124 """
125 date_str = dt.strftime("D:%Y%m%d%H%M%S")
126 if dt.tzinfo is not None:
127 offset = dt.utcoffset()
128 assert offset is not None
129 total_seconds = int(offset.total_seconds())
130 hours, remainder = divmod(abs(total_seconds), 3600)
131 minutes = remainder // 60
132 sign = "+" if total_seconds >= 0 else "-"
133 date_str += f"{sign}{hours:02d}'{minutes:02d}'"
134 return date_str
137def _get_max_pdf_version_header(header1: str, header2: str) -> str:
138 versions = (
139 "%PDF-1.3",
140 "%PDF-1.4",
141 "%PDF-1.5",
142 "%PDF-1.6",
143 "%PDF-1.7",
144 "%PDF-2.0",
145 )
146 pdf_header_indices = []
147 if header1 in versions:
148 pdf_header_indices.append(versions.index(header1))
149 if header2 in versions:
150 pdf_header_indices.append(versions.index(header2))
151 if len(pdf_header_indices) == 0:
152 raise ValueError(f"Neither {header1!r} nor {header2!r} are proper headers")
153 return versions[max(pdf_header_indices)]
156WHITESPACES = (b"\x00", b"\t", b"\n", b"\f", b"\r", b" ")
157WHITESPACES_AS_BYTES = b"".join(WHITESPACES)
158WHITESPACES_AS_REGEXP = b"[" + WHITESPACES_AS_BYTES + b"]"
161def read_until_whitespace(stream: StreamType, max_bytes: Optional[int] = None) -> bytes:
162 """
163 Read non-whitespace characters and return them.
165 Stops upon encountering whitespace or when max_bytes is reached.
167 Args:
168 stream: The data stream from which was read.
169 max_bytes: The maximum number of bytes returned; by default unlimited.
171 Returns:
172 The data which was read.
174 """
175 txt = bytearray()
176 while True:
177 tok = stream.read(1)
178 if tok.isspace() or not tok:
179 break
180 txt += tok
181 if len(txt) == max_bytes:
182 break
183 return bytes(txt)
186def read_non_whitespace(stream: BinaryStreamType) -> bytes:
187 """
188 Find and read the next non-whitespace character (ignores whitespace).
190 Args:
191 stream: The data stream from which was read.
193 Returns:
194 The data which was read.
196 """
197 tok = stream.read(1)
198 while tok in WHITESPACES:
199 tok = stream.read(1)
200 return tok
203def skip_over_whitespace(stream: StreamType) -> bool:
204 """
205 Similar to read_non_whitespace, but return a boolean if at least one
206 whitespace character was read.
208 Args:
209 stream: The data stream from which was read.
211 Returns:
212 True if one or more whitespace was skipped, otherwise return False.
214 """
215 tok = stream.read(1)
216 cnt = 0
217 while tok in WHITESPACES:
218 cnt += 1
219 tok = stream.read(1)
220 return cnt > 0
223def check_if_whitespace_only(value: bytes) -> bool:
224 """
225 Check if the given value consists of whitespace characters only.
227 Args:
228 value: The bytes to check.
230 Returns:
231 True if the value only has whitespace characters, otherwise return False.
233 """
234 return all(b in WHITESPACES_AS_BYTES for b in value)
237NEUTRAL_CHARACTER_RANGES = (
238 ("\x00", "\x2F"), # ASCII control codes, space, and early punctuation (!"#$%)
239 ("\x3A", "\x40"), # ASCII operators and punctuation between digits & A (:;<=>?@)
240 ("\u2000", "\u206F"), # General punctuation
241 ("\u20A0", "\u21FF"), # Currency symbols, diacritical marks, letter-like symbols, number forms, arrows
242)
245def is_char_neutral(char: str, custom_special_characters: str = "") -> bool:
246 """Check if a character is part of neutral character ranges"""
247 if any(start <= char <= end for start, end in NEUTRAL_CHARACTER_RANGES):
248 return True
250 return bool(custom_special_characters and char in custom_special_characters)
253RTL_CHARACTER_RANGES = (
254 ("\u0590", "\u08FF"), # Hebrew, Arabic, Syriac, Thaana, N'Ko, etc.
255 ("\uFB1D", "\uFDFF"), # Hebrew & Arabic Presentation Forms-A
256 ("\uFE70", "\uFEFF"), # Arabic Presentation Forms-B
257)
260def is_char_rtl(char: str, custom_rtl_min: str = "", custom_rtl_max: str = "") -> bool:
261 """Check if a character is part of RTL character ranges"""
262 if any(start <= char <= end for start, end in RTL_CHARACTER_RANGES):
263 return True
265 return bool(custom_rtl_min and custom_rtl_max and (custom_rtl_min <= char <= custom_rtl_max))
268def skip_over_comment(stream: StreamType) -> None:
269 tok = stream.read(1)
270 stream.seek(-1, 1)
271 if tok == b"%":
272 while tok not in (b"\n", b"\r"):
273 tok = stream.read(1)
274 if tok == b"":
275 raise PdfStreamError("File ended unexpectedly.")
278def read_until_regex(*, stream: StreamType, regex: Pattern[bytes], length: int = sys.maxsize) -> bytes:
279 """
280 Read until the regular expression pattern matched (ignore the match).
281 Treats EOF on the underlying stream as the end of the token to be matched.
283 Args:
284 stream: The stream to read from.
285 regex: The pattern to search for.
286 length: The (approximated) maximum number of bytes to read before raising an exception.
288 Returns:
289 The read bytes.
291 """
292 parts: list[bytes] = []
293 total_length = 0
294 tail = b""
295 chunk_size = 16
296 while True:
297 token = stream.read(chunk_size)
298 if not token:
299 return b"".join(parts)
300 token_length = len(token)
301 if (current_length := total_length + token_length) >= length:
302 raise LimitReachedError(
303 f"Read stream length of {current_length} exceeds maximum allowed length of {length}."
304 )
306 # Search overlap of previous tail + new chunk to catch
307 # multi-byte regex matches spanning chunk boundaries.
308 current_buffer = tail + token
309 search_match = regex.search(current_buffer)
310 parts.append(token)
311 if search_match is not None:
312 overlap = len(tail)
313 actual_start = total_length - overlap + search_match.start()
314 stream.seek(actual_start - total_length - token_length, 1)
315 return b"".join(parts)[:actual_start]
316 total_length += token_length
318 # Fixed overlap: 16 bytes is sufficient for the short
319 # delimiter patterns used in PDF parsing.
320 tail = token[-16:]
321 if chunk_size < 8192:
322 chunk_size <<= 1
325def read_block_backwards(stream: BinaryStreamType, to_read: int) -> bytes:
326 """
327 Given a stream at position X, read a block of size to_read ending at position X.
329 This changes the stream's position to the beginning of where the block was
330 read.
332 Args:
333 stream:
334 to_read:
336 Returns:
337 The data which was read.
339 """
340 if stream.tell() < to_read:
341 raise PdfStreamError("Could not read malformed PDF file")
342 # Seek to the start of the block we want to read.
343 stream.seek(-to_read, SEEK_CUR)
344 read = stream.read(to_read)
345 # Seek to the start of the block we read after reading it.
346 stream.seek(-to_read, SEEK_CUR)
347 return read
350def read_previous_line(stream: StreamType) -> bytes:
351 """
352 Given a byte stream with current position X, return the previous line.
354 All characters between the first CR/LF byte found before X
355 (or, the start of the file, if no such byte is found) and position X
356 After this call, the stream will be positioned one byte after the
357 first non-CRLF character found beyond the first CR/LF byte before X,
358 or, if no such byte is found, at the beginning of the stream.
360 Args:
361 stream: StreamType:
363 Returns:
364 The data which was read.
366 """
367 line_content = []
368 found_crlf = False
369 if stream.tell() == 0:
370 raise PdfStreamError(STREAM_TRUNCATED_PREMATURELY)
371 while True:
372 to_read = min(DEFAULT_BUFFER_SIZE, stream.tell())
373 if to_read == 0:
374 break
375 # Read the block. After this, our stream will be one
376 # beyond the initial position.
377 block = read_block_backwards(stream, to_read)
378 idx = len(block) - 1
379 if not found_crlf:
380 # We haven't found our first CR/LF yet.
381 # Read off characters until we hit one.
382 while idx >= 0 and block[idx] not in b"\r\n":
383 idx -= 1
384 if idx >= 0:
385 found_crlf = True
386 if found_crlf:
387 # We found our first CR/LF already (on this block or
388 # a previous one).
389 # Our combined line is the remainder of the block
390 # plus any previously read blocks.
391 line_content.append(block[idx + 1 :])
392 # Continue to read off any more CRLF characters.
393 while idx >= 0 and block[idx] in b"\r\n":
394 idx -= 1
395 else:
396 # Didn't find CR/LF yet - add this block to our
397 # previously read blocks and continue.
398 line_content.append(block)
399 if idx >= 0:
400 # We found the next non-CRLF character.
401 # Set the stream position correctly, then break
402 stream.seek(idx + 1, SEEK_CUR)
403 break
404 # Join all the blocks in the line (which are in reverse order)
405 return b"".join(line_content[::-1])
408def matrix_multiply(
409 a: TransformationMatrixType, b: TransformationMatrixType
410) -> TransformationMatrixType:
411 return tuple( # type: ignore[return-value]
412 tuple(sum(float(i) * float(j) for i, j in zip(row, col)) for col in zip(*b))
413 for row in a
414 )
417def mark_location(stream: StreamType) -> None:
418 """Create text file showing current location in context."""
419 # Mainly for debugging
420 radius = 5000
421 stream.seek(-radius, 1)
422 with open("pypdf_pdfLocation.txt", "wb") as output_fh:
423 output_fh.write(stream.read(radius))
424 output_fh.write(b"HERE")
425 output_fh.write(stream.read(radius))
426 stream.seek(-radius, 1)
429def deprecate(msg: str, stacklevel: int = 3) -> None:
430 warnings.warn(msg, DeprecationWarning, stacklevel=stacklevel)
433def deprecation(msg: str) -> NoReturn:
434 raise DeprecationError(msg)
437def deprecate_with_replacement(old_name: str, new_name: str, removed_in: str) -> None:
438 """Issue a warning that a feature will be removed, but has a replacement."""
439 deprecate(
440 f"{old_name} is deprecated and will be removed in pypdf {removed_in}. Use {new_name} instead.",
441 4,
442 )
445def deprecation_with_replacement(old_name: str, new_name: str, removed_in: str) -> NoReturn:
446 """Raise an exception that a feature was already removed, but has a replacement."""
447 deprecation(
448 f"{old_name} is deprecated and was removed in pypdf {removed_in}. Use {new_name} instead."
449 )
452def deprecate_no_replacement(name: str, removed_in: str) -> None:
453 """Issue a warning that a feature will be removed without replacement."""
454 deprecate(f"{name} is deprecated and will be removed in pypdf {removed_in}.", 4)
457def deprecation_no_replacement(name: str, removed_in: str) -> NoReturn:
458 """Raise an exception that a feature was already removed without replacement."""
459 deprecation(f"{name} is deprecated and was removed in pypdf {removed_in}.")
462def logger_error(message: str, *, source: str, **values: Any) -> None:
463 """
464 Use this instead of logger.error directly.
466 That allows people to overwrite it more easily.
468 See the docs on when to use which:
469 https://pypdf.readthedocs.io/en/latest/user/suppress-warnings.html
470 """
471 if values:
472 logging.getLogger(source).error(message, values)
473 else:
474 logging.getLogger(source).error(message)
477def logger_warning(message: str, *, source: str, **values: Any) -> None:
478 """
479 Use this instead of logger.warning directly.
481 That allows people to overwrite it more easily.
483 ## Exception, warnings.warn, logger_warning
484 - Exceptions should be used if the user should write code that deals with
485 an error case, e.g. the PDF being completely broken.
486 - warnings.warn should be used if the user needs to fix their code, e.g.
487 DeprecationWarnings
488 - logger_warning should be used if the user needs to know that an issue was
489 handled by pypdf, e.g. a non-compliant PDF being read in a way that
490 pypdf could apply a robustness fix to still read it. This applies mainly
491 to strict=False mode.
492 """
493 if values:
494 logging.getLogger(source).warning(message, values)
495 else:
496 # Keep parity with logger_error and support plain warning messages.
497 # Passing an empty dict to logging is not equivalent to passing no args:
498 # plain messages would fail while being formatted.
499 logging.getLogger(source).warning(message)
502def rename_kwargs(
503 func_name: str, kwargs: dict[str, Any], aliases: dict[str, str], fail: bool = False
504) -> None:
505 """
506 Helper function to deprecate arguments.
508 Args:
509 func_name: Name of the function to be deprecated
510 kwargs:
511 aliases:
512 fail:
514 """
515 for old_term, new_term in aliases.items():
516 if old_term in kwargs:
517 if fail:
518 raise DeprecationError(
519 f"{old_term} is deprecated as an argument. Use {new_term} instead"
520 )
521 if new_term in kwargs:
522 raise TypeError(
523 f"{func_name} received both {old_term} and {new_term} as "
524 f"an argument. {old_term} is deprecated. "
525 f"Use {new_term} instead."
526 )
527 kwargs[new_term] = kwargs.pop(old_term)
528 warnings.warn(
529 message=(
530 f"{old_term} is deprecated as an argument. Use {new_term} instead"
531 ),
532 category=DeprecationWarning,
533 stacklevel=3,
534 )
537def _human_readable_bytes(bytes: int) -> str:
538 if bytes < 10**3:
539 return f"{bytes} Byte"
540 if bytes < 10**6:
541 return f"{bytes / 10**3:.1f} kB"
542 if bytes < 10**9:
543 return f"{bytes / 10**6:.1f} MB"
544 return f"{bytes / 10**9:.1f} GB"
547# The following class has been copied from Django:
548# https://github.com/django/django/blob/adae619426b6f50046b3daaa744db52989c9d6db/django/utils/functional.py#L51-L65
549# It received some modifications to comply with our own coding standards.
550#
551# Original license:
552#
553# ---------------------------------------------------------------------------------
554# Copyright (c) Django Software Foundation and individual contributors.
555# All rights reserved.
556#
557# Redistribution and use in source and binary forms, with or without modification,
558# are permitted provided that the following conditions are met:
559#
560# 1. Redistributions of source code must retain the above copyright notice,
561# this list of conditions and the following disclaimer.
562#
563# 2. Redistributions in binary form must reproduce the above copyright
564# notice, this list of conditions and the following disclaimer in the
565# documentation and/or other materials provided with the distribution.
566#
567# 3. Neither the name of Django nor the names of its contributors may be used
568# to endorse or promote products derived from this software without
569# specific prior written permission.
570#
571# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
572# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
573# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
574# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
575# ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
576# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
577# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
578# ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
579# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
580# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
581# ---------------------------------------------------------------------------------
582class classproperty: # noqa: N801
583 """
584 Decorator that converts a method with a single cls argument into a property
585 that can be accessed directly from the class.
586 """
588 def __init__(self, method=None) -> None: # type: ignore # noqa: ANN001
589 self.fget = method
591 def __get__(self, instance, cls=None) -> Any: # type: ignore # noqa: ANN001
592 return self.fget(cls)
594 def getter(self, method) -> Self: # type: ignore # noqa: ANN001
595 self.fget = method
596 return self
599@dataclass
600class File:
601 from .generic import IndirectObject # noqa: PLC0415
603 name: str = ""
604 """
605 Filename as identified within the PDF file.
606 """
607 data: bytes = b""
608 """
609 Data as bytes.
610 """
611 indirect_reference: Optional[IndirectObject] = None
612 """
613 Reference to the object storing the stream.
614 """
616 def __str__(self) -> str:
617 return f"{self.__class__.__name__}(name={self.name}, data: {_human_readable_bytes(len(self.data))})"
619 def __repr__(self) -> str:
620 return self.__str__()[:-1] + f", hash: {hash(self.data)})"
623@functools.total_ordering
624class Version:
625 COMPONENT_PATTERN = re.compile(r"^(\d+)(.*)$")
627 def __init__(self, version_str: str) -> None:
628 self.version_str = version_str
629 self.components = self._parse_version(version_str)
631 def _parse_version(self, version_str: str) -> list[tuple[int, str]]:
632 components = version_str.split(".")
633 parsed_components = []
634 for component in components:
635 match = Version.COMPONENT_PATTERN.match(component)
636 if not match:
637 parsed_components.append((0, component))
638 continue
639 integer_prefix = match.group(1)
640 suffix = match.group(2)
641 if integer_prefix is None:
642 integer_prefix = 0
643 parsed_components.append((int(integer_prefix), suffix))
644 return parsed_components
646 def __eq__(self, other: object) -> bool:
647 if not isinstance(other, Version):
648 return False
649 return self.components == other.components
651 def __hash__(self) -> int:
652 # Convert to tuple as lists cannot be hashed.
653 return hash((self.__class__, tuple(self.components)))
655 def __lt__(self, other: Any) -> bool:
656 if not isinstance(other, Version):
657 raise ValueError(f"Version cannot be compared against {type(other)}")
659 for self_component, other_component in zip(self.components, other.components):
660 self_value, self_suffix = self_component
661 other_value, other_suffix = other_component
663 if self_value < other_value:
664 return True
665 if self_value > other_value:
666 return False
668 if self_suffix < other_suffix:
669 return True
670 if self_suffix > other_suffix:
671 return False
673 return len(self.components) < len(other.components)
676@dataclass
677class _TraversalState:
678 """Sometimes we need mutable objects which just count something."""
679 entry_count: int = 0
680 has_logged: bool = False