Method list_emails

src/services/temp_mail.py:848–900

列出邮箱 Args: limit: 返回数量上限 offset: 分页偏移 **kwargs: 额外查询参数，透传给 admin API Returns: 邮箱列表

(self, limit: int = 50, offset: int = 0, **kwargs)

        return None

    def list_emails(self, limit: int = 50, offset: int = 0, **kwargs) -> List[Dict[str, Any]]:
        """
        列出邮箱

        Args:
            limit: 返回数量上限
            offset: 分页偏移
            **kwargs: 额外查询参数，透传给 admin API

        Returns:
            邮箱列表
        """
        params = {k: v for k, v in kwargs.items() if v is not None}
        raw_limit = params.pop("limit", limit)
        raw_offset = params.pop("offset", offset)
        params["limit"] = self._normalize_admin_limit(raw_limit, default=50)
        params["offset"] = self._normalize_offset(raw_offset, default=0)

        try:
            response = self._request_admin_mails_with_limit_fallback(
                offset=params["offset"],
                extra_params={k: v for k, v in params.items() if k not in ("limit", "offset")},
                preferred_limit=params["limit"],
            )
            mails = response.get("results", [])
            if not isinstance(mails, list):
                raise EmailServiceError(f"API 返回数据格式错误: {response}")

            emails: List[Dict[str, Any]] = []
            for mail in mails:
                address = (mail.get("address") or "").strip()
                mail_id = mail.get("id") or address
                email_info = {
                    "id": mail_id,
                    "service_id": mail_id,
                    "email": address,
                    "subject": mail.get("subject"),
                    "from": mail.get("source"),
                    "created_at": mail.get("createdAt") or mail.get("created_at"),
                    "raw_data": mail,
                }
                emails.append(email_info)

                if address:
                    cached = self._email_cache.get(address, {})
                    self._email_cache[address] = {**cached, **email_info}

            self.update_status(True)
            return emails
        except Exception as e:
            logger.warning(f"列出 TempMail 邮箱失败: {e}")
            self.update_status(False, e)
            return list(self._email_cache.values())

    def delete_email(self, email_id: str) -> bool:
        """