Skip to content

Gcs

GCSReader #

Bases: BasePydanticReader

通用的GCS文件或目录的通用阅读器。

如果未设置key，则会解析整个存储桶（按前缀过滤）。

Args: bucket（str）：您的GCS存储桶的名称 key（可选[str]）：特定文件的名称。如果未提供，则此加载程序将遍历整个存储桶。 prefix（可选[str]）：在加载程序遍历整个存储桶时要按前缀过滤的前缀。默认为空字符串。 recursive（bool）：是否递归搜索子目录。默认为True。 file_extractor（可选[Dict[str, BaseReader]]）：文件扩展名到BaseReader类的映射，指定如何将该文件转换为文本。有关更多详细信息，请参见“SimpleDirectoryReader”。 required_exts（可选[List[str]]）：所需扩展名的列表。默认为None。 num_files_limit（可选[int]）：要读取的最大文件数。默认为None。 file_metadata（可选[Callable[str, Dict]]）：接受文件名并返回文档元数据的字典的函数。默认为None。 service_account_key（可选[Dict[str, str]]）：直接提供GCP服务帐号密钥。 service_account_key_json（可选[str]）：提供GCP服务帐号密钥的JSON字符串。 service_account_key_path（可选[str]）：提供包含GCP服务帐号密钥的文件路径。

Source code in llama_index/readers/gcs/base.py 
 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
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
class GCSReader(BasePydanticReader):
    """通用的GCS文件或目录的通用阅读器。

如果未设置key，则会解析整个存储桶（按前缀过滤）。

Args:
bucket（str）：您的GCS存储桶的名称
key（可选[str]）：特定文件的名称。如果未提供，则此加载程序将遍历整个存储桶。
prefix（可选[str]）：在加载程序遍历整个存储桶时要按前缀过滤的前缀。默认为空字符串。
recursive（bool）：是否递归搜索子目录。默认为True。
file_extractor（可选[Dict[str, BaseReader]]）：文件扩展名到BaseReader类的映射，指定如何将该文件转换为文本。有关更多详细信息，请参见“SimpleDirectoryReader”。
required_exts（可选[List[str]]）：所需扩展名的列表。默认为None。
num_files_limit（可选[int]）：要读取的最大文件数。默认为None。
file_metadata（可选[Callable[str, Dict]]）：接受文件名并返回文档元数据的字典的函数。默认为None。
service_account_key（可选[Dict[str, str]]）：直接提供GCP服务帐号密钥。
service_account_key_json（可选[str]）：提供GCP服务帐号密钥的JSON字符串。
service_account_key_path（可选[str]）：提供包含GCP服务帐号密钥的文件路径。"""

    is_remote: bool = True

    bucket: str
    key: Optional[str] = None
    prefix: Optional[str] = ""
    recursive: bool = True
    file_extractor: Optional[Dict[str, Union[str, BaseReader]]] = Field(
        default=None, exclude=True
    )
    required_exts: Optional[List[str]] = None
    filename_as_id: bool = True
    num_files_limit: Optional[int] = None
    file_metadata: Optional[Callable[[str], Dict]] = Field(default=None, exclude=True)
    service_account_key: Optional[Dict[str, str]] = None
    service_account_key_json: Optional[str] = None
    service_account_key_path: Optional[str] = None

    @classmethod
    def class_name(cls) -> str:
        return "GCSReader"

    def load_gcs_files_as_docs(self) -> List[Document]:
        """从GCS加载文件。"""
        from gcsfs import GCSFileSystem

        if self.service_account_key is not None:
            creds = service_account.Credentials.from_service_account_info(
                self.service_account_key, scopes=SCOPES
            )
        elif self.service_account_key_json is not None:
            creds = service_account.Credentials.from_service_account_info(
                json.loads(self.service_account_key_json), scopes=SCOPES
            )
        elif self.service_account_key_path is not None:
            creds = service_account.Credentials.from_service_account_file(
                self.service_account_key_path, scopes=SCOPES
            )
        else:
            # Use anonymous access if none are specified
            creds = "anon"

        gcsfs = GCSFileSystem(
            token=creds,
        )

        input_dir = self.bucket
        input_files = None

        if self.key:
            input_files = [f"{self.bucket}/{self.key}"]
        elif self.prefix:
            input_dir = f"{input_dir}/{self.prefix}"

        loader = SimpleDirectoryReader(
            input_dir=input_dir,
            input_files=input_files,
            recursive=self.recursive,
            file_extractor=self.file_extractor,
            required_exts=self.required_exts,
            filename_as_id=self.filename_as_id,
            num_files_limit=self.num_files_limit,
            file_metadata=self.file_metadata,
            fs=gcsfs,
        )

        return loader.load_data()

    def load_data(self) -> List[Document]:
        """从GCS加载文件（们）。

返回：
    List[Document]: 从GCS加载的文档列表。
"""
        return self.load_gcs_files_as_docs()

load_gcs_files_as_docs #

load_gcs_files_as_docs() -> List[Document]

从GCS加载文件。

Source code in llama_index/readers/gcs/base.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def load_gcs_files_as_docs(self) -> List[Document]:
    """从GCS加载文件。"""
    from gcsfs import GCSFileSystem

    if self.service_account_key is not None:
        creds = service_account.Credentials.from_service_account_info(
            self.service_account_key, scopes=SCOPES
        )
    elif self.service_account_key_json is not None:
        creds = service_account.Credentials.from_service_account_info(
            json.loads(self.service_account_key_json), scopes=SCOPES
        )
    elif self.service_account_key_path is not None:
        creds = service_account.Credentials.from_service_account_file(
            self.service_account_key_path, scopes=SCOPES
        )
    else:
        # Use anonymous access if none are specified
        creds = "anon"

    gcsfs = GCSFileSystem(
        token=creds,
    )

    input_dir = self.bucket
    input_files = None

    if self.key:
        input_files = [f"{self.bucket}/{self.key}"]
    elif self.prefix:
        input_dir = f"{input_dir}/{self.prefix}"

    loader = SimpleDirectoryReader(
        input_dir=input_dir,
        input_files=input_files,
        recursive=self.recursive,
        file_extractor=self.file_extractor,
        required_exts=self.required_exts,
        filename_as_id=self.filename_as_id,
        num_files_limit=self.num_files_limit,
        file_metadata=self.file_metadata,
        fs=gcsfs,
    )

    return loader.load_data()

load_data #

load_data() -> List[Document]

从GCS加载文件（们）。

返回： List[Document]: 从GCS加载的文档列表。

Source code in llama_index/readers/gcs/base.py 
105
106
107
108
109
110
111
    def load_data(self) -> List[Document]:
        """从GCS加载文件（们）。

返回：
    List[Document]: 从GCS加载的文档列表。
"""
        return self.load_gcs_files_as_docs()