521 lines
No EOL
22 KiB
Python
521 lines
No EOL
22 KiB
Python
"""
|
|
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.chat.methodBase import MethodBase, ActionResult, action
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class MethodSharepoint(MethodBase):
|
|
"""SharePoint method implementation for document operations"""
|
|
|
|
def __init__(self, serviceCenter: Any):
|
|
super().__init__(serviceCenter)
|
|
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")
|
|
expectedDocumentFormats (list, optional): Expected document formats with extension, mimeType, description
|
|
"""
|
|
try:
|
|
connectionReference = parameters.get("connectionReference")
|
|
siteUrl = parameters.get("siteUrl")
|
|
query = parameters.get("query")
|
|
searchScope = parameters.get("searchScope", "all")
|
|
expectedDocumentFormats = parameters.get("expectedDocumentFormats", [])
|
|
|
|
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()
|
|
}
|
|
|
|
# Determine output format based on expected formats
|
|
output_extension = ".json" # Default
|
|
output_mime_type = "application/json" # Default
|
|
|
|
if expectedDocumentFormats and len(expectedDocumentFormats) > 0:
|
|
# Use the first expected format
|
|
expected_format = expectedDocumentFormats[0]
|
|
output_extension = expected_format.get("extension", ".json")
|
|
output_mime_type = expected_format.get("mimeType", "application/json")
|
|
logger.info(f"Using expected format: {output_extension} ({output_mime_type})")
|
|
else:
|
|
logger.info("No expected format specified, using default .json format")
|
|
|
|
return self._createResult(
|
|
success=True,
|
|
data={
|
|
"documents": [
|
|
{
|
|
"documentName": f"sharepoint_find_path_{datetime.now(UTC).strftime('%Y%m%d_%H%M%S')}{output_extension}",
|
|
"documentData": result_data,
|
|
"mimeType": output_mime_type
|
|
}
|
|
]
|
|
}
|
|
)
|
|
|
|
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)
|
|
expectedDocumentFormats (list, optional): Expected document formats with extension, mimeType, description
|
|
"""
|
|
try:
|
|
documentList = parameters.get("documentList")
|
|
connectionReference = parameters.get("connectionReference")
|
|
siteUrl = parameters.get("siteUrl")
|
|
documentPaths = parameters.get("documentPaths")
|
|
includeMetadata = parameters.get("includeMetadata", True)
|
|
expectedDocumentFormats = parameters.get("expectedDocumentFormats", [])
|
|
|
|
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()
|
|
}
|
|
|
|
# Determine output format based on expected formats
|
|
output_extension = ".json" # Default
|
|
output_mime_type = "application/json" # Default
|
|
|
|
if expectedDocumentFormats and len(expectedDocumentFormats) > 0:
|
|
# Use the first expected format
|
|
expected_format = expectedDocumentFormats[0]
|
|
output_extension = expected_format.get("extension", ".json")
|
|
output_mime_type = expected_format.get("mimeType", "application/json")
|
|
logger.info(f"Using expected format: {output_extension} ({output_mime_type})")
|
|
else:
|
|
logger.info("No expected format specified, using default .json format")
|
|
|
|
return self._createResult(
|
|
success=True,
|
|
data={
|
|
"documents": [
|
|
{
|
|
"documentName": f"sharepoint_documents_{datetime.now(UTC).strftime('%Y%m%d_%H%M%S')}{output_extension}",
|
|
"documentData": result_data,
|
|
"mimeType": output_mime_type
|
|
}
|
|
]
|
|
}
|
|
)
|
|
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
|
|
expectedDocumentFormats (list, optional): Expected document formats with extension, mimeType, description
|
|
"""
|
|
try:
|
|
connectionReference = parameters.get("connectionReference")
|
|
siteUrl = parameters.get("siteUrl")
|
|
documentPaths = parameters.get("documentPaths")
|
|
documentList = parameters.get("documentList")
|
|
fileNames = parameters.get("fileNames")
|
|
expectedDocumentFormats = parameters.get("expectedDocumentFormats", [])
|
|
|
|
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()
|
|
}
|
|
|
|
# Determine output format based on expected formats
|
|
output_extension = ".json" # Default
|
|
output_mime_type = "application/json" # Default
|
|
|
|
if expectedDocumentFormats and len(expectedDocumentFormats) > 0:
|
|
# Use the first expected format
|
|
expected_format = expectedDocumentFormats[0]
|
|
output_extension = expected_format.get("extension", ".json")
|
|
output_mime_type = expected_format.get("mimeType", "application/json")
|
|
logger.info(f"Using expected format: {output_extension} ({output_mime_type})")
|
|
else:
|
|
logger.info("No expected format specified, using default .json format")
|
|
|
|
return self._createResult(
|
|
success=True,
|
|
data={
|
|
"documents": [
|
|
{
|
|
"documentName": f"sharepoint_upload_{datetime.now(UTC).strftime('%Y%m%d_%H%M%S')}{output_extension}",
|
|
"documentData": result_data,
|
|
"mimeType": output_mime_type
|
|
}
|
|
]
|
|
}
|
|
)
|
|
|
|
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)
|
|
expectedDocumentFormats (list, optional): Expected document formats with extension, mimeType, description
|
|
"""
|
|
try:
|
|
connectionReference = parameters.get("connectionReference")
|
|
siteUrl = parameters.get("siteUrl")
|
|
folderPaths = parameters.get("folderPaths")
|
|
includeSubfolders = parameters.get("includeSubfolders", False)
|
|
expectedDocumentFormats = parameters.get("expectedDocumentFormats", [])
|
|
|
|
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()
|
|
}
|
|
|
|
# Determine output format based on expected formats
|
|
output_extension = ".json" # Default
|
|
output_mime_type = "application/json" # Default
|
|
|
|
if expectedDocumentFormats and len(expectedDocumentFormats) > 0:
|
|
# Use the first expected format
|
|
expected_format = expectedDocumentFormats[0]
|
|
output_extension = expected_format.get("extension", ".json")
|
|
output_mime_type = expected_format.get("mimeType", "application/json")
|
|
logger.info(f"Using expected format: {output_extension} ({output_mime_type})")
|
|
else:
|
|
logger.info("No expected format specified, using default .json format")
|
|
|
|
return self._createResult(
|
|
success=True,
|
|
data={
|
|
"documents": [
|
|
{
|
|
"documentName": f"sharepoint_document_list_{datetime.now(UTC).strftime('%Y%m%d_%H%M%S')}{output_extension}",
|
|
"documentData": result_data,
|
|
"mimeType": output_mime_type
|
|
}
|
|
]
|
|
}
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error listing SharePoint documents: {str(e)}")
|
|
return self._createResult(
|
|
success=False,
|
|
data={},
|
|
error=str(e)
|
|
) |