blossomtune_gradio/mail.py

import smtplib
from abc import ABC, abstractmethod
from email.mime.text import MIMEText
import requests

from blossomtune_gradio.logs import log
from blossomtune_gradio import config as cfg


class EmailSender(ABC):
    """
    Abstract Base Class for email sending.

    This class defines the interface for all email sending implementations,
    ensuring they have a consistent `send_email` method.
    """

    @abstractmethod
    def send_email(
        self, recipient_email: str, subject: str, body: str
    ) -> tuple[bool, str]:
        """
        Sends an email to the specified recipient.

        Args:
            recipient_email: The email address of the recipient.
            subject: The subject line of the email.
            body: The body content of the email.

        Returns:
            A tuple containing a boolean success status and an error message string.
            The error message is empty if the email was sent successfully.
        """
        pass


class SMTPMailSender(EmailSender):
    """
    Concrete implementation of EmailSender using standard SMTP.
    """

    def send_email(
        self, recipient_email: str, subject: str, body: str
    ) -> tuple[bool, str]:
        """
        Sends an email using the SMTP server configured in `cfg`.
        """
        msg = MIMEText(body)
        msg["Subject"] = subject
        msg["From"] = cfg.SMTP_SENDER
        msg["To"] = recipient_email

        try:
            with smtplib.SMTP(cfg.SMTP_SERVER, cfg.SMTP_PORT) as server:
                if cfg.SMTP_REQUIRE_TLS:
                    server.starttls()
                    server.login(cfg.SMTP_USER, cfg.SMTP_PASSWORD)
                server.send_message(msg)
            log("[Email] SMTP email sent.")
            return True, ""
        except Exception as e:
            log(f"[Email] CRITICAL ERROR sending to {recipient_email} via SMTP: {e}")
            return False, f"Error sending email via SMTP: {e}"


class MailjetSender(EmailSender):
    """
    Concrete implementation of EmailSender using the Mailjet API.
    """

    def send_email(
        self, recipient_email: str, subject: str, body: str
    ) -> tuple[bool, str]:
        """
        Sends an email using the Mailjet transactional API v3.1.
        """
        # Check for truthy values, not just attribute existence.
        if not (cfg.SMTP_USER and cfg.SMTP_PASSWORD):
            error_msg = "Mailjet API keys are not configured."
            log(f"[Email] {error_msg}")
            return False, error_msg

        api_key = cfg.SMTP_USER
        api_secret = cfg.SMTP_PASSWORD

        url = "https://api.mailjet.com/v3.1/send"
        data = {
            "Messages": [
                {
                    "From": {
                        "Email": cfg.SMTP_SENDER,
                        "Name": cfg.SMTP_SENDER.split("@")[0],
                    },
                    "To": [{"Email": recipient_email}],
                    "Subject": subject,
                    "TextPart": body,
                }
            ]
        }

        try:
            response = requests.post(url, auth=(api_key, api_secret), json=data)
            response.raise_for_status()
            log(f"[Email] Mailjet email sent. Status: {response.status_code}")
            return True, ""
        except requests.exceptions.RequestException as e:
            error_msg = f"Error sending email via Mailjet API: {e}. Response: {e.response.text if e.response else 'No response'}"
            log(f"[Email] CRITICAL ERROR: {error_msg}")
            return False, error_msg


def get_email_sender() -> EmailSender:
    """
    Factory function to get the correct email sender implementation.

    This function reads the `EMAIL_PROVIDER` variable from the `config` module
    and returns an appropriate EmailSender instance. Defaults to SMTP.
    """
    provider = getattr(cfg, "EMAIL_PROVIDER", "smtp")
    if provider == "mailjet":
        return MailjetSender()
    # Default to SMTP if the provider is not Mailjet or is missing.
    return SMTPMailSender()


def send_activation_email(
    recipient_email: str, activation_code: str
) -> tuple[bool, str]:
    """
    Sends the activation code to the user using the configured email provider.

    This function uses the factory to get the correct sender and abstracts the
    implementation details.
    """
    subject = "Your BlossomTune Activation Code"
    body = (
        "Welcome to BlossomTune!\n\n"
        "Please use the following code to activate your participation request:\n\n"
        f"{activation_code}\n\n"
        "Thank you!"
    )

    sender = get_email_sender()
    success, error_message = sender.send_email(recipient_email, subject, body)

    if not success:
        return (
            False,
            f"There was an error sending the activation email. Please contact an administrator. Original error: {error_message}",
        )

    return True, ""