Upload 18 files

core/config.py

@@ -49,7 +49,8 @@ class BasicConfig(BaseModel):
     duckmail_base_url: str = Field(default="https://api.duckmail.sbs", description="DuckMail API地址")
     duckmail_api_key: str = Field(default="", description="DuckMail API key")
     duckmail_verify_ssl: bool = Field(default=True, description="DuckMail SSL校验")
-    browser_headless: bool = Field(default=True, description="自动化浏览器无头模式")
+    browser_engine: str = Field(default="dp", description="浏览器引擎：uc (支持无头/有头) 或 dp (仅有头，更稳定)")
+    browser_headless: bool = Field(default=False, description="自动化浏览器无头模式（仅 UC 引擎支持）")
     refresh_window_hours: int = Field(default=1, ge=0, le=24, description="过期刷新窗口（小时）")
     register_default_count: int = Field(default=1, ge=1, le=30, description="默认注册数量")
     register_domain: str = Field(default="", description="默认注册域名（推荐）")
@@ -153,7 +154,8 @@ class ConfigManager:
             duckmail_base_url=basic_data.get("duckmail_base_url") or "https://api.duckmail.sbs",
             duckmail_api_key=str(duckmail_api_key_raw or "").strip(),
             duckmail_verify_ssl=_parse_bool(basic_data.get("duckmail_verify_ssl"), True),
-            browser_headless=_parse_bool(basic_data.get("browser_headless"), True),
+            browser_engine=basic_data.get("browser_engine") or "dp",
+            browser_headless=_parse_bool(basic_data.get("browser_headless"), False),
             refresh_window_hours=int(refresh_window_raw),
             register_default_count=int(register_default_raw),
             register_domain=str(register_domain_raw or "").strip(),

core/duckmail_client.py

@@ -130,6 +130,7 @@ class DuckMailClient:
                 return None
 
         try:
+            self._log("info", "fetching verification code")
             # 获取邮件列表
             res = self._request(
                 "GET",
@@ -138,47 +139,19 @@ class DuckMailClient:
             )
 
             if res.status_code != 200:
-                self._log("warning", f"DuckMail messages request failed: {res.status_code}")
                 return None
 
             data = res.json() if res.content else {}
             messages = data.get("hydra:member", [])
-            self._log("info", f"DuckMail messages count: {len(messages)}")
 
             if not messages:
                 return None
 
-            # 获取第一封邮件的详情（最新的邮件）
+            # 只获取最新一封邮件，不做时间过滤
             msg_id = messages[0].get("id")
-            msg_created_at = messages[0].get("createdAt", "unknown")
             if not msg_id:
                 return None
 
