from django.template.loader import render_to_string from django.utils.translation import gettext_lazy as _ from accounts.models import ChangeSecretRecord from common.tasks import send_mail_attachment_async, upload_backup_to_obj_storage from notifications.notifications import UserMessage from terminal.models.component.storage import ReplayStorage from users.models import User class AccountBackupExecutionTaskMsg(object): subject = _('Notification of account backup route task results') def __init__(self, name: str, user: User): self.name = name self.user = user @property def message(self): name = self.name if self.user.secret_key: return _('{} - The account backup passage task has been completed.' ' See the attachment for details').format(name) else: return _("{} - The account backup passage task has been completed: " "the encryption password has not been set - " "please go to personal information -> Basic file encryption password for preference settings" ).format(name) def publish(self, attachment_list=None): send_mail_attachment_async( self.subject, self.message, [self.user.email], attachment_list ) class AccountBackupByObjStorageExecutionTaskMsg(object): subject = _('Notification of account backup route task results') def __init__(self, name: str, obj_storage: ReplayStorage): self.name = name self.obj_storage = obj_storage @property def message(self): name = self.name return _('{} - The account backup passage task has been completed.' ' See the attachment for details').format(name) def publish(self, attachment_list=None): upload_backup_to_obj_storage( self.obj_storage, attachment_list ) class ChangeSecretExecutionTaskMsg(object): subject = _('Notification of implementation result of encryption change plan') def __init__(self, name: str, user: User, summary): self.name = name self.user = user self.summary = summary @property def message(self): name = self.name if self.user.secret_key: default_message = _('{} - The encryption change task has been completed. ' 'See the attachment for details').format(name) else: default_message = _("{} - The encryption change task has been completed: the encryption " "password has not been set - please go to personal information -> " "set encryption password in preferences").format(name) return self.summary + '\n' + default_message def publish(self, attachments=None): send_mail_attachment_async( self.subject, self.message, [self.user.email], attachments ) class GatherAccountChangeMsg(UserMessage): subject = _('Gather account change information') def __init__(self, user, change_info: dict): self.change_info = change_info super().__init__(user) def get_html_msg(self) -> dict: context = {'change_info': self.change_info} message = render_to_string('accounts/asset_account_change_info.html', context) return { 'subject': str(self.subject), 'message': message } @classmethod def gen_test_msg(cls): user = User.objects.first() return cls(user, {}) class ChangeSecretFailedMsg(UserMessage): subject = _('Change secret or push account failed information') def __init__(self, name, execution_id, user, asset_account_errors: list): self.name = name self.execution_id = execution_id self.asset_account_errors = asset_account_errors super().__init__(user) def get_html_msg(self) -> dict: context = { 'name': self.name, 'recipient': self.user, 'execution_id': self.execution_id, 'asset_account_errors': self.asset_account_errors } message = render_to_string('accounts/change_secret_failed_info.html', context) return { 'subject': str(self.subject), 'message': message } @classmethod def gen_test_msg(cls): name = 'test' user = User.objects.first() record = ChangeSecretRecord.objects.first() execution_id = str(record.execution_id) return cls(name, execution_id, user, [])