File: //proc/1233/root/home/arjun/projects/buyercall/buyercall/blueprints/notification/utilities.py
from buyercall.blueprints.user.models import User
from buyercall.blueprints.partnership.models import Partnership, PartnershipAccount
from buyercall.blueprints.notification.models import NotificationSettings, ActivitySettings
from buyercall.integrations.elasticsearch.utilities import create_notification_data
from buyercall.extensions import es_client, push_engage
from flask import current_app, request, render_template
from buyercall.blueprints import flask_environment
in_app_notify_types = {
'IN_APP_LEAD': [],
'IN_APP_TASK_ASSIGN': ['NEW_TASK_ASSIGNED'],
'IN_APP_ALL_TASKS': ['NEW_TASK_ASSIGNED', 'TASK_PICKED'],
'IN_APP_ACCOUNT_ACTIVITY': ['USER_CREATED', 'ACCOUNT_CREATED', 'USER_DEACTIVATED', 'SOURCE_CREATED'],
'IN_APP_ACCOUNT_SECURITY': ['LOGGED_IN', 'PASSWORD_RESET', 'TWO_FACTOR_AUTH_REQUEST', 'PARTNERSHIP_SETTINGS_EDITED',
'PROFILE_SETTINGS_EDITED', 'EMAIL_NOTIFY_SETTINGS_EDITED', 'SCHEDULE_SETTINGS_EDITED',
'THEME_SETTINGS_EDITED', '2FA_SETTINGS_EDITED', 'SECURITY_INFO_EDITED',
'PERSONAL_SETTINGS_EDITED', 'INAPP_NOTIFY_SETTINGS_EDITED'],
'IN_APP_PROMOTION': [],
}
email_notify_types = {
'EMAIL_LEAD': [],
'EMAIL_TASK_ASSIGN': ['NEW_TASK_ASSIGNED'],
'EMAIL_ALL_TASKS': ['NEW_TASK_ASSIGNED', 'TASK_PICKED'],
'EMAIL_ACCOUNT_ACTIVITY': ['USER_CREATED', 'ACCOUNT_CREATED', 'USER_DEACTIVATED', 'SOURCE_CREATED'],
'EMAIL_ACCOUNT_SECURITY': ['LOGGED_IN', 'PASSWORD_RESET', 'TWO_FACTOR_AUTH_REQUEST', 'PARTNERSHIP_SETTINGS_EDITED',
'PROFILE_SETTINGS_EDITED', 'EMAIL_NOTIFY_SETTINGS_EDITED', 'SCHEDULE_SETTINGS_EDITED',
'THEME_SETTINGS_EDITED', '2FA_SETTINGS_EDITED', 'SECURITY_INFO_EDITED',
'PERSONAL_SETTINGS_EDITED', 'INAPP_NOTIFY_SETTINGS_EDITED'],
'EMAIL_PROMOTION': [],
}
def create_notification_message(message_description, message_variables):
"""
This method contructs the notification message that shows on the notification modal.
:param related_entities: The user or the actor who in involved in the activity. Example: John Doe, +71545856241
:return dict: {user_message: str, other_user_message: str, status: bool}
"""
response = {'status': True}
if not isinstance(message_variables, list):
message_variables = [message_variables]
if message_description and message_variables:
for u_index, u_entity in enumerate(message_variables):
message_description = message_description.replace(f"[{u_index}]", u_entity)
response['message'] = message_description
else:
response['message'] = message_description
return response
def get_users_by_notify_settings(users, notify_type):
if not notify_type:
return users
notify_enabled_users = []
for user in users:
notification_settings = NotificationSettings.get_or_create(user_id=user.id)
if notify_type == 'IN_APP_LEAD' and notification_settings.ian_on_new_lead:
notify_enabled_users.append(user)
elif notify_type == 'IN_APP_TASK_ASSIGN' and notification_settings.ian_on_task_assign:
notify_enabled_users.append(user)
elif notify_type == 'IN_APP_ALL_TASKS' and notification_settings.ian_all_tasks:
notify_enabled_users.append(user)
elif notify_type == 'IN_APP_ACCOUNT_ACTIVITY' and notification_settings.ian_on_account_activity:
notify_enabled_users.append(user)
elif notify_type == 'IN_APP_ACCOUNT_SECURITY' and notification_settings.ian_on_account_security:
notify_enabled_users.append(user)
elif notify_type == 'IN_APP_PROMOTION' and notification_settings.ian_on_promo:
notify_enabled_users.append(user)
if notify_type == 'EMAIL_LEAD' and notification_settings.ian_on_new_lead:
notify_enabled_users.append(user)
elif notify_type == 'EMAIL_TASK_ASSIGN' and notification_settings.ian_on_task_assign:
notify_enabled_users.append(user)
elif notify_type == 'EMAIL_ALL_TASKS' and notification_settings.ian_all_tasks:
notify_enabled_users.append(user)
elif notify_type == 'EMAIL_ACCOUNT_ACTIVITY' and notification_settings.ian_on_account_activity:
notify_enabled_users.append(user)
elif notify_type == 'EMAIL_ACCOUNT_SECURITY' and notification_settings.ian_on_account_security:
notify_enabled_users.append(user)
elif notify_type == 'EMAIL_PROMOTION' and notification_settings.ian_on_promo:
notify_enabled_users.append(user)
else:
pass
return notify_enabled_users
def get_user_sids_by_roles(user, valid_roles, in_app_notify_type=None, email_notify_type=None, *args, **kwargs):
user_ids = []
valid_roles = [] if not valid_roles else valid_roles
role_hierarchy = {
'sysadmin': 1,
'partner': 2,
'admin': 3,
'agent': 4,
'member': 5,
'guest': 6
}
# Get roles above current user
notify_roles = []
for role, position in role_hierarchy.items():
if position < role_hierarchy[user.role]:
notify_roles.append(role)
# Get common roles in valid roles and hierarchical roles
roles = list(set(notify_roles).intersection(valid_roles))
partnership = Partnership.query.filter(Partnership.id == user.partnership_id).first()
if not partnership:
partnership = Partnership.query.get(1)
additional_users = []
for u in kwargs.get('additional_notify_users', []):
if u != user.sid:
additional_users.append(User.query.filter(User.sid == u).first())
# Get users in the same partnership having selected roles
users = User.query.filter(User.partnership_id == partnership.id, User.role.in_(roles)).all()
users += User.query.filter(User.id == user.id).all()
# Add additional users
users += additional_users
inapp_notify_users = get_users_by_notify_settings(users, in_app_notify_type)
email_notify_users = get_users_by_notify_settings(users, email_notify_type)
inapp_notify_user_ids = []
email_notify_user_ids = []
# Get inapp notify user sid list
if inapp_notify_users:
inapp_notify_user_ids = [str(inapp_user.sid) for inapp_user in inapp_notify_users]
# Get email notify user sid list
if email_notify_users:
email_notify_user_ids = [str(email_user.sid) for email_user in email_notify_users]
user_ids = {'inapp_notify_user_ids': list(set(inapp_notify_user_ids)),
'email_notify_user_ids': list(set(email_notify_user_ids))}
return user_ids
def send_notifications(notify_message_type=None, user_related_entities=None, other_user_related_entities=None,
user_id=None, hyperlink=None, index_name='buyercall-notification-log', **kwargs):
try:
response = {'success': True, 'message': 'Successfully inserted the notification.'}
if notify_message_type and user_id:
user = User.query.filter(User.sid == user_id).first()
if user:
partnership = Partnership.query.filter(Partnership.id == user.partnership_id).first()
if not partnership:
partnership = Partnership.query.get(1)
partnership_account = PartnershipAccount.query.filter(
PartnershipAccount.id == user.partnership_account_id).first()
user_ip = user.current_sign_in_ip
user_device = request.headers.get('User-Agent', None)
triggered_user_id = user_id
user_message = ''
other_user_message = ''
activity_settings = ActivitySettings.query.filter(
ActivitySettings.activity_message_type == notify_message_type,
ActivitySettings.is_active == True).first()
if activity_settings:
is_es_inserted = False
valid_user_roles = activity_settings.valid_user_roles
notify_type = activity_settings.activity_type
in_app_notify_type = None
for act_type, msg_types in in_app_notify_types.items():
if notify_message_type in msg_types:
in_app_notify_type = act_type
email_notify_type = None
for act_type, msg_types in email_notify_types.items():
if notify_message_type in msg_types:
email_notify_type = act_type
# Add more users
additional_notify_users = kwargs.get('additional_notify_users', [])
notification_user_ids = get_user_sids_by_roles(
user, valid_user_roles, in_app_notify_type=in_app_notify_type,
email_notify_type=email_notify_type, additional_notify_users=additional_notify_users)
from buyercall.blueprints.user.tasks import put_to_elasticsearch
for sid in notification_user_ids['inapp_notify_user_ids']:
if str(triggered_user_id) == sid:
description = activity_settings.user_message_description
_new_desc_variables = user_related_entities
else:
description = activity_settings.other_user_description
_new_desc_variables = other_user_related_entities
payload = {
"triggered_user_id": triggered_user_id,
"details": _new_desc_variables
}
message = create_notification_message(description, _new_desc_variables)
data = create_notification_data(partnership, partnership_account, user_ip, user_device,
message['message'],
payload, notify_type, notify_message_type, sid, hyperlink)
log_state = flask_environment()
index_name = f'buyercall-notification-log-{log_state}'
es_response = put_to_elasticsearch(data, es_client, index_name)
is_es_inserted = True
if str(triggered_user_id) == sid:
user_message = message['message']
else:
other_user_message = message['message']
# Send push notify to triggered user
if activity_settings.is_web_push:
if str(triggered_user_id) == sid:
notify_settings = NotificationSettings.get_or_create(user.id)
if notify_settings.push_notification:
profile_id = [f'{log_state}-{str(user.sid)}']
response['message'] = push_engage.send_push_notification(
notification_url=hyperlink,
notification_title=activity_settings.message_literal,
notification_message=message['message'], profile_ids=profile_id, **kwargs
)
# Send push notify to other users
if activity_settings.is_web_push:
webpush_enabled_or_not = NotificationSettings.is_webpush_enabled_bulk(
notification_user_ids['inapp_notify_user_ids'])
profile_ids = []
for sid, is_push in webpush_enabled_or_not.items():
if is_push and sid != str(user_id):
profile_ids.append(f'{log_state}-{str(sid)}')
if profile_ids:
response['message'] = push_engage.send_push_notification(
notification_url=hyperlink, notification_title=activity_settings.message_literal,
notification_message=other_user_message, profile_ids=profile_ids, **kwargs
)
# Send email notification
for sid in notification_user_ids['email_notify_user_ids']:
if not is_es_inserted:
if str(triggered_user_id) == sid:
description = activity_settings.user_message_description
_new_desc_variables = user_related_entities
else:
description = activity_settings.other_user_description
_new_desc_variables = other_user_related_entities
payload = {
"triggered_user_id": triggered_user_id,
"details": _new_desc_variables
}
message = create_notification_message(description, _new_desc_variables)
data = create_notification_data(partnership, partnership_account, user_ip, user_device,
message['message'],
payload, notify_type, notify_message_type, sid, hyperlink)
log_state = flask_environment()
index_name = f'buyercall-notification-log-{log_state}'
es_response = put_to_elasticsearch(data, es_client, index_name)
is_es_inserted = True
if str(triggered_user_id) == sid:
user_message = message['message']
else:
other_user_message = message['message']
_user = User.query.filter(User.sid == sid).first()
recipients = [_user.email]
if sid == str(user_id):
mail_message = user_message
else:
mail_message = other_user_message
mail_subject = "You've a new notification"
mail_data = render_template('user/mail/notification.jinja2', message=mail_message,
hyperlink=hyperlink, user=_user, config=current_app.config,
partnership=partnership)
# celery
from buyercall.blueprints.user.tasks import send_notification_mail
mail_status = send_notification_mail.apply_async(
args=(mail_subject, mail_data, recipients))
else:
response = {'success': False, 'message': f'Activity of type {notify_message_type} not found.'}
else:
response = {'success': False, 'message': f'User not found with id {str(user_id)}'}
else:
response = {'success': False, 'message': f'Missing parameters.'}
except Exception as e:
print(e)
response = {'success': False, 'message': 'Error'}
return response