Source code for langchain_community.document_loaders.sitemap

import itertools
import re
from typing import Any, Callable, Generator, Iterable, Iterator, List, Optional, Tuple
from urllib.parse import urlparse

from langchain_core.documents import Document

from langchain_community.document_loaders.web_base import WebBaseLoader


def _default_parsing_function(content: Any) -> str:
    return str(content.get_text())


def _default_meta_function(meta: dict, _content: Any) -> dict:
    return {"source": meta["loc"], **meta}


def _batch_block(iterable: Iterable, size: int) -> Generator[List[dict], None, None]:
    it = iter(iterable)
    while item := list(itertools.islice(it, size)):
        yield item


def _extract_scheme_and_domain(url: str) -> Tuple[str, str]:
    """从给定的URL中提取方案（scheme）+域名（domain）。

参数：
    url（str）：输入的URL。

返回：
    返回一个包含方案和域名的2元组。
"""
    parsed_uri = urlparse(url)
    return parsed_uri.scheme, parsed_uri.netloc


class SitemapLoader(WebBaseLoader):
    """加载站点地图及其URL。

     **安全提示** ：此加载器可用于加载站点地图中指定的所有URL。
        如果恶意行为者获取了站点地图的访问权限，他们可以通过修改站点地图来强制
        服务器加载其他域中的URL。这可能导致服务器端请求伪造（SSRF）攻击；例如，
        攻击者可以强制服务器加载不公开访问的内部服务端点的URL。虽然攻击者
        可能无法立即访问这些数据，但这些数据可能泄漏
        到下游系统中（例如，数据加载器用于为索引加载数据）。

        此加载器是一个网络爬虫，通常不应将网络爬虫部署
        具有对任何内部服务器的网络访问权限。

        控制谁可以提交爬取请求以及爬虫具有什么网络访问权限。

        默认情况下，如果站点地图不是本地文件，则加载器将仅加载与站点地图相同域中的URL。
        如果将 restrict_to_same_domain 设置为 False（不推荐），则可以禁用此功能。

        如果站点地图是本地文件，则默认情况下不会应用此类风险缓解措施。

        使用 filter URLs 参数来限制可以加载的URL。

        请参阅 https://python.langchain.com/docs/security"""

    def __init__(
        self,
        web_path: str,
        filter_urls: Optional[List[str]] = None,
        parsing_function: Optional[Callable] = None,
        blocksize: Optional[int] = None,
        blocknum: int = 0,
        meta_function: Optional[Callable] = None,
        is_local: bool = False,
        continue_on_failure: bool = False,
        restrict_to_same_domain: bool = True,
        **kwargs: Any,
    ):
        """使用网页路径和可选的过滤URL进行初始化。

参数：
    web_path：网站地图的URL。也可以是本地路径。
    filter_urls：正则表达式列表。如果指定，只有匹配其中一个过滤URL的URL将被加载。
        *警告* 过滤URL将被解释为正则表达式。如果不希望其被解释为正则表达式语法，请记得转义特殊字符。例如，`.` 经常出现在URL中，如果要匹配字面上的 `.` 而不是任何字符，则应该转义它。
        当 restrict_to_same_domain 为 True 且网站地图不是本地文件时，restrict_to_same_domain 优先于 filter_urls。
    parsing_function：用于解析 bs4.Soup 输出的函数。
    blocksize：每个块中的站点地图位置数。
    blocknum：应加载的块的编号 - 从零开始计数。默认值：0
    meta_function：用于解析 bs4.Soup 输出的元数据的函数。
        当设置此方法时，还要记得将 metadata["loc"] 复制到 metadata["source"]，如果正在使用此字段。
    is_local：站点地图是否为本地文件。默认值：False
    continue_on_failure：是否在加载URL时发生错误时继续加载站点地图，发出警告而不是引发异常。将此设置为 True 可使加载器更加健壮，但也可能导致数据丢失。默认值：False
    restrict_to_same_domain：是否将加载限制为与站点地图相同域的URL。注意：仅当站点地图不是本地文件时才会应用此选项！
"""

        if blocksize is not None and blocksize < 1:
            raise ValueError("Sitemap blocksize should be at least 1")

        if blocknum < 0:
            raise ValueError("Sitemap blocknum can not be lower then 0")

        try:
            import lxml  # noqa:F401
        except ImportError:
            raise ImportError(
                "lxml package not found, please install it with `pip install lxml`"
            )

        super().__init__(web_paths=[web_path], **kwargs)

        # Define a list of URL patterns (interpreted as regular expressions) that
        # will be allowed to be loaded.
        # restrict_to_same_domain takes precedence over filter_urls when
        # restrict_to_same_domain is True and the sitemap is not a local file.
        self.allow_url_patterns = filter_urls
        self.restrict_to_same_domain = restrict_to_same_domain
        self.parsing_function = parsing_function or _default_parsing_function
        self.meta_function = meta_function or _default_meta_function
        self.blocksize = blocksize
        self.blocknum = blocknum
        self.is_local = is_local
        self.continue_on_failure = continue_on_failure

    def parse_sitemap(self, soup: Any) -> List[dict]:
        """解析站点地图xml并加载到字典列表中。

参数：
    soup：BeautifulSoup对象。

返回：
    字典列表。
"""
        els = []
        for url in soup.find_all("url"):
            loc = url.find("loc")
            if not loc:
                continue

            # Strip leading and trailing whitespace and newlines
            loc_text = loc.text.strip()

            if self.restrict_to_same_domain and not self.is_local:
                if _extract_scheme_and_domain(loc_text) != _extract_scheme_and_domain(
                    self.web_path
                ):
                    continue

            if self.allow_url_patterns and not any(
                re.match(regexp_pattern, loc_text)
                for regexp_pattern in self.allow_url_patterns
            ):
                continue

            els.append(
                {
                    tag: prop.text
                    for tag in ["loc", "lastmod", "changefreq", "priority"]
                    if (prop := url.find(tag))
                }
            )

        for sitemap in soup.find_all("sitemap"):
            loc = sitemap.find("loc")
            if not loc:
                continue
            soup_child = self.scrape_all([loc.text], "xml")[0]

            els.extend(self.parse_sitemap(soup_child))
        return els

    def lazy_load(self) -> Iterator[Document]:
        """加载站点地图。"""
        if self.is_local:
            try:
                import bs4
            except ImportError:
                raise ImportError(
                    "beautifulsoup4 package not found, please install it"
                    " with `pip install beautifulsoup4`"
                )
            fp = open(self.web_path)
            soup = bs4.BeautifulSoup(fp, "xml")
        else:
            soup = self._scrape(self.web_path, parser="xml")

        els = self.parse_sitemap(soup)

        if self.blocksize is not None:
            elblocks = list(_batch_block(els, self.blocksize))
            blockcount = len(elblocks)
            if blockcount - 1 < self.blocknum:
                raise ValueError(
                    "Selected sitemap does not contain enough blocks for given blocknum"
                )
            else:
                els = elblocks[self.blocknum]

        results = self.scrape_all([el["loc"].strip() for el in els if "loc" in el])

        for i, result in enumerate(results):
            yield Document(
                page_content=self.parsing_function(result),
                metadata=self.meta_function(els[i], result),
            )