""" SharePoint method module. Handles SharePoint operations using the SharePoint service. """ import logging from typing import Dict, Any, List, Optional from datetime import datetime, UTC import json import uuid from modules.workflow.methodBase import MethodBase, ActionResult, action logger = logging.getLogger(__name__) class MethodSharepoint(MethodBase): """SharePoint method implementation for document operations""" def __init__(self, serviceContainer: Any): super().__init__(serviceContainer) self.name = "sharepoint" self.description = "Handle Microsoft SharePoint document operations" def _getMicrosoftConnection(self, connectionReference: str) -> Optional[Dict[str, Any]]: """Get Microsoft connection from connection reference""" try: userConnection = self.service.getUserConnectionFromConnectionReference(connectionReference) if not userConnection or userConnection.authority != "msft" or userConnection.status != "active": return None # Get the corresponding token for this user and authority token = self.service.interfaceApp.getToken(userConnection.authority) if not token: logger.warning(f"No token found for user {userConnection.userId} and authority {userConnection.authority}") return None return { "id": userConnection.id, "accessToken": token.tokenAccess, "refreshToken": token.tokenRefresh, "scopes": ["Sites.ReadWrite.All", "User.Read"] # Default Microsoft scopes } except Exception as e: logger.error(f"Error getting Microsoft connection: {str(e)}") return None @action async def findDocumentPath(self, parameters: Dict[str, Any]) -> ActionResult: """ Find document path based on query/description Parameters: connectionReference (str): Reference to the Microsoft connection siteUrl (str): SharePoint site URL query (str): Query or description to find document searchScope (str, optional): Search scope (default: "all") """ try: connectionReference = parameters.get("connectionReference") siteUrl = parameters.get("siteUrl") query = parameters.get("query") searchScope = parameters.get("searchScope", "all") if not connectionReference or not siteUrl or not query: return self._createResult( success=False, data={}, error="Connection reference, site URL, and query are required" ) connection = self._getMicrosoftConnection(connectionReference) if not connection: return self._createResult( success=False, data={}, error="No valid Microsoft connection found for the provided connection reference" ) find_prompt = f""" Simulate finding document paths in Microsoft SharePoint based on a query. Connection: {connection['id']} Site URL: {siteUrl} Query: {query} Search Scope: {searchScope} Please provide: 1. Matching document paths and locations 2. Relevance scores for each match 3. Document metadata and properties 4. Alternative search suggestions 5. Search statistics and coverage """ find_result = await self.service.interfaceAiCalls.callAiTextAdvanced(find_prompt) result_data = { "connectionReference": connectionReference, "siteUrl": siteUrl, "query": query, "searchScope": searchScope, "findResult": find_result, "connection": { "id": connection["id"], "authority": "microsoft", "reference": connectionReference }, "timestamp": datetime.now(UTC).isoformat() } return self._createResult( success=True, data={ "documents": [ { "documentName": f"sharepoint_find_path_{datetime.now(UTC).strftime('%Y%m%d_%H%M%S')}.json", "documentData": result_data } ] } ) except Exception as e: logger.error(f"Error finding document path: {str(e)}") return self._createResult( success=False, data={}, error=str(e) ) @action async def readDocument(self, parameters: Dict[str, Any]) -> ActionResult: """ Read documents from SharePoint Parameters: documentList (str): Reference to the document list to read connectionReference (str): Reference to the Microsoft connection siteUrl (str): SharePoint site URL documentPaths (List[str]): List of paths to the documents in SharePoint includeMetadata (bool, optional): Whether to include metadata (default: True) """ try: documentList = parameters.get("documentList") connectionReference = parameters.get("connectionReference") siteUrl = parameters.get("siteUrl") documentPaths = parameters.get("documentPaths") includeMetadata = parameters.get("includeMetadata", True) if not documentList or not connectionReference or not siteUrl or not documentPaths: return self._createResult( success=False, data={}, error="Document list reference, connection reference, site URL, and document paths are required" ) # Get documents from reference chatDocuments = self.service.getChatDocumentsFromDocumentList(documentList) if not chatDocuments: return self._createResult( success=False, data={}, error="No documents found for the provided reference" ) connection = self._getMicrosoftConnection(connectionReference) if not connection: return self._createResult( success=False, data={}, error="No valid Microsoft connection found for the provided connection reference" ) # Process each document path read_results = [] for i, documentPath in enumerate(documentPaths): if i < len(chatDocuments): chatDocument = chatDocuments[i] fileId = chatDocument.fileId sharepoint_prompt = f""" Simulate reading a document from Microsoft SharePoint. Connection: {connection['id']} Site URL: {siteUrl} Document Path: {documentPath} Include Metadata: {includeMetadata} File ID: {fileId} Please provide: 1. Document content and structure 2. File metadata and properties 3. SharePoint site information 4. Document permissions and sharing 5. Version history if available """ document_data = await self.service.interfaceAiCalls.callAiTextAdvanced(sharepoint_prompt) read_results.append({ "documentPath": documentPath, "fileId": fileId, "documentContent": document_data }) result_data = { "connectionReference": connectionReference, "siteUrl": siteUrl, "documentPaths": documentPaths, "includeMetadata": includeMetadata, "readResults": read_results, "connection": { "id": connection["id"], "authority": "microsoft", "reference": connectionReference }, "timestamp": datetime.now(UTC).isoformat() } return self._createResult( success=True, data={ "documents": [ { "documentName": f"sharepoint_documents_{datetime.now(UTC).strftime('%Y%m%d_%H%M%S')}.json", "documentData": result_data } ] } ) except Exception as e: logger.error(f"Error reading SharePoint documents: {str(e)}") return self._createResult( success=False, data={}, error=str(e) ) @action async def uploadDocument(self, parameters: Dict[str, Any]) -> ActionResult: """ Upload documents to SharePoint Parameters: connectionReference (str): Reference to the Microsoft connection siteUrl (str): SharePoint site URL documentPaths (List[str]): List of paths where to upload the documents documentList (str): Reference to the document list to upload fileNames (List[str]): List of names for the uploaded files """ try: connectionReference = parameters.get("connectionReference") siteUrl = parameters.get("siteUrl") documentPaths = parameters.get("documentPaths") documentList = parameters.get("documentList") fileNames = parameters.get("fileNames") if not connectionReference or not siteUrl or not documentPaths or not documentList or not fileNames: return self._createResult( success=False, data={}, error="Connection reference, site URL, document paths, document list, and file names are required" ) # Get Microsoft connection connection = self._getMicrosoftConnection(connectionReference) if not connection: return self._createResult( success=False, data={}, error="No valid Microsoft connection found for the provided connection reference" ) # Get documents from reference chatDocuments = self.service.getChatDocumentsFromDocumentList(documentList) if not chatDocuments: return self._createResult( success=False, data={}, error="No documents found for the provided reference" ) # Process each document upload upload_results = [] for i, (documentPath, fileName) in enumerate(zip(documentPaths, fileNames)): if i < len(chatDocuments): chatDocument = chatDocuments[i] fileId = chatDocument.fileId file_data = self.service.getFileData(fileId) if not file_data: logger.warning(f"File data not found for fileId: {fileId}") continue # Create SharePoint upload prompt upload_prompt = f""" Simulate uploading a document to Microsoft SharePoint. Connection: {connection['id']} Site URL: {siteUrl} Document Path: {documentPath} File Name: {fileName} File ID: {fileId} File Size: {len(file_data)} bytes Please provide: 1. Upload confirmation and status 2. File metadata and properties 3. SharePoint site integration details 4. Permission and sharing settings 5. Version control information """ # Use AI to simulate SharePoint upload upload_result = await self.service.interfaceAiCalls.callAiTextAdvanced(upload_prompt) upload_results.append({ "documentPath": documentPath, "fileName": fileName, "fileId": fileId, "uploadResult": upload_result }) # Create result data result_data = { "connectionReference": connectionReference, "siteUrl": siteUrl, "documentPaths": documentPaths, "documentList": documentList, "fileNames": fileNames, "uploadResults": upload_results, "connection": { "id": connection["id"], "authority": "microsoft", "reference": connectionReference }, "timestamp": datetime.now(UTC).isoformat() } return self._createResult( success=True, data={ "documents": [ { "documentName": f"sharepoint_upload_{datetime.now(UTC).strftime('%Y%m%d_%H%M%S')}.json", "documentData": result_data } ] } ) except Exception as e: logger.error(f"Error uploading to SharePoint: {str(e)}") return self._createResult( success=False, data={}, error=str(e) ) @action async def listDocuments(self, parameters: Dict[str, Any]) -> ActionResult: """ List documents in SharePoint folder Parameters: connectionReference (str): Reference to the Microsoft connection siteUrl (str): SharePoint site URL folderPaths (List[str]): List of paths to the folders to list includeSubfolders (bool, optional): Whether to include subfolders (default: False) """ try: connectionReference = parameters.get("connectionReference") siteUrl = parameters.get("siteUrl") folderPaths = parameters.get("folderPaths") includeSubfolders = parameters.get("includeSubfolders", False) if not connectionReference or not siteUrl or not folderPaths: return self._createResult( success=False, data={}, error="Connection reference, site URL, and folder paths are required" ) # Get Microsoft connection connection = self._getMicrosoftConnection(connectionReference) if not connection: return self._createResult( success=False, data={}, error="No valid Microsoft connection found for the provided connection reference" ) # Process each folder path list_results = [] for folderPath in folderPaths: # Create SharePoint listing prompt list_prompt = f""" Simulate listing documents in Microsoft SharePoint folder. Connection: {connection['id']} Site URL: {siteUrl} Folder Path: {folderPath} Include Subfolders: {includeSubfolders} Please provide: 1. List of documents and folders 2. File metadata and properties 3. Folder structure and hierarchy 4. Permission and sharing information 5. Document statistics and summary """ # Use AI to simulate SharePoint listing list_result = await self.service.interfaceAiCalls.callAiTextAdvanced(list_prompt) list_results.append({ "folderPath": folderPath, "listResult": list_result }) # Create result data result_data = { "connectionReference": connectionReference, "siteUrl": siteUrl, "folderPaths": folderPaths, "includeSubfolders": includeSubfolders, "listResults": list_results, "connection": { "id": connection["id"], "authority": "microsoft", "reference": connectionReference }, "timestamp": datetime.now(UTC).isoformat() } return self._createResult( success=True, data={ "documents": [ { "documentName": f"sharepoint_document_list_{datetime.now(UTC).strftime('%Y%m%d_%H%M%S')}.json", "documentData": result_data } ] } ) except Exception as e: logger.error(f"Error listing SharePoint documents: {str(e)}") return self._createResult( success=False, data={}, error=str(e) )