Microsoft sharepoint

SharePointReader

Bases: BasePydanticReader, ResourcesReaderMixin, FileSystemReaderMixin

SharePoint阅读器。

从文档下的文件夹中读取SharePoint站点的文件夹。

Parameters:

Name	Type	Description	Default
client_id	str	在Microsoft Azure Portal中注册的应用程序的应用程序ID。 应用程序还必须配置有MS Graph权限"Files.ReadAll"、"Sites.ReadAll"和BrowserSiteLists.Read.All。	required
client_secret	str	在Azure中注册的应用程序的应用程序密钥。	required
tenant_id	str	Azure Active Directory实例的唯一标识符。	required
sharepoint_site_name	Optional[str]	要从中下载的SharePoint站点的名称。	None
sharepoint_folder_path	Optional[str]	要从中下载的SharePoint文件夹的路径。	None
sharepoint_folder_id	Optional[str]	要从中下载的SharePoint文件夹的ID。覆盖sharepoint_folder_path。	None
file_extractor	Optional[Dict[str, BaseReader]]	将文件扩展名映射到BaseReader类的映射，该类指定如何将该文件转换为文本。有关更多详细信息，请参见SimpleDirectoryReader。	None
attach_permission_metadata	bool	如果为True，则阅读器将在文档上附加权限元数据。如果您的向量存储仅支持平面元数据（即没有嵌套字段或列表），或者要避免额外的API调用，则设置为False。	required

Source code in llama_index/readers/microsoft_sharepoint/base.py

