from dataclasses import replace from anydi import singleton from fansifter_common.adapters.ows_account import OwsAccountClient from fansifter_common.adapters.ows_users import OwsUsersClient from fansifter_common.adapters.sendgrid import SendGridClient from fansifter_common.auth.services import AuthService from fansifter_common.auth.types import Permission from fansifter_common.legal_info.services import LegalInfoService from fansifter_common.translation.services import TranslationService from email_campaigns.adapters.db import Repository from email_campaigns.adapters.features import ( AUDIENCE_ENABLE_STRIPO_PLUGIN_V2, Features, ) from email_campaigns.adapters.stripo import AutomatedStripoClientV2 from email_campaigns.automated.models import AutomatedEmail from email_campaigns.emails.exceptions import EmailNotFoundError from email_campaigns.emails.handlers import ( GetEmailPreviewHandler, GetEmailPreviewRequest, ) from email_campaigns.emails.types import EmailPreview from email_campaigns.fonts.services.google_fonts import GoogleFontsService @singleton class GetAutomatedEmailPreviewHandler(GetEmailPreviewHandler[AutomatedEmail]): # TODO: ---> Permission("automated_email", "view") permission = Permission("email_campaign", "view") def __init__( self, auth_service: AuthService, email_repository: Repository[AutomatedEmail], sendgrid_client: SendGridClient, legal_info_service: LegalInfoService, translation_service: TranslationService, ows_account_client: OwsAccountClient, ows_users_client: OwsUsersClient, fonts_service: GoogleFontsService, stripo_v2_client: AutomatedStripoClientV2, features: Features, ) -> None: super().__init__( auth_service=auth_service, email_repository=email_repository, sendgrid_client=sendgrid_client, legal_info_service=legal_info_service, translation_service=translation_service, ows_account_client=ows_account_client, ows_users_client=ows_users_client, fonts_service=fonts_service, ) self.stripo_v2_client = stripo_v2_client self.features = features def handle(self, request: GetEmailPreviewRequest) -> EmailPreview: if not request.with_unapplied_changes or not self.features.is_on_for_identity( AUDIENCE_ENABLE_STRIPO_PLUGIN_V2, identity_id=request.identity_id, ): return super().handle(request) email = self.email_repository.get(request.email_id) if email is None: raise EmailNotFoundError self.auth_service.authorize_account( request.identity_id, account=email.account, permission=self.permission, ) response = self.stripo_v2_client.get_html_css( email_id=email.id, user_id=request.identity_id, ) preview = replace(email.preview, html=response.html, css=response.css) return self.apply_substitutions(preview, email=email, request=request)