import random import string import time from typing import Optional import requests from core.mail_utils import extract_verification_code from core.proxy_utils import request_with_proxy_fallback class FreemailClient: """Freemail 临时邮箱客户端""" def __init__( self, base_url: str = "http://your-freemail-server.com", jwt_token: str = "", proxy: str = "", verify_ssl: bool = True, log_callback=None, ) -> None: self.base_url = base_url.rstrip("/") self.jwt_token = jwt_token.strip() self.verify_ssl = verify_ssl self.proxies = {"http": proxy, "https": proxy} if proxy else None self.log_callback = log_callback self.email: Optional[str] = None def set_credentials(self, email: str, password: str = None) -> None: """设置邮箱凭证（Freemail 不需要密码）""" self.email = email def _request(self, method: str, url: str, **kwargs) -> requests.Response: """发送请求并打印日志""" self._log("info", f"📤 发送 {method} 请求: {url}") if "params" in kwargs: self._log("info", f"🔎 参数: {kwargs['params']}") try: res = request_with_proxy_fallback( requests.request, method, url, proxies=self.proxies, verify=self.verify_ssl, timeout=kwargs.pop("timeout", 15), **kwargs, ) self._log("info", f"📥 收到响应: HTTP {res.status_code}") if 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 register_account(self, domain: Optional[str] = None) -> bool: """创建新的临时邮箱""" try: params = {"admin_token": self.jwt_token} if domain: params["domain"] = domain self._log("info", f"📧 使用域名: {domain}") else: self._log("info", "🔍 自动选择域名...") res = self._request( "POST", f"{self.base_url}/api/generate", params=params, ) if res.status_code in (200, 201): data = res.json() if res.content else {} # Freemail API 返回的字段是 "email" 或 "mailbox" email = data.get("email") or data.get("mailbox") if email: self.email = email self._log("info", f"✅ Freemail 邮箱创建成功: {self.email}") return True else: self._log("error", "❌ 响应中缺少 email 字段") return False elif res.status_code in (401, 403): self._log("error", "❌ Freemail 认证失败 (JWT Token 无效)") return False else: self._log("error", f"❌ Freemail 创建失败: HTTP {res.status_code}") return False except Exception as e: self._log("error", f"❌ Freemail 注册异常: {e}") return False def login(self) -> bool: """登录（Freemail 不需要登录，直接返回 True）""" return True def fetch_verification_code(self, since_time=None) -> Optional[str]: """获取验证码""" if not self.email: self._log("error", "❌ 邮箱地址未设置") return None try: self._log("info", "📬 正在拉取 Freemail 邮件列表...") params = { "mailbox": self.email, "admin_token": self.jwt_token, } res = self._request( "GET", f"{self.base_url}/api/emails", params=params, ) if res.status_code == 401 or res.status_code == 403: self._log("error", "❌ Freemail 认证失败") return None if res.status_code != 200: self._log("error", f"❌ 获取邮件列表失败: HTTP {res.status_code}") return None emails = res.json() if res.content else [] if not isinstance(emails, list): self._log("error", "❌ 响应格式错误（不是列表）") return None if not emails: self._log("info", "📭 邮箱为空，暂无邮件") return None self._log("info", f"📨 收到 {len(emails)} 封邮件，开始检查验证码...") from datetime import datetime, timezone import re def _parse_email_time(email_obj) -> Optional[datetime]: time_keys = ( "created_at", "createdAt", "received_at", "receivedAt", "sent_at", "sentAt", ) raw_time = None for key in time_keys: if email_obj.get(key) is not None: raw_time = email_obj.get(key) break if raw_time is None: return None if isinstance(raw_time, (int, float)): timestamp = float(raw_time) if timestamp > 1e12: timestamp = timestamp / 1000.0 return datetime.fromtimestamp(timestamp).astimezone().replace(tzinfo=None) if isinstance(raw_time, str): raw = raw_time.strip() if not raw: return None if raw.isdigit(): timestamp = float(raw) if timestamp > 1e12: timestamp = timestamp / 1000.0 return datetime.fromtimestamp(timestamp).astimezone().replace(tzinfo=None) # 截断纳秒到微秒（fromisoformat 只支持6位小数） raw = re.sub(r"(\.\d{6})\d+", r"\1", raw) try: parsed = datetime.fromisoformat(raw.replace("Z", "+00:00")) if parsed.tzinfo: return parsed.astimezone().replace(tzinfo=None) return parsed.replace(tzinfo=timezone.utc).astimezone().replace(tzinfo=None) except Exception: return None return None # 按时间倒序，优先检查最新邮件 emails_with_time = [(email_item, _parse_email_time(email_item)) for email_item in emails] if any(item[1] is not None for item in emails_with_time): emails_with_time.sort(key=lambda item: item[1] or datetime.min, reverse=True) emails = [item[0] for item in emails_with_time] skipped_no_time_indexes = [] skipped_expired_indexes = [] def _format_indexes(indexes: list[int]) -> str: if len(indexes) <= 10: return ",".join(str(index) for index in indexes) preview = ",".join(str(index) for index in indexes[:10]) return f"{preview}...(+{len(indexes) - 10})" def _log_skip_summary() -> None: if skipped_no_time_indexes: self._log( "info", f"⏭️ 已跳过 {len(skipped_no_time_indexes)} 封缺少可解析时间的邮件" f"（序号: {_format_indexes(skipped_no_time_indexes)}）", ) if skipped_expired_indexes: self._log( "info", f"⏭️ 已跳过 {len(skipped_expired_indexes)} 封过期邮件" f"（序号: {_format_indexes(skipped_expired_indexes)}）", ) # 从最新一封邮件开始查找 for idx, email_data in enumerate(emails, 1): # 时间过滤 if since_time: email_time = _parse_email_time(email_data) if email_time is None: skipped_no_time_indexes.append(idx) continue if email_time < since_time: skipped_expired_indexes.append(idx) continue # 获取邮件完整内容 email_id = email_data.get("id") if email_id: # 调用详情接口获取完整内容 detail_res = self._request( "GET", f"{self.base_url}/api/email/{email_id}", params={"admin_token": self.jwt_token}, ) if detail_res.status_code == 200: detail_data = detail_res.json() content = detail_data.get("content") or "" html_content = detail_data.get("html_content") or "" else: # 降级：如果详情接口失败，使用列表中的字段 content = email_data.get("content") or "" html_content = email_data.get("html_content") or "" preview = email_data.get("preview") or "" content = content + " " + preview else: # 降级：没有 ID，使用列表中的字段 content = email_data.get("content") or "" html_content = email_data.get("html_content") or "" preview = email_data.get("preview") or "" content = content + " " + preview subject = email_data.get("subject") or "" full_content = subject + " " + content + " " + html_content code = extract_verification_code(full_content) if code: _log_skip_summary() self._log("info", f"✅ 找到验证码: {code}") return code else: self._log("info", f"❌ 邮件 {idx} 中未找到验证码") _log_skip_summary() 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=None, ) -> Optional[str]: """轮询获取验证码""" 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 def _get_domain(self) -> str: """获取可用域名""" try: params = {"admin_token": self.jwt_token} res = self._request( "GET", f"{self.base_url}/api/domains", params=params, ) if res.status_code == 200: domains = res.json() if res.content else [] if isinstance(domains, list) and domains: return domains[0] except Exception: pass return "" def _log(self, level: str, message: str) -> None: """日志回调""" if self.log_callback: try: self.log_callback(level, message) except Exception: pass