class SharePointReader(BasePydanticReader, ResourcesReaderMixin, FileSystemReaderMixin):
    """
    SharePoint阅读器。

    从文档下的文件夹中读取SharePoint站点的文件夹。

    Args:
        client_id (str): 在Microsoft Azure Portal中注册的应用程序的应用程序ID。
            应用程序还必须配置有MS Graph权限"Files.ReadAll"、"Sites.ReadAll"和BrowserSiteLists.Read.All。
        client_secret (str): 在Azure中注册的应用程序的应用程序密钥。
        tenant_id (str): Azure Active Directory实例的唯一标识符。
        sharepoint_site_name (Optional[str]): 要从中下载的SharePoint站点的名称。
        sharepoint_folder_path (Optional[str]): 要从中下载的SharePoint文件夹的路径。
        sharepoint_folder_id (Optional[str]): 要从中下载的SharePoint文件夹的ID。覆盖sharepoint_folder_path。
        file_extractor (Optional[Dict[str, BaseReader]]): 将文件扩展名映射到BaseReader类的映射，该类指定如何将该文件转换为文本。有关更多详细信息，请参见`SimpleDirectoryReader`。
        attach_permission_metadata (bool): 如果为True，则阅读器将在文档上附加权限元数据。如果您的向量存储仅支持平面元数据（即没有嵌套字段或列表），或者要避免额外的API调用，则设置为False。"""

    client_id: str = None
    client_secret: str = None
    tenant_id: str = None
    sharepoint_site_name: Optional[str] = None
    sharepoint_folder_path: Optional[str] = None
    sharepoint_folder_id: Optional[str] = None
    file_extractor: Optional[Dict[str, Union[str, BaseReader]]] = Field(
        default=None, exclude=True
    )
    attach_permission_metadata: bool = True

    _authorization_headers = PrivateAttr()
    _site_id_with_host_name = PrivateAttr()
    _drive_id_endpoint = PrivateAttr()
    _drive_id = PrivateAttr()

    def __init__(
        self,
        client_id: str,
        client_secret: str,
        tenant_id: str,
        sharepoint_site_name: Optional[str] = None,
        sharepoint_folder_path: Optional[str] = None,
        sharepoint_folder_id: Optional[str] = None,
        file_extractor: Optional[Dict[str, Union[str, BaseReader]]] = None,
        **kwargs: Any,
    ) -> None:
        super().__init__(
            client_id=client_id,
            client_secret=client_secret,
            tenant_id=tenant_id,
            sharepoint_site_name=sharepoint_site_name,
            sharepoint_folder_path=sharepoint_folder_path,
            sharepoint_folder_id=sharepoint_folder_id,
            file_extractor=file_extractor,
            **kwargs,
        )

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

    def _get_access_token(self) -> str:
        """获取用于访问SharePoint文件的访问令牌。

返回:
    str: 用于访问文件的访问令牌。

Raises:
    ValueError: 如果获取访问令牌时出现错误。
"""
        authority = f"https://login.microsoftonline.com/{self.tenant_id}/oauth2/token"

        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "resource": "https://graph.microsoft.com/",
        }

        response = requests.post(
            url=authority,
            data=payload,
        )

        if response.status_code == 200 and "access_token" in response.json():
            return response.json()["access_token"]

        else:
            logger.error(response.json()["error"])
            raise ValueError(response.json()["error_description"])

    def _get_site_id_with_host_name(self, access_token, sharepoint_site_name) -> str:
        """检索使用提供的站点名称的SharePoint站点的站点ID。

Args:
    sharepoint_site_name（str）：SharePoint站点的名称。

Returns:
    str：SharePoint站点的ID。

引发：
    Exception：如果未找到指定的SharePoint站点。
"""
        if hasattr(self, "_site_id_with_host_name"):
            return self._site_id_with_host_name

        site_information_endpoint = (
            f"https://graph.microsoft.com/v1.0/sites?search={sharepoint_site_name}"
        )
        self._authorization_headers = {"Authorization": f"Bearer {access_token}"}

        response = requests.get(
            url=site_information_endpoint,
            headers=self._authorization_headers,
        )

        if response.status_code == 200 and "value" in response.json():
            if (
                len(response.json()["value"]) > 0
                and "id" in response.json()["value"][0]
            ):
                return response.json()["value"][0]["id"]
            else:
                raise ValueError(
                    f"The specified sharepoint site {sharepoint_site_name} is not found."
                )
        else:
            if "error_description" in response.json():
                logger.error(response.json()["error"])
                raise ValueError(response.json()["error_description"])
            raise ValueError(response.json()["error"])

    def _get_drive_id(self) -> str:
        """获取SharePoint站点的驱动器ID。

返回：
    str：SharePoint站点驱动器的ID。

引发：
    ValueError：如果获取驱动器ID时出现错误。
"""
        if hasattr(self, "_drive_id"):
            return self._drive_id

        self._drive_id_endpoint = f"https://graph.microsoft.com/v1.0/sites/{self._site_id_with_host_name}/drives"

        response = requests.get(
            url=self._drive_id_endpoint,
            headers=self._authorization_headers,
        )

        if response.status_code == 200 and "value" in response.json():
            if (
                len(response.json()["value"]) > 0
                and "id" in response.json()["value"][0]
            ):
                return response.json()["value"][0]["id"]
            else:
                raise ValueError(
                    "Error occurred while fetching the drives for the sharepoint site."
                )
        else:
            logger.error(response.json()["error"])
            raise ValueError(response.json()["error_description"])

    def _get_sharepoint_folder_id(self, folder_path: str) -> str:
        """获取SharePoint站点的文件夹ID。

Args:
    folder_path（str）：SharePoint站点中文件夹的路径。

Returns:
    str：SharePoint站点文件夹的ID。
"""
        folder_id_endpoint = (
            f"{self._drive_id_endpoint}/{self._drive_id}/root:/{folder_path}"
        )

        response = requests.get(
            url=folder_id_endpoint,
            headers=self._authorization_headers,
        )

        if response.status_code == 200 and "id" in response.json():
            return response.json()["id"]
        else:
            raise ValueError(response.json()["error"])

    def _download_files_and_extract_metadata(
        self,
        folder_id: str,
        download_dir: str,
        current_folder_path: str,
        include_subfolders: bool = False,
    ) -> Dict[str, str]:
        """从指定的文件夹ID下载文件并提取元数据。

Args:
    folder_id（str）：应从中下载文件的文件夹的ID。
    download_dir（str）：应下载文件的目录。
    include_subfolders（bool）：如果为True，则下载所有子文件夹中的文件。

Returns:
    Dict[str, str]：包含已下载文件的元数据的字典。

引发：
    ValueError：如果在下载文件时出现错误。
"""
        folder_info_endpoint = (
            f"{self._drive_id_endpoint}/{self._drive_id}/items/{folder_id}/children"
        )

        response = requests.get(
            url=folder_info_endpoint,
            headers=self._authorization_headers,
        )

        if response.status_code == 200:
            data = response.json()
            metadata = {}
            for item in data["value"]:
                if include_subfolders and "folder" in item:
                    sub_folder_download_dir = os.path.join(download_dir, item["name"])
                    subfolder_metadata = self._download_files_and_extract_metadata(
                        folder_id=item["id"],
                        download_dir=sub_folder_download_dir,
                        current_folder_path=os.path.join(
                            current_folder_path, item["name"]
                        ),
                        include_subfolders=include_subfolders,
                    )

                    metadata.update(subfolder_metadata)

                elif "file" in item:
                    file_metadata = self._download_file(
                        item, download_dir, current_folder_path
                    )
                    metadata.update(file_metadata)
            return metadata
        else:
            logger.error(response.json()["error"])
            raise ValueError(response.json()["error"])

    def _get_file_content_by_url(self, item: Dict[str, Any]) -> bytes:
        """从提供的URL中获取文件的内容。

Args:
    item（Dict[str, Any]）：包含文件元数据的字典。

Returns:
    bytes：文件的内容。
"""
        file_download_url = item["@microsoft.graph.downloadUrl"]
        response = requests.get(file_download_url)
        if response.status_code != 200:
            logger.error(response.json()["error"])
            raise ValueError(response.json()["error_description"])

        return response.content

    def _download_file_by_url(self, item: Dict[str, Any], download_dir: str) -> str:
        """从提供的URL下载文件。

Args:
    item（Dict[str，Any]）：包含文件元数据的字典。
    download_dir（str）：应下载文件的目录。

Returns:
    str：临时目录中已下载文件的路径。
"""
        # Get the download URL for the file.
        file_name = item["name"]

        content = self._get_file_content_by_url(item)

        # Create the directory if it does not exist and save the file.
        if not os.path.exists(download_dir):
            os.makedirs(download_dir)
        file_path = os.path.join(download_dir, file_name)
        with open(file_path, "wb") as f:
            f.write(content)

        return file_path

    def _get_permissions_info(self, item: Dict[str, Any]) -> Dict[str, str]:
        """提取文件的权限信息。更多信息请参见：
https://learn.microsoft.com/en-us/graph/api/resources/permission?view=graph-rest-1.0.

Args:
    item (Dict[str, Any]): 包含文件元数据的字典。

Returns:
    Dict[str, str]: 包含提取的权限信息的字典。
"""
        item_id = item.get("id")
        permissions_info_endpoint = (
            f"{self._drive_id_endpoint}/{self._drive_id}/items/{item_id}/permissions"
        )
        response = requests.get(
            url=permissions_info_endpoint,
            headers=self._authorization_headers,
        )
        permissions = response.json()

        identity_sets = []
        for permission in permissions["value"]:
            # user type permissions
            granted_to = permission.get("grantedToV2", None)
            if granted_to:
                identity_sets.append(granted_to)

            # link type permissions
            granted_to_identities = permission.get("grantedToIdentitiesV2", [])
            for identity in granted_to_identities:
                identity_sets.append(identity)

        # Extract the identity information from each identity set
        # they can be 'application', 'device', 'user', 'group', 'siteUser' or 'siteGroup'
        # 'siteUser' and 'siteGroup' are site-specific, 'group' is for Microsoft 365 groups
        permissions_dict = {}
        for identity_set in identity_sets:
            for identity, identity_info in identity_set.items():
                id = identity_info.get("id")
                display_name = identity_info.get("displayName")
                ids_key = f"allowed_{identity}_ids"
                display_names_key = f"allowed_{identity}_display_names"

                if ids_key not in permissions_dict:
                    permissions_dict[ids_key] = []
                if display_names_key not in permissions_dict:
                    permissions_dict[display_names_key] = []

                permissions_dict[ids_key].append(id)
                permissions_dict[display_names_key].append(display_name)

        # sort to get consistent results, if possible
        for key in permissions_dict:
            try:
                permissions_dict[key] = sorted(permissions_dict[key])
            except TypeError:
                pass

        return permissions_dict

    def _extract_metadata_for_file(self, item: Dict[str, Any]) -> Dict[str, str]:
        """提取与文件相关的元数据。

Args:
- item（Dict[str, str]）：包含文件元数据的字典。

Returns:
- Dict[str, str]：包含提取的元数据的字典。
"""
        # Extract the required metadata for file.
        if self.attach_permission_metadata:
            metadata = self._get_permissions_info(item)
        else:
            metadata = {}

        metadata.update(
            {
                "file_id": item.get("id"),
                "file_name": item.get("name"),
                "url": item.get("webUrl"),
                "file_path": item.get("file_path"),
            }
        )

        return metadata

    def _download_file(
        self,
        item: Dict[str, Any],
        download_dir: str,
        sharepoint_folder_path: str,
    ):
        metadata = {}

        file_path = self._download_file_by_url(item, download_dir)
        item["file_path"] = os.path.join(sharepoint_folder_path, item["name"])

        metadata[file_path] = self._extract_metadata_for_file(item)
        return metadata

    def _download_files_from_sharepoint(
        self,
        download_dir: str,
        sharepoint_site_name: str,
        sharepoint_folder_path: Optional[str],
        sharepoint_folder_id: Optional[str],
        recursive: bool,
    ) -> Dict[str, str]:
        """从指定文件夹下载文件，并返回已下载文件的元数据。

Args:
    download_dir (str): 应下载文件的目录。
    sharepoint_site_name (str): SharePoint站点的名称。
    sharepoint_folder_path (str): SharePoint站点中文件夹的路径。
    recursive (bool): 如果为True，则下载所有子文件夹中的文件。

Returns:
    Dict[str, str]: 包含已下载文件的元数据的字典。
"""
        access_token = self._get_access_token()

        self._site_id_with_host_name = self._get_site_id_with_host_name(
            access_token, sharepoint_site_name
        )

        self._drive_id = self._get_drive_id()

        if not sharepoint_folder_id:
            sharepoint_folder_id = self._get_sharepoint_folder_id(
                sharepoint_folder_path
            )

        return self._download_files_and_extract_metadata(
            sharepoint_folder_id,
            download_dir,
            os.path.join(sharepoint_site_name, sharepoint_folder_path),
            recursive,
        )

    def _exclude_access_control_metadata(
        self, documents: List[Document]
    ) -> List[Document]:
        """从文档中排除访问控制元数据，以便进行嵌入和LLM调用。

Args:
    documents（List[Document]）：文档列表。

Returns:
    List[Document]：排除了访问控制元数据的文档列表。
"""
        for doc in documents:
            access_control_keys = [
                key for key in doc.metadata if key.startswith("allowed_")
            ]

            doc.excluded_embed_metadata_keys.extend(access_control_keys)
            doc.excluded_llm_metadata_keys.extend(access_control_keys)

        return documents

    def _load_documents_with_metadata(
        self,
        files_metadata: Dict[str, Any],
        download_dir: str,
        recursive: bool,
    ) -> List[Document]:
        """从下载的文件中加载文档。

Args:
    files_metadata (Dict[str,Any]): 包含下载文件的元数据的字典。
    download_dir (str): 应该下载文件的目录。
    recursive (bool): 如果为True，则下载所有子文件夹中的文件。

Returns:
    List[Document]: 包含带有元数据的文档的列表。
"""

        def get_metadata(filename: str) -> Any:
            return files_metadata[filename]

        simple_loader = SimpleDirectoryReader(
            download_dir,
            file_extractor=self.file_extractor,
            file_metadata=get_metadata,
            recursive=recursive,
        )
        docs = simple_loader.load_data()
        if self.attach_permission_metadata:
            docs = self._exclude_access_control_metadata(docs)
        return docs

    def load_data(
        self,
        sharepoint_site_name: Optional[str] = None,
        sharepoint_folder_path: Optional[str] = None,
        sharepoint_folder_id: Optional[str] = None,
        recursive: bool = True,
    ) -> List[Document]:
        """从SharePoint站点中的指定文件夹加载文件。

Args:
    sharepoint_site_name（可选[str]）：SharePoint站点的名称。
    sharepoint_folder_path（可选[str]）：SharePoint站点中文件夹的路径。
    recursive（bool）：如果为True，则下载所有子文件夹中的文件。

Returns:
    List[Document]：包含带有元数据的文档的列表。

引发:
    Exception：访问SharePoint站点时发生错误。
"""
        # If no arguments are provided to load_data, default to the object attributes
        if not sharepoint_site_name:
            sharepoint_site_name = self.sharepoint_site_name

        if not sharepoint_folder_path:
            sharepoint_folder_path = self.sharepoint_folder_path

        if not sharepoint_folder_id:
            sharepoint_folder_id = self.sharepoint_folder_id

        # TODO: make both of these values optional — and just default to the client ID defaults
        if not sharepoint_site_name:
            raise ValueError("sharepoint_site_name must be provided.")

        if not sharepoint_folder_path and not sharepoint_folder_id:
            raise ValueError(
                "sharepoint_folder_path or sharepoint_folder_id must be provided."
            )

        try:
            with tempfile.TemporaryDirectory() as temp_dir:
                files_metadata = self._download_files_from_sharepoint(
                    temp_dir,
                    sharepoint_site_name,
                    sharepoint_folder_path,
                    sharepoint_folder_id,
                    recursive,
                )
                # return self.files_metadata
                return self._load_documents_with_metadata(
                    files_metadata, temp_dir, recursive
                )

        except Exception as exp:
            logger.error("An error occurred while accessing SharePoint: %s", exp)

    def _list_folder_contents(
        self, folder_id: str, recursive: bool, current_path: str
    ) -> List[Path]:
        """辅助方法，用于获取文件夹的内容。

Args:
    folder_id（str）：要列出内容的文件夹的ID。
    recursive（bool）：是否递归包含子文件夹。

Returns:
    List[Path]：文件路径列表。
"""
        folder_contents_endpoint = (
            f"{self._drive_id_endpoint}/{self._drive_id}/items/{folder_id}/children"
        )
        response = requests.get(
            url=folder_contents_endpoint,
            headers=self._authorization_headers,
        )
        items = response.json().get("value", [])

        file_paths = []
        for item in items:
            if "folder" in item and recursive:
                # Recursive call for subfolder
                subfolder_id = item["id"]
                subfolder_paths = self._list_folder_contents(
                    subfolder_id, recursive, os.path.join(current_path, item["name"])
                )
                file_paths.extend(subfolder_paths)
            elif "file" in item:
                # Append file path
                file_path = Path(os.path.join(current_path, item["name"]))
                file_paths.append(file_path)

        return file_paths

    def list_resources(
        self,
        sharepoint_site_name: Optional[str] = None,
        sharepoint_folder_path: Optional[str] = None,
        sharepoint_folder_id: Optional[str] = None,
        recursive: bool = True,
    ) -> List[Path]:
        """列出SharePoint站点中指定文件夹中的文件。

Args:
    **kwargs：额外的关键字参数。

Returns:
    List[Path]：指定文件夹中文件的路径列表。

引发：
    Exception：访问SharePoint站点时发生错误。
"""
        # If no arguments are provided to load_data, default to the object attributes
        if not sharepoint_site_name:
            sharepoint_site_name = self.sharepoint_site_name

        if not sharepoint_folder_path:
            sharepoint_folder_path = self.sharepoint_folder_path

        if not sharepoint_folder_id:
            sharepoint_folder_id = self.sharepoint_folder_id

        # TODO: make both of these values optional — and just default to the client ID defaults
        if not sharepoint_site_name:
            raise ValueError("sharepoint_site_name must be provided.")

        if not sharepoint_folder_path and not sharepoint_folder_id:
            raise ValueError(
                "sharepoint_folder_path or sharepoint_folder_id must be provided."
            )

        file_paths = []
        try:
            access_token = self._get_access_token()
            self._site_id_with_host_name = self._get_site_id_with_host_name(
                access_token, sharepoint_site_name
            )
            self._drive_id = self._get_drive_id()
            if not sharepoint_folder_id:
                sharepoint_folder_id = self._get_sharepoint_folder_id(
                    sharepoint_folder_path
                )

            # Fetch folder contents
            folder_contents = self._list_folder_contents(
                sharepoint_folder_id,
                recursive,
                os.path.join(sharepoint_site_name, sharepoint_folder_path),
            )
            file_paths.extend(folder_contents)
            return file_paths

        except Exception as exp:
            logger.error("An error occurred while listing files in SharePoint: %s", exp)
            raise

        return file_paths

    def _get_item_from_path(self, input_file: Path) -> Dict[str, Any]:
        """获取SharePoint中指定文件的项目详情。

Args:
    input_file (Path): SharePoint中文件的路径。
        应包括SharePoint站点名称和文件夹路径。例如 "site_name/folder_path/file_name"。

Returns:
    Dict[str, Any]: 包含项目详情的字典。
"""
        # Get the file ID
        # remove the site_name prefix
        file_path = (
            str(input_file).lstrip("/").replace(f"{self.sharepoint_site_name}/", "", 1)
        )
        endpoint = f"{self._drive_id_endpoint}/{self._drive_id}/root:/{file_path}"

        response = requests.get(
            url=endpoint,
            headers=self._authorization_headers,
        )

        return response.json()

    def get_resource_info(self, resource_id: str, **kwargs) -> Dict:
        """检索SharePoint中指定文件的元数据，而无需下载该文件。

Args:
    input_file（路径）：SharePoint中文件的路径。路径应包括
                        SharePoint站点名称和文件夹路径。例如 "site_name/folder_path/file_name"。
"""
        try:
            item = self._get_item_from_path(Path(resource_id))

            info_dict = {
                "file_path": resource_id,
                "size": item.get("size"),
                "created_at": item.get("createdDateTime"),
                "modified_at": item.get("lastModifiedDateTime"),
                "etag": item.get("eTag"),
            }

            if (
                self.attach_permission_metadata
            ):  # changes in access control should trigger a reingestion of the file
                permissions = self._get_permissions_info(item)
                info_dict.update(permissions)

            return {
                meta_key: meta_value
                for meta_key, meta_value in info_dict.items()
                if meta_value is not None
            }

        except Exception as exp:
            logger.error(
                "An error occurred while fetching file information from SharePoint: %s",
                exp,
            )
            raise

    def load_resource(self, resource_id: str, **kwargs) -> List[Document]:
        try:
            access_token = self._get_access_token()
            self._site_id_with_host_name = self._get_site_id_with_host_name(
                access_token, self.sharepoint_site_name
            )
            self._drive_id = self._get_drive_id()

            path = Path(resource_id)

            item = self._get_item_from_path(path)

            input_file_dir = path.parent

            with tempfile.TemporaryDirectory() as temp_dir:
                metadata = self._download_file(item, temp_dir, input_file_dir)
                return self._load_documents_with_metadata(
                    metadata, temp_dir, recursive=False
                )

        except Exception as exp:
            logger.error(
                "An error occurred while reading file from SharePoint: %s", exp
            )
            raise

    def read_file_content(self, input_file: Path, **kwargs) -> bytes:
        try:
            access_token = self._get_access_token()
            self._site_id_with_host_name = self._get_site_id_with_host_name(
                access_token, self.sharepoint_site_name
            )
            self._drive_id = self._get_drive_id()

            item = self._get_item_from_path(input_file)
            return self._get_file_content_by_url(item)

        except Exception as exp:
            logger.error(
                "An error occurred while reading file content from SharePoint: %s", exp
            )
            raise

