Spaces:
Runtime error
Runtime error
File size: 1,814 Bytes
129cd69 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 |
from typing import Any, Iterator, List
from langchain.docstore.document import Document
from langchain.document_loaders.base import BaseLoader
class BaiduBOSDirectoryLoader(BaseLoader):
"""Load from `Baidu BOS directory`."""
def __init__(self, conf: Any, bucket: str, prefix: str = ""):
"""Initialize with BOS config, bucket and prefix.
:param conf(BosConfig): BOS config.
:param bucket(str): BOS bucket.
:param prefix(str): prefix.
"""
self.conf = conf
self.bucket = bucket
self.prefix = prefix
def load(self) -> List[Document]:
return list(self.lazy_load())
def lazy_load(self) -> Iterator[Document]:
"""Load documents."""
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.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]
|