Method get_verification_code

src/services/web2_mail.py:214–284

(
        self,
        email: str,
        email_id: str = None,
        timeout: int = 120,
        pattern: str = OTP_CODE_PATTERN,
        otp_sent_at: Optional[float] = None,
    )

            raise EmailServiceError(f"创建 Web2 邮箱失败: {e}")

    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]:
        mailbox = self._resolve_mailbox(email, email_id)
        if not mailbox:
            logger.warning("Web2 未找到邮箱缓存: %s", email)
            return None

        token = str(mailbox.get("token") or "").strip()
        if not token:
            logger.warning("Web2 邮箱缺少 token，尝试使用 service_id 兜底: %s", email)
            token = str(mailbox.get("service_id") or "").strip()
        if not token:
            logger.warning("Web2 邮箱缺少 token: %s", email)
            return None

        email_norm = str(email or mailbox.get("email") or "").strip().lower()
        seen_ids = self._seen_message_ids.setdefault(email_norm, set())
        start_time = time.time()

        while time.time() - start_time < timeout:
            try:
                response = self.http_client.get(
                    f"{self.config['base_url']}/messages",
                    headers=self._build_headers({
                        "Cache-Control": "no-cache",
                        "Authorization": f"Bearer {token}",
                    }),
                )
                if response.status_code != 200:
                    time.sleep(3)
                    continue

                payload = response.json()
                if isinstance(payload, dict):
                    messages = payload.get("messages")
                    if messages is None:
                        messages = payload.get("data") or payload.get("items") or payload.get("results") or []
                elif isinstance(payload, list):
                    messages = payload
                else:
                    messages = []
                if not isinstance(messages, list):
                    time.sleep(3)
                    continue

                for message in messages:
                    if not isinstance(message, dict):
                        continue
                    message_id = self._extract_message_id(message)
                    if message_id in seen_ids:
                        continue
                    seen_ids.add(message_id)