Merge branch 'main' of https://github.com/zc-zhangchen/any-auto-register

README.md

@@ -130,6 +130,7 @@
 | TempMail.lol | `tempmail_lol` | 临时邮箱方案，部分地区可能需要代理 |
 | SkyMail (CloudMail) | `skymail` | 通过 API / Token / 域名使用 |
 | YYDS Mail / MaliAPI | `maliapi` | 支持域名与自动域名策略 |
+| GPTMail | `gptmail` | 基于 GPTMail API 生成临时邮箱并轮询邮件，支持已知域名时本地拼装随机地址 |
 | DuckMail | `duckmail` | 临时邮箱方案 |
 | Freemail | `freemail` | 自建邮箱服务 |
 | Laoudo | `laoudo` | 固定邮箱方案 |

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，未设置的返回空字符串

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 自建临时邮箱服务"""
 

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,

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() {
               </Form.Item>
             </>
           )}
+          {mailProvider === 'gptmail' && (
+            <>
+              <Form.Item name="gptmail_base_url" label="API URL">
+                <Input placeholder="https://mail.chatgpt.org.uk" />
+              </Form.Item>
+              <Form.Item name="gptmail_api_key" label="API Key">
+                <Input.Password placeholder="gpt-test" />
+              </Form.Item>
+              <Form.Item
+                name="gptmail_domain"
+                label="邮箱域名（可选）"
+                extra="已知当前可用域名时可直接本地拼装随机地址，省掉一次 generate-email 请求"
+              >
+                <Input placeholder="example.com" />
+              </Form.Item>
+            </>
+          )}
           {mailProvider === 'cfworker' && (
             <>
               <Form.Item name="cfworker_api_url" label="API URL">

frontend/src/pages/Settings.tsx

@@ -25,6 +25,7 @@ const SELECT_FIELDS: Record<string, { label: string; value: string }[]> = {
     { 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'
       }

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()