Response 类¶

def markdown(self, css_selector: Optional[str] = None, main_content_only: bool = False) -> str:
    """Convert the response content to clean Markdown.

    Scripts, styles, and hidden/prompt-injection content are always removed before conversion, which is
    the same cleaning the MCP server does. Requires the "markdownify" package (`pip install "scrapling[rag]"`).

    :param css_selector: CSS selector to convert only the matching elements. All matches are concatenated.
    :param main_content_only: Convert only the content inside the `<body>` tag.
    """
    from scrapling.core.shell import Convertor

    page = (cast(Selector, self.css("body").first) or self) if main_content_only else self
    page = Convertor._sanitize_for_ai(Convertor._strip_noise_tags(page))
    pages = [page] if not css_selector else page.css(css_selector)
    return "\n\n".join(Convertor._convert_to_markdown(element.html_content) for element in pages)

def follow(
    self,
    url: str,
    sid: str = "",
    callback: Callable[["Response"], AsyncGenerator[Union[Dict[str, Any], "Request", None], None]] | None = None,
    priority: int | None = None,
    dont_filter: bool = False,
    meta: dict[str, Any] | None = None,
    referer_flow: bool = True,
    **kwargs: Any,
) -> Any:
    """Create a Request to follow a URL.

    This is a helper method for spiders to easily follow links found in pages.

    **IMPORTANT**: The below arguments if left empty, the corresponding value from the previous request will be used. The only exception is `dont_filter`.

    The session kwargs of the previous request are inherited as well, except `method`, `params`, `data`, `json`, `files`, and `multipart`, so a followed link is a plain GET unless you pass them again.

    :param url: The URL to follow (can be relative, will be joined with current URL)
    :param sid: The session id to use
    :param callback: Spider callback method to use
    :param priority: The priority number to use, the higher the number, the higher priority to be processed first.
    :param dont_filter: If this request has been done before, disable the filter to allow it again.
    :param meta: Additional meta data to included in the request
    :param referer_flow: Enabled by default, set the current response url as referer for the new request url.
    :param kwargs: Additional Request arguments
    :return: Request object ready to be yielded
    """
    from scrapling.spiders import Request

    if not self.request or not isinstance(self.request, Request):
        raise TypeError("This response has no request set yet.")

    # Merge original session kwargs with new kwargs (new takes precedence)
    inherited = {k: v for k, v in self.request._session_kwargs.items() if k not in _NON_INHERITABLE_KWARGS}
    for key in ("headers", "extra_headers"):
        if inherited.get(key):
            inherited[key] = {k: v for k, v in inherited[key].items() if k.lower() != "content-length"}
    session_kwargs = {**inherited, **kwargs}

    if referer_flow:
        # For requests
        headers = session_kwargs.get("headers", {})
        headers["referer"] = self.url
        session_kwargs["headers"] = headers

        # For browsers
        extra_headers = session_kwargs.get("extra_headers", {})
        extra_headers["referer"] = self.url
        session_kwargs["extra_headers"] = extra_headers

        session_kwargs["google_search"] = False

    return Request(
        url=self.urljoin(url),
        sid=sid or self.request.sid,
        callback=callback or self.request.callback,
        priority=priority if priority is not None else self.request.priority,
        dont_filter=dont_filter,
        meta={**(self.meta or {}), **(meta or {})},
        **session_kwargs,
    )

def __str__(self) -> str:
    return f"<{self.status} {self.url}>"

def __getitem__(self, key: str) -> TextHandler:
    if self._is_text_node(self._root):
        raise TypeError("Text nodes do not have attributes")
    return self.attrib[key]

def __contains__(self, key: str) -> bool:
    if self._is_text_node(self._root):
        return False
    return key in self.attrib

def __getstate__(self) -> Any:
    # lxml don't like it :)
    raise TypeError("Can't pickle Selector objects")

