Source code for parsel.selector

"""XPath and JMESPath selectors based on the lxml and jmespath Python
packages."""

import json
import typing
import warnings
from io import BytesIO
from typing import (
    Any,
    Dict,
    List,
    Mapping,
    Optional,
    Pattern,
    Tuple,
    Type,
    TypeVar,
    Union,
)
from warnings import warn

try:
    from typing import TypedDict  # pylint: disable=ungrouped-imports
except ImportError:  # Python 3.7
    from typing_extensions import TypedDict

import jmespath
from lxml import etree, html
from packaging.version import Version

from .csstranslator import GenericTranslator, HTMLTranslator
from .utils import extract_regex, flatten, iflatten, shorten

if typing.TYPE_CHECKING:
    # both require Python 3.8
    from typing import Literal, SupportsIndex

    # simplified _OutputMethodArg from types-lxml
    _TostringMethodType = Literal[
        "html",
        "xml",
    ]


_SelectorType = TypeVar("_SelectorType", bound="Selector")
_ParserType = Union[etree.XMLParser, etree.HTMLParser]

lxml_version = Version(etree.__version__)
lxml_huge_tree_version = Version("4.2")
LXML_SUPPORTS_HUGE_TREE = lxml_version >= lxml_huge_tree_version


class CannotRemoveElementWithoutRoot(Exception):
    pass


class CannotRemoveElementWithoutParent(Exception):
    pass


class CannotDropElementWithoutParent(CannotRemoveElementWithoutParent):
    pass


class SafeXMLParser(etree.XMLParser):  # type: ignore[type-arg]
    def __init__(self, *args: Any, **kwargs: Any) -> None:
        kwargs.setdefault("resolve_entities", False)
        super().__init__(*args, **kwargs)


class CTGroupValue(TypedDict):
    _parser: Union[Type[etree.XMLParser], Type[html.HTMLParser]]  # type: ignore[type-arg]
    _csstranslator: Union[GenericTranslator, HTMLTranslator]
    _tostring_method: str


_ctgroup: Dict[str, CTGroupValue] = {
    "html": {
        "_parser": html.HTMLParser,
        "_csstranslator": HTMLTranslator(),
        "_tostring_method": "html",
    },
    "xml": {
        "_parser": SafeXMLParser,
        "_csstranslator": GenericTranslator(),
        "_tostring_method": "xml",
    },
}


def _xml_or_html(type: Optional[str]) -> str:
    return "xml" if type == "xml" else "html"


def create_root_node(
    text: str,
    parser_cls: Type[_ParserType],
    base_url: Optional[str] = None,
    huge_tree: bool = LXML_SUPPORTS_HUGE_TREE,
    body: bytes = b"",
    encoding: str = "utf8",
) -> etree._Element:
    """Create root node for text using given parser class."""
    if not text:
        body = body.replace(b"\x00", b"").strip()
    else:
        body = text.strip().replace("\x00", "").encode(encoding) or b"<html/>"

    if huge_tree and LXML_SUPPORTS_HUGE_TREE:
        parser = parser_cls(recover=True, encoding=encoding, huge_tree=True)
        root = etree.fromstring(body, parser=parser, base_url=base_url)
    else:
        parser = parser_cls(recover=True, encoding=encoding)
        root = etree.fromstring(body, parser=parser, base_url=base_url)
        for error in parser.error_log:
            if "use XML_PARSE_HUGE option" in error.message:
                warnings.warn(
                    f"Input data is too big. Upgrade to lxml "
                    f"{lxml_huge_tree_version} or later for huge_tree support."
                )
    if root is None:
        root = etree.fromstring(b"<html/>", parser=parser, base_url=base_url)
    return root


class SelectorList(List[_SelectorType]):
    """
    The :class:`SelectorList` class is a subclass of the builtin ``list``
    class, which provides a few additional methods.
    """

    @typing.overload
    def __getitem__(self, pos: "SupportsIndex") -> _SelectorType:
        pass

    @typing.overload
    def __getitem__(self, pos: slice) -> "SelectorList[_SelectorType]":
        pass

    def __getitem__(
        self, pos: Union["SupportsIndex", slice]
    ) -> Union[_SelectorType, "SelectorList[_SelectorType]"]:
        o = super().__getitem__(pos)
        if isinstance(pos, slice):
            return self.__class__(
                typing.cast("SelectorList[_SelectorType]", o)
            )
        else:
            return typing.cast(_SelectorType, o)

    def __getstate__(self) -> None:
        raise TypeError("can't pickle SelectorList objects")

    def jmespath(
        self, query: str, **kwargs: Any
    ) -> "SelectorList[_SelectorType]":
        """
        Call the ``.jmespath()`` method for each element in this list and return
        their results flattened as another :class:`SelectorList`.

        ``query`` is the same argument as the one in :meth:`Selector.jmespath`.

        Any additional named arguments are passed to the underlying
        ``jmespath.search`` call, e.g.::

            selector.jmespath('author.name', options=jmespath.Options(dict_cls=collections.OrderedDict))
        """
        return self.__class__(
            flatten([x.jmespath(query, **kwargs) for x in self])
        )

