_utils.py 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645
  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.
  27. """Utility functions for PDF library."""
  28. __author__ = "Mathieu Fenniak"
  29. __author_email__ = "biziqe@mathieu.fenniak.net"
  30. import functools
  31. import logging
  32. import re
  33. import sys
  34. import warnings
  35. from dataclasses import dataclass
  36. from datetime import datetime, timezone
  37. from io import DEFAULT_BUFFER_SIZE
  38. from os import SEEK_CUR
  39. from re import Pattern
  40. from typing import (
  41. IO,
  42. Any,
  43. Optional,
  44. Union,
  45. overload,
  46. )
  47. if sys.version_info[:2] >= (3, 10):
  48. # Python 3.10+: https://www.python.org/dev/peps/pep-0484/
  49. from typing import TypeAlias
  50. else:
  51. from typing_extensions import TypeAlias
  52. if sys.version_info >= (3, 11):
  53. from typing import Self
  54. else:
  55. from typing_extensions import Self
  56. from .errors import (
  57. STREAM_TRUNCATED_PREMATURELY,
  58. DeprecationError,
  59. PdfStreamError,
  60. )
  61. TransformationMatrixType: TypeAlias = tuple[
  62. tuple[float, float, float], tuple[float, float, float], tuple[float, float, float]
  63. ]
  64. CompressedTransformationMatrix: TypeAlias = tuple[
  65. float, float, float, float, float, float
  66. ]
  67. StreamType = IO[Any]
  68. StrByteType = Union[str, StreamType]
  69. def parse_iso8824_date(text: Optional[str]) -> Optional[datetime]:
  70. orgtext = text
  71. if not text:
  72. return None
  73. if text[0].isdigit():
  74. text = "D:" + text
  75. if text.endswith(("Z", "z")):
  76. text += "0000"
  77. text = text.replace("z", "+").replace("Z", "+").replace("'", "")
  78. i = max(text.find("+"), text.find("-"))
  79. if i > 0 and i != len(text) - 5:
  80. text += "00"
  81. for f in (
  82. "D:%Y",
  83. "D:%Y%m",
  84. "D:%Y%m%d",
  85. "D:%Y%m%d%H",
  86. "D:%Y%m%d%H%M",
  87. "D:%Y%m%d%H%M%S",
  88. "D:%Y%m%d%H%M%S%z",
  89. ):
  90. try:
  91. d = datetime.strptime(text, f) # noqa: DTZ007
  92. except ValueError:
  93. continue
  94. else:
  95. if text.endswith("+0000"):
  96. d = d.replace(tzinfo=timezone.utc)
  97. return d
  98. raise ValueError(f"Can not convert date: {orgtext}")
  99. def format_iso8824_date(dt: datetime) -> str:
  100. """
  101. Convert a datetime object to PDF date string format.
  102. Converts datetime to the PDF date format D:YYYYMMDDHHmmSSOHH'mm
  103. as specified in the PDF Reference.
  104. Args:
  105. dt: A datetime object to convert.
  106. Returns:
  107. A date string in PDF format.
  108. """
  109. date_str = dt.strftime("D:%Y%m%d%H%M%S")
  110. if dt.tzinfo is not None:
  111. offset = dt.utcoffset()
  112. assert offset is not None
  113. total_seconds = int(offset.total_seconds())
  114. hours, remainder = divmod(abs(total_seconds), 3600)
  115. minutes = remainder // 60
  116. sign = "+" if total_seconds >= 0 else "-"
  117. date_str += f"{sign}{hours:02d}'{minutes:02d}'"
  118. return date_str
  119. def _get_max_pdf_version_header(header1: str, header2: str) -> str:
  120. versions = (
  121. "%PDF-1.3",
  122. "%PDF-1.4",
  123. "%PDF-1.5",
  124. "%PDF-1.6",
  125. "%PDF-1.7",
  126. "%PDF-2.0",
  127. )
  128. pdf_header_indices = []
  129. if header1 in versions:
  130. pdf_header_indices.append(versions.index(header1))
  131. if header2 in versions:
  132. pdf_header_indices.append(versions.index(header2))
  133. if len(pdf_header_indices) == 0:
  134. raise ValueError(f"Neither {header1!r} nor {header2!r} are proper headers")
  135. return versions[max(pdf_header_indices)]
  136. WHITESPACES = (b"\x00", b"\t", b"\n", b"\f", b"\r", b" ")
  137. WHITESPACES_AS_BYTES = b"".join(WHITESPACES)
  138. WHITESPACES_AS_REGEXP = b"[" + WHITESPACES_AS_BYTES + b"]"
  139. def read_until_whitespace(stream: StreamType, maxchars: Optional[int] = None) -> bytes:
  140. """
  141. Read non-whitespace characters and return them.
  142. Stops upon encountering whitespace or when maxchars is reached.
  143. Args:
  144. stream: The data stream from which was read.
  145. maxchars: The maximum number of bytes returned; by default unlimited.
  146. Returns:
  147. The data which was read.
  148. """
  149. txt = b""
  150. while True:
  151. tok = stream.read(1)
  152. if tok.isspace() or not tok:
  153. break
  154. txt += tok
  155. if len(txt) == maxchars:
  156. break
  157. return txt
  158. def read_non_whitespace(stream: StreamType) -> bytes:
  159. """
  160. Find and read the next non-whitespace character (ignores whitespace).
  161. Args:
  162. stream: The data stream from which was read.
  163. Returns:
  164. The data which was read.
  165. """
  166. tok = stream.read(1)
  167. while tok in WHITESPACES:
  168. tok = stream.read(1)
  169. return tok
  170. def skip_over_whitespace(stream: StreamType) -> bool:
  171. """
  172. Similar to read_non_whitespace, but return a boolean if at least one
  173. whitespace character was read.
  174. Args:
  175. stream: The data stream from which was read.
  176. Returns:
  177. True if one or more whitespace was skipped, otherwise return False.
  178. """
  179. tok = stream.read(1)
  180. cnt = 0
  181. while tok in WHITESPACES:
  182. cnt += 1
  183. tok = stream.read(1)
  184. return cnt > 0
  185. def check_if_whitespace_only(value: bytes) -> bool:
  186. """
  187. Check if the given value consists of whitespace characters only.
  188. Args:
  189. value: The bytes to check.
  190. Returns:
  191. True if the value only has whitespace characters, otherwise return False.
  192. """
  193. return all(b in WHITESPACES_AS_BYTES for b in value)
  194. def skip_over_comment(stream: StreamType) -> None:
  195. tok = stream.read(1)
  196. stream.seek(-1, 1)
  197. if tok == b"%":
  198. while tok not in (b"\n", b"\r"):
  199. tok = stream.read(1)
  200. if tok == b"":
  201. raise PdfStreamError("File ended unexpectedly.")
  202. def read_until_regex(stream: StreamType, regex: Pattern[bytes]) -> bytes:
  203. """
  204. Read until the regular expression pattern matched (ignore the match).
  205. Treats EOF on the underlying stream as the end of the token to be matched.
  206. Args:
  207. regex: re.Pattern
  208. Returns:
  209. The read bytes.
  210. """
  211. parts: list[bytes] = []
  212. total_len = 0
  213. tail = b""
  214. chunk_size = 16
  215. while True:
  216. tok = stream.read(chunk_size)
  217. if not tok:
  218. return b"".join(parts)
  219. # Search overlap of previous tail + new chunk to catch
  220. # multi-byte regex matches spanning chunk boundaries.
  221. buf = tail + tok
  222. m = regex.search(buf)
  223. if m is not None:
  224. overlap = len(tail)
  225. actual_start = total_len - overlap + m.start()
  226. stream.seek(actual_start - total_len - len(tok), 1)
  227. parts.append(tok)
  228. return b"".join(parts)[:actual_start]
  229. parts.append(tok)
  230. total_len += len(tok)
  231. # Fixed overlap: 16 bytes is sufficient for the short
  232. # delimiter patterns used in PDF parsing.
  233. tail = tok[-16:]
  234. if chunk_size < 8192:
  235. chunk_size <<= 1
  236. return b"".join(parts)
  237. def read_block_backwards(stream: StreamType, to_read: int) -> bytes:
  238. """
  239. Given a stream at position X, read a block of size to_read ending at position X.
  240. This changes the stream's position to the beginning of where the block was
  241. read.
  242. Args:
  243. stream:
  244. to_read:
  245. Returns:
  246. The data which was read.
  247. """
  248. if stream.tell() < to_read:
  249. raise PdfStreamError("Could not read malformed PDF file")
  250. # Seek to the start of the block we want to read.
  251. stream.seek(-to_read, SEEK_CUR)
  252. read = stream.read(to_read)
  253. # Seek to the start of the block we read after reading it.
  254. stream.seek(-to_read, SEEK_CUR)
  255. return read
  256. def read_previous_line(stream: StreamType) -> bytes:
  257. """
  258. Given a byte stream with current position X, return the previous line.
  259. All characters between the first CR/LF byte found before X
  260. (or, the start of the file, if no such byte is found) and position X
  261. After this call, the stream will be positioned one byte after the
  262. first non-CRLF character found beyond the first CR/LF byte before X,
  263. or, if no such byte is found, at the beginning of the stream.
  264. Args:
  265. stream: StreamType:
  266. Returns:
  267. The data which was read.
  268. """
  269. line_content = []
  270. found_crlf = False
  271. if stream.tell() == 0:
  272. raise PdfStreamError(STREAM_TRUNCATED_PREMATURELY)
  273. while True:
  274. to_read = min(DEFAULT_BUFFER_SIZE, stream.tell())
  275. if to_read == 0:
  276. break
  277. # Read the block. After this, our stream will be one
  278. # beyond the initial position.
  279. block = read_block_backwards(stream, to_read)
  280. idx = len(block) - 1
  281. if not found_crlf:
  282. # We haven't found our first CR/LF yet.
  283. # Read off characters until we hit one.
  284. while idx >= 0 and block[idx] not in b"\r\n":
  285. idx -= 1
  286. if idx >= 0:
  287. found_crlf = True
  288. if found_crlf:
  289. # We found our first CR/LF already (on this block or
  290. # a previous one).
  291. # Our combined line is the remainder of the block
  292. # plus any previously read blocks.
  293. line_content.append(block[idx + 1 :])
  294. # Continue to read off any more CRLF characters.
  295. while idx >= 0 and block[idx] in b"\r\n":
  296. idx -= 1
  297. else:
  298. # Didn't find CR/LF yet - add this block to our
  299. # previously read blocks and continue.
  300. line_content.append(block)
  301. if idx >= 0:
  302. # We found the next non-CRLF character.
  303. # Set the stream position correctly, then break
  304. stream.seek(idx + 1, SEEK_CUR)
  305. break
  306. # Join all the blocks in the line (which are in reverse order)
  307. return b"".join(line_content[::-1])
  308. def matrix_multiply(
  309. a: TransformationMatrixType, b: TransformationMatrixType
  310. ) -> TransformationMatrixType:
  311. return tuple( # type: ignore[return-value]
  312. tuple(sum(float(i) * float(j) for i, j in zip(row, col)) for col in zip(*b))
  313. for row in a
  314. )
  315. def mark_location(stream: StreamType) -> None:
  316. """Create text file showing current location in context."""
  317. # Mainly for debugging
  318. radius = 5000
  319. stream.seek(-radius, 1)
  320. with open("pypdf_pdfLocation.txt", "wb") as output_fh:
  321. output_fh.write(stream.read(radius))
  322. output_fh.write(b"HERE")
  323. output_fh.write(stream.read(radius))
  324. stream.seek(-radius, 1)
  325. @overload
  326. def ord_(b: str) -> int:
  327. ...
  328. @overload
  329. def ord_(b: bytes) -> bytes:
  330. ...
  331. @overload
  332. def ord_(b: int) -> int:
  333. ...
  334. def ord_(b: Union[int, str, bytes]) -> Union[int, bytes]:
  335. if isinstance(b, str):
  336. return ord(b)
  337. return b
  338. def deprecate(msg: str, stacklevel: int = 3) -> None:
  339. warnings.warn(msg, DeprecationWarning, stacklevel=stacklevel)
  340. def deprecation(msg: str) -> None:
  341. raise DeprecationError(msg)
  342. def deprecate_with_replacement(old_name: str, new_name: str, removed_in: str) -> None:
  343. """Issue a warning that a feature will be removed, but has a replacement."""
  344. deprecate(
  345. f"{old_name} is deprecated and will be removed in pypdf {removed_in}. Use {new_name} instead.",
  346. 4,
  347. )
  348. def deprecation_with_replacement(old_name: str, new_name: str, removed_in: str) -> None:
  349. """Raise an exception that a feature was already removed, but has a replacement."""
  350. deprecation(
  351. f"{old_name} is deprecated and was removed in pypdf {removed_in}. Use {new_name} instead."
  352. )
  353. def deprecate_no_replacement(name: str, removed_in: str) -> None:
  354. """Issue a warning that a feature will be removed without replacement."""
  355. deprecate(f"{name} is deprecated and will be removed in pypdf {removed_in}.", 4)
  356. def deprecation_no_replacement(name: str, removed_in: str) -> None:
  357. """Raise an exception that a feature was already removed without replacement."""
  358. deprecation(f"{name} is deprecated and was removed in pypdf {removed_in}.")
  359. def logger_error(message: str, *, source: str, **values: Any) -> None:
  360. """
  361. Use this instead of logger.error directly.
  362. That allows people to overwrite it more easily.
  363. See the docs on when to use which:
  364. https://pypdf.readthedocs.io/en/latest/user/suppress-warnings.html
  365. """
  366. logging.getLogger(source).error(message, values)
  367. def logger_warning(msg: str, src: str) -> None:
  368. """
  369. Use this instead of logger.warning directly.
  370. That allows people to overwrite it more easily.
  371. ## Exception, warnings.warn, logger_warning
  372. - Exceptions should be used if the user should write code that deals with
  373. an error case, e.g. the PDF being completely broken.
  374. - warnings.warn should be used if the user needs to fix their code, e.g.
  375. DeprecationWarnings
  376. - logger_warning should be used if the user needs to know that an issue was
  377. handled by pypdf, e.g. a non-compliant PDF being read in a way that
  378. pypdf could apply a robustness fix to still read it. This applies mainly
  379. to strict=False mode.
  380. """
  381. logging.getLogger(src).warning(msg)
  382. def rename_kwargs(
  383. func_name: str, kwargs: dict[str, Any], aliases: dict[str, str], fail: bool = False
  384. ) -> None:
  385. """
  386. Helper function to deprecate arguments.
  387. Args:
  388. func_name: Name of the function to be deprecated
  389. kwargs:
  390. aliases:
  391. fail:
  392. """
  393. for old_term, new_term in aliases.items():
  394. if old_term in kwargs:
  395. if fail:
  396. raise DeprecationError(
  397. f"{old_term} is deprecated as an argument. Use {new_term} instead"
  398. )
  399. if new_term in kwargs:
  400. raise TypeError(
  401. f"{func_name} received both {old_term} and {new_term} as "
  402. f"an argument. {old_term} is deprecated. "
  403. f"Use {new_term} instead."
  404. )
  405. kwargs[new_term] = kwargs.pop(old_term)
  406. warnings.warn(
  407. message=(
  408. f"{old_term} is deprecated as an argument. Use {new_term} instead"
  409. ),
  410. category=DeprecationWarning,
  411. stacklevel=3,
  412. )
  413. def _human_readable_bytes(bytes: int) -> str:
  414. if bytes < 10**3:
  415. return f"{bytes} Byte"
  416. if bytes < 10**6:
  417. return f"{bytes / 10**3:.1f} kB"
  418. if bytes < 10**9:
  419. return f"{bytes / 10**6:.1f} MB"
  420. return f"{bytes / 10**9:.1f} GB"
  421. # The following class has been copied from Django:
  422. # https://github.com/django/django/blob/adae619426b6f50046b3daaa744db52989c9d6db/django/utils/functional.py#L51-L65
  423. # It received some modifications to comply with our own coding standards.
  424. #
  425. # Original license:
  426. #
  427. # ---------------------------------------------------------------------------------
  428. # Copyright (c) Django Software Foundation and individual contributors.
  429. # All rights reserved.
  430. #
  431. # Redistribution and use in source and binary forms, with or without modification,
  432. # are permitted provided that the following conditions are met:
  433. #
  434. # 1. Redistributions of source code must retain the above copyright notice,
  435. # this list of conditions and the following disclaimer.
  436. #
  437. # 2. Redistributions in binary form must reproduce the above copyright
  438. # notice, this list of conditions and the following disclaimer in the
  439. # documentation and/or other materials provided with the distribution.
  440. #
  441. # 3. Neither the name of Django nor the names of its contributors may be used
  442. # to endorse or promote products derived from this software without
  443. # specific prior written permission.
  444. #
  445. # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
  446. # ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  447. # WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
  448. # DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
  449. # ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
  450. # (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
  451. # LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
  452. # ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  453. # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
  454. # SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  455. # ---------------------------------------------------------------------------------
  456. class classproperty: # noqa: N801
  457. """
  458. Decorator that converts a method with a single cls argument into a property
  459. that can be accessed directly from the class.
  460. """
  461. def __init__(self, method=None) -> None: # type: ignore # noqa: ANN001
  462. self.fget = method
  463. def __get__(self, instance, cls=None) -> Any: # type: ignore # noqa: ANN001
  464. return self.fget(cls)
  465. def getter(self, method) -> Self: # type: ignore # noqa: ANN001
  466. self.fget = method
  467. return self
  468. @dataclass
  469. class File:
  470. from .generic import IndirectObject # noqa: PLC0415
  471. name: str = ""
  472. """
  473. Filename as identified within the PDF file.
  474. """
  475. data: bytes = b""
  476. """
  477. Data as bytes.
  478. """
  479. indirect_reference: Optional[IndirectObject] = None
  480. """
  481. Reference to the object storing the stream.
  482. """
  483. def __str__(self) -> str:
  484. return f"{self.__class__.__name__}(name={self.name}, data: {_human_readable_bytes(len(self.data))})"
  485. def __repr__(self) -> str:
  486. return self.__str__()[:-1] + f", hash: {hash(self.data)})"
  487. @functools.total_ordering
  488. class Version:
  489. COMPONENT_PATTERN = re.compile(r"^(\d+)(.*)$")
  490. def __init__(self, version_str: str) -> None:
  491. self.version_str = version_str
  492. self.components = self._parse_version(version_str)
  493. def _parse_version(self, version_str: str) -> list[tuple[int, str]]:
  494. components = version_str.split(".")
  495. parsed_components = []
  496. for component in components:
  497. match = Version.COMPONENT_PATTERN.match(component)
  498. if not match:
  499. parsed_components.append((0, component))
  500. continue
  501. integer_prefix = match.group(1)
  502. suffix = match.group(2)
  503. if integer_prefix is None:
  504. integer_prefix = 0
  505. parsed_components.append((int(integer_prefix), suffix))
  506. return parsed_components
  507. def __eq__(self, other: object) -> bool:
  508. if not isinstance(other, Version):
  509. return False
  510. return self.components == other.components
  511. def __hash__(self) -> int:
  512. # Convert to tuple as lists cannot be hashed.
  513. return hash((self.__class__, tuple(self.components)))
  514. def __lt__(self, other: Any) -> bool:
  515. if not isinstance(other, Version):
  516. raise ValueError(f"Version cannot be compared against {type(other)}")
  517. for self_component, other_component in zip(self.components, other.components):
  518. self_value, self_suffix = self_component
  519. other_value, other_suffix = other_component
  520. if self_value < other_value:
  521. return True
  522. if self_value > other_value:
  523. return False
  524. if self_suffix < other_suffix:
  525. return True
  526. if self_suffix > other_suffix:
  527. return False
  528. return len(self.components) < len(other.components)