# # Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one # or more contributor license agreements. Licensed under the Elastic License 2.0; # you may not use this file except in compliance with the Elastic License 2.0. # """ This module fetches all the messages, attachments, chat tabs, and meeting recordings from Microsoft Teams. """ from collections import defaultdict import requests from . import constant from .microsoft_teams_client import MSTeamsClient from .utils import (extract_api_response, get_schema_fields, html_to_text, url_encode) USER_CHAT_ATTACHMENT = "User Chat Attachments" MEETING_RECORDING = "Meeting Recording" USER_CHAT_TABS = "User Chat Tabs" class MSTeamsUserMessage: """Fetches users details from the Microsoft Teams.""" def __init__(self, access_token, logger, config, local_storage): self.token = access_token self.client = MSTeamsClient(logger, self.token, config) self.logger = logger self.is_permission_sync_enabled = config.get_value("enable_document_permission") self.config = config self.object_type_to_index = config.get_value('object_type_to_index') self.local_storage = local_storage def get_attachments( self, user_id, prefix, attachment_name, attachment_id, chat_id, updated_date, ids_list, user_drive, attachment_client ): """Fetches all the attachments of a user chat :param user_id: Id of the user :param prefix: Title of the chat message :param attachment_name: Name of the attachment :param attachment_id: Id of the attachment :param chat_id: Id of chat :param updated_date: date of chat updated :param ids_list: List of ids :param user_drive: Dictionary of user id with drive id :param attachment_client: Object of Microsoft team client Returns: attachment_list: Documents to be indexed in Workplace Search """ try: item_id = None # Checking if the user_id is present in user_drive dictionary for saving the else iteration. If user_id is # present in the dictionary then we'll directly use its drive_id for fetching attachments. if user_drive.get(user_id): drive_id = list(user_drive[user_id].keys())[0] else: user_drive_response = attachment_client.get_user_chat_attachment_drive( f"{constant.GRAPH_BASE_URL}/users/{user_id}/drive" ) if user_drive_response: # Logic to append user for deletion self.local_storage.insert_document_into_doc_id_storage( ids_list, user_id, constant.USER, "", "" ) drive_id = user_drive_response["id"] user_drive[user_id] = {drive_id: None} # Logic to append user drive for deletion self.local_storage.insert_document_into_doc_id_storage( ids_list, drive_id, constant.USER_CHAT_DRIVE, user_id, "" ) if user_drive: # Check that item_id is present with the drive id or not for saving the else iteration. if user_drive.get(user_id).get(drive_id): item_id = user_drive.get(user_id).get(drive_id) else: user_root_response_data = attachment_client.get_user_chat_attachment_drive_children( f"{constant.GRAPH_BASE_URL}/drives/{drive_id}/items/root/children" ) if user_root_response_data: for child in user_root_response_data: if child["name"] == "Microsoft Teams Chat Files": item_id = child["id"] user_drive[user_id][drive_id] = item_id break attachment_list = [] if not item_id: return [] # Logic to append user drive item for deletion self.local_storage.insert_document_into_doc_id_storage( ids_list, item_id, constant.USER_CHAT_DRIVE_ITEM, drive_id, user_id ) final_attachment_url = f"{constant.GRAPH_BASE_URL}/drives/{drive_id}/items/{item_id}/children?" \ f"$filter=name eq '{url_encode(attachment_name)}'" attachment_response_data = attachment_client.get_user_chat_attachment_drive_children( final_attachment_url) if attachment_response_data: document = attachment_response_data[0] attachment_dict = {"type": USER_CHAT_ATTACHMENT} is_file = document.get("file", {}) if is_file and type(is_file) != float: mimetype = is_file.get("mimeType") if mimetype not in constant.MIMETYPES: attachment_content_response = requests.get( document.get("@microsoft.graph.downloadUrl") ) if attachment_content_response: attachment_content = extract_api_response( attachment_content_response.content ) attachment_dict["id"] = attachment_id attachment_dict["title"] = f"{prefix}-{attachment_name}" attachment_dict["body"] = attachment_content or "" attachment_dict["url"] = document.get("webUrl") attachment_dict["last_updated"] = updated_date attachment_dict["_allow_permissions"] = [] if self.is_permission_sync_enabled: attachment_dict["_allow_permissions"] = [chat_id] attachment_list.append(attachment_dict) # Logic to append user chat attachment for deletion self.local_storage.insert_document_into_doc_id_storage( ids_list, attachment_id, USER_CHAT_ATTACHMENT, item_id, drive_id, ) return attachment_list except Exception as exception: self.logger.exception( f"[Fail] Error while fetching attachments for the user chats. Error: {exception}" ) def fetch_tabs(self, chat_id, ids_list, start_time, end_time): """Fetches user chat tabs from the Microsoft Teams :param chat_id: Id of the chat :param ids_list: List of ids :param start_time: Starting time for fetching data :param end_time: Ending time for fetching data Returns: documents: Documents to be indexed in Workplace Search """ try: documents = [] tab_detail_response = self.client.get_user_chat_tabs( f"{constant.GRAPH_BASE_URL}/chats/{chat_id}/tabs", start_time, end_time, chat_id ) if tab_detail_response: tab_schema = get_schema_fields("user_tabs", self.object_type_to_index) for tab in tab_detail_response: tab_dict = {"type": USER_CHAT_TABS} for ws_field, ms_fields in tab_schema.items(): tab_dict[ws_field] = tab[ms_fields] tab_dict["url"] = tab["configuration"]["websiteUrl"] tab_dict["_allow_permissions"] = [] if self.is_permission_sync_enabled: tab_dict["_allow_permissions"] = [chat_id] documents.append(tab_dict) self.local_storage.insert_document_into_doc_id_storage( ids_list, tab["id"], USER_CHAT_TABS, chat_id, "" ) return documents except Exception as exception: self.logger.exception( f"[Fail] Error while fetching user tabs from teams. Error: {exception}" ) raise def fetch_meeting_recording(self, chat_id, chat): """Fetches meeting recording from the Microsoft Teams :param chat_id: Id of the chat :param chat: dictionary of the user chat Returns: recording_dict: Document to be indexed in Workplace Search """ if ( chat["eventDetail"] and chat["eventDetail"][ "@odata.type"] == "#microsoft.graph.callRecordingEventMessageDetail" ): url = chat["eventDetail"].get("callRecordingUrl") if url and ".sharepoint.com" in url: recording_dict = {"type": MEETING_RECORDING} recording_dict["id"] = chat["eventDetail"]["callId"] recording_dict["title"] = chat["eventDetail"][ "callRecordingDisplayName" ] recording_dict["url"] = url recording_dict["_allow_permissions"] = [] if self.is_permission_sync_enabled: recording_dict["_allow_permissions"] = [chat_id] return recording_dict def get_user_chats(self, ids_list): """Fetches user chats by calling '/Chats' api :param ids_list: List of ids Returns: member_dict: List of dictionaries containing chat id and their members documents: Documents to be indexed in Workplace Search """ self.logger.debug("Fetching the users chats") documents = [] chat_response_data = self.client.get_user_chats(f"{constant.GRAPH_BASE_URL}/chats?$expand=members") if chat_response_data: self.logger.info( "Fetched the user chat metadata. Attempting to extract the messages from the chats, " "attachments and meeting recordings.." ) # member_dict: Dictionary of members with their id for adding permissions member_dict = defaultdict(list) for chat in chat_response_data: for member in chat["members"]: display_name = member["displayName"] if display_name: member_dict[display_name].append(chat["id"]) # Logic to append chat for deletion self.local_storage.insert_document_into_doc_id_storage( ids_list, chat["id"], constant.CHATS, "", "" ) documents.append(chat) return member_dict, documents def get_user_chat_messages( self, ids_list, user_drive, chat_response_data, start_time, end_time, user_attachment_token, ): """Fetches the user chat messages from Microsoft Teams :param ids_list: List of ids :param user_drive: Dictionary of dictionary :param chat_response_data: Chats data for fetching chat messages :param start_time: Starting time for fetching data :param end_time: Ending time for fetching data :param user_attachment_token: Access token for fetching the attachments Returns: documents: Documents to be indexed in Workplace Search """ documents = [] user_schema = get_schema_fields("user_chats", self.object_type_to_index) attachment_client = MSTeamsClient( self.logger, user_attachment_token, self.config ) for val in chat_response_data: member_title = [] for member in val["members"]: display_name = member["displayName"] if display_name: member_title.append(display_name) # Logic to append chat for deletion try: chat_detail_response = self.client.get_user_chat_messages( f'{constant.GRAPH_BASE_URL}/chats/{val["id"]}/messages', start_time, end_time, val['id'] ) if chat_detail_response: for chat in chat_detail_response: if not chat["deletedDateTime"]: title = ( val.get("topic") if val.get("topic") else ",".join(member_title) ) sender = chat["from"] user_name = "" if sender and sender["user"]: user_id = sender.get("user", {}).get("id") user_name = sender.get("user", {}).get("displayName") for attachment in chat["attachments"]: name = attachment["name"] if name and attachment["contentType"] == "reference": attachment_document = self.get_attachments( user_id, title, name, attachment["id"], val["id"], chat["lastModifiedDateTime"], ids_list, user_drive, attachment_client ) if attachment_document: documents.extend(attachment_document) content = chat["body"]["content"] chat_message = html_to_text(self.logger, content) if chat_message: # Logic to append chat message for deletion self.local_storage.insert_document_into_doc_id_storage( ids_list, chat["id"], constant.USER_CHATS_MESSAGE, val["id"], "", ) user_dict = {"type": constant.USER_CHATS_MESSAGE} for ws_field, ms_fields in user_schema.items(): user_dict[ws_field] = chat[ms_fields] user_dict["title"] = title user_dict["body"] = ( f"{user_name} - {chat_message}" if user_name else chat_message ) user_dict["url"] = val["webUrl"] user_dict["_allow_permissions"] = [] if self.is_permission_sync_enabled: user_dict["_allow_permissions"] = [val["id"]] documents.append(user_dict) else: self.logger.info( f"the message for the chat {chat['id']} is empty" ) meeting_recordings = self.fetch_meeting_recording( val["id"], chat ) if meeting_recordings: documents.append(meeting_recordings) except Exception as exception: self.logger.exception( f"[Fail] Error while fetching user chats details from teams. Error: {exception}" ) raise self.logger.info( f"Fetched chats, attachments and meeting recordings metadata. Attempting to fetch tabs " f"for chat: {val['id']}" ) tabs_document = self.fetch_tabs(val["id"], ids_list, start_time, end_time) documents.extend(tabs_document) self.logger.info("Fetched the user chat tabs") return documents