# By: Riasat Ullah # This file contains Microsoft Teams integration related variables and function. from cache_queries import cache_task_instances from dbqueries import db_integrations, db_services, db_users from dbqueries.integrations import db_microsoft_teams from taskcallrest import settings from threading import Thread from translators import label_translator as _lt from utils import constants, errors, helpers, info, integration_type_names as intt, key_manager, label_names as lnm,\ logging, s3, times, tokenizer, url_paths, var_names import configuration import datetime import json import requests # MS Teams S3 credential file variables ms_teams_s3_bucket = 'taskcall-prod-data' ms_teams_s3_key = 'credentials/microsoft_teams_credentials.json' # MS Teams API url formats url_access_token = 'https://login.microsoftonline.com/botframework.com/oauth2/v2.0/token' url_proactive_message = '{0}/v3/conversations' url_reply_message = '{0}/v3/conversations/{1}/activities/{2}' url_ms_teams_consent = 'https://login.microsoftonline.com/{0}/adminconsent?client_id={1}&state={2}&redirect_uri={3}' url_tc_ms_teams_authorization = settings.WEB_APP_BASE_URL +\ '/configurations/services/integrations/microsoft-teams/authorize' url_tc_ms_teams_integration_guide = 'https://docs.taskcallapp.com/integrations/v1/microsoft-teams-integration-guide' # MS Teams payload types payload_type_installation_update = 'installationUpdate' payload_type_message = 'message' # MS Teams commands ms_teams_command = 'TaskCall' ms_teams_bot_command = 'TaskCall Bot' ms_teams_create_incident_cmd = 'create-incident' ms_teams_get_user_cmd = 'get-user' ms_teams_help_cmd = 'help' ms_teams_remove_user_cmd = 'remove-user' ms_teams_request_authorization = 'request-authorization' ms_teams_verify_user_cmd = 'verify-user' ms_teams_valid_instructions = [ms_teams_create_incident_cmd, ms_teams_get_user_cmd, ms_teams_help_cmd, ms_teams_remove_user_cmd, ms_teams_request_authorization, ms_teams_verify_user_cmd] # MS Teams string variables var_id = 'id' var_join_url = 'joinUrl' var_ms_teams_message_id = 'ms_teams_message_id' var_service_url = 'serviceUrl' var_tenant_id = 'tenant_id' # MS Teams error messages ms_teams_api_error = 'MS Teams could not process request. Please check if response-url is still valid.' unknown_action_error = 'Unknown action received. Cannot process request.' unknown_command_error = 'Unknown command was received. Cannot process request' # MS Teams conference bridge API path formats url_tenant_temp_token = 'https://login.microsoftonline.com/{0}/oauth2/v2.0/token' url_online_meetings = 'https://graph.microsoft.com/v1.0/users/{0}/onlineMeetings' def get_ms_teams_credentials(): ''' Get the credentials needed for handling and making API calls for MS Teams. :return: (dict) of credentials ''' creds = s3.read_json(ms_teams_s3_bucket, ms_teams_s3_key) return creds def get_ms_teams_request_headers(access_token): ''' Get the headers to be sent with MS Teams requests. :param access_token: (str) bearer access token :return: (dict) of request header ''' headers = {'Content-type': constants.content_type_json, 'Authorization': 'Bearer ' + access_token} return headers def renew_ms_teams_access_token(): ''' Renews MS Teams access token. :return: new access token; None if renewal fails ''' creds = get_ms_teams_credentials() content = { 'grant_type': 'client_credentials', 'client_id': creds[var_names.bot_id], 'client_secret': creds[var_names.bot_secret_value], 'scope': 'https://api.botframework.com/.default' } response = requests.post(url_access_token, content) if response.status_code == 200: access_token = response.json()[var_names.access_token] creds[var_names.access_token] = access_token s3.update_json(ms_teams_s3_bucket, ms_teams_s3_key, creds) return access_token else: logging.error('Could not update MS Teams access token') logging.error(response.json()) return None def ms_teams_post_api_request(url, access_token, message): ''' Make a post request to a MS Teams API endpoint. :param url: API endpoint url :param access_token: access token :param message: message to send :return: (tuple) status code, json response ''' response = requests.post(url, headers=get_ms_teams_request_headers(access_token), data=json.dumps(message)) if response.status_code == 401: new_access_token = renew_ms_teams_access_token() if new_access_token is not None: new_response = requests.post(url, headers=get_ms_teams_request_headers(new_access_token), data=json.dumps(message)) return new_response.status_code, new_response.json() return response.status_code, response.json() def get_ms_teams_adaptive_reply_body(bot_id, channel_id, reply_to_id, adaptive_card): ''' Gets body to be sent as JSON response reply to a message sent by MS Teams. :param bot_id: ID of the TaskCall bot (this is the recipient ID in the original message) :param channel_id: the channel ID :param reply_to_id: ID of the original message :param adaptive_card: (dict) of the adaptive card bod to include as attachment :return: (dict) of final response ''' body = { 'type': 'message', 'bot': {'id': bot_id}, 'channelData': {'channel': {'id': channel_id}}, 'replyToId': reply_to_id, 'attachmentLayout': 'list', 'attachments': [ { 'contentType': 'application/vnd.microsoft.card.adaptive', 'content': adaptive_card } ] } return body def get_ms_teams_text_reply_body(bot_id, channel_id, reply_to_id, text): ''' Gets body to be sent as JSON response reply to a message sent by MS Teams. :param bot_id: ID of the TaskCall bot (this is the recipient ID in the original message) :param channel_id: ID of the channel :param reply_to_id: ID of the original message :param text: the text to send :return: (dict) of final response ''' body = { 'type': 'message', 'bot': {'id': bot_id}, 'channelData': {'channel': {'id': channel_id}}, 'replyToId': reply_to_id, 'text': text } return body def get_ms_teams_proactive_adaptive_body(bot_id, channel_id, adaptive_card): ''' Gets the body of the proactive message that will be sent as JSON to MS Teams. :param bot_id: TaskCall bot ID :param channel_id: ID of the channel to sent to :param adaptive_card: (dict) adaptive card :return: (dict) of message to be sent ''' body = { 'bot': {'id': bot_id}, 'isGroup': True, 'topicName': 'New Incident', 'channelData': {'channel': {'id': channel_id}}, 'activity': { 'type': 'message', 'attachments': [ { 'contentType': 'application/vnd.microsoft.card.adaptive', 'content': adaptive_card, } ], 'entities': [] } } return body def get_ms_teams_proactive_text(bot_id, channel_id, text): ''' Gets the body of a simple proactive text message that will be sent as JSON to MS Teams. :param bot_id: TaskCall bot ID :param channel_id: ID of the channel to sent to :param text: (str) text message :return: (dict) of message to be sent ''' body = { 'bot': {'id': bot_id}, 'isGroup': True, 'topicName': 'New Message', 'channelData': {'channel': {'id': channel_id}}, 'activity': { 'type': 'message', 'text': text, 'attachments': [], 'entities': [] } } return body def get_incident_details_card(lang, instance_id, org_instance_id, title, text_msg, assigned_to_str, urgency, service_name, snapshots, conference_bridge=None): ''' Puts the incident details card together. :param lang: language the card should be in :param instance_id: ID of the instance :param org_instance_id: organization specific instance ID :param title: instance title :param text_msg: description of the instance :param assigned_to_str: names of the people who the instance is assigned to :param urgency: instance urgency :param service_name: name of the service :param snapshots: snapshots to be included with the details :param conference_bridge: (dict) -> {conference_url: .., conference_phone: ..} :return: (dict) of the adaptive card ''' facts = [ { "title": _lt.get_label(lnm.det_assigned_to, lang) + ":", "value": assigned_to_str }, { "title": _lt.get_label(lnm.det_service, lang) + ":", "value": service_name }, { "title": _lt.get_label(lnm.det_urgency, lang) + ":", "value": helpers.get_urgency_map(urgency, lang) } ] if conference_bridge is not None: if var_names.conference_url in conference_bridge and conference_bridge[var_names.conference_url] is not None: conf_url = conference_bridge[var_names.conference_url] facts.append({ "title": _lt.get_label(lnm.ttl_conference_url, lang) + ":", "value": "[" + conf_url + "](" + conf_url + ")" }) if var_names.conference_phone in conference_bridge\ and conference_bridge[var_names.conference_phone] is not None: conf_phone = conference_bridge[var_names.conference_phone] facts.append({ "title": _lt.get_label(lnm.ttl_dial_in_number, lang) + ":", "value": "[" + conf_phone + "](tel:" + conf_phone + ")" }) details_item_list = [ { "type": "FactSet", "facts": facts } ] if text_msg is not None and text_msg != '': details_item_list.append( { "type": "TextBlock", "text": _lt.get_label(lnm.ttl_details, lang), "weight": "Bolder", "spacing": "Medium", "wrap": True } ) details_item_list.append( { "type": "TextBlock", "text": text_msg, "wrap": True } ) if snapshots is not None and len(snapshots) > 0: for img_url in snapshots: details_item_list.append( { "type": "Image", "url": img_url } ) actions_list = [ { "type": "Action.Submit", "title": _lt.get_label(lnm.ttl_acknowledge, lang), "data": { var_names.event_type: constants.acknowledge_event, var_names.instance_id: instance_id } }, { "type": "Action.Submit", "title": _lt.get_label(lnm.ttl_resolve, lang), "data": { var_names.event_type: constants.resolve_event, var_names.instance_id: instance_id } }, { "type": "Action.ShowCard", "title": _lt.get_label(lnm.ttl_add_note, lang), "card": { "type": "AdaptiveCard", "version": "1.0", "body": [ { "type": "Input.Text", "id": var_names.notes, "isMultiline": True } ], "actions": [ { "type": "Action.Submit", "title": _lt.get_label(lnm.ins_submit, lang), "data": { var_names.event_type: constants.notate_event, var_names.instance_id: instance_id } } ] } } ] adaptive_card = { "$schema": "http://adaptivecards.io/schemas/adaptive-card.json", "type": "AdaptiveCard", "version": "1.0", "body": [ { "type": "Container", "items": [ { "type": "TextBlock", "text": _lt.get_label(lnm.ttl_incident, lang) + ' #' + str(org_instance_id), "weight": "bolder", "size": "small", "color": "attention" }, { "type": "TextBlock", "text": "[" + title + "](" + url_paths.web_incidents_details + '/' + str(org_instance_id) + ")", "weight": "bolder", "size": "large" } ] }, { "type": "Container", "items": details_item_list } ], "actions": actions_list } return adaptive_card def get_ms_teams_tenant_temp_token(tenant_id): ''' Get a temporary token for an MS Teams tenant. This token can be used to create online meetings. :param tenant_id: (str) MS Teams tenant ID :return: (str) temporary token ''' creds = s3.read_json(ms_teams_s3_bucket, ms_teams_s3_key) url = url_tenant_temp_token.format(tenant_id) headers = {'Content-Type': constants.content_type_url_encoded} body = {'client_id': creds[var_names.bot_id], 'client_secret': creds[var_names.bot_secret_value], 'scope': 'https://graph.microsoft.com/.default', 'grant_type': 'client_credentials'} response = requests.post(url, headers=headers, data=body) if response.status_code in (200, 201): return response.json()[var_names.access_token] else: logging.error('Failed to obtain temporary token for Microsoft Teams tenant. Error ' + str(response.status_code)) logging.error(str(response.json())) return None def create_meeting(timestamp, tenant_id, user_object_id, title, assignee_emails): ''' Create a Microsoft Teams online meeting. :param timestamp: timestamp when this request is being made :param tenant_id: tenant ID of the Microsoft Teams account where the online meeting will be created :param user_object_id: Microsoft AAD object ID of the user on behalf of whom the meeting will be created :param title: title to set for the meeting (should be the title of the incident) :param assignee_emails: (list) of email addresses of assignees ''' meeting_id, conf_url = None, None if len(assignee_emails) > 0: # We will not be using invitees in the actual request at the moment. Error: 'Request payload cannot be null' # invitees = [{'emailAddress': {'address': item}} for item in assignee_emails] temp_token = get_ms_teams_tenant_temp_token(tenant_id) if temp_token is not None: headers = {'Content-Type': constants.content_type_json, 'Authorization': 'Bearer ' + temp_token} url = url_online_meetings.format(user_object_id) start_time = timestamp + datetime.timedelta(minutes=1) end_time = start_time + datetime.timedelta(minutes=30) body = { 'subject': title, 'startDateTime': start_time.strftime(constants.json_timestamp_milli_format) + '+00:00', 'endDateTime': end_time.strftime(constants.json_timestamp_milli_format) + '+00:00' } resp = requests.post(url, headers=headers, data=json.dumps(body)) if resp.status_code in (200, 201): resp_body = resp.json() meeting_id = resp_body[var_id] conf_url = resp_body[var_join_url] else: logging.error('Failed to create MS Teams conference bridge. Status code - ' + str(resp.status_code)) logging.error(str(resp.json())) else: logging.error('Failed to create Microsoft Teams conference bridge because temporary token ' + 'to request with could not be obtained.') return meeting_id, conf_url class MsTeamsInstallationWelcomeHandler(Thread): ''' This class sends a welcome message to the TaskCall MS Teams bot when it is installed for the first time. The activity_id is the ID of the original message. ''' def __init__(self, conn, access_token, tenant_id, team_id, mst_service_url, bot_id, conversation_id, activity_id, channel_id, is_group=True): self.conn = conn self.access_token = access_token self.tenant_id = tenant_id self.team_id = team_id self.mst_service_url = mst_service_url self.bot_id = bot_id self.conversation_id = conversation_id self.activity_id = activity_id self.channel_id = channel_id self.is_group = is_group self.response_url = url_reply_message.format(mst_service_url, conversation_id, activity_id) self.current_time = times.get_current_timestamp() self.temporary_token = key_manager.generate_random_string_key(length=configuration.temporary_token_length) self.token_validity_end = self.current_time + datetime.timedelta(minutes=configuration.temporary_token_lifetime) self.token_details = {var_tenant_id: self.tenant_id, var_names.team_id: self.team_id, var_service_url: mst_service_url, var_names.bot_id: self.bot_id, var_names.channel_id: self.channel_id, var_names.is_group: self.is_group} Thread.__init__(self) def run(self): message = get_ms_teams_text_reply_body(self.bot_id, self.channel_id, self.activity_id, _lt.get_label(errors.err_system_error, constants.lang_en)) try: adaptive_card = { "$schema": "http://adaptivecards.io/schemas/adaptive-card.json", "type": "AdaptiveCard", "version": "1.0", "body": [ { "type": "Container", "items": [{ "type": "TextBlock", "text": _lt.get_label(info.msg_thank_you_for_installing_taskcall_app, constants.lang_en), "weight": "bolder", "size": "large" }] }, { "type": "Container", "items": [{ "type": "TextBlock", "text": _lt.get_label(info.msg_ms_teams_welcome_message, constants.lang_en), "wrap": True }] }, { "type": "Container", "items": [{ "type": "TextBlock", "text": _lt.get_label(lnm.det_for_more_information_refer_to, constants.lang_en) + ' ' + "[" + _lt.get_label(lnm.dsm_integration_guide, constants.lang_en) + "]" + "(" + url_tc_ms_teams_integration_guide + ").", "wrap": True }] } ], "actions": [ { "type": "Action.OpenUrl", "iconUrl": "https://taskcallapp.com/images/vendors/microsoft-teams/MicrosoftTeamsPopOut.png", "title": _lt.get_label(lnm.ins_authorize, constants.lang_en), "url": url_ms_teams_consent.format(self.tenant_id, self.bot_id, self.temporary_token, url_tc_ms_teams_authorization) } ] } message = get_ms_teams_adaptive_reply_body(self.bot_id, self.channel_id, self.activity_id, adaptive_card) except Exception as e: logging.exception(str(e)) finally: status, output = ms_teams_post_api_request(self.response_url, self.access_token, message) if status in (200, 201): db_integrations.store_temporary_verification_token( self.conn, self.current_time, self.token_validity_end, self.temporary_token, intt.microsoft_teams, self.token_details, expire_prior_tokens_by=[(var_names.channel_id, self.channel_id), (var_tenant_id, self.tenant_id)] ) else: logging.exception(str(output)) class MsTeamsUserVerifier(Thread): def __init__(self, conn, access_token, tenant_id, mst_service_url, bot_id, recipient_id, recipient_object_id, conversation_id, activity_id, channel_id, taskcall_pref_name): # check if the MS Teams tenant ID is associated to an organization or not # check if preferred username exists in the organization or not self.conn = conn self.access_token = access_token self.tenant_id = tenant_id self.mst_service_url = mst_service_url self.bot_id = bot_id self.recipient_id = recipient_id self.recipient_object_id = recipient_object_id self.conversation_id = conversation_id self.activity_id = activity_id self.channel_id = channel_id self.taskcall_pref_name = taskcall_pref_name self.response_url = url_reply_message.format(mst_service_url, conversation_id, activity_id) self.current_time = times.get_current_timestamp() Thread.__init__(self) def run(self): message = get_ms_teams_text_reply_body(self.bot_id, self.channel_id, self.activity_id, _lt.get_label(errors.err_verification_failed, constants.lang_en)) try: org_id, service_id, integ_id, external_info = db_microsoft_teams.get_microsoft_teams_integration_details( self.conn, self.current_time, self.tenant_id, self.channel_id) if service_id is not None: org_pref_maps = db_users.get_preferred_username_id_and_permissions(self.conn, self.current_time, org_id) if self.taskcall_pref_name in org_pref_maps: taskcall_user_id = org_pref_maps[self.taskcall_pref_name][0] if external_info is None: external_info = dict() if var_names.users not in external_info: external_info[var_names.users] = dict() if var_names.user_object_id not in external_info: external_info[var_names.user_object_id] = dict() external_info[var_names.users][self.recipient_id] = taskcall_user_id external_info[var_names.user_object_id][self.recipient_object_id] = taskcall_user_id db_services.update_organization_integration_type_details( self.conn, self.current_time, org_id, intt.microsoft_teams, external_info, external_id=self.tenant_id) message = get_ms_teams_text_reply_body( self.bot_id, self.channel_id, self.activity_id, _lt.get_label(info.msg_success, constants.lang_en) ) except Exception as e: logging.exception(str(e)) message = get_ms_teams_text_reply_body(self.bot_id, self.channel_id, self.activity_id, _lt.get_label(errors.err_system_error, constants.lang_en)) finally: status, output = ms_teams_post_api_request(self.response_url, self.access_token, message) if status not in (200, 201): logging.exception(str(output)) class MsTeamsUserRemover(Thread): ''' Removes an associated MS Teams user from the database. ''' def __init__(self, conn, access_token, tenant_id, mst_service_url, bot_id, recipient_id, conversation_id, activity_id, channel_id): self.conn = conn self.access_token = access_token self.tenant_id = tenant_id self.mst_service_url = mst_service_url self.bot_id = bot_id self.recipient_id = recipient_id self.conversation_id = conversation_id self.activity_id = activity_id self.channel_id = channel_id self.response_url = url_reply_message.format(mst_service_url, conversation_id, activity_id) self.current_time = times.get_current_timestamp() Thread.__init__(self) def run(self): message = get_ms_teams_text_reply_body(self.bot_id, self.channel_id, self.activity_id, _lt.get_label(errors.err_unknown_resource, constants.lang_en)) try: org_id, service_id, integ_id, external_info = db_microsoft_teams.get_microsoft_teams_integration_details( self.conn, self.current_time, self.tenant_id, self.channel_id) if service_id is not None and external_info is not None and var_names.users in external_info\ and self.recipient_id in external_info[var_names.users]: del external_info[var_names.users][self.recipient_id] db_services.update_organization_integration_type_details( self.conn, self.current_time, org_id, intt.microsoft_teams, external_info, external_id=self.tenant_id) message = get_ms_teams_text_reply_body( self.bot_id, self.channel_id, self.activity_id, _lt.get_label(info.msg_success, constants.lang_en) ) else: message = get_ms_teams_text_reply_body( self.bot_id, self.channel_id, self.activity_id, _lt.get_label(info.msg_ms_teams_user_unverified_or_integration_missing, constants.lang_en)) except Exception as e: logging.exception(str(e)) message = get_ms_teams_text_reply_body(self.bot_id, self.channel_id, self.activity_id, _lt.get_label(errors.err_system_error, constants.lang_en)) finally: status, output = ms_teams_post_api_request(self.response_url, self.access_token, message) if status not in (200, 201): logging.exception(str(output)) class MsTeamsUserInfo(Thread): ''' Gets the TaskCall username of an associated MS Teams account. ''' def __init__(self, conn, access_token, tenant_id, mst_service_url, bot_id, recipient_id, conversation_id, activity_id, channel_id): self.conn = conn self.access_token = access_token self.tenant_id = tenant_id self.mst_service_url = mst_service_url self.bot_id = bot_id self.recipient_id = recipient_id self.conversation_id = conversation_id self.activity_id = activity_id self.channel_id = channel_id self.response_url = url_reply_message.format(mst_service_url, conversation_id, activity_id) self.current_time = times.get_current_timestamp() Thread.__init__(self) def run(self): message = get_ms_teams_text_reply_body(self.bot_id, self.channel_id, self.activity_id, _lt.get_label(errors.err_unknown_resource, constants.lang_en)) try: org_id, service_id, integ_id, external_info = db_microsoft_teams.get_microsoft_teams_integration_details( self.conn, self.current_time, self.tenant_id, self.channel_id) if service_id is not None and external_info is not None and var_names.users in external_info\ and self.recipient_id in external_info[var_names.users]: req_user_id = external_info[var_names.users][self.recipient_id] user_details = db_users.get_user_details(self.conn, self.current_time, org_id, user_id=req_user_id) if user_details is not None: message = get_ms_teams_text_reply_body( self.bot_id, self.channel_id, self.activity_id, user_details[var_names.preferred_username] ) else: message = get_ms_teams_text_reply_body( self.bot_id, self.channel_id, self.activity_id, _lt.get_label(info.msg_ms_teams_user_unverified_or_integration_missing, constants.lang_en)) except Exception as e: logging.exception(str(e)) message = get_ms_teams_text_reply_body(self.bot_id, self.channel_id, self.activity_id, _lt.get_label(errors.err_system_error, constants.lang_en)) finally: status, output = ms_teams_post_api_request(self.response_url, self.access_token, message) if status not in (200, 201): logging.exception(str(output)) class MsTeamsTextReplier(Thread): ''' Sends a simple text reply to a MS Teams message. ''' def __init__(self, access_token, mst_service_url, bot_id, conversation_id, activity_id, channel_id, text_label): self.access_token = access_token self.mst_service_url = mst_service_url self.bot_id = bot_id self.conversation_id = conversation_id self.activity_id = activity_id self.channel_id = channel_id self.text_label = text_label self.response_url = url_reply_message.format(mst_service_url, conversation_id, activity_id) Thread.__init__(self) def run(self): message = get_ms_teams_text_reply_body(self.bot_id, self.channel_id, self.activity_id, _lt.get_label(self.text_label, constants.lang_en)) status, output = ms_teams_post_api_request(self.response_url, self.access_token, message) if status not in (200, 201): logging.exception(str(output)) class MsTeamsIncidentActionProcessor(Thread): ''' Handles actions that are received from MS Teams interactive messages. ''' def __init__(self, conn, access_token, tenant_id, mst_service_url, bot_id, conversation_id, activity_id, channel_id, recipient_id, recipient_name, action_value): self.conn = conn self.access_token = access_token self.tenant_id = tenant_id self.mst_service_url = mst_service_url self.bot_id = bot_id self.conversation_id = conversation_id self.activity_id = activity_id self.channel_id = channel_id self.recipient_id = recipient_id self.recipient_name = recipient_name self.action_value = action_value self.response_url = url_reply_message.format(mst_service_url, conversation_id, activity_id) self.current_time = times.get_current_timestamp() Thread.__init__(self) def run(self): to_respond = True message = get_ms_teams_text_reply_body(self.bot_id, self.channel_id, self.activity_id, _lt.get_label(errors.err_system_error, constants.lang_en)) try: org_id, service_id, integ_id, external_info = db_microsoft_teams.get_microsoft_teams_integration_details( self.conn, self.current_time, self.tenant_id, self.channel_id) if service_id is not None and external_info is not None and var_names.users in external_info\ and self.recipient_id in external_info[var_names.users]: taskcall_user_id = external_info[var_names.users][self.recipient_id] org_id, user_perm, org_perm = db_users.get_user_token_details( self.conn, self.current_time, taskcall_user_id)[:3] token_exp = times.get_current_timestamp() + datetime.timedelta(minutes=1) user_token = tokenizer.create_token(taskcall_user_id, org_id, user_perm, org_perm, token_exp) instance_id = self.action_value[var_names.instance_id] event_type = self.action_value[var_names.event_type] body = { var_names.instance_id: instance_id, var_names.access_method: constants.integrations_api } if event_type == constants.acknowledge_event: api_url = url_paths.incidents_acknowledge status, output = helpers.post_api_request(api_url, body, user_token) if status == 200: to_respond = False else: message = get_ms_teams_text_reply_body(self.bot_id, self.channel_id, self.activity_id, output) elif event_type == constants.resolve_event: api_url = url_paths.incidents_resolve status, output = helpers.post_api_request(api_url, body, user_token) if status == 200: to_respond = False else: message = get_ms_teams_text_reply_body(self.bot_id, self.channel_id, self.activity_id, output) elif event_type == constants.notate_event: note = self.action_value[var_names.notes] body[var_names.notes] = note api_url = url_paths.incidents_notate status, output = helpers.post_api_request(api_url, body, user_token) if status == 200: to_respond = False else: message = get_ms_teams_text_reply_body(self.bot_id, self.channel_id, self.activity_id, output) else: message = get_ms_teams_text_reply_body(self.bot_id, self.channel_id, self.activity_id, unknown_action_error) else: message = get_ms_teams_text_reply_body(self.bot_id, self.channel_id, self.activity_id, _lt.get_label(errors.err_verification_failed, constants.lang_en)) except Exception as e: logging.exception(str(e)) finally: if to_respond: status, output = ms_teams_post_api_request(self.response_url, self.access_token, message) if status not in (200, 201): logging.exception(str(output)) class MsTeamsIncidentDispatcher(object): ''' This is a MS Teams incident details dispatcher class. This should be used to send incident details to a specific MS Teams channel. ''' def __init__(self, access_token, mst_service_url, bot_id, channel_id, instance_id, org_instance_id, instance_title, text_msg, service_name, assigned_to_str, urgency, snapshots, lang=constants.lang_en, cache=None, conference_bridge=None): self.access_token = access_token self.bot_id = bot_id self.channel_id = channel_id self.instance_id = instance_id self.org_instance_id = org_instance_id self.instance_title = instance_title self.text_msg = text_msg self.service_name = service_name self.assigned_to_str = assigned_to_str self.urgency = urgency self.snapshots = snapshots self.conference_bridge = conference_bridge self.lang = lang self.cache = cache self.ms_teams_api_url = url_proactive_message.format(mst_service_url) def send_message(self): adaptive_card = get_incident_details_card( self.lang, self.instance_id, self.org_instance_id, self.instance_title, self.text_msg, self.assigned_to_str, self.urgency, self.service_name, self.snapshots, self.conference_bridge) message = get_ms_teams_proactive_adaptive_body(self.bot_id, self.channel_id, adaptive_card) try: status, output = ms_teams_post_api_request(self.ms_teams_api_url, self.access_token, message) if status in (200, 201): activity_id = output['activityId'] logging.info('Microsoft Teams notification activity ID: ' + activity_id) # Only store the activity ID in cache. No need to store it in the database. if self.cache is not None and settings.CACHE_ON: inst = cache_task_instances.get_single_instance(self.cache, self.instance_id) if inst is not None: inst.task.details[var_ms_teams_message_id] = activity_id cache_task_instances.store_single_instance(self.cache, inst) else: logging.exception(str(output)) except Exception as e: logging.exception(str(e)) raise def send_event_update(access_token, mst_service_url, bot_id, channel_id, evn_type, org_inst_id, inst_title, lang=constants.lang_en, user_display_name=None, extra_info=None): ''' Notify a MS Teams channel about an action that has been performed on an incident. :param access_token: access token :param mst_service_url: URL to send the message to :param bot_id: TaskCall bot ID :param channel_id: ID of the channel to sent to :param evn_type: type of event (ACKNOWLEDGE, RESOLVE, etc) :param org_inst_id: organization instance ID :param inst_title: title of the instance :param lang: language the message should be sent in :param user_display_name: name of the user who is taking the action :param extra_info: extra details to add in the chat message :return: status of the HTTP request ''' if evn_type == constants.acknowledge_event: evn_lbl = lnm.evn_acknowledge elif evn_type == constants.notate_event: evn_lbl = lnm.evn_notate elif evn_type == constants.resolve_event: evn_lbl = lnm.evn_resolve elif evn_type == constants.status_update_event: evn_lbl = lnm.evn_status_update elif evn_type == constants.un_acknowledge_event: evn_lbl = lnm.evn_un_acknowledge elif evn_type == constants.urgency_amendment_event: evn_lbl = lnm.evn_urgency_amendment else: return trimmed_title = '' if inst_title is not None: trimmed_title = inst_title[:40] + '...' if len(inst_title) > 40 else inst_title event_line = "[" + _lt.get_label(lnm.ttl_incident, lang) + " #" + str(org_inst_id) + "]" + \ "(" + url_paths.web_incidents_details + "/" + str(org_inst_id) + ") " + trimmed_title + ": "\ "**" + _lt.get_label(evn_lbl, constants.lang_en) + "**" if user_display_name is not None: event_line += " " + _lt.get_label(lnm.dsm_by, lang) + ' ' + user_display_name if extra_info is not None: event_line += ' - ' + extra_info message = get_ms_teams_proactive_text(bot_id, channel_id, event_line) ms_teams_api_url = url_proactive_message.format(mst_service_url) status, output = ms_teams_post_api_request(ms_teams_api_url, access_token, message) if status not in (200, 201): logging.exception(str(output))