Upload 5 files

Dockerfile

@@ -0,0 +1,17 @@
+FROM python:3.10-slim
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# 安装 Playwright 系统依赖和浏览器
+RUN playwright install-deps chromium && playwright install chromium
+
+COPY app.py register.py .
+
+RUN mkdir -p /app/tokens
+
+EXPOSE 7860
+
+CMD ["python", "app.py"]

README.md

@@ -1,10 +1,13 @@
----
-title: Try1
-emoji: 👁
-colorFrom: yellow
-colorTo: pink
-sdk: docker
-pinned: false
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+---
+title: OpenAI 自动注册器
+emoji: 🤖
+colorFrom: blue
+colorTo: green
+sdk: docker
+app_file: app.py
+pinned: false
+---
+
+# OpenAI 自动注册系统
+
+使用 Playwright 自动化注册 OpenAI 账号，支持代理。点击按钮开始注册，成功后会显示邮箱和 Account ID，并保存 Token 文件。

app.py

@@ -0,0 +1,221 @@
+import gradio as gr
+import asyncio
+import sys
+import json
+import random
+from pathlib import Path
+import zipfile
+from datetime import datetime
+
+TOKEN_DIR = Path("/app/tokens")
+TOKEN_DIR.mkdir(exist_ok=True)
+ACCOUNTS_FILE = TOKEN_DIR / "accounts.txt"
+
+# 全局状态
+current_process = None
+
+async def run_registration(proxy: str, loop_mode: bool = False, progress=gr.Progress()):
+    """
+    执行注册，loop_mode=False 表示单次（带 --once），loop_mode=True 表示无限循环（不带 --once）
+    返回实时日志的异步生成器
+    """
+    global current_process
+    cmd = [sys.executable, "/app/register.py"]
+    if not loop_mode:
+        cmd.append("--once")
+    if proxy and proxy.strip():
+        cmd.extend(["--proxy", proxy.strip()])
+
+    process = await asyncio.create_subprocess_exec(
+        *cmd,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.STDOUT,
+    )
+    current_process = process
+
+    lines = []
+    async for line in process.stdout:
+        decoded = line.decode('utf-8', errors='replace').rstrip()
+        lines.append(decoded)
+        if len(lines) % 10 == 0:
+            progress(0.5, desc="运行中...")
+        yield "\n".join(lines)
+
+    await process.wait()
+    current_process = None
+    progress(1.0, desc="进程结束")
+    yield "\n".join(lines) + "\n\n✅ 进程已结束。"
+
+def stop_registration():
+    global current_process
+    if current_process and current_process.returncode is None:
+        current_process.terminate()
+        return "⏹️ 已发送停止信号"
+    return "没有正在运行的进程"
+
+def list_token_files():
+    files = list(TOKEN_DIR.glob("token_*.json"))
+    return [str(f) for f in files]
+
+def get_accounts_content():
+    if ACCOUNTS_FILE.exists():
+        return ACCOUNTS_FILE.read_text(encoding='utf-8')
+    return "暂无账号信息"
+
+def download_accounts():
+    if ACCOUNTS_FILE.exists():
+        return str(ACCOUNTS_FILE)
+    return None
+
+def download_all_files():
+    token_files = list(TOKEN_DIR.glob("token_*.json"))
+    if not token_files and not ACCOUNTS_FILE.exists():
+        return None
+
+    zip_path = TOKEN_DIR / f"all_files_{datetime.now().strftime('%Y%m%d_%H%M%S')}.zip"
+    with zipfile.ZipFile(zip_path, 'w') as zipf:
+        for f in token_files:
+            zipf.write(f, arcname=f.name)
+        if ACCOUNTS_FILE.exists():
+            zipf.write(ACCOUNTS_FILE, arcname="accounts.txt")
+    return str(zip_path)
+
+def download_latest():
+    files = list(TOKEN_DIR.glob("token_*.json"))
+    if files:
+        latest = max(files, key=lambda p: p.stat().st_mtime)
+        return latest
+    return None
+
+def update_file_list():
+    files = list_token_files()
+    count = len(files)
+    return gr.update(value=files), gr.update(value=count)
+
+# 构建 Gradio 界面
+with gr.Blocks(title="OpenAI 自动注册 (双模式版)") as demo:
+    gr.Markdown("""
+    # 🤖 OpenAI 自动注册 (双模式版)
+    - **单次注册**：执行一次注册，完成后停止。
+    - **无限循环**：启动后 `register.py` 自己无限循环注册，可随时停止。
+    - 所有生成的 token 文件及 `accounts.txt` 均可单独下载或打包下载全部。
+    """)
+
+    with gr.Row():
+        proxy_input = gr.Textbox(
+            label="代理地址 (可选)",
+            placeholder="例如 http://127.0.0.1:7890"
+        )
+
+    with gr.Row():
+        single_btn = gr.Button("▶️ 单次注册", variant="primary")
+        loop_btn = gr.Button("🔄 无限循环", variant="primary")
+        stop_btn = gr.Button("⏹️ 停止", variant="secondary")
+
+    output = gr.Textbox(label="实时日志", lines=20, interactive=False)
+    progress_bar = gr.Progress()
+
+    gr.Markdown("---\n### 📁 生成的 Token 文件")
+    with gr.Row():
+        file_count = gr.Number(value=0, label="当前 Token 数量", interactive=False)
+        file_list = gr.Files(label="所有 Token 文件", file_count="multiple")
+    with gr.Row():
+        refresh_btn = gr.Button("🔄 刷新文件列表")
+        download_latest_btn = gr.Button("📥 下载最新 Token")
+        download_all_btn = gr.Button("📦 下载全部文件")
+
+    gr.Markdown("---\n### 📋 账号信息 (accounts.txt)")
+    with gr.Row():
+        accounts_display = gr.Textbox(label="accounts.txt 内容", lines=10, interactive=False)
+        refresh_accounts_btn = gr.Button("🔄 刷新账号列表")
+        download_accounts_btn = gr.Button("📥 下载 accounts.txt")
+
+    # 单次注册
+    async def single_run(proxy):
+        yield gr.update(interactive=False), gr.update(interactive=False), gr.update(interactive=True), None
+        logs = ""
+        async for log in run_registration(proxy, loop_mode=False):
+            logs = log
+            yield gr.update(), gr.update(), gr.update(), logs
+        yield gr.update(interactive=True), gr.update(interactive=True), gr.update(interactive=False), logs
+
+    single_btn.click(
+        fn=single_run,
+        inputs=proxy_input,
+        outputs=[single_btn, loop_btn, stop_btn, output],
+        queue=True
+    ).then(
+        fn=update_file_list,
+        outputs=[file_list, file_count]
+    ).then(
+        fn=get_accounts_content,
+        outputs=accounts_display
+    )
+
+    # 无限循环
+    async def loop_run(proxy):
+        yield gr.update(interactive=False), gr.update(interactive=False), gr.update(interactive=True), None
+        logs = ""
+        async for log in run_registration(proxy, loop_mode=True):
+            logs = log
+            yield gr.update(), gr.update(), gr.update(), logs
+        yield gr.update(interactive=True), gr.update(interactive=True), gr.update(interactive=False), logs
+
+    loop_btn.click(
+        fn=loop_run,
+        inputs=proxy_input,
+        outputs=[single_btn, loop_btn, stop_btn, output],
+        queue=True
+    ).then(
+        fn=update_file_list,
+        outputs=[file_list, file_count]
+    ).then(
+        fn=get_accounts_content,
+        outputs=accounts_display
+    )
+
+    # 停止
+    def stop_fn():
+        msg = stop_registration()
+        return gr.update(interactive=False), msg
+
+    stop_btn.click(
+        fn=stop_fn,
+        outputs=[stop_btn, output]
+    ).then(
+        fn=update_file_list,
+        outputs=[file_list, file_count]
+    ).then(
+        fn=get_accounts_content,
+        outputs=accounts_display
+    )
+
+    # 刷新文件列表
+    refresh_btn.click(
+        fn=update_file_list,
+        outputs=[file_list, file_count]
+    )
+
+    download_latest_btn.click(
+        fn=download_latest,
+        outputs=gr.File(label="下载最新 Token")
+    )
+
+    download_all_btn.click(
+        fn=download_all_files,
+        outputs=gr.File(label="下载全部文件")
+    )
+
+    # 账号相关
+    refresh_accounts_btn.click(
+        fn=get_accounts_content,
+        outputs=accounts_display
+    )
+
+    download_accounts_btn.click(
+        fn=download_accounts,
+        outputs=gr.File(label="下载 accounts.txt")
+    )
+
+demo.queue()
+demo.launch(server_name="0.0.0.0", server_port=7860)

