refactor: migrate direct mail.send usages to EmailI18nService

- Updated mail_enterprise_task.py to use EmailI18nService.send_raw_email()
- Updated queue_monitor_task.py to use EmailI18nService with new QUEUE_MONITOR_ALERT type
- Updated mail_clean_document_notify_task.py to use EmailI18nService with new DOCUMENT_CLEAN_NOTIFY type
- Added send_raw_email() method for backward compatibility with pre-rendered HTML emails
- Added new email types: ENTERPRISE_CUSTOM, QUEUE_MONITOR_ALERT, DOCUMENT_CLEAN_NOTIFY
- Added email template configurations for the new email types
- Removed unused imports (render_template, mail) from refactored files

This change centralizes email handling through EmailI18nService while maintaining
backward compatibility for legacy email sending patterns.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

api/libs/email_i18n.py

@@ -30,6 +30,9 @@ class EmailType(Enum):
     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):
@@ -215,6 +218,30 @@ class EmailI18nService:
             },
         )
 
+    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,
@@ -377,6 +404,30 @@ def create_default_email_config() -> EmailI18nConfig:
                 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)

api/schedule/mail_clean_document_notify_task.py

@@ -3,12 +3,12 @@ import time
 from collections import defaultdict
 
 import click
-from flask import render_template  # type: ignore
 
 import app
 from configs import dify_config
 from extensions.ext_database import db
 from extensions.ext_mail import mail
+from libs.email_i18n import EmailType, get_email_i18n_service
 from models.account import Account, Tenant, TenantAccountJoin
 from models.dataset import Dataset, DatasetAutoDisableLog
 from services.feature_service import FeatureService
@@ -72,14 +72,16 @@ def mail_clean_document_notify_task():
                         document_count = len(document_ids)
                         knowledge_details.append(rf"Knowledge base {dataset.name}: {document_count} documents")
                 if knowledge_details:
-                    html_content = render_template(
+                    email_service = get_email_i18n_service()
-                        "clean_document_job_mail_template-US.html",
+                    email_service.send_email(
-                        userName=account.email,
+                        email_type=EmailType.DOCUMENT_CLEAN_NOTIFY,
-                        knowledge_details=knowledge_details,
+                        language_code="en-US",
-                        url=url,
+                        to=account.email,
-                    )
+                        template_context={
-                    mail.send(
+                            "userName": account.email,
-                        to=account.email, subject="Dify Knowledge base auto disable notification", html=html_content
+                            "knowledge_details": knowledge_details,
+                            "url": url,
+                        },
                     )
 
             # update notified to True

api/schedule/queue_monitor_task.py

@@ -3,13 +3,12 @@ from datetime import datetime
 from urllib.parse import urlparse
 
 import click
-from flask import render_template
 from redis import Redis
 
 import app
 from configs import dify_config
 from extensions.ext_database import db
-from extensions.ext_mail import mail
+from libs.email_i18n import EmailType, get_email_i18n_service
 
 # Create a dedicated Redis connection (using the same configuration as Celery)
 celery_broker_url = dify_config.CELERY_BROKER_URL
@@ -39,18 +38,20 @@ def queue_monitor_task():
             alter_emails = dify_config.QUEUE_MONITOR_ALERT_EMAILS
             if alter_emails:
                 to_list = alter_emails.split(",")
+                email_service = get_email_i18n_service()
                 for to in to_list:
                     try:
                         current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-                        html_content = render_template(
+                        email_service.send_email(
-                            "queue_monitor_alert_email_template_en-US.html",
+                            email_type=EmailType.QUEUE_MONITOR_ALERT,
-                            queue_name=queue_name,
+                            language_code="en-US",
-                            queue_length=queue_length,
+                            to=to,
-                            threshold=threshold,
+                            template_context={
-                            alert_time=current_time,
+                                "queue_name": queue_name,
-                        )
+                                "queue_length": queue_length,
-                        mail.send(
+                                "threshold": threshold,
-                            to=to, subject="Alert: Dataset Queue pending tasks exceeded the limit", html=html_content
+                                "alert_time": current_time,
+                            },
                         )
                     except Exception as e:
                         logging.exception(click.style("Exception occurred during sending email", fg="red"))

api/tasks/mail_enterprise_task.py

@@ -1,15 +1,17 @@
 import logging
 import time
+from collections.abc import Mapping
 
 import click
 from celery import shared_task  # type: ignore
 from flask import render_template_string
 
 from extensions.ext_mail import mail
+from libs.email_i18n import get_email_i18n_service
 
 
 @shared_task(queue="mail")
-def send_enterprise_email_task(to, subject, body, substitutions):
+def send_enterprise_email_task(to: list[str], subject: str, body: str, substitutions: Mapping[str, str]):
     if not mail.is_inited():
         return
 
@@ -19,11 +21,8 @@ def send_enterprise_email_task(to, subject, body, substitutions):
     try:
         html_content = render_template_string(body, **substitutions)
 
-        if isinstance(to, list):
+        email_service = get_email_i18n_service()
-            for t in to:
+        email_service.send_raw_email(to=to, subject=subject, html_content=html_content)
-                mail.send(to=t, subject=subject, html=html_content)
-        else:
-            mail.send(to=to, subject=subject, html=html_content)
 
         end_at = time.perf_counter()
         logging.info(