Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/pypdf/_utils.py: 51%
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, maxchars: Optional[int] = None) -> bytes:
162 """
163 Read non-whitespace characters and return them.
165 Stops upon encountering whitespace or when maxchars is reached.
167 Args:
168 stream: The data stream from which was read.
169 maxchars: The maximum number of bytes returned; by default unlimited.
171 Returns:
172 The data which was read.
174 """
175 txt = b""
176 while True:
177 tok = stream.read(1)
178 if tok.isspace() or not tok:
179 break
180 txt += tok
181 if len(txt) == maxchars:
182 break
183 return 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)
237def skip_over_comment(stream: StreamType) -> None:
238 tok = stream.read(1)
239 stream.seek(-1, 1)
240 if tok == b"%":
241 while tok not in (b"\n", b"\r"):
242 tok = stream.read(1)
243 if tok == b"":
244 raise PdfStreamError("File ended unexpectedly.")
247def read_until_regex(*, stream: StreamType, regex: Pattern[bytes], length: int = sys.maxsize) -> bytes:
248 """
249 Read until the regular expression pattern matched (ignore the match).
250 Treats EOF on the underlying stream as the end of the token to be matched.
252 Args:
253 stream: The stream to read from.
254 regex: The pattern to search for.
255 length: The (approximated) maximum number of bytes to read before raising an exception.
257 Returns:
258 The read bytes.
260 """
261 parts: list[bytes] = []
262 total_length = 0
263 tail = b""
264 chunk_size = 16
265 while True:
266 token = stream.read(chunk_size)
267 if not token:
268 return b"".join(parts)
269 token_length = len(token)
270 if (current_length := total_length + token_length) >= length:
271 raise LimitReachedError(
272 f"Read stream length of {current_length} exceeds maximum allowed length of {length}."
273 )
275 # Search overlap of previous tail + new chunk to catch
276 # multi-byte regex matches spanning chunk boundaries.
277 current_buffer = tail + token
278 search_match = regex.search(current_buffer)
279 parts.append(token)
280 if search_match is not None:
281 overlap = len(tail)
282 actual_start = total_length - overlap + search_match.start()
283 stream.seek(actual_start - total_length - token_length, 1)
284 return b"".join(parts)[:actual_start]
285 total_length += token_length
287 # Fixed overlap: 16 bytes is sufficient for the short
288 # delimiter patterns used in PDF parsing.
289 tail = token[-16:]
290 if chunk_size < 8192:
291 chunk_size <<= 1
294def read_block_backwards(stream: BinaryStreamType, to_read: int) -> bytes:
295 """
296 Given a stream at position X, read a block of size to_read ending at position X.
298 This changes the stream's position to the beginning of where the block was
299 read.
301 Args:
302 stream:
303 to_read:
305 Returns:
306 The data which was read.
308 """
309 if stream.tell() < to_read:
310 raise PdfStreamError("Could not read malformed PDF file")
311 # Seek to the start of the block we want to read.
312 stream.seek(-to_read, SEEK_CUR)
313 read = stream.read(to_read)
314 # Seek to the start of the block we read after reading it.
315 stream.seek(-to_read, SEEK_CUR)
316 return read
319def read_previous_line(stream: StreamType) -> bytes:
320 """
321 Given a byte stream with current position X, return the previous line.
323 All characters between the first CR/LF byte found before X
324 (or, the start of the file, if no such byte is found) and position X
325 After this call, the stream will be positioned one byte after the
326 first non-CRLF character found beyond the first CR/LF byte before X,
327 or, if no such byte is found, at the beginning of the stream.
329 Args:
330 stream: StreamType:
332 Returns:
333 The data which was read.
335 """
336 line_content = []
337 found_crlf = False
338 if stream.tell() == 0:
339 raise PdfStreamError(STREAM_TRUNCATED_PREMATURELY)
340 while True:
341 to_read = min(DEFAULT_BUFFER_SIZE, stream.tell())
342 if to_read == 0:
343 break
344 # Read the block. After this, our stream will be one
345 # beyond the initial position.
346 block = read_block_backwards(stream, to_read)
347 idx = len(block) - 1
348 if not found_crlf:
349 # We haven't found our first CR/LF yet.
350 # Read off characters until we hit one.
351 while idx >= 0 and block[idx] not in b"\r\n":
352 idx -= 1
353 if idx >= 0:
354 found_crlf = True
355 if found_crlf:
356 # We found our first CR/LF already (on this block or
357 # a previous one).
358 # Our combined line is the remainder of the block
359 # plus any previously read blocks.
360 line_content.append(block[idx + 1 :])
361 # Continue to read off any more CRLF characters.
362 while idx >= 0 and block[idx] in b"\r\n":
363 idx -= 1
364 else:
365 # Didn't find CR/LF yet - add this block to our
366 # previously read blocks and continue.
367 line_content.append(block)
368 if idx >= 0:
369 # We found the next non-CRLF character.
370 # Set the stream position correctly, then break
371 stream.seek(idx + 1, SEEK_CUR)
372 break
373 # Join all the blocks in the line (which are in reverse order)
374 return b"".join(line_content[::-1])
377def matrix_multiply(
378 a: TransformationMatrixType, b: TransformationMatrixType
379) -> TransformationMatrixType:
380 return tuple( # type: ignore[return-value]
381 tuple(sum(float(i) * float(j) for i, j in zip(row, col)) for col in zip(*b))
382 for row in a
383 )
386def mark_location(stream: StreamType) -> None:
387 """Create text file showing current location in context."""
388 # Mainly for debugging
389 radius = 5000
390 stream.seek(-radius, 1)
391 with open("pypdf_pdfLocation.txt", "wb") as output_fh:
392 output_fh.write(stream.read(radius))
393 output_fh.write(b"HERE")
394 output_fh.write(stream.read(radius))
395 stream.seek(-radius, 1)
398def deprecate(msg: str, stacklevel: int = 3) -> None:
399 warnings.warn(msg, DeprecationWarning, stacklevel=stacklevel)
402def deprecation(msg: str) -> NoReturn:
403 raise DeprecationError(msg)
406def deprecate_with_replacement(old_name: str, new_name: str, removed_in: str) -> None:
407 """Issue a warning that a feature will be removed, but has a replacement."""
408 deprecate(
409 f"{old_name} is deprecated and will be removed in pypdf {removed_in}. Use {new_name} instead.",
410 4,
411 )
414def deprecation_with_replacement(old_name: str, new_name: str, removed_in: str) -> NoReturn:
415 """Raise an exception that a feature was already removed, but has a replacement."""
416 deprecation(
417 f"{old_name} is deprecated and was removed in pypdf {removed_in}. Use {new_name} instead."
418 )
421def deprecate_no_replacement(name: str, removed_in: str) -> None:
422 """Issue a warning that a feature will be removed without replacement."""
423 deprecate(f"{name} is deprecated and will be removed in pypdf {removed_in}.", 4)
426def deprecation_no_replacement(name: str, removed_in: str) -> NoReturn:
427 """Raise an exception that a feature was already removed without replacement."""
428 deprecation(f"{name} is deprecated and was removed in pypdf {removed_in}.")
431def logger_error(message: str, *, source: str, **values: Any) -> None:
432 """
433 Use this instead of logger.error directly.
435 That allows people to overwrite it more easily.
437 See the docs on when to use which:
438 https://pypdf.readthedocs.io/en/latest/user/suppress-warnings.html
439 """
440 if values:
441 logging.getLogger(source).error(message, values)
442 else:
443 logging.getLogger(source).error(message)
446def logger_warning(message: str, *, source: str, **values: Any) -> None:
447 """
448 Use this instead of logger.warning directly.
450 That allows people to overwrite it more easily.
452 ## Exception, warnings.warn, logger_warning
453 - Exceptions should be used if the user should write code that deals with
454 an error case, e.g. the PDF being completely broken.
455 - warnings.warn should be used if the user needs to fix their code, e.g.
456 DeprecationWarnings
457 - logger_warning should be used if the user needs to know that an issue was
458 handled by pypdf, e.g. a non-compliant PDF being read in a way that
459 pypdf could apply a robustness fix to still read it. This applies mainly
460 to strict=False mode.
461 """
462 if values:
463 logging.getLogger(source).warning(message, values)
464 else:
465 # Keep parity with logger_error and support plain warning messages.
466 # Passing an empty dict to logging is not equivalent to passing no args:
467 # plain messages would fail while being formatted.
468 logging.getLogger(source).warning(message)
471def rename_kwargs(
472 func_name: str, kwargs: dict[str, Any], aliases: dict[str, str], fail: bool = False
473) -> None:
474 """
475 Helper function to deprecate arguments.
477 Args:
478 func_name: Name of the function to be deprecated
479 kwargs:
480 aliases:
481 fail:
483 """
484 for old_term, new_term in aliases.items():
485 if old_term in kwargs:
486 if fail:
487 raise DeprecationError(
488 f"{old_term} is deprecated as an argument. Use {new_term} instead"
489 )
490 if new_term in kwargs:
491 raise TypeError(
492 f"{func_name} received both {old_term} and {new_term} as "
493 f"an argument. {old_term} is deprecated. "
494 f"Use {new_term} instead."
495 )
496 kwargs[new_term] = kwargs.pop(old_term)
497 warnings.warn(
498 message=(
499 f"{old_term} is deprecated as an argument. Use {new_term} instead"
500 ),
501 category=DeprecationWarning,
502 stacklevel=3,
503 )
506def _human_readable_bytes(bytes: int) -> str:
507 if bytes < 10**3:
508 return f"{bytes} Byte"
509 if bytes < 10**6:
510 return f"{bytes / 10**3:.1f} kB"
511 if bytes < 10**9:
512 return f"{bytes / 10**6:.1f} MB"
513 return f"{bytes / 10**9:.1f} GB"
516# The following class has been copied from Django:
517# https://github.com/django/django/blob/adae619426b6f50046b3daaa744db52989c9d6db/django/utils/functional.py#L51-L65
518# It received some modifications to comply with our own coding standards.
519#
520# Original license:
521#
522# ---------------------------------------------------------------------------------
523# Copyright (c) Django Software Foundation and individual contributors.
524# All rights reserved.
525#
526# Redistribution and use in source and binary forms, with or without modification,
527# are permitted provided that the following conditions are met:
528#
529# 1. Redistributions of source code must retain the above copyright notice,
530# this list of conditions and the following disclaimer.
531#
532# 2. Redistributions in binary form must reproduce the above copyright
533# notice, this list of conditions and the following disclaimer in the
534# documentation and/or other materials provided with the distribution.
535#
536# 3. Neither the name of Django nor the names of its contributors may be used
537# to endorse or promote products derived from this software without
538# specific prior written permission.
539#
540# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
541# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
542# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
543# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
544# ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
545# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
546# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
547# ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
548# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
549# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
550# ---------------------------------------------------------------------------------
551class classproperty: # noqa: N801
552 """
553 Decorator that converts a method with a single cls argument into a property
554 that can be accessed directly from the class.
555 """
557 def __init__(self, method=None) -> None: # type: ignore # noqa: ANN001
558 self.fget = method
560 def __get__(self, instance, cls=None) -> Any: # type: ignore # noqa: ANN001
561 return self.fget(cls)
563 def getter(self, method) -> Self: # type: ignore # noqa: ANN001
564 self.fget = method
565 return self
568@dataclass
569class File:
570 from .generic import IndirectObject # noqa: PLC0415
572 name: str = ""
573 """
574 Filename as identified within the PDF file.
575 """
576 data: bytes = b""
577 """
578 Data as bytes.
579 """
580 indirect_reference: Optional[IndirectObject] = None
581 """
582 Reference to the object storing the stream.
583 """
585 def __str__(self) -> str:
586 return f"{self.__class__.__name__}(name={self.name}, data: {_human_readable_bytes(len(self.data))})"
588 def __repr__(self) -> str:
589 return self.__str__()[:-1] + f", hash: {hash(self.data)})"
592@functools.total_ordering
593class Version:
594 COMPONENT_PATTERN = re.compile(r"^(\d+)(.*)$")
596 def __init__(self, version_str: str) -> None:
597 self.version_str = version_str
598 self.components = self._parse_version(version_str)
600 def _parse_version(self, version_str: str) -> list[tuple[int, str]]:
601 components = version_str.split(".")
602 parsed_components = []
603 for component in components:
604 match = Version.COMPONENT_PATTERN.match(component)
605 if not match:
606 parsed_components.append((0, component))
607 continue
608 integer_prefix = match.group(1)
609 suffix = match.group(2)
610 if integer_prefix is None:
611 integer_prefix = 0
612 parsed_components.append((int(integer_prefix), suffix))
613 return parsed_components
615 def __eq__(self, other: object) -> bool:
616 if not isinstance(other, Version):
617 return False
618 return self.components == other.components
620 def __hash__(self) -> int:
621 # Convert to tuple as lists cannot be hashed.
622 return hash((self.__class__, tuple(self.components)))
624 def __lt__(self, other: Any) -> bool:
625 if not isinstance(other, Version):
626 raise ValueError(f"Version cannot be compared against {type(other)}")
628 for self_component, other_component in zip(self.components, other.components):
629 self_value, self_suffix = self_component
630 other_value, other_suffix = other_component
632 if self_value < other_value:
633 return True
634 if self_value > other_value:
635 return False
637 if self_suffix < other_suffix:
638 return True
639 if self_suffix > other_suffix:
640 return False
642 return len(self.components) < len(other.components)