-            self._log("info", f"DuckMail fetching message: {msg_id} (created: {msg_created_at})")
-
-            # 检查邮件时间是否在 since_time 之后
-            self._log("info", f"DuckMail since_time check: since_time={since_time}, msg_created_at={msg_created_at}")
-            if since_time and msg_created_at != "unknown":
-                try:
-                    from dateutil import parser
-                    email_time = parser.parse(msg_created_at)
-                    # 移除时区信息进行比较
-                    if email_time.tzinfo:
-                        email_time = email_time.replace(tzinfo=None)
-                    if since_time.tzinfo:
-                        since_time = since_time.replace(tzinfo=None)
-
-                    self._log("info", f"DuckMail comparing times: email={email_time}, since={since_time}")
-                    if email_time < since_time:
-                        self._log("info", f"DuckMail email too old: {email_time} < {since_time}")
-                        return None
-                    else:
-                        self._log("info", f"DuckMail email is new: {email_time} >= {since_time}")
-                except Exception as e:
-                    self._log("warning", f"DuckMail time comparison failed: {e}")
-            else:
-                self._log("info", f"DuckMail skipping time check (since_time={since_time}, msg_created_at={msg_created_at})")
-
             detail = self._request(
                 "GET",
                 f"{self.base_url}/messages/{msg_id}",
@@ -189,34 +162,24 @@ class DuckMailClient:
                 return None
 
             payload = detail.json() if detail.content else {}
-            subject = payload.get("subject", "")
-            created_at = payload.get("createdAt", "unknown")
-            self._log("info", f"DuckMail message subject: {subject} (created: {created_at})")
 
-            # 获取邮件内容（text可能是字符串，html可能是列表）
+            # 获取邮件内容
             text_content = payload.get("text") or ""
             html_content = payload.get("html") or ""
 
-            # 如果html是列表，转换为字符串
             if isinstance(html_content, list):
                 html_content = "".join(str(item) for item in html_content)
             if isinstance(text_content, list):
                 text_content = "".join(str(item) for item in text_content)
 
             content = text_content + html_content
-            self._log("info", f"DuckMail email content length: {len(content)} chars")
             code = extract_verification_code(content)
             if code:
-                self._log("info", f"DuckMail extracted code: {code}")
-            else:
-                self._log("warning", f"DuckMail no code found in message")
-                # 打印部分内容用于调试
-                preview = content[:200] if content else "(empty)"
-                self._log("warning", f"DuckMail content preview: {preview}")
+                self._log("info", f"code found: {code}")
             return code
 
         except Exception as e:
-            self._log("error", f"DuckMail fetch code failed: {e}")
+            self._log("error", f"fetch code failed: {e}")
             return None
 
     def poll_for_code(
@@ -228,23 +191,19 @@ class DuckMailClient:
         """轮询获取验证码"""
         if not self.token:
             if not self.login():
-                self._log("error", "DuckMail token missing")
                 return None
 
-        self._log("info", "DuckMail polling for code")
         max_retries = timeout // interval
 
         for i in range(1, max_retries + 1):
-            self._log("info", f"DuckMail attempt {i}/{max_retries}")
             code = self.fetch_verification_code(since_time=since_time)
             if code:
-                self._log("info", f"DuckMail code found: {code}")
                 return code
 
             if i < max_retries:
                 time.sleep(interval)
 
-        self._log("error", "DuckMail code timeout")
+        self._log("error", "verification code timeout")
         return None
 
     def _get_domain(self) -> str:

core/gemini_automation.py

@@ -132,7 +132,11 @@ class GeminiAutomation:
         if has_business_params:
             return self._extract_config(page, email)
 
-        # Step 3: 尝试触发发送验证码
+        # Step 3: 记录发送验证码的时间并触发发送
+        from datetime import datetime
+        send_time = datetime.now()
+
+        self._log("info", "clicking send verification code button")
         if not self._click_send_code_button(page):
             self._log("error", "send code button not found")
             self._save_screenshot(page, "send_code_button_missing")
@@ -145,11 +149,9 @@ class GeminiAutomation:
             self._save_screenshot(page, "code_input_missing")
             return {"success": False, "error": "code input not found"}
 
-        time.sleep(3)
-
-        # Step 5: 轮询邮件获取验证码
+        # Step 5: 轮询邮件获取验证码（传入发送时间）
         self._log("info", "polling for verification code")
-        code = mail_client.poll_for_code(timeout=40, interval=4)
+        code = mail_client.poll_for_code(timeout=40, interval=4, since_time=send_time)
 
         if not code:
             self._log("error", "verification code timeout")
@@ -184,7 +186,8 @@ class GeminiAutomation:
         # Step 7: 处理协议页面（如果有）
         self._handle_agreement_page(page)
 
-        # Step 8: 导航到业务页面
+        # Step 8: 导航到业务页面并等待参数生成
+        self._log("info", "navigating to business page")
         page.get("https://business.gemini.google/", timeout=self.timeout)
         time.sleep(3)
 
@@ -193,14 +196,20 @@ class GeminiAutomation:
             if self._handle_username_setup(page):
                 time.sleep(3)
 
-        # Step 10: 提取配置
-        if "cid" in page.url or self._wait_for_cid(page):
-            self._log("info", "login success")
-            return self._extract_config(page, email)
-
-        self._log("error", "login failed")
-        self._save_screenshot(page, "login_failed")
-        return {"success": False, "error": "login failed"}
+        # Step 10: 等待 URL 参数生成（csesidx 和 cid）
+        self._log("info", "waiting for URL parameters")
+        if not self._wait_for_business_params(page):
+            self._log("warning", "URL parameters not generated, trying refresh")
+            page.refresh()
+            time.sleep(3)
+            if not self._wait_for_business_params(page):
+                self._log("error", "URL parameters generation failed")
+                self._save_screenshot(page, "params_missing")
+                return {"success": False, "error": "URL parameters not found"}
+
+        # Step 11: 提取配置
+        self._log("info", "login success")
+        return self._extract_config(page, email)
 
     def _click_send_code_button(self, page) -> bool:
         """点击发送验证码按钮（如果需要）"""
@@ -284,6 +293,16 @@ class GeminiAutomation:
             time.sleep(1)
         return False
 
+    def _wait_for_business_params(self, page, timeout: int = 30) -> bool:
+        """等待业务页面参数生成（csesidx 和 cid）"""
+        for _ in range(timeout):
+            url = page.url
+            if "csesidx=" in url and "/cid/" in url:
+                self._log("info", f"business params ready: {url}")
+                return True
+            time.sleep(1)
+        return False
+
     def _handle_username_setup(self, page) -> bool:
         """处理用户名设置页面"""
         current_url = page.url

core/gemini_automation_uc.py

@@ -130,6 +130,17 @@ class GeminiAutomationUC:
             self._save_screenshot("continue_button_failed")
             return {"success": False, "error": f"failed to click continue: {e}"}
 
+        # 记录发送验证码的时间
+        from datetime import datetime
+        send_time = datetime.now()
+
+        # 检查是否需要点击"发送验证码"按钮
+        self._log("info", "clicking send verification code button")
+        if not self._click_send_code_button():
+            self._log("error", "send code button not found")
+            self._save_screenshot("send_code_button_missing")
+            return {"success": False, "error": "send code button not found"}
+
         # 等待验证码输入框出现
         code_input = self._wait_for_code_input()
         if not code_input:
@@ -137,11 +148,9 @@ class GeminiAutomationUC:
             self._save_screenshot("code_input_missing")
             return {"success": False, "error": "code input not found"}
 
-        time.sleep(3)
-
-        # 获取验证码
+        # 获取验证码（传入发送时间）
         self._log("info", "polling for verification code")
-        code = mail_client.poll_for_code(timeout=40, interval=4)
+        code = mail_client.poll_for_code(timeout=40, interval=4, since_time=send_time)
 
         if not code:
             self._log("error", "verification code timeout")
@@ -192,7 +201,8 @@ class GeminiAutomationUC:
         # 处理协议页面
         self._handle_agreement_page()
 
-        # 导航到业务页面
+        # 导航到业务页面并等待参数生成
+        self._log("info", "navigating to business page")
         self.driver.get("https://business.gemini.google/")
         time.sleep(3)
 
@@ -201,14 +211,58 @@ class GeminiAutomationUC:
             if self._handle_username_setup():
                 time.sleep(3)
 
+        # 等待 URL 参数生成（csesidx 和 cid）
+        self._log("info", "waiting for URL parameters")
+        if not self._wait_for_business_params():
+            self._log("warning", "URL parameters not generated, trying refresh")
+            self.driver.refresh()
+            time.sleep(3)
+            if not self._wait_for_business_params():
+                self._log("error", "URL parameters generation failed")
+                self._save_screenshot("params_missing")
+                return {"success": False, "error": "URL parameters not found"}
+
         # 提取配置
-        if "cid" in self.driver.current_url or self._wait_for_cid():
-            self._log("info", "login success")
-            return self._extract_config(email)
+        self._log("info", "login success")
+        return self._extract_config(email)
+
+    def _click_send_code_button(self) -> bool:
+        """点击发送验证码按钮（如果需要）"""
+        time.sleep(2)
+
+        # 方法1: 直接通过ID查找
+        try:
+            direct_btn = WebDriverWait(self.driver, 5).until(
+                EC.element_to_be_clickable((By.ID, "sign-in-with-email"))
+            )
+            self.driver.execute_script("arguments[0].click();", direct_btn)
+            time.sleep(2)
+            return True
+        except TimeoutException:
+            pass
+
+        # 方法2: 通过关键词查找按钮
+        keywords = ["通过电子邮件发送验证码", "通过电子邮件发送", "email", "Email", "Send code", "Send verification", "Verification code"]
+        try:
+            buttons = self.driver.find_elements(By.TAG_NAME, "button")
+            for btn in buttons:
+                text = btn.text.strip() if btn.text else ""
+                if text and any(kw in text for kw in keywords):
+                    self.driver.execute_script("arguments[0].click();", btn)
+                    time.sleep(2)
+                    return True
+        except Exception:
+            pass
+
+        # 方法3: 检查是否已经在验证码输入页面
+        try:
+            code_input = self.driver.find_element(By.CSS_SELECTOR, "input[name='pinInput']")
+            if code_input:
+                return True
+        except NoSuchElementException:
+            pass
 
-        self._log("error", "login failed")
-        self._save_screenshot("login_failed")
-        return {"success": False, "error": "login failed"}
+        return False
 
     def _wait_for_code_input(self, timeout: int = 30):
         """等待验证码输入框出现"""
@@ -265,6 +319,16 @@ class GeminiAutomationUC:
             time.sleep(1)
         return False
 
+    def _wait_for_business_params(self, timeout: int = 30) -> bool:
+        """等待业务页面参数生成（csesidx 和 cid）"""
+        for _ in range(timeout):
+            url = self.driver.current_url
+            if "csesidx=" in url and "/cid/" in url:
+                self._log("info", f"business params ready: {url}")
+                return True
+            time.sleep(1)
+        return False
+
     def _handle_username_setup(self) -> bool:
         """处理用户名设置页面"""
         current_url = self.driver.current_url

core/login_service.py

@@ -152,12 +152,24 @@ class LoginService(BaseTaskService[LoginTask]):
         else:
             return {"success": False, "email": account_id, "error": f"unsupported mail provider: {mail_provider}"}
 
-        automation = GeminiAutomationUC(
-            user_agent=self.user_agent,
-            proxy=config.basic.proxy,
-            headless=config.basic.browser_headless,
-            log_callback=log_cb,
-        )
+        # 根据配置选择浏览器引擎
+        browser_engine = (config.basic.browser_engine or "dp").lower()
+        if browser_engine == "dp":
+            # DrissionPage 引擎：仅有头模式，更稳定
+            automation = GeminiAutomation(
+                user_agent=self.user_agent,
+                proxy=config.basic.proxy,
+                headless=False,  # DP 不支持无头模式
+                log_callback=log_cb,
+            )
+        else:
+            # undetected-chromedriver 引擎：支持有头和无头
+            automation = GeminiAutomationUC(
+                user_agent=self.user_agent,
+                proxy=config.basic.proxy,
+                headless=config.basic.browser_headless,
+                log_callback=log_cb,
+            )
         try:
             result = automation.login_and_extract(account_id, client)
         except Exception as exc:

core/mail_utils.py

@@ -8,7 +8,6 @@ def extract_verification_code(text: str) -> Optional[str]:
         return None
 
     # 策略1: 上下文关键词匹配（中英文冒号）
-    # 排除 CSS 样式值（如 14px, 16pt 等）
     context_pattern = r"(?:验证码|code|verification|passcode|pin).*?[:：]\s*([A-Za-z0-9]{4,8})\b"
     match = re.search(context_pattern, text, re.IGNORECASE)
     if match:
@@ -17,17 +16,14 @@ def extract_verification_code(text: str) -> Optional[str]:
         if not re.match(r"^\d+(?:px|pt|em|rem|vh|vw|%)$", candidate, re.IGNORECASE):
             return candidate
 
-    # 策略2: 6位数字
+    # 策略2: 6位字母数字混合（与测试代码一致，优先级提高）
+    match = re.search(r"[A-Z0-9]{6}", text)
+    if match:
+        return match.group(0)
+
+    # 策略3: 6位数字（降级为备选）
     digits = re.findall(r"\b\d{6}\b", text)
     if digits:
         return digits[0]
 
-    # 策略3: 6位字母数字混合
-    alphanumeric = re.findall(r"\b[A-Z0-9]{6}\b", text)
-    for candidate in alphanumeric:
-        has_letter = any(c.isalpha() for c in candidate)
-        has_digit = any(c.isdigit() for c in candidate)
-        if has_letter and has_digit:
-            return candidate
-
     return None

core/microsoft_mail_client.py

@@ -54,6 +54,8 @@ class MicrosoftMailClient:
     def fetch_verification_code(self, since_time: Optional[datetime] = None) -> Optional[str]:
         if not self.email:
             return None
+
+        self._log("info", "fetching verification code")
         token = self._get_access_token()
         if not token:
             return None
@@ -63,15 +65,14 @@ class MicrosoftMailClient:
         try:
             client.authenticate("XOAUTH2", lambda _: auth_string)
         except Exception as exc:
-            self._log("error", f"Microsoft IMAP auth failed: {exc}")
+            self._log("error", f"IMAP auth failed: {exc}")
             try:
                 client.logout()
             except Exception:
                 pass
             return None
 
-        search_since = since_time or (datetime.utcnow() - timedelta(minutes=5))
-        since_str = search_since.strftime("%d-%b-%Y")
+        search_since = since_time or (datetime.now() - timedelta(minutes=5))
 
         try:
             for mailbox in ("INBOX", "Junk"):
@@ -82,12 +83,13 @@ class MicrosoftMailClient:
                 except Exception:
                     continue
 
-                status, data = client.search(None, "SINCE", since_str)
+                # 搜索所有邮件
+                status, data = client.search(None, "ALL")
                 if status != "OK" or not data or not data[0]:
                     continue
 
-                ids = data[0].split()
-                ids = ids[-20:]
+                ids = data[0].split()[-5:]  # 只检查最近 5 封
+
                 for msg_id in reversed(ids):
                     status, msg_data = client.fetch(msg_id, "(RFC822)")
                     if status != "OK" or not msg_data:
@@ -102,12 +104,17 @@ class MicrosoftMailClient:
 
                     msg = message_from_bytes(raw_bytes)
                     msg_date = self._parse_message_date(msg.get("Date"))
+
+                    # 按时间过滤
                     if msg_date and msg_date < search_since:
                         continue
 
                     content = self._message_to_text(msg)
-                    code = extract_verification_code(content)
-                    if code:
+                    import re
+                    match = re.search(r'[A-Z0-9]{6}', content)
+                    if match:
+                        code = match.group(0)
+                        self._log("info", f"code found in {mailbox}: {code}")
                         return code
         finally:
             try:
@@ -127,12 +134,15 @@ class MicrosoftMailClient:
             return None
 
         max_retries = max(1, timeout // interval)
-        for _ in range(max_retries):
+
+        for i in range(1, max_retries + 1):
             code = self.fetch_verification_code(since_time=since_time)
             if code:
                 return code
-            if interval:
+            if i < max_retries:
                 time.sleep(interval)
+
+        self._log("error", "verification code timeout")
         return None
 
     @staticmethod

core/register_service.py

@@ -116,12 +116,24 @@ class RegisterService(BaseTaskService[RegisterTask]):
         if not client.register_account(domain=domain):
             return {"success": False, "error": "duckmail register failed"}
 
-        automation = GeminiAutomationUC(
-            user_agent=self.user_agent,
-            proxy=config.basic.proxy,
-            headless=config.basic.browser_headless,
-            log_callback=log_cb,
-        )
+        # 根据配置选择浏览器引擎
+        browser_engine = (config.basic.browser_engine or "dp").lower()
+        if browser_engine == "dp":
+            # DrissionPage 引擎：仅有头模式，更稳定
+            automation = GeminiAutomation(
+                user_agent=self.user_agent,
+                proxy=config.basic.proxy,
+                headless=False,  # DP 不支持无头模式
+                log_callback=log_cb,
+            )
+        else:
+            # undetected-chromedriver 引擎：支持有头和无头
+            automation = GeminiAutomationUC(
+                user_agent=self.user_agent,
+                proxy=config.basic.proxy,
+                headless=config.basic.browser_headless,
+                log_callback=log_cb,
+            )
 
         try:
             result = automation.login_and_extract(client.email, client)