Source code for langchain_community.document_loaders.obs_file

# coding:utf-8

import os
import tempfile
from typing import Any, List, Optional

from langchain_core.documents import Document

from langchain_community.document_loaders.base import BaseLoader
from langchain_community.document_loaders.unstructured import UnstructuredFileLoader


class OBSFileLoader(BaseLoader):
    """从`华为OBS文件`加载。"""

    def __init__(
        self,
        bucket: str,
        key: str,
        client: Any = None,
        endpoint: str = "",
        config: Optional[dict] = None,
    ) -> None:
        """使用指定的设置初始化OBSFileLoader。

参数：
    bucket（str）：要使用的OBS存储桶的名称。
    key（str）：OBS存储桶中对象的名称。
    client（ObsClient，可选）：ObsClient的一个实例，用于连接到OBS。
    endpoint（str，可选）：OBS存储桶的端点URL。如果未提供`client`，则此参数是必需的。
    config（dict，可选）：用字典形式提供连接到OBS的参数。如果提供了`client`，则此参数将被忽略。字典可以包含以下键：
        - “ak”（str，可选）：您的OBS访问密钥（如果`get_token_from_ecs`为False且存储桶策略不是公共读取，则为必需）。
        - “sk”（str，可选）：您的OBS秘密密钥（如果`get_token_from_ecs`为False且存储桶策略不是公共读取，则为必需）。
        - “token”（str，可选）：您的安全令牌（如果使用临时凭证，则为必需）。
        - “get_token_from_ecs”（bool，可选）：是否从ECS检索安全令牌。如果未提供，则默认为False。如果设置为True，则将忽略`ak`，`sk`和`token`。

引发：
    ValueError：如果未安装`esdk-obs-python`包。
    TypeError：如果提供的`client`不是ObsClient的实例。
    ValueError：如果未提供`client`，但缺少`endpoint`。

注意：
    在使用此类之前，请确保已在OBS注册并具有必要的凭据。除非`get_token_from_ecs`为True或存储桶策略为公共读取，否则`ak`，`sk`和`endpoint`值是必需的。在使用临时凭证时，`token`是必需的。

示例：
    创建一个带有新客户端的新OBSFileLoader：
    ```
    config = {
        "ak": "your-access-key",
        "sk": "your-secret-key"
    }
    obs_loader = OBSFileLoader("your-bucket-name", "your-object-key", config=config)
    ```

    创建一个带有现有客户端的新OBSFileLoader：
    ```
    from obs import ObsClient

    # 假设您有一个现有的ObsClient对象'obs_client'
    obs_loader = OBSFileLoader("your-bucket-name", "your-object-key", client=obs_client)
    ```

    创建一个不带现有客户端的新OBSFileLoader：
    ```
    obs_loader = OBSFileLoader("your-bucket-name", "your-object-key", endpoint="your-endpoint-url")
    ```
"""  # noqa: E501
        try:
            from obs import ObsClient
        except ImportError:
            raise ImportError(
                "Could not import esdk-obs-python python package. "
                "Please install it with `pip install esdk-obs-python`."
            )
        if not client:
            if not endpoint:
                raise ValueError("Either OBSClient or endpoint must be provided.")
            if not config:
                config = dict()
            if config.get("get_token_from_ecs"):
                client = ObsClient(server=endpoint, security_provider_policy="ECS")
            else:
                client = ObsClient(
                    access_key_id=config.get("ak"),
                    secret_access_key=config.get("sk"),
                    security_token=config.get("token"),
                    server=endpoint,
                )
        if not isinstance(client, ObsClient):
            raise TypeError("Client must be ObsClient type")
        self.client = client
        self.bucket = bucket
        self.key = key

    def load(self) -> List[Document]:
        """加载文档。"""
        with tempfile.TemporaryDirectory() as temp_dir:
            file_path = f"{temp_dir}/{self.bucket}/{self.key}"
            os.makedirs(os.path.dirname(file_path), exist_ok=True)
            # Download the file to a destination
            self.client.downloadFile(
                bucketName=self.bucket, objectKey=self.key, downloadFile=file_path
            )
            loader = UnstructuredFileLoader(file_path)
            return loader.load()