Merge remote-tracking branch 'origin/master'

src/config/constants.py

@@ -35,6 +35,7 @@ class EmailServiceType(str, Enum):
     OUTLOOK = "outlook"
     CUSTOM_DOMAIN = "custom_domain"
     TEMP_MAIL = "temp_mail"
+    DUCK_MAIL = "duck_mail"
 
 
 # ============================================================================
@@ -112,6 +113,13 @@ EMAIL_SERVICE_DEFAULTS = {
         "api_key_header": "X-API-Key",
         "timeout": 30,
         "max_retries": 3,
+    },
+    "duck_mail": {
+        "base_url": "",
+        "default_domain": "",
+        "password_length": 12,
+        "timeout": 30,
+        "max_retries": 3,
     }
 }
 
@@ -368,4 +376,4 @@ MICROSOFT_SCOPES = {
 }
 
 # Outlook 提供者默认优先级
-OUTLOOK_PROVIDER_PRIORITY = ["imap_new", "imap_old", "graph_api"]
+OUTLOOK_PROVIDER_PRIORITY = ["imap_new", "imap_old", "graph_api"]

src/core/upload/cpa_upload.py

@@ -6,6 +6,7 @@ import json
 import logging
 from typing import List, Dict, Any, Tuple, Optional
 from datetime import datetime
+from urllib.parse import quote
 
 from curl_cffi import requests as cffi_requests
 from curl_cffi import CurlMime
@@ -17,6 +18,77 @@ from ...config.settings import get_settings
 logger = logging.getLogger(__name__)
 
 
