_base.py 32 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946
  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. import binascii
  28. import codecs
  29. import hashlib
  30. import re
  31. import sys
  32. from collections.abc import Sequence
  33. from math import log10
  34. from struct import iter_unpack
  35. from typing import Any, Callable, ClassVar, Optional, Union, cast
  36. if sys.version_info[:2] >= (3, 10):
  37. from typing import TypeGuard
  38. else:
  39. from typing_extensions import TypeGuard # PEP 647
  40. if sys.version_info >= (3, 11):
  41. from typing import Self
  42. else:
  43. from typing_extensions import Self
  44. from .._codecs import _pdfdoc_encoding_rev
  45. from .._protocols import PdfObjectProtocol, PdfWriterProtocol
  46. from .._utils import (
  47. StreamType,
  48. classproperty,
  49. deprecation_no_replacement,
  50. deprecation_with_replacement,
  51. logger_warning,
  52. read_non_whitespace,
  53. read_until_regex,
  54. )
  55. from ..errors import STREAM_TRUNCATED_PREMATURELY, PdfReadError, PdfStreamError
  56. __author__ = "Mathieu Fenniak"
  57. __author_email__ = "biziqe@mathieu.fenniak.net"
  58. class PdfObject(PdfObjectProtocol):
  59. # function for calculating a hash value
  60. hash_func: Callable[..., "hashlib._Hash"] = hashlib.sha1
  61. indirect_reference: Optional["IndirectObject"]
  62. def hash_bin(self) -> int:
  63. """
  64. Used to detect modified object.
  65. Returns:
  66. Hash considering type and value.
  67. """
  68. raise NotImplementedError(
  69. f"{self.__class__.__name__} does not implement .hash_bin() so far"
  70. )
  71. def hash_value_data(self) -> bytes:
  72. return f"{self}".encode()
  73. def hash_value(self) -> bytes:
  74. return (
  75. f"{self.__class__.__name__}:"
  76. f"{self.hash_func(self.hash_value_data()).hexdigest()}"
  77. ).encode()
  78. def replicate(
  79. self,
  80. pdf_dest: PdfWriterProtocol,
  81. ) -> "PdfObject":
  82. """
  83. Clone object into pdf_dest (PdfWriterProtocol which is an interface for PdfWriter)
  84. without ensuring links. This is used in clone_document_from_root with incremental = True.
  85. Args:
  86. pdf_dest: Target to clone to.
  87. Returns:
  88. The cloned PdfObject
  89. """
  90. return self.clone(pdf_dest)
  91. def clone(
  92. self,
  93. pdf_dest: PdfWriterProtocol,
  94. force_duplicate: bool = False,
  95. ignore_fields: Optional[Sequence[Union[str, int]]] = (),
  96. ) -> "PdfObject":
  97. """
  98. Clone object into pdf_dest (PdfWriterProtocol which is an interface for PdfWriter).
  99. By default, this method will call ``_reference_clone`` (see ``_reference``).
  100. Args:
  101. pdf_dest: Target to clone to.
  102. force_duplicate: By default, if the object has already been cloned and referenced,
  103. the copy will be returned; when ``True``, a new copy will be created.
  104. (Default value = ``False``)
  105. ignore_fields: List/tuple of field names (for dictionaries) that will be ignored
  106. during cloning (applies to children duplication as well). If fields are to be
  107. considered for a limited number of levels, you have to add it as integer, for
  108. example ``[1,"/B","/TOTO"]`` means that ``"/B"`` will be ignored at the first
  109. level only but ``"/TOTO"`` on all levels.
  110. Returns:
  111. The cloned PdfObject
  112. """
  113. raise NotImplementedError(
  114. f"{self.__class__.__name__} does not implement .clone so far"
  115. )
  116. def _reference_clone(
  117. self, clone: Any, pdf_dest: PdfWriterProtocol, force_duplicate: bool = False
  118. ) -> PdfObjectProtocol:
  119. """
  120. Reference the object within the _objects of pdf_dest only if
  121. indirect_reference attribute exists (which means the objects was
  122. already identified in xref/xobjstm) if object has been already
  123. referenced do nothing.
  124. Args:
  125. clone:
  126. pdf_dest:
  127. Returns:
  128. The clone
  129. """
  130. try:
  131. if not force_duplicate and clone.indirect_reference.pdf == pdf_dest:
  132. return clone
  133. except Exception:
  134. pass
  135. # if hasattr(clone, "indirect_reference"):
  136. try:
  137. ind = self.indirect_reference
  138. except AttributeError:
  139. return clone
  140. if (
  141. pdf_dest.incremental
  142. and ind is not None
  143. and ind.pdf == pdf_dest._reader
  144. and ind.idnum <= len(pdf_dest._objects)
  145. ):
  146. i = ind.idnum
  147. else:
  148. i = len(pdf_dest._objects) + 1
  149. if ind is not None:
  150. if id(ind.pdf) not in pdf_dest._id_translated:
  151. pdf_dest._id_translated[id(ind.pdf)] = {}
  152. pdf_dest._id_translated[id(ind.pdf)]["PreventGC"] = ind.pdf # type: ignore[index]
  153. if (
  154. not force_duplicate
  155. and ind.idnum in pdf_dest._id_translated[id(ind.pdf)]
  156. ):
  157. obj = pdf_dest.get_object(
  158. pdf_dest._id_translated[id(ind.pdf)][ind.idnum]
  159. )
  160. assert obj is not None
  161. return obj
  162. pdf_dest._id_translated[id(ind.pdf)][ind.idnum] = i
  163. try:
  164. pdf_dest._objects[i - 1] = clone
  165. except IndexError:
  166. pdf_dest._objects.append(clone)
  167. i = len(pdf_dest._objects)
  168. clone.indirect_reference = IndirectObject(i, 0, pdf_dest)
  169. return clone
  170. def get_object(self) -> Optional["PdfObject"]:
  171. """Resolve indirect references."""
  172. return self
  173. def write_to_stream(
  174. self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
  175. ) -> None:
  176. raise NotImplementedError
  177. class NullObject(PdfObject):
  178. def clone(
  179. self,
  180. pdf_dest: PdfWriterProtocol,
  181. force_duplicate: bool = False,
  182. ignore_fields: Optional[Sequence[Union[str, int]]] = (),
  183. ) -> "NullObject":
  184. """Clone object into pdf_dest."""
  185. return cast(
  186. "NullObject", self._reference_clone(NullObject(), pdf_dest, force_duplicate)
  187. )
  188. def hash_bin(self) -> int:
  189. """
  190. Used to detect modified object.
  191. Returns:
  192. Hash considering type and value.
  193. """
  194. return hash((self.__class__,))
  195. def write_to_stream(
  196. self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
  197. ) -> None:
  198. if encryption_key is not None: # deprecated
  199. deprecation_no_replacement(
  200. "the encryption_key parameter of write_to_stream", "5.0.0"
  201. )
  202. stream.write(b"null")
  203. @staticmethod
  204. def read_from_stream(stream: StreamType) -> "NullObject":
  205. nulltxt = stream.read(4)
  206. if nulltxt != b"null":
  207. raise PdfReadError("Could not read Null object")
  208. return NullObject()
  209. def __repr__(self) -> str:
  210. return "NullObject"
  211. def __eq__(self, other: object) -> bool:
  212. return isinstance(other, NullObject)
  213. def __hash__(self) -> int:
  214. return self.hash_bin()
  215. class BooleanObject(PdfObject):
  216. def __init__(self, value: Any) -> None:
  217. self.value = value
  218. def clone(
  219. self,
  220. pdf_dest: PdfWriterProtocol,
  221. force_duplicate: bool = False,
  222. ignore_fields: Optional[Sequence[Union[str, int]]] = (),
  223. ) -> "BooleanObject":
  224. """Clone object into pdf_dest."""
  225. return cast(
  226. "BooleanObject",
  227. self._reference_clone(BooleanObject(self.value), pdf_dest, force_duplicate),
  228. )
  229. def hash_bin(self) -> int:
  230. """
  231. Used to detect modified object.
  232. Returns:
  233. Hash considering type and value.
  234. """
  235. return hash((self.__class__, self.value))
  236. def __eq__(self, o: object, /) -> bool:
  237. if isinstance(o, BooleanObject):
  238. return self.value == o.value
  239. if isinstance(o, bool):
  240. return self.value == o
  241. return False
  242. def __hash__(self) -> int:
  243. return self.hash_bin()
  244. def __repr__(self) -> str:
  245. return "True" if self.value else "False"
  246. def write_to_stream(
  247. self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
  248. ) -> None:
  249. if encryption_key is not None: # deprecated
  250. deprecation_no_replacement(
  251. "the encryption_key parameter of write_to_stream", "5.0.0"
  252. )
  253. if self.value:
  254. stream.write(b"true")
  255. else:
  256. stream.write(b"false")
  257. @staticmethod
  258. def read_from_stream(stream: StreamType) -> "BooleanObject":
  259. word = stream.read(4)
  260. if word == b"true":
  261. return BooleanObject(True)
  262. if word == b"fals":
  263. stream.read(1)
  264. return BooleanObject(False)
  265. raise PdfReadError("Could not read Boolean object")
  266. class IndirectObject(PdfObject):
  267. def __init__(self, idnum: int, generation: int, pdf: Any) -> None: # PdfReader
  268. self.idnum = idnum
  269. self.generation = generation
  270. self.pdf = pdf
  271. def __hash__(self) -> int:
  272. return hash((self.idnum, self.generation, id(self.pdf)))
  273. def hash_bin(self) -> int:
  274. """
  275. Used to detect modified object.
  276. Returns:
  277. Hash considering type and value.
  278. """
  279. return hash((self.__class__, self.idnum, self.generation, id(self.pdf)))
  280. def replicate(
  281. self,
  282. pdf_dest: PdfWriterProtocol,
  283. ) -> "PdfObject":
  284. return IndirectObject(self.idnum, self.generation, pdf_dest)
  285. def clone(
  286. self,
  287. pdf_dest: PdfWriterProtocol,
  288. force_duplicate: bool = False,
  289. ignore_fields: Optional[Sequence[Union[str, int]]] = (),
  290. ) -> "IndirectObject":
  291. """Clone object into pdf_dest."""
  292. if self.pdf == pdf_dest and not force_duplicate:
  293. # Already duplicated and no extra duplication required
  294. return self
  295. if id(self.pdf) not in pdf_dest._id_translated:
  296. pdf_dest._id_translated[id(self.pdf)] = {}
  297. pdf_dest._id_translated[id(self.pdf)]["PreventGC"] = self.pdf # type: ignore[index]
  298. if self.idnum in pdf_dest._id_translated[id(self.pdf)]:
  299. dup = pdf_dest.get_object(pdf_dest._id_translated[id(self.pdf)][self.idnum])
  300. if force_duplicate:
  301. assert dup is not None
  302. assert dup.indirect_reference is not None
  303. idref = dup.indirect_reference
  304. return IndirectObject(idref.idnum, idref.generation, idref.pdf)
  305. else:
  306. obj = self.get_object()
  307. # case observed : a pointed object can not be found
  308. if obj is None:
  309. # this normally
  310. obj = NullObject()
  311. assert isinstance(self, (IndirectObject,))
  312. obj.indirect_reference = self
  313. dup = pdf_dest._add_object(
  314. obj.clone(pdf_dest, force_duplicate, ignore_fields)
  315. )
  316. assert dup is not None, "mypy"
  317. assert dup.indirect_reference is not None, "mypy"
  318. return dup.indirect_reference
  319. @property
  320. def indirect_reference(self) -> "IndirectObject": # type: ignore[override]
  321. return self
  322. def get_object(self) -> Optional["PdfObject"]:
  323. return self.pdf.get_object(self)
  324. def __deepcopy__(self, memo: Any) -> "IndirectObject":
  325. return IndirectObject(self.idnum, self.generation, self.pdf)
  326. def _get_object_with_check(self) -> Optional["PdfObject"]:
  327. o = self.get_object()
  328. # the check is done here to not slow down get_object()
  329. if isinstance(o, IndirectObject):
  330. raise PdfStreamError(
  331. f"{self.__repr__()} references an IndirectObject {o.__repr__()}"
  332. )
  333. return o
  334. def __getattr__(self, name: str) -> Any:
  335. # Attribute not found in object: look in pointed object
  336. try:
  337. return getattr(self._get_object_with_check(), name)
  338. except AttributeError:
  339. raise AttributeError(
  340. f"No attribute {name} found in IndirectObject or pointed object"
  341. )
  342. def __getitem__(self, key: Any) -> Any:
  343. # items should be extracted from pointed Object
  344. return self._get_object_with_check()[key] # type: ignore
  345. def __contains__(self, key: Any) -> bool:
  346. return key in self._get_object_with_check() # type: ignore
  347. def __iter__(self) -> Any:
  348. return self._get_object_with_check().__iter__() # type: ignore
  349. def __float__(self) -> str:
  350. # in this case we are looking for the pointed data
  351. return self.get_object().__float__() # type: ignore
  352. def __int__(self) -> int:
  353. # in this case we are looking for the pointed data
  354. return self.get_object().__int__() # type: ignore
  355. def __str__(self) -> str:
  356. # in this case we are looking for the pointed data
  357. return self.get_object().__str__()
  358. def __repr__(self) -> str:
  359. return f"IndirectObject({self.idnum!r}, {self.generation!r}, {id(self.pdf)})"
  360. def __eq__(self, other: object) -> bool:
  361. return (
  362. other is not None
  363. and isinstance(other, IndirectObject)
  364. and self.idnum == other.idnum
  365. and self.generation == other.generation
  366. and self.pdf is other.pdf
  367. )
  368. def __ne__(self, other: object) -> bool:
  369. return not self.__eq__(other)
  370. def write_to_stream(
  371. self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
  372. ) -> None:
  373. if encryption_key is not None: # deprecated
  374. deprecation_no_replacement(
  375. "the encryption_key parameter of write_to_stream", "5.0.0"
  376. )
  377. stream.write(f"{self.idnum} {self.generation} R".encode())
  378. @staticmethod
  379. def read_from_stream(stream: StreamType, pdf: Any) -> "IndirectObject": # PdfReader
  380. idnum = b""
  381. while True:
  382. tok = stream.read(1)
  383. if not tok:
  384. raise PdfStreamError(STREAM_TRUNCATED_PREMATURELY)
  385. if tok.isspace():
  386. break
  387. idnum += tok
  388. generation = b""
  389. while True:
  390. tok = stream.read(1)
  391. if not tok:
  392. raise PdfStreamError(STREAM_TRUNCATED_PREMATURELY)
  393. if tok.isspace():
  394. if not generation:
  395. continue
  396. break
  397. generation += tok
  398. r = read_non_whitespace(stream)
  399. if r != b"R":
  400. raise PdfReadError(
  401. f"Error reading indirect object reference at byte {hex(stream.tell())}"
  402. )
  403. return IndirectObject(int(idnum), int(generation), pdf)
  404. FLOAT_WRITE_PRECISION = 8 # shall be min 5 digits max, allow user adj
  405. class FloatObject(float, PdfObject):
  406. def __new__(
  407. cls, value: Any = "0.0", context: Optional[Any] = None
  408. ) -> Self:
  409. try:
  410. value = float(value)
  411. return float.__new__(cls, value)
  412. except Exception as e:
  413. # If this isn't a valid decimal (happens in malformed PDFs)
  414. # fallback to 0
  415. logger_warning(
  416. f"{e} : FloatObject ({value}) invalid; use 0.0 instead", __name__
  417. )
  418. return float.__new__(cls, 0.0)
  419. def clone(
  420. self,
  421. pdf_dest: Any,
  422. force_duplicate: bool = False,
  423. ignore_fields: Optional[Sequence[Union[str, int]]] = (),
  424. ) -> "FloatObject":
  425. """Clone object into pdf_dest."""
  426. return cast(
  427. "FloatObject",
  428. self._reference_clone(FloatObject(self), pdf_dest, force_duplicate),
  429. )
  430. def hash_bin(self) -> int:
  431. """
  432. Used to detect modified object.
  433. Returns:
  434. Hash considering type and value.
  435. """
  436. return hash((self.__class__, self.as_numeric))
  437. def myrepr(self) -> str:
  438. if self == 0:
  439. return "0.0"
  440. nb = FLOAT_WRITE_PRECISION - int(log10(abs(self)))
  441. return f"{self:.{max(1, nb)}f}".rstrip("0").rstrip(".")
  442. def __repr__(self) -> str:
  443. return self.myrepr() # repr(float(self))
  444. def as_numeric(self) -> float:
  445. return float(self)
  446. def write_to_stream(
  447. self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
  448. ) -> None:
  449. if encryption_key is not None: # deprecated
  450. deprecation_no_replacement(
  451. "the encryption_key parameter of write_to_stream", "5.0.0"
  452. )
  453. stream.write(self.myrepr().encode("utf8"))
  454. class NumberObject(int, PdfObject):
  455. NumberPattern = re.compile(b"[^+-.0-9]")
  456. def __new__(cls, value: Any) -> Self:
  457. try:
  458. return int.__new__(cls, int(value))
  459. except ValueError:
  460. logger_warning(f"NumberObject({value}) invalid; use 0 instead", __name__)
  461. return int.__new__(cls, 0)
  462. def clone(
  463. self,
  464. pdf_dest: Any,
  465. force_duplicate: bool = False,
  466. ignore_fields: Optional[Sequence[Union[str, int]]] = (),
  467. ) -> "NumberObject":
  468. """Clone object into pdf_dest."""
  469. return cast(
  470. "NumberObject",
  471. self._reference_clone(NumberObject(self), pdf_dest, force_duplicate),
  472. )
  473. def hash_bin(self) -> int:
  474. """
  475. Used to detect modified object.
  476. Returns:
  477. Hash considering type and value.
  478. """
  479. return hash((self.__class__, self.as_numeric()))
  480. def as_numeric(self) -> int:
  481. return int(repr(self).encode("utf8"))
  482. def write_to_stream(
  483. self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
  484. ) -> None:
  485. if encryption_key is not None: # deprecated
  486. deprecation_no_replacement(
  487. "the encryption_key parameter of write_to_stream", "5.0.0"
  488. )
  489. stream.write(repr(self).encode("utf8"))
  490. @staticmethod
  491. def read_from_stream(stream: StreamType) -> Union["NumberObject", "FloatObject"]:
  492. num = read_until_regex(stream, NumberObject.NumberPattern)
  493. if b"." in num:
  494. return FloatObject(num)
  495. return NumberObject(num)
  496. class ByteStringObject(bytes, PdfObject):
  497. """
  498. Represents a string object where the text encoding could not be determined.
  499. This occurs quite often, as the PDF spec doesn't provide an alternate way to
  500. represent strings -- for example, the encryption data stored in files (like
  501. /O) is clearly not text, but is still stored in a "String" object.
  502. """
  503. def clone(
  504. self,
  505. pdf_dest: Any,
  506. force_duplicate: bool = False,
  507. ignore_fields: Optional[Sequence[Union[str, int]]] = (),
  508. ) -> "ByteStringObject":
  509. """Clone object into pdf_dest."""
  510. return cast(
  511. "ByteStringObject",
  512. self._reference_clone(
  513. ByteStringObject(bytes(self)), pdf_dest, force_duplicate
  514. ),
  515. )
  516. def hash_bin(self) -> int:
  517. """
  518. Used to detect modified object.
  519. Returns:
  520. Hash considering type and value.
  521. """
  522. return hash((self.__class__, bytes(self)))
  523. @property
  524. def original_bytes(self) -> bytes:
  525. """For compatibility with TextStringObject.original_bytes."""
  526. return self
  527. def write_to_stream(
  528. self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
  529. ) -> None:
  530. if encryption_key is not None: # deprecated
  531. deprecation_no_replacement(
  532. "the encryption_key parameter of write_to_stream", "5.0.0"
  533. )
  534. stream.write(b"<")
  535. stream.write(binascii.hexlify(self))
  536. stream.write(b">")
  537. def __str__(self) -> str:
  538. charset_to_try = ["utf-16", *list(NameObject.CHARSETS)]
  539. for enc in charset_to_try:
  540. try:
  541. return self.decode(enc)
  542. except UnicodeDecodeError:
  543. pass
  544. raise PdfReadError("Cannot decode ByteStringObject.")
  545. class TextStringObject(str, PdfObject): # noqa: SLOT000
  546. """
  547. A string object that has been decoded into a real unicode string.
  548. If read from a PDF document, this string appeared to match the
  549. PDFDocEncoding, or contained a UTF-16BE BOM mark to cause UTF-16 decoding
  550. to occur.
  551. """
  552. autodetect_pdfdocencoding: bool
  553. autodetect_utf16: bool
  554. utf16_bom: bytes
  555. _original_bytes: Optional[bytes] = None
  556. def __new__(cls, value: Any) -> Self:
  557. original_bytes = None
  558. if isinstance(value, bytes):
  559. original_bytes = value
  560. value = value.decode("charmap")
  561. text_string_object = str.__new__(cls, value)
  562. text_string_object._original_bytes = original_bytes
  563. text_string_object.autodetect_utf16 = False
  564. text_string_object.autodetect_pdfdocencoding = False
  565. text_string_object.utf16_bom = b""
  566. if original_bytes is not None and original_bytes[:2] in {codecs.BOM_UTF16_LE, codecs.BOM_UTF16_BE}:
  567. # The value of `original_bytes` is only set for inputs being `bytes`.
  568. # If this is UTF-16 data according to the BOM (first two characters),
  569. # perform special handling. All other cases should not need any special conversion
  570. # due to already being a string.
  571. try:
  572. text_string_object = str.__new__(cls, original_bytes.decode("utf-16"))
  573. except UnicodeDecodeError as exception:
  574. logger_warning(
  575. f"{exception!s}\ninitial string:{exception.object!r}",
  576. __name__,
  577. )
  578. text_string_object = str.__new__(cls, exception.object[: exception.start].decode("utf-16"))
  579. text_string_object._original_bytes = original_bytes
  580. text_string_object.autodetect_utf16 = True
  581. text_string_object.utf16_bom = original_bytes[:2]
  582. else:
  583. try:
  584. encode_pdfdocencoding(text_string_object)
  585. text_string_object.autodetect_pdfdocencoding = True
  586. except UnicodeEncodeError:
  587. text_string_object.autodetect_utf16 = True
  588. text_string_object.utf16_bom = codecs.BOM_UTF16_BE
  589. return text_string_object
  590. def clone(
  591. self,
  592. pdf_dest: Any,
  593. force_duplicate: bool = False,
  594. ignore_fields: Optional[Sequence[Union[str, int]]] = (),
  595. ) -> "TextStringObject":
  596. """Clone object into pdf_dest."""
  597. obj = TextStringObject(self)
  598. obj._original_bytes = self._original_bytes
  599. obj.autodetect_pdfdocencoding = self.autodetect_pdfdocencoding
  600. obj.autodetect_utf16 = self.autodetect_utf16
  601. obj.utf16_bom = self.utf16_bom
  602. return cast(
  603. "TextStringObject", self._reference_clone(obj, pdf_dest, force_duplicate)
  604. )
  605. def hash_bin(self) -> int:
  606. """
  607. Used to detect modified object.
  608. Returns:
  609. Hash considering type and value.
  610. """
  611. return hash((self.__class__, self.original_bytes))
  612. @property
  613. def original_bytes(self) -> bytes:
  614. """
  615. It is occasionally possible that a text string object gets created where
  616. a byte string object was expected due to the autodetection mechanism --
  617. if that occurs, this "original_bytes" property can be used to
  618. back-calculate what the original encoded bytes were.
  619. """
  620. if self._original_bytes is not None:
  621. return self._original_bytes
  622. return self.get_original_bytes()
  623. def get_original_bytes(self) -> bytes:
  624. # We're a text string object, but the library is trying to get our raw
  625. # bytes. This can happen if we auto-detected this string as text, but
  626. # we were wrong. It's pretty common. Return the original bytes that
  627. # would have been used to create this object, based upon the autodetect
  628. # method.
  629. if self.autodetect_utf16:
  630. if self.utf16_bom == codecs.BOM_UTF16_LE:
  631. return codecs.BOM_UTF16_LE + self.encode("utf-16le")
  632. if self.utf16_bom == codecs.BOM_UTF16_BE:
  633. return codecs.BOM_UTF16_BE + self.encode("utf-16be")
  634. return self.encode("utf-16be")
  635. if self.autodetect_pdfdocencoding:
  636. return encode_pdfdocencoding(self)
  637. raise Exception("no information about original bytes") # pragma: no cover
  638. def get_encoded_bytes(self) -> bytes:
  639. # Try to write the string out as a PDFDocEncoding encoded string. It's
  640. # nicer to look at in the PDF file. Sadly, we take a performance hit
  641. # here for trying...
  642. try:
  643. if self._original_bytes is not None:
  644. return self._original_bytes
  645. if self.autodetect_utf16:
  646. raise UnicodeEncodeError("", "forced", -1, -1, "")
  647. bytearr = encode_pdfdocencoding(self)
  648. except UnicodeEncodeError:
  649. if self.utf16_bom == codecs.BOM_UTF16_LE:
  650. bytearr = codecs.BOM_UTF16_LE + self.encode("utf-16le")
  651. elif self.utf16_bom == codecs.BOM_UTF16_BE:
  652. bytearr = codecs.BOM_UTF16_BE + self.encode("utf-16be")
  653. else:
  654. bytearr = self.encode("utf-16be")
  655. return bytearr
  656. def write_to_stream(
  657. self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
  658. ) -> None:
  659. if encryption_key is not None: # deprecated
  660. deprecation_no_replacement(
  661. "the encryption_key parameter of write_to_stream", "5.0.0"
  662. )
  663. bytearr = self.get_encoded_bytes()
  664. stream.write(b"(")
  665. for c_ in iter_unpack("c", bytearr):
  666. c = cast(bytes, c_[0])
  667. if not c.isalnum() and c != b" ":
  668. # This:
  669. # stream.write(rf"\{c:0>3o}".encode())
  670. # gives
  671. # https://github.com/davidhalter/parso/issues/207
  672. stream.write(b"\\%03o" % ord(c))
  673. else:
  674. stream.write(c)
  675. stream.write(b")")
  676. class NameObject(str, PdfObject): # noqa: SLOT000
  677. delimiter_pattern = re.compile(rb"\s+|[\(\)<>\[\]{}/%]")
  678. prefix = b"/"
  679. renumber_table: ClassVar[dict[str, bytes]] = {
  680. **{chr(i): f"#{i:02X}".encode() for i in b"#()<>[]{}/%"},
  681. **{chr(i): f"#{i:02X}".encode() for i in range(33)},
  682. }
  683. def clone(
  684. self,
  685. pdf_dest: Any,
  686. force_duplicate: bool = False,
  687. ignore_fields: Optional[Sequence[Union[str, int]]] = (),
  688. ) -> "NameObject":
  689. """Clone object into pdf_dest."""
  690. return cast(
  691. "NameObject",
  692. self._reference_clone(NameObject(self), pdf_dest, force_duplicate),
  693. )
  694. def hash_bin(self) -> int:
  695. """
  696. Used to detect modified object.
  697. Returns:
  698. Hash considering type and value.
  699. """
  700. return hash((self.__class__, self))
  701. def write_to_stream(
  702. self, stream: StreamType, encryption_key: Union[None, str, bytes] = None
  703. ) -> None:
  704. if encryption_key is not None: # deprecated
  705. deprecation_no_replacement(
  706. "the encryption_key parameter of write_to_stream", "5.0.0"
  707. )
  708. stream.write(self.renumber())
  709. def renumber(self) -> bytes:
  710. out = self[0].encode("utf-8")
  711. if out != b"/":
  712. deprecation_no_replacement(
  713. f"Incorrect first char in NameObject, should start with '/': ({self})",
  714. "5.0.0",
  715. )
  716. parts = [out]
  717. for c in self[1:]:
  718. if c > "~":
  719. parts.extend(f"#{x:02X}".encode() for x in c.encode("utf-8"))
  720. else:
  721. try:
  722. parts.append(self.renumber_table[c])
  723. except KeyError:
  724. parts.append(c.encode("utf-8"))
  725. return b"".join(parts)
  726. def _sanitize(self) -> "NameObject":
  727. """
  728. Sanitize the NameObject's name to be a valid PDF name part
  729. (alphanumeric, underscore, hyphen). The _sanitize method replaces
  730. spaces and any non-alphanumeric/non-underscore/non-hyphen with
  731. underscores.
  732. Returns:
  733. NameObject with sanitized name.
  734. """
  735. name = str(self).removeprefix("/")
  736. name = re.sub(r"\ ", "_", name)
  737. name = re.sub(r"[^a-zA-Z0-9_-]", "_", name)
  738. return NameObject("/" + name)
  739. @classproperty
  740. def surfix(cls) -> bytes: # noqa: N805
  741. deprecation_with_replacement("surfix", "prefix", "5.0.0")
  742. return b"/"
  743. @staticmethod
  744. def unnumber(sin: bytes) -> bytes:
  745. result = bytearray()
  746. i = 0
  747. while i < len(sin):
  748. if sin[i:i + 1] == b"#":
  749. try:
  750. result.append(int(sin[i + 1 : i + 3], 16))
  751. i += 3
  752. continue
  753. except (ValueError, IndexError):
  754. # if the 2 characters after # can not be converted to hex
  755. # we change nothing and carry on
  756. pass
  757. result.append(sin[i])
  758. i += 1
  759. return bytes(result)
  760. CHARSETS = ("utf-8", "gbk", "latin1")
  761. @staticmethod
  762. def read_from_stream(stream: StreamType, pdf: Any) -> "NameObject": # PdfReader
  763. name = stream.read(1)
  764. if name != NameObject.prefix:
  765. raise PdfReadError("Name read error")
  766. name += read_until_regex(stream, NameObject.delimiter_pattern)
  767. try:
  768. # Name objects should represent irregular characters
  769. # with a '#' followed by the symbol's hex number
  770. name = NameObject.unnumber(name)
  771. for enc in NameObject.CHARSETS:
  772. try:
  773. ret = name.decode(enc)
  774. return NameObject(ret)
  775. except Exception:
  776. pass
  777. raise UnicodeDecodeError("", name, 0, 0, "Code Not Found")
  778. except (UnicodeEncodeError, UnicodeDecodeError) as e:
  779. if not pdf.strict:
  780. logger_warning(
  781. f"Illegal character in NameObject ({name!r}), "
  782. "you may need to adjust NameObject.CHARSETS",
  783. __name__,
  784. )
  785. return NameObject(name.decode("charmap"))
  786. raise PdfReadError(
  787. f"Illegal character in NameObject ({name!r}). "
  788. "You may need to adjust NameObject.CHARSETS.",
  789. ) from e
  790. def encode_pdfdocencoding(unicode_string: str) -> bytes:
  791. try:
  792. return bytes([_pdfdoc_encoding_rev[k] for k in unicode_string])
  793. except KeyError:
  794. raise UnicodeEncodeError(
  795. "pdfdocencoding",
  796. unicode_string,
  797. -1,
  798. -1,
  799. "does not exist in translation table",
  800. )
  801. def is_null_or_none(x: Any) -> TypeGuard[Union[None, NullObject, IndirectObject]]:
  802. """
  803. Returns:
  804. True if x is None or NullObject.
  805. """
  806. return x is None or (
  807. isinstance(x, PdfObject)
  808. and (x.get_object() is None or isinstance(x.get_object(), NullObject))
  809. )