# By: Riasat Ullah
# This file contains Google Chat 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_google_chat
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from httplib2 import Http
from oauth2client.service_account import ServiceAccountCredentials
from taskcallrest import settings
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
# Google Chat S3 credential file variables
gc_s3_bucket = 'taskcall-prod-data'
gc_s3_key = 'credentials/google_chat_credentials.json'
# Google Chat API url formats
url_tc_gc_authorization = settings.WEB_APP_BASE_URL +\
'/configurations/services/integrations/google-chat/authorize?token={0}'
url_tc_gc_integration_guide = 'https://docs.taskcallapp.com/integrations/v1/google-chat-integration-guide'
# Google Chat payload types
payload_type_unspecified = 'UNSPECIFIED'
payload_type_message = 'MESSAGE'
payload_type_added_to_space = 'ADDED_TO_SPACE'
payload_type_remove_from_space = 'REMOVED_FROM_SPACE'
payload_type_card_clicked = 'CARD_CLICKED'
# Google Chat commands
gc_create_incident_cmd = '/create-incident'
gc_get_user_cmd = '/get-user'
gc_help_cmd = '/help'
gc_remove_user_cmd = '/remove-user'
gc_request_authorization = '/request-authorization'
gc_verify_user_cmd = '/verify-user'
gc_valid_commands = [gc_create_incident_cmd, gc_get_user_cmd, gc_help_cmd,
gc_remove_user_cmd, gc_request_authorization, gc_verify_user_cmd]
# Google Chat string variables
var_domain_id = 'domainId'
var_google_chat_message_id = 'google_chat_message_id'
var_stringInputs = 'stringInputs'
# Google Chat custom event type
custom_event_create_space = 'create_space'
custom_event_notes_dialog = 'notes_dialog'
# names of functions that return or process Dialogs
func_create_incident = 'create_incident'
func_get_new_incident_dialog = 'new_incident_dialog'
func_get_note_dialog = 'get_note_dialog'
def get_google_chat_credentials():
'''
Get the credentials needed for handling and making API calls for MS Teams.
:return: (dict) of credentials
'''
creds = s3.read_json(gc_s3_bucket, gc_s3_key)
return creds
def create_google_chat_message(space_name, message):
'''
Make a post request to a MS Teams API endpoint.
:param space_name: name of the space to create the message in
:param message: message to send
:return: (str) ID of the message created
:error: Google Http error
'''
scopes = ['https://www.googleapis.com/auth/chat.bot']
chat_credentials = ServiceAccountCredentials.from_json_keyfile_dict(get_google_chat_credentials(), scopes)
chat = build('chat', 'v1', http=chat_credentials.authorize(Http()))
try:
response = chat.spaces().messages().create(parent=space_name, body=message).execute()
message_id = response['name']
logging.info('Google Chat notification message ID: ' + message_id)
return message_id
except HttpError as e:
logging.exception('Failed to create Google Chat message')
logging.exception(str(e))
return None
def update_google_chat_message(message_id, message):
'''
Updates an existing Google Chat card sent from TaskCall.
:param message_id: ID of the message to update
:param message: Card object
:error: Google Http error
'''
scopes = ['https://www.googleapis.com/auth/chat.bot']
chat_credentials = ServiceAccountCredentials.from_json_keyfile_dict(get_google_chat_credentials(), scopes)
chat = build('chat', 'v1', http=chat_credentials.authorize(Http()))
try:
chat.spaces().messages().update(name=message_id, body=message, updateMask="cardsV2").execute()
except HttpError as e:
logging.exception('Failed to create Google Chat message')
logging.exception(str(e))
return None
def get_simple_text_reply(text):
'''
Gets the simple text message reply to sent to Google Chat.
:param text: the text to send
:return: (dict) of final response
'''
body = {'text': text}
return body
def get_simple_dialog_reply(error_message=None):
'''
Get the simple reply to send after a dialog submission. If there is no error message, then OK response is sent.
:param error_message: (optional) the error message to send if there is one.
:return: (dict) -> Google Chat Dialog response
'''
if error_message is None:
action_status = "OK"
else:
action_status = error_message
dialog = {
"actionResponse": {
"type": "DIALOG",
"dialogAction": {
"actionStatus": action_status
}
}
}
return dialog
def is_request_dialog_event(request):
'''
Checks if a message payload is a REQUEST_DIALOG event or not
:param request: Http request
:return: (boolean) True or False
'''
if 'isDialogEvent' in request.data and request.data['isDialogEvent'] and\
'dialogEventType' in request.data and request.data['dialogEventType'] == 'REQUEST_DIALOG':
return True
return False
def is_submit_dialog_event(request):
'''
Checks if a message payload is a SUBMIT_DIALOG event or not
:param request: Http request
:return: (boolean) True or False
'''
if 'isDialogEvent' in request.data and request.data['isDialogEvent'] and\
'dialogEventType' in request.data and request.data['dialogEventType'] == 'SUBMIT_DIALOG':
return True
return False
def get_incident_details_card(lang, instance_id, org_instance_id, title, text_msg, assigned_to_str, status, 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 status: instance status
: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
'''
detail_widgets = [
{
"decoratedText": {
"topLabel": _lt.get_label(lnm.det_assigned_to, lang),
"text": assigned_to_str,
"wrapText": True
}
},
{
"decoratedText": {
"topLabel": _lt.get_label(lnm.det_service, lang),
"text": service_name,
"wrapText": True
}
},
{
"decoratedText": {
"topLabel": _lt.get_label(lnm.det_urgency, lang),
"text": helpers.get_urgency_map(urgency, lang),
"wrapText": True
}
}
]
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]
detail_widgets.append({
"topLabel": _lt.get_label(lnm.ttl_conference_url, lang),
"text": "" + conf_url + "",
"wrapText": True
})
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]
detail_widgets.append({
"topLabel": _lt.get_label(lnm.ttl_conference_url, lang),
"text": "" + conf_phone + "",
"wrapText": True
})
if text_msg is not None and text_msg != '':
detail_widgets.append(
{
"decoratedText": {
"topLabel": _lt.get_label(lnm.ttl_details, lang),
"text": text_msg,
"wrapText": True
}
}
)
if snapshots is not None and len(snapshots) > 0:
for img_url in snapshots:
detail_widgets.append(
{
"image": {
"imageUrl": img_url
}
}
)
sections = [{
"widgets": [{
"decoratedText": {
"topLabel": helpers.get_instance_status_map(status, lang),
"text": "#" + str(org_instance_id) + ': ' + title + "",
"wrapText": True
}
}]
}, {
"widgets": detail_widgets
}]
if status != constants.resolved_state:
sections.append({
"widgets": [{
"buttonList": {
"buttons": [{
"text": _lt.get_label(lnm.ttl_acknowledge, lang),
"onClick": {
"action": {
"function": "acknowledge",
"parameters": [{
"key": var_names.event_type,
"value": constants.acknowledge_event
}, {
"key": var_names.instance_id,
"value": str(instance_id)
}]
}
}
}, {
"text": _lt.get_label(lnm.ttl_resolve, lang),
"onClick": {
"action": {
"function": "resolve",
"parameters": [{
"key": var_names.event_type,
"value": constants.resolve_event
}, {
"key": var_names.instance_id,
"value": str(instance_id)
}]
}
}
}]
}
}]
})
card = {
"cards_v2": [{
"card": {
"sections": sections
}
}]
}
return card
def update_incident_details_card(lang, curr_card, updates, new_status=None):
'''
Update the details of a card and return it.
:param lang: language the card should be in
:param curr_card: current state of the card
:param updates: (list of html string) -> update statement containing tag where needed
:param new_status: (str) new status of the incident if there has been a status change
:return: (dict) -> card object
'''
# for item in updates:
# item_section = {
# "widgets": [{
# "textParagraph": {
# "text": item
# }
# }]
# }
# curr_card[0]["card"]["sections"].append(item_section)
if new_status is not None:
curr_card[0]["card"]["sections"][0]["widgets"][0]["decoratedText"]["topLabel"] =\
helpers.get_instance_status_map(new_status, lang)
new_card = {
'cards_v2': curr_card
}
return new_card
def get_new_incident_dialog(conn, org_id, lang=constants.lang_en):
'''
Get the dialog to be sent to Google Chat when the slash command /create-incident
is triggered to create a new incident.
:param conn: db connection
:param org_id: ID of the organization
:param lang: language to prepare the dialog in
:return: (dict) -> Google Chat Dialog
:errors: DatabaseError
'''
current_time = times.get_current_timestamp()
org_services = db_services.get_basic_services_list(conn, current_time, org_id)
service_items = [{"text": x[0], "value": x[1]} for x in org_services]
urgency_items = [{"text": helpers.get_urgency_map(urg, lang), "value": str(urg)}
for urg in configuration.allowed_urgency_levels]
dialog = {
"actionResponse": {
"type": "DIALOG",
"dialogAction": {
"dialog": {
"body": {
"sections": [{
"header": _lt.get_label(lnm.ttl_new_incident, lang),
"widgets": [{
"textInput": {
"label": _lt.get_label(lnm.ttl_incident_title, lang),
"type": "SINGLE_LINE",
"name": var_names.task_title
}
}, {
"selectionInput": {
"label": _lt.get_label(lnm.det_service, lang),
"type": "DROPDOWN",
"name": var_names.service_ref_id,
"items": service_items
}
}, {
"selectionInput": {
"label": _lt.get_label(lnm.det_urgency, lang),
"type": "DROPDOWN",
"name": var_names.urgency_level,
"items": urgency_items
}
}, {
"textInput": {
"label": _lt.get_label(lnm.det_description, lang),
"type": "MULTIPLE_LINE",
"name": var_names.text_msg
}
}, {
"buttonList": {
"buttons": [{
"text": _lt.get_label(lnm.ttl_create_incident, lang),
"onClick": {
"action": {
"function": "create_incident",
}
}
}]
}
}]
}]
}
}
}
}
}
return dialog
def get_text_input_dialog(instance_id, lang=constants.lang_en):
dialog = {
"actionResponse": {
"type": "DIALOG",
"dialogAction": {
"dialog": {
"body": {
"sections": [{
"header": _lt.get_label(lnm.ttl_add_note, lang),
"widgets": [{
"textInput": {
"label": _lt.get_label(lnm.ttl_notes, lang),
"type": "MULTIPLE_LINE",
"name": "notes"
}
}, {
"buttonList": {
"buttons": [{
"text": _lt.get_label(lnm.ttl_add_note, lang),
"onClick": {
"action": {
"function": "notate",
"parameters": [{
"key": var_names.event_type,
"value": constants.notate_event
}, {
"key": var_names.instance_id,
"value": str(instance_id)
}]
}
}
}]
}
}]
}]
}
}
}
}
}
return dialog
def get_google_chat_authorization_verification_token(conn, domain_id, space_name):
'''
Get and store a token to verify a specific Google Chat authorization process. When an authorization request is made,
a token is sent in the response with which the integration process can be verified. The issuance of the token
requires it to be stored in the database.
:param conn: db connection
:param domain_id: Google Workspace domain ID
:param space_name: name of the space to integrate with
:return: (str) token | if the token cannot be stored successfully then None is sent
:errors: DatabaseError
'''
current_time = times.get_current_timestamp()
temporary_token = key_manager.generate_random_string_key(length=configuration.temporary_token_length)
token_validity_end = current_time + datetime.timedelta(minutes=configuration.temporary_token_lifetime)
token_details = {var_domain_id: domain_id, var_names.channel_id: space_name}
try:
db_integrations.store_temporary_verification_token(
conn, current_time, token_validity_end, temporary_token, intt.google_chat, token_details,
expire_prior_tokens_by=[(var_domain_id, domain_id), (var_names.channel_id, space_name)]
)
return temporary_token
except Exception as e:
logging.exception(str(e))
return None
def get_google_chat_welcome_card(temp_token):
'''
Gets the welcome message that is sent to Google Chat after the installation of the TaskCall app in a space.
:param temp_token: temporary token that is used to verify the authorization request
:return: (dict) -> Google Chat card
'''
card = {
"cards_v2": [{
"card": {
"header": {
"title": _lt.get_label(info.msg_thank_you_for_installing_taskcall_app, constants.lang_en)
},
"sections": [{
"widgets": [{
"textParagraph": {
"text": _lt.get_label(info.msg_google_chat_welcome_message, constants.lang_en)
}
}, {
"decoratedText": {
"text": _lt.get_label(lnm.det_for_more_information_refer_to, constants.lang_en) +
" " +
_lt.get_label(lnm.dsm_integration_guide, constants.lang_en) + ''
}
}]
}, {
"widgets": [{
"buttonList": {
"buttons": [{
"text": _lt.get_label(lnm.ins_authorize, constants.lang_en),
"onClick": {
"openLink": {
"url": url_tc_gc_authorization.format(temp_token)
}
}
}]
}
}]
}]
}
}]
}
return card
def verify_google_chat_user(conn, domain_id, space_name, google_chat_username, taskcall_pref_name,
lang=constants.lang_en):
'''
Verify and map a Google Chat user with a TaskCall user.
:param conn: db connection
:param domain_id: Google Workspace domain ID
:param space_name: name of the Google Chat space the message was received from
:param google_chat_username: the username of the user in Google Chat
:param taskcall_pref_name: the preferred name of the user in TaskCall
:param lang: language to prepare the response message in
:return: (dict) -> Google Chat format simple text response
:errors: DatabaseError
'''
message = get_simple_text_reply(_lt.get_label(errors.err_verification_failed, lang))
try:
current_time = times.get_current_timestamp()
org_id, service_id, integ_id, external_info = db_google_chat.get_google_chat_integration_details(
conn, current_time, domain_id, space_name)
if service_id is not None:
org_pref_maps = db_users.get_preferred_username_id_and_permissions(conn, current_time, org_id)
if taskcall_pref_name in org_pref_maps:
taskcall_user_id = org_pref_maps[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()
external_info[var_names.users][google_chat_username] = taskcall_user_id
db_services.update_organization_integration_type_details(
conn, current_time, org_id, intt.google_chat, external_info, external_id=domain_id)
message = get_simple_text_reply(_lt.get_label(info.msg_success, lang))
except Exception as e:
logging.exception(str(e))
message = get_simple_text_reply(_lt.get_label(errors.err_system_error, lang))
finally:
return message
def remove_google_chat_user(conn, domain_id, space_name, google_chat_username, lang=constants.lang_en):
'''
Removes a verified Google Chat user from the mapping between Google Chat users and TaskCall users.
:param conn: db connection
:param domain_id: Google Workspace domain ID
:param space_name: name of the Google Chat space the message was received from
:param google_chat_username: the username of the user in Google Chat
:param lang: language to prepare the response message in
:return: (dict) -> Google Chat format simple text response
:errors: DatabaseError
'''
message = get_simple_text_reply(_lt.get_label(errors.err_unknown_resource, lang))
try:
current_time = times.get_current_timestamp()
org_id, service_id, integ_id, external_info = db_google_chat.get_google_chat_integration_details(
conn, current_time, domain_id, space_name)
if service_id is not None and external_info is not None and var_names.users in external_info\
and google_chat_username in external_info[var_names.users]:
del external_info[var_names.users][google_chat_username]
db_services.update_organization_integration_type_details(
conn, current_time, org_id, intt.google_chat, external_info, external_id=domain_id)
message = get_simple_text_reply(_lt.get_label(info.msg_success, lang))
else:
message = get_simple_text_reply(
_lt.get_label(info.msg_google_chat_user_unverified_or_integration_missing, lang))
except Exception as e:
logging.exception(str(e))
message = get_simple_text_reply(_lt.get_label(errors.err_system_error, lang))
finally:
return message
def get_google_chat_user_info(conn, domain_id, space_name, google_chat_username, lang=constants.lang_en):
'''
Gets the TaskCall preferred username of a verified (mapped) Google Chat user.
:param conn: db connection
:param domain_id: Google Workspace domain ID
:param space_name: name of the Google Chat space the message was received from
:param google_chat_username: the username of the user in Google Chat
:param lang: language to prepare the response message in
:return: (dict) -> Google Chat format simple text response
:errors: DatabaseError
'''
message = get_simple_text_reply(_lt.get_label(errors.err_unknown_resource, lang))
try:
current_time = times.get_current_timestamp()
org_id, service_id, integ_id, external_info = db_google_chat.get_google_chat_integration_details(
conn, current_time, domain_id, space_name)
if service_id is not None and external_info is not None and var_names.users in external_info\
and google_chat_username in external_info[var_names.users]:
req_user_id = external_info[var_names.users][google_chat_username]
user_details = db_users.get_user_details(conn, current_time, org_id, user_id=req_user_id)
if user_details is not None:
message = get_simple_text_reply(user_details[var_names.preferred_username])
else:
message = get_simple_text_reply(
_lt.get_label(info.msg_google_chat_user_unverified_or_integration_missing, lang))
except Exception as e:
logging.exception(str(e))
message = get_simple_text_reply(_lt.get_label(errors.err_system_error, lang))
finally:
return message
def process_google_chat_incident_actions(conn, domain_id, space_name, google_chat_username, google_chat_display_name,
action_value, message_id, current_card, lang=constants.lang_en):
'''
Handle actions that are taken from the card sent to Google Chat.
:param conn: db connection
:param domain_id: Google Workspace domain ID
:param space_name: name of the Google Chat space
:param google_chat_username: username of the user in Google Chat
:param google_chat_display_name: display name of the user in Google Chat
:param action_value: action parameters sent with the event from Google Chat {event_type: , instance_id: }
:param message_id: ID of the message (this will be used to update the original card sent)
:param current_card: (dict) state of the current card
:param lang: language to process the event in
:return: (dict) -> reply message (card or text)
:errors: DatabaseError
'''
message = get_simple_text_reply(_lt.get_label(errors.err_system_error, lang))
try:
current_time = times.get_current_timestamp()
org_id, service_id, integ_id, external_info = db_google_chat.get_google_chat_integration_details(
conn, current_time, domain_id, space_name)
if service_id is not None and external_info is not None and var_names.users in external_info\
and google_chat_username in external_info[var_names.users]:
taskcall_user_id = external_info[var_names.users][google_chat_username]
org_id, user_perm, org_perm = db_users.get_user_token_details(
conn, 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 = int(action_value[var_names.instance_id])
event_type = 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:
event_line = ' '.join([
"" + _lt.get_label(lnm.dsm_acknowledged, constants.lang_en).capitalize() + "",
_lt.get_label(lnm.dsm_by, constants.lang_en), google_chat_display_name
])
new_card = update_incident_details_card(lang, current_card, [event_line],
constants.acknowledged_state)
update_google_chat_message(message_id, new_card)
message = info.msg_success
else:
message = get_simple_text_reply(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:
event_line = ' '.join([
"" + _lt.get_label(lnm.dsm_resolved, constants.lang_en).capitalize() + "",
_lt.get_label(lnm.dsm_by, constants.lang_en), google_chat_display_name
])
new_card = update_incident_details_card(lang, current_card, [event_line], constants.resolved_state)
update_google_chat_message(message_id, new_card)
message = info.msg_success
else:
message = get_simple_text_reply(output)
else:
message = get_simple_text_reply(_lt.get_label(errors.err_unknown_action, lang))
else:
message = get_simple_text_reply(_lt.get_label(errors.err_verification_failed, constants.lang_en))
except Exception as e:
logging.exception(str(e))
finally:
return message
class GoogleChatIncidentDispatcher(object):
'''
This is a Google Chat incident detail dispatcher class. This should be used to send
incident details to a specific Google Chat space.
'''
def __init__(self, space_name, instance_id, org_instance_id, instance_title, text_msg, service_name,
assigned_to_str, status, urgency, snapshots, lang=constants.lang_en, cache=None,
conference_bridge=None):
self.space_name = space_name
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.status = status
self.urgency = urgency
self.snapshots = snapshots
self.conference_bridge = conference_bridge
self.lang = lang
self.cache = cache
def send_message(self):
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.status, self.urgency, self.service_name, self.snapshots, self.conference_bridge
)
try:
message_id = create_google_chat_message(self.space_name, card)
# Only store the message 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_google_chat_message_id] = message_id
cache_task_instances.store_single_instance(self.cache, inst)
except Exception as e:
logging.exception(str(e))
raise
def send_event_update(space_name, evn_type, org_inst_id, inst_title, lang=constants.lang_en, user_display_name=None,
extra_info=None):
'''
Notify a Google Chat channel about an action that has been performed on an incident.
:param space_name: name of the space to create the message in
: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 = "<" + url_paths.web_incidents_details + "/" + str(org_inst_id) + "|" + \
_lt.get_label(lnm.ttl_incident, lang) + " #" + 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
body = {'text': event_line}
create_google_chat_message(space_name, body)