(self, config: Dict[str, Any] = None)
| 176 | return None |
| 177 | |
| 178 | def create_email(self, config: Dict[str, Any] = None) -> Dict[str, Any]: |
| 179 | try: |
| 180 | response = self.http_client.post( |
| 181 | f"{self.config['base_url']}/mailbox", |
| 182 | headers=self._build_headers(), |
| 183 | json={}, |
| 184 | ) |
| 185 | if response.status_code != 200: |
| 186 | raise EmailServiceError(f"web2 创建邮箱失败,状态码: {response.status_code}") |
| 187 | |
| 188 | data = response.json() |
| 189 | mailbox = self._normalize_mailbox_payload(data) |
| 190 | |
| 191 | email = str(mailbox.get("email") or mailbox.get("address") or "").strip() |
| 192 | token = str(mailbox.get("token") or "").strip() |
| 193 | service_id = str(mailbox.get("service_id") or token or email).strip() |
| 194 | if not email: |
| 195 | raise EmailServiceError(f"web2 返回数据不完整: {data}") |
| 196 | |
| 197 | mailbox.update({ |
| 198 | "email": email, |
| 199 | "address": email, |
| 200 | "token": token, |
| 201 | "service_id": service_id, |
| 202 | "created_at": time.time(), |
| 203 | }) |
| 204 | self._cache_mailbox(mailbox) |
| 205 | self.update_status(True) |
| 206 | logger.info("Web2 邮箱创建成功: %s", email) |
| 207 | return mailbox |
| 208 | except Exception as e: |
| 209 | self.update_status(False, e) |
| 210 | if isinstance(e, EmailServiceError): |
| 211 | raise |
| 212 | raise EmailServiceError(f"创建 Web2 邮箱失败: {e}") |
| 213 | |
| 214 | def get_verification_code( |
| 215 | self, |
nothing calls this directly
no test coverage detected