from dataclasses import dataclass 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 fansifter_common.auth.validators import AccountValidator from email_campaigns.adapters.db import DB from email_campaigns.automated.enums import AutomatedEmailType 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.services import EmailDomainService from email_campaigns.emails.validators import EmailValidator from email_campaigns.rosters.services import FanDataListService from email_campaigns.rosters.types import FanDataListId @dataclass(kw_only=True) class CreateAutomatedEmailRequest(AuthAccountRequest): name: str type: AutomatedEmailType fandata_list_id: FanDataListId html_content: str | None css_content: str | None @singleton class CreateAutomatedEmailHandler: # TODO: ---> Permission("automated_email", "create") permission = Permission("email_campaign", "create") def __init__( self, db: DB, auth_service: AuthService, account_validator: AccountValidator, email_domain_service: EmailDomainService, fandata_list_service: FanDataListService, automated_email_repository: AutomatedEmailRepository, automated_email_validator: AutomatedEmailValidator, email_validator: EmailValidator, ) -> None: self.db = db self.auth_service = auth_service self.account_validator = account_validator self.email_domain_service = email_domain_service self.fandata_list_service = fandata_list_service self.automated_email_repository = automated_email_repository self.automated_email_validator = automated_email_validator self.email_validator = email_validator def handle(self, request: CreateAutomatedEmailRequest) -> AutomatedEmail: with self.db.transaction(): # Check user permissions self.auth_service.authorize_account( request.identity_id, account=request.account, permission=self.permission, ) # Validate account self.account_validator.validate_account( request.account, allow_subaccount=False ) # Validate name uniqueness (subclass-specific) self.automated_email_validator.validate_name( request.name, account=request.account ) # Validate required content blocks if html_content is provided if request.html_content is not None: self.email_validator.validate_privacy_footer_block_presence( request.html_content ) if request.type == AutomatedEmailType.OPT_IN: self.email_validator.validate_opt_in_button_block_presence( request.html_content ) # Ensure account has access to artist fandata list. Local reps (IRLs) # are allowed to manage automated emails for their artists. self.fandata_list_service.ensure_artist_access( request.fandata_list_id, account=request.account, allow_local_rep=True, ) # Resolve fandata list and set sender name sender_name = "" if request.fandata_list_id.is_artist: global_participant = ( self.fandata_list_service.resolve_global_participant( request.fandata_list_id.id ) ) sender_name = global_participant.name elif request.fandata_list_id.is_custom_list: custom_list = self.fandata_list_service.resolve_allowed_custom_list( request.fandata_list_id.id, account=request.account ) sender_name = custom_list.name # Setup email domain and username email_domain = self.email_domain_service.get_allowed_domain( identity_id=request.identity_id, account=request.account ) email_username = None if email_domain and sender_name: email_username = self.email_domain_service.create_email_username( sender_name=sender_name, domain_name=email_domain.domain ) # Create automated email automated_email = AutomatedEmail( name=request.name, type=request.type, vendor_id=request.vendor_id, subaccount_id=request.subaccount_id, created_by=request.identity_id, updated_by=request.identity_id, ) automated_email.sender_name = sender_name automated_email.email_username = email_username automated_email.email_domain = email_domain automated_email.fandata_list_id = request.fandata_list_id automated_email.css_content = request.css_content automated_email.html_content = request.html_content automated_email.source_campaign_connections = [] # Save automated email self.automated_email_repository.save(automated_email) return automated_email