Create nb2.py

nb2.py

@@ -0,0 +1,331 @@
+from selenium import webdriver
+from selenium.webdriver.chrome.service import Service
+from selenium.webdriver.chrome.options import Options
+from selenium.webdriver.common.by import By
+from selenium.webdriver.support.ui import WebDriverWait
+from selenium.webdriver.support import expected_conditions as EC
+from selenium.common.exceptions import TimeoutException
+from selenium.webdriver.common.keys import Keys
+import time
+import random
+import string
+import logging
+import requests
+import json
+import re
+import os
+from fastapi import FastAPI
+import uvicorn
+
+# 配置日志
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger(__name__)
+
+# FastAPI应用
+app = FastAPI()
+
+class MailGw:
+    def __init__(self):
+        self.base_url = "https://api.mail.gw"
+        self.token = None
+        self.email = None
+        self.password = "Temp123!@#"
+
+    def create_email(self):
+        try:
+            # 获取可用域名列表
+            domains_response = requests.get(f"{self.base_url}/domains")
+            domains = domains_response.json()['hydra:member']
+            
+            # 打印所有可用域名
+            logger.info(f"可用域名列表: {[d['domain'] for d in domains]}")
+            
+            # 随机选择一个域名
+            domain = random.choice(domains)['domain']
+            logger.info(f"选择的域名: {domain}")
+
+            # 生成随机邮箱地址
+            random_name = ''.join(random.choices(string.ascii_lowercase + string.digits, k=10))
+            email = f"{random_name}@{domain}"
+
+            # 创建邮箱账户
+            account_response = requests.post(
+                f"{self.base_url}/accounts",
+                json={
+                    "address": email,
+                    "password": self.password
+                }
+            )
+
+            if account_response.status_code != 201:
+                logger.error(f"创建邮箱失败: {account_response.text}")
+                return None
+
+            self.email = email
+
+            # 获取token
+            token_response = requests.post(
+                f"{self.base_url}/token",
+                json={
+                    "address": email,
+                    "password": self.password
+                }
+            )
+
+            if token_response.status_code != 200:
+                logger.error(f"获取token失败: {token_response.text}")
+                return None
+
+            self.token = token_response.json()['token']
+            logger.info(f"创建邮箱成功: {email}")
+            return email
+
+        except Exception as e:
+            logger.error(f"创建邮箱时发生错误: {str(e)}")
+            return None
+
+    def get_verification_code(self, max_retries=10, delay=2):
+        if not self.token:
+            logger.error("Token不存在")
+            return None
+
+        headers = {"Authorization": f"Bearer {self.token}"}
+
+        for _ in range(max_retries):
+            try:
+                # 获取邮件列表
+                messages_response = requests.get(
+                    f"{self.base_url}/messages",
+                    headers=headers
+                )
+
+                messages = messages_response.json()['hydra:member']
+
+                if messages:
+                    # 获取最新邮件的ID
+                    message_id = messages[0]['id']
+                    
+                    # 获取邮件内容
+                    message_response = requests.get(
+                        f"{self.base_url}/messages/{message_id}",
+                        headers=headers
+                    )
+                    
+                    # 从邮件内容中提取验证码
+                    text = message_response.json()['text']
+                    
+                    # 使用正则表达式匹配6位数字验证码
+                    match = re.search(r'\b\d{6}\b', text)
+                    if match:
+                        verification_code = match.group(0)
+                        logger.info(f"获取到验证码: {verification_code}")
+                        return verification_code
+
+            except Exception as e:
+                logger.error(f"获取验证码时发生错误: {str(e)}")
+
+            time.sleep(delay)
+            logger.info(f"等待验证码，重试第 {_ + 1} 次")
+
+        logger.error("未能获取到验证码")
+        return None
+
+def setup_driver():
+    """设置Chrome驱动"""
+    chrome_options = Options()
+    chrome_options.add_argument('--headless')
+    chrome_options.add_argument('--no-sandbox')
+    chrome_options.add_argument('--disable-dev-shm-usage')
+    chrome_options.binary_location = os.getenv('CHROME_BIN', '/usr/bin/chromium')
+    
+    service = Service(os.getenv('CHROMEDRIVER_PATH', '/usr/bin/chromedriver'))
+    driver = webdriver.Chrome(service=service, options=chrome_options)
+    return driver
+
+def get_cookie(driver):
+    try:
+        cookie_script = """
+        var cookieString = document.cookie;
+        var essentialCookies = cookieString.split('; ').filter(function(cookie) {
+            return cookie.startsWith('daily_query_') || 
+                   cookie.startsWith('DSR=') || 
+                   cookie.startsWith('DS=') || 
+                   cookie.startsWith('uuid_guest=') ||
+                   cookie.startsWith('ai_model=');
+        }).join('; ');
+        console.log('Essential Cookies:', essentialCookies);
+        return essentialCookies;
+        """
+        cookie = driver.execute_script(cookie_script)
+        
+        logger.info(f"获取的 Cookie: {cookie}")
+        
+        with open('cookies.txt', 'a') as file:
+            file.write(f"    - '{cookie}'\n")
+        
+        logger.info("Cookie 已追加到 cookies.txt 文件")
+        return cookie
+    except Exception as e:
+        logger.error(f"获取cookie错误: {str(e)}")
+        return None
+
+def wait_for_url(driver, url, timeout=10):
+    try:
+        start_time = time.time()
+        while time.time() - start_time < timeout:
+            current_url = driver.current_url.rstrip('/')
+            target_url = url.rstrip('/')
+            if current_url == target_url:
+                return True
+            time.sleep(0.5)
+        return False
+    except TimeoutException:
+        return False
+
+def wait_for_email_input(driver, timeout=10):
+    """等待邮箱输入框出现并返回"""
+    start_time = time.time()
+    while time.time() - start_time < timeout:
+        try:
+            input_js = """
+            return document.querySelector("#AppProvider_Wrapper > form > descope-wc")
+                .shadowRoot.querySelector("#TExZpnv5m6")
+                .shadowRoot.querySelector("#input-vaadin-email-field-3")
+            """
+            email_input = driver.execute_script(input_js)
+            if email_input:
+                logger.info("邮箱输入框已加载")
+                return True
+        except Exception:
+            pass
+        time.sleep(0.5)
+    logger.error("邮箱输入框未出现")
+    return False
+
+def input_email(driver, email):
+    """使用JavaScript输入邮箱"""
+    try:
+        input_js = f"""
+        var input = document.querySelector("#AppProvider_Wrapper > form > descope-wc")
+            .shadowRoot.querySelector("#TExZpnv5m6")
+            .shadowRoot.querySelector("#input-vaadin-email-field-3");
+        input.value = "{email}";
+        input.dispatchEvent(new Event('input', {{ bubbles: true }}));
+        input.dispatchEvent(new Event('change', {{ bubbles: true }}));
+        """
+        driver.execute_script(input_js)
+        logger.info("输入邮箱成功")
+        return True
+    except Exception as e:
+        logger.error(f"输入邮箱失败: {str(e)}")
+        return False
+
+def input_verification_code(driver, code):
+    """使用JavaScript输入验证码"""
+    try:
+        input_js = f"""
+        var input = document.querySelector("#AppProvider_Wrapper > form > descope-wc")
+            .shadowRoot.querySelector("#oneTimeCodeId")
+            .shadowRoot.querySelector("#input-vaadin-text-field-7 > div.wrapper > descope-text-field:nth-child(1)")
+            .shadowRoot.querySelector("#input-vaadin-text-field-35");
+        input.value = "{code}";
+        input.dispatchEvent(new Event('input', {{ bubbles: true }}));
+        input.dispatchEvent(new Event('change', {{ bubbles: true }}));
+        """
+        driver.execute_script(input_js)
+        logger.info("输入验证码成功")
+        return True
+    except Exception as e:
+        logger.error(f"输入验证码失败: {str(e)}")
+        return False
+
+def registration_process(driver, wait):
+    try:
+        # 创建临时邮箱
+        mail_client = MailGw()
+        email = mail_client.create_email()
+        if not email:
+            logger.error("创建临时邮箱失败")
+            return False
+
+        # 第一步：打开认证页面
+        driver.get("https://you.com/authenticate")
+        logger.info("打开认证页面")
+
+        # 等待邮箱输入框出现
+        if not wait_for_email_input(driver):
+            logger.error("等待邮箱输入框超时")
+            return False
+
+        # 输入邮箱
+        if not input_email(driver, email):
+            logger.error("输入邮箱失败")
+            return False
+
+        time.sleep(0.5)
+
+        # 按回车继续
+        actions = webdriver.ActionChains(driver)
+        actions.send_keys(Keys.RETURN)
+        actions.perform()
+        logger.info("按下回车键")
+        time.sleep(2)
+
+        # 获取验证码
+        verification_code = mail_client.get_verification_code()
+        if not verification_code:
+            logger.error("获取验证码失败")
+            return False
+
+        logger.info(f"获取验证码: {verification_code}")
+
+        # 输入验证码
+        if not input_verification_code(driver, verification_code):
+            logger.error("输入验证码失败")
+            return False
+
+        time.sleep(2)
+
+        # 等待URL变化并获取cookie
+        if wait_for_url(driver, "https://you.com"):
+            logger.info("URL已变更为 https://you.com")
+            time.sleep(1)
+            cookie = get_cookie(driver)
+            if cookie:
+                logger.info("成功获取cookie")
+                return cookie
+            else:
+                logger.error("获取cookie失败")
+                return None
+        else:
+            logger.error("URL未变化为预期值")
+            return None
+
+    except Exception as e:
+        logger.error(f"注册过程错误: {str(e)}")
+        return None
+
+@app.get("/")
+def read_root():
+    return {"status": "running"}
+
+@app.get("/generate_cookie")
+async def generate_cookie():
+    try:
+        driver = setup_driver()
+        wait = WebDriverWait(driver, 20)
+        cookie = registration_process(driver, wait)
+        driver.quit()
+        
+        if cookie:
+            return {"status": "success", "cookie": cookie}
+        else:
+            return {"status": "error", "message": "Failed to generate cookie"}
+    except Exception as e:
+        return {"status": "error", "message": str(e)}
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=7860)