Source code for langchain_community.document_loaders.baiducloud_bos_directory
from typing import Any, Iterator
from langchain_core.documents import Document
from langchain_community.document_loaders.base import BaseLoader
[docs]class BaiduBOSDirectoryLoader(BaseLoader):
"""从`百度BOS目录`加载。"""
[docs] def __init__(self, conf: Any, bucket: str, prefix: str = ""):
"""使用BOS配置、存储桶和前缀进行初始化。
:param conf(BosConfig): BOS配置。
:param bucket(str): BOS存储桶。
:param prefix(str): 前缀。
"""
self.conf = conf
self.bucket = bucket
self.prefix = prefix
[docs] def lazy_load(self) -> Iterator[Document]:
"""加载文档。"""
try:
from baidubce.services.bos.bos_client import BosClient
except ImportError:
raise ImportError(
"Please install bce-python-sdk with `pip install bce-python-sdk`."
)
client = BosClient(self.conf)
contents = []
marker = ""
while True:
response = client.list_objects(
bucket_name=self.bucket,
prefix=self.prefix,
marker=marker,
max_keys=1000,
)
contents_len = len(response.contents)
contents.extend(response.contents)
if response.is_truncated or contents_len < int(str(response.max_keys)):
break
marker = response.next_marker
from langchain_community.document_loaders.baiducloud_bos_file import (
BaiduBOSFileLoader,
)
for content in contents:
if str(content.key).endswith("/"):
continue
loader = BaiduBOSFileLoader(self.conf, self.bucket, str(content.key))
yield loader.load()[0]