_lxml.py 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501
  1. # encoding: utf-8
  2. from __future__ import annotations
  3. # Use of this source code is governed by the MIT license.
  4. __license__ = "MIT"
  5. __all__ = [
  6. "LXMLTreeBuilderForXML",
  7. "LXMLTreeBuilder",
  8. ]
  9. from typing import (
  10. Any,
  11. Dict,
  12. Iterable,
  13. List,
  14. Optional,
  15. Set,
  16. Tuple,
  17. Type,
  18. TYPE_CHECKING,
  19. Union,
  20. )
  21. from io import BytesIO
  22. from io import StringIO
  23. from typing_extensions import TypeAlias
  24. from lxml import etree # type:ignore
  25. from bs4.element import (
  26. AttributeDict,
  27. XMLAttributeDict,
  28. Comment,
  29. Doctype,
  30. NamespacedAttribute,
  31. ProcessingInstruction,
  32. XMLProcessingInstruction,
  33. )
  34. from bs4.builder import (
  35. DetectsXMLParsedAsHTML,
  36. FAST,
  37. HTML,
  38. HTMLTreeBuilder,
  39. PERMISSIVE,
  40. TreeBuilder,
  41. XML,
  42. )
  43. from bs4.dammit import EncodingDetector
  44. from bs4.exceptions import ParserRejectedMarkup
  45. if TYPE_CHECKING:
  46. from bs4._typing import (
  47. _Encoding,
  48. _Encodings,
  49. _NamespacePrefix,
  50. _NamespaceURL,
  51. _NamespaceMapping,
  52. _InvertedNamespaceMapping,
  53. _RawMarkup,
  54. )
  55. from bs4 import BeautifulSoup
  56. LXML: str = "lxml"
  57. def _invert(d: dict[Any, Any]) -> dict[Any, Any]:
  58. "Invert a dictionary."
  59. return dict((v, k) for k, v in list(d.items()))
  60. _LXMLParser: TypeAlias = Union[etree.XMLParser, etree.HTMLParser]
  61. _ParserOrParserClass: TypeAlias = Union[
  62. _LXMLParser, Type[etree.XMLParser], Type[etree.HTMLParser]
  63. ]
  64. class LXMLTreeBuilderForXML(TreeBuilder):
  65. DEFAULT_PARSER_CLASS: Type[etree.XMLParser] = etree.XMLParser
  66. is_xml: bool = True
  67. #: Set this to true (probably by passing huge_tree=True into the :
  68. #: BeautifulSoup constructor) to enable the lxml feature "disable security
  69. #: restrictions and support very deep trees and very long text
  70. #: content".
  71. huge_tree: bool
  72. processing_instruction_class: Type[ProcessingInstruction]
  73. NAME: str = "lxml-xml"
  74. ALTERNATE_NAMES: Iterable[str] = ["xml"]
  75. # Well, it's permissive by XML parser standards.
  76. features: Iterable[str] = [NAME, LXML, XML, FAST, PERMISSIVE]
  77. CHUNK_SIZE: int = 512
  78. # This namespace mapping is specified in the XML Namespace
  79. # standard.
  80. DEFAULT_NSMAPS: _NamespaceMapping = dict(xml="http://www.w3.org/XML/1998/namespace")
  81. DEFAULT_NSMAPS_INVERTED: _InvertedNamespaceMapping = _invert(DEFAULT_NSMAPS)
  82. nsmaps: List[Optional[_InvertedNamespaceMapping]]
  83. empty_element_tags: Optional[Set[str]]
  84. parser: Any
  85. _default_parser: Optional[etree.XMLParser]
  86. # NOTE: If we parsed Element objects and looked at .sourceline,
  87. # we'd be able to see the line numbers from the original document.
  88. # But instead we build an XMLParser or HTMLParser object to serve
  89. # as the target of parse messages, and those messages don't include
  90. # line numbers.
  91. # See: https://bugs.launchpad.net/lxml/+bug/1846906
  92. def initialize_soup(self, soup: BeautifulSoup) -> None:
  93. """Let the BeautifulSoup object know about the standard namespace
  94. mapping.
  95. :param soup: A `BeautifulSoup`.
  96. """
  97. # Beyond this point, self.soup is set, so we can assume (and
  98. # assert) it's not None whenever necessary.
  99. super(LXMLTreeBuilderForXML, self).initialize_soup(soup)
  100. self._register_namespaces(self.DEFAULT_NSMAPS)
  101. def _register_namespaces(self, mapping: Dict[str, str]) -> None:
  102. """Let the BeautifulSoup object know about namespaces encountered
  103. while parsing the document.
  104. This might be useful later on when creating CSS selectors.
  105. This will track (almost) all namespaces, even ones that were
  106. only in scope for part of the document. If two namespaces have
  107. the same prefix, only the first one encountered will be
  108. tracked. Un-prefixed namespaces are not tracked.
  109. :param mapping: A dictionary mapping namespace prefixes to URIs.
  110. """
  111. assert self.soup is not None
  112. for key, value in list(mapping.items()):
  113. # This is 'if key' and not 'if key is not None' because we
  114. # don't track un-prefixed namespaces. Soupselect will
  115. # treat an un-prefixed namespace as the default, which
  116. # causes confusion in some cases.
  117. if key and key not in self.soup._namespaces:
  118. # Let the BeautifulSoup object know about a new namespace.
  119. # If there are multiple namespaces defined with the same
  120. # prefix, the first one in the document takes precedence.
  121. self.soup._namespaces[key] = value
  122. def default_parser(self, encoding: Optional[_Encoding]) -> _ParserOrParserClass:
  123. """Find the default parser for the given encoding.
  124. :return: Either a parser object or a class, which
  125. will be instantiated with default arguments.
  126. """
  127. if self._default_parser is not None:
  128. return self._default_parser
  129. return self.DEFAULT_PARSER_CLASS(target=self, recover=True, huge_tree=self.huge_tree, encoding=encoding)
  130. def parser_for(self, encoding: Optional[_Encoding]) -> _LXMLParser:
  131. """Instantiate an appropriate parser for the given encoding.
  132. :param encoding: A string.
  133. :return: A parser object such as an `etree.XMLParser`.
  134. """
  135. # Use the default parser.
  136. parser = self.default_parser(encoding)
  137. if callable(parser):
  138. # Instantiate the parser with default arguments
  139. parser = parser(target=self, recover=True, huge_tree=self.huge_tree, encoding=encoding)
  140. return parser
  141. def __init__(
  142. self,
  143. parser: Optional[etree.XMLParser] = None,
  144. empty_element_tags: Optional[Set[str]] = None,
  145. huge_tree: bool = False,
  146. **kwargs: Any,
  147. ):
  148. # TODO: Issue a warning if parser is present but not a
  149. # callable, since that means there's no way to create new
  150. # parsers for different encodings.
  151. self._default_parser = parser
  152. self.soup = None
  153. self.nsmaps = [self.DEFAULT_NSMAPS_INVERTED]
  154. self.active_namespace_prefixes = [dict(self.DEFAULT_NSMAPS)]
  155. if self.is_xml:
  156. self.processing_instruction_class = XMLProcessingInstruction
  157. else:
  158. self.processing_instruction_class = ProcessingInstruction
  159. if "attribute_dict_class" not in kwargs:
  160. kwargs["attribute_dict_class"] = XMLAttributeDict
  161. self.huge_tree = huge_tree
  162. super(LXMLTreeBuilderForXML, self).__init__(**kwargs)
  163. def _getNsTag(self, tag: str) -> Tuple[Optional[str], str]:
  164. # Split the namespace URL out of a fully-qualified lxml tag
  165. # name. Copied from lxml's src/lxml/sax.py.
  166. if tag[0] == "{" and "}" in tag:
  167. namespace, name = tag[1:].split("}", 1)
  168. return (namespace, name)
  169. return (None, tag)
  170. def prepare_markup(
  171. self,
  172. markup: _RawMarkup,
  173. user_specified_encoding: Optional[_Encoding] = None,
  174. document_declared_encoding: Optional[_Encoding] = None,
  175. exclude_encodings: Optional[_Encodings] = None,
  176. ) -> Iterable[
  177. Tuple[Union[str, bytes], Optional[_Encoding], Optional[_Encoding], bool]
  178. ]:
  179. """Run any preliminary steps necessary to make incoming markup
  180. acceptable to the parser.
  181. lxml really wants to get a bytestring and convert it to
  182. Unicode itself. So instead of using UnicodeDammit to convert
  183. the bytestring to Unicode using different encodings, this
  184. implementation uses EncodingDetector to iterate over the
  185. encodings, and tell lxml to try to parse the document as each
  186. one in turn.
  187. :param markup: Some markup -- hopefully a bytestring.
  188. :param user_specified_encoding: The user asked to try this encoding.
  189. :param document_declared_encoding: The markup itself claims to be
  190. in this encoding.
  191. :param exclude_encodings: The user asked _not_ to try any of
  192. these encodings.
  193. :yield: A series of 4-tuples: (markup, encoding, declared encoding,
  194. has undergone character replacement)
  195. Each 4-tuple represents a strategy for converting the
  196. document to Unicode and parsing it. Each strategy will be tried
  197. in turn.
  198. """
  199. if not self.is_xml:
  200. # We're in HTML mode, so if we're given XML, that's worth
  201. # noting.
  202. DetectsXMLParsedAsHTML.warn_if_markup_looks_like_xml(markup, stacklevel=3)
  203. if isinstance(markup, str):
  204. # We were given Unicode. Maybe lxml can parse Unicode on
  205. # this system?
  206. # TODO: This is a workaround for
  207. # https://bugs.launchpad.net/lxml/+bug/1948551.
  208. # We can remove it once the upstream issue is fixed.
  209. if len(markup) > 0 and markup[0] == "\N{BYTE ORDER MARK}":
  210. markup = markup[1:]
  211. yield markup, None, document_declared_encoding, False
  212. if isinstance(markup, str):
  213. # No, apparently not. Convert the Unicode to UTF-8 and
  214. # tell lxml to parse it as UTF-8.
  215. yield (markup.encode("utf8"), "utf8", document_declared_encoding, False)
  216. # Since the document was Unicode in the first place, there
  217. # is no need to try any more strategies; we know this will
  218. # work.
  219. return
  220. known_definite_encodings: List[_Encoding] = []
  221. if user_specified_encoding:
  222. # This was provided by the end-user; treat it as a known
  223. # definite encoding per the algorithm laid out in the
  224. # HTML5 spec. (See the EncodingDetector class for
  225. # details.)
  226. known_definite_encodings.append(user_specified_encoding)
  227. user_encodings: List[_Encoding] = []
  228. if document_declared_encoding:
  229. # This was found in the document; treat it as a slightly
  230. # lower-priority user encoding.
  231. user_encodings.append(document_declared_encoding)
  232. detector = EncodingDetector(
  233. markup,
  234. known_definite_encodings=known_definite_encodings,
  235. user_encodings=user_encodings,
  236. is_html=not self.is_xml,
  237. exclude_encodings=exclude_encodings,
  238. )
  239. for encoding in detector.encodings:
  240. yield (detector.markup, encoding, document_declared_encoding, False)
  241. def feed(self, markup: _RawMarkup) -> None:
  242. io: Union[BytesIO, StringIO]
  243. if isinstance(markup, bytes):
  244. io = BytesIO(markup)
  245. elif isinstance(markup, str):
  246. io = StringIO(markup)
  247. # initialize_soup is called before feed, so we know this
  248. # is not None.
  249. assert self.soup is not None
  250. # Call feed() at least once, even if the markup is empty,
  251. # or the parser won't be initialized.
  252. data = io.read(self.CHUNK_SIZE)
  253. try:
  254. self.parser = self.parser_for(self.soup.original_encoding)
  255. self.parser.feed(data)
  256. while len(data) != 0:
  257. # Now call feed() on the rest of the data, chunk by chunk.
  258. data = io.read(self.CHUNK_SIZE)
  259. if len(data) != 0:
  260. self.parser.feed(data)
  261. self.parser.close()
  262. except (UnicodeDecodeError, LookupError, etree.ParserError) as e:
  263. raise ParserRejectedMarkup(e)
  264. def close(self) -> None:
  265. self.nsmaps = [self.DEFAULT_NSMAPS_INVERTED]
  266. def start(
  267. self,
  268. tag: str | bytes,
  269. attrib: Dict[str | bytes, str | bytes],
  270. nsmap: _NamespaceMapping = {},
  271. ) -> None:
  272. # This is called by lxml code as a result of calling
  273. # BeautifulSoup.feed(), and we know self.soup is set by the time feed()
  274. # is called.
  275. assert self.soup is not None
  276. assert isinstance(tag, str)
  277. # We need to recreate the attribute dict for three
  278. # reasons. First, for type checking, so we can assert there
  279. # are no bytestrings in the keys or values. Second, because we
  280. # need a mutable dict--lxml might send us an immutable
  281. # dictproxy. Third, so we can handle namespaced attribute
  282. # names by converting the keys to NamespacedAttributes.
  283. new_attrib: Dict[Union[str, NamespacedAttribute], str] = (
  284. self.attribute_dict_class()
  285. )
  286. for k, v in attrib.items():
  287. assert isinstance(k, str)
  288. assert isinstance(v, str)
  289. new_attrib[k] = v
  290. nsprefix: Optional[_NamespacePrefix] = None
  291. namespace: Optional[_NamespaceURL] = None
  292. # Invert each namespace map as it comes in.
  293. if len(nsmap) == 0 and len(self.nsmaps) > 1:
  294. # There are no new namespaces for this tag, but
  295. # non-default namespaces are in play, so we need a
  296. # separate tag stack to know when they end.
  297. self.nsmaps.append(None)
  298. elif len(nsmap) > 0:
  299. # A new namespace mapping has come into play.
  300. # First, Let the BeautifulSoup object know about it.
  301. self._register_namespaces(nsmap)
  302. # Then, add it to our running list of inverted namespace
  303. # mappings.
  304. self.nsmaps.append(_invert(nsmap))
  305. # The currently active namespace prefixes have
  306. # changed. Calculate the new mapping so it can be stored
  307. # with all Tag objects created while these prefixes are in
  308. # scope.
  309. current_mapping = dict(self.active_namespace_prefixes[-1])
  310. current_mapping.update(nsmap)
  311. # We should not track un-prefixed namespaces as we can only hold one
  312. # and it will be recognized as the default namespace by soupsieve,
  313. # which may be confusing in some situations.
  314. if "" in current_mapping:
  315. del current_mapping[""]
  316. self.active_namespace_prefixes.append(current_mapping)
  317. # Also treat the namespace mapping as a set of attributes on the
  318. # tag, so we can recreate it later.
  319. for prefix, namespace in list(nsmap.items()):
  320. attribute = NamespacedAttribute(
  321. "xmlns", prefix, "http://www.w3.org/2000/xmlns/"
  322. )
  323. new_attrib[attribute] = namespace
  324. # Namespaces are in play. Find any attributes that came in
  325. # from lxml with namespaces attached to their names, and
  326. # turn then into NamespacedAttribute objects.
  327. final_attrib: AttributeDict = self.attribute_dict_class()
  328. for attr, value in list(new_attrib.items()):
  329. namespace, attr = self._getNsTag(attr)
  330. if namespace is None:
  331. final_attrib[attr] = value
  332. else:
  333. nsprefix = self._prefix_for_namespace(namespace)
  334. attr = NamespacedAttribute(nsprefix, attr, namespace)
  335. final_attrib[attr] = value
  336. namespace, tag = self._getNsTag(tag)
  337. nsprefix = self._prefix_for_namespace(namespace)
  338. self.soup.handle_starttag(
  339. tag,
  340. namespace,
  341. nsprefix,
  342. final_attrib,
  343. namespaces=self.active_namespace_prefixes[-1],
  344. )
  345. def _prefix_for_namespace(
  346. self, namespace: Optional[_NamespaceURL]
  347. ) -> Optional[_NamespacePrefix]:
  348. """Find the currently active prefix for the given namespace."""
  349. if namespace is None:
  350. return None
  351. for inverted_nsmap in reversed(self.nsmaps):
  352. if inverted_nsmap is not None and namespace in inverted_nsmap:
  353. return inverted_nsmap[namespace]
  354. return None
  355. def end(self, tag: str | bytes) -> None:
  356. assert self.soup is not None
  357. assert isinstance(tag, str)
  358. self.soup.endData()
  359. namespace, tag = self._getNsTag(tag)
  360. nsprefix = None
  361. if namespace is not None:
  362. for inverted_nsmap in reversed(self.nsmaps):
  363. if inverted_nsmap is not None and namespace in inverted_nsmap:
  364. nsprefix = inverted_nsmap[namespace]
  365. break
  366. self.soup.handle_endtag(tag, nsprefix)
  367. if len(self.nsmaps) > 1:
  368. # This tag, or one of its parents, introduced a namespace
  369. # mapping, so pop it off the stack.
  370. out_of_scope_nsmap = self.nsmaps.pop()
  371. if out_of_scope_nsmap is not None:
  372. # This tag introduced a namespace mapping which is no
  373. # longer in scope. Recalculate the currently active
  374. # namespace prefixes.
  375. self.active_namespace_prefixes.pop()
  376. def pi(self, target: str, data: str) -> None:
  377. assert self.soup is not None
  378. self.soup.endData()
  379. data = target + " " + data
  380. self.soup.handle_data(data)
  381. self.soup.endData(self.processing_instruction_class)
  382. def data(self, data: str | bytes) -> None:
  383. assert self.soup is not None
  384. assert isinstance(data, str)
  385. self.soup.handle_data(data)
  386. def doctype(self, name: str, pubid: str, system: str) -> None:
  387. assert self.soup is not None
  388. self.soup.endData()
  389. doctype_string = Doctype._string_for_name_and_ids(name, pubid, system)
  390. self.soup.handle_data(doctype_string)
  391. self.soup.endData(containerClass=Doctype)
  392. def comment(self, text: str | bytes) -> None:
  393. "Handle comments as Comment objects."
  394. assert self.soup is not None
  395. assert isinstance(text, str)
  396. self.soup.endData()
  397. self.soup.handle_data(text)
  398. self.soup.endData(Comment)
  399. def test_fragment_to_document(self, fragment: str) -> str:
  400. """See `TreeBuilder`."""
  401. return '<?xml version="1.0" encoding="utf-8"?>\n%s' % fragment
  402. class LXMLTreeBuilder(HTMLTreeBuilder, LXMLTreeBuilderForXML):
  403. NAME: str = LXML
  404. ALTERNATE_NAMES: Iterable[str] = ["lxml-html"]
  405. features: Iterable[str] = list(ALTERNATE_NAMES) + [NAME, HTML, FAST, PERMISSIVE]
  406. is_xml: bool = False
  407. def default_parser(self, encoding: Optional[_Encoding]) -> _ParserOrParserClass:
  408. return etree.HTMLParser
  409. def feed(self, markup: _RawMarkup) -> None:
  410. # We know self.soup is set by the time feed() is called.
  411. assert self.soup is not None
  412. encoding = self.soup.original_encoding
  413. try:
  414. self.parser = self.parser_for(encoding)
  415. self.parser.feed(markup)
  416. self.parser.close()
  417. except (UnicodeDecodeError, LookupError, etree.ParserError) as e:
  418. raise ParserRejectedMarkup(e)
  419. def test_fragment_to_document(self, fragment: str) -> str:
  420. """See `TreeBuilder`."""
  421. return "<html><body>%s</body></html>" % fragment