dify/api/libs/email_i18n.py

"""
Email Internationalization Module

This module provides a centralized, elegant way to handle email internationalization
in Dify. It follows Domain-Driven Design principles with proper type hints and
eliminates the need for repetitive language switching logic.
"""

from dataclasses import dataclass
from enum import Enum
from typing import Any, Optional, Protocol

from flask import render_template
from pydantic import BaseModel, Field

from extensions.ext_mail import mail
from services.feature_service import BrandingModel, FeatureService


class EmailType(Enum):
    """Enumeration of supported email types."""

    RESET_PASSWORD = "reset_password"
    INVITE_MEMBER = "invite_member"
    EMAIL_CODE_LOGIN = "email_code_login"
    CHANGE_EMAIL_OLD = "change_email_old"
    CHANGE_EMAIL_NEW = "change_email_new"
    OWNER_TRANSFER_CONFIRM = "owner_transfer_confirm"
    OWNER_TRANSFER_OLD_NOTIFY = "owner_transfer_old_notify"
    OWNER_TRANSFER_NEW_NOTIFY = "owner_transfer_new_notify"
    ACCOUNT_DELETION_SUCCESS = "account_deletion_success"
    ACCOUNT_DELETION_VERIFICATION = "account_deletion_verification"
    ENTERPRISE_CUSTOM = "enterprise_custom"
    QUEUE_MONITOR_ALERT = "queue_monitor_alert"
    DOCUMENT_CLEAN_NOTIFY = "document_clean_notify"


class EmailLanguage(Enum):
    """Supported email languages with fallback handling."""

    EN_US = "en-US"
    ZH_HANS = "zh-Hans"

    @classmethod
    def from_language_code(cls, language_code: str) -> "EmailLanguage":
        """Convert a language code to EmailLanguage with fallback to English."""
        if language_code == "zh-Hans":
            return cls.ZH_HANS
        return cls.EN_US


@dataclass(frozen=True)
class EmailTemplate:
    """Immutable value object representing an email template configuration."""

    subject: str
    template_path: str
    branded_template_path: str


@dataclass(frozen=True)
class EmailContent:
    """Immutable value object containing rendered email content."""

    subject: str
    html_content: str
    template_context: dict[str, Any]


class EmailI18nConfig(BaseModel):
    """Configuration for email internationalization."""

    model_config = {"frozen": True, "extra": "forbid"}

    templates: dict[EmailType, dict[EmailLanguage, EmailTemplate]] = Field(
        default_factory=dict, description="Mapping of email types to language-specific templates"
    )

    def get_template(self, email_type: EmailType, language: EmailLanguage) -> EmailTemplate:
        """Get template configuration for specific email type and language."""
        type_templates = self.templates.get(email_type)
        if not type_templates:
            raise ValueError(f"No templates configured for email type: {email_type}")

        template = type_templates.get(language)
        if not template:
            # Fallback to English if specific language not found
            template = type_templates.get(EmailLanguage.EN_US)
            if not template:
                raise ValueError(f"No template found for {email_type} in {language} or English")

        return template


class EmailRenderer(Protocol):
    """Protocol for email template renderers."""

    def render_template(self, template_path: str, **context: Any) -> str:
        """Render email template with given context."""
        ...


class FlaskEmailRenderer:
    """Flask-based email template renderer."""

    def render_template(self, template_path: str, **context: Any) -> str:
        """Render email template using Flask's render_template."""
        return render_template(template_path, **context)


class BrandingService(Protocol):
    """Protocol for branding service abstraction."""

    def get_branding_config(self) -> BrandingModel:
        """Get current branding configuration."""
        ...


class FeatureBrandingService:
    """Feature service based branding implementation."""

    def get_branding_config(self) -> BrandingModel:
        """Get branding configuration from feature service."""
        return FeatureService.get_system_features().branding


class EmailSender(Protocol):
    """Protocol for email sending abstraction."""

    def send_email(self, to: str, subject: str, html_content: str) -> None:
        """Send email with given parameters."""
        ...


