Update register.py

register.py

@@ -1,242 +1,243 @@
-"""
-Gemini Business 自动注册模块 (Docker / HF Space 版)
-"""
-
-# ================= 标准库 =================
-import json
-import time
-import random
-from pathlib import Path
-from datetime import datetime
-from urllib.parse import urlparse, parse_qs
-from concurrent.futures import ThreadPoolExecutor
-
-# ================= 第三方 =================
-import requests
-from bs4 import BeautifulSoup
-import undetected_chromedriver as uc
-from selenium.webdriver.common.by import By
-from selenium.webdriver.common.keys import Keys
-from selenium.webdriver.support.ui import WebDriverWait
-from selenium.webdriver.support import expected_conditions as EC
-
-# ================= 项目内 =================
-from logger import log
-
-
-# ==================== 配置 ====================
-LOGIN_URL = "https://auth.business.gemini.google/login?continueUrl=https:%2F%2Fbusiness.gemini.google%2F"
-
-MAIL_API = "https://mail.chatgpt.org.uk"
-MAIL_KEY = "gpt-test"
-
-ACCOUNTS_FILE = "accounts.json"
-
-HEADLESS_MODE = True
-
-XPATH = {
-    "email_input": "//input[@type='email']",
-    "continue_btn": "//button",
-}
-
-NAMES = [
-    "James Smith", "John Johnson", "Robert Williams", "Michael Brown",
-    "William Jones", "David Garcia", "Mary Miller", "Patricia Davis",
-    "Jennifer Rodriguez", "Linda Martinez", "Elizabeth Taylor"
-]
-
-
-# ==================== 邮箱 ====================
-email_queue = []
-
-
-def create_temp_email():
-    try:
-        r = requests.get(
-            f"{MAIL_API}/api/generate-email",
-            headers={"X-API-Key": MAIL_KEY},
-            timeout=20
-        )
-        if r.status_code == 200 and r.json().get("success"):
-            return r.json()["data"]["email"]
-    except Exception as e:
-        log(f"邮箱创建失败: {e}", "ERROR")
-    return None
-
-
-def prefetch_email():
-    email = create_temp_email()
-    if email:
-        email_queue.append(email)
-
-
-def get_email():
-    if email_queue:
-        email = email_queue.pop(0)
-        log(f"使用预取邮箱: {email}")
-        return email
-    email = create_temp_email()
-    log(f"新邮箱: {email}")
-    return email
-
-
-def fetch_verification_code(email, timeout=90):
-    log("等待验证码...")
-    start = time.time()
-    while time.time() - start < timeout:
-        try:
-            r = requests.get(
-                f"{MAIL_API}/api/emails",
-                params={"email": email},
-                headers={"X-API-Key": MAIL_KEY},
-                timeout=10
-            )
-            if r.status_code == 200:
-                emails = r.json().get("data", {}).get("emails", [])
-                if emails:
-                    soup = BeautifulSoup(
-                        emails[0].get("html_content", ""),
-                        "html.parser"
-                    )
-                    span = soup.find("span", class_="verification-code")
-                    if span:
-                        code = span.text.strip()
-                        if len(code) == 6:
-                            log(f"验证码: {code}", "OK")
-                            return code
-        except:
-            pass
-        time.sleep(2)
-
-    log("验证码超时", "ERROR")
-    return None
-
-
-# ==================== 浏览器 ====================
-def create_driver():
-    options = uc.ChromeOptions()
-    options.add_argument("--headless=new")
-    options.add_argument("--no-sandbox")
-    options.add_argument("--disable-dev-shm-usage")
-    options.add_argument("--disable-gpu")
-    options.add_argument("--window-size=1280,800")
-
-    return uc.Chrome(options=options, use_subprocess=True)
-
-
-# ==================== 保存配置 ====================
-def save_account_config(email, driver):
-    cookies = driver.get_cookies()
-    url = driver.current_url
-    parsed = urlparse(url)
-
-    config_id = None
-    for p in url.split("/"):
-        if len(p) > 10 and p.isalnum():
-            config_id = p
-
-    cookie_map = {c["name"]: c["value"] for c in cookies}
-
-    csesidx = parse_qs(parsed.query).get("csesidx", [None])[0]
-
-    if not all([config_id, csesidx]):
-        log("配置提取失败", "ERROR")
-        return None
-
-    data = {
-        "id": email,
-        "config_id": config_id,
-        "csesidx": csesidx,
-        "cookies": cookie_map,
-        "expires_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-    }
-
-    accounts = []
-    if Path(ACCOUNTS_FILE).exists():
-        accounts = json.load(open(ACCOUNTS_FILE, "r", encoding="utf-8"))
-
-    accounts.append(data)
-    json.dump(accounts, open(ACCOUNTS_FILE, "w", encoding="utf-8"), indent=2)
-
-    log(f"账号保存成功: {email}", "OK")
-    return data
-
-
-# ==================== 注册单个 ====================
-def register_single(driver, executor):
-    email = get_email()
-    wait = WebDriverWait(driver, 30)
-
-    try:
-        driver.get(LOGIN_URL)
-        time.sleep(2)
-
-        inp = wait.until(EC.element_to_be_clickable((By.XPATH, XPATH["email_input"])))
-        inp.clear()
-        inp.send_keys(email)
-        inp.send_keys(Keys.ENTER)
-
-        executor.submit(prefetch_email)
-
-        code = fetch_verification_code(email)
-        if not code:
-            return False
-
-        time.sleep(1)
-        driver.switch_to.active_element.send_keys(code)
-
-        time.sleep(2)
-
-        name = random.choice(NAMES)
-        driver.switch_to.active_element.send_keys(name)
-        driver.switch_to.active_element.send_keys(Keys.ENTER)
-
-        time.sleep(5)
-
-        save_account_config(email, driver)
-        log(f"注册成功: {email}", "OK")
-        return True
-
-    except Exception as e:
-        log(f"注册异常: {e}", "ERROR")
-        return False
-
-
-# ==================== 批量注册 ====================
-def run_batch_registration(target=5):
-    if Path(ACCOUNTS_FILE).exists():
-        Path(ACCOUNTS_FILE).unlink()
-
-    driver = create_driver()
-    executor = ThreadPoolExecutor(max_workers=2)
-
-    success = 0
-    attempts = 0
-
-    executor.submit(prefetch_email)
-
-    while success < target:
-        attempts += 1
-        log(f"开始注册第 {attempts} 次")
-        ok = register_single(driver, executor)
-        if ok:
-            success += 1
-        time.sleep(random.randint(2, 4))
-
-    driver.quit()
-    executor.shutdown(wait=False)
-
-    return {
-        "success": success,
-        "attempts": attempts,
-        "is_ok": success > 0
-    }
-
-
-# ==================== 对外接口 ====================
-def handle_task_execution(count, uploader):
-    stats = run_batch_registration(count)
-    if stats["is_ok"]:
-        uploader.login()
-        uploader.upload_and_merge(ACCOUNTS_FILE)
-    return stats
+"""
+Gemini Business 自动注册模块 (Docker / HF Space 版)
+"""
+
+# ================= 标准库 =================
+import json
+import time
+import random
+from pathlib import Path
+from datetime import datetime
+from urllib.parse import urlparse, parse_qs
+from concurrent.futures import ThreadPoolExecutor
+
+# ================= 第三方 =================
+import requests
+from bs4 import BeautifulSoup
+import undetected_chromedriver as uc
+from selenium.webdriver.common.by import By
+from selenium.webdriver.common.keys import Keys
+from selenium.webdriver.support.ui import WebDriverWait
+from selenium.webdriver.support import expected_conditions as EC
+
+# ================= 项目内 =================
+from logger import log
+
+
+# ==================== 配置 ====================
+LOGIN_URL = "https://auth.business.gemini.google/login?continueUrl=https:%2F%2Fbusiness.gemini.google%2F"
+
+MAIL_API = "https://mail.chatgpt.org.uk"
+MAIL_KEY = "gpt-test"
+
+ACCOUNTS_FILE = "accounts.json"
+
+HEADLESS_MODE = False
+# HEADLESS_MODE = True
+
+XPATH = {
+    "email_input": "//input[@type='email']",
+    "continue_btn": "//button",
+}
+
+NAMES = [
+    "James Smith", "John Johnson", "Robert Williams", "Michael Brown",
+    "William Jones", "David Garcia", "Mary Miller", "Patricia Davis",
+    "Jennifer Rodriguez", "Linda Martinez", "Elizabeth Taylor"
+]
+
+
+# ==================== 邮箱 ====================
+email_queue = []
+
+
+def create_temp_email():
+    try:
+        r = requests.get(
+            f"{MAIL_API}/api/generate-email",
+            headers={"X-API-Key": MAIL_KEY},
+            timeout=20
+        )
+        if r.status_code == 200 and r.json().get("success"):
+            return r.json()["data"]["email"]
+    except Exception as e:
+        log(f"邮箱创建失败: {e}", "ERROR")
+    return None
+
+
+def prefetch_email():
+    email = create_temp_email()
+    if email:
+        email_queue.append(email)
+
+
+def get_email():
+    if email_queue:
+        email = email_queue.pop(0)
+        log(f"使用预取邮箱: {email}")
+        return email
+    email = create_temp_email()
+    log(f"新邮箱: {email}")
+    return email
+
+
+def fetch_verification_code(email, timeout=90):
+    log("等待验证码...")
+    start = time.time()
+    while time.time() - start < timeout:
+        try:
+            r = requests.get(
+                f"{MAIL_API}/api/emails",
+                params={"email": email},
+                headers={"X-API-Key": MAIL_KEY},
+                timeout=10
+            )
+            if r.status_code == 200:
+                emails = r.json().get("data", {}).get("emails", [])
+                if emails:
+                    soup = BeautifulSoup(
+                        emails[0].get("html_content", ""),
+                        "html.parser"
+                    )
+                    span = soup.find("span", class_="verification-code")
+                    if span:
+                        code = span.text.strip()
+                        if len(code) == 6:
+                            log(f"验证码: {code}", "OK")
+                            return code
+        except:
+            pass
+        time.sleep(2)
+
+    log("验证码超时", "ERROR")
+    return None
+
+
+# ==================== 浏览器 ====================
+def create_driver():
+    options = uc.ChromeOptions()
+    options.add_argument("--headless=new")
+    options.add_argument("--no-sandbox")
+    options.add_argument("--disable-dev-shm-usage")
+    options.add_argument("--disable-gpu")
+    options.add_argument("--window-size=1280,800")
+
+    return uc.Chrome(options=options, use_subprocess=True)
+
+
+# ==================== 保存配置 ====================
+def save_account_config(email, driver):
+    cookies = driver.get_cookies()
+    url = driver.current_url
+    parsed = urlparse(url)
+
+    config_id = None
+    for p in url.split("/"):
+        if len(p) > 10 and p.isalnum():
+            config_id = p
+
+    cookie_map = {c["name"]: c["value"] for c in cookies}
+
+    csesidx = parse_qs(parsed.query).get("csesidx", [None])[0]
+
+    if not all([config_id, csesidx]):
+        log("配置提取失败", "ERROR")
+        return None
+
+    data = {
+        "id": email,
+        "config_id": config_id,
+        "csesidx": csesidx,
+        "cookies": cookie_map,
+        "expires_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    }
+
+    accounts = []
+    if Path(ACCOUNTS_FILE).exists():
+        accounts = json.load(open(ACCOUNTS_FILE, "r", encoding="utf-8"))
+
+    accounts.append(data)
+    json.dump(accounts, open(ACCOUNTS_FILE, "w", encoding="utf-8"), indent=2)
+
+    log(f"账号保存成功: {email}", "OK")
+    return data
+
+
+# ==================== 注册单个 ====================
+def register_single(driver, executor):
+    email = get_email()
+    wait = WebDriverWait(driver, 30)
+
+    try:
+        driver.get(LOGIN_URL)
+        time.sleep(2)
+
+        inp = wait.until(EC.element_to_be_clickable((By.XPATH, XPATH["email_input"])))
+        inp.clear()
+        inp.send_keys(email)
+        inp.send_keys(Keys.ENTER)
+
+        executor.submit(prefetch_email)
+
+        code = fetch_verification_code(email)
+        if not code:
+            return False
+
+        time.sleep(1)
+        driver.switch_to.active_element.send_keys(code)
+
+        time.sleep(2)
+
+        name = random.choice(NAMES)
+        driver.switch_to.active_element.send_keys(name)
+        driver.switch_to.active_element.send_keys(Keys.ENTER)
+
+        time.sleep(5)
+
+        save_account_config(email, driver)
+        log(f"注册成功: {email}", "OK")
+        return True
+
+    except Exception as e:
+        log(f"注册异常: {e}", "ERROR")
+        return False
+
+
+# ==================== 批量注册 ====================
+def run_batch_registration(target=5):
+    if Path(ACCOUNTS_FILE).exists():
+        Path(ACCOUNTS_FILE).unlink()
+
+    driver = create_driver()
+    executor = ThreadPoolExecutor(max_workers=2)
+
+    success = 0
+    attempts = 0
+
+    executor.submit(prefetch_email)
+
+    while success < target:
+        attempts += 1
+        log(f"开始注册第 {attempts} 次")
+        ok = register_single(driver, executor)
+        if ok:
+            success += 1
+        time.sleep(random.randint(2, 4))
+
+    driver.quit()
+    executor.shutdown(wait=False)
+
+    return {
+        "success": success,
+        "attempts": attempts,
+        "is_ok": success > 0
+    }
+
+
+# ==================== 对外接口 ====================
+def handle_task_execution(count, uploader):
+    stats = run_batch_registration(count)
+    if stats["is_ok"]:
+        uploader.login()
+        uploader.upload_and_merge(ACCOUNTS_FILE)
+    return stats