load_data

load_data(
    sharepoint_site_name: Optional[str] = None,
    sharepoint_folder_path: Optional[str] = None,
    sharepoint_folder_id: Optional[str] = None,
    recursive: bool = True,
) -> List[Document]

从SharePoint站点中的指定文件夹加载文件。

Returns:

Type	Description
List[Document]	List[Document]：包含带有元数据的文档的列表。

引发

Exception：访问SharePoint站点时发生错误。

Source code in llama_index/readers/microsoft_sharepoint/base.py

    def load_data(
        self,
        sharepoint_site_name: Optional[str] = None,
        sharepoint_folder_path: Optional[str] = None,
        sharepoint_folder_id: Optional[str] = None,
        recursive: bool = True,
    ) -> List[Document]:
        """从SharePoint站点中的指定文件夹加载文件。

Args:
    sharepoint_site_name（可选[str]）：SharePoint站点的名称。
    sharepoint_folder_path（可选[str]）：SharePoint站点中文件夹的路径。
    recursive（bool）：如果为True，则下载所有子文件夹中的文件。

Returns:
    List[Document]：包含带有元数据的文档的列表。

引发:
    Exception：访问SharePoint站点时发生错误。
"""
        # If no arguments are provided to load_data, default to the object attributes
        if not sharepoint_site_name:
            sharepoint_site_name = self.sharepoint_site_name

        if not sharepoint_folder_path:
            sharepoint_folder_path = self.sharepoint_folder_path

        if not sharepoint_folder_id:
            sharepoint_folder_id = self.sharepoint_folder_id

        # TODO: make both of these values optional — and just default to the client ID defaults
        if not sharepoint_site_name:
            raise ValueError("sharepoint_site_name must be provided.")

        if not sharepoint_folder_path and not sharepoint_folder_id:
            raise ValueError(
                "sharepoint_folder_path or sharepoint_folder_id must be provided."
            )

        try:
            with tempfile.TemporaryDirectory() as temp_dir:
                files_metadata = self._download_files_from_sharepoint(
                    temp_dir,
                    sharepoint_site_name,
                    sharepoint_folder_path,
                    sharepoint_folder_id,
                    recursive,
                )
                # return self.files_metadata
                return self._load_documents_with_metadata(
                    files_metadata, temp_dir, recursive
                )

        except Exception as exp:
            logger.error("An error occurred while accessing SharePoint: %s", exp)