register.py

@@ -0,0 +1,700 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+OpenAI 自动注册脚本（异步最终版，含弹窗关闭）
+"""
+import json
+import os
+import re
+import sys
+import time
+import random
+import string
+import secrets
+import hashlib
+import base64
+import argparse
+import asyncio
+from pathlib import Path
+from datetime import datetime, timedelta
+from dataclasses import dataclass
+from typing import Any, Dict, Optional, List, Tuple
+
+import urllib.parse
+import urllib.request
+import urllib.error
+
+from curl_cffi import requests
+from playwright.async_api import async_playwright
+
+# ==========================================
+# 常量配置
+# ==========================================
+OUT_DIR = Path(__file__).parent.resolve()
+TOKEN_DIR = OUT_DIR / "tokens"
+TOKEN_DIR.mkdir(parents=True, exist_ok=True)
+
+AUTH_URL = "https://auth.openai.com/oauth/authorize"
+TOKEN_URL = "https://auth.openai.com/oauth/token"
+CLIENT_ID = "app_EMoamEEZ73f0CkXaXp7hrann"
+DEFAULT_REDIRECT_URI = "http://localhost:1455/auth/callback"
+DEFAULT_SCOPE = "openid email profile offline_access"
+
+# ==========================================
+# 辅助函数
+# ==========================================
+def _gen_password() -> str:
+    alphabet = string.ascii_letters + string.digits
+    special = "!@#$%^&*.-"
+    base = [
+        random.choice(string.ascii_lowercase),
+        random.choice(string.ascii_uppercase),
+        random.choice(string.digits),
+        random.choice(special),
+    ]
+    base += [random.choice(alphabet + special) for _ in range(12)]
+    random.shuffle(base)
+    return "".join(base)
+
+def _random_name() -> str:
+    letters = string.ascii_lowercase
+    n = random.randint(5, 9)
+    s = ''.join(random.choice(letters) for _ in range(n))
+    return s.capitalize()
+
+def _random_birthdate() -> str:
+    start = datetime(1970,1,1); end = datetime(1999,12,31)
+    delta = end - start
+    d = start + timedelta(days=random.randrange(delta.days + 1))
+    return d.strftime('%Y-%m-%d')
+
+def _b64url_no_pad(raw: bytes) -> str:
+    return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
+
+def _sha256_b64url_no_pad(s: str) -> str:
+    return _b64url_no_pad(hashlib.sha256(s.encode("ascii")).digest())
+
+def _random_state(nbytes: int = 16) -> str:
+    return secrets.token_urlsafe(nbytes)
+
+def _pkce_verifier() -> str:
+    return secrets.token_urlsafe(64)
+
+def _parse_callback_url(callback_url: str) -> Dict[str, Any]:
+    candidate = callback_url.strip()
+    if not candidate:
+        return {"code": "","state": "","error": "","error_description": ""}
+    if "://" not in candidate:
+        if candidate.startswith("?"):
+            candidate = f"http://localhost{candidate}"
+        elif any(ch in candidate for ch in "/?#") or ":" in candidate:
+            candidate = f"http://{candidate}"
+        elif "=" in candidate:
+            candidate = f"http://localhost/?{candidate}"
+    parsed = urllib.parse.urlparse(candidate)
+    query = urllib.parse.parse_qs(parsed.query, keep_blank_values=True)
+    fragment = urllib.parse.parse_qs(parsed.fragment, keep_blank_values=True)
+    for key, values in fragment.items():
+        if key not in query or not query[key] or not (query[key][0] or "").strip():
+            query[key] = values
+    def get1(k: str) -> str:
+        v = query.get(k, [""])
+        return (v[0] or "").strip()
+    code = get1("code"); state = get1("state")
+    error = get1("error"); error_description = get1("error_description")
+    if code and not state and "#" in code:
+        code, state = code.split("#",1)
+    if not error and error_description:
+        error, error_description = error_description, ""
+    return {"code": code,"state": state,"error": error,"error_description": error_description}
+
+def _jwt_claims_no_verify(id_token: str) -> Dict[str, Any]:
+    if not id_token or id_token.count(".") < 2:
+        return {}
+    payload_b64 = id_token.split(".")[1]
+    pad = "=" * ((4 - (len(payload_b64) % 4)) % 4)
+    try:
+        payload = base64.urlsafe_b64decode((payload_b64 + pad).encode("ascii"))
+        return json.loads(payload.decode("utf-8"))
+    except Exception:
+        return {}
+
+def _decode_jwt_segment(seg: str) -> Dict[str, Any]:
+    raw = (seg or "").strip()
+    if not raw: return {}
+    pad = "=" * ((4 - (len(raw) % 4)) % 4)
+    try:
+        decoded = base64.urlsafe_b64decode((raw + pad).encode("ascii"))
+        return json.loads(decoded.decode("utf-8"))
+    except Exception:
+        return {}
+
+def _to_int(v: Any) -> int:
+    try: return int(v)
+    except (TypeError, ValueError): return 0
+
+def _post_form(url: str, data: Dict[str, str], timeout: int = 30) -> Dict[str, Any]:
+    body = urllib.parse.urlencode(data).encode("utf-8")
+    req = urllib.request.Request(
+        url, data=body, method="POST",
+        headers={"Content-Type": "application/x-www-form-urlencoded","Accept": "application/json"},
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
+            raw = resp.read()
+            if resp.status != 200:
+                raise RuntimeError(f"token exchange failed: {resp.status}: {raw.decode('utf-8','replace')}")
+            return json.loads(raw.decode("utf-8"))
+    except urllib.error.HTTPError as exc:
+        raw = exc.read()
+        raise RuntimeError(f"token exchange failed: {exc.code}: {raw.decode('utf-8','replace')}") from exc
+
+@dataclass(frozen=True)
+class OAuthStart:
+    auth_url: str
+    state: str
+    code_verifier: str
+    redirect_uri: str
+
+def generate_oauth_url(*, redirect_uri: str = DEFAULT_REDIRECT_URI, scope: str = DEFAULT_SCOPE) -> OAuthStart:
+    state = _random_state()
+    code_verifier = _pkce_verifier()
+    code_challenge = _sha256_b64url_no_pad(code_verifier)
+    params = {
+        "client_id": CLIENT_ID,
+        "response_type": "code",
+        "redirect_uri": redirect_uri,
+        "scope": scope,
+        "state": state,
+        "code_challenge": code_challenge,
+        "code_challenge_method": "S256",
+        "prompt": "login",
+        "id_token_add_organizations": "true",
+        "codex_cli_simplified_flow": "true",
+    }
+    auth_url = f"{AUTH_URL}?{urllib.parse.urlencode(params)}"
+    return OAuthStart(auth_url=auth_url, state=state, code_verifier=code_verifier, redirect_uri=redirect_uri)
+
+def fetch_sentinel_token(*, flow: str, did: str, proxies: Any = None) -> Optional[str]:
+    try:
+        body = json.dumps({"p": "", "id": did, "flow": flow})
+        resp = requests.post(
+            "https://sentinel.openai.com/backend-api/sentinel/req",
+            headers={
+                "origin": "https://sentinel.openai.com",
+                "referer": "https://sentinel.openai.com/backend-api/sentinel/frame.html?sv=20260219f9f6",
+                "content-type": "text/plain;charset=UTF-8",
+            },
+            data=body,
+            proxies=proxies,
+            impersonate="chrome",
+            timeout=15,
+        )
+        if resp.status_code != 200:
+            print(f"[Error] Sentinel flow={flow} 状态码: {resp.status_code}")
+            try:
+                print(resp.text)
+            except Exception:
+                pass
+            return None
+        return resp.json().get("token")
+    except Exception as e:
+        print(f"[Error] Sentinel flow={flow} 获取失败: {e}")
+        return None
+
+def submit_callback_url(*, callback_url: str, expected_state: str, code_verifier: str, redirect_uri: str = DEFAULT_REDIRECT_URI) -> str:
+    cb = _parse_callback_url(callback_url)
+    if cb["error"]:
+        desc = cb["error_description"]
+        raise RuntimeError(f"oauth error: {cb['error']}: {desc}".strip())
+    if not cb["code"]:
+        raise ValueError("callback url missing ?code=")
+    if not cb["state"]:
+        raise ValueError("callback url missing ?state=")
+    if cb["state"] != expected_state:
+        raise ValueError("state mismatch")
+
+    token_resp = _post_form(
+        TOKEN_URL,
+        {
+            "grant_type": "authorization_code",
+            "client_id": CLIENT_ID,
+            "code": cb["code"],
+            "redirect_uri": redirect_uri,
+            "code_verifier": code_verifier,
+        },
+    )
+    access_token = (token_resp.get("access_token") or "").strip()
+    refresh_token = (token_resp.get("refresh_token") or "").strip()
+    id_token = (token_resp.get("id_token") or "").strip()
+    expires_in = _to_int(token_resp.get("expires_in"))
+
+    claims = _jwt_claims_no_verify(id_token)
+    email = str(claims.get("email") or "").strip()
+    auth_claims = claims.get("https://api.openai.com/auth") or {}
+    account_id = str(auth_claims.get("chatgpt_account_id") or "").strip()
+
+    now = int(time.time())
+    expired_rfc3339 = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(now + max(expires_in, 0)))
+    now_rfc3339 = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(now))
+
+    config = {
+        "id_token": id_token,
+        "access_token": access_token,
+        "refresh_token": refresh_token,
+        "account_id": account_id,
+        "last_refresh": now_rfc3339,
+        "email": email,
+        "type": "codex",
+        "expired": expired_rfc3339,
+    }
+    return json.dumps(config, ensure_ascii=False, separators=(",", ":"))
+
+# ==========================================
+# Playwright 邮箱功能（含弹窗关闭）
+# ==========================================
+async def generate_email_by_click(mail_page) -> Tuple[str, str]:
+    """关闭“我知道了”弹窗，然后点击生成按钮，返回 (email, token)"""
+    try:
+        close_btn = await mail_page.query_selector('button.announcement-close-btn, button:has-text("我知道了")')
+        if close_btn:
+            is_visible = await close_btn.is_visible()
+            if is_visible:
+                await close_btn.click()
+                print("[*] 已点击“我知道了”关闭按钮")
+                await asyncio.sleep(1)
+    except Exception as e:
+        print(f"[*] 关闭弹窗异常（可忽略）: {e}")
+
+    button_selectors = [
+        'button.action-btn',
+        'button:has-text("随机生成")',
+        'button:has-text("Generate")',
+        'button:has-text("刷新")',
+        'button:has-text("新邮箱")',
+        '.generate-btn',
+        '#generate'
+    ]
+    button = None
+    for selector in button_selectors:
+        try:
+            button = await mail_page.wait_for_selector(selector, timeout=5000)
+            print(f"找到按钮: {selector}")
+            break
+        except:
+            continue
+
+    if not button:
+        raise RuntimeError("未找到生成按钮，请检查页面元素")
+
+    async with mail_page.expect_response(lambda response: "/api/generate-email" in response.url and response.status == 200) as response_info:
+        await button.click()
+    response = await response_info.value
+    data = await response.json()
+    if data.get('success'):
+        email = data['data']['email']
+        token = data['auth']['token']
+        return email, token
+    else:
+        raise RuntimeError(f"生成邮箱失败: {data.get('error')}")
+
+async def playwright_fetch_code(mail_page, email: str, token: str, timeout: int = 180) -> str:
+    """在邮箱页面轮询获取验证码，返回验证码字符串（增强版）"""
+    try:
+        await mail_page.evaluate("""
+            if (typeof refreshInbox === 'function') {
+                refreshInbox();
+            }
+        """)
+        refresh_btn = await mail_page.query_selector('button:has-text("手动刷新")')
+        if refresh_btn:
+            await refresh_btn.click()
+            print("[*] 点击了手动刷新按钮")
+        await asyncio.sleep(2)
+    except Exception as e:
+        print(f"[*] 首次刷新异常: {e}")
+
+    start_time = time.time()
+    attempt = 0
+    while time.time() - start_time < timeout:
+        attempt += 1
+        try:
+            await mail_page.evaluate("refreshInbox?.()")
+            refresh_btn = await mail_page.query_selector('button:has-text("手动刷新")')
+            if refresh_btn:
+                await refresh_btn.click()
+        except:
+            pass
+
+        result = await mail_page.evaluate("""
+            async ([email, token]) => {
+                try {
+                    const response = await fetch(`/api/emails?email=${encodeURIComponent(email)}`, {
+                        method: 'GET',
+                        headers: {
+                            'Accept': 'application/json',
+                            'X-Inbox-Token': token
+                        }
+                    });
+                    if (!response.ok) {
+                        return { success: false, error: `HTTP ${response.status}` };
+                    }
+                    return await response.json();
+                } catch (error) {
+                    return { success: false, error: error.toString() };
+                }
+            }
+        """, [email, token])
+
+        if not result.get('success'):
+            print(f"[*] 第 {attempt} 次获取邮件列表失败: {result.get('error')}")
+            await asyncio.sleep(3)
+            continue
+
+        emails = result.get('data', {}).get('emails', [])
+        emails.sort(key=lambda x: x.get('timestamp', 0), reverse=True)
+        print(f"[*] 第 {attempt} 次获取到 {len(emails)} 封邮件")
+
+        for idx, mail in enumerate(emails):
+            subject = mail.get('subject', '')
+            content = mail.get('html_content', '') or mail.get('content', '') or mail.get('text', '')
+            print(f"   邮件 {idx+1} 主题: {subject[:50]}")
+            print(f"   内容预览: {content[:100].replace(chr(10), ' ')}")
+
+            patterns = [
+                (r'(?<!\d)(\d{6})(?!\d)', 1),
+                (r'代码为\s*(\d{6})', 1),
+                (r'verification code[:\s]*(\d{6})', 1),
+                (r'is[:\s]*(\d{6})', 1),
+                (r'<[^>]*>(\d)[^>]*>(\d)[^>]*>(\d)[^>]*>(\d)[^>]*>(\d)[^>]*>(\d)', 6),
+            ]
+            for pattern, group_count in patterns:
+                if group_count == 6:
+                    match = re.search(pattern, content, re.DOTALL)
+                    if match:
+                        code = ''.join(match.groups())
+                        if len(code) == 6 and code.isdigit():
+                            print(f"[✓] 从分隔数字中提取验证码: {code}")
+                            return code
+                else:
+                    match = re.search(pattern, subject + ' ' + content, re.IGNORECASE)
+                    if match:
+                        code = match.group(1)
+                        if len(code) == 6 and code.isdigit():
+                            print(f"[✓] 提取验证码: {code}")
+                            return code
+
+        print("[*] 未找到验证码，等待后重试...")
+        await asyncio.sleep(3)
+
+    return ""
+
+# ==========================================
+# 核心注册函数
+# ==========================================
+async def async_run_registration(proxy: Optional[str]) -> Optional[Tuple[str, str, str]]:
+    proxies = None
+    if proxy:
+        proxies = {"http": proxy, "https": proxy}
+
+    async with async_playwright() as p:
+        launch_args = [
+            '--disable-blink-features=AutomationControlled',
+            '--no-sandbox',
+            '--disable-gpu',
+            '--disable-dev-shm-usage',
+        ]
+        browser = await p.chromium.launch(headless=True, args=launch_args)
+        context_kwargs = {}
+        if proxy:
+            context_kwargs["proxy"] = {"server": proxy}
+        context = await browser.new_context(**context_kwargs)
+        mail_page = await context.new_page()
+
+        await mail_page.goto("https://mail.chatgpt.org.uk/", timeout=60000)
+        await mail_page.wait_for_load_state("networkidle")
+
+        email, token = await generate_email_by_click(mail_page)
+        print(f"[*] 成功获取临时邮箱: {email}")
+
+        s = requests.Session(proxies=proxies, impersonate="chrome")
+        s.headers.update({"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/145.0.0.0 Safari/537.36"})
+
+        try:
+            trace = s.get("https://cloudflare.com/cdn-cgi/trace", timeout=10)
+            loc_re = re.search(r"^loc=(.+)$", trace.text, re.MULTILINE)
+            loc = loc_re.group(1) if loc_re else None
+            print(f"[*] 当前 IP 所在地: {loc}")
+            if loc in ("CN", "HK"):
+                raise RuntimeError("代理所在地不支持 CN/HK")
+        except Exception as e:
+            print(f"[Error] 网络连接检查失败: {e}")
+            await browser.close()
+            return None
+
+        oauth = generate_oauth_url()
+        url = oauth.auth_url
+
+        try:
+            resp = s.get(url, timeout=15)
+            did = s.cookies.get("oai-did")
+            print(f"[*] Device ID: {did}")
+
+            signup_body = json.dumps({"username": {"value": email, "kind": "email"}, "screen_hint": "signup"})
+            sen_req_body = json.dumps({"p": "", "id": did, "flow": "authorize_continue"})
+
+            sen_resp = requests.post(
+                "https://sentinel.openai.com/backend-api/sentinel/req",
+                headers={
+                    "origin": "https://sentinel.openai.com",
+                    "referer": "https://sentinel.openai.com/backend-api/sentinel/frame.html?sv=20260219f9f6",
+                    "content-type": "text/plain;charset=UTF-8",
+                    "user-agent": s.headers["user-agent"],
+                },
+                data=sen_req_body,
+                proxies=proxies,
+                impersonate="chrome",
+                timeout=15,
+            )
+            print(f"[*] sentinel authorize_continue 状态: {sen_resp.status_code}")
+            if sen_resp.status_code != 200:
+                print(f"[Error] Sentinel 异常拦截")
+                await browser.close()
+                return None
+
+            sen_token = sen_resp.json().get("token")
+            sentinel = json.dumps({"p": "", "t": "", "c": sen_token, "id": did, "flow": "authorize_continue"}) if sen_token else None
+
+            so_token = fetch_sentinel_token(flow="oauth_create_account", did=did, proxies=proxies)
+            print(f"[*] sentinel oauth_create_account token: {bool(so_token)}")
+
+            signup_headers = {
+                "referer": "https://auth.openai.com/create-account",
+                "accept": "application/json",
+                "content-type": "application/json",
+            }
+            if sentinel:
+                signup_headers["openai-sentinel-token"] = sentinel
+            signup_resp = s.post(
+                "https://auth.openai.com/api/accounts/authorize/continue",
+                headers=signup_headers,
+                data=signup_body,
+            )
+            print(f"[*] authorize/continue 状态: {signup_resp.status_code}")
+            if signup_resp.status_code != 200:
+                print("[Error] 注册表单提交失败")
+                await browser.close()
+                return None
+
+            password = _gen_password()
+            register_headers = {
+                "referer": "https://auth.openai.com/create-account/password",
+                "accept": "application/json",
+                "content-type": "application/json",
+            }
+            if sentinel:
+                register_headers["openai-sentinel-token"] = sentinel
+            reg_resp = s.post(
+                "https://auth.openai.com/api/accounts/user/register",
+                headers=register_headers,
+                data=json.dumps({"password": password, "username": email}),
+            )
+            print(f"[*] 设置密码状态: {reg_resp.status_code}")
+            if reg_resp.status_code != 200:
+                print("[Error] 设置密码失败")
+                await browser.close()
+                return None
+
+            for attempt in range(3):
+                try:
+                    send_headers = {
+                        "referer": "https://auth.openai.com/create-account/password",
+                        "accept": "application/json",
+                    }
+                    send_resp = s.get(
+                        "https://auth.openai.com/api/accounts/email-otp/send",
+                        headers=send_headers,
+                        timeout=30,
+                    )
+                    print(f"[*] 触发发送验证码状态: {send_resp.status_code}")
+                    if send_resp.status_code == 200:
+                        break
+                except Exception as e:
+                    print(f"[Warn] send 调用异常 (尝试 {attempt+1}/3): {e}")
+                    await asyncio.sleep(2)
+            else:
+                print("[Error] 触发验证码发送失败，重试耗尽")
+                await browser.close()
+                return None
+
+            code = await playwright_fetch_code(mail_page, email, token, timeout=180)
+            if not code:
+                print("[Error] 未能收到验证码")
+                await browser.close()
+                return None
+            print(f"[*] 收到验证码: {code}")
+
+            code_body = json.dumps({"code": code})
+            validate_headers = {
+                "referer": "https://auth.openai.com/email-verification",
+                "accept": "application/json",
+                "content-type": "application/json",
+            }
+            if sentinel:
+                validate_headers["openai-sentinel-token"] = sentinel
+
+            code_resp = s.post(
+                "https://auth.openai.com/api/accounts/email-otp/validate",
+                headers=validate_headers,
+                data=code_body,
+            )
+            print(f"[*] 验证码校验状态: {code_resp.status_code}")
+            if code_resp.status_code != 200:
+                print("[Error] 验证码校验失败")
+                await browser.close()
+                return None
+
+            create_account_body = json.dumps({"name": _random_name(), "birthdate": _random_birthdate()})
+            create_headers = {
+                "referer": "https://auth.openai.com/about-you",
+                "accept": "application/json",
+                "content-type": "application/json",
+            }
+            if so_token:
+                create_headers["openai-sentinel-so-token"] = so_token
+
+            create_account_resp = s.post(
+                "https://auth.openai.com/api/accounts/create_account",
+                headers=create_headers,
+                data=create_account_body,
+            )
+            print(f"[*] 账户创建状态: {create_account_resp.status_code}")
+            if create_account_resp.status_code != 200:
+                print("[Error] 账户创建失败")
+                await browser.close()
+                return None
+
+            auth_cookie = s.cookies.get("oai-client-auth-session")
+            if not auth_cookie:
+                print("[Error] 未能获取到授权 Cookie")
+                await browser.close()
+                return None
+
+            auth_json = _decode_jwt_segment(auth_cookie.split(".")[0])
+            workspaces = auth_json.get("workspaces") or []
+            if not workspaces:
+                print("[Error] 授权 Cookie 里没有 workspace 信息")
+                await browser.close()
+                return None
+            workspace_id = str((workspaces[0] or {}).get("id") or "").strip()
+            if not workspace_id:
+                print("[Error] 无法解析 workspace_id")
+                await browser.close()
+                return None
+
+            select_body = json.dumps({"workspace_id": workspace_id})
+            select_resp = s.post(
+                "https://auth.openai.com/api/accounts/workspace/select",
+                headers={
+                    "referer": "https://auth.openai.com/sign-in-with-chatgpt/codex/consent",
+                    "content-type": "application/json",
+                },
+                data=select_body,
+            )
+            if select_resp.status_code != 200:
+                print(f"[Error] 选择 workspace 失败")
+                await browser.close()
+                return None
+
+            continue_url = str((select_resp.json() or {}).get("continue_url") or "").strip()
+            if not continue_url:
+                print("[Error] workspace/select 响应缺少 continue_url")
+                await browser.close()
+                return None
+
+            current_url = continue_url
+            for _ in range(6):
+                final_resp = s.get(current_url, allow_redirects=False, timeout=15)
+                location = final_resp.headers.get("Location") or ""
+                if final_resp.status_code not in [301, 302, 303, 307, 308]:
+                    break
+                if not location:
+                    break
+                next_url = urllib.parse.urljoin(current_url, location)
+                if "code=" in next_url and "state=" in next_url:
+                    token_json = submit_callback_url(
+                        callback_url=next_url,
+                        code_verifier=oauth.code_verifier,
+                        redirect_uri=oauth.redirect_uri,
+                        expected_state=oauth.state,
+                    )
+                    await browser.close()
+                    return token_json, email, password
+                current_url = next_url
+
+            print("[Error] 未能在重定向链中捕获到最终 Callback URL")
+            await browser.close()
+            return None
+
+        except Exception as e:
+            print(f"[Error] 运行时发生错误: {e}")
+            await browser.close()
+            return None
+
+# ==========================================
+# 主函数
+# ==========================================
+async def main():
+    parser = argparse.ArgumentParser(description="OpenAI 自动注册脚本（异步最终版）")
+    parser.add_argument("--proxy", default=None, help="代理地址，如 http://127.0.0.1:7890")
+    parser.add_argument("--once", action="store_true", help="只运行一次")
+    parser.add_argument("--sleep-min", type=int, default=5, help="循环模式最短等待秒数")
+    parser.add_argument("--sleep-max", type=int, default=30, help="循环模式最长等待秒数")
+    args = parser.parse_args()
+
+    sleep_min = max(1, args.sleep_min)
+    sleep_max = max(sleep_min, args.sleep_max)
+
+    count = 0
+    print("[Info] OpenAI 自动注册脚本 (异步最终版) 启动")
+    OUT_DIR.mkdir(parents=True, exist_ok=True)
+
+    while True:
+        count += 1
+        print(f"\n[{datetime.now().strftime('%H:%M:%S')}] >>> 开始第 {count} 次注册流程 <<<")
+        try:
+            result = await async_run_registration(args.proxy)
+            if result:
+                token_json, email, password = result
+                try:
+                    t_data = json.loads(token_json)
+                    fname_email = t_data.get("email", "unknown").replace("@", "_")
+                except Exception:
+                    fname_email = "unknown"
+
+                file_path = TOKEN_DIR / f"token_{fname_email}_{int(time.time())}.json"
+                try:
+                    file_path.write_text(token_json, encoding="utf-8")
+                    print(f"[*] 成功! Token 已保存至: {file_path}")
+                except Exception as e:
+                    print(f"[Error] 保存 token 失败: {e}")
+
+                acc_file = TOKEN_DIR / "accounts.txt"
+                try:
+                    with open(acc_file, "a", encoding="utf-8") as f:
+                        f.write(f"{email}----{password}\n")
+                except Exception as e:
+                    print(f"[Error] 保存账号信息失败: {e}")
+            else:
+                print("[-] 本次注册失败。")
+        except Exception as e:
+            print(f"[Error] 发生未捕获异常: {e}")
+
+        if args.once:
+            break
+        wait_time = random.randint(sleep_min, sleep_max)
+        print(f"[*] 休息 {wait_time} 秒...")
+        await asyncio.sleep(wait_time)
+
+if __name__ == "__main__":
+    asyncio.run(main())

requirements.txt

@@ -0,0 +1,3 @@
+gradio
+playwright
+curl_cffi