[docs] def xpath( self, xpath: str, namespaces: Optional[Mapping[str, str]] = None, **kwargs: Any, ) -> "SelectorList[_SelectorType]": """ Call the ``.xpath()`` method for each element in this list and return their results flattened as another :class:`SelectorList`. ``xpath`` is the same argument as the one in :meth:`Selector.xpath` ``namespaces`` is an optional ``prefix: namespace-uri`` mapping (dict) for additional prefixes to those registered with ``register_namespace(prefix, uri)``. Contrary to ``register_namespace()``, these prefixes are not saved for future calls. Any additional named arguments can be used to pass values for XPath variables in the XPath expression, e.g.:: selector.xpath('//a[href=$url]', url="http://www.example.com") """ return self.__class__( flatten( [x.xpath(xpath, namespaces=namespaces, **kwargs) for x in self] ) )
[docs] def css(self, query: str) -> "SelectorList[_SelectorType]": """ Call the ``.css()`` method for each element in this list and return their results flattened as another :class:`SelectorList`. ``query`` is the same argument as the one in :meth:`Selector.css` """ return self.__class__(flatten([x.css(query) for x in self]))
[docs] def re( self, regex: Union[str, Pattern[str]], replace_entities: bool = True ) -> List[str]: """ Call the ``.re()`` method for each element in this list and return their results flattened, as a list of strings. By default, character entity references are replaced by their corresponding character (except for ``&amp;`` and ``&lt;``. Passing ``replace_entities`` as ``False`` switches off these replacements. """ return flatten( [x.re(regex, replace_entities=replace_entities) for x in self] )
@typing.overload def re_first( self, regex: Union[str, Pattern[str]], default: None = None, replace_entities: bool = True, ) -> Optional[str]: pass @typing.overload def re_first( self, regex: Union[str, Pattern[str]], default: str, replace_entities: bool = True, ) -> str: pass
[docs] def re_first( self, regex: Union[str, Pattern[str]], default: Optional[str] = None, replace_entities: bool = True, ) -> Optional[str]: """ Call the ``.re()`` method for the first element in this list and return the result in an string. If the list is empty or the regex doesn't match anything, return the default value (``None`` if the argument is not provided). By default, character entity references are replaced by their corresponding character (except for ``&amp;`` and ``&lt;``. Passing ``replace_entities`` as ``False`` switches off these replacements. """ for el in iflatten( x.re(regex, replace_entities=replace_entities) for x in self ): return typing.cast(str, el) return default
[docs] def getall(self) -> List[str]: """ Call the ``.get()`` method for each element is this list and return their results flattened, as a list of strings. """ return [x.get() for x in self]
extract = getall @typing.overload def get(self, default: None = None) -> Optional[str]: pass @typing.overload def get(self, default: str) -> str: pass
[docs] def get(self, default: Optional[str] = None) -> Any: """ Return the result of ``.get()`` for the first element in this list. If the list is empty, return the default value. """ for x in self: return x.get() return default
extract_first = get @property def attrib(self) -> Mapping[str, str]: """Return the attributes dictionary for the first element. If the list is empty, return an empty dict. """ for x in self: return x.attrib return {} def remove(self) -> None: # type: ignore[override] """ Remove matched nodes from the parent for each element in this list. """ warn( "Method parsel.selector.SelectorList.remove is deprecated, please use parsel.selector.SelectorList.drop method instead", category=DeprecationWarning, stacklevel=2, ) for x in self: x.remove() def drop(self) -> None: """ Drop matched nodes from the parent for each element in this list. """ for x in self: x.drop() _NOT_SET = object() def _get_root_from_text( text: str, *, type: str, **lxml_kwargs: Any ) -> etree._Element: return create_root_node(text, _ctgroup[type]["_parser"], **lxml_kwargs) def _get_root_and_type_from_bytes( body: bytes, encoding: str, *, input_type: Optional[str], **lxml_kwargs: Any, ) -> Tuple[Any, str]: if input_type == "text": return body.decode(encoding), input_type if encoding == "utf8": try: data = json.load(BytesIO(body)) except ValueError: data = _NOT_SET if data is not _NOT_SET: return data, "json" if input_type == "json": return None, "json" assert input_type in ("html", "xml", None) # nosec type = _xml_or_html(input_type) root = create_root_node( text="", body=body, encoding=encoding, parser_cls=_ctgroup[type]["_parser"], **lxml_kwargs, ) return root, type def _get_root_and_type_from_text( text: str, *, input_type: Optional[str], **lxml_kwargs: Any ) -> Tuple[Any, str]: if input_type == "text": return text, input_type try: data = json.loads(text) except ValueError: data = _NOT_SET if data is not _NOT_SET: return data, "json" if input_type == "json": return None, "json" assert input_type in ("html", "xml", None) # nosec type = _xml_or_html(input_type) root = _get_root_from_text(text, type=type, **lxml_kwargs) return root, type def _get_root_type(root: Any, *, input_type: Optional[str]) -> str: if isinstance(root, etree._Element): # pylint: disable=protected-access if input_type in {"json", "text"}: raise ValueError( f"Selector got an lxml.etree._Element object as root, " f"and {input_type!r} as type." ) return _xml_or_html(input_type) elif isinstance(root, (dict, list)) or _is_valid_json(root): return "json" return input_type or "json" def _is_valid_json(text: str) -> bool: try: json.loads(text) except (TypeError, ValueError): return False else: return True def _load_json_or_none(text: str) -> Any: if isinstance(text, (str, bytes, bytearray)): try: return json.loads(text) except ValueError: return None return None class Selector: """Wrapper for input data in HTML, JSON, or XML format, that allows selecting parts of it using selection expressions. You can write selection expressions in CSS or XPath for HTML and XML inputs, or in JMESPath for JSON inputs. ``text`` is an ``str`` object. ``body`` is a ``bytes`` object. It can be used together with the ``encoding`` argument instead of the ``text`` argument. ``type`` defines the selector type. It can be ``"html"`` (default), ``"json"``, or ``"xml"``. ``base_url`` allows setting a URL for the document. This is needed when looking up external entities with relative paths. See the documentation for :func:`lxml.etree.fromstring` for more information. ``huge_tree`` controls the lxml/libxml2 feature that forbids parsing certain large documents to protect from possible memory exhaustion. The argument is ``True`` by default if the installed lxml version supports it, which disables the protection to allow parsing such documents. Set it to ``False`` if you want to enable the protection. See `this lxml FAQ entry <https://lxml.de/FAQ.html#is-lxml-vulnerable-to-xml-bombs>`_ for more information. """ __slots__ = [ "namespaces", "type", "_expr", "_huge_tree", "root", "_text", "body", "__weakref__", ] _default_namespaces = { "re": "http://exslt.org/regular-expressions", # supported in libxslt: # set:difference # set:has-same-node # set:intersection # set:leading # set:trailing "set": "http://exslt.org/sets", } _lxml_smart_strings = False selectorlist_cls = SelectorList["Selector"] def __init__( self, text: Optional[str] = None, type: Optional[str] = None, body: bytes = b"", encoding: str = "utf8", namespaces: Optional[Mapping[str, str]] = None, root: Optional[Any] = _NOT_SET, base_url: Optional[str] = None, _expr: Optional[str] = None, huge_tree: bool = LXML_SUPPORTS_HUGE_TREE, ) -> None: self.root: Any if type not in ("html", "json", "text", "xml", None): raise ValueError(f"Invalid type: {type}") if text is None and not body and root is _NOT_SET: raise ValueError("Selector needs text, body, or root arguments") if text is not None and not isinstance(text, str): msg = f"text argument should be of type str, got {text.__class__}" raise TypeError(msg) if text is not None: if root is not _NOT_SET: warnings.warn( "Selector got both text and root, root is being ignored.", stacklevel=2, ) if not isinstance(text, str): msg = f"text argument should be of type str, got {text.__class__}" raise TypeError(msg) root, type = _get_root_and_type_from_text( text, input_type=type, base_url=base_url, huge_tree=huge_tree, ) self.root = root self.type = type elif body: if not isinstance(body, bytes): msg = f"body argument should be of type bytes, got {body.__class__}" raise TypeError(msg) root, type = _get_root_and_type_from_bytes( body=body, encoding=encoding, input_type=type, base_url=base_url, huge_tree=huge_tree, ) self.root = root self.type = type elif root is _NOT_SET: raise ValueError("Selector needs text, body, or root arguments") else: self.root = root self.type = _get_root_type(root, input_type=type) self.namespaces = dict(self._default_namespaces) if namespaces is not None: self.namespaces.update(namespaces) self._expr = _expr self._huge_tree = huge_tree self._text = text def __getstate__(self) -> Any: raise TypeError("can't pickle Selector objects") def _get_root( self, text: str = "", base_url: Optional[str] = None, huge_tree: bool = LXML_SUPPORTS_HUGE_TREE, type: Optional[str] = None, body: bytes = b"", encoding: str = "utf8", ) -> etree._Element: return create_root_node( text, body=body, encoding=encoding, parser_cls=_ctgroup[type or self.type]["_parser"], base_url=base_url, huge_tree=huge_tree, ) def jmespath( self: _SelectorType, query: str, **kwargs: Any, ) -> SelectorList[_SelectorType]: """ Find objects matching the JMESPath ``query`` and return the result as a :class:`SelectorList` instance with all elements flattened. List elements implement :class:`Selector` interface too. ``query`` is a string containing the `JMESPath <https://jmespath.org/>`_ query to apply. Any additional named arguments are passed to the underlying ``jmespath.search`` call, e.g.:: selector.jmespath('author.name', options=jmespath.Options(dict_cls=collections.OrderedDict)) """ if self.type == "json": if isinstance(self.root, str): # Selector received a JSON string as root. data = _load_json_or_none(self.root) else: data = self.root else: assert self.type in {"html", "xml"} # nosec data = _load_json_or_none(self.root.text) result = jmespath.search(query, data, **kwargs) if result is None: result = [] elif not isinstance(result, list): result = [result] def make_selector(x: Any) -> _SelectorType: # closure function if isinstance(x, str): return self.__class__(text=x, _expr=query, type="text") else: return self.__class__(root=x, _expr=query) result = [make_selector(x) for x in result] return typing.cast( SelectorList[_SelectorType], self.selectorlist_cls(result) )
[docs] def xpath( self: _SelectorType, query: str, namespaces: Optional[Mapping[str, str]] = None, **kwargs: Any, ) -> SelectorList[_SelectorType]: """ Find nodes matching the xpath ``query`` and return the result as a :class:`SelectorList` instance with all elements flattened. List elements implement :class:`Selector` interface too. ``query`` is a string containing the XPATH query to apply. ``namespaces`` is an optional ``prefix: namespace-uri`` mapping (dict) for additional prefixes to those registered with ``register_namespace(prefix, uri)``. Contrary to ``register_namespace()``, these prefixes are not saved for future calls. Any additional named arguments can be used to pass values for XPath variables in the XPath expression, e.g.:: selector.xpath('//a[href=$url]', url="http://www.example.com") """ if self.type not in ("html", "xml", "text"): raise ValueError( f"Cannot use xpath on a Selector of type {self.type!r}" ) if self.type in ("html", "xml"): try: xpathev = self.root.xpath except AttributeError: return typing.cast( SelectorList[_SelectorType], self.selectorlist_cls([]) ) else: try: xpathev = self._get_root(self._text or "", type="html").xpath except AttributeError: return typing.cast( SelectorList[_SelectorType], self.selectorlist_cls([]) ) nsp = dict(self.namespaces) if namespaces is not None: nsp.update(namespaces) try: result = xpathev( query, namespaces=nsp, smart_strings=self._lxml_smart_strings, **kwargs, ) except etree.XPathError as exc: raise ValueError(f"XPath error: {exc} in {query}") if type(result) is not list: result = [result] result = [ self.__class__( root=x, _expr=query, namespaces=self.namespaces, type=_xml_or_html(self.type), ) for x in result ] return typing.cast( SelectorList[_SelectorType], self.selectorlist_cls(result) )
[docs] def css(self: _SelectorType, query: str) -> SelectorList[_SelectorType]: """ Apply the given CSS selector and return a :class:`SelectorList` instance. ``query`` is a string containing the CSS selector to apply. In the background, CSS queries are translated into XPath queries using `cssselect`_ library and run ``.xpath()`` method. .. _cssselect: https://pypi.python.org/pypi/cssselect/ """ if self.type not in ("html", "xml", "text"): raise ValueError( f"Cannot use css on a Selector of type {self.type!r}" ) return self.xpath(self._css2xpath(query))
def _css2xpath(self, query: str) -> str: type = _xml_or_html(self.type) return _ctgroup[type]["_csstranslator"].css_to_xpath(query)
[docs] def re( self, regex: Union[str, Pattern[str]], replace_entities: bool = True ) -> List[str]: """ Apply the given regex and return a list of strings with the matches. ``regex`` can be either a compiled regular expression or a string which will be compiled to a regular expression using ``re.compile(regex)``. By default, character entity references are replaced by their corresponding character (except for ``&amp;`` and ``&lt;``). Passing ``replace_entities`` as ``False`` switches off these replacements. """ data = self.get() return extract_regex(regex, data, replace_entities=replace_entities)
@typing.overload def re_first( self, regex: Union[str, Pattern[str]], default: None = None, replace_entities: bool = True, ) -> Optional[str]: pass @typing.overload def re_first( self, regex: Union[str, Pattern[str]], default: str, replace_entities: bool = True, ) -> str: pass
[docs] def re_first( self, regex: Union[str, Pattern[str]], default: Optional[str] = None, replace_entities: bool = True, ) -> Optional[str]: """ Apply the given regex and return the first string which matches. If there is no match, return the default value (``None`` if the argument is not provided). By default, character entity references are replaced by their corresponding character (except for ``&amp;`` and ``&lt;``). Passing ``replace_entities`` as ``False`` switches off these replacements. """ return next( iflatten(self.re(regex, replace_entities=replace_entities)), default, )
[docs] def get(self) -> Any: """ Serialize and return the matched nodes in a single string. Percent encoded content is unquoted. """ if self.type in ("text", "json"): return self.root try: return typing.cast( str, etree.tostring( self.root, method=_ctgroup[self.type]["_tostring_method"], encoding="unicode", with_tail=False, ), ) except (AttributeError, TypeError): if self.root is True: return "1" elif self.root is False: return "0" else: return str(self.root)
extract = get
[docs] def getall(self) -> List[str]: """ Serialize and return the matched node in a 1-element list of strings. """ return [self.get()]
[docs] def register_namespace(self, prefix: str, uri: str) -> None: """ Register the given namespace to be used in this :class:`Selector`. Without registering namespaces you can't select or extract data from non-standard namespaces. See :ref:`selector-examples-xml`. """ self.namespaces[prefix] = uri
[docs] def remove_namespaces(self) -> None: """ Remove all namespaces, allowing to traverse the document using namespace-less xpaths. See :ref:`removing-namespaces`. """ for el in self.root.iter("*"): if el.tag.startswith("{"): el.tag = el.tag.split("}", 1)[1] # loop on element attributes also for an in el.attrib: if an.startswith("{"): el.attrib[an.split("}", 1)[1]] = el.attrib.pop(an) # remove namespace declarations etree.cleanup_namespaces(self.root)
def remove(self) -> None: """ Remove matched nodes from the parent element. """ warn( "Method parsel.selector.Selector.remove is deprecated, please use parsel.selector.Selector.drop method instead", category=DeprecationWarning, stacklevel=2, ) try: parent = self.root.getparent() except AttributeError: # 'str' object has no attribute 'getparent' raise CannotRemoveElementWithoutRoot( "The node you're trying to remove has no root, " "are you trying to remove a pseudo-element? " "Try to use 'li' as a selector instead of 'li::text' or " "'//li' instead of '//li/text()', for example." ) try: parent.remove(self.root) except AttributeError: # 'NoneType' object has no attribute 'remove' raise CannotRemoveElementWithoutParent( "The node you're trying to remove has no parent, " "are you trying to remove a root element?" ) def drop(self) -> None: """ Drop matched nodes from the parent element. """ try: parent = self.root.getparent() except AttributeError: # 'str' object has no attribute 'getparent' raise CannotRemoveElementWithoutRoot( "The node you're trying to drop has no root, " "are you trying to drop a pseudo-element? " "Try to use 'li' as a selector instead of 'li::text' or " "'//li' instead of '//li/text()', for example." ) try: if self.type == "xml": if parent is None: raise ValueError("This node has no parent") parent.remove(self.root) else: typing.cast(html.HtmlElement, self.root).drop_tree() except (AttributeError, AssertionError): # 'NoneType' object has no attribute 'drop' raise CannotDropElementWithoutParent( "The node you're trying to remove has no parent, " "are you trying to remove a root element?" ) @property def attrib(self) -> Dict[str, str]: """Return the attributes dictionary for underlying element.""" return dict(self.root.attrib)
[docs] def __bool__(self) -> bool: """ Return ``True`` if there is any real content selected or ``False`` otherwise. In other words, the boolean value of a :class:`Selector` is given by the contents it selects. """ return bool(self.get())
__nonzero__ = __bool__ def __str__(self) -> str: data = repr(shorten(self.get(), width=40)) return f"<{type(self).__name__} query={self._expr!r} data={data}>" __repr__ = __str__