list_resources

list_resources(
    sharepoint_site_name: Optional[str] = None,
    sharepoint_folder_path: Optional[str] = None,
    sharepoint_folder_id: Optional[str] = None,
    recursive: bool = True,
) -> List[Path]

列出SharePoint站点中指定文件夹中的文件。

Returns:

Type	Description
List[Path]	List[Path]：指定文件夹中文件的路径列表。

引发： Exception：访问SharePoint站点时发生错误。

Source code in llama_index/readers/microsoft_sharepoint/base.py

    def list_resources(
        self,
        sharepoint_site_name: Optional[str] = None,
        sharepoint_folder_path: Optional[str] = None,
        sharepoint_folder_id: Optional[str] = None,
        recursive: bool = True,
    ) -> List[Path]:
        """列出SharePoint站点中指定文件夹中的文件。

Args:
    **kwargs：额外的关键字参数。

Returns:
    List[Path]：指定文件夹中文件的路径列表。

引发：
    Exception：访问SharePoint站点时发生错误。
"""
        # If no arguments are provided to load_data, default to the object attributes
        if not sharepoint_site_name:
            sharepoint_site_name = self.sharepoint_site_name

        if not sharepoint_folder_path:
            sharepoint_folder_path = self.sharepoint_folder_path

        if not sharepoint_folder_id:
            sharepoint_folder_id = self.sharepoint_folder_id

        # TODO: make both of these values optional — and just default to the client ID defaults
        if not sharepoint_site_name:
            raise ValueError("sharepoint_site_name must be provided.")

        if not sharepoint_folder_path and not sharepoint_folder_id:
            raise ValueError(
                "sharepoint_folder_path or sharepoint_folder_id must be provided."
            )

        file_paths = []
        try:
            access_token = self._get_access_token()
            self._site_id_with_host_name = self._get_site_id_with_host_name(
                access_token, sharepoint_site_name
            )
            self._drive_id = self._get_drive_id()
            if not sharepoint_folder_id:
                sharepoint_folder_id = self._get_sharepoint_folder_id(
                    sharepoint_folder_path
                )

            # Fetch folder contents
            folder_contents = self._list_folder_contents(
                sharepoint_folder_id,
                recursive,
                os.path.join(sharepoint_site_name, sharepoint_folder_path),
            )
            file_paths.extend(folder_contents)
            return file_paths

        except Exception as exp:
            logger.error("An error occurred while listing files in SharePoint: %s", exp)
            raise

        return file_paths

get_resource_info

get_resource_info(resource_id: str, **kwargs) -> Dict

检索SharePoint中指定文件的元数据，而无需下载该文件。

Source code in llama_index/readers/microsoft_sharepoint/base.py

    def get_resource_info(self, resource_id: str, **kwargs) -> Dict:
        """检索SharePoint中指定文件的元数据，而无需下载该文件。

Args:
    input_file（路径）：SharePoint中文件的路径。路径应包括
                        SharePoint站点名称和文件夹路径。例如 "site_name/folder_path/file_name"。
"""
        try:
            item = self._get_item_from_path(Path(resource_id))

            info_dict = {
                "file_path": resource_id,
                "size": item.get("size"),
                "created_at": item.get("createdDateTime"),
                "modified_at": item.get("lastModifiedDateTime"),
                "etag": item.get("eTag"),
            }

            if (
                self.attach_permission_metadata
            ):  # changes in access control should trigger a reingestion of the file
                permissions = self._get_permissions_info(item)
                info_dict.update(permissions)

            return {
                meta_key: meta_value
                for meta_key, meta_value in info_dict.items()
                if meta_value is not None
            }

        except Exception as exp:
            logger.error(
                "An error occurred while fetching file information from SharePoint: %s",
                exp,
            )
            raise