class FlaskMailSender:
    """Flask-Mail based email sender."""

    def send_email(self, to: str, subject: str, html_content: str) -> None:
        """Send email using Flask-Mail."""
        if mail.is_inited():
            mail.send(to=to, subject=subject, html=html_content)


class EmailI18nService:
    """
    Main service for internationalized email handling.

    This service provides a clean API for sending internationalized emails
    with proper branding support and template management.
    """

    def __init__(
        self,
        config: EmailI18nConfig,
        renderer: EmailRenderer,
        branding_service: BrandingService,
        sender: EmailSender,
    ) -> None:
        self._config = config
        self._renderer = renderer
        self._branding_service = branding_service
        self._sender = sender

    def send_email(
        self,
        email_type: EmailType,
        language_code: str,
        to: str,
        template_context: Optional[dict[str, Any]] = None,
    ) -> None:
        """
        Send internationalized email with branding support.

        Args:
            email_type: Type of email to send
            language_code: Target language code
            to: Recipient email address
            template_context: Additional context for template rendering
        """
        if template_context is None:
            template_context = {}

        language = EmailLanguage.from_language_code(language_code)
        email_content = self._render_email_content(email_type, language, template_context)

        self._sender.send_email(to=to, subject=email_content.subject, html_content=email_content.html_content)

    def send_change_email(
        self,
        language_code: str,
        to: str,
        code: str,
        phase: str,
    ) -> None:
        """
        Send change email notification with phase-specific handling.

        Args:
            language_code: Target language code
            to: Recipient email address
            code: Verification code
            phase: Either 'old_email' or 'new_email'
        """
        if phase == "old_email":
            email_type = EmailType.CHANGE_EMAIL_OLD
        elif phase == "new_email":
            email_type = EmailType.CHANGE_EMAIL_NEW
        else:
            raise ValueError(f"Invalid phase: {phase}. Must be 'old_email' or 'new_email'")

        self.send_email(
            email_type=email_type,
            language_code=language_code,
            to=to,
            template_context={
                "to": to,
                "code": code,
            },
        )

    def send_raw_email(
        self,
        to: str | list[str],
        subject: str,
        html_content: str,
    ) -> None:
        """
        Send a raw email directly without template processing.

        This method is provided for backward compatibility with legacy email
        sending that uses pre-rendered HTML content (e.g., enterprise emails
        with custom templates).

        Args:
            to: Recipient email address(es)
            subject: Email subject
            html_content: Pre-rendered HTML content
        """
        if isinstance(to, list):
            for recipient in to:
                self._sender.send_email(to=recipient, subject=subject, html_content=html_content)
        else:
            self._sender.send_email(to=to, subject=subject, html_content=html_content)

    def _render_email_content(
        self,
        email_type: EmailType,
        language: EmailLanguage,
        template_context: dict[str, Any],
    ) -> EmailContent:
        """Render email content with branding and internationalization."""
        template_config = self._config.get_template(email_type, language)
        branding = self._branding_service.get_branding_config()

        # Determine template path based on branding
        template_path = template_config.branded_template_path if branding.enabled else template_config.template_path

        # Prepare template context with branding information
        full_context = {
            **template_context,
            "branding_enabled": branding.enabled,
            "application_title": branding.application_title if branding.enabled else "Dify",
        }

        # Render template
        html_content = self._renderer.render_template(template_path, **full_context)

        # Apply templating to subject with all context variables
        subject = template_config.subject
        try:
            subject = subject.format(**full_context)
        except KeyError:
            # If template variables are missing, fall back to basic formatting
            if branding.enabled and "{application_title}" in subject:
                subject = subject.format(application_title=branding.application_title)

        return EmailContent(
            subject=subject,
            html_content=html_content,
            template_context=full_context,
        )


