Class TempMailService

src/services/temp_mail.py:28–947

Temp-Mail 邮箱服务 基于自部署 Cloudflare Worker 的临时邮箱，admin 模式管理邮箱 不走代理，不使用 requests 库

class TempMailService(BaseEmailService):
    """
    Temp-Mail 邮箱服务
    基于自部署 Cloudflare Worker 的临时邮箱，admin 模式管理邮箱
    不走代理，不使用 requests 库
    """

    def __init__(self, config: Dict[str, Any] = None, name: str = None):
        """
        初始化 TempMail 服务

        Args:
            config: 配置字典，支持以下键:
                - base_url: Worker 域名地址，如 https://mail.example.com (必需)
                - admin_password: Admin 密码，对应 x-admin-auth header (必需)
                - domain: 邮箱域名，如 example.com (必需)
                - enable_prefix: 是否启用前缀，默认 True
                - timeout: 请求超时时间，默认 30
                - max_retries: 最大重试次数，默认 3
            name: 服务名称
        """
        super().__init__(EmailServiceType.TEMP_MAIL, name)

        required_keys = ["base_url", "admin_password", "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 = {
            "enable_prefix": True,
            "timeout": 30,
            "max_retries": 3,
        }
        self.config = {**default_config, **(config or {})}

        # 不走代理，proxy_url=None
        http_config = RequestConfig(
            timeout=self.config["timeout"],
            max_retries=self.config["max_retries"],
        )
        self.http_client = HTTPClient(proxy_url=None, config=http_config)

        # 邮箱缓存：email -> {jwt, address}
        self._email_cache: Dict[str, Dict[str, Any]] = {}
        # 记录每个邮箱上一次成功使用的邮件 ID，避免重复使用旧验证码
        self._last_used_mail_ids: Dict[str, str] = {}
        # /admin/mails 接口对 limit 参数较严格，这里统一限制上限，避免 400 Invalid limit
        self._admin_mails_limit_max = 50

    def _normalize_admin_limit(self, value: Any, default: int = 50) -> int:
        try:
            number = int(value)
        except Exception:
            number = int(default)
        if number <= 0:
            number = int(default)
        return max(1, min(number, int(self._admin_mails_limit_max)))