codekingpro/portable-devtools

1"""Loader that loads data from Sharepoint Document Library"""2 3from __future__ import annotations4 5import json6from pathlib import Path7from typing import Any, Dict, Iterator, List, Optional8 9import requests10from langchain_core.document_loaders import BaseLoader11from langchain_core.documents import Document12from pydantic import Field13 14from langchain_community.document_loaders.base_o365 import (15    O365BaseLoader,16)17 18 19class SharePointLoader(O365BaseLoader, BaseLoader):20    """Load  from `SharePoint`."""21 22    document_library_id: str = Field(...)23    """ The ID of the SharePoint document library to load data from."""24    folder_path: Optional[str] = None25    """ The path to the folder to load data from."""26    object_ids: Optional[List[str]] = None27    """ The IDs of the objects to load data from."""28    folder_id: Optional[str] = None29    """ The ID of the folder to load data from."""30    load_auth: Optional[bool] = False31    """ Whether to load authorization identities."""32    token_path: Path = Path.home() / ".credentials" / "o365_token.txt"33    """ The path to the token to make api calls"""34    load_extended_metadata: Optional[bool] = False35    """ Whether to load extended metadata. Size, Owner and full_path."""36 37    @property38    def _scopes(self) -> List[str]:39        """Return required scopes.40        Returns:41            List[str]: A list of required scopes.42        """43        return ["sharepoint", "basic"]44 45    def lazy_load(self) -> Iterator[Document]:46        """47        Load documents lazily. Use this when working at a large scale.48        Yields:49            Document: A document object representing the parsed blob.50        """51        try:52            from O365.drive import Drive, Folder53        except ImportError:54            raise ImportError(55                "O365 package not found, please install it with `pip install o365`"56            )57        drive = self._auth().storage().get_drive(self.document_library_id)58        if not isinstance(drive, Drive):59            raise ValueError(f"There isn't a Drive with id {self.document_library_id}.")60        if self.folder_path:61            target_folder = drive.get_item_by_path(self.folder_path)62            if not isinstance(target_folder, Folder):63                raise ValueError(f"There isn't a folder with path {self.folder_path}.")64            for blob in self._load_from_folder(target_folder):65                file_id = str(blob.metadata.get("id"))66                if self.load_auth is True:67                    auth_identities = self.authorized_identities(file_id)68                if self.load_extended_metadata is True:69                    extended_metadata = self.get_extended_metadata(file_id)70                    extended_metadata.update({"source_full_url": target_folder.web_url})71                for parsed_blob in self._blob_parser.lazy_parse(blob):72                    if self.load_auth is True:73                        parsed_blob.metadata["authorized_identities"] = auth_identities74                    if self.load_extended_metadata is True:75                        parsed_blob.metadata.update(extended_metadata)76                    yield parsed_blob77        if self.folder_id:78            target_folder = drive.get_item(self.folder_id)79            if not isinstance(target_folder, Folder):80                raise ValueError(f"There isn't a folder with path {self.folder_path}.")81            for blob in self._load_from_folder(target_folder):82                file_id = str(blob.metadata.get("id"))83                if self.load_auth is True:84                    auth_identities = self.authorized_identities(file_id)85                if self.load_extended_metadata is True:86                    extended_metadata = self.get_extended_metadata(file_id)87                    extended_metadata.update({"source_full_url": target_folder.web_url})88                for parsed_blob in self._blob_parser.lazy_parse(blob):89                    if self.load_auth is True:90                        parsed_blob.metadata["authorized_identities"] = auth_identities91                    if self.load_extended_metadata is True:92                        parsed_blob.metadata.update(extended_metadata)93                    yield parsed_blob94        if self.object_ids:95            for blob in self._load_from_object_ids(drive, self.object_ids):96                file_id = str(blob.metadata.get("id"))97                if self.load_auth is True:98                    auth_identities = self.authorized_identities(file_id)99                if self.load_extended_metadata is True:100                    extended_metadata = self.get_extended_metadata(file_id)101                for parsed_blob in self._blob_parser.lazy_parse(blob):102                    if self.load_auth is True:103                        parsed_blob.metadata["authorized_identities"] = auth_identities104                    if self.load_extended_metadata is True:105                        parsed_blob.metadata.update(extended_metadata)106                    yield parsed_blob107 108        if not (self.folder_path or self.folder_id or self.object_ids):109            target_folder = drive.get_root_folder()110            if not isinstance(target_folder, Folder):111                raise ValueError("Unable to fetch root folder")112            for blob in self._load_from_folder(target_folder):113                file_id = str(blob.metadata.get("id"))114                if self.load_auth is True:115                    auth_identities = self.authorized_identities(file_id)116                if self.load_extended_metadata is True:117                    extended_metadata = self.get_extended_metadata(file_id)118                for blob_part in self._blob_parser.lazy_parse(blob):119                    blob_part.metadata.update(blob.metadata)120                    if self.load_auth is True:121                        blob_part.metadata["authorized_identities"] = auth_identities122                    if self.load_extended_metadata is True:123                        blob_part.metadata.update(extended_metadata)124                        blob_part.metadata.update(125                            {"source_full_url": target_folder.web_url}126                        )127                    yield blob_part128 129    def authorized_identities(self, file_id: str) -> List:130        """131        Retrieve the access identities (user/group emails) for a given file.132        Args:133            file_id (str): The ID of the file.134        Returns:135            List: A list of group names (email addresses) that have136                  access to the file.137        """138        data = self._fetch_access_token()139        access_token = data.get("access_token")140        url = (141            "https://graph.microsoft.com/v1.0/drives"142            f"/{self.document_library_id}/items/{file_id}/permissions"143        )144        headers = {"Authorization": f"Bearer {access_token}"}145        response = requests.request("GET", url, headers=headers)146        access_list = response.json()147 148        group_names = []149 150        for access_data in access_list.get("value"):151            if access_data.get("grantedToV2"):152                site_data = (153                    (access_data.get("grantedToV2").get("siteUser"))154                    or (access_data.get("grantedToV2").get("user"))155                    or (access_data.get("grantedToV2").get("group"))156                )157                if site_data:158                    email = site_data.get("email")159                    if email:160                        group_names.append(email)161        return group_names162 163    def _fetch_access_token(self) -> Any:164        """165        Fetch the access token from the token file.166        Returns:167            The access token as a dictionary.168        """169        with open(self.token_path, encoding="utf-8") as f:170            s = f.read()171        data = json.loads(s)172        return data173 174    def get_extended_metadata(self, file_id: str) -> Dict:175        """176        Retrieve extended metadata for a file in SharePoint.177        As of today, following fields are supported in the extended metadata:178        - size: size of the source file.179        - owner: display name of the owner of the source file.180        - full_path: pretty human readable path of the source file.181        Args:182            file_id (str): The ID of the file.183        Returns:184            `dict` containing the extended metadata of the file, including size, owner,185                and full path.186        """187        data = self._fetch_access_token()188        access_token = data.get("access_token")189        url = (190            "https://graph.microsoft.com/v1.0/drives/"191            f"{self.document_library_id}/items/{file_id}"192            "?$select=size,createdBy,parentReference,name"193        )194        headers = {"Authorization": f"Bearer {access_token}"}195        response = requests.request("GET", url, headers=headers)196        metadata = response.json()197        staged_metadata = {198            "size": metadata.get("size", 0),199            "owner": metadata.get("createdBy", {})200            .get("user", {})201            .get("displayName", ""),202            "full_path": metadata.get("parentReference", {})203            .get("path", "")204            .split(":")[-1]205            + "/"206            + metadata.get("name", ""),207        }208        return staged_metadata209