def create_default_email_config() -> EmailI18nConfig:
    """Create default email i18n configuration with all supported templates."""
    templates: dict[EmailType, dict[EmailLanguage, EmailTemplate]] = {
        EmailType.RESET_PASSWORD: {
            EmailLanguage.EN_US: EmailTemplate(
                subject="Set Your {application_title} Password",
                template_path="reset_password_mail_template_en-US.html",
                branded_template_path="without-brand/reset_password_mail_template_en-US.html",
            ),
            EmailLanguage.ZH_HANS: EmailTemplate(
                subject="设置您的 {application_title} 密码",
                template_path="reset_password_mail_template_zh-CN.html",
                branded_template_path="without-brand/reset_password_mail_template_zh-CN.html",
            ),
        },
        EmailType.INVITE_MEMBER: {
            EmailLanguage.EN_US: EmailTemplate(
                subject="Join {application_title} Workspace Now",
                template_path="invite_member_mail_template_en-US.html",
                branded_template_path="without-brand/invite_member_mail_template_en-US.html",
            ),
            EmailLanguage.ZH_HANS: EmailTemplate(
                subject="立即加入 {application_title} 工作空间",
                template_path="invite_member_mail_template_zh-CN.html",
                branded_template_path="without-brand/invite_member_mail_template_zh-CN.html",
            ),
        },
        EmailType.EMAIL_CODE_LOGIN: {
            EmailLanguage.EN_US: EmailTemplate(
                subject="{application_title} Login Code",
                template_path="email_code_login_mail_template_en-US.html",
                branded_template_path="without-brand/email_code_login_mail_template_en-US.html",
            ),
            EmailLanguage.ZH_HANS: EmailTemplate(
                subject="{application_title} 登录验证码",
                template_path="email_code_login_mail_template_zh-CN.html",
                branded_template_path="without-brand/email_code_login_mail_template_zh-CN.html",
            ),
        },
        EmailType.CHANGE_EMAIL_OLD: {
            EmailLanguage.EN_US: EmailTemplate(
                subject="Check your current email",
                template_path="change_mail_confirm_old_template_en-US.html",
                branded_template_path="without-brand/change_mail_confirm_old_template_en-US.html",
            ),
            EmailLanguage.ZH_HANS: EmailTemplate(
                subject="检测您现在的邮箱",
                template_path="change_mail_confirm_old_template_zh-CN.html",
                branded_template_path="without-brand/change_mail_confirm_old_template_zh-CN.html",
            ),
        },
        EmailType.CHANGE_EMAIL_NEW: {
            EmailLanguage.EN_US: EmailTemplate(
                subject="Confirm your new email address",
                template_path="change_mail_confirm_new_template_en-US.html",
                branded_template_path="without-brand/change_mail_confirm_new_template_en-US.html",
            ),
            EmailLanguage.ZH_HANS: EmailTemplate(
                subject="确认您的邮箱地址变更",
                template_path="change_mail_confirm_new_template_zh-CN.html",
                branded_template_path="without-brand/change_mail_confirm_new_template_zh-CN.html",
            ),
        },
        EmailType.OWNER_TRANSFER_CONFIRM: {
            EmailLanguage.EN_US: EmailTemplate(
                subject="Verify Your Request to Transfer Workspace Ownership",
                template_path="transfer_workspace_owner_confirm_template_en-US.html",
                branded_template_path="without-brand/transfer_workspace_owner_confirm_template_en-US.html",
            ),
            EmailLanguage.ZH_HANS: EmailTemplate(
                subject="验证您转移工作空间所有权的请求",
                template_path="transfer_workspace_owner_confirm_template_zh-CN.html",
                branded_template_path="without-brand/transfer_workspace_owner_confirm_template_zh-CN.html",
            ),
        },
        EmailType.OWNER_TRANSFER_OLD_NOTIFY: {
            EmailLanguage.EN_US: EmailTemplate(
                subject="Workspace ownership has been transferred",
                template_path="transfer_workspace_old_owner_notify_template_en-US.html",
                branded_template_path="without-brand/transfer_workspace_old_owner_notify_template_en-US.html",
            ),
            EmailLanguage.ZH_HANS: EmailTemplate(
                subject="工作区所有权已转移",
                template_path="transfer_workspace_old_owner_notify_template_zh-CN.html",
                branded_template_path="without-brand/transfer_workspace_old_owner_notify_template_zh-CN.html",
            ),
        },
        EmailType.OWNER_TRANSFER_NEW_NOTIFY: {
            EmailLanguage.EN_US: EmailTemplate(
                subject="You are now the owner of {WorkspaceName}",
                template_path="transfer_workspace_new_owner_notify_template_en-US.html",
                branded_template_path="without-brand/transfer_workspace_new_owner_notify_template_en-US.html",
            ),
            EmailLanguage.ZH_HANS: EmailTemplate(
                subject="您现在是 {WorkspaceName} 的所有者",
                template_path="transfer_workspace_new_owner_notify_template_zh-CN.html",
                branded_template_path="without-brand/transfer_workspace_new_owner_notify_template_zh-CN.html",
            ),
        },
        EmailType.ACCOUNT_DELETION_SUCCESS: {
            EmailLanguage.EN_US: EmailTemplate(
                subject="Your Dify.AI Account Has Been Successfully Deleted",
                template_path="delete_account_success_template_en-US.html",
                branded_template_path="delete_account_success_template_en-US.html",
            ),
            EmailLanguage.ZH_HANS: EmailTemplate(
                subject="您的 Dify.AI 账户已成功删除",
                template_path="delete_account_success_template_zh-CN.html",
                branded_template_path="delete_account_success_template_zh-CN.html",
            ),
        },
        EmailType.ACCOUNT_DELETION_VERIFICATION: {
            EmailLanguage.EN_US: EmailTemplate(
                subject="Dify.AI Account Deletion and Verification",
                template_path="delete_account_code_email_template_en-US.html",
                branded_template_path="delete_account_code_email_template_en-US.html",
            ),
            EmailLanguage.ZH_HANS: EmailTemplate(
                subject="Dify.AI 账户删除和验证",
                template_path="delete_account_code_email_template_zh-CN.html",
                branded_template_path="delete_account_code_email_template_zh-CN.html",
            ),
        },
        EmailType.QUEUE_MONITOR_ALERT: {
            EmailLanguage.EN_US: EmailTemplate(
                subject="Alert: Dataset Queue pending tasks exceeded the limit",
                template_path="queue_monitor_alert_email_template_en-US.html",
                branded_template_path="queue_monitor_alert_email_template_en-US.html",
            ),
            EmailLanguage.ZH_HANS: EmailTemplate(
                subject="警报：数据集队列待处理任务超过限制",
                template_path="queue_monitor_alert_email_template_zh-CN.html",
                branded_template_path="queue_monitor_alert_email_template_zh-CN.html",
            ),
        },
        EmailType.DOCUMENT_CLEAN_NOTIFY: {
            EmailLanguage.EN_US: EmailTemplate(
                subject="Dify Knowledge base auto disable notification",
                template_path="clean_document_job_mail_template-US.html",
                branded_template_path="clean_document_job_mail_template-US.html",
            ),
            EmailLanguage.ZH_HANS: EmailTemplate(
                subject="Dify 知识库自动禁用通知",
                template_path="clean_document_job_mail_template_zh-CN.html",
                branded_template_path="clean_document_job_mail_template_zh-CN.html",
            ),
        },
    }

    return EmailI18nConfig(templates=templates)


# Singleton instance for application-wide use
def get_default_email_i18n_service() -> EmailI18nService:
    """Get configured email i18n service with default dependencies."""
    config = create_default_email_config()
    renderer = FlaskEmailRenderer()
    branding_service = FeatureBrandingService()
    sender = FlaskMailSender()

    return EmailI18nService(
        config=config,
        renderer=renderer,
        branding_service=branding_service,
        sender=sender,
    )


# Global instance
_email_i18n_service: Optional[EmailI18nService] = None


def get_email_i18n_service() -> EmailI18nService:
    """Get global email i18n service instance."""
    global _email_i18n_service
    if _email_i18n_service is None:
        _email_i18n_service = get_default_email_i18n_service()
    return _email_i18n_service