#!/usr/bin/env python # -*- coding: utf-8 -*- """ Grok 注册机 - TTK GUI 版本 整合 DrissionPage_example.py, openai_register.py, batch_open_nsfw.py """ try: import tkinter as tk from tkinter import ttk, messagebox, scrolledtext TK_AVAILABLE = True TK_IMPORT_ERROR = None except ImportError as exc: tk = None ttk = None messagebox = None scrolledtext = None TK_AVAILABLE = False TK_IMPORT_ERROR = exc import threading import datetime import time import os import sys import gc import queue import secrets import struct import random import re import string import json import base64 import select import socket import socketserver import ssl import urllib.parse import tempfile import traceback os.environ.setdefault("TK_SILENCE_DEPRECATION", "1") from DrissionPage import Chromium, ChromiumOptions from DrissionPage.errors import PageDisconnectedError from curl_cffi import requests CONFIG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.json") MEMORY_CLEANUP_INTERVAL = 5 UI_BG = "#242424" UI_PANEL_BG = "#2b2b2b" UI_FG = "#f2f2f2" UI_MUTED_FG = "#b8b8b8" UI_ENTRY_BG = "#333333" UI_BUTTON_BG = "#3a3a3a" UI_ACTIVE_BG = "#4a6078" DEFAULT_CONFIG = { "duckmail_api_key": "", "cloudflare_api_base": "", "cloudflare_api_key": "", "cloudflare_auth_mode": "none", "cloudflare_path_domains": "/api/domains", "cloudflare_path_accounts": "/api/new_address", "cloudflare_path_token": "/api/token", "cloudflare_path_messages": "/api/mails", "cloudmail_api_base": "", "cloudmail_public_token": "", "cloudmail_domains": "", "cloudmail_path_messages": "/api/public/emailList", "proxy": "", "enable_nsfw": True, "register_count": 1, "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36", "grok2api_auto_add_local": False, "grok2api_local_token_file": "", "grok2api_pool_name": "ssoBasic", "grok2api_auto_add_remote": False, "grok2api_remote_base": "", "grok2api_remote_app_key": "", "api_reverse_tools": "", "cpa_export_enabled": False, "cpa_auth_dir": "./cpa_auths", "cpa_copy_to_hotload": False, "cpa_hotload_dir": "", "cpa_base_url": "https://cli-chat-proxy.grok.com/v1", "cpa_proxy": "", "cpa_headless": False, "cpa_force_standalone": True, "cpa_mint_timeout_sec": 300, "cpa_mint_cookie_inject": True, "cpa_oidc_request_timeout_sec": 15, "cpa_oidc_poll_timeout_sec": 15, "grok2api_allow_legacy_full_save": False, "email_provider": "duckmail", "yyds_api_key": "", "yyds_jwt": "", "defaultDomains": "", } config = DEFAULT_CONFIG.copy() _cf_domain_index = 0 _cloudmail_domain_index = 0 class RegistrationCancelled(Exception): pass class AccountRetryNeeded(Exception): pass class ConfigError(RuntimeError): pass class RemoteTokenCompatibilityError(RuntimeError): pass class RemoteTokenRequestError(RuntimeError): pass def log_exception(context, exc, log_callback=None): message = f"{context}: {exc.__class__.__name__}: {exc}" if log_callback: log_callback(f"[!] {message}") else: print(f"[!] {message}", file=sys.stderr) return message def _require_bool(cfg, key): value = cfg.get(key) if type(value) is not bool: raise ConfigError(f"配置项 {key} 必须是布尔值 true/false") return value def _require_int(cfg, key, minimum, maximum): value = cfg.get(key) if type(value) is not int: raise ConfigError(f"配置项 {key} 必须是整数") if not minimum <= value <= maximum: raise ConfigError(f"配置项 {key} 必须在 {minimum} 到 {maximum} 之间") return value def _require_string(cfg, key, path=False): value = cfg.get(key) if not isinstance(value, str): raise ConfigError(f"配置项 {key} 必须是字符串") value = value.strip() if key not in ("user_agent",) else value if "\x00" in value: raise ConfigError(f"配置项 {key} 包含非法空字符") if path and value: os.path.expanduser(value) return value def validate_config_structure(raw): if not isinstance(raw, dict): raise ConfigError("config root must be a JSON object") cfg = {**DEFAULT_CONFIG, **raw} bool_keys = ( "enable_nsfw", "grok2api_auto_add_local", "grok2api_auto_add_remote", "grok2api_allow_legacy_full_save", "cpa_export_enabled", "cpa_copy_to_hotload", "cpa_headless", "cpa_force_standalone", "cpa_mint_cookie_inject", ) for key in bool_keys: cfg[key] = _require_bool(cfg, key) cfg["register_count"] = _require_int(cfg, "register_count", 1, 2500) cfg["cpa_mint_timeout_sec"] = _require_int(cfg, "cpa_mint_timeout_sec", 30, 1800) cfg["cpa_oidc_request_timeout_sec"] = _require_int(cfg, "cpa_oidc_request_timeout_sec", 3, 120) cfg["cpa_oidc_poll_timeout_sec"] = _require_int(cfg, "cpa_oidc_poll_timeout_sec", 3, 120) string_keys = tuple(key for key, value in DEFAULT_CONFIG.items() if isinstance(value, str)) path_keys = {"grok2api_local_token_file", "api_reverse_tools", "cpa_auth_dir", "cpa_hotload_dir"} for key in string_keys: cfg[key] = _require_string(cfg, key, path=key in path_keys) enums = { "email_provider": {"duckmail", "yyds", "cloudflare", "cloudmail"}, "cloudflare_auth_mode": {"query-key", "bearer", "x-api-key", "x-admin-auth", "none"}, "grok2api_pool_name": {"ssoBasic", "ssoSuper"}, } for key, allowed in enums.items(): value = cfg.get(key, DEFAULT_CONFIG.get(key, "")) if value not in allowed: raise ConfigError(f"配置项 {key} 的值无效: {value!r}; 允许值: {sorted(allowed)}") cfg[key] = value api_path_keys = { "cloudflare_path_domains", "cloudflare_path_accounts", "cloudflare_path_token", "cloudflare_path_messages", "cloudmail_path_messages", } for key in api_path_keys: value = cfg[key] if value and not value.startswith("/"): value = "/" + value cfg[key] = value url_keys = { "cloudflare_api_base", "cloudmail_api_base", "grok2api_remote_base", "cpa_base_url", } for key in url_keys: value = cfg[key] if not value: continue parsed = urllib.parse.urlsplit(value) if parsed.scheme not in ("http", "https") or not parsed.netloc: raise ConfigError(f"配置项 {key} 必须是有效的 http/https URL") for key in path_keys: value = cfg[key] if value.startswith("~"): cfg[key] = os.path.expanduser(value) return cfg def validate_run_requirements(cfg): cfg = validate_config_structure(cfg) provider = cfg["email_provider"] if provider == "cloudflare" and not cfg["cloudflare_api_base"]: raise ConfigError("Cloudflare 模式需要配置 cloudflare_api_base") if provider == "cloudmail": missing = [ key for key in ("cloudmail_api_base", "cloudmail_public_token", "cloudmail_domains") if not cfg[key] ] if missing: raise ConfigError("Cloud Mail 模式缺少必需配置: " + ", ".join(missing)) if provider == "yyds" and not (cfg["yyds_api_key"] or cfg["yyds_jwt"]): raise ConfigError("YYDS 模式需要至少配置 yyds_api_key 或 yyds_jwt") if cfg["grok2api_auto_add_remote"]: missing = [ key for key in ("grok2api_remote_base", "grok2api_remote_app_key") if not cfg[key] ] if missing: raise ConfigError("远端 token 入池缺少必需配置: " + ", ".join(missing)) if cfg["cpa_copy_to_hotload"] and not cfg["cpa_hotload_dir"]: raise ConfigError("启用 CPA 热加载复制时必须配置 cpa_hotload_dir") return cfg def validate_config(raw): """Backward-compatible full validation used before a run or save.""" return validate_run_requirements(raw) def load_config(): global config if os.path.exists(CONFIG_FILE): try: with open(CONFIG_FILE, "r", encoding="utf-8") as f: loaded = json.load(f) config = validate_config_structure(loaded) except ConfigError: raise except Exception as exc: raise ConfigError(f"配置文件解析失败: {CONFIG_FILE}: {exc}") from exc else: config = validate_config_structure(DEFAULT_CONFIG.copy()) return config def save_config(): global config config = validate_config_structure(config) config_dir = os.path.dirname(os.path.abspath(CONFIG_FILE)) os.makedirs(config_dir, exist_ok=True) fd = None temp_path = None try: fd, temp_path = tempfile.mkstemp(prefix=".config-", suffix=".json.tmp", dir=config_dir) with os.fdopen(fd, "w", encoding="utf-8") as f: fd = None json.dump(config, f, indent=4, ensure_ascii=False) f.write("\n") f.flush() os.fsync(f.fileno()) try: os.chmod(temp_path, 0o600) except Exception: pass os.replace(temp_path, CONFIG_FILE) temp_path = None try: os.chmod(CONFIG_FILE, 0o600) except Exception: pass except Exception as exc: raise ConfigError(f"保存配置失败: {exc}") from exc finally: if fd is not None: try: os.close(fd) except Exception: pass if temp_path and os.path.exists(temp_path): try: os.unlink(temp_path) except Exception: pass def ensure_stable_python_runtime(): if sys.version_info < (3, 14) or os.environ.get("DPE_REEXEC_DONE") == "1": return local_app_data = os.environ.get("LOCALAPPDATA", "") candidates = [ os.path.join(local_app_data, "Programs", "Python", "Python312", "python.exe"), os.path.join(local_app_data, "Programs", "Python", "Python313", "python.exe"), ] current_python = os.path.normcase(os.path.abspath(sys.executable)) for candidate in candidates: if not os.path.isfile(candidate): continue if os.path.normcase(os.path.abspath(candidate)) == current_python: return print( f"[*] 检测到 Python {sys.version.split()[0]}，自动切换到更稳定的解释器: {candidate}" ) env = os.environ.copy() env["DPE_REEXEC_DONE"] = "1" os.execve(candidate, [candidate, os.path.abspath(__file__), *sys.argv[1:]], env) def warn_runtime_compatibility(): if sys.version_info >= (3, 14): print( "[提示] 当前 Python 为 3.14+；若出现 Mail.tm TLS 异常，建议改用 Python 3.12 或 3.13。" ) ensure_stable_python_runtime() warn_runtime_compatibility() EXTENSION_PATH = os.path.abspath( os.path.join(os.path.dirname(__file__), "turnstilePatch") ) DUCKMAIL_API_BASE = "https://api.duckmail.sbs" def get_configured_proxy(): return str(config.get("proxy", "") or "").strip() def get_proxies(): proxy = get_configured_proxy() if proxy: return {"http": proxy, "https": proxy} return {} def _parse_proxy_url(proxy): raw = str(proxy or "").strip() if not raw: return None if "://" not in raw: raw = "http://" + raw try: return urllib.parse.urlsplit(raw) except Exception: return None def _safe_proxy_port(parsed): try: return parsed.port except Exception: return None def _proxy_has_auth(proxy): parsed = _parse_proxy_url(proxy) return bool(parsed and parsed.hostname and (parsed.username is not None or parsed.password is not None)) def _strip_proxy_auth(proxy): raw = str(proxy or "").strip() parsed = _parse_proxy_url(raw) if not parsed or not parsed.hostname: return raw host = parsed.hostname if ":" in host and not host.startswith("["): host = f"[{host}]" port = _safe_proxy_port(parsed) netloc = f"{host}:{port}" if port else host stripped = urllib.parse.urlunsplit((parsed.scheme or "http", netloc, parsed.path, parsed.query, parsed.fragment)) if "://" not in raw: return stripped.split("://", 1)[1] return stripped def _proxy_endpoint_terms(proxy=None): parsed = _parse_proxy_url(proxy or get_configured_proxy()) if not parsed or not parsed.hostname: return [] terms = [parsed.hostname] port = _safe_proxy_port(parsed) if port: terms.append(f"{parsed.hostname}:{port}") terms.append(f"port {port}") return [x.lower() for x in terms if x] def is_proxy_connection_error(exc): if not get_configured_proxy(): return False err = str(exc or "").lower() if not err: return False if any(x in err for x in ("proxy", "tunnel", "socks")): return True connect_markers = ( "could not connect", "failed to connect", "connection refused", "connection reset", "connect error", "timed out", "timeout", ) if any(x in err for x in connect_markers): terms = _proxy_endpoint_terms() if not terms or any(t in err for t in terms): return True return False def page_has_proxy_error(page_obj): try: url = str(getattr(page_obj, "url", "") or "") title = str(page_obj.run_js("return document.title || ''") or "") body = str(page_obj.run_js("return document.body ? document.body.innerText.slice(0, 2000) : ''") or "") except Exception: return False text = f"{url}\n{title}\n{body}".lower() return any( marker in text for marker in ( "err_proxy", "proxy connection failed", "proxy server", "proxy authentication", "tunnel connection failed", "无法连接到代理服务器", "代理服务器", ) ) class _ReusableThreadingTCPServer(socketserver.ThreadingTCPServer): allow_reuse_address = True daemon_threads = True def _proxy_recv_until_headers(sock, timeout=20, limit=65536): sock.settimeout(timeout) data = b"" while b"\r\n\r\n" not in data and len(data) < limit: chunk = sock.recv(4096) if not chunk: break data += chunk return data def _proxy_relay(left, right, timeout=60): left.settimeout(timeout) right.settimeout(timeout) sockets = [left, right] while True: readable, _, _ = select.select(sockets, [], [], timeout) if not readable: return for sock in readable: data = sock.recv(65536) if not data: return peer = right if sock is left else left peer.sendall(data) class _LocalAuthProxyBridgeHandler(socketserver.BaseRequestHandler): def handle(self): bridge = self.server.bridge upstream = None try: initial = _proxy_recv_until_headers(self.request, timeout=bridge.timeout) if not initial: return first_line = initial.split(b"\r\n", 1)[0].decode("latin1", "ignore") if first_line.upper().startswith("CONNECT "): target = first_line.split()[1] upstream = bridge.open_upstream() req = [f"CONNECT {target} HTTP/1.1", f"Host: {target}"] if bridge.auth_header: req.append(f"Proxy-Authorization: Basic {bridge.auth_header}") upstream.sendall(("\r\n".join(req) + "\r\n\r\n").encode("latin1")) response = _proxy_recv_until_headers(upstream, timeout=bridge.timeout) if response: self.request.sendall(response) status = response.split(b"\r\n", 1)[0] if b" 200 " not in status: return _proxy_relay(self.request, upstream, timeout=bridge.relay_timeout) else: upstream = bridge.open_upstream() upstream.sendall(bridge.inject_proxy_auth(initial)) _proxy_relay(self.request, upstream, timeout=bridge.relay_timeout) except Exception: return finally: if upstream is not None: try: upstream.close() except Exception: pass class LocalAuthProxyBridge: def __init__(self, proxy_url): parsed = _parse_proxy_url(proxy_url) if not parsed or not parsed.hostname: raise ValueError("认证代理地址格式无效") if (parsed.scheme or "http").lower() not in ("http", "https"): raise ValueError("Chromium 本地认证代理桥仅支持 http/https 上游代理") self.upstream_scheme = (parsed.scheme or "http").lower() self.upstream_host = parsed.hostname self.upstream_port = _safe_proxy_port(parsed) or (443 if self.upstream_scheme == "https" else 80) username = urllib.parse.unquote(parsed.username or "") password = urllib.parse.unquote(parsed.password or "") raw_auth = f"{username}:{password}".encode("utf-8") self.auth_header = base64.b64encode(raw_auth).decode("ascii") if (username or password) else "" self.timeout = 20 self.relay_timeout = 90 self.server = None self.thread = None self.local_proxy = "" def open_upstream(self): sock = socket.create_connection((self.upstream_host, self.upstream_port), timeout=self.timeout) if self.upstream_scheme == "https": context = ssl.create_default_context() sock = context.wrap_socket(sock, server_hostname=self.upstream_host) sock.settimeout(self.timeout) return sock def inject_proxy_auth(self, data): if not self.auth_header or b"\r\n\r\n" not in data: return data if b"\r\nproxy-authorization:" in data.lower(): return data head, body = data.split(b"\r\n\r\n", 1) auth_line = f"Proxy-Authorization: Basic {self.auth_header}".encode("latin1") return head + b"\r\n" + auth_line + b"\r\n\r\n" + body def start(self): self.server = _ReusableThreadingTCPServer(("127.0.0.1", 0), _LocalAuthProxyBridgeHandler) self.server.bridge = self port = self.server.server_address[1] self.local_proxy = f"http://127.0.0.1:{port}" self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) self.thread.start() return self.local_proxy def stop(self): if self.server is not None: try: self.server.shutdown() self.server.server_close() except Exception: pass self.server = None self.thread = None self.local_proxy = "" def stop_browser_proxy_bridge(): global browser_proxy_bridge if browser_proxy_bridge is not None: try: browser_proxy_bridge.stop() except Exception: pass browser_proxy_bridge = None def prepare_browser_proxy(use_proxy=True, log_callback=None): proxy = get_configured_proxy() if not use_proxy or not proxy: return "", None if _proxy_has_auth(proxy): parsed = _parse_proxy_url(proxy) scheme = (parsed.scheme or "http").lower() if parsed else "" if scheme in ("http", "https"): bridge = LocalAuthProxyBridge(proxy) browser_proxy = bridge.start() if log_callback: log_callback(f"[*] 已为 Chromium 启动本地认证代理桥: {browser_proxy}") return browser_proxy, bridge stripped = _strip_proxy_auth(proxy) if log_callback: log_callback("[!] Chromium 暂不直接支持该认证代理协议，已使用去认证代理地址，失败将回退直连") return stripped, None return proxy, None def get_duckmail_api_key(): return config.get("duckmail_api_key", "") def get_cloudflare_api_base(): return str(config.get("cloudflare_api_base", "") or "").rstrip("/") def get_cloudflare_api_key(): return config.get("cloudflare_api_key", "") def get_cloudflare_auth_mode(): return str(config.get("cloudflare_auth_mode", "none") or "none").lower() def get_cloudflare_path(key, default_path): raw = str(config.get(key, default_path) or default_path).strip() if not raw.startswith("/"): raw = "/" + raw return raw def cloudflare_build_headers(content_type=False): headers = {"Content-Type": "application/json"} if content_type else {} key = get_cloudflare_api_key() mode = get_cloudflare_auth_mode() if key: if mode == "x-api-key": headers["X-API-Key"] = key elif mode == "x-admin-auth": headers["x-admin-auth"] = key elif mode != "none": headers["Authorization"] = f"Bearer {key}" return headers def cloudflare_apply_auth_params(params=None): merged = dict(params or {}) key = get_cloudflare_api_key() mode = get_cloudflare_auth_mode() if key and mode == "query-key": merged["key"] = key return merged def cloudflare_next_default_domain(): """按配置轮换选择 Cloudflare 临时邮箱域名。""" global _cf_domain_index domains = [x.strip() for x in str(config.get("defaultDomains", "") or "").split(",") if x.strip()] if not domains: return "" domain = domains[_cf_domain_index % len(domains)] _cf_domain_index += 1 return domain def cloudflare_is_admin_create_path(path): """判断当前创建邮箱路径是否为 cloudflare_temp_email 管理员创建接口。""" return str(path or "").rstrip("/").lower() == "/admin/new_address" def _pick_list_payload(data): if isinstance(data, list): return data if isinstance(data, dict): if isinstance(data.get("results"), list): return data.get("results") if isinstance(data.get("hydra:member"), list): return data.get("hydra:member") if isinstance(data.get("data"), list): return data.get("data") if isinstance(data.get("messages"), list): return data.get("messages") if isinstance(data.get("data"), dict): nested = data.get("data") if isinstance(nested.get("messages"), list): return nested.get("messages") return [] def cloudflare_create_temp_address(api_base): """适配 cloudflare_temp_email 新建地址接口并兼容 admin 创建模式。""" path = get_cloudflare_path("cloudflare_path_accounts", "/api/new_address") url = f"{api_base}{path}" domain = cloudflare_next_default_domain() is_admin_create = cloudflare_is_admin_create_path(path) if is_admin_create: payload = {"name": generate_username(10), "enablePrefix": True} if domain: payload["domain"] = domain headers = cloudflare_build_headers(content_type=True) else: payload = {} if domain: payload["domain"] = domain headers = {"Content-Type": "application/json"} resp = http_post(url, json=payload, headers=headers) resp.raise_for_status() try: data = resp.json() except Exception: raise Exception(f"Cloudflare {path} 返回非JSON: {resp.text[:300]}") address = data.get("address") jwt = data.get("jwt") if not address or not jwt: raise Exception(f"Cloudflare {path} 缺少 address/jwt: {data}") return address, jwt def get_cloudmail_api_base(): return str(config.get("cloudmail_api_base", "") or "").strip().rstrip("/") def get_cloudmail_public_token(): return str(config.get("cloudmail_public_token", "") or "").strip() def get_cloudmail_path(): raw = str( config.get("cloudmail_path_messages", "/api/public/emailList") or "/api/public/emailList" ).strip() return raw if raw.startswith("/") else "/" + raw def cloudmail_next_domain(): """按配置轮换选择 Cloud Mail 无人收件域名。""" global _cloudmail_domain_index domains = [ item.strip().lstrip("@") for item in str(config.get("cloudmail_domains", "") or "").split(",") if item.strip().lstrip("@") ] if not domains: return "" domain = domains[_cloudmail_domain_index % len(domains)] _cloudmail_domain_index += 1 return domain def cloudmail_get_email_and_token(): """生成无需预创建账号的 Cloud Mail 收件地址。""" if not get_cloudmail_api_base(): raise Exception("Cloud Mail API Base 未配置") if not get_cloudmail_public_token(): raise Exception("Cloud Mail Public Token 未配置") domain = cloudmail_next_domain() if not domain: raise Exception("Cloud Mail 收件域名未配置") address = f"{generate_username(12)}@{domain}" # 仅返回非敏感占位凭证；公共 Token 始终只从 config.json 读取。 return address, f"cloudmail:{address}" def cloudmail_get_messages(address): api_base = get_cloudmail_api_base() public_token = get_cloudmail_public_token() if not api_base: raise Exception("Cloud Mail API Base 未配置") if not public_token: raise Exception("Cloud Mail Public Token 未配置") payload = { "toEmail": address, "type": 0, "isDel": 0, "timeSort": "desc", "num": 1, "size": 20, } resp = http_post( f"{api_base}{get_cloudmail_path()}", headers={ "Authorization": public_token, "Content-Type": "application/json", }, json=payload, timeout=20, ) resp.raise_for_status() try: data = resp.json() except Exception: raise Exception(f"Cloud Mail 邮件接口返回非JSON: {resp.text[:300]}") if not isinstance(data, dict): raise Exception(f"Cloud Mail 邮件接口返回格式错误: {data}") result_code = data.get("code") if result_code not in (None, 200, "200"): raise Exception( f"Cloud Mail 邮件接口失败: code={result_code}, message={data.get('message', '')}" ) messages = data.get("data") if isinstance(messages, list): return messages return _pick_list_payload(data) def get_user_agent(): return config.get( "user_agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36", ) def resolve_grok2api_local_token_file(): configured = str(config.get("grok2api_local_token_file", "") or "").strip() if configured: return configured return os.path.join(os.path.dirname(__file__), "token.json") def _normalize_sso_token(raw_token): token = str(raw_token or "").strip() if token.startswith("sso="): token = token[4:] return token def add_token_to_grok2api_local_pool(raw_token, email="", log_callback=None): token = _normalize_sso_token(raw_token) if not token: return False token_file = os.path.abspath(resolve_grok2api_local_token_file()) pool_name = str(config.get("grok2api_pool_name", "ssoBasic") or "ssoBasic").strip() or "ssoBasic" parent = os.path.dirname(token_file) os.makedirs(parent, exist_ok=True) lock_path = token_file + ".lock" try: with open(lock_path, "a", encoding="utf-8"): pass os.chmod(lock_path, 0o600) except Exception: pass try: from filelock import FileLock except Exception as exc: raise RuntimeError(f"filelock 依赖不可用，拒绝非原子写入 token 池: {exc}") with FileLock(lock_path, timeout=30): data = {} if os.path.exists(token_file): try: with open(token_file, "r", encoding="utf-8") as f: data = json.load(f) or {} except Exception as exc: broken_path = token_file + f".broken-{int(time.time())}" try: os.replace(token_file, broken_path) except Exception: broken_path = token_file raise RuntimeError(f"本地 token 文件 JSON 解析失败，已停止写入以避免覆盖: {broken_path}: {exc}") if not isinstance(data, dict): raise RuntimeError("本地 token 文件根节点不是 JSON object，拒绝覆盖") pool = data.get(pool_name) if pool is None: pool = [] elif not isinstance(pool, list): raise RuntimeError(f"本地 token 池 {pool_name} 不是列表，拒绝覆盖") existing = set() for item in pool: if isinstance(item, str): existing.add(_normalize_sso_token(item)) elif isinstance(item, dict): existing.add(_normalize_sso_token(item.get("token", ""))) if token in existing: if log_callback: log_callback(f"[*] grok2api 本地池已存在 token: {pool_name}") return True pool.append({"token": token, "tags": ["auto-register"], "note": email}) data[pool_name] = pool if os.path.exists(token_file): backup_path = token_file + ".bak" try: with open(token_file, "rb") as src, open(backup_path, "wb") as dst: dst.write(src.read()) dst.flush() os.fsync(dst.fileno()) try: os.chmod(backup_path, 0o600) except Exception: pass except Exception as exc: raise RuntimeError(f"创建本地 token 备份失败，拒绝继续写入: {exc}") fd, temp_path = tempfile.mkstemp(prefix=".token-", suffix=".tmp", dir=parent) try: with os.fdopen(fd, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) f.write("\n") f.flush() os.fsync(f.fileno()) try: os.chmod(temp_path, 0o600) except Exception: pass os.replace(temp_path, token_file) temp_path = None try: os.chmod(token_file, 0o600) except Exception: pass finally: if temp_path and os.path.exists(temp_path): try: os.unlink(temp_path) except Exception: pass if log_callback: log_callback(f"[+] 已写入 grok2api 本地池: {pool_name} ({token_file})") return True def get_grok2api_remote_api_bases(base): """生成 grok2api 管理 API 候选根路径。 参数: - base str: 用户配置的 grok2api 远端地址 返回: - list[str]: 依次尝试的管理 API 根路径 """ normalized = str(base or "").strip().rstrip("/") if not normalized: return [] lower = normalized.lower() candidates = [normalized] if lower.endswith("/admin/api"): return candidates if lower.endswith("/admin"): candidates.append(f"{normalized}/api") else: candidates.append(f"{normalized}/admin/api") seen = set() unique = [] for item in candidates: if item not in seen: unique.append(item) seen.add(item) return unique def add_token_to_grok2api_remote_pool(raw_token, email="", log_callback=None): token = _normalize_sso_token(raw_token) if not token: return False base = str(config.get("grok2api_remote_base", "") or "").strip().rstrip("/") app_key = str(config.get("grok2api_remote_app_key", "") or "").strip() pool_name = str(config.get("grok2api_pool_name", "ssoBasic") or "ssoBasic").strip() if not base or not app_key: raise RemoteTokenRequestError("grok2api 远端未配置 base/app_key") headers = {"Content-Type": "application/json"} query = {"app_key": app_key} remote_pool = {"ssoBasic": "basic", "ssoSuper": "super"}[pool_name] api_bases = get_grok2api_remote_api_bases(base) incompatible = [] add_payload = {"tokens": [token], "pool": remote_pool, "tags": ["auto-register"]} for api_base in api_bases: endpoint = f"{api_base}/tokens/add" try: response = http_post(endpoint, headers=headers, params=query, json=add_payload, timeout=30) except Exception as exc: raise RemoteTokenRequestError(f"远端 /tokens/add 网络请求失败: {endpoint}: {exc}") from exc status = int(getattr(response, "status_code", 0) or 0) if 200 <= status < 300: if log_callback: log_callback(f"[+] 已写入 grok2api 远端池: {pool_name} ({endpoint})") return True if status in (404, 405): incompatible.append(f"{endpoint}: HTTP {status}") continue body = str(getattr(response, "text", "") or "")[:300] raise RemoteTokenRequestError(f"远端 /tokens/add 请求失败，不允许全量回退: {endpoint}: HTTP {status}: {body}") if not bool(config.get("grok2api_allow_legacy_full_save", False)): raise RemoteTokenCompatibilityError( "/tokens/add 不受支持，旧版全量保存默认禁用以避免并发覆盖: " + "; ".join(incompatible) ) current = None fallback_base = None etag = None load_errors = [] for api_base in api_bases or [base]: endpoint = f"{api_base}/tokens" try: response = http_get(endpoint, headers=headers, params=query, timeout=20) except Exception as exc: raise RemoteTokenRequestError(f"旧版远端池读取网络失败: {endpoint}: {exc}") from exc status = int(getattr(response, "status_code", 0) or 0) if status != 200: load_errors.append(f"{endpoint}: HTTP {status}") continue payload = response.json() candidate = payload.get("tokens") if isinstance(payload, dict) and "tokens" in payload else payload if not isinstance(candidate, dict): load_errors.append(f"{endpoint}: unexpected payload") continue current = candidate fallback_base = api_base response_headers = getattr(response, "headers", {}) or {} etag = response_headers.get("ETag") or response_headers.get("etag") break if current is None or fallback_base is None: raise RemoteTokenRequestError("无法安全读取旧版远端 token 池: " + "; ".join(load_errors)) pool = current.get(pool_name) if pool is None: pool = [] elif not isinstance(pool, list): raise RemoteTokenRequestError(f"远端 token 池 {pool_name} 不是列表，拒绝全量覆盖") existing = { _normalize_sso_token(item if isinstance(item, str) else item.get("token", "")) for item in pool if isinstance(item, (str, dict)) } if token not in existing: pool.append({"token": token, "tags": ["auto-register"], "note": email}) current[pool_name] = pool if not etag: raise RemoteTokenCompatibilityError( "旧版远端接口未提供 ETag，无法保证并发安全，已拒绝全量保存" ) save_headers = dict(headers) save_headers["If-Match"] = etag endpoint = f"{fallback_base}/tokens" try: response = http_post(endpoint, headers=save_headers, params=query, json=current, timeout=30) except Exception as exc: raise RemoteTokenRequestError(f"旧版远端池保存网络失败: {endpoint}: {exc}") from exc status = int(getattr(response, "status_code", 0) or 0) if not 200 <= status < 300: raise RemoteTokenRequestError(f"旧版远端池保存失败: {endpoint}: HTTP {status}") if log_callback: log_callback(f"[+] 已写入 grok2api 远端池（旧版兼容）: {pool_name} ({endpoint})") return True def add_token_to_grok2api_pools(raw_token, email="", log_callback=None): result = { "local": {"enabled": bool(config.get("grok2api_auto_add_local", False)), "ok": None, "error": None}, "remote": {"enabled": bool(config.get("grok2api_auto_add_remote", False)), "ok": None, "error": None}, } if result["local"]["enabled"]: try: result["local"]["ok"] = bool(add_token_to_grok2api_local_pool(raw_token, email=email, log_callback=log_callback)) except Exception as exc: result["local"]["ok"] = False result["local"]["error"] = log_exception("写入 grok2api 本地池失败", exc, log_callback) if result["remote"]["enabled"]: try: result["remote"]["ok"] = bool(add_token_to_grok2api_remote_pool(raw_token, email=email, log_callback=log_callback)) except Exception as exc: result["remote"]["ok"] = False result["remote"]["error"] = log_exception("写入 grok2api 远端池失败", exc, log_callback) return result def apply_browser_proxy_option(options, proxy): if not proxy: return if hasattr(options, "set_proxy"): try: options.set_proxy(proxy) return except Exception: pass if not hasattr(options, "set_argument"): raise AttributeError("当前 DrissionPage ChromiumOptions 不支持设置浏览器代理") try: options.set_argument(f"--proxy-server={proxy}") except TypeError: options.set_argument("--proxy-server", proxy) def create_browser_options(browser_proxy=""): options = ChromiumOptions() options.auto_port() options.set_timeouts(base=1) apply_browser_proxy_option(options, browser_proxy) if os.path.exists(EXTENSION_PATH): options.add_extension(EXTENSION_PATH) return options def _build_request_kwargs(**kwargs): request_kwargs = dict(kwargs) proxies = request_kwargs.pop("proxies", None) if proxies is None: proxies = get_proxies() if proxies: request_kwargs["proxies"] = proxies request_kwargs.setdefault("timeout", 15) return request_kwargs def http_get(url, **kwargs): request_kwargs = _build_request_kwargs(**kwargs) try: return requests.get(url, **request_kwargs) except Exception as exc: if request_kwargs.get("proxies") and is_proxy_connection_error(exc): retry_kwargs = dict(kwargs) retry_kwargs["proxies"] = {} return requests.get(url, **_build_request_kwargs(**retry_kwargs)) raise def http_post(url, **kwargs): request_kwargs = _build_request_kwargs(**kwargs) try: return requests.post(url, **request_kwargs) except Exception as exc: if request_kwargs.get("proxies") and is_proxy_connection_error(exc): retry_kwargs = dict(kwargs) retry_kwargs["proxies"] = {} return requests.post(url, **_build_request_kwargs(**retry_kwargs)) raise def raise_if_cancelled(cancel_callback=None): if cancel_callback and cancel_callback(): raise RegistrationCancelled("用户停止注册") def sleep_with_cancel(seconds, cancel_callback=None): deadline = time.time() + max(seconds, 0) while True: raise_if_cancelled(cancel_callback) remaining = deadline - time.time() if remaining <= 0: return time.sleep(min(0.2, remaining)) def get_domains(api_key=None): headers = {} key = api_key or get_duckmail_api_key() if key: headers["Authorization"] = f"Bearer {key}" resp = http_get(f"{DUCKMAIL_API_BASE}/domains", headers=headers) resp.raise_for_status() return resp.json().get("hydra:member", []) def create_account(address, password, api_key=None, expires_in=0): headers = {"Content-Type": "application/json"} key = api_key or get_duckmail_api_key() if key: headers["Authorization"] = f"Bearer {key}" data = {"address": address, "password": password, "expiresIn": expires_in} resp = http_post(f"{DUCKMAIL_API_BASE}/accounts", json=data, headers=headers) resp.raise_for_status() return resp.json() def get_token(address, password): data = {"address": address, "password": password} resp = http_post(f"{DUCKMAIL_API_BASE}/token", json=data) resp.raise_for_status() return resp.json().get("token") def get_messages(token): headers = {"Authorization": f"Bearer {token}"} resp = http_get(f"{DUCKMAIL_API_BASE}/messages", headers=headers) resp.raise_for_status() return resp.json().get("hydra:member", []) def get_message_detail(token, message_id): headers = {"Authorization": f"Bearer {token}"} resp = http_get(f"{DUCKMAIL_API_BASE}/messages/{message_id}", headers=headers) resp.raise_for_status() return resp.json() def cloudflare_get_domains(api_base, api_key=None): headers = cloudflare_build_headers(content_type=False) if api_key and "Authorization" in headers: headers["Authorization"] = f"Bearer {api_key}" if api_key and "X-API-Key" in headers: headers["X-API-Key"] = api_key path = get_cloudflare_path("cloudflare_path_domains", "/domains") params = cloudflare_apply_auth_params() resp = http_get(f"{api_base}{path}", headers=headers, params=params) resp.raise_for_status() return _pick_list_payload(resp.json()) def cloudflare_create_account(api_base, address, password, api_key=None, expires_in=0): headers = cloudflare_build_headers(content_type=True) if api_key and "Authorization" in headers: headers["Authorization"] = f"Bearer {api_key}" if api_key and "X-API-Key" in headers: headers["X-API-Key"] = api_key payload = {"address": address, "password": password, "expiresIn": expires_in} path = get_cloudflare_path("cloudflare_path_accounts", "/accounts") params = cloudflare_apply_auth_params() resp = http_post(f"{api_base}{path}", json=payload, headers=headers, params=params) resp.raise_for_status() return resp.json() def cloudflare_get_token(api_base, address, password, api_key=None): headers = cloudflare_build_headers(content_type=True) if api_key and "Authorization" in headers: headers["Authorization"] = f"Bearer {api_key}" if api_key and "X-API-Key" in headers: headers["X-API-Key"] = api_key path = get_cloudflare_path("cloudflare_path_token", "/token") resp = http_post( f"{api_base}{path}", json={"address": address, "password": password}, headers=headers, params=cloudflare_apply_auth_params(), ) resp.raise_for_status() data = resp.json() if isinstance(data, dict): if data.get("token"): return data.get("token") if isinstance(data.get("data"), dict) and data["data"].get("token"): return data["data"].get("token") return None def cloudflare_get_messages(api_base, token): headers = {"Authorization": f"Bearer {token}"} path = get_cloudflare_path("cloudflare_path_messages", "/messages") params = {"limit": 20, "offset": 0} params = cloudflare_apply_auth_params(params) resp = http_get(f"{api_base}{path}", headers=headers, params=params) resp.raise_for_status() try: data = resp.json() except Exception: raise Exception(f"Cloudflare messages 返回非JSON: {resp.text[:300]}") return _pick_list_payload(data) def cloudflare_get_message_detail(api_base, token, message_id): headers = {"Authorization": f"Bearer {token}"} candidates = [ f"{api_base}/api/mail/{message_id}", f"{api_base}{get_cloudflare_path('cloudflare_path_messages', '/messages')}/{message_id}", ] last_err = None for url in candidates: try: resp = http_get( url, headers=headers, params=cloudflare_apply_auth_params(), ) resp.raise_for_status() data = resp.json() if isinstance(data, dict) and isinstance(data.get("data"), dict): return data["data"] return data except Exception as exc: last_err = exc continue raise Exception(f"Cloudflare 获取邮件详情失败: {last_err}") YYDS_API_BASE = "https://maliapi.215.im/v1" def get_yyds_api_key(): return config.get("yyds_api_key", "") def get_yyds_jwt(): return config.get("yyds_jwt", "") def yyds_get_domains(api_key=None, jwt=None): key = api_key or get_yyds_api_key() token = jwt or get_yyds_jwt() headers = {} if token: headers["Authorization"] = f"Bearer {token}" elif key: headers["X-API-Key"] = key resp = http_get(f"{YYDS_API_BASE}/domains", headers=headers) resp.raise_for_status() data = resp.json() return data.get("data", []) if data.get("success") else [] def yyds_create_account(address=None, domain=None, api_key=None, jwt=None): key = api_key or get_yyds_api_key() token = jwt or get_yyds_jwt() headers = {"Content-Type": "application/json"} if token: headers["Authorization"] = f"Bearer {token}" elif key: headers["X-API-Key"] = key payload = {} if address: payload["address"] = address if domain: payload["domain"] = domain elif key or token: payload["autoDomainStrategy"] = "prefer_owned" resp = http_post(f"{YYDS_API_BASE}/accounts", json=payload, headers=headers) resp.raise_for_status() data = resp.json() if data.get("success"): return data.get("data", {}) raise Exception(f"YYDS 创建邮箱失败: {data}") def yyds_get_token(address, api_key=None, jwt=None): key = api_key or get_yyds_api_key() token = jwt or get_yyds_jwt() headers = {"Content-Type": "application/json"} if token: headers["Authorization"] = f"Bearer {token}" elif key: headers["X-API-Key"] = key resp = http_post( f"{YYDS_API_BASE}/token", json={"address": address}, headers=headers ) resp.raise_for_status() data = resp.json() if data.get("success"): return data.get("data", {}).get("token") raise Exception(f"YYDS 获取token失败: {data}") def yyds_get_messages(address, token=None, api_key=None, jwt=None): key = api_key or get_yyds_api_key() temp_token = token or jwt or get_yyds_jwt() headers = {} if temp_token: headers["Authorization"] = f"Bearer {temp_token}" elif key: headers["X-API-Key"] = key resp = http_get( f"{YYDS_API_BASE}/messages", params={"address": address}, headers=headers, ) resp.raise_for_status() data = resp.json() if data.get("success"): return data.get("data", {}).get("messages", []) return [] def yyds_get_message_detail(message_id, token=None, api_key=None, jwt=None): key = api_key or get_yyds_api_key() temp_token = token or jwt or get_yyds_jwt() headers = {} if temp_token: headers["Authorization"] = f"Bearer {temp_token}" elif key: headers["X-API-Key"] = key resp = http_get(f"{YYDS_API_BASE}/messages/{message_id}", headers=headers) resp.raise_for_status() data = resp.json() if data.get("success"): return data.get("data", {}) raise Exception(f"YYDS 获取邮件详情失败: {data}") def yyds_generate_username(length=10): chars = string.ascii_lowercase + string.digits return "".join(secrets.choice(chars) for _ in range(length)) def yyds_pick_domain(api_key=None, jwt=None): domains = yyds_get_domains(api_key=api_key, jwt=jwt) if not domains: raise Exception("YYDS 没有返回任何可用域名") private = [d for d in domains if d.get("isVerified") and not d.get("isPublic")] if private: return private[0]["domain"] public = [d for d in domains if d.get("isVerified") and d.get("isPublic")] if public: return public[0]["domain"] verified = [d for d in domains if d.get("isVerified")] if verified: return verified[0]["domain"] raise Exception("YYDS 无已验证域名可用") def yyds_get_email_and_token(api_key=None, jwt=None): key = api_key or get_yyds_api_key() token = jwt or get_yyds_jwt() if not token and not key: raise Exception("YYDS API Key 或 JWT 未配置") domain = yyds_pick_domain(api_key=key, jwt=token) username = yyds_generate_username(10) result = yyds_create_account( address=username, domain=domain, api_key=key, jwt=token ) address = result.get("address") or f"{username}@{domain}" temp_token = result.get("token") if not temp_token: temp_token = yyds_get_token(address, api_key=key, jwt=token) if not temp_token: raise Exception("获取 YYDS token 失败") print(f"[*] 已创建 YYDS 邮箱: {address}") return address, temp_token def yyds_get_oai_code( token, address, timeout=180, poll_interval=3, log_callback=None, jwt=None, cancel_callback=None, ): deadline = time.time() + timeout seen_ids = set() while time.time() < deadline: raise_if_cancelled(cancel_callback) try: messages = yyds_get_messages(address, token=token, jwt=jwt) except Exception as exc: if log_callback: log_callback(f"[Debug] YYDS 拉取邮件列表失败: {exc}") sleep_with_cancel(poll_interval, cancel_callback) continue for msg in messages: msg_id = msg.get("id") if not msg_id or msg_id in seen_ids: continue seen_ids.add(msg_id) to_addrs = [t.get("address", "").lower() for t in (msg.get("to") or [])] if address.lower() not in to_addrs: continue try: detail = yyds_get_message_detail(msg_id, token=token, jwt=jwt) except Exception as exc: if log_callback: log_callback(f"[Debug] YYDS 获取邮件详情失败: {exc}") continue parts = [] text_body = detail.get("text") or "" if text_body: parts.append(text_body) html_list = detail.get("html") or [] for h in html_list: parts.append(re.sub(r"<[^>]+>", " ", h)) combined = "\n".join(parts) subject = detail.get("subject", "") if log_callback: log_callback(f"[Debug] YYDS 收到邮件: {subject}") code = extract_verification_code(combined, subject) if code: if log_callback: log_callback(f"[*] YYDS 从邮件中提取到验证码: {code}") return code sleep_with_cancel(poll_interval, cancel_callback) raise Exception(f"YYDS 在 {timeout}s 内未收到验证码邮件") def generate_username(length=10): chars = string.ascii_lowercase + string.digits return "".join(secrets.choice(chars) for _ in range(length)) def pick_domain(api_key=None): domains = get_domains(api_key=api_key) if not domains: raise Exception("DuckMail 没有返回任何可用域名") private = [d for d in domains if d.get("ownerId")] verified_private = [d for d in private if d.get("isVerified")] if verified_private: return verified_private[0]["domain"] public = [d for d in domains if d.get("isVerified")] if public: return public[0]["domain"] raise Exception("DuckMail 无已验证域名可用") def get_email_provider(): return config.get("email_provider", "duckmail") def get_email_and_token(api_key=None): provider = get_email_provider() if provider == "yyds": return yyds_get_email_and_token(api_key=api_key, jwt=get_yyds_jwt()) if provider == "cloudmail": return cloudmail_get_email_and_token() if provider == "cloudflare": api_base = get_cloudflare_api_base() if not api_base: raise Exception("Cloudflare API Base 未配置") try: # cloudflare_temp_email 专用模式 return cloudflare_create_temp_address(api_base) except Exception as primary_exc: # 兜底回退到 Mail.tm 风格 key = api_key or get_cloudflare_api_key() domains = cloudflare_get_domains(api_base, api_key=key) if not domains: raise Exception(f"Cloudflare 创建邮箱失败: {primary_exc}") verified = [d for d in domains if d.get("isVerified")] target = verified[0] if verified else domains[0] domain = target.get("domain") if not domain: raise Exception("Cloudflare 域名数据格式错误，缺少 domain 字段") username = generate_username(10) address = f"{username}@{domain}" password = secrets.token_urlsafe(12) cloudflare_create_account( api_base, address, password, api_key=key, expires_in=0 ) token = cloudflare_get_token(api_base, address, password, api_key=key) if not token: raise Exception("获取 Cloudflare 邮箱 token 失败") return address, token key = api_key or get_duckmail_api_key() domain = pick_domain(api_key=key) username = generate_username(10) address = f"{username}@{domain}" password = secrets.token_urlsafe(12) create_account(address, password, api_key=key, expires_in=0) token = get_token(address, password) if not token: raise Exception("获取 DuckMail token 失败") return address, token def get_oai_code( dev_token, email, timeout=180, poll_interval=3, log_callback=None, cancel_callback=None, resend_callback=None, ): provider = get_email_provider() if provider == "yyds": return yyds_get_oai_code( dev_token, email, timeout=timeout, poll_interval=poll_interval, log_callback=log_callback, jwt=get_yyds_jwt(), cancel_callback=cancel_callback, ) if provider == "cloudmail": return cloudmail_get_oai_code( dev_token, email, timeout=timeout, poll_interval=poll_interval, log_callback=log_callback, cancel_callback=cancel_callback, resend_callback=resend_callback, ) if provider == "cloudflare": return cloudflare_get_oai_code( dev_token, email, timeout=timeout, poll_interval=poll_interval, log_callback=log_callback, cancel_callback=cancel_callback, resend_callback=resend_callback, ) return duckmail_get_oai_code( dev_token, email, timeout=timeout, poll_interval=poll_interval, log_callback=log_callback, cancel_callback=cancel_callback, ) def extract_verification_code(text, subject=""): if subject: match = re.search(r"^([A-Z0-9]{3}-[A-Z0-9]{3})\s+xAI", subject, re.IGNORECASE) if match: return match.group(1) match = re.search(r"\b([A-Z0-9]{3}-[A-Z0-9]{3})\b", text, re.IGNORECASE) if match: return match.group(1) patterns = [ r"verification\s+code[:\s]+(\d{4,8})", r"your\s+code[:\s]+(\d{4,8})", r"confirm(?:ation)?\s+code[:\s]+(\d{4,8})", ] for pattern in patterns: match = re.search(pattern, text, re.IGNORECASE) if match: return match.group(1) return None def cloudmail_get_oai_code( dev_token, email, timeout=180, poll_interval=3, log_callback=None, cancel_callback=None, resend_callback=None, ): # dev_token 是为了保持现有邮箱 Provider 调用契约；Cloud Mail 使用配置中的公共 Token。 _ = dev_token deadline = time.time() + timeout seen_attempts = {} next_resend_at = time.time() + 35 while time.time() < deadline: raise_if_cancelled(cancel_callback) if resend_callback and time.time() >= next_resend_at: try: resend_callback() if log_callback: log_callback("[*] 已触发重新发送验证码") except Exception as exc: if log_callback: log_callback(f"[Debug] 触发重发验证码失败: {exc}") next_resend_at = time.time() + 35 try: messages = cloudmail_get_messages(email) except Exception as exc: if log_callback: log_callback(f"[Debug] Cloud Mail 拉取邮件列表失败: {exc}") sleep_with_cancel(poll_interval, cancel_callback) continue if log_callback: log_callback(f"[Debug] Cloud Mail 本轮邮件数量: {len(messages)}") for msg in messages: msg_id = msg.get("emailId") or msg.get("email_id") or msg.get("id") if not msg_id: continue attempt = int(seen_attempts.get(msg_id, 0)) if attempt >= 5: continue seen_attempts[msg_id] = attempt + 1 target_address = str( msg.get("toEmail") or msg.get("to_email") or "" ).strip().lower() if target_address and target_address != email.lower(): continue parts = [] code_value = str(msg.get("code", "") or "").strip() if code_value: parts.append(f"verification code: {code_value}") for field in ("text", "content", "html", "body", "snippet"): value = msg.get(field) values = value if isinstance(value, list) else [value] for item in values: if isinstance(item, str) and item.strip(): parts.append(re.sub(r"<[^>]+>", " ", item)) subject = str(msg.get("subject", "") or "") combined = "\n".join(parts) if log_callback: log_callback(f"[Debug] Cloud Mail 收到邮件: {subject}") code = extract_verification_code(combined, subject) if code: if log_callback: log_callback(f"[*] Cloud Mail 从邮件中提取到验证码: {code}") return code if log_callback: log_callback( f"[Debug] Cloud Mail 邮件已解析但未提取到验证码 " f"id={msg_id} attempt={seen_attempts[msg_id]}" ) sleep_with_cancel(poll_interval, cancel_callback) raise Exception(f"Cloud Mail 在 {timeout}s 内未收到验证码邮件") def duckmail_get_oai_code( dev_token, email, timeout=180, poll_interval=3, log_callback=None, cancel_callback=None, ): deadline = time.time() + timeout seen_ids = set() while time.time() < deadline: raise_if_cancelled(cancel_callback) try: messages = get_messages(dev_token) except Exception as exc: if log_callback: log_callback(f"[Debug] 拉取邮件列表失败: {exc}") sleep_with_cancel(poll_interval, cancel_callback) continue for msg in messages: msg_id = msg.get("id") or msg.get("msgid") if not msg_id or msg_id in seen_ids: continue seen_ids.add(msg_id) recipients = [t.get("address", "").lower() for t in (msg.get("to") or [])] if email.lower() not in recipients: continue try: detail = get_message_detail(dev_token, msg_id) except Exception as exc: if log_callback: log_callback(f"[Debug] 获取邮件详情失败: {exc}") continue parts = [] text_body = detail.get("text") or "" if text_body: parts.append(text_body) html_list = detail.get("html") or [] for h in html_list: parts.append(re.sub(r"<[^>]+>", " ", h)) combined = "\n".join(parts) subject = detail.get("subject", "") if log_callback: log_callback(f"[Debug] 收到邮件: {subject}") code = extract_verification_code(combined, subject) if code: if log_callback: log_callback(f"[*] 从邮件中提取到验证码: {code}") return code sleep_with_cancel(poll_interval, cancel_callback) raise Exception(f"在 {timeout}s 内未收到验证码邮件") def cloudflare_get_oai_code( dev_token, email, timeout=180, poll_interval=3, log_callback=None, cancel_callback=None, resend_callback=None, ): api_base = get_cloudflare_api_base() if not api_base: raise Exception("Cloudflare API Base 未配置") deadline = time.time() + timeout # 同一封邮件正文可能延迟可读，允许多次重试解析，避免偶发漏码 seen_attempts = {} next_resend_at = time.time() + 35 while time.time() < deadline: raise_if_cancelled(cancel_callback) if resend_callback and time.time() >= next_resend_at: try: resend_callback() if log_callback: log_callback("[*] 已触发重新发送验证码") except Exception as exc: if log_callback: log_callback(f"[Debug] 触发重发验证码失败: {exc}") next_resend_at = time.time() + 35 try: messages = cloudflare_get_messages(api_base, dev_token) except Exception as exc: if log_callback: log_callback(f"[Debug] Cloudflare 拉取邮件列表失败: {exc}") sleep_with_cancel(poll_interval, cancel_callback) continue if log_callback: log_callback(f"[Debug] Cloudflare 本轮邮件数量: {len(messages)}") for msg in messages: msg_id = msg.get("id") or msg.get("msgid") if not msg_id: continue attempt = int(seen_attempts.get(msg_id, 0)) if attempt >= 5: continue seen_attempts[msg_id] = attempt + 1 recipients = [t.get("address", "").lower() for t in (msg.get("to") or [])] msg_addr = str(msg.get("address", "")).lower() # 优先匹配目标邮箱；若结构不一致也允许继续解析，避免接口字段漂移导致漏码 address_matched = True if recipients: address_matched = email.lower() in recipients elif msg_addr: address_matched = msg_addr == email.lower() if not address_matched and log_callback: log_callback(f"[Debug] 跳过疑似非目标邮件 id={msg_id} address={msg_addr} to={recipients}") continue parts = [] # 先直接从列表项取内容，避免 detail 接口差异导致漏码 for field in ("text", "raw", "content", "intro", "body", "snippet"): value = msg.get(field) if isinstance(value, str) and value.strip(): parts.append(value) html_list = msg.get("html") or [] if isinstance(html_list, str): html_list = [html_list] for h in html_list: parts.append(re.sub(r"<[^>]+>", " ", h)) subject = str(msg.get("subject", "") or "") combined = "\n".join(parts) # 再尝试 detail 接口补全内容 try: detail = cloudflare_get_message_detail(api_base, dev_token, msg_id) for field in ("text", "raw", "content", "intro", "body", "snippet"): value = detail.get(field) if isinstance(value, str) and value.strip(): combined += "\n" + value html_list2 = detail.get("html") or [] if isinstance(html_list2, str): html_list2 = [html_list2] for h in html_list2: combined += "\n" + re.sub(r"<[^>]+>", " ", h) if not subject: subject = str(detail.get("subject", "") or "") except Exception as exc: if log_callback: log_callback(f"[Debug] Cloudflare detail接口失败，改用列表内容解析: {exc}") if log_callback: log_callback(f"[Debug] Cloudflare 收到邮件: {subject}") code = extract_verification_code(combined, subject) if code: if log_callback: log_callback(f"[*] Cloudflare 从邮件中提取到验证码: {code}") return code elif log_callback: log_callback(f"[Debug] 邮件已解析但未提取到验证码 id={msg_id} attempt={seen_attempts[msg_id]}") sleep_with_cancel(poll_interval, cancel_callback) raise Exception(f"Cloudflare 在 {timeout}s 内未收到验证码邮件") def generate_random_birthdate(): import datetime as dt today = dt.date.today() age = random.randint(20, 40) birth_year = today.year - age birth_month = random.randint(1, 12) birth_day = random.randint(1, 28) return f"{birth_year}-{birth_month:02d}-{birth_day:02d}T16:00:00.000Z" def response_preview(res, limit=200): try: text = str(res.text or "") except Exception: text = "" text = re.sub(r"\s+", " ", text).strip() return text[:limit] def is_cloudflare_block_response(res): try: headers = {str(k).lower(): str(v).lower() for k, v in dict(res.headers).items()} text = str(res.text or "").lower() server = headers.get("server", "") content_type = headers.get("content-type", "") return ( res.status_code in (403, 429, 503) and ( "cloudflare" in server or "cloudflare" in text or "cf-error" in text or "__cf_chl" in text or "text/html" in content_type ) ) except Exception: return False def set_birth_date(session, log_callback=None): url = "https://grok.com/rest/auth/set-birth-date" new_headers = { "content-type": "application/json", "origin": "https://grok.com", "referer": "https://grok.com/", } payload = {"birthDate": generate_random_birthdate()} try: res = session.post(url, json=payload, headers=new_headers, timeout=15) if log_callback: log_callback( f"[Debug] set_birth_date status: {res.status_code}, body: {response_preview(res)}" ) if 200 <= res.status_code < 300: return True, "ok" if is_cloudflare_block_response(res): return ( False, "set_birth_date 被 grok.com 的 Cloudflare 防护拦截，HTTP " f"{res.status_code}", ) return False, f"set_birth_date HTTP {res.status_code}: {response_preview(res)}" except Exception as e: if log_callback: log_callback(f"[set_birth_date] 异常: {e}") return False, f"set_birth_date 异常: {e}" def set_tos_accepted(session, log_callback=None): url = "https://accounts.x.ai/auth_mgmt.AuthManagement/SetTosAcceptedVersion" payload = struct.pack("B", (2 << 3) | 0) + struct.pack("B", 1) data = b"\x00" + struct.pack(">I", len(payload)) + payload new_headers = { "content-type": "application/grpc-web+proto", "x-grpc-web": "1", "x-user-agent": "connect-es/2.1.1", "origin": "https://accounts.x.ai", "referer": "https://accounts.x.ai/accept-tos", } try: res = session.post(url, data=data, headers=new_headers, timeout=15) if log_callback: log_callback(f"[Debug] set_tos_accepted status: {res.status_code}") if 200 <= res.status_code < 300: return True, "ok" if is_cloudflare_block_response(res): return ( False, "set_tos_accepted 被 accounts.x.ai 的 Cloudflare 防护拦截，HTTP " f"{res.status_code}", ) return False, f"set_tos_accepted HTTP {res.status_code}: {response_preview(res)}" except Exception as e: if log_callback: log_callback(f"[set_tos_accepted] 异常: {e}") return False, f"set_tos_accepted 异常: {e}" def encode_grpc_nsfw_settings(): field1_content = bytes([0x10, 0x01]) field1 = bytes([0x0A, len(field1_content)]) + field1_content nsfw_string = b"always_show_nsfw_content" field2_inner = bytes([0x0A, len(nsfw_string)]) + nsfw_string field2 = bytes([0x12, len(field2_inner)]) + field2_inner payload = field1 + field2 return b"\x00" + struct.pack(">I", len(payload)) + payload def update_nsfw_settings(session, log_callback=None): url = "https://grok.com/auth_mgmt.AuthManagement/UpdateUserFeatureControls" data = encode_grpc_nsfw_settings() new_headers = { "content-type": "application/grpc-web+proto", "x-grpc-web": "1", "origin": "https://grok.com", "referer": "https://grok.com/", } try: res = session.post(url, data=data, headers=new_headers, timeout=15) if log_callback: log_callback( f"[Debug] update_nsfw status: {res.status_code}, body: {response_preview(res)}" ) if 200 <= res.status_code < 300: return True, "ok" if is_cloudflare_block_response(res): return ( False, "update_nsfw_settings 被 grok.com 的 Cloudflare 防护拦截，HTTP " f"{res.status_code}", ) return False, f"update_nsfw_settings HTTP {res.status_code}: {response_preview(res)}" except Exception as e: if log_callback: log_callback(f"[update_nsfw] 异常: {e}") return False, f"update_nsfw_settings 异常: {e}" def enable_nsfw_for_token(token, cf_clearance="", log_callback=None): proxies = get_proxies() user_agent = get_user_agent() try: with requests.Session(impersonate="chrome120", proxies=proxies) as session: cookie_parts = [f"sso={token}", f"sso-rw={token}"] if cf_clearance: cookie_parts.append(f"cf_clearance={cf_clearance}") session.headers.update( { "user-agent": user_agent, "cookie": "; ".join(cookie_parts), } ) ok, message = set_tos_accepted(session, log_callback) if not ok: return False, message ok, message = set_birth_date(session, log_callback) if not ok: return False, message ok, message = update_nsfw_settings(session, log_callback) if not ok: return False, message return True, "成功开启 NSFW" except Exception as e: return False, f"异常: {str(e)}" SIGNUP_URL = "https://accounts.x.ai/sign-up?redirect=grok-com" browser = None page = None browser_proxy_bridge = None browser_started_with_proxy = False def setup_light_theme(root): try: root.option_add("*Background", UI_BG) root.option_add("*Foreground", UI_FG) root.option_add("*selectBackground", UI_ACTIVE_BG) root.option_add("*selectForeground", UI_FG) root.option_add("*insertBackground", UI_FG) root.option_add("*Entry.Background", UI_ENTRY_BG) root.option_add("*Text.Background", UI_ENTRY_BG) root.option_add("*Menu.Background", UI_ENTRY_BG) root.option_add("*Menu.Foreground", UI_FG) style = ttk.Style(root) available = set(style.theme_names()) if "clam" in available: style.theme_use("clam") elif "default" in available: style.theme_use("default") root.configure(bg=UI_BG) style.configure(".", background=UI_BG, foreground=UI_FG, fieldbackground=UI_ENTRY_BG) style.configure("TFrame", background=UI_BG) style.configure("TLabelframe", background=UI_BG, foreground=UI_FG) style.configure("TLabelframe.Label", background=UI_BG, foreground=UI_FG) style.configure("TLabel", background=UI_BG, foreground=UI_FG) style.configure("TCheckbutton", background=UI_BG, foreground=UI_FG) style.configure("TButton", background=UI_BUTTON_BG, foreground=UI_FG) style.configure("TEntry", fieldbackground=UI_ENTRY_BG, foreground=UI_FG) style.configure("TCombobox", fieldbackground=UI_ENTRY_BG, foreground=UI_FG) style.configure("TSpinbox", fieldbackground=UI_ENTRY_BG, foreground=UI_FG) except Exception: pass def tk_label(parent, text="", **kwargs): return tk.Label(parent, text=text, bg=kwargs.pop("bg", UI_BG), fg=kwargs.pop("fg", UI_FG), **kwargs) def tk_entry(parent, textvariable=None, width=30, **kwargs): return tk.Entry( parent, textvariable=textvariable, width=width, bg=UI_ENTRY_BG, fg=UI_FG, insertbackground=UI_FG, disabledbackground="#2f2f2f", disabledforeground=UI_MUTED_FG, highlightthickness=1, highlightbackground="#555555", relief=tk.SOLID, **kwargs, ) def tk_button(parent, text="", command=None, state="normal", **kwargs): return tk.Button( parent, text=text, command=command, state=state, bg=UI_BUTTON_BG, fg=UI_FG, activebackground=UI_ACTIVE_BG, activeforeground=UI_FG, disabledforeground="#777777", relief=tk.RAISED, padx=10, pady=3, **kwargs, ) def tk_checkbutton(parent, text="", variable=None, **kwargs): return tk.Checkbutton( parent, text=text, variable=variable, bg=UI_BG, fg=UI_FG, activebackground=UI_BG, activeforeground=UI_FG, selectcolor="#3d7be0", **kwargs, ) def tk_option_menu(parent, variable, values, width=12): menu = tk.OptionMenu(parent, variable, *values) menu.configure( width=width, bg=UI_ENTRY_BG, fg=UI_FG, activebackground=UI_ACTIVE_BG, activeforeground=UI_FG, highlightthickness=1, highlightbackground="#555555", relief=tk.SOLID, ) menu["menu"].configure(bg=UI_ENTRY_BG, fg=UI_FG, activebackground=UI_ACTIVE_BG, activeforeground=UI_FG) return menu def start_browser(log_callback=None, use_proxy=True): global browser, page, browser_proxy_bridge, browser_started_with_proxy last_exc = None proxy_enabled = bool(use_proxy and get_configured_proxy()) for attempt in range(1, 5): bridge = None try: browser_proxy, bridge = prepare_browser_proxy(use_proxy=use_proxy, log_callback=log_callback) browser = Chromium(create_browser_options(browser_proxy=browser_proxy)) browser_proxy_bridge = bridge browser_started_with_proxy = bool(browser_proxy) tabs = browser.get_tabs() page = tabs[-1] if tabs else browser.new_tab() if log_callback and getattr(browser, "user_data_path", None): log_callback(f"[Debug] 当前浏览器资料目录: {browser.user_data_path}") if log_callback and get_configured_proxy(): mode = "代理" if browser_started_with_proxy else "直连" log_callback(f"[*] 浏览器网络模式: {mode}") if log_callback and attempt > 1: log_callback(f"[*] 浏览器第 {attempt} 次启动成功") return browser, page except Exception as exc: last_exc = exc if bridge is not None: try: bridge.stop() except Exception: pass if log_callback: mode = "代理" if proxy_enabled else "直连" log_callback(f"[Debug] 浏览器{mode}启动失败(第{attempt}/4次): {exc}") try: if browser is not None: browser.quit(del_data=True) except Exception: pass browser = None page = None browser_proxy_bridge = None browser_started_with_proxy = False time.sleep(min(1.5 * attempt, 4)) raise Exception(f"浏览器启动失败，已重试4次: {last_exc}") def stop_browser(): global browser, page, browser_started_with_proxy if browser is not None: try: browser.quit(del_data=True) except Exception: pass stop_browser_proxy_bridge() browser = None page = None browser_started_with_proxy = False def restart_browser(log_callback=None, use_proxy=True): stop_browser() return start_browser(log_callback=log_callback, use_proxy=use_proxy) def cleanup_runtime_memory(log_callback=None, reason="定期清理"): if log_callback: log_callback(f"[*] {reason}: 关闭浏览器并清理内存") stop_browser() try: from cpa_xai.browser_confirm import shutdown_mint_browsers shutdown_mint_browsers() except Exception as exc: if log_callback: log_callback(f"[Debug] CPA 浏览器清理失败: {exc}") collected = gc.collect() if log_callback: log_callback(f"[*] Python GC 已回收对象数: {collected}") def refresh_active_page(): global browser, page if browser is None: restart_browser() try: tabs = browser.get_tabs() if tabs: page = tabs[-1] else: page = browser.new_tab() except Exception: restart_browser() return page def click_email_signup_button(timeout=10, log_callback=None, cancel_callback=None): global page deadline = time.time() + timeout while time.time() < deadline: raise_if_cancelled(cancel_callback) if log_callback: log_callback("[Debug] 尝试查找“使用邮箱注册”按钮...") 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; } function nodeText(node) { return [ node.innerText, node.textContent, node.getAttribute('aria-label'), node.getAttribute('title'), node.getAttribute('href'), ].filter(Boolean).join(' ').replace(/\s+/g, ' ').trim(); } function scoreEntry(node) { const compact = nodeText(node).replace(/\s+/g, ''); const lower = compact.toLowerCase(); if (compact.includes('使用邮箱注册')) return 100; if (lower.includes('signupwithemail')) return 95; if (lower.includes('continuewithemail')) return 90; if (lower.includes('email') && (lower.includes('sign') || lower.includes('continue') || lower.includes('use') || lower.includes('with'))) return 80; if (lower === 'email' || lower.includes('邮箱')) return 70; return 0; } const candidates = Array.from(document.querySelectorAll('button, a, [role="button"]')) .filter((node) => isVisible(node) && !node.disabled && node.getAttribute('aria-disabled') !== 'true') .map((node) => ({ node, score: scoreEntry(node), text: nodeText(node) })) .filter((item) => item.score > 0) .sort((a, b) => b.score - a.score); const target = candidates[0]?.node || null; if (!target) { return false; } target.click(); return candidates[0].text || true; """) if clicked: if log_callback: detail = f": {clicked}" if isinstance(clicked, str) else "" log_callback(f"[*] 已点击「使用邮箱注册」按钮{detail}") sleep_with_cancel(2, cancel_callback) return True if log_callback: current_url = page.url if page else "none" log_callback(f"[Debug] 当前URL: {current_url}") sleep_with_cancel(1, cancel_callback) if log_callback: page_html = page.html[:500] if page else "no page" log_callback(f"[Debug] 页面内容片段: {page_html}") raise Exception("未找到「使用邮箱注册」按钮") def open_signup_page(log_callback=None, cancel_callback=None): global browser, page raise_if_cancelled(cancel_callback) if browser is None: start_browser(log_callback=log_callback) if log_callback: log_callback("[*] 浏览器已启动") def _open_with_current_browser(): global page try: page = browser.get_tab(0) page.get(SIGNUP_URL) except Exception as e: if log_callback: log_callback(f"[Debug] 打开URL异常: {e}") page = browser.new_tab(SIGNUP_URL) page.wait.doc_loaded() try: _open_with_current_browser() except Exception as e: if browser_started_with_proxy and get_configured_proxy(): if log_callback: log_callback(f"[!] 浏览器代理访问注册页失败，自动回退直连: {e}") restart_browser(log_callback=log_callback, use_proxy=False) _open_with_current_browser() else: raise if browser_started_with_proxy and page_has_proxy_error(page): if log_callback: log_callback("[!] 浏览器页面显示代理错误，自动回退直连") restart_browser(log_callback=log_callback, use_proxy=False) _open_with_current_browser() sleep_with_cancel(2, cancel_callback) if log_callback: log_callback(f"[*] 当前URL: {page.url}") click_email_signup_button( log_callback=log_callback, cancel_callback=cancel_callback ) def has_profile_form(log_callback=None): refresh_active_page() try: return bool( page.run_js( """ const givenInput = document.querySelector('input[data-testid="givenName"], input[name="givenName"], input[autocomplete="given-name"]'); const familyInput = document.querySelector('input[data-testid="familyName"], input[name="familyName"], input[autocomplete="family-name"]'); const passwordInput = document.querySelector('input[data-testid="password"], input[name="password"], input[type="password"]'); return !!(givenInput && familyInput && passwordInput); """ ) ) except Exception: return False def fill_email_and_submit(timeout=45, log_callback=None, cancel_callback=None): raise_if_cancelled(cancel_callback) email, dev_token = get_email_and_token() if not email or not dev_token: raise Exception("获取邮箱失败") if log_callback: log_callback(f"[*] 已创建邮箱: {email}") deadline = time.time() + timeout last_diag_time = 0 last_reclick_time = 0 last_snapshot = None while time.time() < deadline: raise_if_cancelled(cancel_callback) filled = page.run_js( """ const email = arguments[0]; 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 textOf(node) { return [ node.innerText, node.textContent, node.getAttribute('aria-label'), node.getAttribute('title'), node.getAttribute('placeholder'), node.getAttribute('data-testid'), node.getAttribute('name'), node.getAttribute('id'), node.getAttribute('autocomplete'), ].filter(Boolean).join(' ').replace(/\s+/g, ' ').trim(); } function describeInput(node) { return [ `type=${node.getAttribute('type') || ''}`, `name=${node.getAttribute('name') || ''}`, `id=${node.getAttribute('id') || ''}`, `placeholder=${node.getAttribute('placeholder') || ''}`, `aria=${node.getAttribute('aria-label') || ''}`, `testid=${node.getAttribute('data-testid') || ''}`, ].join(' ').replace(/\s+/g, ' ').trim().slice(0, 160); } function describeAction(node) { return textOf(node).slice(0, 120); } function emailCandidates() { const direct = Array.from(document.querySelectorAll('input[data-testid="email"], input[name="email"], input[type="email"], input[autocomplete="email"], input[placeholder*="mail" i], input[aria-label*="mail" i]')); const all = Array.from(document.querySelectorAll('input, textarea')); for (const node of all) { const type = (node.getAttribute('type') || '').toLowerCase(); if (['hidden', 'submit', 'button', 'checkbox', 'radio', 'file', 'search'].includes(type)) continue; const meta = textOf(node).toLowerCase(); if (meta.includes('email') || meta.includes('e-mail') || meta.includes('mail') || meta.includes('邮箱') || meta.includes('电子邮件')) { direct.push(node); } } return Array.from(new Set(direct)); } const visibleInputs = Array.from(document.querySelectorAll('input, textarea')) .filter((node) => isVisible(node) && !node.disabled && !node.readOnly) .map(describeInput) .slice(0, 8); const visibleActions = Array.from(document.querySelectorAll('button, a, [role="button"]')) .filter((node) => isVisible(node) && !node.disabled && node.getAttribute('aria-disabled') !== 'true') .map(describeAction) .filter(Boolean) .slice(0, 10); const input = emailCandidates().find((node) => isVisible(node) && !node.disabled && !node.readOnly) || null; if (!input) { return { state: 'not-ready', url: location.href, title: document.title, inputs: visibleInputs, buttons: visibleActions, }; } input.focus(); input.click(); const valueProto = input instanceof HTMLTextAreaElement ? HTMLTextAreaElement.prototype : HTMLInputElement.prototype; const valueSetter = Object.getOwnPropertyDescriptor(valueProto, 'value')?.set; const tracker = input._valueTracker; if (tracker) tracker.setValue(''); if (valueSetter) valueSetter.call(input, email); else input.value = email; input.dispatchEvent(new InputEvent('beforeinput', { bubbles: true, data: email, inputType: 'insertText' })); input.dispatchEvent(new InputEvent('input', { bubbles: true, data: email, inputType: 'insertText' })); input.dispatchEvent(new Event('change', { bubbles: true })); const inputType = (input.getAttribute('type') || '').toLowerCase(); const isValid = inputType !== 'email' || input.checkValidity(); if ((input.value || '').trim() !== email || !isValid) { return { state: 'fill-failed', value: input.value || '', valid: isValid, input: describeInput(input), url: location.href, }; } input.blur(); return { state: 'filled', input: describeInput(input), url: location.href, }; """, email, ) state = filled.get("state") if isinstance(filled, dict) else filled if isinstance(filled, dict): last_snapshot = filled if state == "not-ready": now = time.time() if now - last_reclick_time >= 3: reclicked = 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; } function nodeText(node) { return [ node.innerText, node.textContent, node.getAttribute('aria-label'), node.getAttribute('title'), node.getAttribute('href'), ].filter(Boolean).join(' ').replace(/\s+/g, ' ').trim(); } function scoreEntry(node) { const compact = nodeText(node).replace(/\s+/g, ''); const lower = compact.toLowerCase(); if (compact.includes('使用邮箱注册')) return 100; if (lower.includes('signupwithemail')) return 95; if (lower.includes('continuewithemail')) return 90; if (lower.includes('email') && (lower.includes('sign') || lower.includes('continue') || lower.includes('use') || lower.includes('with'))) return 80; if (lower === 'email' || lower.includes('邮箱')) return 70; return 0; } const candidates = Array.from(document.querySelectorAll('button, a, [role="button"]')) .filter((node) => isVisible(node) && !node.disabled && node.getAttribute('aria-disabled') !== 'true') .map((node) => ({ node, score: scoreEntry(node), text: nodeText(node) })) .filter((item) => item.score > 0) .sort((a, b) => b.score - a.score); if (!candidates.length) return false; candidates[0].node.click(); return candidates[0].text || true; """) last_reclick_time = now if reclicked and log_callback: detail = f": {reclicked}" if isinstance(reclicked, str) else "" log_callback(f"[Debug] 邮箱输入框未出现，已再次触发邮箱注册入口{detail}") if log_callback and now - last_diag_time >= 5: last_diag_time = now inputs = " | ".join((filled or {}).get("inputs", [])[:6]) if isinstance(filled, dict) else "" buttons = " | ".join((filled or {}).get("buttons", [])[:8]) if isinstance(filled, dict) else "" url = (filled or {}).get("url", page.url if page else "") if isinstance(filled, dict) else (page.url if page else "") log_callback(f"[Debug] 等待邮箱输入框: url={url}; inputs={inputs or 'none'}; buttons={buttons or 'none'}") sleep_with_cancel(0.5, cancel_callback) continue if state != "filled": if log_callback: log_callback(f"[Debug] 邮箱输入框已出现，但写入失败: {filled}") sleep_with_cancel(0.5, cancel_callback) continue sleep_with_cancel(0.8, cancel_callback) 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; } function textOf(node) { return [ node.innerText, node.textContent, node.getAttribute('aria-label'), node.getAttribute('title'), node.getAttribute('placeholder'), node.getAttribute('data-testid'), node.getAttribute('name'), node.getAttribute('id'), node.getAttribute('autocomplete'), ].filter(Boolean).join(' ').replace(/\s+/g, ' ').trim(); } function emailCandidates() { const direct = Array.from(document.querySelectorAll('input[data-testid="email"], input[name="email"], input[type="email"], input[autocomplete="email"], input[placeholder*="mail" i], input[aria-label*="mail" i]')); const all = Array.from(document.querySelectorAll('input, textarea')); for (const node of all) { const type = (node.getAttribute('type') || '').toLowerCase(); if (['hidden', 'submit', 'button', 'checkbox', 'radio', 'file', 'search'].includes(type)) continue; const meta = textOf(node).toLowerCase(); if (meta.includes('email') || meta.includes('e-mail') || meta.includes('mail') || meta.includes('邮箱') || meta.includes('电子邮件')) { direct.push(node); } } return Array.from(new Set(direct)); } const input = emailCandidates().find((node) => isVisible(node) && !node.disabled && !node.readOnly) || null; if (!input || !(input.value || '').trim()) return false; const inputType = (input.getAttribute('type') || '').toLowerCase(); if (inputType === 'email' && !input.checkValidity()) return false; const buttons = Array.from(document.querySelectorAll('button[type="submit"], button, [role="button"], input[type="submit"]')) .filter((node) => isVisible(node) && !node.disabled && node.getAttribute('aria-disabled') !== 'true'); const submitButton = buttons.find((node) => { const text = textOf(node).replace(/\s+/g, ''); const lower = text.toLowerCase(); return ( text === '注册' || text.includes('注册') || text.includes('继续') || text.includes('下一步') || text.includes('确认') || lower.includes('signup') || lower.includes('sign up') || lower.includes('continue') || lower.includes('next') || lower.includes('createaccount') || lower.includes('submit') ); }); if (submitButton) { submitButton.click(); return textOf(submitButton) || true; } const form = input.closest('form'); if (form) { if (form.requestSubmit) form.requestSubmit(); else form.dispatchEvent(new Event('submit', { bubbles: true, cancelable: true })); return 'form-submit'; } input.focus(); input.dispatchEvent(new KeyboardEvent('keydown', { key: 'Enter', code: 'Enter', bubbles: true, cancelable: true })); input.dispatchEvent(new KeyboardEvent('keyup', { key: 'Enter', code: 'Enter', bubbles: true, cancelable: true })); return 'enter'; """ ) if clicked: if log_callback: detail = f" ({clicked})" if isinstance(clicked, str) else "" log_callback(f"[*] 已填写邮箱并提交: {email}{detail}") return email, dev_token sleep_with_cancel(0.5, cancel_callback) if last_snapshot: inputs = " | ".join(last_snapshot.get("inputs", [])[:6]) buttons = " | ".join(last_snapshot.get("buttons", [])[:8]) url = last_snapshot.get("url", page.url if page else "") raise Exception( f"未找到邮箱输入框或注册按钮，最后页面: url={url}; inputs={inputs or 'none'}; buttons={buttons or 'none'}" ) raise Exception("未找到邮箱输入框或注册按钮") def fill_code_and_submit(email, dev_token, timeout=180, log_callback=None, cancel_callback=None): def _resend_code(): page.run_js( r""" const nodes = Array.from(document.querySelectorAll('button, a, [role="button"]')); const target = nodes.find((node) => { const t = (node.innerText || node.textContent || '').replace(/\s+/g, '').toLowerCase(); return t.includes('重新发送') || t.includes('resend') || t.includes('再次发送'); }); if (target && !target.disabled) { target.click(); return true; } return false; """ ) code = get_oai_code( dev_token, email, log_callback=log_callback, cancel_callback=cancel_callback, resend_callback=_resend_code, ) if not code: raise Exception("获取验证码失败") clean_code = str(code).replace("-", "").strip() deadline = time.time() + timeout while time.time() < deadline: raise_if_cancelled(cancel_callback) filled = page.run_js( """ const code = String(arguments[0] || '').trim(); if (!code) return 'empty-code'; 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 setInputValue(input, value) { const nativeSetter = Object.getOwnPropertyDescriptor(window.HTMLInputElement.prototype, 'value')?.set; const tracker = input._valueTracker; if (tracker) tracker.setValue(''); if (nativeSetter) nativeSetter.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 })); } const aggregate = Array.from(document.querySelectorAll( 'input[data-input-otp=\"true\"], input[name=\"code\"], input[autocomplete=\"one-time-code\"], input[inputmode=\"numeric\"], input[inputmode=\"text\"]' )).find((node) => isVisible(node) && !node.disabled && !node.readOnly && Number(node.maxLength || 6) > 1); if (aggregate) { aggregate.focus(); aggregate.click(); setInputValue(aggregate, code); return String(aggregate.value || '').replace(/\\s+/g, '') ? 'filled-aggregate' : 'aggregate-failed'; } const otpBoxes = Array.from(document.querySelectorAll('input')).filter((node) => { if (!isVisible(node) || node.disabled || node.readOnly) return false; const maxLength = Number(node.maxLength || 0); const ac = String(node.autocomplete || '').toLowerCase(); return maxLength === 1 || ac === 'one-time-code'; }); if (otpBoxes.length >= code.length) { for (let i = 0; i < code.length; i += 1) { const ch = code[i] || ''; const box = otpBoxes[i]; box.focus(); box.click(); setInputValue(box, ch); box.dispatchEvent(new KeyboardEvent('keydown', { bubbles: true, key: ch })); box.dispatchEvent(new KeyboardEvent('keyup', { bubbles: true, key: ch })); } const merged = otpBoxes.slice(0, code.length).map((x) => String(x.value || '').trim()).join(''); return merged.length ? 'filled-boxes' : 'boxes-failed'; } return 'not-ready'; """, clean_code, ) if filled == "not-ready": sleep_with_cancel(0.5, cancel_callback) continue if "failed" in str(filled): if log_callback: log_callback(f"[Debug] 验证码填写失败: {filled}") sleep_with_cancel(0.5, cancel_callback) continue 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 buttons = Array.from(document.querySelectorAll('button[type=\"submit\"], button')).filter((node) => { return isVisible(node) && !node.disabled && node.getAttribute('aria-disabled') !== 'true'; }); const btn = buttons.find((node) => { const t = (node.innerText || node.textContent || '').replace(/\\s+/g, '').toLowerCase(); return ( t.includes('确认邮箱') || t.includes('继续') || t.includes('下一步') || t.includes('confirm') || t.includes('continue') || t.includes('next') ); }); if (!btn) return 'no-button'; btn.focus(); btn.click(); return 'clicked'; """ ) if clicked == "clicked" or clicked == "no-button": if log_callback: log_callback(f"[*] 已填写验证码并提交: {code}") sleep_with_cancel(1.5, cancel_callback) return code sleep_with_cancel(0.5, cancel_callback) raise Exception("验证码已获取，但自动填写/提交失败") def getTurnstileToken(log_callback=None, cancel_callback=None): global page if page is None: raise Exception("页面未就绪，无法执行 Turnstile") try: page.run_js( "try { if (window.turnstile && typeof turnstile.reset === 'function') turnstile.reset(); } catch(e) {}" ) except Exception: pass for _ in range(0, 20): raise_if_cancelled(cancel_callback) try: 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 ''; } """ ) token = str(token or "").strip() if len(token) >= 80: if log_callback: log_callback(f"[*] Turnstile 已通过，token长度={len(token)}") return token challenge_input = page.ele("@name=cf-turnstile-response") if challenge_input: wrapper = challenge_input.parent() iframe = None try: iframe = wrapper.shadow_root.ele("tag:iframe") except Exception: iframe = None 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() except Exception: pass else: # 兜底：尝试触发页面上可见的 Turnstile 容器 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 sleep_with_cancel(1, cancel_callback) raise Exception("Turnstile 获取 token 失败") def build_profile(): given_name_pool = [ "Neo", "Ethan", "Liam", "Noah", "Lucas", "Mason", "Ryan", "Leo", "Owen", "Aiden", "Elio", "Aron", "Ivan", "Nolan", "Evan", "Kai", "Caleb", "Adam", "Ezra", "Miles", "Logan", "Carter", "Hunter", "Jason", "Brian", "Dylan", "Alex", "Colin", "Blake", "Gavin", "Henry", "Julian", "Kevin", "Louis", "Marcus", "Nathan", "Oscar", "Peter", "Quinn", "Robin", "Simon", "Tristan", "Victor", "Wesley", "Xavier", "Yuri", "Zane", "Felix", "Aaron", "Damian", ] family_name_pool = [ "Lin", "Wang", "Zhao", "Liu", "Chen", "Zhang", "Xu", "Sun", "Guo", "He", "Yang", "Wu", "Zhou", "Tang", "Qin", "Shi", "Fang", "Peng", "Cao", "Deng", "Fan", "Fu", "Gao", "Han", "Hu", "Jiang", "Kong", "Lu", "Ma", "Nie", "Pan", "Qiao", "Ren", "Shao", "Tian", "Xie", "Yan", "Yao", "Yu", "Zeng", "Bai", "Duan", "Hou", "Jin", "Kang", "Luo", "Mao", "Song", "Wei", "Xiong", ] given_name = random.choice(given_name_pool) family_name = random.choice(family_name_pool) password = "N" + secrets.token_hex(4) + "!a7#" + secrets.token_urlsafe(6) return given_name, family_name, password def fill_profile_and_submit(timeout=120, log_callback=None, cancel_callback=None): given_name, family_name, password = build_profile() deadline = time.time() + timeout form_filled_once = False wait_cf_since = None last_cf_retry_at = 0.0 while time.time() < deadline: raise_if_cancelled(cancel_callback) if not form_filled_once: filled = page.run_js( """ 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 pickInput(selector) { return Array.from(document.querySelectorAll(selector)).find((node) => { return isVisible(node) && !node.disabled && !node.readOnly; }) || null; } function setInputValue(input, value) { if (!input) return false; input.focus(); input.click(); const nativeSetter = Object.getOwnPropertyDescriptor(window.HTMLInputElement.prototype, 'value')?.set; const tracker = input._valueTracker; if (tracker) tracker.setValue(''); if (nativeSetter) nativeSetter.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 = pickInput('input[data-testid="givenName"], input[name="givenName"], input[autocomplete="given-name"], input[aria-label*="名"]'); const familyInput = pickInput('input[data-testid="familyName"], input[name="familyName"], input[autocomplete="family-name"], input[aria-label*="姓"]'); const passwordInput = pickInput('input[data-testid="password"], input[name="password"], input[type="password"], input[autocomplete="new-password"]'); if (!givenInput || !familyInput || !passwordInput) return 'not-ready'; const ok1 = setInputValue(givenInput, givenName); const ok2 = setInputValue(familyInput, familyName); const ok3 = setInputValue(passwordInput, password); if (!ok1 || !ok2 || !ok3) return 'fill-failed'; const buttons = Array.from(document.querySelectorAll('button[type="submit"], button, [role="button"], input[type="submit"]')).filter((node) => { return isVisible(node) && !node.disabled && node.getAttribute('aria-disabled') !== 'true'; }); const submitBtn = buttons.find((node) => { const t = (node.innerText || node.textContent || '').replace(/\\s+/g, '').toLowerCase(); return t.includes('完成注册') || t.includes('创建账户') || t.includes('signup') || t.includes('createaccount'); }); // 必须等待 Cloudflare 校验通过后再提交 const cfInput = document.querySelector('input[name="cf-turnstile-response"]'); const cfPresent = !!cfInput || !!document.querySelector('iframe[src*="turnstile"], div.cf-turnstile, [data-sitekey], script[src*="turnstile"]'); if (cfPresent) { const token = String((cfInput && cfInput.value) || '').trim(); const solvedByToken = token.length >= 80; if (!solvedByToken) return 'wait-cloudflare:' + token.length; } if (submitBtn) { return 'ready-to-submit'; } return 'filled-no-submit'; """, given_name, family_name, password, ) if isinstance(filled, str) and filled.startswith("wait-cloudflare"): form_filled_once = True if log_callback: token_len = filled.split(":", 1)[1] if ":" in filled else "0" log_callback(f"[*] 资料已填写，等待 Cloudflare 人机验证通过... 当前token长度={token_len}") if token_len == "0": pause_seconds = random.uniform(1, 3) if log_callback: log_callback(f"[*] Cloudflare token 为空，暂停 {pause_seconds:.1f}s 后继续检测") sleep_with_cancel(pause_seconds, cancel_callback) now = time.time() if wait_cf_since is None: wait_cf_since = now # 卡住后自动二次复用 Turnstile 组件 if now - wait_cf_since >= 12 and now - last_cf_retry_at >= 8: if log_callback: log_callback("[*] Cloudflare 验证卡住，开始二次复用 Turnstile...") try: token = getTurnstileToken(log_callback=log_callback, cancel_callback=cancel_callback) if token: synced = page.run_js( """ const token = String(arguments[0] || '').trim(); const cfInput = document.querySelector('input[name="cf-turnstile-response"]'); if (!cfInput || !token) return false; const nativeSetter = Object.getOwnPropertyDescriptor(window.HTMLInputElement.prototype, 'value')?.set; if (nativeSetter) nativeSetter.call(cfInput, token); else cfInput.value = token; cfInput.dispatchEvent(new Event('input', { bubbles: true })); cfInput.dispatchEvent(new Event('change', { bubbles: true })); return String(cfInput.value || '').trim().length; """, token, ) if log_callback: log_callback(f"[*] Turnstile 二次复用完成，回填长度={synced}") except Exception as cf_exc: if log_callback: log_callback(f"[Debug] Turnstile 二次复用失败: {cf_exc}") last_cf_retry_at = now sleep_with_cancel(0.8, cancel_callback) continue if filled in ("ready-to-submit", "filled-no-submit"): form_filled_once = True elif filled == "fill-failed" and log_callback: log_callback("[Debug] 资料输入失败，重试中...") sleep_with_cancel(0.5, cancel_callback) continue elif filled == "not-ready": sleep_with_cancel(0.5, cancel_callback) continue submit_state = 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 cfInput = document.querySelector('input[name="cf-turnstile-response"]'); const cfPresent = !!cfInput || !!document.querySelector('iframe[src*="turnstile"], div.cf-turnstile, [data-sitekey], script[src*="turnstile"]'); if (cfPresent) { const token = String((cfInput && cfInput.value) || '').trim(); const solvedByToken = token.length >= 80; if (!solvedByToken) return 'wait-cloudflare:' + token.length; } function buttonText(node) { return [ node.innerText, node.textContent, node.getAttribute('value'), node.getAttribute('aria-label'), node.getAttribute('title'), ].filter(Boolean).join(' ').replace(/\s+/g, ' ').trim(); } const buttons = Array.from(document.querySelectorAll('button[type="submit"], button, [role="button"], input[type="submit"]')).filter((node) => { return isVisible(node) && !node.disabled && node.getAttribute('aria-disabled') !== 'true'; }); const submitBtn = buttons.find((node) => { const t = buttonText(node).replace(/\s+/g, '').toLowerCase(); return t.includes('完成注册') || t.includes('创建账户') || t.includes('signup') || t.includes('createaccount'); }); if (!submitBtn) { const visibleTexts = buttons.map(buttonText).filter(Boolean).slice(0, 8).join(' | '); return 'no-submit-button:' + visibleTexts; } submitBtn.focus(); submitBtn.click(); return 'submitted'; """ ) if isinstance(submit_state, str) and submit_state.startswith("wait-cloudflare"): if log_callback: token_len = submit_state.split(":", 1)[1] if ":" in submit_state else "0" log_callback(f"[*] 等待 Cloudflare 人机验证通过后再提交... 当前token长度={token_len}") now = time.time() if wait_cf_since is None: wait_cf_since = now if now - wait_cf_since >= 12 and now - last_cf_retry_at >= 8: if log_callback: log_callback("[*] 提交前仍卡住，自动再次复用 Turnstile...") try: token = getTurnstileToken(log_callback=log_callback, cancel_callback=cancel_callback) if token: synced = page.run_js( """ const token = String(arguments[0] || '').trim(); const cfInput = document.querySelector('input[name="cf-turnstile-response"]'); if (!cfInput || !token) return false; const nativeSetter = Object.getOwnPropertyDescriptor(window.HTMLInputElement.prototype, 'value')?.set; if (nativeSetter) nativeSetter.call(cfInput, token); else cfInput.value = token; cfInput.dispatchEvent(new Event('input', { bubbles: true })); cfInput.dispatchEvent(new Event('change', { bubbles: true })); return String(cfInput.value || '').trim().length; """, token, ) if log_callback: log_callback(f"[*] Turnstile 二次复用完成，回填长度={synced}") except Exception as cf_exc: if log_callback: log_callback(f"[Debug] Turnstile 二次复用失败: {cf_exc}") last_cf_retry_at = now sleep_with_cancel(0.8, cancel_callback) continue if submit_state == "submitted": if log_callback: log_callback(f"[*] 已填写注册资料并提交: {given_name} {family_name}") return {"given_name": given_name, "family_name": family_name, "password": password} wait_cf_since = None if isinstance(submit_state, str) and submit_state.startswith("no-submit-button") and log_callback: visible_buttons = submit_state.split(":", 1)[1] if ":" in submit_state else "" suffix = f" 可见按钮: {visible_buttons}" if visible_buttons else "" log_callback(f"[Debug] 未找到提交按钮，继续等待页面稳定...{suffix}") sleep_with_cancel(0.5, cancel_callback) raise Exception("最终注册页资料填写失败") def wait_for_sso_cookie(timeout=120, log_callback=None, cancel_callback=None): deadline = time.time() + timeout last_seen_names = set() last_submit_retry = 0.0 last_cf_retry_at = 0.0 final_no_submit_state = "" final_no_submit_since = None final_no_submit_timeout = 25 last_wait_exception_message = "" last_wait_exception_at = 0.0 while time.time() < deadline: raise_if_cancelled(cancel_callback) try: refresh_active_page() if page is None: sleep_with_cancel(1, cancel_callback) continue # 仍停留在“完成注册”页时，若 Cloudflare 已通过，周期性重试点击提交 now = time.time() if now - last_submit_retry >= 2.5: retried = 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, ''); const lower = t.toLowerCase(); return t.includes('完成注册') || lower.includes('completeyoursignup') || lower.includes('completesignup'); }); if (!titleHit) return 'not-final-page'; const cfInput = document.querySelector('input[name="cf-turnstile-response"]'); const cfPresent = !!cfInput || !!document.querySelector('iframe[src*="turnstile"], div.cf-turnstile, [data-sitekey], script[src*="turnstile"]'); if (cfPresent) { const token = String((cfInput && cfInput.value) || '').trim(); const solved = token.length >= 80; if (!solved) return 'final-page-wait-cf:' + token.length; } function buttonText(node) { return [ node.innerText, node.textContent, node.getAttribute('value'), node.getAttribute('aria-label'), node.getAttribute('title'), ].filter(Boolean).join(' ').replace(/\s+/g, ' ').trim(); } const buttons = Array.from(document.querySelectorAll('button[type="submit"], button, [role="button"], input[type="submit"]')).filter((node) => { return isVisible(node) && !node.disabled && node.getAttribute('aria-disabled') !== 'true'; }); const submitBtn = buttons.find((node) => { const t = buttonText(node).replace(/\s+/g, '').toLowerCase(); return t.includes('完成注册') || t.includes('创建账户') || t.includes('signup') || t.includes('createaccount'); }); if (!submitBtn) { const visibleTexts = buttons.map(buttonText).filter(Boolean).slice(0, 8).join(' | '); return 'final-page-no-submit:' + visibleTexts; } submitBtn.focus(); submitBtn.click(); return 'final-page-clicked-submit'; """ ) last_submit_retry = now if log_callback and (retried == "final-page-clicked-submit" or (isinstance(retried, str) and retried.startswith("final-page-no-submit"))): log_callback(f"[Debug] 最终页状态: {retried}") if isinstance(retried, str) and retried.startswith("final-page-no-submit"): if retried != final_no_submit_state: final_no_submit_state = retried final_no_submit_since = now elif final_no_submit_since and now - final_no_submit_since >= final_no_submit_timeout: raise AccountRetryNeeded( f"最终注册页状态 {final_no_submit_timeout}s 未变化且未找到提交按钮，重试当前账号: {retried}" ) else: final_no_submit_state = "" final_no_submit_since = None if log_callback and isinstance(retried, str) and retried.startswith("final-page-wait-cf"): token_len = retried.split(":", 1)[1] if ":" in retried else "0" log_callback(f"[Debug] 最终页状态: final-page-wait-cf, token长度={token_len}") if now - last_cf_retry_at >= 10: if log_callback: log_callback("[*] 最终页 Cloudflare 卡住，自动二次复用 Turnstile...") try: token = getTurnstileToken(log_callback=log_callback, cancel_callback=cancel_callback) if token: synced = page.run_js( """ const token = String(arguments[0] || '').trim(); const cfInput = document.querySelector('input[name="cf-turnstile-response"]'); if (!cfInput || !token) return false; const nativeSetter = Object.getOwnPropertyDescriptor(window.HTMLInputElement.prototype, 'value')?.set; if (nativeSetter) nativeSetter.call(cfInput, token); else cfInput.value = token; cfInput.dispatchEvent(new Event('input', { bubbles: true })); cfInput.dispatchEvent(new Event('change', { bubbles: true })); return String(cfInput.value || '').trim().length; """, token, ) if log_callback: log_callback(f"[*] 最终页 Turnstile 二次复用完成，回填长度={synced}") except Exception as cf_exc: if log_callback: log_callback(f"[Debug] 最终页 Turnstile 二次复用失败: {cf_exc}") last_cf_retry_at = now cookies = page.cookies(all_domains=True, all_info=True) or [] for item in cookies: if isinstance(item, dict): name = str(item.get("name", "")).strip() value = str(item.get("value", "")).strip() else: name = str(getattr(item, "name", "")).strip() value = str(getattr(item, "value", "")).strip() if name: last_seen_names.add(name) if name == "sso" and value: if log_callback: log_callback("[*] 已获取到 sso cookie") return value except PageDisconnectedError: refresh_active_page() except AccountRetryNeeded: raise except Exception as exc: if log_callback: now = time.time() message = f"{exc.__class__.__name__}: {exc}" if message != last_wait_exception_message or now - last_wait_exception_at >= 10: log_callback(f"[Debug] 等待 sso cookie 时出现异常，将继续等待: {message}") last_wait_exception_message = message last_wait_exception_at = now sleep_with_cancel(1, cancel_callback) raise Exception( f"等待超时：未获取到 sso cookie。已看到 cookies: {sorted(last_seen_names)}" ) def maybe_export_cpa_xai_after_success(email, password, sso="", log_callback=None, cancel_callback=None): if not bool(config.get("cpa_export_enabled", False)): return {"ok": False, "skipped": True, "reason": "disabled"} logger = log_callback or (lambda message: None) try: from cpa_export import export_cpa_xai_for_account except Exception as exc: logger(f"[!] CPA 模块导入失败，已跳过 OIDC 导出: {exc}") return {"ok": False, "error": str(exc)} current_page = None try: current_page = page except Exception: current_page = None try: result = export_cpa_xai_for_account( email=email, password=password, page=current_page, sso=sso, config=config, log_callback=logger, cancel_callback=cancel_callback, ) except Exception as exc: logger(f"[!] CPA OIDC 导出失败，账号已保留: {exc}") return {"ok": False, "error": str(exc)} if result.get("ok"): exported_path = result.get("hotload_path") or result.get("path") or "" suffix = f": {exported_path}" if exported_path else "" logger(f"[+] CPA OIDC 导出成功{suffix}") elif not result.get("skipped"): logger(f"[!] CPA OIDC 导出失败，账号已保留: {result.get('error') or result}") return result def _save_mail_credential(email, credential, log_callback=None): from account_outputs import save_mail_credential try: return save_mail_credential(os.path.dirname(__file__), email, credential) except Exception as exc: log_exception("保存邮箱凭据失败", exc, log_callback) return False def _append_account_line(path, email, password, sso): from account_outputs import append_account_line return append_account_line(path, email, password, sso) def _queue_unsaved_account(path, payload, error, log_callback=None): from account_outputs import queue_unsaved_account try: return queue_unsaved_account(path, payload, error) except Exception as exc: log_exception("写入账号 pending 队列失败", exc, log_callback) return False def retry_pending_file(pending_path, output_path=None, log_callback=None): from account_outputs import retry_pending_file as _retry_pending_file return _retry_pending_file(pending_path, output_path=output_path, log_callback=log_callback) def run_registration_common(count, log_callback, cancel_callback, accounts_output_file, observer): from registration_flow import RegistrationCallbacks, RegistrationOperations, run_batch callbacks = RegistrationCallbacks(log=log_callback, cancelled=cancel_callback) operations = RegistrationOperations( start_browser=lambda: start_browser(log_callback=log_callback), restart_browser=lambda: restart_browser(log_callback=log_callback), browser_missing=lambda: browser is None, open_signup_page=lambda: open_signup_page(log_callback=log_callback, cancel_callback=cancel_callback), fill_email_and_submit=lambda: fill_email_and_submit(log_callback=log_callback, cancel_callback=cancel_callback), save_mail_credential=lambda email, token: _save_mail_credential(email, token, log_callback), fill_code_and_submit=lambda email, token: fill_code_and_submit(email, token, log_callback=log_callback, cancel_callback=cancel_callback), fill_profile_and_submit=lambda: fill_profile_and_submit(log_callback=log_callback, cancel_callback=cancel_callback), wait_for_sso_cookie=lambda: wait_for_sso_cookie(log_callback=log_callback, cancel_callback=cancel_callback), enable_nsfw=lambda sso: enable_nsfw_for_token(sso, log_callback=log_callback), persist_account_line=lambda email, password, sso: _append_account_line(accounts_output_file, email, password, sso), queue_unsaved_result=lambda payload, error: _queue_unsaved_account(accounts_output_file, payload, error, log_callback), add_tokens=lambda sso, email: add_token_to_grok2api_pools(sso, email=email, log_callback=log_callback), export_cpa=lambda email, password, sso: maybe_export_cpa_xai_after_success( email=email, password=password, sso=sso, log_callback=log_callback, cancel_callback=cancel_callback, ), cleanup=lambda reason: cleanup_runtime_memory(log_callback=log_callback, reason=reason), sleep=lambda seconds: sleep_with_cancel(seconds, cancel_callback), cancelled_exception=RegistrationCancelled, retry_exception=AccountRetryNeeded, ) return run_batch( count=count, callbacks=callbacks, observer=observer, ops=operations, enable_nsfw=bool(config.get("enable_nsfw", True)), cleanup_interval=MEMORY_CLEANUP_INTERVAL, max_slot_retry=3, max_mail_retry=3, ) class GrokRegisterGUI: def __init__(self, root): self.root = root self.root.title("Grok 注册机") self.root.geometry("1120x900") self.root.minsize(960, 700) self.is_running = False self.batch_count = 0 self.success_count = 0 self.fail_count = 0 self.registered_unsaved_count = 0 self.postprocess_warning_count = 0 self.results = [] self.stop_requested = False self.ui_queue = queue.Queue() self.accounts_output_file = "" self.setup_ui() self.root.after(50, self.process_ui_queue) def setup_ui(self): load_config() main_frame = tk.Frame(self.root, bg=UI_BG, padx=10, pady=10) main_frame.pack(fill=tk.BOTH, expand=True) main_frame.grid_columnconfigure(0, weight=1) main_frame.grid_rowconfigure(3, weight=1) config_frame = tk.LabelFrame( main_frame, text="配置", bg=UI_PANEL_BG, fg=UI_FG, padx=10, pady=10, relief=tk.GROOVE, borderwidth=1, ) config_frame.grid(row=0, column=0, sticky=tk.EW, pady=(0, 8)) config_frame.grid_columnconfigure(1, weight=1, minsize=260) config_frame.grid_columnconfigure(3, weight=1, minsize=260) def add_label(row, column, text): tk_label(config_frame, text=text, bg=UI_PANEL_BG).grid( row=row, column=column, sticky=tk.W, padx=(0, 6), pady=3, ) def add_field(widget, row, column, columnspan=1, sticky=tk.EW): widget.grid( row=row, column=column, columnspan=columnspan, sticky=sticky, padx=(0, 14), pady=3, ) add_label(0, 0, "邮箱服务商:") self.email_provider_var = tk.StringVar(value=config.get("email_provider", "duckmail")) self.email_provider_combo = tk_option_menu(config_frame, self.email_provider_var, ["duckmail", "yyds", "cloudflare", "cloudmail"], width=12) add_field(self.email_provider_combo, 0, 1, sticky=tk.W) add_label(0, 2, "注册数量:") self.count_var = tk.StringVar(value=str(config.get("register_count", 1))) self.count_spinbox = tk.Spinbox( config_frame, from_=1, to=2500, width=8, textvariable=self.count_var, bg=UI_ENTRY_BG, fg=UI_FG, insertbackground=UI_FG, buttonbackground=UI_BUTTON_BG, disabledbackground="#2f2f2f", disabledforeground=UI_MUTED_FG, relief=tk.SOLID, ) add_field(self.count_spinbox, 0, 3, sticky=tk.W) add_label(1, 0, "注册选项:") self.nsfw_var = tk.BooleanVar(value=config.get("enable_nsfw", True)) self.nsfw_check = tk_checkbutton(config_frame, text="注册后开启 NSFW", variable=self.nsfw_var) add_field(self.nsfw_check, 1, 1, sticky=tk.W) add_label(1, 2, "代理（可选）:") self.proxy_var = tk.StringVar(value=config.get("proxy", "")) self.proxy_entry = tk_entry(config_frame, textvariable=self.proxy_var, width=34) add_field(self.proxy_entry, 1, 3) add_label(2, 0, "DuckMail API Key:") self.api_key_var = tk.StringVar(value=config.get("duckmail_api_key", "")) self.api_key_entry = tk_entry(config_frame, textvariable=self.api_key_var, width=34) add_field(self.api_key_entry, 2, 1) add_label(2, 2, "Cloudflare 鉴权模式:") self.cloudflare_auth_mode_var = tk.StringVar(value=config.get("cloudflare_auth_mode", "none")) self.cloudflare_auth_mode_combo = tk_option_menu( config_frame, self.cloudflare_auth_mode_var, ["query-key", "bearer", "x-api-key", "x-admin-auth", "none"], width=12 ) add_field(self.cloudflare_auth_mode_combo, 2, 3, sticky=tk.W) add_label(3, 0, "Cloudflare API Base:") self.cloudflare_api_base_var = tk.StringVar(value=config.get("cloudflare_api_base", "")) self.cloudflare_api_base_entry = tk_entry(config_frame, textvariable=self.cloudflare_api_base_var, width=72) add_field(self.cloudflare_api_base_entry, 3, 1, columnspan=3) add_label(4, 0, "Cloudflare API Key:") self.cloudflare_api_key_var = tk.StringVar(value=config.get("cloudflare_api_key", "")) self.cloudflare_api_key_entry = tk_entry(config_frame, textvariable=self.cloudflare_api_key_var, width=34) add_field(self.cloudflare_api_key_entry, 4, 1) add_label(4, 2, "CF 路径:") self.cloudflare_paths_var = tk.StringVar( value=",".join( [ config.get("cloudflare_path_domains", "/api/domains"), config.get("cloudflare_path_accounts", "/api/new_address"), config.get("cloudflare_path_token", "/api/token"), config.get("cloudflare_path_messages", "/api/mails"), ] ) ) self.cloudflare_paths_entry = tk_entry(config_frame, textvariable=self.cloudflare_paths_var, width=34) add_field(self.cloudflare_paths_entry, 4, 3) add_label(5, 0, "Cloud Mail API Base:") self.cloudmail_api_base_var = tk.StringVar(value=config.get("cloudmail_api_base", "")) self.cloudmail_api_base_entry = tk_entry(config_frame, textvariable=self.cloudmail_api_base_var, width=34) add_field(self.cloudmail_api_base_entry, 5, 1) add_label(5, 2, "Cloud Mail 域名:") self.cloudmail_domains_var = tk.StringVar(value=config.get("cloudmail_domains", "")) self.cloudmail_domains_entry = tk_entry(config_frame, textvariable=self.cloudmail_domains_var, width=34) add_field(self.cloudmail_domains_entry, 5, 3) add_label(6, 0, "Cloud Mail Public Token:") self.cloudmail_public_token_var = tk.StringVar(value=config.get("cloudmail_public_token", "")) self.cloudmail_public_token_entry = tk_entry(config_frame, textvariable=self.cloudmail_public_token_var, width=72) add_field(self.cloudmail_public_token_entry, 6, 1, columnspan=3) add_label(7, 0, "grok2api 本地入池:") self.grok2api_local_auto_var = tk.BooleanVar(value=bool(config.get("grok2api_auto_add_local", True))) self.grok2api_local_auto_check = tk_checkbutton(config_frame, variable=self.grok2api_local_auto_var) add_field(self.grok2api_local_auto_check, 7, 1, sticky=tk.W) add_label(7, 2, "grok2api 池名:") self.grok2api_pool_name_var = tk.StringVar(value=str(config.get("grok2api_pool_name", "ssoBasic"))) self.grok2api_pool_name_combo = tk_option_menu( config_frame, self.grok2api_pool_name_var, ["ssoBasic", "ssoSuper"], width=12 ) add_field(self.grok2api_pool_name_combo, 7, 3, sticky=tk.W) add_label(8, 0, "本地 token.json:") self.grok2api_local_file_var = tk.StringVar(value=str(config.get("grok2api_local_token_file", ""))) self.grok2api_local_file_entry = tk_entry(config_frame, textvariable=self.grok2api_local_file_var, width=72) add_field(self.grok2api_local_file_entry, 8, 1, columnspan=3) add_label(9, 0, "grok2api 远端入池:") self.grok2api_remote_auto_var = tk.BooleanVar(value=bool(config.get("grok2api_auto_add_remote", False))) self.grok2api_remote_auto_check = tk_checkbutton(config_frame, variable=self.grok2api_remote_auto_var) add_field(self.grok2api_remote_auto_check, 9, 1, sticky=tk.W) add_label(10, 0, "grok2api 远端 Base:") self.grok2api_remote_base_var = tk.StringVar(value=str(config.get("grok2api_remote_base", ""))) self.grok2api_remote_base_entry = tk_entry(config_frame, textvariable=self.grok2api_remote_base_var, width=72) add_field(self.grok2api_remote_base_entry, 10, 1, columnspan=3) add_label(11, 0, "grok2api 远端 app_key:") self.grok2api_remote_key_var = tk.StringVar(value=str(config.get("grok2api_remote_app_key", ""))) self.grok2api_remote_key_entry = tk_entry(config_frame, textvariable=self.grok2api_remote_key_var, width=72) add_field(self.grok2api_remote_key_entry, 11, 1, columnspan=3) add_label(12, 0, "OIDC / CPA:") self.cpa_export_var = tk.BooleanVar(value=bool(config.get("cpa_export_enabled", False))) self.cpa_export_check = tk_checkbutton(config_frame, text="注册成功后导出 CPA xAI OIDC", variable=self.cpa_export_var) add_field(self.cpa_export_check, 12, 1, sticky=tk.W) add_label(12, 2, "CPA 输出目录:") self.cpa_auth_dir_var = tk.StringVar(value=str(config.get("cpa_auth_dir", "./cpa_auths"))) self.cpa_auth_dir_entry = tk_entry(config_frame, textvariable=self.cpa_auth_dir_var, width=34) add_field(self.cpa_auth_dir_entry, 12, 3) btn_frame = tk.Frame(main_frame, bg=UI_BG) btn_frame.grid(row=1, column=0, sticky=tk.EW, pady=(0, 6)) self.start_btn = tk_button(btn_frame, text="开始注册", command=self.start_registration) self.start_btn.pack(side=tk.LEFT, padx=5) self.stop_btn = tk_button(btn_frame, text="停止", command=self.stop_registration, state=tk.DISABLED) self.stop_btn.pack(side=tk.LEFT, padx=5) self.clear_btn = tk_button(btn_frame, text="清空日志", command=self.clear_log) self.clear_btn.pack(side=tk.LEFT, padx=5) status_frame = tk.Frame(main_frame, bg=UI_BG) status_frame.grid(row=2, column=0, sticky=tk.EW, pady=(0, 6)) self.status_var = tk.StringVar(value="就绪") tk_label(status_frame, text="状态: ").pack(side=tk.LEFT) self.status_label = tk.Label(status_frame, textvariable=self.status_var, bg=UI_BG, fg="green") self.status_label.pack(side=tk.LEFT) self.stats_var = tk.StringVar(value="成功: 0 | 失败: 0 | 待恢复: 0 | 后处理警告: 0") tk.Label(status_frame, textvariable=self.stats_var, bg=UI_BG, fg=UI_FG).pack(side=tk.RIGHT) log_frame = tk.LabelFrame( main_frame, text="日志", bg=UI_PANEL_BG, fg=UI_FG, padx=5, pady=5, relief=tk.GROOVE, borderwidth=1, ) log_frame.grid(row=3, column=0, sticky=tk.NSEW) log_frame.grid_columnconfigure(0, weight=1) log_frame.grid_rowconfigure(0, weight=1) self.log_text = scrolledtext.ScrolledText( log_frame, height=18, width=60, bg="#111111", fg="#f5f5f5", insertbackground="#f5f5f5", selectbackground="#345a8a", selectforeground="#ffffff", relief=tk.SOLID, borderwidth=1, highlightthickness=1, highlightbackground="#555555", ) self.log_text.grid(row=0, column=0, sticky=tk.NSEW) self.log("[*] GUI 已就绪，配置已加载") self.log(f"[*] 当前邮箱服务商: {self.email_provider_var.get()} | 注册数量: {self.count_var.get()}") def process_ui_queue(self): try: while True: event = self.ui_queue.get_nowait() kind = event[0] if kind == "log": line = event[1] self.log_text.insert(tk.END, f"{line}\n") self.log_text.see(tk.END) elif kind == "clear_log": self.log_text.delete(1.0, tk.END) elif kind == "stats": self.stats_var.set(f"成功: {event[1]} | 失败: {event[2]} | 待恢复: {event[3]} | 后处理警告: {event[4]}") elif kind == "running": running = bool(event[1]) self.start_btn.config(state=tk.DISABLED if running else tk.NORMAL) self.stop_btn.config(state=tk.NORMAL if running else tk.DISABLED) self.status_var.set("运行中..." if running else "就绪") self.status_label.config(foreground="blue" if running else "green") elif kind == "error": messagebox.showerror(event[1], event[2]) except queue.Empty: pass except Exception as exc: print(f"[!] UI 队列处理失败: {exc}", file=sys.stderr) finally: try: self.root.after(50, self.process_ui_queue) except Exception: pass def log(self, message): timestamp = datetime.datetime.now().strftime("%H:%M:%S") line = f"[{timestamp}] {message}" print(line, flush=True) self.ui_queue.put(("log", line)) def clear_log(self): self.ui_queue.put(("clear_log",)) def update_stats(self): self.ui_queue.put(("stats", self.success_count, self.fail_count, self.registered_unsaved_count, self.postprocess_warning_count)) def _set_running_ui(self, running): self.is_running = bool(running) self.ui_queue.put(("running", self.is_running)) def should_stop(self): return self.stop_requested or not self.is_running def start_registration(self): if self.is_running: self.log("[!] 当前已有任务在运行") return config["email_provider"] = self.email_provider_var.get().strip() or "duckmail" config["enable_nsfw"] = bool(self.nsfw_var.get()) config["proxy"] = self.proxy_var.get().strip() config["duckmail_api_key"] = self.api_key_var.get().strip() config["cloudflare_api_base"] = self.cloudflare_api_base_var.get().strip() config["cloudflare_api_key"] = self.cloudflare_api_key_var.get().strip() config["cloudflare_auth_mode"] = self.cloudflare_auth_mode_var.get().strip() or "none" config["cloudmail_api_base"] = self.cloudmail_api_base_var.get().strip() config["cloudmail_public_token"] = self.cloudmail_public_token_var.get().strip() config["cloudmail_domains"] = self.cloudmail_domains_var.get().strip() config["grok2api_auto_add_local"] = bool(self.grok2api_local_auto_var.get()) config["grok2api_local_token_file"] = self.grok2api_local_file_var.get().strip() config["grok2api_pool_name"] = self.grok2api_pool_name_var.get().strip() or "ssoBasic" config["grok2api_auto_add_remote"] = bool(self.grok2api_remote_auto_var.get()) config["grok2api_remote_base"] = self.grok2api_remote_base_var.get().strip() config["grok2api_remote_app_key"] = self.grok2api_remote_key_var.get().strip() config["cpa_export_enabled"] = bool(self.cpa_export_var.get()) config["cpa_auth_dir"] = self.cpa_auth_dir_var.get().strip() or "./cpa_auths" raw_paths = [x.strip() for x in self.cloudflare_paths_var.get().split(",") if x.strip()] if len(raw_paths) >= 4: config["cloudflare_path_domains"] = raw_paths[0] if raw_paths[0].startswith("/") else ("/" + raw_paths[0]) config["cloudflare_path_accounts"] = raw_paths[1] if raw_paths[1].startswith("/") else ("/" + raw_paths[1]) config["cloudflare_path_token"] = raw_paths[2] if raw_paths[2].startswith("/") else ("/" + raw_paths[2]) config["cloudflare_path_messages"] = raw_paths[3] if raw_paths[3].startswith("/") else ("/" + raw_paths[3]) try: count = int(self.count_var.get()) config["register_count"] = count validated = validate_run_requirements(config) config.clear() config.update(validated) save_config() except (ValueError, ConfigError) as exc: self.log(f"[!] 配置无效或保存失败: {exc}") return self.stop_requested = False self.success_count = 0 self.fail_count = 0 self.results = [] now = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") self.accounts_output_file = os.path.join( os.path.dirname(__file__), f"accounts_{now}.txt" ) self.update_stats() self._set_running_ui(True) self.log(f"[*] 配置已保存，开始执行。目标数量: {count}") self.log(f"[*] 成功账号将实时保存到: {self.accounts_output_file}") threading.Thread( target=self.run_registration, args=(count,), daemon=True, ).start() def stop_registration(self): self.stop_requested = True self.log("[!] 用户停止注册") def run_registration(self, count): def observer(batch, account, output): self.success_count = batch.success_count self.fail_count = batch.fail_count self.registered_unsaved_count = batch.registered_unsaved_count self.postprocess_warning_count = batch.postprocess_warning_count if account is not None: self.results.append({"email": account.email, "sso": account.sso, "profile": account.profile, "output": output}) self.update_stats() try: batch = run_registration_common( count=count, log_callback=self.log, cancel_callback=self.should_stop, accounts_output_file=self.accounts_output_file, observer=observer, ) self.success_count = batch.success_count self.fail_count = batch.fail_count self.registered_unsaved_count = batch.registered_unsaved_count self.postprocess_warning_count = batch.postprocess_warning_count self.update_stats() except Exception as exc: log_exception("任务异常", exc, self.log) finally: self._set_running_ui(False) self.log("[*] 任务结束") class CliStopController: def __init__(self): self.stop_requested = False def should_stop(self): return self.stop_requested def stop(self): self.stop_requested = True def cli_log(message): timestamp = datetime.datetime.now().strftime("%H:%M:%S") print(f"[{timestamp}] {message}", flush=True) def run_registration_cli(count): controller = CliStopController() accounts_output_file = os.path.join( os.path.dirname(__file__), f"accounts_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.txt", ) cli_log(f"[*] 终端模式启动，目标数量: {count}") cli_log(f"[*] 成功账号将实时保存到: {accounts_output_file}") last_stats = {"success": 0, "fail": 0, "pending": 0, "warnings": 0} def observer(batch, account, output): last_stats["success"] = batch.success_count last_stats["fail"] = batch.fail_count last_stats["pending"] = batch.registered_unsaved_count last_stats["warnings"] = batch.postprocess_warning_count cli_log(f"[*] 当前统计: 成功 {batch.success_count} | 失败 {batch.fail_count} | 待恢复 {batch.registered_unsaved_count} | 后处理警告 {batch.postprocess_warning_count}") try: batch = run_registration_common( count=count, log_callback=cli_log, cancel_callback=controller.should_stop, accounts_output_file=accounts_output_file, observer=observer, ) last_stats["success"] = batch.success_count last_stats["fail"] = batch.fail_count last_stats["pending"] = batch.registered_unsaved_count last_stats["warnings"] = batch.postprocess_warning_count except KeyboardInterrupt: controller.stop() cli_log("[!] 收到 Ctrl+C，正在停止并清理") except Exception as exc: log_exception("任务异常", exc, cli_log) finally: cli_log(f"[*] 任务结束。成功 {last_stats['success']} | 失败 {last_stats['fail']} | 待恢复 {last_stats['pending']} | 后处理警告 {last_stats['warnings']}") def main_cli(): try: load_config() except ConfigError as exc: cli_log(f"[!] {exc}") return try: validated = validate_run_requirements(config) config.clear() config.update(validated) except ConfigError as exc: cli_log(f"[!] {exc}") return count = int(config.get("register_count", 1) or 1) cli_log("[*] CLI 已加载配置") cli_log(f"[*] 当前邮箱服务商: {config.get('email_provider', 'duckmail')} | 注册数量: {count}") cli_log("[*] 输入 start 后开始；按 Ctrl+C 可强制停止") try: command = input("> ").strip().lower() except KeyboardInterrupt: cli_log("[!] 已取消") return if command != "start": cli_log("[!] 未输入 start，已退出") return run_registration_cli(count) def main(): if len(sys.argv) > 1 and sys.argv[1].strip().lower() == "retry-pending": if len(sys.argv) < 3: print("用法: python grok_register_ttk.py retry-pending [输出文件]", file=sys.stderr) return try: summary = retry_pending_file( sys.argv[2], output_path=sys.argv[3] if len(sys.argv) > 3 else None, log_callback=cli_log, ) cli_log( f"[*] pending 恢复完成: 已恢复 {summary['restored']} | 剩余 {summary['remaining']} | 输出 {summary['output_path']}" ) except Exception as exc: log_exception("pending 恢复失败", exc, cli_log) return if len(sys.argv) > 1 and sys.argv[1].strip().lower() in ("start", "cli", "--cli"): main_cli() return if not TK_AVAILABLE: print(f"[!] GUI 模式需要 Tkinter，但当前环境不可用: {TK_IMPORT_ERROR}", file=sys.stderr) print("[*] 可改用 CLI 模式: python grok_register_ttk.py cli", file=sys.stderr) return root = tk.Tk() setup_light_theme(root) try: app = GrokRegisterGUI(root) except ConfigError as exc: print(f"[!] {exc}", file=sys.stderr) try: messagebox.showerror("配置错误", str(exc)) except Exception: pass root.destroy() return root.mainloop() if __name__ == "__main__": main()