def get_all_text(
    self,
    separator: str = "\n",
    strip: bool = False,
    ignore_tags: Tuple = (
        "script",
        "style",
    ),
    valid_values: bool = True,
) -> TextHandler:
    """Get all child strings of this element, concatenated using the given separator.

    :param separator: Strings will be concatenated using this separator.
    :param strip: If True, strings will be stripped before being concatenated.
    :param ignore_tags: A tuple of all tag names you want to ignore
    :param valid_values: If enabled, elements with text-content that is empty or only whitespaces will be ignored

    :return: A TextHandler
    """
    if self._is_text_node(self._root):
        return TextHandler(str(self._root))

    ignored_elements: set[Any] = set()
    if ignore_tags:
        for element in self._root.iter(*ignore_tags):
            if element not in ignored_elements:
                ignored_elements.update(element.iter())

    _all_strings = []

    def append_text(text: str) -> None:
        processed_text = text.strip() if strip else text
        if not valid_values or processed_text.strip():
            _all_strings.append(processed_text)

    def is_visible_text_node(text_node: _ElementUnicodeResult) -> bool:
        parent = text_node.getparent()
        if parent is None:
            return False

        owner = parent.getparent() if text_node.is_tail else parent
        return owner not in ignored_elements

    for text_node in cast(list[_ElementUnicodeResult], _find_all_text_nodes(self._root)):
        text = str(text_node)
        if text and is_visible_text_node(text_node):
            append_text(text)

    return cast(TextHandler, TextHandler(separator).join(_all_strings))

def urljoin(self, relative_url: str) -> str:
    """Join this Selector's url with a relative url to form an absolute full URL."""
    return urljoin(self.url, relative_url)

def prettify(self) -> TextHandler:
    """Return a prettified version of the element's inner html-code"""
    if self._is_text_node(self._root):
        return TextHandler(str(self._root))
    content = tostring(
        self._root,
        encoding=self.encoding,
        pretty_print=True,
        method="html",
        with_tail=False,
    )
    if isinstance(content, bytes):
        content = content.strip().decode(self.encoding)
    return TextHandler(content)

def has_class(self, class_name: str) -> bool:
    """Check if the element has a specific class
    :param class_name: The class name to check for
    :return: True if element has class with that name otherwise False
    """
    if self._is_text_node(self._root):
        return False
    return class_name in self._root.classes

def iterancestors(self) -> Generator["Selector", None, None]:
    """Return a generator that loops over all ancestors of the element, starting with the element's parent."""
    if self._is_text_node(self._root):
        return
    for ancestor in self._root.iterancestors():
        yield self.__element_convertor(ancestor)

def find_ancestor(self, func: Callable[["Selector"], bool]) -> Optional["Selector"]:
    """Loop over all ancestors of the element till one match the passed function
    :param func: A function that takes each ancestor as an argument and returns True/False
    :return: The first ancestor that match the function or ``None`` otherwise.
    """
    for ancestor in self.iterancestors():
        if func(ancestor):
            return ancestor
    return None

def get(self) -> TextHandler:
    """
    Serialize this element to a string.
    For text nodes, returns the text value. For HTML elements, returns the outer HTML.
    """
    if self._is_text_node(self._root):
        return TextHandler(str(self._root))
    return self.html_content

def getall(self) -> TextHandlers:
    """Return a single-element list containing this element's serialized string."""
    return TextHandlers([self.get()])

def __repr__(self) -> str:
    length_limit = 40

    if self._is_text_node(self._root):
        text = str(self._root)
        if len(text) > length_limit:
            text = text[:length_limit].strip() + "..."
        return f"<text='{text}'>"

    content = clean_spaces(self.html_content)
    if len(content) > length_limit:
        content = content[:length_limit].strip() + "..."
    data = f"<data='{content}'"

    if self.parent:
        parent_content = clean_spaces(self.parent.html_content)
        if len(parent_content) > length_limit:
            parent_content = parent_content[:length_limit].strip() + "..."

        data += f" parent='{parent_content}'"

    return data + ">"

