Source code for langchain_community.document_loaders.rss

import logging
from typing import Any, Iterator, List, Optional, Sequence

from langchain_core.documents import Document

from langchain_community.document_loaders.base import BaseLoader
from langchain_community.document_loaders.news import NewsURLLoader

logger = logging.getLogger(__name__)


class RSSFeedLoader(BaseLoader):
    """从`RSS`源使用`Unstructured`加载新闻文章。

    Args:
        urls: 要加载的RSS源的URL。每个源中的文章都加载到自己的文档中。
        opml: 从中加载源URL的OPML文件。应提供urls或opml中的一个。该值可以是URL字符串，也可以是字节或字符串形式的OPML标记内容。
        continue_on_failure: 如果为True，则即使对特定URL加载失败，也会继续加载文档。
        show_progress_bar: 如果为True，则使用tqdm显示加载进度条。需要安装tqdm，``pip install tqdm``。
        **newsloader_kwargs: 要传递给NewsURLLoader的任何其他命名参数。

    Example:
        .. code-block:: python

            from langchain_community.document_loaders import RSSFeedLoader

            loader = RSSFeedLoader(
                urls=["<url-1>", "<url-2>"],
            )
            docs = loader.load()

    该加载器使用feedparser来解析RSS源。feedparser库默认未安装，因此如果使用此加载器，应安装它：
    https://pythonhosted.org/feedparser/

    如果使用OPML，还应安装listparser：
    https://pythonhosted.org/listparser/

    最后，使用newspaper来处理每篇文章：
    https://newspaper.readthedocs.io/en/latest/"""  # noqa: E501

    def __init__(
        self,
        urls: Optional[Sequence[str]] = None,
        opml: Optional[str] = None,
        continue_on_failure: bool = True,
        show_progress_bar: bool = False,
        **newsloader_kwargs: Any,
    ) -> None:
        """使用URL或OPML进行初始化。"""
        if (urls is None) == (
            opml is None
        ):  # This is True if both are None or neither is None
            raise ValueError(
                "Provide either the urls or the opml argument, but not both."
            )
        self.urls = urls
        self.opml = opml
        self.continue_on_failure = continue_on_failure
        self.show_progress_bar = show_progress_bar
        self.newsloader_kwargs = newsloader_kwargs

    def load(self) -> List[Document]:
        iter = self.lazy_load()
        if self.show_progress_bar:
            try:
                from tqdm import tqdm
            except ImportError as e:
                raise ImportError(
                    "Package tqdm must be installed if show_progress_bar=True. "
                    "Please install with 'pip install tqdm' or set "
                    "show_progress_bar=False."
                ) from e
            iter = tqdm(iter)
        return list(iter)

    @property
    def _get_urls(self) -> Sequence[str]:
        if self.urls:
            return self.urls
        try:
            import listparser
        except ImportError as e:
            raise ImportError(
                "Package listparser must be installed if the opml arg is used. "
                "Please install with 'pip install listparser' or use the "
                "urls arg instead."
            ) from e
        rss = listparser.parse(self.opml)
        return [feed.url for feed in rss.feeds]

    def lazy_load(self) -> Iterator[Document]:
        try:
            import feedparser
        except ImportError:
            raise ImportError(
                "feedparser package not found, please install it with "
                "`pip install feedparser`"
            )

        for url in self._get_urls:
            try:
                feed = feedparser.parse(url)
                if getattr(feed, "bozo", False):
                    raise ValueError(
                        f"Error fetching {url}, exception: {feed.bozo_exception}"
                    )
            except Exception as e:
                if self.continue_on_failure:
                    logger.error(f"Error fetching {url}, exception: {e}")
                    continue
                else:
                    raise e
            try:
                for entry in feed.entries:
                    loader = NewsURLLoader(
                        urls=[entry.link],
                        **self.newsloader_kwargs,
                    )
                    article = loader.load()[0]
                    article.metadata["feed"] = url
                    yield article
            except Exception as e:
                if self.continue_on_failure:
                    logger.error(f"Error processing entry {entry.link}, exception: {e}")
                    continue
                else:
                    raise e