""" DuckMailTempMailProvider ~~~~~~~~~~~~~~~~~~~~~~~~~ 对接 DuckMail (https://github.com/MoonWeSif/DuckMail) 的 TempMailProviderBase 实现。 认证模型 -------- - 创建邮箱：POST /accounts（不需要鉴权，但需要有效的域名） - 获取 Token：POST /token（使用邮箱地址和密码） - 所有操作接口：Authorization: Bearer 配置项（settings 表 key） ------------------------- - duckmail_base_url : API 基础地址（默认 https://api.duckmail.sbs） - duckmail_api_key : 可选，私有域名访问 API Key（格式 dk_xxx） - duckmail_default_domain : 默认域名（留空则用 duckmail.sbs） - duckmail_expires_in : 账户过期时间（秒），0=-1=永不过期，留空=24h 字段映射 -------- DuckMail 使用以下字段名，本模块统一转换为平台标准字段名： - from.address -> from_address - subject -> subject - text -> content - html[0] -> html_content - createdAt -> timestamp (ISO 8601 -> int unix) - id -> message_id（加 dk_ 前缀避免冲突） """ from __future__ import annotations import json import logging import re import secrets import string from datetime import datetime, timezone from typing import Any import requests from outlook_web.repositories import settings as settings_repo from outlook_web.services.temp_mail_provider_base import TempMailProviderBase, register_provider logger = logging.getLogger(__name__) _DK_REQUEST_TIMEOUT = 30 # seconds DEFAULT_BASE_URL = "https://api.duckmail.sbs" DEFAULT_DOMAIN = "duckmail.sbs" DEFAULT_PREFIX_RULES = { "min_length": 3, "max_length": 32, "pattern": r"^[a-z0-9][a-z0-9._-]*$", } # --------------------------------------------------------------------------- # 异常 # --------------------------------------------------------------------------- class DuckMailProviderError(Exception): def __init__(self, code: str, message: str, *, data: dict[str, Any] | None = None): super().__init__(message) self.code = code self.message = message self.data = data or {} # --------------------------------------------------------------------------- # 工具函数 # --------------------------------------------------------------------------- def _map_dk_http_error(status_code: int, text: str = "") -> str: if status_code in (401, 403): return "UNAUTHORIZED" if status_code == 404: return "NOT_FOUND" if status_code == 409: return "CONFLICT" if status_code == 422: return "VALIDATION_FAILED" if status_code == 429: return "UPSTREAM_RATE_LIMITED" if status_code >= 500: return "UPSTREAM_SERVER_ERROR" return "UPSTREAM_BAD_PAYLOAD" def _iso_to_timestamp(iso_str: str) -> int: """将 ISO 8601 字符串转换为 Unix timestamp（整数）。""" try: clean = iso_str.replace("Z", "+00:00") if "." in clean: clean = clean[: clean.index(".")] + clean[clean.index("+") if "+" in clean else len(clean):] return int(datetime.fromisoformat(clean).timestamp()) except (ValueError, AttributeError, IndexError): return 0 def _normalize_domain_entries(raw_domains: Any, default_domain: str) -> list[dict[str, Any]]: domains: list[dict[str, Any]] = [] seen: set[str] = set() values: list[Any] = raw_domains if isinstance(raw_domains, list) else [] for item in values: if isinstance(item, dict): name = str(item.get("name") or "").strip() enabled = bool(item.get("enabled", True)) else: name = str(item or "").strip() enabled = True if not name or name in seen: continue seen.add(name) domains.append({ "name": name, "enabled": enabled, "is_default": bool(default_domain and name == default_domain), }) if default_domain and default_domain not in seen: domains.append({"name": default_domain, "enabled": True, "is_default": True}) return domains def _get_dk_base_url() -> str: """读取 DuckMail API 基础地址。""" url = settings_repo.get_setting("duckmail_base_url", "").strip() return url or DEFAULT_BASE_URL def _get_dk_api_key() -> str: """读取 DuckMail API Key（可选，用于私有域名）。""" return settings_repo.get_setting("duckmail_api_key", "").strip() def _get_dk_default_domain() -> str: """读取默认域名。""" return settings_repo.get_setting("duckmail_default_domain", "").strip() or DEFAULT_DOMAIN def _get_dk_expires_in() -> str: """读取账户过期时间（秒）。""" return settings_repo.get_setting("duckmail_expires_in", "").strip() # --------------------------------------------------------------------------- # Provider 实现 # --------------------------------------------------------------------------- @register_provider class DuckMailTempMailProvider(TempMailProviderBase): """ 对接 DuckMail 临时邮箱服务的 Provider 实现。 配置读取（来自 settings 表）： - duckmail_base_url : API 基础地址（默认 https://api.duckmail.sbs） - duckmail_api_key : 可选，私有域名访问 API Key（格式 dk_xxx） - duckmail_default_domain: 默认域名（默认 duckmail.sbs） - duckmail_expires_in : 账户过期时间（秒），0/-1=永不过期，留空=24h """ provider_name = "duckmail" provider_label = "DuckMail" provider_version = "1.0.0" provider_author = "OutlookMail Plus" config_schema = { "title": "DuckMail 配置", "fields": [ { "key": "base_url", "label": "API 基础地址", "type": "text", "placeholder": "https://api.duckmail.sbs", "default": DEFAULT_BASE_URL, "help": "DuckMail API 的基础 URL", }, { "key": "api_key", "label": "API Key（可选）", "type": "text", "placeholder": "dk_xxx", "default": "", "help": "用于访问私有域名，格式 dk_xxx", }, { "key": "default_domain", "label": "默认域名", "type": "text", "placeholder": "duckmail.sbs", "default": DEFAULT_DOMAIN, "help": "创建邮箱时使用的默认域名", }, { "key": "expires_in", "label": "账户过期时间（秒）", "type": "number", "placeholder": "留空=24小时", "default": "", "help": "0 或 -1 表示永不过期，正数表示秒数，留空表示 24 小时自动清理", }, ], } def __init__(self, *, provider_name: str | None = None): self.provider_name = provider_name or "duckmail" # ------------------------------------------------------------------ # 内部辅助 # ------------------------------------------------------------------ def _base_url(self) -> str: return _get_dk_base_url() def _user_headers(self, token: str) -> dict[str, str]: """构建用户认证请求头（Bearer Token）。""" return {"Content-Type": "application/json", "Authorization": f"Bearer {token}"} def _admin_headers(self, api_key: str | None = None) -> dict[str, str]: """构建管理/创建请求头。优先使用传入的 api_key，否则读取配置。""" headers = {"Content-Type": "application/json"} key = api_key or _get_dk_api_key() if key: headers["Authorization"] = f"Bearer {key}" return headers def _coerce_email(self, mailbox: dict[str, Any] | str) -> str: if isinstance(mailbox, dict): return str(mailbox.get("email") or "").strip() return str(mailbox or "").strip() def _get_token(self, mailbox: dict[str, Any]) -> str: """从 mailbox.meta 中取出 auth_token。""" meta = mailbox.get("meta") or {} if isinstance(meta, str): try: meta = json.loads(meta) except Exception: meta = {} return str(meta.get("auth_token") or "").strip() def _get_account_id(self, mailbox: dict[str, Any]) -> str: """从 mailbox.meta 中取出 provider_mailbox_id（即 DuckMail 账户 ID）。""" meta = mailbox.get("meta") or {} if isinstance(meta, str): try: meta = json.loads(meta) except Exception: meta = {} return str(meta.get("provider_mailbox_id") or meta.get("accountId") or "").strip() def _build_meta(self, *, token: str = "", account_id: str = "", address: str = "", password: str = "") -> dict[str, Any]: return { "provider_name": self.provider_name, "provider_mailbox_id": account_id, "auth_token": token, "email": address, "password": password, "provider_cursor": "", "provider_labels": [], "provider_capabilities": { "delete_mailbox": True, "delete_message": True, "clear_messages": False, }, "provider_debug": {"bridge": "duckmail"}, } def _raise_http_error(self, resp: requests.Response, *, operation: str) -> None: code = _map_dk_http_error(resp.status_code, resp.text) raise DuckMailProviderError( code, f"DuckMail {operation} 失败 HTTP {resp.status_code}", data={"status_code": resp.status_code, "body": resp.text[:500]}, ) def _parse_expires_in(self) -> int | None: """解析 expires_in 配置。""" raw = _get_dk_expires_in() if not raw: return None # 默认 24h try: val = int(raw) return val except (ValueError, TypeError): return None def _do_request(self, base_url: str, method: str, path: str, *, headers: dict[str, str] | None = None, params: dict | None = None) -> requests.Response | None: """发送 HTTP 请求的封装，统一处理超时和网络异常。""" url = f"{base_url}{path}" try: resp = requests.request(method, url, headers=headers, params=params, timeout=_DK_REQUEST_TIMEOUT) return resp except requests.Timeout: logger.warning("[duckmail] %s %s timeout", method, url) return None except requests.RequestException as exc: logger.warning("[duckmail] %s %s network error: %s", method, url, exc) return None def _refresh_token(self, base_url: str, mailbox: dict[str, Any]) -> str | None: """ 尝试用 mailbox.meta 中的地址和密码重新获取 token。 如果成功，更新 mailbox.meta 并返回新 token；失败返回 None。 """ meta = mailbox.get("meta") or {} if isinstance(meta, str): try: meta = json.loads(meta) except Exception: meta = {} email_addr = meta.get("email", "") password = meta.get("password", "") if not email_addr or not password: logger.warning("[duckmail] refresh_token: no email/password in meta for %s", email_addr) return None try: resp = requests.post( f"{base_url}/token", headers={"Content-Type": "application/json"}, json={"address": email_addr, "password": password}, timeout=_DK_REQUEST_TIMEOUT, ) except Exception as exc: logger.warning("[duckmail] refresh_token request failed: %s", exc) return None if not resp.ok: logger.warning("[duckmail] refresh_token returned %d", resp.status_code) return None try: token_data = resp.json() new_token = str(token_data.get("token") or "").strip() except Exception: return None if not new_token: return None # 更新 meta 中的 token meta["auth_token"] = new_token mailbox["meta"] = meta logger.info("[duckmail] token refreshed for %s", email_addr) return new_token # ------------------------------------------------------------------ # TempMailProviderBase 接口实现 # ------------------------------------------------------------------ def get_options(self) -> dict[str, Any]: default_domain = _get_dk_default_domain() # 从 API 获取可用域名列表 domains_list: list[dict[str, Any]] = [{"name": default_domain, "enabled": True, "is_default": True}] base_url = self._base_url() try: resp = requests.get( f"{base_url}/domains", headers=self._admin_headers(), timeout=_DK_REQUEST_TIMEOUT, ) if resp.ok: data = resp.json() hydra_members = data.get("hydra:member", []) if isinstance(hydra_members, list): for entry in hydra_members: domain_name = str(entry.get("domain") or "").strip() if domain_name and domain_name not in [d["name"] for d in domains_list]: is_verified = entry.get("isVerified", True) owner_id = entry.get("ownerId") # 只显示已验证的系统域名（ownerId=null）或已验证的私有域名 if is_verified: domains_list.append({ "name": domain_name, "enabled": True, "is_default": bool(domain_name == default_domain), }) except Exception as exc: logger.warning("[duckmail] get_options domain fetch failed: %s", exc) return { "domain_strategy": "auto_or_manual", "default_mode": "auto", "domains": domains_list, "prefix_rules": { "min_length": DEFAULT_PREFIX_RULES["min_length"], "max_length": DEFAULT_PREFIX_RULES["max_length"], "pattern": DEFAULT_PREFIX_RULES["pattern"], }, "provider": self.provider_name, "provider_name": self.provider_name, "provider_label": "duckmail", "api_base_url": base_url, } def create_mailbox(self, *, prefix: str | None = None, domain: str | None = None) -> dict[str, Any]: """ 调用 POST /accounts 创建邮箱，然后 POST /token 获取认证 token。 返回格式： - 成功：{"success": True, "email": "...", "meta": {...}} - 失败：{"success": False, "error": "...", "error_code": "..."} """ base_url = self._base_url() if not base_url: return { "success": False, "error": "DuckMail base_url 未配置", "error_code": "TEMP_MAIL_PROVIDER_NOT_CONFIGURED", } # 确定目标域名 target_domain = (domain or "").strip() if not target_domain: target_domain = _get_dk_default_domain() # 生成用户名 effective_name = (prefix or "").strip() if not effective_name: alphabet = string.ascii_lowercase + string.digits effective_name = "".join(secrets.choice(alphabet) for _ in range(8)) # 构造邮箱地址 email_addr = f"{effective_name}@{target_domain}" # 设置密码 password = "".join(secrets.choice(string.ascii_letters + string.digits + "!@#$") for _ in range(10)) # 构建创建请求体 create_payload: dict[str, Any] = { "address": email_addr, "password": password, } expires_in = self._parse_expires_in() if expires_in is not None: create_payload["expiresIn"] = expires_in # 第 1 步：创建账户 try: resp = requests.post( f"{base_url}/accounts", headers=self._admin_headers(), json=create_payload, timeout=_DK_REQUEST_TIMEOUT, ) except requests.Timeout: return { "success": False, "error": "DuckMail 请求超时", "error_code": "UPSTREAM_TIMEOUT", } except requests.RequestException as exc: return { "success": False, "error": f"DuckMail 网络错误: {exc}", "error_code": "UPSTREAM_SERVER_ERROR", } if not resp.ok: code = _map_dk_http_error(resp.status_code, resp.text) return { "success": False, "error": f"DuckMail 创建邮箱失败 HTTP {resp.status_code}: {resp.text[:200]}", "error_code": code, } try: account_data = resp.json() except Exception: return { "success": False, "error": "DuckMail 返回非 JSON 响应", "error_code": "UPSTREAM_BAD_PAYLOAD", } account_id = str(account_data.get("id") or "").strip() if not account_id: return { "success": False, "error": "DuckMail 未返回账户 ID", "error_code": "UPSTREAM_BAD_PAYLOAD", } # 第 2 步：获取认证 token try: token_resp = requests.post( f"{base_url}/token", headers={"Content-Type": "application/json"}, json={"address": email_addr, "password": password}, timeout=_DK_REQUEST_TIMEOUT, ) except requests.Timeout: # token 获取失败但不影响邮箱创建，返回不带 token 的结果 logger.warning("[duckmail] token request failed after account created: %s", email_addr) return { "success": True, "email": email_addr, "meta": self._build_meta(account_id=account_id, address=email_addr, password=password), } except requests.RequestException as exc: logger.warning("[duckmail] token request network error: %s", exc) return { "success": True, "email": email_addr, "meta": self._build_meta(account_id=account_id, address=email_addr, password=password), } auth_token = "" if token_resp.ok: try: token_data = token_resp.json() auth_token = str(token_data.get("token") or "").strip() except Exception: pass return { "success": True, "email": email_addr, "meta": self._build_meta(token=auth_token, account_id=account_id, address=email_addr, password=password), } def delete_mailbox(self, mailbox: dict[str, Any]) -> bool: """调用 DELETE /accounts/{id} 删除邮箱。""" account_id = self._get_account_id(mailbox) token = self._get_token(mailbox) if not account_id: logger.warning("[duckmail] delete_mailbox: no account_id for %s", self._coerce_email(mailbox)) return False base_url = self._base_url() try: resp = requests.delete( f"{base_url}/accounts/{account_id}", headers=self._user_headers(token) if token else self._admin_headers(), timeout=_DK_REQUEST_TIMEOUT, ) return resp.ok or resp.status_code == 204 except requests.RequestException as exc: logger.warning("[duckmail] delete_mailbox failed id=%s err=%s", account_id, exc) return False def list_messages(self, mailbox: dict[str, Any] | str) -> list[dict[str, Any]]: """ 调用 GET /messages 获取邮件列表。 返回列表中每项的字段符合平台标准： - id, message_id, from_address, subject, content, html_content, has_html, timestamp 如果 token 失效，尝试用邮箱地址+密码重新获取 token（token refresh）。 """ email_addr = self._coerce_email(mailbox) token = self._get_token(mailbox) if isinstance(mailbox, dict) else "" if not token: raise DuckMailProviderError( "UNAUTHORIZED", f"邮箱 {email_addr} 缺少 auth_token，无法读取邮件", data={"email": email_addr}, ) base_url = self._base_url() # 尝试读取邮件 resp = self._do_request(base_url, "GET", "/messages", headers=self._user_headers(token), params={"page": 1}) # token 失效（401）时尝试刷新 if resp and not resp.ok and resp.status_code == 401: refreshed_token = self._refresh_token(base_url, mailbox) if refreshed_token: resp = self._do_request(base_url, "GET", "/messages", headers=self._user_headers(refreshed_token), params={"page": 1}) if resp is None or not resp.ok: status = resp.status_code if resp else 0 body = resp.text[:200] if resp else "" code = _map_dk_http_error(status, body) raise DuckMailProviderError( code, f"DuckMail 读取邮件失败 HTTP {status}", data={"email": email_addr, "status_code": status, "body": body}, ) try: data = resp.json() except Exception: raise DuckMailProviderError( "UPSTREAM_BAD_PAYLOAD", "DuckMail 邮件列表返回非 JSON 响应", data={"email": email_addr}, ) dk_msgs = data.get("hydra:member", []) if not isinstance(dk_msgs, list): raise DuckMailProviderError( "UPSTREAM_BAD_PAYLOAD", "DuckMail 邮件列表字段格式错误", data={"email": email_addr}, ) results: list[dict[str, Any]] = [] for dk_msg in dk_msgs: try: results.append(self._normalize_dk_message(dk_msg)) except Exception as exc: logger.warning( "[duckmail] failed to parse dk_msg id=%s err=%s", dk_msg.get("id"), exc, ) return results def _normalize_dk_message(self, dk_msg: dict[str, Any]) -> dict[str, Any]: """将 DuckMail 原始邮件结构转换为平台标准结构。""" dk_id = dk_msg.get("id") message_id = f"dk_{dk_id}" if dk_id is not None else "" created_at_str = str(dk_msg.get("createdAt") or "") timestamp = _iso_to_timestamp(created_at_str) if created_at_str else 0 # from 字段 from_info = dk_msg.get("from") or {} if isinstance(from_info, dict): from_address = str(from_info.get("address") or "").strip() else: from_address = str(from_info).strip() # subject subject = str(dk_msg.get("subject") or "").strip() # 列表视图不含 body，content 和 html_content 留空 # 详情需要通过 get_message_detail 获取 return { "id": message_id, "message_id": message_id, "from_address": from_address, "subject": subject, "content": "", "html_content": "", "has_html": False, "timestamp": timestamp, "seen": bool(dk_msg.get("seen", False)), "has_attachments": bool(dk_msg.get("hasAttachments", False)), "size": int(dk_msg.get("size") or 0), } def get_message_detail(self, mailbox: dict[str, Any] | str, message_id: str) -> dict[str, Any] | None: """ 调用 GET /messages/{id} 获取邮件详情。 如果 token 失效，自动尝试刷新。 """ email_addr = self._coerce_email(mailbox) token = self._get_token(mailbox) if isinstance(mailbox, dict) else "" if not token: raise DuckMailProviderError( "UNAUTHORIZED", f"邮箱 {email_addr} 缺少 auth_token，无法读取邮件详情", data={"email": email_addr}, ) # 还原 DuckMail 原始 ID dk_id = message_id if message_id.startswith("dk_"): dk_id = message_id[3:] base_url = self._base_url() resp = self._do_request(base_url, "GET", f"/messages/{dk_id}", headers=self._user_headers(token)) # token 失效时刷新 if resp and not resp.ok and resp.status_code == 401: refreshed_token = self._refresh_token(base_url, mailbox) if refreshed_token: resp = self._do_request(base_url, "GET", f"/messages/{dk_id}", headers=self._user_headers(refreshed_token)) if resp is None or not resp.ok: status = resp.status_code if resp else 0 body = resp.text[:200] if resp else "" code = _map_dk_http_error(status, body) raise DuckMailProviderError( code, f"DuckMail 读取邮件详情失败 HTTP {status}", data={"email": email_addr, "message_id": message_id, "status_code": status}, ) try: data = resp.json() except Exception: raise DuckMailProviderError( "UPSTREAM_BAD_PAYLOAD", "DuckMail 邮件详情返回非 JSON 响应", data={"email": email_addr, "message_id": message_id}, ) return self._detail_to_standard(data) def _detail_to_standard(self, dk_detail: dict[str, Any]) -> dict[str, Any]: """将 DuckMail 邮件详情转换为平台标准结构。""" dk_id = dk_detail.get("id") message_id = f"dk_{dk_id}" if dk_id is not None else "" created_at_str = str(dk_detail.get("createdAt") or "") timestamp = _iso_to_timestamp(created_at_str) if created_at_str else 0 from_info = dk_detail.get("from") or {} if isinstance(from_info, dict): from_address = str(from_info.get("address") or "").strip() else: from_address = str(from_info).strip() subject = str(dk_detail.get("subject") or "").strip() # body text_body = str(dk_detail.get("text") or "").strip() html_list = dk_detail.get("html") or [] if isinstance(html_list, list) and html_list: html_content = str(html_list[0] or "").strip() else: html_content = "" has_html = bool(html_content) return { "id": message_id, "message_id": message_id, "from_address": from_address, "subject": subject, "content": text_body, "html_content": html_content, "has_html": has_html, "timestamp": timestamp, "seen": bool(dk_detail.get("seen", False)), "has_attachments": bool(dk_detail.get("hasAttachments", False)), "size": int(dk_detail.get("size") or 0), "attachments": dk_detail.get("attachments", []), } def delete_message(self, mailbox: dict[str, Any] | str, message_id: str) -> bool: """ 调用 DELETE /messages/{id} 删除单封邮件。 如果 token 失效，尝试刷新。 """ token = self._get_token(mailbox) if isinstance(mailbox, dict) else "" email_addr = self._coerce_email(mailbox) if not token: logger.warning("[duckmail] delete_message: no token for %s", email_addr) return False # 还原 DuckMail 原始 ID dk_id = message_id if message_id.startswith("dk_"): dk_id = message_id[3:] base_url = self._base_url() resp = self._do_request(base_url, "DELETE", f"/messages/{dk_id}", headers=self._user_headers(token)) if resp and not resp.ok and resp.status_code == 401: refreshed_token = self._refresh_token(base_url, mailbox) if refreshed_token: resp = self._do_request(base_url, "DELETE", f"/messages/{dk_id}", headers=self._user_headers(refreshed_token)) if resp is None: return False return resp.ok or resp.status_code == 204 def clear_messages(self, mailbox: dict[str, Any]) -> bool: """ DuckMail 没有批量清空接口。 通过逐封删除实现（注意性能，最多删 30 封）。 如果 token 失效，尝试刷新。 """ token = self._get_token(mailbox) email_addr = self._coerce_email(mailbox) if not token: logger.warning("[duckmail] clear_messages: no token for %s", email_addr) return False base_url = self._base_url() resp = self._do_request(base_url, "GET", "/messages", headers=self._user_headers(token), params={"page": 1}) if resp and not resp.ok and resp.status_code == 401: refreshed_token = self._refresh_token(base_url, mailbox) if refreshed_token: resp = self._do_request(base_url, "GET", "/messages", headers=self._user_headers(refreshed_token), params={"page": 1}) if resp is None or not resp.ok: return False data = resp.json() members = data.get("hydra:member", []) if not isinstance(members, list): return False # 最多删 30 封 for msg in members[:30]: dk_id = str(msg.get("id") or "") if dk_id: try: self._do_request(base_url, "DELETE", f"/messages/{dk_id}", headers=self._user_headers(token)) except Exception: pass return True