# Copyright (c) 2025 Patrick Motsch # All rights reserved. import logging import time import json import requests from typing import Dict, Any from modules.datamodels.datamodelChat import ActionResult, ActionDocument logger = logging.getLogger(__name__) async def readEmails(self, parameters: Dict[str, Any]) -> ActionResult: operationId = None try: # Init progress logger workflowId = self.services.workflow.id if self.services.workflow else f"no-workflow-{int(time.time())}" operationId = f"outlook_read_{workflowId}_{int(time.time())}" # Start progress tracking parentOperationId = parameters.get('parentOperationId') self.services.chat.progressLogStart( operationId, "Read Emails", "Outlook Email Reading", f"Folder: {parameters.get('folder', 'Inbox')}", parentOperationId=parentOperationId ) connectionReference = parameters.get("connectionReference") folder = parameters.get("folder", "Inbox") limit = parameters.get("limit", 1000) filter = parameters.get("filter") outputMimeType = parameters.get("outputMimeType", "application/json") if not connectionReference: if operationId: self.services.chat.progressLogFinish(operationId, False) return ActionResult.isFailure(error="Connection reference is required") self.services.chat.progressLogUpdate(operationId, 0.2, "Validating parameters") # Validate limit parameter if limit <= 0: limit = 1000 logger.warning(f"Invalid limit value ({limit}), using default value 1000") # Validate filter parameter if provided if filter: # Remove any potentially dangerous characters that could break the filter filter = filter.strip() if len(filter) > 500: logger.warning(f"Filter too long ({len(filter)} chars), truncating to 500 characters") filter = filter[:500] # Get Microsoft connection self.services.chat.progressLogUpdate(operationId, 0.3, "Getting Microsoft connection") connection = self.connection.getMicrosoftConnection(connectionReference) if not connection: self.services.chat.progressLogFinish(operationId, False) return ActionResult.isFailure(error="No valid Microsoft connection found for the provided connection reference") # Read emails using Microsoft Graph API self.services.chat.progressLogUpdate(operationId, 0.4, "Reading emails from Microsoft Graph API") try: # Microsoft Graph API endpoint for messages graph_url = "https://graph.microsoft.com/v1.0" headers = { "Authorization": f"Bearer {connection['accessToken']}", "Content-Type": "application/json" } # Get the folder ID for the specified folder folder_id = self.folderManagement.getFolderId(folder, connection) if folder_id: # Build the API request with folder ID api_url = f"{graph_url}/me/mailFolders/{folder_id}/messages" else: # Fallback: use folder name directly (for well-known folders like "Inbox") api_url = f"{graph_url}/me/mailFolders/{folder}/messages" logger.warning(f"Could not find folder ID for '{folder}', using folder name directly") params = { "$top": limit, "$orderby": "receivedDateTime desc" } if filter: # Build proper Graph API filter parameters filter_params = self.emailProcessing.buildGraphFilter(filter) params.update(filter_params) # If using $search, remove $orderby as they can't be combined if "$search" in params: params.pop("$orderby", None) # If using $filter with contains(), remove $orderby as they can't be combined # Microsoft Graph API doesn't support contains() with orderby if "$filter" in params and "contains(" in params["$filter"].lower(): params.pop("$orderby", None) # Filter applied # Make the API call response = requests.get(api_url, headers=headers, params=params) if response.status_code != 200: logger.error(f"Graph API error: {response.status_code} - {response.text}") logger.error(f"Request URL: {response.url}") logger.error(f"Request params: {params}") response.raise_for_status() self.services.chat.progressLogUpdate(operationId, 0.7, "Processing email data") emails_data = response.json() email_data = { "emails": emails_data.get("value", []), "count": len(emails_data.get("value", [])), "folder": folder, "filter": filter, "apiMetadata": { "@odata.context": emails_data.get("@odata.context"), "@odata.count": emails_data.get("@odata.count"), "@odata.nextLink": emails_data.get("@odata.nextLink") } } except ImportError: logger.error("requests module not available") if operationId: self.services.chat.progressLogFinish(operationId, False) return ActionResult.isFailure(error="requests module not available") except requests.exceptions.HTTPError as e: if e.response.status_code == 400: logger.error(f"Bad Request (400) - Invalid filter or parameter: {e.response.text}") if operationId: self.services.chat.progressLogFinish(operationId, False) return ActionResult.isFailure(error=f"Invalid filter syntax. Please check your filter parameter. Error: {e.response.text}") elif e.response.status_code == 401: logger.error("Unauthorized (401) - Access token may be expired or invalid") if operationId: self.services.chat.progressLogFinish(operationId, False) return ActionResult.isFailure(error="Authentication failed. Please check your connection and try again.") elif e.response.status_code == 403: logger.error("Forbidden (403) - Insufficient permissions to access emails") if operationId: self.services.chat.progressLogFinish(operationId, False) return ActionResult.isFailure(error="Insufficient permissions to read emails from this folder.") else: logger.error(f"HTTP Error {e.response.status_code}: {e.response.text}") if operationId: self.services.chat.progressLogFinish(operationId, False) return ActionResult.isFailure(error=f"HTTP Error {e.response.status_code}: {e.response.text}") except Exception as e: logger.error(f"Error reading emails from Microsoft Graph API: {str(e)}") if operationId: self.services.chat.progressLogFinish(operationId, False) return ActionResult.isFailure(error=f"Failed to read emails: {str(e)}") # Determine output format based on MIME type mime_type_mapping = { "application/json": ".json", "text/plain": ".txt", "text/csv": ".csv" } output_extension = mime_type_mapping.get(outputMimeType, ".json") output_mime_type = outputMimeType logger.info(f"Using output format: {output_extension} ({output_mime_type})") # Create result data as JSON string result_data = { "connectionReference": connectionReference, "folder": folder, "limit": limit, "filter": filter, "emails": email_data, "connection": { "id": connection["id"], "authority": "microsoft", "reference": connectionReference }, "timestamp": self.services.utils.timestampGetUtc() } validationMetadata = { "actionType": "outlook.readEmails", "connectionReference": connectionReference, "folder": folder, "limit": limit, "filter": filter, "emailCount": email_data.get("count", 0), "outputMimeType": outputMimeType } self.services.chat.progressLogUpdate(operationId, 0.9, f"Found {email_data.get('count', 0)} emails") self.services.chat.progressLogFinish(operationId, True) return ActionResult.isSuccess( documents=[ActionDocument( documentName=f"outlook_emails_{self._format_timestamp_for_filename()}.json", documentData=json.dumps(result_data, indent=2), mimeType="application/json", validationMetadata=validationMetadata )] ) except Exception as e: logger.error(f"Error reading emails: {str(e)}") if operationId: try: self.services.chat.progressLogFinish(operationId, False) except Exception: pass return ActionResult.isFailure( error=str(e) )