from dataclasses import dataclass import email_validator from anydi import singleton from fansifter_common.auth.requests import AuthAccountRequest from fansifter_common.auth.services import AuthService from fansifter_common.auth.types import Permission from email_campaigns.adapters.db.decorators import transactional from email_campaigns.automated.enums import AutomatedEmailType from email_campaigns.automated.exceptions import ( AutomatedEmailSenderEmailNotValidError, AutomatedEmailTypeChangeNotAllowedError, ) from email_campaigns.automated.models import AutomatedEmail from email_campaigns.automated.repositories import AutomatedEmailRepository from email_campaigns.automated.validators import AutomatedEmailValidator from email_campaigns.emails.cache import EmailPublicPreviewCache from email_campaigns.emails.exceptions import EmailNotFoundError from email_campaigns.emails.services import EmailDomainService from email_campaigns.rosters.services import FanDataListService from email_campaigns.rosters.types import FanDataListId @dataclass(kw_only=True) class UpdateAutomatedEmailRequest(AuthAccountRequest): email_id: str name: str fandata_list_id: FanDataListId type: AutomatedEmailType sender_name: str | None email_domain_id: str | None email_username: str | None subject: str | None preview_text: str | None @singleton class UpdateAutomatedEmailHandler: # TODO: ---> Permission("automated_email", "edit") permission = Permission("email_campaign", "edit") def __init__( self, auth_service: AuthService, automated_email_repository: AutomatedEmailRepository, automated_email_validator: AutomatedEmailValidator, email_domain_service: EmailDomainService, fandata_list_service: FanDataListService, preview_cache: EmailPublicPreviewCache, ) -> None: self.auth_service = auth_service self.automated_email_repository = automated_email_repository self.automated_email_validator = automated_email_validator self.email_domain_service = email_domain_service self.fandata_list_service = fandata_list_service self.preview_cache = preview_cache @transactional def handle(self, request: UpdateAutomatedEmailRequest) -> AutomatedEmail: automated_email = self.automated_email_repository.get(request.email_id) if automated_email is None: raise EmailNotFoundError # Check if user has permission to edit email campaign self.auth_service.authorize_account( request.identity_id, account=request.account, permission=self.permission, ) # Disallow changing the automated email type after creation if request.type != automated_email.type: raise AutomatedEmailTypeChangeNotAllowedError # Validate name self.automated_email_validator.validate_name( name=request.name, account=request.account, automated_email=automated_email, ) # Validate global participant id if changed if ( request.fandata_list_id.is_artist and request.fandata_list_id.id != automated_email.global_participant_id ): self.fandata_list_service.resolve_global_participant( request.fandata_list_id.id ) elif ( request.fandata_list_id.is_custom_list and request.fandata_list_id.id != automated_email.custom_list_id ): self.fandata_list_service.resolve_allowed_custom_list( request.fandata_list_id.id, account=request.account, ) # Resolve allowed email domain email_domain = self.email_domain_service.resolve_allowed_domain( email_domain_id=request.email_domain_id, identity_id=request.identity_id, account=request.account, ) # Validate sender email if email_domain and request.sender_name and request.email_username: try: email_validator.validate_email( f"{request.email_username}@{email_domain.domain}", check_deliverability=False, allow_smtputf8=False, ) except email_validator.EmailNotValidError as exc: raise AutomatedEmailSenderEmailNotValidError( f"Invalid email username '{request.email_username}': {exc}" ) from exc # Validate main rep only artists if changed if request.fandata_list_id.is_artist and ( request.fandata_list_id.id != automated_email.global_participant_id ): # Local reps (IRLs) are allowed to manage automated emails. self.fandata_list_service.ensure_artist_access( request.fandata_list_id, account=request.account, allow_local_rep=True, ) automated_email.name = request.name automated_email.account = request.account automated_email.fandata_list_id = request.fandata_list_id automated_email.sender_name = request.sender_name automated_email.email_domain_id = request.email_domain_id automated_email.email_domain = email_domain automated_email.email_username = request.email_username automated_email.subject = request.subject automated_email.preview_text = request.preview_text automated_email.updated_by = request.identity_id self.automated_email_repository.save(automated_email) # Invalidate cache after campaign update self.preview_cache.delete( automated_email.id, email_type=automated_email.email_type ) return automated_email