@@ -3,6 +3,7 @@ from authentication.exceptions.business_host_exceptions.already_account import (
)
from authentication.exceptions.business_host_exceptions.already_has_plan import (
AlreadyHasPlan,
+ InviteeHasPlan
)
from authentication.exceptions.business_host_exceptions.already_host import (
AlreadyHost,
@@ -10,6 +11,7 @@ from authentication.exceptions.business_host_exceptions.already_host import (
__all__ = (
'AlreadyHasPlan',
+ 'InviteeHasPlan',
'AlreadyHost',
'AlreadyAccount',
)
@@ -1,12 +1,6 @@
-from authentication.exceptions.business_host_exceptions.base_already import (
- BaseAlready,
-)
+from django.utils.translation import gettext as _
-class AlreadyAccount(BaseAlready):
- def msg(self) -> dict:
- return dict(
- message='user is already a business account for other company',
- user_id=self.user.uid,
- parent_company=self.user.business_account.parent_company.uid,
- )
+class AlreadyAccount(Exception):
+ def __str__(self):
+ return _('User is already a business account for another company')
@@ -1,17 +1,19 @@
-from authentication.exceptions.business_host_exceptions.base_already import (
- BaseAlready,
-)
-from django.utils.translation import gettext_lazy as _
+from django.utils.translation import gettext as _
-class AlreadyHasPlan(BaseAlready):
- def msg(self):
- return dict(
- message=_(
- 'You already have an active tariff plan. You must request a '
- 'cancellation of your current tariff plan, after which '
- 'you will be able to create a Corporate Account.'
- ),
- user_id=self.user.uid,
- plan_id=self.user.payment_plan.plan.uid,
+class AlreadyHasPlan(Exception):
+ def __str__(self) -> str:
+ return _(
+ 'You already have an active tariff plan. You must request a '
+ 'cancellation of your current tariff plan (via the \"Report an error\" button), after which '
+ 'you will be able to create a Corporate Account'
+ )
+
+
+class InviteeHasPlan(Exception):
+ def __str__(self) -> str:
+ return _(
+ 'Invitee already has an active tariff plan. Invitee must request a '
+ 'cancellation of his current tariff plan (via the \"Report an error\" button), after which '
+ 'you will be able to invite him'
)
@@ -1,12 +1,6 @@
-from authentication.exceptions.business_host_exceptions.base_already import (
- BaseAlready,
-)
+from django.utils.translation import gettext as _
-class AlreadyHost(BaseAlready):
- def msg(self) -> dict:
- return dict(
- message='user already has a host account',
- user_id=self.user.uid,
- company_id=self.user.host_account.uid,
- )
+class AlreadyHost(Exception):
+ def __str__(self):
+ return _('User already has a host account')
@@ -1,12 +0,0 @@
-from typing import TYPE_CHECKING
-
-if TYPE_CHECKING:
- # TODO: убрать кринж на рефакторинге
- from authentication.models import CustomUserModel
-
-
-class BaseAlready(Exception):
- def __init__(self, user: 'CustomUserModel'):
- self.user = user
-
- def msg(self): ...
@@ -1,8 +1,4 @@
-from authentication.exceptions.business_host_exceptions.base_already import (
- BaseAlready,
-)
-
-__all__ = ('BaseAlready',)
+__all__ = ()
class InvalidPassword(Exception): ...
@@ -4,3 +4,43 @@ from django.utils.translation import gettext as _
class BusinessAccountNotFound(Exception):
def __str__(self):
return _('Business account not found')
+
+
+class AdminReinviteForbidden(Exception):
+ def __str__(self):
+ return _('You cannot reinvite other administrators')
+
+
+class UnconfirmedUserChangePass(Exception):
+ def __str__(self):
+ return _('You cannot change the password of an unconfirmed e-mail user.')
+
+
+class AdminPasswordChangeForbidden(Exception):
+ def __str__(self):
+ return _('Admin staff cannot change the password of another admin staff member')
+
+
+class AdminUpdateForbidden(Exception):
+ def __str__(self):
+ return _('Admin staff cannot update other admin staff members')
+
+
+class AdminCreateForbidden(Exception):
+ def __str__(self):
+ return _('Admin staff cannot create other admin staff members')
+
+
+class AdminDeleteForbidden(Exception):
+ def __str__(self):
+ return _('Admin staff cannot delete other admin staff members')
+
+
+class AdminPromoteForbidden(Exception):
+ def __str__(self):
+ return _('Admin staff cannot promote other staff members to admin')
+
+
+class PasswordChangeRestricted(Exception):
+ def __str__(self):
+ return _('You can only change the password for regular employees or security staff members')
@@ -11,6 +11,11 @@ class WrongPassword(Exception):
return _('Wrong password')
+class PasswordsDoNotMatch(Exception):
+ def __str__(self):
+ return _("Passwords don't match")
+
+
class EmailNotConfirmed(Exception):
def __str__(self):
return _('User has not confirmed his email yet')
@@ -89,6 +89,10 @@ class BusinessUserHost(BaseModel):
is_log_history_enabled = models.BooleanField(default=False, verbose_name=_('Log history enabled'))
+ @property
+ def display_name(self) -> str:
+ return self.company_name or self.user.email
+
@property
def accounts(self) -> QuerySet[BusinessAccount]:
return self.accounts
@@ -1,16 +1,19 @@
-from decimal import Decimal
from typing import Any, OrderedDict, Tuple
from django.utils.translation import gettext_lazy as _
-from rest_framework.request import Request
+from authentication.exceptions.business_account import (
+ AdminPasswordChangeForbidden,
+ PasswordChangeRestricted,
+ UnconfirmedUserChangePass,
+)
+from authentication.exceptions.user import PasswordsDoNotMatch
from authentication.models import (
BusinessAccount,
BusinessUserHost,
CustomUserModel,
)
from authentication.models.choices import InvitationStatus
-from authentication.serializers import ChangePasswordSerializer
class BusinessAccountService:
@@ -35,22 +38,13 @@ class BusinessAccountService:
cls,
user: CustomUserModel,
host: BusinessUserHost,
- status: Tuple[str, Any] | None = None,
- account_privileges: Tuple[str, Any] | None = None,
- token_limit: Decimal | None = None,
+ account_privileges: str,
):
- account: BusinessAccount = BusinessAccount.objects.create(
+ account, _ = BusinessAccount.objects.update_or_create(
user=user,
parent_company=host,
+ defaults={'account_privileges': account_privileges},
)
-
- if status is not None:
- account.acceptance_status = status
- if account_privileges is not None:
- account.account_privileges = account_privileges
- if token_limit is not None:
- account.token_limit = token_limit
- account.save()
return cls(account)
@classmethod
@@ -63,9 +57,7 @@ class BusinessAccountService:
@classmethod
def get_company_name(cls, user: CustomUserModel):
- if user.account_type == 'business_host':
- return user.host_account
- return user.business_account.parent_company
+ return user.host or user.employee.parent_company
def update(self, data: OrderedDict) -> BusinessAccount:
if data['status'] == InvitationStatus.ACCEPTED:
@@ -86,34 +78,17 @@ class BusinessAccountService:
def reject(self):
self.update_status(InvitationStatus.REJECTED)
- def update_limit(self, new_balance: Decimal | None):
- self.account.token_limit = new_balance
- self.account.save()
-
- def update_privileges(self, new_privileges: str):
- self.account.account_privileges = new_privileges
- self.account.save()
-
- def update_user_password(self, request: Request):
- serializer = ChangePasswordSerializer(data=request.data)
- serializer.is_valid(raise_exception=True)
-
- if serializer.validated_data['password_1'] != serializer.validated_data['password_2']:
- raise Exception(_("Passwords don't match"))
-
- if (
- self.account.user.account_type == 'business_security'
- and request.user.account_type == 'business_security'
- ):
- raise Exception(_('You do not have sufficient rights to perform this action'))
+ def update_user_password(self, user: CustomUserModel, password_1: str, password_2: str):
+ if self.account.acceptance_status != InvitationStatus.ACCEPTED:
+ raise UnconfirmedUserChangePass
- if self.account.user.account_type not in ('business_account', 'business_security'):
- raise Exception(_('You do not have sufficient rights to perform this action'))
+ if password_1 != password_2:
+ raise PasswordsDoNotMatch
- if self.account.acceptance_status != InvitationStatus.ACCEPTED:
- raise Exception(_('You cannot change the password of an unconfirmed e-mail user.'))
+ if self.account.user.account_type == 'business_admin' == user.account_type:
+ raise AdminPasswordChangeForbidden
- self.account.user.set_password(serializer.validated_data['password_1'])
+ self.account.user.set_password(password_1)
self.account.user.save()
def update_status(self, status: Tuple[str, Any]):
@@ -1,17 +1,19 @@
from decimal import Decimal
from typing import Any, Tuple
+from uuid import UUID
-from django.db import IntegrityError
-from django.utils.translation import gettext_lazy as _
from rest_framework.request import Request
from authentication.exceptions import business_host_exceptions
-from authentication.exceptions.business_account import BusinessAccountNotFound
-from authentication.exceptions.business_host_exceptions import (
- AlreadyAccount,
- AlreadyHasPlan,
+from authentication.exceptions.business_account import (
+ AdminCreateForbidden,
+ AdminDeleteForbidden,
+ AdminPromoteForbidden,
+ AdminReinviteForbidden,
+ AdminUpdateForbidden,
+ BusinessAccountNotFound,
)
-from authentication.exceptions.business_host_exceptions.access_denied import AccessDenied
+from authentication.exceptions.business_host_exceptions import AlreadyAccount, InviteeHasPlan
from authentication.exceptions.business_host_exceptions.already_host import (
AlreadyHost,
)
@@ -19,8 +21,8 @@ from authentication.models import (
BusinessAccount,
BusinessUserHost,
CustomUserModel,
+ BusinessGroup,
)
-from authentication.models.choices import InvitationStatus
from authentication.selectors.account_status_selector import (
AccountStatusSelector,
)
@@ -32,13 +34,11 @@ from authentication.selectors.business_host_selector import (
)
from authentication.selectors.user_selector import UserSelector
from authentication.serializers import (
- AccountDataUpdateSerializer,
AddModelsSerializer,
BusinessAccountDataSerializer,
BusinessHostSerializer,
BusinessHostUpdateSerializer,
DeleteBusinessAccountSerializer,
- DeletedAccountDataSerializer,
DeleteModelsSerializer,
NewBusinessAccountSerializer,
NewBusinessHostSerializer,
@@ -67,66 +67,46 @@ class BusinessHostService:
self.user = user
def create_account(
- self,
- email: str,
- token_limit: Decimal | None = None,
- account_privileges: Tuple[str, Any] | None = None,
+ self, email: str, host: BusinessUserHost, account_privileges: str
) -> BusinessAccountService:
password = generate_token(15)
user = CustomUserModel.objects.create_user(email=email, password=password)
- user.save()
PaymentPlanService(user).subscribe_user_to_plan(PaymentPlan.objects.get(price=0, is_corporate=False))
- account_service = BusinessAccountService.create(
- user,
- self.user.host_account,
- account_privileges=account_privileges,
- )
- if token_limit is not None:
- account_service.update_limit(token_limit)
- EmailService(self.user).send_corporate_greeting_email(account_service.account, password)
+ account_service = BusinessAccountService.create(user, host, account_privileges=account_privileges)
+ EmailService.send_corporate_greeting_email(account_service.account, password)
return account_service
- def create_existing(
- self,
- email: str,
- token_limit: Decimal | None = None,
- account_privileges: Tuple[str, Any] | None = None,
- ) -> BusinessAccountService:
+ def create_existing(self, email: str, account_privileges: str, group: UUID | None = None) -> BusinessAccountService:
+ company = self.user.host or self.user.employee.parent_company
+ if self.user.account_type == 'business_admin' and account_privileges == 'admin':
+ raise AdminCreateForbidden
try:
user = UserSelector.get_by_email(email.lower())
- if AccountStatusSelector(user).is_business_host() and not user.is_deleted:
- raise AlreadyHost(user)
- if AccountStatusSelector(user).is_business_account() and not user.is_deleted:
- raise AlreadyAccount(user)
+ if user.host:
+ raise AlreadyHost
+ if user.employee and user.employee.parent_company != company:
+ raise AlreadyAccount
if PaymentPlanSelector(user).is_plan_paid():
- raise AlreadyHasPlan(user)
+ raise InviteeHasPlan
except CustomUserModel.DoesNotExist:
- return self.create_account(
- email=email,
- token_limit=token_limit,
- account_privileges=account_privileges,
- )
- try:
- account_service = BusinessAccountService.create(
- user,
- self.user.host_account,
- account_privileges=account_privileges,
- )
- except IntegrityError:
- account_service = BusinessAccountService.from_user(user)
- if token_limit is not None:
- account_service.update_limit(token_limit)
+ return self.create_account(email=email, host=company, account_privileges=account_privileges)
+ account_service = BusinessAccountService.create(
+ user,
+ company,
+ account_privileges=account_privileges,
+ )
+ account_service.account.group = BusinessGroup.objects.filter(uid=group, parent_company=company).first()
+ account_service.account.save()
if not user.is_deleted:
- EmailService(self.user).send_corporate_invitation_email(account_service.account)
+ EmailService.send_corporate_invitation_email(account_service.account, company)
else:
password = generate_token(15)
user.set_password(password)
user.is_deleted = False
user.save()
- EmailService(self.user).send_corporate_greeting_email(account_service.account, password)
-
+ EmailService.send_corporate_greeting_email(account_service.account, password)
return account_service
def create(self, request: Request) -> BusinessAccountDataSerializer:
@@ -151,51 +131,29 @@ class BusinessHostService:
logger.exception("Ошибка при создании учетной записи:")
raise APIException(f"Произошла ошибка при создании учетной записи: {exc}") from exc
- def update_token_limit(
- self,
- user: CustomUserModel,
- amount: Decimal,
- ):
- BusinessAccountSelector.from_user(user, company=self.user.host_account).to_service().update_limit(
- amount
- )
-
def update_user_invitation_status(self, user: CustomUserModel, new_status: Tuple[str, Any]):
BusinessAccountSelector.from_user(user, company=self.user.host_account).to_service().update_status(
new_status
)
- def update_privileges(self, user: CustomUserModel, new_privileges: str):
- BusinessAccountSelector.from_user(
- user, company=self.user.host_account
- ).to_service().update_privileges(new_privileges)
+ def update(self, email: str, token_limit: Decimal | None, privileges: str) -> BusinessAccount:
+ business_account = UserSelector.get_by_email(email).employee
+ if not business_account:
+ raise BusinessAccountNotFound
+
+ if business_account.user.account_type == 'business_admin' == self.user.account_type:
+ raise AdminUpdateForbidden
- def update(self, request: Request, **kwargs):
- serializer = AccountDataUpdateSerializer(data=request.data)
- serializer.is_valid(raise_exception=True)
+ if self.user.account_type == 'business_admin' and privileges == 'admin':
+ raise AdminPromoteForbidden
- user_email = kwargs.get('user_email', None)
- if user_email is None:
- raise Exception(_('No user_email is provided'))
-
- user = UserSelector.get_by_email(user_email)
- if not AccountStatusSelector(user).is_business_account():
- raise Exception("Business Account for this user doesn't exist")
- if token_limit := serializer.validated_data.get('token_limit', None):
- self.update_token_limit(
- user,
- token_limit,
- )
- if status := serializer.validated_data.get('status', False):
- self.update_user_invitation_status(user, status)
- if privileges := serializer.validated_data.get('account_privileges', None):
- self.update_privileges(user, privileges)
-
- account_type = UserSelector(user).check_account_type()
- if status == InvitationStatus.CANCELLED:
- return DeletedAccountDataSerializer(user, context={'account_type': 'regular'})
-
- return BusinessAccountDataSerializer(user.business_account, context={'account_type': account_type})
+ company = self.user.host or self.user.employee.parent_company
+ if business_account.parent_company != company:
+ raise AlreadyAccount
+ business_account.token_limit = token_limit
+ business_account.account_privileges = privileges
+ business_account.save()
+ return business_account
def update_self(self, request: Request, serialize: bool = True) -> BusinessHostSerializer:
serializer = BusinessHostUpdateSerializer(data=request.data)
@@ -220,36 +178,29 @@ class BusinessHostService:
return BusinessHostSerializer(company, context={'token_cap_enabled': company.token_cap_enabled})
return company
- def delete(self, request: Request):
- serializer = DeleteBusinessAccountSerializer(data=request.data)
- serializer.is_valid(raise_exception=True)
- account = BusinessAccount.objects.filter(
- parent_company=self.user.host_account, user__uid=serializer.validated_data['uid']
- ).first()
+ def delete(self, employee_user_uid: UUID):
+ company = self.user.host or self.user.employee.parent_company
+ account = BusinessAccount.objects.filter(parent_company=company, user__uid=employee_user_uid).first()
if not account:
raise BusinessAccountNotFound
+ if self.user.account_type == 'business_admin' == account.user.account_type:
+ raise AdminDeleteForbidden
+
account.delete()
def create_host(self, request: Request) -> UserDataSerializer:
serializer = NewBusinessHostSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
- if PaymentPlanSelector(self.user).is_plan_paid():
- raise business_host_exceptions.AlreadyHasPlan(self.user)
-
if AccountStatusSelector(self.user).is_business_host():
- raise business_host_exceptions.AlreadyHost(self.user)
+ raise business_host_exceptions.AlreadyHost()
if AccountStatusSelector(self.user).is_business_account():
- raise business_host_exceptions.AlreadyAccount(self.user)
+ raise business_host_exceptions.AlreadyAccount()
host = BusinessUserHost.objects.create(user=self.user, **serializer.validated_data)
- host.save()
- PaymentPlanService(self.user).subscribe_user_to_plan(
- PaymentPlan.objects.get(price=0, is_corporate=True)
- )
EmailService(self.user).send_copr_purchase_email(host)
account_type = UserSelector(self.user).check_account_type()
@@ -282,12 +233,10 @@ class BusinessHostService:
return BusinessHostSelector(self.user).get_allowed_models()
def reinvite_business_account(self, business_account: BusinessAccount) -> None:
- if (
- self.user.account_type == business_account.user.account_type
- and self.user.account_type == 'business_admin'
- ):
- raise AccessDenied
+ if self.user.account_type == 'business_admin' == business_account.user.account_type:
+ raise AdminReinviteForbidden
password = generate_token(15)
business_account.user.set_password(password)
+ business_account.user.is_deleted = False
business_account.user.save()
EmailService(self.user).send_reinvited_email(business_account, password)
@@ -1,11 +1,15 @@
import logging
+from datetime import datetime
+from decimal import Decimal
+from typing import Any, Sequence
import dns.resolver
from django.conf import settings
from django.core.mail import EmailMessage, send_mail
from django.template import TemplateDoesNotExist
from django.template.loader import get_template
-from django.utils.html import format_html, strip_tags
+from django.utils.html import strip_tags
+from django.utils.safestring import SafeString
from django.utils.translation import gettext_lazy as _
from authentication.exceptions.email_exceptions import LetterNotFound, LetterUnknownException
@@ -13,12 +17,7 @@ from authentication.exceptions.user import DomainNotFound
from authentication.models import BusinessAccount, BusinessUserHost
from authentication.models.user import CustomUserModel
from authentication.services.email_token_service import EmailTokenService
-from core.minio_service import MinIOService
-from reports.models.error_report import ErrorReport
-
-from django.core.mail import send_mail
-from django.utils.html import strip_tags
-
+from reports.domain import Report
logger = logging.getLogger(__name__)
@@ -32,16 +31,18 @@ class EmailService:
self.user.save()
return self.user.is_subscribed_to_emails
- def send_email(self, subject: str, message: str, user_email: str):
+ @staticmethod
+ def send_email(subject: str, message: str, user_emails: Sequence[str]) -> None:
try:
- email_domain = user_email.split('@')[-1]
- dns.resolver.resolve(email_domain, 'MX')
+ for user_email in user_emails:
+ email_domain = user_email.split('@')[-1]
+ dns.resolver.resolve(email_domain, 'MX')
send_mail(
subject=subject,
message=strip_tags(message),
html_message=message,
from_email=settings.EMAIL_HOST_USER,
- recipient_list=(user_email,),
+ recipient_list=user_emails,
auth_password=settings.EMAIL_HOST_PASSWORD,
)
except (dns.resolver.NoAnswer, dns.resolver.NXDOMAIN, dns.exception.Timeout):
@@ -50,27 +51,29 @@ class EmailService:
logger.exception(exc)
raise Exception(_('Error occured when proceed email sending'))
+ @staticmethod
+ def _render_letter_template(template_name: str, context: dict[str, Any]) -> SafeString:
+ try:
+ template = get_template(f'{template_name}.html')
+ html_message = template.render(context=context)
+ except TemplateDoesNotExist:
+ raise LetterNotFound
+ except Exception as exc:
+ raise LetterUnknownException from exc
+ return html_message
+
def send_reg_conf_email(self):
token = EmailTokenService(self.user).generate_user_token()
+ html_message = self._render_letter_template(
+ template_name='authentication/account_confirmation',
+ context={
+ 'email': self.user.email,
+ 'confirmation_url': settings.USER_CONFIRMATION_URL,
+ 'confirmation_token': token.key,
+ },
+ )
self.send_email(
- subject='Подтверждение аккаунта',
- message=format_html(
- """
- Добро пожаловать на платформу AIR!
-
- Ваша почта на платформе - {}
- Для того, чтобы подтвердить регистрацию - перейдите по ссылке ниже:
- [Нажмите для активации]
- Если ссылка не открывается по кнопке, попробуйте вставить ссылку в адресную строку:
- {}?token={}
- """,
- self.user.email,
- settings.USER_CONFIRMATION_URL,
- token.key,
- settings.USER_CONFIRMATION_URL,
- token.key,
- ),
- user_email=self.user.email,
+ subject='Подтверждение аккаунта', message=html_message, user_emails=(self.user.email,)
)
def send_password_reset_email(self):
@@ -78,106 +81,101 @@ class EmailService:
self.send_email(
subject='Сброс пароля',
message=f'{settings.USER_PASSWORD_RESET_URL}?token={token.key}',
- user_email=self.user.email,
+ user_emails=(self.user.email,),
)
- def send_error_email(self, error: ErrorReport):
- message = f"""
- Пользователь {error.author.email} сообщил об ошибке:
- {error.report_text}
- Скриншоты ошибки (если пользователь их приложил) - во вложении
- """
-
+ def send_error_email(self, report: Report) -> None:
+ html_message = self._render_letter_template(
+ template_name='authentication/error_email',
+ context={'email': self.user.email, 'report_text': report.message},
+ )
mail = EmailMessage(
- subject=f'Ошибка у пользователя {error.author.email}!',
- body=message,
+ subject=f'Ошибка у пользователя {self.user.email}!',
+ body=html_message,
from_email=settings.EMAIL_HOST_USER,
to=settings.ERROR_EMAIL_RECIPIENTS,
- reply_to=(error.author.email,),
+ # reply_to=(error.author.email,),
)
- if error.additional_images is not None:
- for _, img in error.additional_images.items():
- image = MinIOService().get_object('air-errors', img)
- mail.attach(img, image)
-
+ for img in report.attachments:
+ mail.attach(img.name, img.read(), img.content_type)
+ img.close()
mail.send(fail_silently=True)
def send_copr_purchase_email(self, host: BusinessUserHost):
- message = f"""
- Пользователь {host.user.email} зарегистрировал корпоративный аккаунт:
-
- Компания: {host.company_name}
- Сектор: {host.company_sector}
- ИНН: {host.ITN}
- ОГРН: {host.PSRN}
- Контактное лицо:
- {host.preferred_name}, {host.job_title}
- {host.corporate_phone}
- {host.corporate_email}
- """
-
- mail = EmailMessage(
+ html_message = self._render_letter_template(
+ template_name='authentication/corporate_purchase_email', context={'host': host}
+ )
+ self.send_email(
subject='Новый корпоративный аккаунт',
- body=message,
- from_email=settings.EMAIL_HOST_USER,
- to=('a.ippolitov@air.fail', 't.bikbov@air.fail'),
+ message=html_message,
+ user_emails=('a.ippolitov@air.fail', 't.bikbov@air.fail'),
)
- mail.send(fail_silently=True)
- def send_corporate_greeting_email(self, account: BusinessAccount, password: str | None = None):
- if self.user.host_account is None:
- raise Exception(_('Regular users cannot send introductory letters'))
+ @classmethod
+ def send_corporate_greeting_email(cls, account: BusinessAccount, password: str | None = None):
token = EmailTokenService(account.user).generate_user_token()
- try:
- template = get_template('authentication/corporate_greeting_email.html')
- html_message = template.render(
- context={
- 'company_name': self.user.host_account.company_name,
- 'invitation_url': settings.INVITATION_RESPONSE_URL,
- 'token': token.key,
- 'email': account.user.email,
- 'password': password,
- }
- )
- except TemplateDoesNotExist:
- raise LetterNotFound
- except Exception as exc:
- raise LetterUnknownException from exc
- self.send_email(
- subject='Ваш аккаунт на платформе AIR', message=html_message, user_email=account.user.email
+ html_message = cls._render_letter_template(
+ template_name='authentication/corporate_greeting_email',
+ context={
+ 'business_account': account,
+ 'invitation_url': settings.INVITATION_RESPONSE_URL,
+ 'token': token.key,
+ 'password': password,
+ },
+ )
+ cls.send_email(
+ subject='Ваш аккаунт на платформе AIR', message=html_message, user_emails=(account.user.email,)
)
def send_reinvited_email(self, account: BusinessAccount, password: str | None = None) -> None:
- try:
- template = get_template('authentication/reinvited_employee_letter.html')
- html_message = template.render(context={'email': account.user.email, 'password': password})
- except TemplateDoesNotExist:
- raise LetterNotFound
- except Exception as exc:
- raise LetterUnknownException from exc
+ token = EmailTokenService(account.user).generate_user_token()
+ html_message = self._render_letter_template(
+ template_name='authentication/corporate_greeting_email',
+ context={
+ 'business_account': account,
+ 'invitation_url': settings.INVITATION_RESPONSE_URL,
+ 'token': token.key,
+ 'password': password,
+ },
+ )
self.send_email(
- subject='Ваш аккаунт на платформе AIR', message=html_message, user_email=account.user.email
+ subject='Ваш аккаунт на платформе AIR', message=html_message, user_emails=(account.user.email,)
)
- def send_corporate_invitation_email(self, account: BusinessAccount):
- if self.user.host_account is None:
- raise Exception(_('Regular users cannot send invitation letters'))
+ @classmethod
+ def send_corporate_invitation_email(cls, account: BusinessAccount, host: BusinessUserHost) -> None:
token = EmailTokenService(account.user).generate_user_token()
- message = format_html(
- """
- Добрый день, {}!
- Компания {} приглашает вас получить доступ
- к корпоративному аккаунту на нашей платформе.
- Чтобы ответить на приглашение - перейдите по ссылке ниже:
- [Нажмите для подтверждения]
- Если ссылка не открывается по кнопке, попробуйте вставить ссылку в адресную строку:
- {}?token={}
- """,
- account.user.email,
- self.user.host_account.company_name,
- settings.INVITATION_RESPONSE_URL,
- token.key,
- settings.INVITATION_RESPONSE_URL,
- token.key,
+ html_message = cls._render_letter_template(
+ template_name='authentication/corporate_invitation_email',
+ context={
+ 'email': account.user.email,
+ 'host': host,
+ 'invite_url': settings.INVITATION_RESPONSE_URL,
+ 'token': token.key,
+ },
+ )
+ cls.send_email('Приглашение на платформу AIR', html_message, (account.user.email,))
+
+ @classmethod
+ def send_low_balance_email(cls, host: BusinessUserHost) -> None:
+ html_message = cls._render_letter_template(
+ template_name='authentication/low_balance_email', context={'host': host}
+ )
+ cls.send_email(
+ f'AIR: баланс корпоративного аккаунта ниже {host.token_cap}', html_message, host.token_cap_emails
)
- self.send_email('', message, account.user.email)
+
+ @classmethod
+ def send_payment_email(
+ cls, product_title: str, next_payment_at: datetime | str, amount: Decimal, payer_email: str, status: str
+ ) -> None:
+ templates = {
+ 'succeeded': 'authentication/success_payment_notification',
+ 'canceled': 'authentication/canceled_payment_notification',
+ }
+ html_message = cls._render_letter_template(
+ template_name=templates[status],
+ context={'product_title': product_title, 'next_payment_at': next_payment_at, 'amount': amount},
+ )
+ cls.send_email('Детали платежа', html_message, (payer_email,))
+
@@ -16,7 +16,7 @@ class EmailTokenService:
return EmailToken.objects.create(user=self.user, key=generate_token(22))
@classmethod
- def get_token(self, key: str) -> EmailToken:
+ def get_token(cls, key: str) -> EmailToken:
try:
return EmailToken.objects.get(key=key)
except EmailToken.DoesNotExist:
@@ -89,7 +89,6 @@ class UserService:
)
except CustomUserModel.DoesNotExist:
pass
-
return user
def create_user_telegram(self, request: Request) -> TelegramUser:
@@ -218,7 +217,7 @@ class UserService:
token = EmailTokenSelector.get_email_token(key=key)
if token is None:
- raise Exception(_("No token like this in a database"))
+ raise EmailTokenNotFound
serializer = ChangePasswordSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
@@ -0,0 +1,16 @@
+
+
+
+
+ Подтверждение аккаунта
+
+
+ Добро пожаловать на платформу AIR!
+
+ Ваша почта на платформе - {{ email }}
+ Для того, чтобы подтвердить регистрацию - перейдите по ссылке ниже:
+ [Нажмите для активации]
+ Если ссылка не открывается по кнопке, попробуйте вставить ссылку в адресную строку:
+ {{ confirmation_url }}?token={{ confirmation_token }}
+
+
\ No newline at end of file
@@ -0,0 +1,22 @@
+
+
+
+
+ Неуспешный платеж
+
+
+ Здравствуйте!
+
+ К сожалению, платёж за восстановление баланса по плану на сумму {{ amount }} не был успешно
+ обработан.
+
+ Следующая попытка списания будет выполнена: {{ next_payment_at }}.
+
+
+ Пожалуйста, убедитесь, что на вашем счёте достаточно средств или
+ обновите платёжные данные, чтобы избежать прерывания доступа.
+
+
+ Если у вас возникнут вопросы, вы всегда можете обратиться в нашу службу поддержки.
+
+
@@ -7,7 +7,7 @@
Добро пожаловать на платформу AIR!
- Этот аккаунт был создан для вас {{ company_name }} в рабочих целях.
+ Этот аккаунт был создан для вас {{ business_account.parent_company.display_name }} в рабочих целях.
Вы можете использовать все ресурсы корпоративного аккаунта — оплачивать
подписку нет необходимости.
@@ -18,8 +18,7 @@
{{ invitation_url }}?token={{ token }}
Ваши данные для входа:
- E-mail: {{ email }}
+ E-mail: {{ business_account.user.email }}
Пароль: {{ password }}
- Всегда с вами, команда AIR