| # Use of this source code is governed by the MIT license. |
| __license__ = "MIT" |
| |
| __all__ = [ |
| 'LXMLTreeBuilderForXML', |
| 'LXMLTreeBuilder', |
| ] |
| |
| try: |
| from collections.abc import Callable # Python 3.6 |
| except ImportError as e: |
| from collections import Callable |
| |
| from io import BytesIO |
| from io import StringIO |
| from lxml import etree |
| from bs4.element import ( |
| Comment, |
| Doctype, |
| NamespacedAttribute, |
| ProcessingInstruction, |
| XMLProcessingInstruction, |
| ) |
| from bs4.builder import ( |
| DetectsXMLParsedAsHTML, |
| FAST, |
| HTML, |
| HTMLTreeBuilder, |
| PERMISSIVE, |
| ParserRejectedMarkup, |
| TreeBuilder, |
| XML) |
| from bs4.dammit import EncodingDetector |
| |
| LXML = 'lxml' |
| |
| def _invert(d): |
| "Invert a dictionary." |
| return dict((v,k) for k, v in list(d.items())) |
| |
| class LXMLTreeBuilderForXML(TreeBuilder): |
| DEFAULT_PARSER_CLASS = etree.XMLParser |
| |
| is_xml = True |
| processing_instruction_class = XMLProcessingInstruction |
| |
| NAME = "lxml-xml" |
| ALTERNATE_NAMES = ["xml"] |
| |
| # Well, it's permissive by XML parser standards. |
| features = [NAME, LXML, XML, FAST, PERMISSIVE] |
| |
| CHUNK_SIZE = 512 |
| |
| # This namespace mapping is specified in the XML Namespace |
| # standard. |
| DEFAULT_NSMAPS = dict(xml='http://www.w3.org/XML/1998/namespace') |
| |
| DEFAULT_NSMAPS_INVERTED = _invert(DEFAULT_NSMAPS) |
| |
| # NOTE: If we parsed Element objects and looked at .sourceline, |
| # we'd be able to see the line numbers from the original document. |
| # But instead we build an XMLParser or HTMLParser object to serve |
| # as the target of parse messages, and those messages don't include |
| # line numbers. |
| # See: https://bugs.launchpad.net/lxml/+bug/1846906 |
| |
| def initialize_soup(self, soup): |
| """Let the BeautifulSoup object know about the standard namespace |
| mapping. |
| |
| :param soup: A `BeautifulSoup`. |
| """ |
| super(LXMLTreeBuilderForXML, self).initialize_soup(soup) |
| self._register_namespaces(self.DEFAULT_NSMAPS) |
| |
| def _register_namespaces(self, mapping): |
| """Let the BeautifulSoup object know about namespaces encountered |
| while parsing the document. |
| |
| This might be useful later on when creating CSS selectors. |
| |
| This will track (almost) all namespaces, even ones that were |
| only in scope for part of the document. If two namespaces have |
| the same prefix, only the first one encountered will be |
| tracked. Un-prefixed namespaces are not tracked. |
| |
| :param mapping: A dictionary mapping namespace prefixes to URIs. |
| """ |
| for key, value in list(mapping.items()): |
| # This is 'if key' and not 'if key is not None' because we |
| # don't track un-prefixed namespaces. Soupselect will |
| # treat an un-prefixed namespace as the default, which |
| # causes confusion in some cases. |
| if key and key not in self.soup._namespaces: |
| # Let the BeautifulSoup object know about a new namespace. |
| # If there are multiple namespaces defined with the same |
| # prefix, the first one in the document takes precedence. |
| self.soup._namespaces[key] = value |
| |
| def default_parser(self, encoding): |
| """Find the default parser for the given encoding. |
| |
| :param encoding: A string. |
| :return: Either a parser object or a class, which |
| will be instantiated with default arguments. |
| """ |
| if self._default_parser is not None: |
| return self._default_parser |
| return etree.XMLParser( |
| target=self, strip_cdata=False, recover=True, encoding=encoding) |
| |
| def parser_for(self, encoding): |
| """Instantiate an appropriate parser for the given encoding. |
| |
| :param encoding: A string. |
| :return: A parser object such as an `etree.XMLParser`. |
| """ |
| # Use the default parser. |
| parser = self.default_parser(encoding) |
| |
| if isinstance(parser, Callable): |
| # Instantiate the parser with default arguments |
| parser = parser( |
| target=self, strip_cdata=False, recover=True, encoding=encoding |
| ) |
| return parser |
| |
| def __init__(self, parser=None, empty_element_tags=None, **kwargs): |
| # TODO: Issue a warning if parser is present but not a |
| # callable, since that means there's no way to create new |
| # parsers for different encodings. |
| self._default_parser = parser |
| if empty_element_tags is not None: |
| self.empty_element_tags = set(empty_element_tags) |
| self.soup = None |
| self.nsmaps = [self.DEFAULT_NSMAPS_INVERTED] |
| self.active_namespace_prefixes = [dict(self.DEFAULT_NSMAPS)] |
| super(LXMLTreeBuilderForXML, self).__init__(**kwargs) |
| |
| def _getNsTag(self, tag): |
| # Split the namespace URL out of a fully-qualified lxml tag |
| # name. Copied from lxml's src/lxml/sax.py. |
| if tag[0] == '{': |
| return tuple(tag[1:].split('}', 1)) |
| else: |
| return (None, tag) |
| |
| def prepare_markup(self, markup, user_specified_encoding=None, |
| exclude_encodings=None, |
| document_declared_encoding=None): |
| """Run any preliminary steps necessary to make incoming markup |
| acceptable to the parser. |
| |
| lxml really wants to get a bytestring and convert it to |
| Unicode itself. So instead of using UnicodeDammit to convert |
| the bytestring to Unicode using different encodings, this |
| implementation uses EncodingDetector to iterate over the |
| encodings, and tell lxml to try to parse the document as each |
| one in turn. |
| |
| :param markup: Some markup -- hopefully a bytestring. |
| :param user_specified_encoding: The user asked to try this encoding. |
| :param document_declared_encoding: The markup itself claims to be |
| in this encoding. |
| :param exclude_encodings: The user asked _not_ to try any of |
| these encodings. |
| |
| :yield: A series of 4-tuples: |
| (markup, encoding, declared encoding, |
| has undergone character replacement) |
| |
| Each 4-tuple represents a strategy for converting the |
| document to Unicode and parsing it. Each strategy will be tried |
| in turn. |
| """ |
| is_html = not self.is_xml |
| if is_html: |
| self.processing_instruction_class = ProcessingInstruction |
| # We're in HTML mode, so if we're given XML, that's worth |
| # noting. |
| DetectsXMLParsedAsHTML.warn_if_markup_looks_like_xml( |
| markup, stacklevel=3 |
| ) |
| else: |
| self.processing_instruction_class = XMLProcessingInstruction |
| |
| if isinstance(markup, str): |
| # We were given Unicode. Maybe lxml can parse Unicode on |
| # this system? |
| |
| # TODO: This is a workaround for |
| # https://bugs.launchpad.net/lxml/+bug/1948551. |
| # We can remove it once the upstream issue is fixed. |
| if len(markup) > 0 and markup[0] == u'\N{BYTE ORDER MARK}': |
| markup = markup[1:] |
| yield markup, None, document_declared_encoding, False |
| |
| if isinstance(markup, str): |
| # No, apparently not. Convert the Unicode to UTF-8 and |
| # tell lxml to parse it as UTF-8. |
| yield (markup.encode("utf8"), "utf8", |
| document_declared_encoding, False) |
| |
| # This was provided by the end-user; treat it as a known |
| # definite encoding per the algorithm laid out in the HTML5 |
| # spec. (See the EncodingDetector class for details.) |
| known_definite_encodings = [user_specified_encoding] |
| |
| # This was found in the document; treat it as a slightly lower-priority |
| # user encoding. |
| user_encodings = [document_declared_encoding] |
| detector = EncodingDetector( |
| markup, known_definite_encodings=known_definite_encodings, |
| user_encodings=user_encodings, is_html=is_html, |
| exclude_encodings=exclude_encodings |
| ) |
| for encoding in detector.encodings: |
| yield (detector.markup, encoding, document_declared_encoding, False) |
| |
| def feed(self, markup): |
| if isinstance(markup, bytes): |
| markup = BytesIO(markup) |
| elif isinstance(markup, str): |
| markup = StringIO(markup) |
| |
| # Call feed() at least once, even if the markup is empty, |
| # or the parser won't be initialized. |
| data = markup.read(self.CHUNK_SIZE) |
| try: |
| self.parser = self.parser_for(self.soup.original_encoding) |
| self.parser.feed(data) |
| while len(data) != 0: |
| # Now call feed() on the rest of the data, chunk by chunk. |
| data = markup.read(self.CHUNK_SIZE) |
| if len(data) != 0: |
| self.parser.feed(data) |
| self.parser.close() |
| except (UnicodeDecodeError, LookupError, etree.ParserError) as e: |
| raise ParserRejectedMarkup(e) |
| |
| def close(self): |
| self.nsmaps = [self.DEFAULT_NSMAPS_INVERTED] |
| |
| def start(self, name, attrs, nsmap={}): |
| # Make sure attrs is a mutable dict--lxml may send an immutable dictproxy. |
| attrs = dict(attrs) |
| nsprefix = None |
| # Invert each namespace map as it comes in. |
| if len(nsmap) == 0 and len(self.nsmaps) > 1: |
| # There are no new namespaces for this tag, but |
| # non-default namespaces are in play, so we need a |
| # separate tag stack to know when they end. |
| self.nsmaps.append(None) |
| elif len(nsmap) > 0: |
| # A new namespace mapping has come into play. |
| |
| # First, Let the BeautifulSoup object know about it. |
| self._register_namespaces(nsmap) |
| |
| # Then, add it to our running list of inverted namespace |
| # mappings. |
| self.nsmaps.append(_invert(nsmap)) |
| |
| # The currently active namespace prefixes have |
| # changed. Calculate the new mapping so it can be stored |
| # with all Tag objects created while these prefixes are in |
| # scope. |
| current_mapping = dict(self.active_namespace_prefixes[-1]) |
| current_mapping.update(nsmap) |
| |
| # We should not track un-prefixed namespaces as we can only hold one |
| # and it will be recognized as the default namespace by soupsieve, |
| # which may be confusing in some situations. |
| if '' in current_mapping: |
| del current_mapping[''] |
| self.active_namespace_prefixes.append(current_mapping) |
| |
| # Also treat the namespace mapping as a set of attributes on the |
| # tag, so we can recreate it later. |
| attrs = attrs.copy() |
| for prefix, namespace in list(nsmap.items()): |
| attribute = NamespacedAttribute( |
| "xmlns", prefix, "http://www.w3.org/2000/xmlns/") |
| attrs[attribute] = namespace |
| |
| # Namespaces are in play. Find any attributes that came in |
| # from lxml with namespaces attached to their names, and |
| # turn then into NamespacedAttribute objects. |
| new_attrs = {} |
| for attr, value in list(attrs.items()): |
| namespace, attr = self._getNsTag(attr) |
| if namespace is None: |
| new_attrs[attr] = value |
| else: |
| nsprefix = self._prefix_for_namespace(namespace) |
| attr = NamespacedAttribute(nsprefix, attr, namespace) |
| new_attrs[attr] = value |
| attrs = new_attrs |
| |
| namespace, name = self._getNsTag(name) |
| nsprefix = self._prefix_for_namespace(namespace) |
| self.soup.handle_starttag( |
| name, namespace, nsprefix, attrs, |
| namespaces=self.active_namespace_prefixes[-1] |
| ) |
| |
| def _prefix_for_namespace(self, namespace): |
| """Find the currently active prefix for the given namespace.""" |
| if namespace is None: |
| return None |
| for inverted_nsmap in reversed(self.nsmaps): |
| if inverted_nsmap is not None and namespace in inverted_nsmap: |
| return inverted_nsmap[namespace] |
| return None |
| |
| def end(self, name): |
| self.soup.endData() |
| completed_tag = self.soup.tagStack[-1] |
| namespace, name = self._getNsTag(name) |
| nsprefix = None |
| if namespace is not None: |
| for inverted_nsmap in reversed(self.nsmaps): |
| if inverted_nsmap is not None and namespace in inverted_nsmap: |
| nsprefix = inverted_nsmap[namespace] |
| break |
| self.soup.handle_endtag(name, nsprefix) |
| if len(self.nsmaps) > 1: |
| # This tag, or one of its parents, introduced a namespace |
| # mapping, so pop it off the stack. |
| out_of_scope_nsmap = self.nsmaps.pop() |
| |
| if out_of_scope_nsmap is not None: |
| # This tag introduced a namespace mapping which is no |
| # longer in scope. Recalculate the currently active |
| # namespace prefixes. |
| self.active_namespace_prefixes.pop() |
| |
| def pi(self, target, data): |
| self.soup.endData() |
| data = target + ' ' + data |
| self.soup.handle_data(data) |
| self.soup.endData(self.processing_instruction_class) |
| |
| def data(self, content): |
| self.soup.handle_data(content) |
| |
| def doctype(self, name, pubid, system): |
| self.soup.endData() |
| doctype = Doctype.for_name_and_ids(name, pubid, system) |
| self.soup.object_was_parsed(doctype) |
| |
| def comment(self, content): |
| "Handle comments as Comment objects." |
| self.soup.endData() |
| self.soup.handle_data(content) |
| self.soup.endData(Comment) |
| |
| def test_fragment_to_document(self, fragment): |
| """See `TreeBuilder`.""" |
| return '<?xml version="1.0" encoding="utf-8"?>\n%s' % fragment |
| |
| |
| class LXMLTreeBuilder(HTMLTreeBuilder, LXMLTreeBuilderForXML): |
| |
| NAME = LXML |
| ALTERNATE_NAMES = ["lxml-html"] |
| |
| features = ALTERNATE_NAMES + [NAME, HTML, FAST, PERMISSIVE] |
| is_xml = False |
| processing_instruction_class = ProcessingInstruction |
| |
| def default_parser(self, encoding): |
| return etree.HTMLParser |
| |
| def feed(self, markup): |
| encoding = self.soup.original_encoding |
| try: |
| self.parser = self.parser_for(encoding) |
| self.parser.feed(markup) |
| self.parser.close() |
| except (UnicodeDecodeError, LookupError, etree.ParserError) as e: |
| raise ParserRejectedMarkup(e) |
| |
| |
| def test_fragment_to_document(self, fragment): |
| """See `TreeBuilder`.""" |
| return '<html><body>%s</body></html>' % fragment |
