from typing import Any import email_validator from anydi import singleton from bs4 import BeautifulSoup from fansifter_common.exceptions import Error, FansifterError from pydantic import BaseModel, ConfigDict, Field, ValidationError from email_campaigns.emails.exceptions import ( EmailContentRequiredFieldsError, EmailOptInButtonMissingError, EmailPrivacyFooterMissingError, EmailSenderEmailNotValidError, ) from email_campaigns.emails.models import BaseEmail from email_campaigns.emails.types import EmailContentErrorSummary BlockCheck = tuple[str, type[FansifterError]] class EmailContentSchema(BaseModel): model_config = ConfigDict(str_strip_whitespace=True) sender_name: str = Field(min_length=1) subject: str = Field(min_length=1) email_domain_id: str email_username: str = Field(min_length=1) html_content: str = Field(min_length=1) @singleton class EmailValidator: privacy_footer_class_name = "esd-privacy-links" opt_in_button_class_name = "esd-cta-button-block" @staticmethod def validate_required_content( email: BaseEmail, raise_on_error: bool = False, data: dict[str, Any] | None = None, ) -> EmailContentErrorSummary | None: """Validate email content.""" data = {**email.filled_data, **(data or {})} try: EmailContentSchema.model_validate(data) except ValidationError as exc: errors = { error["loc"][-1]: Error(code=error["type"], message=error["msg"]) for error in exc.errors() } field_errors = EmailContentErrorSummary( sender_name=errors.get("sender_name"), subject=errors.get("subject"), email_domain_id=errors.get("email_domain_id"), email_username=errors.get("email_username"), html_content=errors.get("html_content"), ) if raise_on_error: raise EmailContentRequiredFieldsError(field_errors) from exc return field_errors return None @staticmethod def validate_block_presence( html_content: str | None, checks: list[BlockCheck], ) -> None: if html_content is None: _, error_cls = checks[0] raise error_cls soup = BeautifulSoup(html_content, "html.parser") for class_name, error_cls in checks: if not soup.find(class_=class_name): raise error_cls def validate_privacy_footer_block_presence(self, html_content: str | None) -> None: self.validate_block_presence( html_content, [(self.privacy_footer_class_name, EmailPrivacyFooterMissingError)], ) def validate_opt_in_button_block_presence(self, html_content: str | None) -> None: self.validate_block_presence( html_content, [(self.opt_in_button_class_name, EmailOptInButtonMissingError)], ) @staticmethod def validate_sender_email(sender_email: str) -> None: """Validate that sender email is valid.""" try: email_validator.validate_email(sender_email, check_deliverability=False) except email_validator.EmailNotValidError as exc: raise EmailSenderEmailNotValidError from exc