From f5eb1b8bc102a1ce0bf48a86e035a94104c1b6fd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=81=E4=BA=91?= <58023926@qq.com> Date: Thu, 2 Apr 2026 10:31:25 +0800 Subject: [PATCH] =?UTF-8?q?feat(mail):=20=E6=B7=BB=E5=8A=A0=20GPTMail=20?= =?UTF-8?q?=E9=82=AE=E7=AE=B1=E6=B8=A0=E9=81=93=E6=94=AF=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 1 + api/config.py | 5 + core/base_mailbox.py | 202 ++++++++++++++++++++++++ frontend/src/pages/Accounts.tsx | 3 + frontend/src/pages/RegisterTaskPage.tsx | 25 +++ frontend/src/pages/Settings.tsx | 13 ++ tests/test_gptmail_mailbox.py | 150 ++++++++++++++++++ 7 files changed, 399 insertions(+) create mode 100644 tests/test_gptmail_mailbox.py diff --git a/README.md b/README.md index 97a453a..42997d3 100644 --- a/README.md +++ b/README.md @@ -129,6 +129,7 @@ | TempMail.lol | `tempmail_lol` | 临时邮箱方案,部分地区可能需要代理 | | SkyMail (CloudMail) | `skymail` | 通过 API / Token / 域名使用 | | YYDS Mail / MaliAPI | `maliapi` | 支持域名与自动域名策略 | +| GPTMail | `gptmail` | 基于 GPTMail API 生成临时邮箱并轮询邮件,支持已知域名时本地拼装随机地址 | | DuckMail | `duckmail` | 临时邮箱方案 | | Freemail | `freemail` | 自建邮箱服务 | | Laoudo | `laoudo` | 固定邮箱方案 | diff --git a/api/config.py b/api/config.py index 81e69f4..110dd03 100644 --- a/api/config.py +++ b/api/config.py @@ -30,6 +30,9 @@ CONFIG_KEYS = [ "maliapi_api_key", "maliapi_domain", "maliapi_auto_domain_strategy", + "gptmail_base_url", + "gptmail_api_key", + "gptmail_domain", "cfworker_api_url", "cfworker_admin_token", "cfworker_custom_auth", @@ -83,6 +86,8 @@ def get_config(): all_cfg = config_store.get_all() if not all_cfg.get("mail_provider"): all_cfg["mail_provider"] = "luckmail" + if not all_cfg.get("gptmail_base_url"): + all_cfg["gptmail_base_url"] = "https://mail.chatgpt.org.uk" if not all_cfg.get("luckmail_base_url"): all_cfg["luckmail_base_url"] = "https://mails.luckyous.com/" # 只返回已知 key,未设置的返回空字符串 diff --git a/core/base_mailbox.py b/core/base_mailbox.py index 9bab56e..1ded7bc 100644 --- a/core/base_mailbox.py +++ b/core/base_mailbox.py @@ -187,6 +187,13 @@ def create_mailbox( auto_domain_strategy=extra.get("maliapi_auto_domain_strategy", ""), proxy=proxy, ) + elif provider == "gptmail": + return GPTMailMailbox( + api_url=extra.get("gptmail_base_url", "https://mail.chatgpt.org.uk"), + api_key=extra.get("gptmail_api_key", ""), + domain=extra.get("gptmail_domain", ""), + proxy=proxy, + ) elif provider == "cfworker": return CFWorkerMailbox( api_url=extra.get("cfworker_api_url", ""), @@ -987,6 +994,201 @@ class MaliAPIMailbox(BaseMailbox): ) +class GPTMailMailbox(BaseMailbox): + """GPTMail 临时邮箱服务""" + + def __init__( + self, + api_url: str = "https://mail.chatgpt.org.uk", + api_key: str = "", + domain: str = "", + proxy: str = None, + ): + self.api = (api_url or "https://mail.chatgpt.org.uk").rstrip("/") + self.api_key = str(api_key or "").strip() + self.domain = self._normalize_domain(domain) + self.proxy = build_requests_proxy_config(proxy) + self._email = None + + @staticmethod + def _normalize_domain(value: Any) -> str: + domain = str(value or "").strip().lower() + if domain.startswith("@"): + domain = domain[1:] + return domain + + @staticmethod + def _generate_local_part() -> str: + import string + + prefix = "".join(random.choices(string.ascii_lowercase, k=6)) + suffix = "".join(random.choices(string.digits, k=4)) + return f"{prefix}{suffix}" + + def _headers(self) -> dict[str, str]: + headers = {"accept": "application/json"} + if self.api_key: + headers["X-API-Key"] = self.api_key + return headers + + def _request_json( + self, + method: str, + path: str, + *, + params: dict | None = None, + json_body: dict | None = None, + timeout: int = 15, + ) -> Any: + import requests + + response = requests.request( + method, + f"{self.api}{path}", + params=params, + json=json_body, + headers=self._headers(), + proxies=self.proxy, + timeout=timeout, + ) + try: + payload = response.json() + except Exception as exc: + preview = (response.text or "")[:200] + raise RuntimeError( + f"GPTMail API {path} 返回非 JSON: HTTP {response.status_code} {preview}" + ) from exc + + if response.status_code >= 400: + error = payload.get("error") if isinstance(payload, dict) else "" + message = str(error or response.text or f"HTTP {response.status_code}").strip() + raise RuntimeError(f"GPTMail API {path} 失败: {message}") + + if isinstance(payload, dict) and payload.get("success") is False: + error = str(payload.get("error") or "unknown error").strip() + raise RuntimeError(f"GPTMail API {path} 失败: {error}") + + if isinstance(payload, dict) and "data" in payload: + return payload.get("data") + return payload + + def _list_messages(self, email: str) -> list[dict]: + data = self._request_json("GET", "/api/emails", params={"email": email}, timeout=10) + if isinstance(data, dict): + messages = data.get("emails", []) + else: + messages = data + return [item for item in (messages or []) if isinstance(item, dict)] + + def _get_message_detail(self, message_id: str) -> dict[str, Any]: + data = self._request_json("GET", f"/api/email/{message_id}", timeout=10) + return data if isinstance(data, dict) else {} + + def get_email(self) -> MailboxAccount: + if self.domain: + email = f"{self._generate_local_part()}@{self.domain}" + self._email = email + self._log(f"[GPTMail] 本地拼装邮箱: {email}") + return MailboxAccount( + email=email, + account_id=email, + extra={"provider": "gptmail", "domain": self.domain, "local_address": True}, + ) + + data = self._request_json("GET", "/api/generate-email") + if not isinstance(data, dict): + raise RuntimeError(f"GPTMail 返回异常: {data}") + + email = str(data.get("email") or "").strip() + if not email: + raise RuntimeError(f"GPTMail 返回空邮箱: {data}") + + self._email = email + self._log(f"[GPTMail] 生成邮箱: {email}") + return MailboxAccount( + email=email, + account_id=email, + extra={"provider": "gptmail"}, + ) + + def get_current_ids(self, account: MailboxAccount) -> set: + try: + return { + str(message.get("id")) + for message in self._list_messages(account.email) + if message.get("id") is not None + } + except Exception: + return set() + + def wait_for_code( + self, + account: MailboxAccount, + keyword: str = "", + timeout: int = 120, + before_ids: set = None, + code_pattern: str = None, + **kwargs, + ) -> str: + import re + + seen = {str(mid) for mid in (before_ids or set())} + exclude_codes = { + str(code) for code in (kwargs.get("exclude_codes") or set()) if code + } + + def poll_once() -> Optional[str]: + try: + messages = self._list_messages(account.email) + for message in messages: + message_id = str(message.get("id") or "").strip() + if not message_id or message_id in seen: + continue + seen.add(message_id) + + try: + detail = self._get_message_detail(message_id) + except Exception: + detail = {} + + search_text = " ".join( + [ + str(message.get("subject") or ""), + str(message.get("from_address") or ""), + str(message.get("content") or ""), + str(message.get("html_content") or ""), + str(detail.get("subject") or ""), + str(detail.get("content") or ""), + str(detail.get("html_content") or ""), + str(detail.get("raw_headers") or ""), + ] + ).strip() + search_text = self._decode_raw_content(search_text) or search_text + search_text = re.sub( + r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", + "", + search_text, + ) + if keyword and keyword.lower() not in search_text.lower(): + continue + + code = self._safe_extract(search_text, code_pattern) + if code and code in exclude_codes: + continue + if code: + self._log(f"[GPTMail] 收到验证码: {code}") + return code + except Exception: + pass + return None + + return self._run_polling_wait( + timeout=timeout, + poll_interval=3, + poll_once=poll_once, + ) + + class CFWorkerMailbox(BaseMailbox): """Cloudflare Worker 自建临时邮箱服务""" diff --git a/frontend/src/pages/Accounts.tsx b/frontend/src/pages/Accounts.tsx index 6c11a57..24bbf6b 100644 --- a/frontend/src/pages/Accounts.tsx +++ b/frontend/src/pages/Accounts.tsx @@ -632,6 +632,9 @@ export default function Accounts() { laoudo_auth: cfg.laoudo_auth, laoudo_email: cfg.laoudo_email, laoudo_account_id: cfg.laoudo_account_id, + gptmail_base_url: cfg.gptmail_base_url, + gptmail_api_key: cfg.gptmail_api_key, + gptmail_domain: cfg.gptmail_domain, maliapi_base_url: cfg.maliapi_base_url, maliapi_api_key: cfg.maliapi_api_key, maliapi_domain: cfg.maliapi_domain, diff --git a/frontend/src/pages/RegisterTaskPage.tsx b/frontend/src/pages/RegisterTaskPage.tsx index dcc259c..227bd30 100644 --- a/frontend/src/pages/RegisterTaskPage.tsx +++ b/frontend/src/pages/RegisterTaskPage.tsx @@ -50,6 +50,9 @@ export default function RegisterTaskPage() { laoudo_auth: cfg.laoudo_auth || '', laoudo_email: cfg.laoudo_email || '', laoudo_account_id: cfg.laoudo_account_id || '', + gptmail_base_url: cfg.gptmail_base_url || 'https://mail.chatgpt.org.uk', + gptmail_api_key: cfg.gptmail_api_key || '', + gptmail_domain: cfg.gptmail_domain || '', maliapi_base_url: cfg.maliapi_base_url || 'https://maliapi.215.im/v1', maliapi_api_key: cfg.maliapi_api_key || '', maliapi_domain: cfg.maliapi_domain || '', @@ -89,6 +92,9 @@ export default function RegisterTaskPage() { laoudo_auth: values.laoudo_auth, laoudo_email: values.laoudo_email, laoudo_account_id: values.laoudo_account_id, + gptmail_base_url: values.gptmail_base_url, + gptmail_api_key: values.gptmail_api_key, + gptmail_domain: values.gptmail_domain, maliapi_base_url: values.maliapi_base_url, maliapi_api_key: values.maliapi_api_key, maliapi_domain: values.maliapi_domain, @@ -191,6 +197,7 @@ export default function RegisterTaskPage() { executor_type: 'protocol', captcha_solver: 'yescaptcha', mail_provider: 'luckmail', + gptmail_base_url: 'https://mail.chatgpt.org.uk', count: 1, register_delay_seconds: 0, maliapi_base_url: 'https://maliapi.215.im/v1', @@ -255,6 +262,7 @@ export default function RegisterTaskPage() { { value: 'tempmail_lol', label: 'TempMail.lol' }, { value: 'skymail', label: 'SkyMail (CloudMail)' }, { value: 'maliapi', label: 'YYDS Mail / MaliAPI' }, + { value: 'gptmail', label: 'GPTMail' }, { value: 'duckmail', label: 'DuckMail' }, { value: 'freemail', label: 'Freemail' }, { value: 'laoudo', label: 'Laoudo' }, @@ -310,6 +318,23 @@ export default function RegisterTaskPage() { )} + {mailProvider === 'gptmail' && ( + <> + + + + + + + + + + + )} {mailProvider === 'cfworker' && ( <> diff --git a/frontend/src/pages/Settings.tsx b/frontend/src/pages/Settings.tsx index 0bd61fd..df1cf65 100644 --- a/frontend/src/pages/Settings.tsx +++ b/frontend/src/pages/Settings.tsx @@ -25,6 +25,7 @@ const SELECT_FIELDS: Record = { { label: 'DuckMail(自动生成)', value: 'duckmail' }, { label: 'MoeMail (sall.cc)', value: 'moemail' }, { label: 'YYDS Mail / MaliAPI', value: 'maliapi' }, + { label: 'GPTMail', value: 'gptmail' }, { label: 'Freemail(自建 CF Worker)', value: 'freemail' }, { label: 'CF Worker(自建域名)', value: 'cfworker' }, ], @@ -119,6 +120,15 @@ const TAB_ITEMS = [ { key: 'maliapi_auto_domain_strategy', label: '自动域名策略', type: 'select' }, ], }, + { + title: 'GPTMail', + desc: '基于 GPTMail API 生成临时邮箱并轮询邮件;若已知本站可用域名,也可本地拼装随机地址', + fields: [ + { key: 'gptmail_base_url', label: 'API URL', placeholder: 'https://mail.chatgpt.org.uk' }, + { key: 'gptmail_api_key', label: 'API Key', secret: true, placeholder: 'gpt-test' }, + { key: 'gptmail_domain', label: '邮箱域名(可选)', placeholder: 'example.com' }, + ], + }, { title: 'TempMail.lol', desc: '自动生成邮箱,无需配置,需要代理访问(CN IP 被封)', @@ -1038,6 +1048,9 @@ export default function Settings() { if (!data.mail_provider) { data.mail_provider = 'luckmail' } + if (!data.gptmail_base_url) { + data.gptmail_base_url = 'https://mail.chatgpt.org.uk' + } if (!data.maliapi_base_url) { data.maliapi_base_url = 'https://maliapi.215.im/v1' } diff --git a/tests/test_gptmail_mailbox.py b/tests/test_gptmail_mailbox.py new file mode 100644 index 0000000..3000656 --- /dev/null +++ b/tests/test_gptmail_mailbox.py @@ -0,0 +1,150 @@ +import unittest +from unittest.mock import patch + +from core.base_mailbox import MailboxAccount, create_mailbox + + +class GPTMailMailboxTests(unittest.TestCase): + def _build_mailbox(self, **extra): + config = { + "gptmail_base_url": "https://mail.chatgpt.org.uk", + "gptmail_api_key": "gpt-test", + } + config.update(extra) + return create_mailbox("gptmail", extra=config) + + @patch("requests.request") + def test_get_email_issues_generate_request(self, mock_request): + mock_request.return_value.status_code = 200 + mock_request.return_value.json.return_value = { + "success": True, + "data": {"email": "demo@example.com"}, + } + + mailbox = self._build_mailbox() + account = mailbox.get_email() + + self.assertEqual(account.email, "demo@example.com") + self.assertEqual(account.account_id, "demo@example.com") + mock_request.assert_called_once_with( + "GET", + "https://mail.chatgpt.org.uk/api/generate-email", + params=None, + json=None, + headers={ + "accept": "application/json", + "X-API-Key": "gpt-test", + }, + proxies=None, + timeout=15, + ) + + @patch("requests.request") + def test_get_email_can_compose_local_address_when_domain_configured(self, mock_request): + mailbox = self._build_mailbox(gptmail_domain="known.example") + + with patch.object(type(mailbox), "_generate_local_part", return_value="demo1234"): + account = mailbox.get_email() + + self.assertEqual(account.email, "demo1234@known.example") + self.assertEqual(account.extra["domain"], "known.example") + mock_request.assert_not_called() + + @patch("requests.request") + def test_get_current_ids_reads_inbox_messages(self, mock_request): + mock_request.return_value.status_code = 200 + mock_request.return_value.json.return_value = { + "success": True, + "data": { + "emails": [ + {"id": "m1", "subject": "Hello"}, + {"id": "m2", "subject": "World"}, + ] + }, + } + + mailbox = self._build_mailbox() + ids = mailbox.get_current_ids(MailboxAccount(email="demo@example.com")) + + self.assertEqual(ids, {"m1", "m2"}) + mock_request.assert_called_once_with( + "GET", + "https://mail.chatgpt.org.uk/api/emails", + params={"email": "demo@example.com"}, + json=None, + headers={ + "accept": "application/json", + "X-API-Key": "gpt-test", + }, + proxies=None, + timeout=10, + ) + + @patch("time.sleep", return_value=None) + @patch("requests.request") + def test_wait_for_code_skips_excluded_codes_and_fetches_detail(self, mock_request, _sleep): + mock_request.side_effect = [ + _response( + { + "success": True, + "data": { + "emails": [ + {"id": "m1", "subject": "Your code: 111111"}, + ] + }, + } + ), + _response( + { + "success": True, + "data": { + "id": "m1", + "subject": "Your code: 111111", + "content": "111111", + }, + } + ), + _response( + { + "success": True, + "data": { + "emails": [ + {"id": "m1", "subject": "Your code: 111111"}, + {"id": "m2", "subject": "Your code: 222222"}, + ] + }, + } + ), + _response( + { + "success": True, + "data": { + "id": "m2", + "subject": "Your code: 222222", + "content": "verification code 222222", + }, + } + ), + ] + + mailbox = self._build_mailbox() + code = mailbox.wait_for_code( + MailboxAccount(email="demo@example.com"), + timeout=5, + exclude_codes={"111111"}, + ) + + self.assertEqual(code, "222222") + self.assertEqual(mock_request.call_count, 4) + + +def _response(payload, status_code=200): + response = unittest.mock.Mock() + response.status_code = status_code + response.json.return_value = payload + response.text = "" + return response + + +if __name__ == "__main__": + unittest.main()