import base64
from datetime import datetime, timedelta
from typing import List
from lxml.builder import E
from lxml.etree import tostring
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.tools.decorators import dangerous
from .db_object import DbObject
[docs]
def create_timestamps():
now = datetime.now()
broadcast_expiration_time = now + timedelta(days=365)
datetime_format = "%Y-%m-%dT%H:%M:%S"
creation_string = now.strftime(datetime_format)
expiration_string = broadcast_expiration_time.strftime(datetime_format)
return creation_string, expiration_string
[docs]
def create_stored_proc_parameters(
group_id, subject, body, expire, creation, attachment_name, attachment_data
):
parameters = [
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "FromGroupId", group_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar, "Subject", subject
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Text, "Body", body
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.DateTime, "ExpirationDate", expire
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.DateTime, "CreateDate", creation
),
]
if attachment_name:
parameters.extend([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"AttachmentName",
attachment_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Image,
"AttachmentData",
str(base64.b64encode(base64.b64encode(attachment_data)), "utf-8"),
),
])
return parameters
[docs]
class SecureMessage(DbObject):
[docs]
def get_user_list_param(self, list_of_users):
xml_list = []
for user_id in list_of_users:
xml_list.append(E.user(id=str(user_id)))
xml_string = tostring(E.userList(*xml_list), encoding="unicode")
return ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Xml, "UserList", xml_string
)
[docs]
def get_group_id_param(self, group_id):
return ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "ToGroupId", group_id
)
[docs]
def get_customer_id_param(self, customer_id):
return ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "ToCustomerId", customer_id
)
@dangerous("This has the potential to target hundreds or thousands of users")
async def broadcast_with_attachment(
self,
from_group_id: int,
subject: str,
body: str,
attachment_name: str = None,
attachment_data: bytes = None,
group_id: int = None,
customer_id: int = None,
user_list: List[int] = None,
):
"""
This sends a secure message to an entire group of End Users.
NOTE: This is NOT a secure message viewable in Central by CSRs,
but rather one visible by end users in the Messages menu
in OnlineBanking
"""
assert any([
group_id,
customer_id,
user_list,
]), "Error: no group, customer, or user sent in function call"
broadcast_creation_string, broadcast_expiration_string = create_timestamps()
parameters = create_stored_proc_parameters(
from_group_id,
subject,
body,
broadcast_expiration_string,
broadcast_creation_string,
attachment_name,
attachment_data,
)
if user_list:
parameters.append(self.get_user_list_param(user_list))
elif group_id:
parameters.append(self.get_group_id_param(group_id))
elif customer_id:
parameters.append(self.get_customer_id_param(customer_id))
await self.call_hq(
"sdk_CreateBroadcastMessageWithAttachment",
ExecuteStoredProcedure.SqlParameters(parameters),
)