def relocate(
    self,
    element: Union[Dict, HtmlElement, "Selector"],
    percentage: int = 40,
    selector_type: bool = False,
) -> Union[List[HtmlElement], "Selectors"]:
    """This function will search again for the element in the page tree, used automatically on page structure change

    :param element: The element we want to relocate in the tree
    :param percentage: The minimum percentage to accept and not going lower than that. Be aware that the percentage
     calculation depends solely on the page structure, so don't play with this number unless you must know
     what you are doing!
    :param selector_type: If True, the return result will be converted to `Selectors` object
    :return: List of pure HTML elements that got the highest matching score or 'Selectors' object
    """
    score_table: Dict[float, List[Any]] = {}
    # Note: `element` will most likely always be a dictionary at this point.
    if isinstance(element, self.__class__):
        element = element._root

    if issubclass(type(element), HtmlElement):
        element = _StorageTools.element_to_dict(element)

    for node in cast(List, _find_all_elements(self._root)):
        # Collect all elements in the page, then for each element get the matching score of it against the node.
        # Hence: the code doesn't stop even if the score was 100%
        # because there might be another element(s) left in page with the same score
        score = self.__calculate_similarity_score(cast(Dict, element), node)
        score_table.setdefault(score, []).append(node)

    if score_table:
        highest_probability = max(score_table.keys())
        if score_table[highest_probability] and highest_probability >= percentage:
            if log.getEffectiveLevel() < 20:
                # No need to execute this part if the logging level is not debugging
                log.debug(f"Highest probability was {highest_probability}%")
                log.debug("Top 5 best matching elements are: ")
                for percent in tuple(sorted(score_table.keys(), reverse=True))[:5]:
                    log.debug(f"{percent} -> {self.__elements_convertor(score_table[percent])}")

            if not selector_type:
                return score_table[highest_probability]
            return self.__elements_convertor(score_table[highest_probability])
        log.warning(
            f"Adaptive relocation found no element above the {percentage}% threshold "
            f"(top score: {highest_probability}%). Lower `percentage` if this is the right element."
        )
    return []

def css(
    self,
    selector: str,
    identifier: str = "",
    adaptive: bool = False,
    auto_save: bool = False,
    percentage: int = 40,
) -> "Selectors":
    """Search the current tree with CSS3 selectors

    **Important:
    It's recommended to use the identifier argument if you plan to use a different selector later
    and want to relocate the same element(s)**

    :param selector: The CSS3 selector to be used.
    :param adaptive: Enabled will make the function try to relocate the element if it was 'saved' before
    :param identifier: A string that will be used to save/retrieve element's data in adaptive,
     otherwise the selector will be used.
    :param auto_save: Automatically save new elements for `adaptive` later
    :param percentage: The minimum percentage to accept while `adaptive` is working and not going lower than that.
     Be aware that the percentage calculation depends solely on the page structure, so don't play with this
     number unless you must know what you are doing!

    :return: `Selectors` class.
    """
    if self._is_text_node(self._root):
        return Selectors()

    try:
        if not self.__adaptive_enabled or "," not in selector:
            # No need to split selectors in this case, let's save some CPU cycles :)
            xpath_selector = _css_to_xpath(selector)
            return self.xpath(
                xpath_selector,
                identifier or selector,
                adaptive,
                auto_save,
                percentage,
            )

        results = Selectors()
        for single_selector in split_selectors(selector):
            # I'm doing this only so the `save` function saves data correctly for combined selectors
            # Like using the ',' to combine two different selectors that point to different elements.
            xpath_selector = _css_to_xpath(single_selector.canonical())
            results += self.xpath(
                xpath_selector,
                identifier or single_selector.canonical(),
                adaptive,
                auto_save,
                percentage,
            )

        return Selectors(results)
    except (
        SelectorError,
        SelectorSyntaxError,
    ) as e:
        raise SelectorSyntaxError(f"Invalid CSS selector '{selector}': {str(e)}") from e

