chaos
767516745e
readlink/realpath of /snap/bin/chromium becomes /usr/bin/snap, which is not a browser and breaks DrissionPage launch. Keep the snap wrapper path, reject snap host basenames, and prefer real deb chromium binaries.
862 lines
34 KiB
Python
Executable File
862 lines
34 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
自动注册脚本 v2（Cloudflare catch-all + Gmail IMAP）
|
||
流程：生成随机别名 -> Chromium注册 -> 直接IMAP取验证码 -> 填码完成注册
|
||
"""
|
||
import argparse
|
||
import email as email_lib
|
||
import imaplib
|
||
import json
|
||
import os
|
||
import random
|
||
import re
|
||
import string
|
||
import sys
|
||
import time
|
||
from datetime import datetime, timezone
|
||
from email.header import decode_header as _decode_header
|
||
from email.utils import parsedate_to_datetime
|
||
from pathlib import Path
|
||
|
||
from DrissionPage import Chromium, ChromiumOptions
|
||
|
||
# ---- 配置 ----
|
||
BASE_URL = "https://accounts.x.ai"
|
||
SIGNUP_URL = "https://accounts.x.ai/sign-up?redirect=grok-com"
|
||
ACCOUNTS_FILE = "accounts_cli.txt"
|
||
CPA_DIR = "cpa_auths"
|
||
DOMAIN = "zhangyunuo.net"
|
||
CODE_TIMEOUT = 180
|
||
CODE_POLL = 3
|
||
IMAP_HOST = "imap.gmail.com"
|
||
IMAP_PORT = 993
|
||
EXTENSION_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "turnstilePatch")
|
||
|
||
CHROMIUM_SLIM_FLAGS = [
|
||
"--disable-gpu",
|
||
"--disable-software-rasterizer",
|
||
"--no-sandbox",
|
||
"--disable-dev-shm-usage",
|
||
"--disable-images",
|
||
"--mute-audio",
|
||
"--disable-background-networking",
|
||
"--no-first-run",
|
||
]
|
||
|
||
CODE_PATTERNS = [
|
||
re.compile(r"verification\s+code[:\s]+(\d{4,8})", re.I),
|
||
re.compile(r"your\s+code[:\s]+(\d{4,8})", re.I),
|
||
re.compile(r"confirm(?:ation)?\s+code[:\s]+(\d{4,8})", re.I),
|
||
re.compile(r"code[:\s]+(\d{4,8})", re.I),
|
||
re.compile(r"验证码[：:\\s]+(\d{4,8})"),
|
||
]
|
||
|
||
def load_env(path=".env"):
|
||
data = {}
|
||
if not os.path.exists(path):
|
||
return data
|
||
with open(path) as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
if "=" in line:
|
||
k, v = line.split("=", 1)
|
||
data[k.strip()] = v.strip().strip('"').strip("'")
|
||
return data
|
||
|
||
def gen_alias(domain: str, length: int = 10) -> str:
|
||
alias = "".join(random.choices(string.ascii_lowercase + string.digits, k=length))
|
||
return f"{alias}@{domain}"
|
||
|
||
def gen_password(length: int = 16) -> str:
|
||
# Avoid form-hostile punctuation; keep strong enough for xAI signup rules.
|
||
lower = string.ascii_lowercase
|
||
upper = string.ascii_uppercase
|
||
digits = string.digits
|
||
special = "!@#$%^&*"
|
||
if length < 8:
|
||
length = 8
|
||
base = [
|
||
random.choice(lower),
|
||
random.choice(upper),
|
||
random.choice(digits),
|
||
random.choice(special),
|
||
]
|
||
pool = lower + upper + digits + special
|
||
base += random.choices(pool, k=length - len(base))
|
||
random.shuffle(base)
|
||
return "".join(base)
|
||
|
||
def decode_header(raw: str) -> str:
|
||
parts = _decode_header(raw or "")
|
||
out = []
|
||
for part, charset in parts:
|
||
if isinstance(part, bytes):
|
||
out.append(part.decode(charset or "utf-8", errors="replace"))
|
||
else:
|
||
out.append(part)
|
||
return "".join(out)
|
||
|
||
def message_body(msg) -> str:
|
||
body = ""
|
||
if msg.is_multipart():
|
||
for part in msg.walk():
|
||
ct = part.get_content_type()
|
||
if ct == "text/plain":
|
||
payload = part.get_payload(decode=True)
|
||
if payload:
|
||
body += payload.decode("utf-8", errors="replace")
|
||
else:
|
||
payload = msg.get_payload(decode=True)
|
||
if payload:
|
||
body += payload.decode("utf-8", errors="replace")
|
||
return body
|
||
|
||
def extract_code(text: str) -> str | None:
|
||
for pat in CODE_PATTERNS:
|
||
m = pat.search(text)
|
||
if m:
|
||
return m.group(1)
|
||
return None
|
||
|
||
def imap_login(user: str, password: str):
|
||
imap = imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT, timeout=45)
|
||
imap.login(user, password)
|
||
imap.select("INBOX")
|
||
return imap
|
||
|
||
def fetch_code(imap, target_email: str, timeout: int = CODE_TIMEOUT, poll: int = CODE_POLL) -> str | None:
|
||
deadline = time.time() + timeout
|
||
target_lower = target_email.lower().strip()
|
||
while time.time() < deadline:
|
||
try:
|
||
status, data = imap.search(None, "ALL")
|
||
if status != "OK" or not data or not data[0]:
|
||
time.sleep(poll)
|
||
continue
|
||
msg_ids = data[0].split()[-30:]
|
||
filter_ts = int((time.time() - 900) * 1000)
|
||
for mid in reversed(msg_ids):
|
||
try:
|
||
_, msg_data = imap.fetch(mid, "(RFC822)")
|
||
except Exception:
|
||
continue
|
||
if not msg_data or not msg_data[0] or not isinstance(msg_data[0][1], bytes):
|
||
continue
|
||
msg = email_lib.message_from_bytes(msg_data[0][1])
|
||
date_str = msg.get("Date")
|
||
if date_str:
|
||
try:
|
||
dt = parsedate_to_datetime(date_str)
|
||
if dt.tzinfo is None:
|
||
dt = dt.replace(tzinfo=timezone.utc)
|
||
if int(dt.timestamp() * 1000) < filter_ts:
|
||
continue
|
||
except Exception:
|
||
pass
|
||
subject = decode_header(msg.get("Subject", ""))
|
||
sender = decode_header(msg.get("From", ""))
|
||
recipient_blob = " ".join(
|
||
decode_header(msg.get(h, ""))
|
||
for h in ("To", "Cc", "Delivered-To", "X-Original-To", "Original-Recipient", "Envelope-To")
|
||
).lower()
|
||
if target_lower not in recipient_blob:
|
||
continue
|
||
body = message_body(msg)
|
||
combined = f"{subject}\n{sender}\n{recipient_blob}\n{body}".lower()
|
||
if not any(kw in combined for kw in ["x.ai", "xai", "grok", "verification", "code", "confirm", "验证码", "确认"]):
|
||
continue
|
||
code = extract_code(combined)
|
||
if code:
|
||
print(f"[+] Code found for {target_email}: {code}")
|
||
return code
|
||
except Exception as e:
|
||
print(f"[!] IMAP poll error: {e}")
|
||
time.sleep(poll)
|
||
return None
|
||
|
||
def save_account(email: str, password: str, sso: str = ""):
|
||
Path(ACCOUNTS_FILE).parent.mkdir(parents=True, exist_ok=True)
|
||
with open(ACCOUNTS_FILE, "a") as f:
|
||
f.write(f"{email}----{password}----{sso}\n")
|
||
print(f"[+] Saved: {email}")
|
||
|
||
def _normalize_browser_path(path: str | None) -> str | None:
|
||
"""Keep /snap/bin/chromium; never collapse to /usr/bin/snap."""
|
||
if not path:
|
||
return None
|
||
path = str(path).strip()
|
||
if not path:
|
||
return None
|
||
if not (os.path.isfile(path) or os.path.islink(path) or os.path.exists(path)):
|
||
return None
|
||
base = os.path.basename(path)
|
||
if base in ("snap", "snap-confine", "snapd", "env", "bash", "sh"):
|
||
return None
|
||
low = path.replace("\\", "/").lower()
|
||
if "chromium" not in low and "chrome" not in low:
|
||
return None
|
||
if path.startswith("/snap/bin/"):
|
||
return path
|
||
try:
|
||
real = os.path.realpath(path)
|
||
if os.path.basename(real) in ("snap", "snap-confine", "snapd"):
|
||
return path if ("chromium" in low or "chrome" in low) else None
|
||
if "chromium" in real.lower() or "chrome" in real.lower():
|
||
return real
|
||
except Exception:
|
||
pass
|
||
return path
|
||
|
||
|
||
def resolve_browser_path() -> str | None:
|
||
# Prefer Chromium: Google Chrome blocks --load-extension (turnstilePatch).
|
||
# Never use realpath of /snap/bin/chromium → /usr/bin/snap (breaks launch).
|
||
for cand in (
|
||
"/usr/bin/chromium",
|
||
"/usr/bin/chromium-browser",
|
||
"/usr/lib/chromium/chromium",
|
||
"/snap/bin/chromium",
|
||
"/usr/bin/google-chrome-stable",
|
||
"/usr/bin/google-chrome",
|
||
):
|
||
norm = _normalize_browser_path(cand)
|
||
if norm:
|
||
return norm
|
||
return None
|
||
|
||
|
||
def has_display() -> bool:
|
||
"""True if Chromium can open a real GUI window (Windows always; else DISPLAY/WAYLAND)."""
|
||
if sys.platform == "win32":
|
||
return True
|
||
return bool((os.environ.get("DISPLAY") or "").strip() or (os.environ.get("WAYLAND_DISPLAY") or "").strip())
|
||
|
||
|
||
def create_browser_options(headless: bool = False, proxy: str = "") -> ChromiumOptions:
|
||
opts = ChromiumOptions()
|
||
opts.auto_port()
|
||
for flag in CHROMIUM_SLIM_FLAGS:
|
||
opts.set_argument(flag)
|
||
browser_path = resolve_browser_path()
|
||
if browser_path:
|
||
try:
|
||
opts.set_browser_path(browser_path)
|
||
except Exception:
|
||
pass
|
||
if os.path.exists(EXTENSION_PATH):
|
||
is_chrome = bool(
|
||
browser_path
|
||
and "chrome" in os.path.basename(browser_path)
|
||
and "chromium" not in browser_path
|
||
)
|
||
if is_chrome:
|
||
print("[!] skip turnstilePatch: Google Chrome disallows --load-extension; use Chromium")
|
||
else:
|
||
opts.add_extension(EXTENSION_PATH)
|
||
# Auto headless when no DISPLAY (unless caller already requested headless)
|
||
use_headless = bool(headless) or not has_display()
|
||
if use_headless:
|
||
try:
|
||
opts.headless()
|
||
except Exception:
|
||
opts.set_argument("--headless=new")
|
||
for flag in ("--headless=new", "--disable-gpu", "--no-sandbox", "--disable-dev-shm-usage"):
|
||
try:
|
||
opts.set_argument(flag)
|
||
except Exception:
|
||
pass
|
||
if not headless and not has_display():
|
||
print(
|
||
f"[*] 无图形界面 (DISPLAY={os.environ.get('DISPLAY', '')!r})，已自动启用 headless。\n"
|
||
" 推荐: xvfb-run -a python auto_register.py ... （过 CF 更稳）",
|
||
flush=True,
|
||
)
|
||
else:
|
||
print("[*] browser headless=True", flush=True)
|
||
if proxy:
|
||
try:
|
||
from urllib.parse import urlparse
|
||
u = urlparse(proxy if "://" in proxy else f"http://{proxy}")
|
||
host = u.hostname or ""
|
||
if host:
|
||
port = u.port or (443 if (u.scheme or "http") == "https" else 80)
|
||
scheme = u.scheme or "http"
|
||
opts.set_argument(f"--proxy-server={scheme}://{host}:{port}")
|
||
except Exception as e:
|
||
print(f"[!] Proxy setup failed: {e}")
|
||
return opts
|
||
|
||
def wait_for_tab(browser: Chromium, timeout: int = 15) -> any:
|
||
"""Wait for browser tab to be ready and return it."""
|
||
deadline = time.time() + timeout
|
||
while time.time() < deadline:
|
||
try:
|
||
tab_ids = browser.tab_ids
|
||
if tab_ids:
|
||
tab = browser.get_tab(tab_ids[0])
|
||
if tab is not None:
|
||
return tab
|
||
except Exception:
|
||
pass
|
||
time.sleep(0.5)
|
||
# Fallback
|
||
return browser.latest_tab
|
||
|
||
def bypass_turnstile(page) -> None:
|
||
"""Try to bypass Cloudflare Turnstile using multiple methods."""
|
||
try:
|
||
# Method 1: Check if Turnstile token already exists
|
||
token = page.run_js("""
|
||
try {
|
||
const byInput = String((document.querySelector('input[name="cf-turnstile-response"]') || {}).value || '').trim();
|
||
if (byInput) return byInput;
|
||
if (window.turnstile && typeof turnstile.getResponse === 'function') {
|
||
return String(turnstile.getResponse() || '').trim();
|
||
}
|
||
return '';
|
||
} catch(e) { return ''; }
|
||
""")
|
||
if token and len(token) >= 80:
|
||
print(f"[+] Turnstile token already present (len={len(token)})")
|
||
return
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
# Method 2: Try to find and click Turnstile checkbox
|
||
challenge_input = page.ele("@name=cf-turnstile-response", timeout=2)
|
||
if challenge_input:
|
||
try:
|
||
wrapper = challenge_input.parent()
|
||
iframe = None
|
||
try:
|
||
iframe = wrapper.shadow_root.ele("tag:iframe")
|
||
except Exception:
|
||
pass
|
||
if iframe:
|
||
try:
|
||
iframe.run_js("""
|
||
window.dtp = 1;
|
||
function getRandomInt(min, max) { return Math.floor(Math.random() * (max - min + 1)) + min; }
|
||
let sx = getRandomInt(800, 1200);
|
||
let sy = getRandomInt(400, 700);
|
||
Object.defineProperty(MouseEvent.prototype, 'screenX', { value: sx });
|
||
Object.defineProperty(MouseEvent.prototype, 'screenY', { value: sy });
|
||
""")
|
||
except Exception:
|
||
pass
|
||
try:
|
||
body_sr = iframe.ele("tag:body").shadow_root
|
||
btn = body_sr.ele("tag:input")
|
||
if btn:
|
||
btn.click()
|
||
print("[+] Clicked Turnstile checkbox via shadow DOM")
|
||
except Exception:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
# Method 3: Click any visible Turnstile container
|
||
page.run_js("""
|
||
const nodes = Array.from(document.querySelectorAll('div,span,iframe')).filter((n) => {
|
||
const txt = (n.className || '') + ' ' + (n.id || '') + ' ' + (n.getAttribute?.('src') || '');
|
||
return String(txt).toLowerCase().includes('turnstile');
|
||
});
|
||
if (nodes.length && typeof nodes[0].click === 'function') nodes[0].click();
|
||
""")
|
||
except Exception:
|
||
pass
|
||
|
||
def click_email_signup_button(page, timeout: int = 15) -> bool:
|
||
"""Click the 'Sign up with email' button on the landing page."""
|
||
deadline = time.time() + timeout
|
||
while time.time() < deadline:
|
||
clicked = page.run_js("""
|
||
const candidates = Array.from(document.querySelectorAll('button, a, [role="button"]'));
|
||
const target = candidates.find((node) => {
|
||
const text = (node.innerText || node.textContent || '').replace(/\\s+/g, '');
|
||
const lower = text.toLowerCase();
|
||
return (
|
||
text.includes('使用邮箱注册') ||
|
||
lower.includes('signupwithemail') ||
|
||
lower.includes('continuewithemail') ||
|
||
lower.includes('email')
|
||
);
|
||
});
|
||
if (!target) return false;
|
||
target.click();
|
||
return true;
|
||
""")
|
||
if clicked:
|
||
print("[+] Clicked 'Sign up with email' button")
|
||
return True
|
||
time.sleep(1)
|
||
return False
|
||
|
||
def fill_email_js(page, email: str) -> bool:
|
||
"""Fill email input using React-compatible JS injection."""
|
||
deadline = time.time() + 15
|
||
while time.time() < deadline:
|
||
result = page.run_js("""
|
||
function isVisible(node) {
|
||
if (!node) return false;
|
||
const style = window.getComputedStyle(node);
|
||
if (style.display === 'none' || style.visibility === 'hidden' || style.opacity === '0') return false;
|
||
const rect = node.getBoundingClientRect();
|
||
return rect.width > 0 && rect.height > 0;
|
||
}
|
||
const input = Array.from(document.querySelectorAll('input[data-testid="email"], input[name="email"], input[type="email"], input[autocomplete="email"]')).find((node) => isVisible(node) && !node.disabled && !node.readOnly) || null;
|
||
if (!input) return 'not-ready';
|
||
input.focus(); input.click();
|
||
const valueSetter = Object.getOwnPropertyDescriptor(HTMLInputElement.prototype, 'value')?.set;
|
||
const tracker = input._valueTracker;
|
||
if (tracker) tracker.setValue(arguments[0]);
|
||
if (valueSetter) valueSetter.call(input, arguments[0]); else input.value = arguments[0];
|
||
input.dispatchEvent(new Event('focus', { bubbles: true }));
|
||
input.dispatchEvent(new InputEvent('beforeinput', { bubbles: true, data: arguments[0], inputType: 'insertText' }));
|
||
input.dispatchEvent(new InputEvent('input', { bubbles: true, data: arguments[0], inputType: 'insertText' }));
|
||
input.dispatchEvent(new Event('change', { bubbles: true }));
|
||
input.dispatchEvent(new Event('blur', { bubbles: true }));
|
||
const current = (input.value || '').trim();
|
||
if (current === arguments[0]) return 'filled';
|
||
input.value = '';
|
||
input.dispatchEvent(new Event('input', { bubbles: true }));
|
||
for (const ch of arguments[0]) {
|
||
input.dispatchEvent(new KeyboardEvent('keydown', { key: ch, bubbles: true }));
|
||
input.value += ch;
|
||
input.dispatchEvent(new InputEvent('input', { bubbles: true, data: ch, inputType: 'insertText' }));
|
||
input.dispatchEvent(new KeyboardEvent('keyup', { key: ch, bubbles: true }));
|
||
}
|
||
input.dispatchEvent(new Event('change', { bubbles: true }));
|
||
if ((input.value || '').trim() === arguments[0]) return 'filled';
|
||
return input.value;
|
||
""", email)
|
||
if result == "not-ready":
|
||
time.sleep(0.5)
|
||
continue
|
||
if result == "filled":
|
||
return True
|
||
time.sleep(0.5)
|
||
return False
|
||
|
||
def click_continue_button(page) -> bool:
|
||
"""Click continue/signup button after email is filled."""
|
||
clicked = page.run_js(r"""
|
||
function isVisible(node) {
|
||
if (!node) return false;
|
||
const style = window.getComputedStyle(node);
|
||
if (style.display === 'none' || style.visibility === 'hidden' || style.opacity === '0') return false;
|
||
const rect = node.getBoundingClientRect();
|
||
return rect.width > 0 && rect.height > 0;
|
||
}
|
||
const input = Array.from(document.querySelectorAll('input[data-testid="email"], input[name="email"], input[type="email"], input[autocomplete="email"]')).find((node) => isVisible(node) && !node.disabled && !node.readOnly) || null;
|
||
if (!input || !(input.value || '').trim()) return false;
|
||
const buttons = Array.from(document.querySelectorAll('button[type="submit"], button')).filter((node) => isVisible(node) && !node.disabled && node.getAttribute('aria-disabled') !== 'true');
|
||
const submitButton = buttons.find((node) => {
|
||
const raw = (node.innerText || node.textContent || '').trim();
|
||
const compact = raw.replace(/\s+/g, '').toLowerCase();
|
||
return (
|
||
raw.includes('注册') ||
|
||
compact === 'signup' ||
|
||
compact.includes('signup') ||
|
||
compact.includes('continue') ||
|
||
compact.includes('next') ||
|
||
compact.includes('createaccount') ||
|
||
compact.includes('submit')
|
||
);
|
||
});
|
||
if (!submitButton || submitButton.disabled) return false;
|
||
submitButton.click();
|
||
return true;
|
||
""")
|
||
return bool(clicked)
|
||
|
||
def gen_name() -> tuple[str, str]:
|
||
first = ["Alex", "Sam", "Jordan", "Taylor", "Casey", "Riley", "Morgan", "Avery", "Quinn", "Jamie"]
|
||
last = ["Smith", "Johnson", "Brown", "Taylor", "Anderson", "Thomas", "Jackson", "White", "Harris", "Martin"]
|
||
return random.choice(first), random.choice(last)
|
||
|
||
|
||
def fill_profile_js(page, password: str, timeout: int = 120) -> dict:
|
||
"""Fill given/family name + password and submit after Turnstile is ready."""
|
||
given_name, family_name = gen_name()
|
||
deadline = time.time() + timeout
|
||
form_filled = False
|
||
while time.time() < deadline:
|
||
state = page.run_js(
|
||
r"""
|
||
const givenName = arguments[0];
|
||
const familyName = arguments[1];
|
||
const password = arguments[2];
|
||
function isVisible(node) {
|
||
if (!node) return false;
|
||
const style = window.getComputedStyle(node);
|
||
if (style.display === 'none' || style.visibility === 'hidden' || style.opacity === '0') return false;
|
||
const rect = node.getBoundingClientRect();
|
||
return rect.width > 0 && rect.height > 0;
|
||
}
|
||
function pick(sel) {
|
||
return Array.from(document.querySelectorAll(sel)).find((n) => isVisible(n) && !n.disabled && !n.readOnly) || null;
|
||
}
|
||
function setVal(input, value) {
|
||
if (!input) return false;
|
||
input.focus(); input.click();
|
||
const setter = Object.getOwnPropertyDescriptor(HTMLInputElement.prototype, 'value')?.set;
|
||
const tracker = input._valueTracker;
|
||
if (tracker) tracker.setValue('');
|
||
if (setter) setter.call(input, value); else input.value = value;
|
||
input.dispatchEvent(new InputEvent('beforeinput', { bubbles: true, data: value, inputType: 'insertText' }));
|
||
input.dispatchEvent(new InputEvent('input', { bubbles: true, data: value, inputType: 'insertText' }));
|
||
input.dispatchEvent(new Event('change', { bubbles: true }));
|
||
input.blur();
|
||
return String(input.value || '').trim() === String(value || '').trim();
|
||
}
|
||
const givenInput = pick('input[data-testid="givenName"], input[name="givenName"], input[autocomplete="given-name"]');
|
||
const familyInput = pick('input[data-testid="familyName"], input[name="familyName"], input[autocomplete="family-name"]');
|
||
const passwordInput = pick('input[data-testid="password"], input[name="password"], input[type="password"], input[autocomplete="new-password"]');
|
||
if (!givenInput || !familyInput || !passwordInput) return 'not-ready';
|
||
const ok1 = setVal(givenInput, givenName);
|
||
const ok2 = setVal(familyInput, familyName);
|
||
const ok3 = setVal(passwordInput, password);
|
||
if (!ok1 || !ok2 || !ok3) return 'fill-failed';
|
||
const cfInput = document.querySelector('input[name="cf-turnstile-response"]');
|
||
const cfPresent = !!cfInput || !!document.querySelector('iframe[src*="turnstile"], div.cf-turnstile, [data-sitekey]');
|
||
if (cfPresent) {
|
||
const token = String((cfInput && cfInput.value) || '').trim();
|
||
if (token.length < 80) return 'wait-cf:' + token.length;
|
||
}
|
||
const buttons = Array.from(document.querySelectorAll('button[type="submit"], button')).filter((n) => isVisible(n) && !n.disabled && n.getAttribute('aria-disabled') !== 'true');
|
||
const submitBtn = buttons.find((n) => {
|
||
const t = (n.innerText || n.textContent || '').replace(/\s+/g, '').toLowerCase();
|
||
return t.includes('完成注册') || t.includes('创建账户') || t.includes('signup') || t.includes('signup') || t.includes('createaccount') || t.includes('continue') || t.includes('confirm');
|
||
});
|
||
if (!submitBtn) return 'filled-no-submit';
|
||
submitBtn.click();
|
||
return 'submitted';
|
||
""",
|
||
given_name,
|
||
family_name,
|
||
password,
|
||
)
|
||
if state == "not-ready":
|
||
time.sleep(0.8)
|
||
continue
|
||
if state == "fill-failed":
|
||
time.sleep(0.5)
|
||
continue
|
||
if isinstance(state, str) and state.startswith("wait-cf"):
|
||
form_filled = True
|
||
bypass_turnstile(page)
|
||
time.sleep(1.5)
|
||
continue
|
||
if state == "filled-no-submit":
|
||
form_filled = True
|
||
time.sleep(1)
|
||
continue
|
||
if state == "submitted":
|
||
return {"ok": True, "given_name": given_name, "family_name": family_name, "password": password}
|
||
time.sleep(0.8)
|
||
return {"ok": False, "error": "timeout waiting profile/turnstile", "given_name": given_name, "family_name": family_name}
|
||
|
||
|
||
def wait_for_sso(page, timeout: int = 180) -> str:
|
||
deadline = time.time() + timeout
|
||
last_click = 0.0
|
||
while time.time() < deadline:
|
||
# periodically re-click final submit if still on complete-registration page
|
||
now = time.time()
|
||
if now - last_click >= 3:
|
||
try:
|
||
page.run_js(r"""
|
||
function isVisible(node) {
|
||
if (!node) return false;
|
||
const style = window.getComputedStyle(node);
|
||
if (style.display === 'none' || style.visibility === 'hidden' || style.opacity === '0') return false;
|
||
const rect = node.getBoundingClientRect();
|
||
return rect.width > 0 && rect.height > 0;
|
||
}
|
||
const titleHit = !!Array.from(document.querySelectorAll('h1,h2,div,span')).find((el) => {
|
||
const t = (el.textContent || '').replace(/\s+/g, '');
|
||
return t.includes('完成注册') || t.toLowerCase().includes('complete');
|
||
});
|
||
if (!titleHit) return false;
|
||
const cfInput = document.querySelector('input[name="cf-turnstile-response"]');
|
||
if (cfInput && String(cfInput.value || '').trim().length < 80) return false;
|
||
const buttons = Array.from(document.querySelectorAll('button[type="submit"], button')).filter((n) => isVisible(n) && !n.disabled);
|
||
const submitBtn = buttons.find((n) => {
|
||
const t = (n.innerText || n.textContent || '').replace(/\s+/g, '').toLowerCase();
|
||
return t.includes('完成注册') || t.includes('创建账户') || t.includes('signup') || t.includes('signup') || t.includes('createaccount');
|
||
});
|
||
if (submitBtn) { submitBtn.click(); return true; }
|
||
return false;
|
||
""")
|
||
except Exception:
|
||
pass
|
||
last_click = now
|
||
try:
|
||
cookies = None
|
||
for getter in (
|
||
lambda: page.cookies(all_domains=True, all_info=True),
|
||
lambda: page.cookies(all_domains=True),
|
||
lambda: page.cookies(),
|
||
):
|
||
try:
|
||
cookies = getter()
|
||
if cookies:
|
||
break
|
||
except TypeError:
|
||
continue
|
||
except Exception:
|
||
continue
|
||
if isinstance(cookies, list):
|
||
for c in cookies:
|
||
if not isinstance(c, dict):
|
||
continue
|
||
if str(c.get("name", "")).strip() == "sso" and c.get("value"):
|
||
return str(c.get("value")).strip()
|
||
except Exception:
|
||
pass
|
||
time.sleep(1)
|
||
return ""
|
||
|
||
|
||
def maybe_export_cpa(email: str, password: str, sso: str, page) -> str | None:
|
||
"""Best-effort CPA export using project cpa_export + config.json."""
|
||
cfg_path = Path(__file__).resolve().parent / "config.json"
|
||
if not cfg_path.is_file():
|
||
return None
|
||
try:
|
||
cfg = json.loads(cfg_path.read_text(encoding="utf-8"))
|
||
except Exception:
|
||
return None
|
||
if not cfg.get("cpa_export_enabled", True):
|
||
return None
|
||
try:
|
||
import cpa_export
|
||
cookies = cpa_export.export_cookies_from_page(page)
|
||
result = cpa_export.export_cpa_xai_for_account(
|
||
email,
|
||
password,
|
||
page=None,
|
||
cookies=cookies,
|
||
sso=sso,
|
||
config=cfg,
|
||
log_callback=lambda m: print(f"[cpa] {m}"),
|
||
)
|
||
if result.get("ok") and result.get("path"):
|
||
return str(result["path"])
|
||
except Exception as exc:
|
||
print(f"[!] CPA export exception: {exc}")
|
||
return None
|
||
|
||
|
||
def do_register(alias: str, password: str, imap_user: str, imap_password: str, headless: bool = False, proxy: str = "") -> dict:
|
||
email = alias
|
||
print(f"[*] Registering: {email}")
|
||
|
||
opts = create_browser_options(headless=headless, proxy=proxy)
|
||
browser = None
|
||
imap_conn = None
|
||
try:
|
||
browser = Chromium(opts)
|
||
page = wait_for_tab(browser, timeout=20)
|
||
if page is None:
|
||
raise Exception("无法获取浏览器标签页")
|
||
|
||
page.get(SIGNUP_URL)
|
||
page.wait.doc_loaded(timeout=20)
|
||
time.sleep(2)
|
||
print(f"[+] Page loaded: {page.url}")
|
||
|
||
# 尝试绕过 Turnstile
|
||
bypass_turnstile(page)
|
||
time.sleep(1)
|
||
|
||
# 点击"使用邮箱注册"按钮
|
||
if not click_email_signup_button(page, timeout=10):
|
||
print("[!] No email signup button found, continuing anyway...")
|
||
time.sleep(2)
|
||
|
||
# 填写邮箱
|
||
if not fill_email_js(page, email):
|
||
raise Exception("未找到邮箱输入框或写入失败")
|
||
print(f"[+] Email submitted: {email}")
|
||
time.sleep(0.8)
|
||
|
||
# 点击继续
|
||
if not click_continue_button(page):
|
||
raise Exception("未找到注册按钮")
|
||
print(f"[+] Continue clicked")
|
||
time.sleep(3)
|
||
|
||
# 连接 IMAP 准备取验证码
|
||
print(f"[*] Connecting IMAP {imap_user}...")
|
||
imap_conn = imap_login(imap_user, imap_password)
|
||
|
||
# 等待验证码
|
||
print(f"[*] Waiting for code...")
|
||
code = fetch_code(imap_conn, email, timeout=CODE_TIMEOUT, poll=CODE_POLL)
|
||
if not code:
|
||
raise Exception(f"未收到验证码: {email}")
|
||
|
||
# 填写验证码
|
||
code_input = page.ele(
|
||
"css:input[data-input-otp='true'], input[name='code'], input[autocomplete='one-time-code']",
|
||
timeout=5,
|
||
)
|
||
if code_input:
|
||
code_input.clear()
|
||
code_input.input(code)
|
||
else:
|
||
boxes = page.eles("css:input[inputmode='numeric']")
|
||
for i, ch in enumerate(code):
|
||
if i < len(boxes):
|
||
boxes[i].input(ch)
|
||
time.sleep(0.1)
|
||
print(f"[+] Code filled: {code}")
|
||
time.sleep(0.5)
|
||
|
||
# 点击确认
|
||
page.run_js("""
|
||
const btns = Array.from(document.querySelectorAll('button'));
|
||
const btn = btns.find(b => {
|
||
const t = (b.innerText || b.textContent || '').replace(/\\s+/g, '').toLowerCase();
|
||
return t.includes('continue') || t.includes('confirm') || t.includes('下一步');
|
||
});
|
||
if (btn) btn.click();
|
||
""")
|
||
time.sleep(3)
|
||
|
||
# 验证码后通常进入资料页：姓名 + 密码 + Turnstile
|
||
print("[*] Filling profile (name/password)...")
|
||
profile = fill_profile_js(page, password, timeout=120)
|
||
if not profile.get("ok"):
|
||
raise Exception(f"资料填写失败: {profile.get('error') or 'unknown'}")
|
||
print(f"[+] Profile filled: {profile.get('given_name')} {profile.get('family_name')}")
|
||
|
||
# 等待 sso cookie（最终提交后）
|
||
print("[*] Waiting for sso cookie...")
|
||
sso_cookie = wait_for_sso(page, timeout=180)
|
||
current_url = page.url
|
||
print(f"[+] Current URL: {current_url}")
|
||
if not sso_cookie:
|
||
raise Exception("未获取到 sso cookie")
|
||
|
||
save_account(email, password, sso_cookie)
|
||
result = {"ok": True, "email": email, "password": password, "sso": sso_cookie, "url": current_url}
|
||
|
||
# 可选：协议优先 CPA mint
|
||
if os.environ.get("CPA_EXPORT", "1") not in ("0", "false", "False"):
|
||
try:
|
||
cpa_path = maybe_export_cpa(email, password, sso_cookie, page)
|
||
if cpa_path:
|
||
result["cpa_path"] = cpa_path
|
||
print(f"[+] CPA auth: {cpa_path}")
|
||
except Exception as cpa_exc:
|
||
print(f"[!] CPA export failed: {cpa_exc}")
|
||
result["cpa_error"] = str(cpa_exc)
|
||
return result
|
||
|
||
except Exception as e:
|
||
print(f"[-] Failed: {e}")
|
||
try:
|
||
if browser is not None:
|
||
tab = wait_for_tab(browser, timeout=5)
|
||
if tab is not None:
|
||
tab.get_screenshot(path="register_error.png", full_page=False)
|
||
print(f"[!] Screenshot: register_error.png")
|
||
except Exception:
|
||
pass
|
||
return {"ok": False, "error": str(e), "email": email}
|
||
finally:
|
||
if imap_conn:
|
||
try:
|
||
imap_conn.close()
|
||
except Exception:
|
||
pass
|
||
try:
|
||
imap_conn.logout()
|
||
except Exception:
|
||
pass
|
||
if browser is not None:
|
||
try:
|
||
browser.quit()
|
||
except Exception:
|
||
pass
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="Grok 自动注册（Cloudflare catch-all + Gmail IMAP）")
|
||
parser.add_argument("--count", type=int, default=1)
|
||
parser.add_argument("--threads", type=int, default=1)
|
||
parser.add_argument("--headless", action="store_true")
|
||
parser.add_argument("--proxy", default=None)
|
||
parser.add_argument("--domain", default=DOMAIN)
|
||
parser.add_argument("--imap-user", default=None)
|
||
parser.add_argument("--imap-password", default=None)
|
||
args = parser.parse_args()
|
||
|
||
env = load_env()
|
||
proxy = args.proxy or env.get("proxy") or env.get("cpa_proxy") or ""
|
||
if proxy == "http://127.0.0.1:7890":
|
||
proxy = ""
|
||
imap_user = args.imap_user or env.get("GMAIL_USER") or ""
|
||
imap_password = args.imap_password or env.get("GMAIL_PASSWORD") or ""
|
||
if not imap_user or not imap_password:
|
||
print("Error: IMAP credentials required via --imap-user/--imap-password or GMAIL_USER/GMAIL_PASSWORD")
|
||
return 1
|
||
|
||
print(f"[*] Domain: {args.domain}")
|
||
print(f"[*] Proxy: {proxy or 'direct'}")
|
||
print(f"[*] IMAP: {imap_user}")
|
||
print(f"[*] Count: {args.count}, Threads: {args.threads}")
|
||
|
||
tasks = []
|
||
for i in range(args.count):
|
||
tasks.append((i + 1, gen_alias(args.domain), gen_password()))
|
||
|
||
results = []
|
||
threads = max(1, min(int(args.threads or 1), 10))
|
||
if threads == 1:
|
||
for idx, alias, password in tasks:
|
||
print(f"\n=== Account {idx}/{args.count} ===")
|
||
res = do_register(alias, password, imap_user, imap_password, headless=args.headless, proxy=proxy or None)
|
||
results.append(res)
|
||
if idx < args.count:
|
||
time.sleep(3)
|
||
else:
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
print(f"[*] Running with {threads} threads")
|
||
with ThreadPoolExecutor(max_workers=threads) as ex:
|
||
futs = {
|
||
ex.submit(
|
||
do_register, alias, password, imap_user, imap_password, args.headless, proxy or None
|
||
): (idx, alias)
|
||
for idx, alias, password in tasks
|
||
}
|
||
for fut in as_completed(futs):
|
||
idx, alias = futs[fut]
|
||
try:
|
||
res = fut.result()
|
||
except Exception as exc:
|
||
res = {"ok": False, "email": alias, "error": str(exc)}
|
||
results.append(res)
|
||
status = "OK" if res.get("ok") else "FAIL"
|
||
print(f"[{status}] #{idx} {res.get('email')} {res.get('error', '')[:80]}")
|
||
|
||
print(f"\n=== Summary ===")
|
||
ok = sum(1 for r in results if r.get("ok"))
|
||
print(f"Success: {ok}/{len(results)}")
|
||
for r in results:
|
||
status = "OK" if r.get("ok") else "FAIL"
|
||
extra = r.get("cpa_path") or r.get("error", "")
|
||
print(f" [{status}] {r.get('email')} - {str(extra)[:100]}")
|
||
return 0 if ok else 1
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main() or 0)
|