langchain_community.document_loaders.tencent_cos_directory

from typing import Any, Iterator

from langchain_core.documents import Document

from langchain_community.document_loaders.base import BaseLoader
from langchain_community.document_loaders.tencent_cos_file import TencentCOSFileLoader


[docs]class TencentCOSDirectoryLoader(BaseLoader):
    """从`腾讯云 COS`目录加载。"""

[docs]    def __init__(self, conf: Any, bucket: str, prefix: str = ""):
        """使用COS配置、存储桶和前缀进行初始化。
:param conf(CosConfig): COS配置。
:param bucket(str): COS存储桶。
:param prefix(str): 前缀。
"""
        self.conf = conf
        self.bucket = bucket
        self.prefix = prefix

[docs]    def lazy_load(self) -> Iterator[Document]:
        """加载文档。"""
        try:
            from qcloud_cos import CosS3Client
        except ImportError:
            raise ImportError(
                "Could not import cos-python-sdk-v5 python package. "
                "Please install it with `pip install cos-python-sdk-v5`."
            )
        client = CosS3Client(self.conf)
        contents = []
        marker = ""
        while True:
            response = client.list_objects(
                Bucket=self.bucket, Prefix=self.prefix, Marker=marker, MaxKeys=1000
            )
            if "Contents" in response:
                contents.extend(response["Contents"])
            if response["IsTruncated"] == "false":
                break
            marker = response["NextMarker"]
        for content in contents:
            if content["Key"].endswith("/"):
                continue
            loader = TencentCOSFileLoader(self.conf, self.bucket, content["Key"])
            yield loader.load()[0]
Source code for langchain_community.document_loaders.tencent_cos_directory
