Spaces:
Sleeping
Sleeping
| """ | |
| Cloudflare Temp Email 临时邮箱客户端 | |
| API 文档参考 (基于 Hono 框架，JWT 认证): | |
| - 获取公开配置: GET /open_api/settings | |
| - 创建新邮箱: POST /api/new_address body: {name, domain} → {address, jwt} | |
| - 获取邮件列表: GET /api/mails Authorization: Bearer {jwt} | |
| - 获取邮件详情: GET /api/mail/:mail_id Authorization: Bearer {jwt} | |
| """ | |
| import random | |
| import string | |
| import time | |
| from datetime import datetime | |
| from typing import Optional | |
| import requests | |
| from core.mail_utils import extract_verification_code | |
| from core.proxy_utils import request_with_proxy_fallback | |
| class CloudflareMailClient: | |
| """Cloudflare Temp Email 临时邮箱客户端""" | |
| def __init__( | |
| self, | |
| base_url: str = "", | |
| proxy: str = "", | |
| api_key: str = "", | |
| domain: str = "", | |
| verify_ssl: bool = True, | |
| log_callback=None, | |
| ) -> None: | |
| self.base_url = (base_url or "").rstrip("/") | |
| self.proxy_url = (proxy or "").strip() | |
| self.api_key = (api_key or "").strip() # x-custom-auth 密码 | |
| self.domain = (domain or "").strip() | |
| self.verify_ssl = verify_ssl | |
| self.log_callback = log_callback | |
| self.email: Optional[str] = None | |
| self.password: Optional[str] = None # 兼容接口，存储 JWT token | |
| self.jwt_token: Optional[str] = None # 创建地址时返回的 JWT | |
| self._available_domains: list = [] | |
| # ------------------------------------------------------------------ | |
| # 内部工具 | |
| # ------------------------------------------------------------------ | |
| def _log(self, level: str, message: str) -> None: | |
| if self.log_callback: | |
| try: | |
| self.log_callback(level, message) | |
| except Exception: | |
| pass | |
| def _request(self, method: str, url: str, **kwargs) -> requests.Response: | |
| headers = kwargs.pop("headers", None) or {} | |
| # 实例密码认证（admin 路由使用 x-admin-auth） | |
| if self.api_key and "x-admin-auth" not in {k.lower() for k in headers}: | |
| headers["x-admin-auth"] = self.api_key | |
| # 邮件操作时使用 JWT Bearer 认证 | |
| if self.jwt_token and "authorization" not in {k.lower() for k in headers}: | |
| headers["Authorization"] = f"Bearer {self.jwt_token}" | |
| kwargs["headers"] = headers | |
| self._log("info", f"📤 发送 {method} 请求: {url}") | |
| if "json" in kwargs and kwargs["json"] is not None: | |
| self._log("info", f"📦 请求体: {kwargs['json']}") | |
| proxies = {"http": self.proxy_url, "https": self.proxy_url} if self.proxy_url else None | |
| try: | |
| res = request_with_proxy_fallback( | |
| requests.request, | |
| method, | |
| url, | |
| proxies=proxies, | |
| verify=self.verify_ssl, | |
| timeout=kwargs.pop("timeout", 30), | |
| **kwargs, | |
| ) | |
| self._log("info", f"📥 收到响应: HTTP {res.status_code}") | |
| if res.content and res.status_code >= 400: | |
| try: | |
| self._log("error", f"📄 响应内容: {res.text[:500]}") | |
| except Exception: | |
| pass | |
| return res | |
| except Exception as e: | |
| self._log("error", f"❌ 网络请求失败: {e}") | |
| raise | |
| # ------------------------------------------------------------------ | |
| # 公开接口 | |
| # ------------------------------------------------------------------ | |
| def set_credentials(self, email: str, password: str = "") -> None: | |
| """设置凭据（兼容接口）。password 存储 JWT token。""" | |
| self.email = email | |
| self.password = password | |
| if password: | |
| self.jwt_token = password | |
| def _get_available_domains(self) -> list: | |
| """GET /open_api/settings 获取可用域名列表""" | |
| if self._available_domains: | |
| return self._available_domains | |
| try: | |
| res = self._request("GET", f"{self.base_url}/open_api/settings") | |
| if res.status_code == 200: | |
| data = res.json() if res.content else {} | |
| domains = data.get("domains", []) | |
| if isinstance(domains, list) and domains: | |
| self._available_domains = [str(d).strip() for d in domains if d] | |
| self._log("info", f"🌐 CFMail 可用域名: {self._available_domains}") | |
| return self._available_domains | |
| except Exception as e: | |
| self._log("error", f"❌ 获取可用域名失败: {e}") | |
| return self._available_domains | |
| def register_account(self, domain: Optional[str] = None) -> bool: | |
| """POST /api/new_address 创建新邮箱地址""" | |
| if not self.base_url: | |
| self._log("error", "❌ cfmail_base_url 未配置") | |
| return False | |
| # 确定域名 | |
| selected_domain = domain or self.domain | |
| if not selected_domain: | |
| available = self._get_available_domains() | |
| if available: | |
| selected_domain = random.choice(available) | |
| # 生成随机用户名 | |
| rand = "".join(random.choices(string.ascii_lowercase + string.digits, k=10)) | |
| timestamp = str(int(time.time()))[-4:] | |
| name = f"t{timestamp}{rand}" | |
| payload = {"name": name} | |
| if selected_domain: | |
| payload["domain"] = selected_domain | |
| self._log("info", f"📧 使用域名: {selected_domain}") | |
| self._log("info", f"🎲 创建邮箱: {name}") | |
| try: | |
| res = self._request("POST", f"{self.base_url}/admin/new_address", json=payload) | |
| if res.status_code in (200, 201): | |
| data = res.json() if res.content else {} | |
| address = data.get("address", "") | |
| jwt = data.get("jwt", "") | |
| if address: | |
| self.email = address | |
| self.jwt_token = jwt | |
| self.password = jwt # 兼容接口 | |
| self._log("info", f"✅ CFMail 注册成功: {self.email}") | |
| return True | |
| self._log("error", f"❌ CFMail 注册失败: HTTP {res.status_code}") | |
| return False | |
| except Exception as e: | |
| self._log("error", f"❌ CFMail 注册异常: {e}") | |
| return False | |
| def login(self) -> bool: | |
| """无需登录，直接返回 True""" | |
| return True | |
| def _extract_body_from_raw(raw: str) -> str: | |
| """从原始邮件中提取正文（text/plain + text/html），跳过 header""" | |
| if not raw: | |
| return "" | |
| import email as _email | |
| try: | |
| msg = _email.message_from_string(raw) | |
| parts = [] | |
| if msg.is_multipart(): | |
| for part in msg.walk(): | |
| ct = part.get_content_type() | |
| if ct in ("text/plain", "text/html"): | |
| payload = part.get_payload(decode=True) | |
| if payload: | |
| charset = part.get_content_charset() or "utf-8" | |
| parts.append(payload.decode(charset, errors="replace")) | |
| else: | |
| payload = msg.get_payload(decode=True) | |
| if payload: | |
| charset = msg.get_content_charset() or "utf-8" | |
| parts.append(payload.decode(charset, errors="replace")) | |
| return "".join(parts) | |
| except Exception: | |
| return "" | |
| def fetch_verification_code(self, since_time: Optional[datetime] = None) -> Optional[str]: | |
| """GET /api/mails 获取邮件列表，再 GET /api/mail/:id 获取详情，提取验证码""" | |
| if not self.jwt_token: | |
| self._log("error", "❌ 缺少 JWT token，无法获取邮件") | |
| return None | |
| try: | |
| self._log("info", "📬 正在拉取 CFMail 邮件列表...") | |
| res = self._request("GET", f"{self.base_url}/api/mails", params={"limit": 20, "offset": 0}) | |
| if res.status_code != 200: | |
| self._log("error", f"❌ 获取邮件列表失败: HTTP {res.status_code}") | |
| return None | |
| data = res.json() if res.content else {} | |
| # 响应格式: {"results": [...], "total": N} | |
| messages = data.get("results", []) | |
| if not isinstance(messages, list): | |
| messages = [] | |
| if not messages: | |
| self._log("info", "📭 邮箱为空，暂无邮件") | |
| return None | |
| self._log("info", f"📨 收到 {len(messages)} 封邮件，开始检查验证码...") | |
| # 按 id 降序（新邮件优先） | |
| try: | |
| messages = sorted(messages, key=lambda m: int(m.get("id") or 0), reverse=True) | |
| except Exception: | |
| pass | |
| for idx, msg in enumerate(messages, 1): | |
| msg_id = msg.get("id") | |
| if not msg_id: | |
| continue | |
| # 时间过滤 | |
| if since_time: | |
| raw_time = msg.get("created_at") or msg.get("createdAt") | |
| if raw_time: | |
| try: | |
| if isinstance(raw_time, (int, float)): | |
| ts = float(raw_time) | |
| if ts > 1e12: | |
| ts /= 1000.0 | |
| msg_time = datetime.fromtimestamp(ts) | |
| else: | |
| import re | |
| raw_time = re.sub(r"(\.\d{6})\d+", r"\1", str(raw_time)) | |
| # cfmail 的 created_at 是 UTC 无时区标记，显式加 +00:00 再转本地时间 | |
| if not raw_time.endswith("Z") and "+" not in raw_time and raw_time.count("-") <= 2: | |
| raw_time = raw_time + "+00:00" | |
| msg_time = datetime.fromisoformat(raw_time.replace("Z", "+00:00")).astimezone().replace(tzinfo=None) | |
| if msg_time < since_time: | |
| continue | |
| except Exception: | |
| pass | |
| # 列表响应已包含 raw 字段，直接解析正文提取验证码 | |
| raw_in_list = msg.get("raw") or "" | |
| if raw_in_list: | |
| body = self._extract_body_from_raw(raw_in_list) | |
| code = extract_verification_code(body) | |
| if code: | |
| self._log("info", f"✅ 找到验证码: {code}") | |
| return code | |
| # 兜底：尝试从其他摘要字段提取 | |
| summary = (msg.get("subject") or "") + (msg.get("text") or "") + (msg.get("html") or "") | |
| if summary: | |
| code = extract_verification_code(summary) | |
| if code: | |
| self._log("info", f"✅ 找到验证码: {code}") | |
| return code | |
| # 获取邮件详情 | |
| self._log("info", f"🔍 正在读取邮件 {idx}/{len(messages)} 详情...") | |
| detail_res = self._request("GET", f"{self.base_url}/api/mail/{msg_id}") | |
| if detail_res.status_code != 200: | |
| self._log("warning", f"⚠️ 读取邮件详情失败: HTTP {detail_res.status_code}") | |
| continue | |
| detail = detail_res.json() if detail_res.content else {} | |
| content = self._extract_body_from_raw(detail.get("raw") or "") | |
| if content: | |
| code = extract_verification_code(content) | |
| if code: | |
| self._log("info", f"✅ 找到验证码: {code}") | |
| return code | |
| else: | |
| self._log("info", f"❌ 邮件 {idx} 中未找到验证码") | |
| self._log("warning", "⚠️ 所有邮件中均未找到验证码") | |
| return None | |
| except Exception as e: | |
| self._log("error", f"❌ 获取验证码异常: {e}") | |
| return None | |
| def poll_for_code( | |
| self, | |
| timeout: int = 120, | |
| interval: int = 4, | |
| since_time: Optional[datetime] = None, | |
| ) -> Optional[str]: | |
| """轮询获取验证码""" | |
| if not self.email: | |
| return None | |
| max_retries = max(1, timeout // interval) | |
| self._log("info", f"⏱️ 开始轮询验证码 (超时 {timeout}秒, 间隔 {interval}秒, 最多 {max_retries} 次)") | |
| for i in range(1, max_retries + 1): | |
| self._log("info", f"🔄 第 {i}/{max_retries} 次轮询...") | |
| code = self.fetch_verification_code(since_time=since_time) | |
| if code: | |
| self._log("info", f"🎉 验证码获取成功: {code}") | |
| return code | |
| if i < max_retries: | |
| self._log("info", f"⏳ 等待 {interval} 秒后重试...") | |
| time.sleep(interval) | |
| self._log("error", f"⏰ 验证码获取超时 ({timeout}秒)") | |
| return None | |