def xpath(
    self,
    selector: str,
    identifier: str = "",
    adaptive: bool = False,
    auto_save: bool = False,
    percentage: int = 40,
    **kwargs: Any,
) -> "Selectors":
    """Search the current tree with XPath selectors

    **Important:
    It's recommended to use the identifier argument if you plan to use a different selector later
    and want to relocate the same element(s)**

     Note: **Additional keyword arguments will be passed as XPath variables in the XPath expression!**

    :param selector: The XPath selector to be used.
    :param adaptive: Enabled will make the function try to relocate the element if it was 'saved' before
    :param identifier: A string that will be used to save/retrieve element's data in adaptive,
     otherwise the selector will be used.
    :param auto_save: Automatically save new elements for `adaptive` later
    :param percentage: The minimum percentage to accept while `adaptive` is working and not going lower than that.
     Be aware that the percentage calculation depends solely on the page structure, so don't play with this
     number unless you must know what you are doing!

    :return: `Selectors` class.
    """
    if self._is_text_node(self._root):
        return Selectors()

    try:
        elements = self._root.xpath(selector, **kwargs)
        # string(), count(), boolean() and the like return one value instead of a list of nodes
        if not isinstance(elements, list):
            value = ("1" if elements else "0") if isinstance(elements, bool) else str(elements)
            return self.__elements_convertor([_ElementUnicodeResult(value)])

        if elements:
            if not self.__adaptive_enabled and auto_save:
                log.warning(
                    "Argument `auto_save` will be ignored because `adaptive` wasn't enabled on initialization. Check docs for more info."
                )
            elif self.__adaptive_enabled and auto_save:
                self.save(elements[0], identifier or selector)

            return self.__handle_elements(elements)
        elif self.__adaptive_enabled:
            if adaptive:
                element_data = self.retrieve(identifier or selector)
                if element_data:
                    elements = self.relocate(element_data, percentage)
                    if elements and auto_save:
                        self.save(elements[0], identifier or selector)

            return self.__handle_elements(elements)
        else:
            if adaptive:
                log.warning(
                    "Argument `adaptive` will be ignored because `adaptive` wasn't enabled on initialization. Check docs for more info."
                )
            elif auto_save:
                log.warning(
                    "Argument `auto_save` will be ignored because `adaptive` wasn't enabled on initialization. Check docs for more info."
                )

            return self.__handle_elements(elements)

    except (
        SelectorError,
        SelectorSyntaxError,
        XPathError,
        XPathEvalError,
    ) as e:
        raise SelectorSyntaxError(f"Invalid XPath selector: {selector}") from e

def find_all(
    self,
    *args: str | Iterable[str] | Pattern | Callable | Dict[str, str],
    **kwargs: str,
) -> "Selectors":
    """Find elements by filters of your creations for ease.

    :param args: Tag name(s), iterable of tag names, regex patterns, function, or a dictionary of elements' attributes. Leave empty for selecting all.
    :param kwargs: The attributes you want to filter elements based on it.
    :return: The `Selectors` object of the elements or empty list
    """
    if self._is_text_node(self._root):
        return Selectors()

    if not args and not kwargs:
        raise TypeError("You have to pass something to search with, like tag name(s), tag attributes, or both.")

    attributes: Dict[str, Any] = dict()
    tags: Set[str] = set()
    patterns: Set[Pattern] = set()
    results, functions, selectors = Selectors(), [], []

    # Brace yourself for a wonderful journey!
    for arg in args:
        if isinstance(arg, str):
            tags.add(arg)

        elif type(arg) in (list, tuple, set):
            arg = cast(Iterable, arg)  # Type narrowing for type checkers like pyright
            if not all(map(lambda x: isinstance(x, str), arg)):
                raise TypeError("Nested Iterables are not accepted, only iterables of tag names are accepted")
            tags.update(set(arg))

        elif isinstance(arg, dict):
            if not all([(isinstance(k, str) and isinstance(v, str)) for k, v in arg.items()]):
                raise TypeError(
                    "Nested dictionaries are not accepted, only string keys and string values are accepted"
                )
            attributes.update(arg)

        elif isinstance(arg, re_Pattern):
            patterns.add(arg)

        elif callable(arg):
            if len(signature(arg).parameters) > 0:
                functions.append(arg)
            else:
                raise TypeError(
                    "Callable filter function must have at least one argument to take `Selector` objects."
                )

        else:
            raise TypeError(f'Argument with type "{type(arg)}" is not accepted, please read the docs.')

    if not all([(isinstance(k, str) and isinstance(v, str)) for k, v in kwargs.items()]):
        raise TypeError("Only string values are accepted for arguments")

    for attribute_name, value in kwargs.items():
        # Only replace names for kwargs, replacing them in dictionaries doesn't make sense
        attribute_name = _whitelisted.get(attribute_name, attribute_name)
        attributes[attribute_name] = value

    # It's easier and faster to build a selector than traversing the tree
    tags = tags or set("*")
    for tag in tags:
        selector = tag
        for key, value in attributes.items():
            # Not escaping anything with the key so the user can pass patterns like {'href*': '/p/'} or get errors :)
            if key == "class" and (class_names := value.split()):
                # `class` is a space-separated list, so exact-match [class="x"] misses class="x y"; match each name with ~=
                # An empty/blank value has no names to match, so it falls through to the exact match below
                selector += "".join('[class~="{}"]'.format(_escape_css_string(name)) for name in class_names)
            else:
                selector += '[{}="{}"]'.format(key, _escape_css_string(value))
        if selector != "*":
            selectors.append(selector)

    if selectors:
        results = cast(Selectors, self.css(", ".join(selectors)))
        if results:
            # From the results, get the ones that fulfill passed regex patterns
            for pattern in patterns:
                results = results.filter(lambda e: e.text.re(pattern, check_match=True))

            # From the results, get the ones that fulfill passed functions
            for function in functions:
                results = results.filter(function)
    else:
        results = results or self.below_elements
        for pattern in patterns:
            results = results.filter(lambda e: e.text.re(pattern, check_match=True))

        # Collect an element if it fulfills the passed function otherwise
        for function in functions:
            results = results.filter(function)

    return results