+def _normalize_cpa_auth_files_url(api_url: str) -> str:
+    """将用户填写的 CPA 地址规范化为 auth-files 接口地址。"""
+    normalized = (api_url or "").strip().rstrip("/")
+    lower_url = normalized.lower()
+
+    if not normalized:
+        return ""
+
+    if lower_url.endswith("/auth-files"):
+        return normalized
+
+    if lower_url.endswith("/v0/management") or lower_url.endswith("/management"):
+        return f"{normalized}/auth-files"
+
+    if lower_url.endswith("/v0"):
+        return f"{normalized}/management/auth-files"
+
+    return f"{normalized}/v0/management/auth-files"
+
+
+def _build_cpa_headers(api_token: str, content_type: Optional[str] = None) -> dict:
+    headers = {
+        "Authorization": f"Bearer {api_token}",
+    }
+    if content_type:
+        headers["Content-Type"] = content_type
+    return headers
+
+
+def _extract_cpa_error(response) -> str:
+    error_msg = f"上传失败: HTTP {response.status_code}"
+    try:
+        error_detail = response.json()
+        if isinstance(error_detail, dict):
+            error_msg = error_detail.get("message", error_msg)
+    except Exception:
+        error_msg = f"{error_msg} - {response.text[:200]}"
+    return error_msg
+
+
+def _post_cpa_auth_file_multipart(upload_url: str, filename: str, file_content: bytes, api_token: str):
+    mime = CurlMime()
+    mime.addpart(
+        name="file",
+        data=file_content,
+        filename=filename,
+        content_type="application/json",
+    )
+
+    return cffi_requests.post(
+        upload_url,
+        multipart=mime,
+        headers=_build_cpa_headers(api_token),
+        proxies=None,
+        timeout=30,
+        impersonate="chrome110",
+    )
+
+
+def _post_cpa_auth_file_raw_json(upload_url: str, filename: str, file_content: bytes, api_token: str):
+    raw_upload_url = f"{upload_url}?name={quote(filename)}"
+    return cffi_requests.post(
+        raw_upload_url,
+        data=file_content,
+        headers=_build_cpa_headers(api_token, content_type="application/json"),
+        proxies=None,
+        timeout=30,
+        impersonate="chrome110",
+    )
+
+
 def generate_token_json(account: Account) -> dict:
     """
     生成 CPA 格式的 Token JSON
@@ -73,45 +145,35 @@ def upload_to_cpa(
     if not effective_token:
         return False, "CPA API Token 未配置"
 
-    api_url = effective_url.rstrip("/")
-    upload_url = f"{api_url}/v0/management/auth-files"
+    upload_url = _normalize_cpa_auth_files_url(effective_url)
 
     filename = f"{token_data['email']}.json"
     file_content = json.dumps(token_data, ensure_ascii=False, indent=2).encode("utf-8")
 
-    headers = {
-        "Authorization": f"Bearer {effective_token}",
-    }
-
     try:
-        mime = CurlMime()
-        mime.addpart(
-            name="file",
-            data=file_content,
-            filename=filename,
-            content_type="application/json",
-        )
-
-        response = cffi_requests.post(
+        response = _post_cpa_auth_file_multipart(
             upload_url,
-            multipart=mime,
-            headers=headers,
-            proxies=None,
-            timeout=30,
-            impersonate="chrome110",
+            filename,
+            file_content,
+            effective_token,
         )
 
         if response.status_code in (200, 201):
             return True, "上传成功"
 
-        error_msg = f"上传失败: HTTP {response.status_code}"
-        try:
-            error_detail = response.json()
-            if isinstance(error_detail, dict):
-                error_msg = error_detail.get("message", error_msg)
-        except Exception:
-            error_msg = f"{error_msg} - {response.text[:200]}"
-        return False, error_msg
+        if response.status_code in (404, 405, 415):
+            logger.warning("CPA multipart 上传失败，尝试原始 JSON 回退: %s", response.status_code)
+            fallback_response = _post_cpa_auth_file_raw_json(
+                upload_url,
+                filename,
+                file_content,
+                effective_token,
+            )
+            if fallback_response.status_code in (200, 201):
+                return True, "上传成功"
+            response = fallback_response
+
+        return False, _extract_cpa_error(response)
 
     except Exception as e:
         logger.error(f"CPA 上传异常: {e}")
@@ -217,12 +279,11 @@ def test_cpa_connection(api_url: str, api_token: str, proxy: str = None) -> Tupl
     if not api_token:
         return False, "API Token 不能为空"
 
-    api_url = api_url.rstrip("/")
-    test_url = f"{api_url}/v0/management/auth-files"
-    headers = {"Authorization": f"Bearer {api_token}"}
+    test_url = _normalize_cpa_auth_files_url(api_url)
+    headers = _build_cpa_headers(api_token)
 
     try:
-        response = cffi_requests.options(
+        response = cffi_requests.get(
             test_url,
             headers=headers,
             proxies=None,
@@ -230,10 +291,16 @@ def test_cpa_connection(api_url: str, api_token: str, proxy: str = None) -> Tupl
             impersonate="chrome110",
         )
 
-        if response.status_code in (200, 204, 401, 403, 405):
-            if response.status_code == 401:
-                return False, "连接成功，但 API Token 无效"
+        if response.status_code == 200:
             return True, "CPA 连接测试成功"
+        if response.status_code == 401:
+            return False, "连接成功，但 API Token 无效"
+        if response.status_code == 403:
+            return False, "连接成功，但服务端未启用远程管理或当前 Token 无权限"
+        if response.status_code == 404:
+            return False, "未找到 CPA auth-files 接口，请检查 API URL 是否填写为根地址、/v0/management 或完整 auth-files 地址"
+        if response.status_code == 503:
+            return False, "连接成功，但服务端认证管理器不可用"
 
         return False, f"服务器返回异常状态码: {response.status_code}"
 

src/services/__init__.py

@@ -14,12 +14,14 @@ from .tempmail import TempmailService
 from .outlook import OutlookService
 from .moe_mail import MeoMailEmailService
 from .temp_mail import TempMailService
+from .duck_mail import DuckMailService
 
 # 注册服务
 EmailServiceFactory.register(EmailServiceType.TEMPMAIL, TempmailService)
 EmailServiceFactory.register(EmailServiceType.OUTLOOK, OutlookService)
 EmailServiceFactory.register(EmailServiceType.CUSTOM_DOMAIN, MeoMailEmailService)
 EmailServiceFactory.register(EmailServiceType.TEMP_MAIL, TempMailService)
+EmailServiceFactory.register(EmailServiceType.DUCK_MAIL, DuckMailService)
 
 # 导出 Outlook 模块的额外内容
 from .outlook.base import (
@@ -50,6 +52,7 @@ __all__ = [
     'OutlookService',
     'MeoMailEmailService',
     'TempMailService',
+    'DuckMailService',
     # Outlook 模块
     'ProviderType',
     'EmailMessage',
@@ -61,4 +64,4 @@ __all__ = [
     'IMAPOldProvider',
     'IMAPNewProvider',
     'GraphAPIProvider',
-]
+]

src/services/duck_mail.py

@@ -0,0 +1,366 @@
+"""
+DuckMail 邮箱服务实现
+兼容 DuckMail 的 accounts/token/messages 接口模型
+"""
+
+import logging
+import random
+import re
+import string
+import time
+from datetime import datetime, timezone
+from html import unescape
+from typing import Any, Dict, List, Optional
+
+from .base import BaseEmailService, EmailServiceError, EmailServiceType
+from ..config.constants import OTP_CODE_PATTERN
+from ..core.http_client import HTTPClient, RequestConfig
+
+
+logger = logging.getLogger(__name__)
+
+
+class DuckMailService(BaseEmailService):
+    """DuckMail 邮箱服务"""
+
+    def __init__(self, config: Dict[str, Any] = None, name: str = None):
+        super().__init__(EmailServiceType.DUCK_MAIL, name)
+
+        required_keys = ["base_url", "default_domain"]
+        missing_keys = [key for key in required_keys if not (config or {}).get(key)]
+        if missing_keys:
+            raise ValueError(f"缺少必需配置: {missing_keys}")
+
+        default_config = {
+            "api_key": "",
+            "password_length": 12,
+            "expires_in": None,
+            "timeout": 30,
+            "max_retries": 3,
+            "proxy_url": None,
+        }
+        self.config = {**default_config, **(config or {})}
+        self.config["base_url"] = str(self.config["base_url"]).rstrip("/")
+        self.config["default_domain"] = str(self.config["default_domain"]).strip().lstrip("@")
+
+        http_config = RequestConfig(
+            timeout=self.config["timeout"],
+            max_retries=self.config["max_retries"],
+        )
+        self.http_client = HTTPClient(
+            proxy_url=self.config.get("proxy_url"),
+            config=http_config,
+        )
+
+        self._accounts_by_id: Dict[str, Dict[str, Any]] = {}
+        self._accounts_by_email: Dict[str, Dict[str, Any]] = {}
+
+    def _build_headers(
+        self,
+        token: Optional[str] = None,
+        use_api_key: bool = False,
+        extra_headers: Optional[Dict[str, str]] = None,
+    ) -> Dict[str, str]:
+        headers = {
+            "Accept": "application/json",
+            "Content-Type": "application/json",
+        }
+
+        auth_token = token
+        if not auth_token and use_api_key and self.config.get("api_key"):
+            auth_token = self.config["api_key"]
+
+        if auth_token:
+            headers["Authorization"] = f"Bearer {auth_token}"
+
+        if extra_headers:
+            headers.update(extra_headers)
+
+        return headers
+
+    def _make_request(
+        self,
+        method: str,
+        path: str,
+        token: Optional[str] = None,
+        use_api_key: bool = False,
+        **kwargs,
+    ) -> Dict[str, Any]:
+        url = f"{self.config['base_url']}{path}"
+        kwargs["headers"] = self._build_headers(
+            token=token,
+            use_api_key=use_api_key,
+            extra_headers=kwargs.get("headers"),
+        )
+
+        try:
+            response = self.http_client.request(method, url, **kwargs)
+            if response.status_code >= 400:
+                error_message = f"API 请求失败: {response.status_code}"
+                try:
+                    error_payload = response.json()
+                    error_message = f"{error_message} - {error_payload}"
+                except Exception:
+                    error_message = f"{error_message} - {response.text[:200]}"
+                raise EmailServiceError(error_message)
+
+            try:
+                return response.json()
+            except Exception:
+                return {"raw_response": response.text}
+        except Exception as e:
+            self.update_status(False, e)
+            if isinstance(e, EmailServiceError):
+                raise
+            raise EmailServiceError(f"请求失败: {method} {path} - {e}")
+
+    def _generate_local_part(self) -> str:
+        first = random.choice(string.ascii_lowercase)
+        rest = "".join(random.choices(string.ascii_lowercase + string.digits, k=7))
+        return f"{first}{rest}"
+
+    def _generate_password(self) -> str:
+        length = max(6, int(self.config.get("password_length") or 12))
+        alphabet = string.ascii_letters + string.digits
+        return "".join(random.choices(alphabet, k=length))
+
+    def _cache_account(self, account_info: Dict[str, Any]) -> None:
+        account_id = str(account_info.get("account_id") or account_info.get("service_id") or "").strip()
+        email = str(account_info.get("email") or "").strip().lower()
+
+        if account_id:
+            self._accounts_by_id[account_id] = account_info
+        if email:
+            self._accounts_by_email[email] = account_info
+
+    def _get_account_info(self, email: Optional[str] = None, email_id: Optional[str] = None) -> Optional[Dict[str, Any]]:
+        if email_id:
+            cached = self._accounts_by_id.get(str(email_id))
+            if cached:
+                return cached
+
+        if email:
+            cached = self._accounts_by_email.get(str(email).strip().lower())
+            if cached:
+                return cached
+
+        return None
+
+    def _strip_html(self, html_content: Any) -> str:
+        if isinstance(html_content, list):
+            html_content = "\n".join(str(item) for item in html_content if item)
+        text = str(html_content or "")
+        return unescape(re.sub(r"<[^>]+>", " ", text))
+
+    def _parse_message_time(self, value: Optional[str]) -> Optional[float]:
+        if not value:
+            return None
+        try:
+            normalized = value.replace("Z", "+00:00")
+            return datetime.fromisoformat(normalized).astimezone(timezone.utc).timestamp()
+        except Exception:
+            return None
+
+    def _message_search_text(self, summary: Dict[str, Any], detail: Dict[str, Any]) -> str:
+        sender = summary.get("from") or detail.get("from") or {}
+        if isinstance(sender, dict):
+            sender_text = " ".join(
+                str(sender.get(key) or "") for key in ("name", "address")
+            ).strip()
+        else:
+            sender_text = str(sender)
+
+        subject = str(summary.get("subject") or detail.get("subject") or "")
+        text_body = str(detail.get("text") or "")
+        html_body = self._strip_html(detail.get("html"))
+        return "\n".join(part for part in [sender_text, subject, text_body, html_body] if part).strip()
+
+    def create_email(self, config: Dict[str, Any] = None) -> Dict[str, Any]:
+        request_config = config or {}
+        local_part = str(request_config.get("name") or self._generate_local_part()).strip()
+        domain = str(request_config.get("default_domain") or request_config.get("domain") or self.config["default_domain"]).strip().lstrip("@")
+        address = f"{local_part}@{domain}"
+        password = self._generate_password()
+
+        payload: Dict[str, Any] = {
+            "address": address,
+            "password": password,
+        }
+
+        expires_in = request_config.get("expiresIn", request_config.get("expires_in", self.config.get("expires_in")))
+        if expires_in is not None:
+            payload["expiresIn"] = expires_in
+
+        account_response = self._make_request(
+            "POST",
+            "/accounts",
+            json=payload,
+            use_api_key=bool(self.config.get("api_key")),
+        )
+        token_response = self._make_request(
+            "POST",
+            "/token",
+            json={
+                "address": account_response.get("address", address),
+                "password": password,
+            },
+        )
+
+        account_id = str(account_response.get("id") or token_response.get("id") or "").strip()
+        resolved_address = str(account_response.get("address") or address).strip()
+        token = str(token_response.get("token") or "").strip()
+
+        if not account_id or not resolved_address or not token:
+            raise EmailServiceError("DuckMail 返回数据不完整")
+
+        email_info = {
+            "email": resolved_address,
+            "service_id": account_id,
+            "id": account_id,
+            "account_id": account_id,
+            "token": token,
+            "password": password,
+            "created_at": time.time(),
+            "raw_account": account_response,
+        }
+
+        self._cache_account(email_info)
+        self.update_status(True)
+        return email_info
+
+    def get_verification_code(
+        self,
+        email: str,
+        email_id: str = None,
+        timeout: int = 120,
+        pattern: str = OTP_CODE_PATTERN,
+        otp_sent_at: Optional[float] = None,
+    ) -> Optional[str]:
+        account_info = self._get_account_info(email=email, email_id=email_id)
+        if not account_info:
+            logger.warning(f"DuckMail 未找到邮箱缓存: {email}, {email_id}")
+            return None
+
+        token = account_info.get("token")
+        if not token:
+            logger.warning(f"DuckMail 邮箱缺少访问 token: {email}")
+            return None
+
+        start_time = time.time()
+        seen_message_ids = set()
+
+        while time.time() - start_time < timeout:
+            try:
+                response = self._make_request(
+                    "GET",
+                    "/messages",
+                    token=token,
+                    params={"page": 1},
+                )
+                messages = response.get("hydra:member", [])
+
+                for message in messages:
+                    message_id = str(message.get("id") or "").strip()
+                    if not message_id or message_id in seen_message_ids:
+                        continue
+
+                    created_at = self._parse_message_time(message.get("createdAt"))
+                    if otp_sent_at and created_at and created_at + 1 < otp_sent_at:
+                        continue
+
+                    seen_message_ids.add(message_id)
+                    detail = self._make_request(
+                        "GET",
+                        f"/messages/{message_id}",
+                        token=token,
+                    )
+
+                    content = self._message_search_text(message, detail)
+                    if "openai" not in content.lower():
+                        continue
+
+                    match = re.search(pattern, content)
+                    if match:
+                        self.update_status(True)
+                        return match.group(1)
+            except Exception as e:
+                logger.debug(f"DuckMail 轮询验证码失败: {e}")
+
+            time.sleep(3)
+
+        return None
+
+    def list_emails(self, **kwargs) -> List[Dict[str, Any]]:
+        return list(self._accounts_by_email.values())
+
+    def delete_email(self, email_id: str) -> bool:
+        account_info = self._get_account_info(email_id=email_id) or self._get_account_info(email=email_id)
+        if not account_info:
+            return False
+
+        token = account_info.get("token")
+        account_id = account_info.get("account_id") or account_info.get("service_id")
+        if not token or not account_id:
+            return False
+
+        try:
+            self._make_request(
+                "DELETE",
+                f"/accounts/{account_id}",
+                token=token,
+            )
+            self._accounts_by_id.pop(str(account_id), None)
+            self._accounts_by_email.pop(str(account_info.get("email") or "").lower(), None)
+            self.update_status(True)
+            return True
+        except Exception as e:
+            logger.warning(f"DuckMail 删除邮箱失败: {e}")
+            self.update_status(False, e)
+            return False
+
+    def check_health(self) -> bool:
+        try:
+            self._make_request(
+                "GET",
+                "/domains",
+                params={"page": 1},
+                use_api_key=bool(self.config.get("api_key")),
+            )
+            self.update_status(True)
+            return True
+        except Exception as e:
+            logger.warning(f"DuckMail 健康检查失败: {e}")
+            self.update_status(False, e)
+            return False
+
+    def get_email_messages(self, email_id: str, **kwargs) -> List[Dict[str, Any]]:
+        account_info = self._get_account_info(email_id=email_id) or self._get_account_info(email=email_id)
+        if not account_info or not account_info.get("token"):
+            return []
+        response = self._make_request(
+            "GET",
+            "/messages",
+            token=account_info["token"],
+            params={"page": kwargs.get("page", 1)},
+        )
+        return response.get("hydra:member", [])
+
+    def get_message_detail(self, email_id: str, message_id: str) -> Optional[Dict[str, Any]]:
+        account_info = self._get_account_info(email_id=email_id) or self._get_account_info(email=email_id)
+        if not account_info or not account_info.get("token"):
+            return None
+        return self._make_request(
+            "GET",
+            f"/messages/{message_id}",
+            token=account_info["token"],
+        )
+
+    def get_service_info(self) -> Dict[str, Any]:
+        return {
+            "service_type": self.service_type.value,
+            "name": self.name,
+            "base_url": self.config["base_url"],
+            "default_domain": self.config["default_domain"],
+            "cached_accounts": len(self._accounts_by_email),
+            "status": self.status.value,
+        }

src/web/app.py

@@ -36,6 +36,16 @@ STATIC_DIR = _RESOURCE_ROOT / "static"
 TEMPLATES_DIR = _RESOURCE_ROOT / "templates"
 
 
+def _build_static_asset_version(static_dir: Path) -> str:
+    """基于静态文件最后修改时间生成版本号，避免部署后浏览器继续使用旧缓存。"""
+    latest_mtime = 0
+    if static_dir.exists():
+        for path in static_dir.rglob("*"):
+            if path.is_file():
+                latest_mtime = max(latest_mtime, int(path.stat().st_mtime))
+    return str(latest_mtime or 1)
+
+
 def create_app() -> FastAPI:
     """创建 FastAPI 应用实例"""
     settings = get_settings()
@@ -80,6 +90,7 @@ def create_app() -> FastAPI:
 
     # 模板引擎
     templates = Jinja2Templates(directory=str(TEMPLATES_DIR))
+    templates.env.globals["static_version"] = _build_static_asset_version(STATIC_DIR)
 
     def _auth_token(password: str) -> str:
         secret = get_settings().webui_secret_key.get_secret_value().encode("utf-8")

src/web/routes/email.py

@@ -144,6 +144,7 @@ async def get_email_services_stats():
             'outlook_count': 0,
             'custom_count': 0,
             'temp_mail_count': 0,
+            'duck_mail_count': 0,
             'tempmail_available': True,  # 临时邮箱始终可用
             'enabled_count': enabled_count
         }
@@ -155,6 +156,8 @@ async def get_email_services_stats():
                 stats['custom_count'] = count
             elif service_type == 'temp_mail':
                 stats['temp_mail_count'] = count
+            elif service_type == 'duck_mail':
+                stats['duck_mail_count'] = count
 
         return stats
 
@@ -204,6 +207,17 @@ async def get_service_types():
                     {"name": "domain", "label": "邮箱域名", "required": True, "placeholder": "example.com"},
                     {"name": "enable_prefix", "label": "启用前缀", "required": False, "default": True},
                 ]
+            },
+            {
+                "value": "duck_mail",
+                "label": "DuckMail",
+                "description": "DuckMail 接口邮箱服务，支持 API Key 私有域名访问",
+                "config_fields": [
+                    {"name": "base_url", "label": "API 地址", "required": True, "placeholder": "https://api.duckmail.sbs"},
+                    {"name": "default_domain", "label": "默认域名", "required": True, "placeholder": "duckmail.sbs"},
+                    {"name": "api_key", "label": "API Key", "required": False, "secret": True},
+                    {"name": "password_length", "label": "随机密码长度", "required": False, "default": 12},
+                ]
             }
         ]
     }

src/web/routes/registration.py

@@ -211,6 +211,9 @@ def _normalize_email_service_config(
     elif service_type == EmailServiceType.TEMP_MAIL:
         if 'default_domain' in normalized and 'domain' not in normalized:
             normalized['domain'] = normalized.pop('default_domain')
+    elif service_type == EmailServiceType.DUCK_MAIL:
+        if 'domain' in normalized and 'default_domain' not in normalized:
+            normalized['default_domain'] = normalized.pop('domain')
 
     if proxy_url and 'proxy_url' not in normalized:
         normalized['proxy_url'] = proxy_url
@@ -341,6 +344,20 @@ def _run_sync_registration_task(task_uuid: str, email_service_type: str, proxy:
                         logger.info(f"使用数据库 Outlook 账户: {selected_service.name}")
                     else:
                         raise ValueError("所有 Outlook 账户都已注册过 OpenAI 账号，请添加新的 Outlook 账户")
+                elif service_type == EmailServiceType.DUCK_MAIL:
+                    from ...database.models import EmailService as EmailServiceModel
+
+                    db_service = db.query(EmailServiceModel).filter(
+                        EmailServiceModel.service_type == "duck_mail",
+                        EmailServiceModel.enabled == True
+                    ).order_by(EmailServiceModel.priority.asc()).first()
+
+                    if db_service and db_service.config:
+                        config = _normalize_email_service_config(service_type, db_service.config, actual_proxy_url)
+                        crud.update_registration_task(db, task_uuid, email_service_id=db_service.id)
+                        logger.info(f"使用数据库 DuckMail 服务: {db_service.name}")
+                    else:
+                        raise ValueError("没有可用的 DuckMail 邮箱服务，请先在邮箱服务页面添加服务")
                 else:
                     config = email_service_config or {}
 
@@ -1069,6 +1086,11 @@ async def get_available_email_services():
             "available": False,
             "count": 0,
             "services": []
+        },
+        "duck_mail": {
+            "available": False,
+            "count": 0,
+            "services": []
         }
     }
 
@@ -1142,6 +1164,24 @@ async def get_available_email_services():
         result["temp_mail"]["count"] = len(temp_mail_services)
         result["temp_mail"]["available"] = len(temp_mail_services) > 0
 
+        duck_mail_services = db.query(EmailServiceModel).filter(
+            EmailServiceModel.service_type == "duck_mail",
+            EmailServiceModel.enabled == True
+        ).order_by(EmailServiceModel.priority.asc()).all()
+
+        for service in duck_mail_services:
+            config = service.config or {}
+            result["duck_mail"]["services"].append({
+                "id": service.id,
+                "name": service.name,
+                "type": "duck_mail",
+                "default_domain": config.get("default_domain"),
+                "priority": service.priority
+            })
+
+        result["duck_mail"]["count"] = len(duck_mail_services)
+        result["duck_mail"]["available"] = len(duck_mail_services) > 0
+
     return result