def find(
    self,
    *args: str | Iterable[str] | Pattern | Callable | Dict[str, str],
    **kwargs: str,
) -> Optional["Selector"]:
    """Find elements by filters of your creations for ease, then return the first result. Otherwise return `None`.

    :param args: Tag name(s), iterable of tag names, regex patterns, function, or a dictionary of elements' attributes. Leave empty for selecting all.
    :param kwargs: The attributes you want to filter elements based on it.
    :return: The `Selector` object of the element or `None` if the result didn't match
    """
    for element in self.find_all(*args, **kwargs):
        return element
    return None

def save(self, element: HtmlElement, identifier: str) -> None:
    """Saves the element's unique properties to the storage for retrieval and relocation later

    :param element: The element itself that we want to save to storage, it can be a ` Selector ` or pure ` HtmlElement `
    :param identifier: This is the identifier that will be used to retrieve the element later from the storage. See
        the docs for more info.
    """
    if self.__adaptive_enabled and self._storage:
        target_element: Any = element
        if isinstance(target_element, self.__class__):
            target_element = target_element._root

        if self._is_text_node(target_element):
            target_element = target_element.getparent()

        self._storage.save(target_element, identifier)
    else:
        raise RuntimeError(
            "Can't use `adaptive` features while it's disabled globally, you have to start a new class instance."
        )

def retrieve(self, identifier: str) -> Optional[Dict[str, Any]]:
    """Using the identifier, we search the storage and return the unique properties of the element

    :param identifier: This is the identifier that will be used to retrieve the element from the storage. See
        the docs for more info.
    :return: A dictionary of the unique properties
    """
    if self.__adaptive_enabled and self._storage:
        return self._storage.retrieve(identifier)

    raise RuntimeError(
        "Can't use `adaptive` features while it's disabled globally, you have to start a new class instance."
    )

def json(self) -> Dict:
    """Return JSON response if the response is jsonable otherwise throws error"""
    if self._is_text_node(self._root):
        return TextHandler(str(self._root)).json()
    if self._raw_body and isinstance(self._raw_body, (str, bytes)):
        if isinstance(self._raw_body, str):
            return TextHandler(self._raw_body).json()
        else:
            if TYPE_CHECKING:
                assert isinstance(self._raw_body, bytes)
            return TextHandler(self._raw_body.decode()).json()
    elif self.text:
        return self.text.json()
    else:
        return self.get_all_text(strip=True).json()