Upload 18 files

core/account.py

@@ -38,6 +38,12 @@ class AccountConfig:
     config_id: str
     expires_at: Optional[str] = None  # 账户过期时间 (格式: "2025-12-23 10:59:21")
     disabled: bool = False  # 手动禁用状态
+    mail_provider: Optional[str] = None
+    mail_address: Optional[str] = None
+    mail_password: Optional[str] = None
+    mail_client_id: Optional[str] = None
+    mail_refresh_token: Optional[str] = None
+    mail_tenant: Optional[str] = None
 
     def get_remaining_hours(self) -> Optional[float]:
         """计算账户剩余小时数"""
@@ -440,7 +446,13 @@ def load_multi_account_config(
             csesidx=acc["csesidx"],
             config_id=acc["config_id"],
             expires_at=acc.get("expires_at"),
-            disabled=acc.get("disabled", False)  # 读取手动禁用状态，默认为 False
+            disabled=acc.get("disabled", False),  # 读取手动禁用状态，默认为False
+            mail_provider=acc.get("mail_provider"),
+            mail_address=acc.get("mail_address"),
+            mail_password=acc.get("mail_password") or acc.get("email_password"),
+            mail_client_id=acc.get("mail_client_id"),
+            mail_refresh_token=acc.get("mail_refresh_token"),
+            mail_tenant=acc.get("mail_tenant"),
         )
 
         # 检查账户是否已过期（已过期也加载到管理面板）

core/base_task_service.py

@@ -0,0 +1,165 @@
+"""
+基础任务服务类
+提供通用的任务管理、日志记录和账户更新功能
+"""
+import asyncio
+import logging
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Any, Callable, Dict, Generic, List, Optional, TypeVar
+
+from core.account import update_accounts_config
+
+logger = logging.getLogger("gemini.base_task")
+
+
+class TaskStatus(str, Enum):
+    """任务状态枚举"""
+    PENDING = "pending"
+    RUNNING = "running"
+    SUCCESS = "success"
+    FAILED = "failed"
+
+
+@dataclass
+class BaseTask:
+    """基础任务数据类"""
+    id: str
+    status: TaskStatus = TaskStatus.PENDING
+    progress: int = 0
+    success_count: int = 0
+    fail_count: int = 0
+    created_at: float = field(default_factory=time.time)
+    finished_at: Optional[float] = None
+    results: List[Dict[str, Any]] = field(default_factory=list)
+    error: Optional[str] = None
+    logs: List[Dict[str, str]] = field(default_factory=list)
+
+    def to_dict(self) -> dict:
+        """转换为字典"""
+        return {
+            "id": self.id,
+            "status": self.status.value,
+            "progress": self.progress,
+            "success_count": self.success_count,
+            "fail_count": self.fail_count,
+            "created_at": self.created_at,
+            "finished_at": self.finished_at,
+            "results": self.results,
+            "error": self.error,
+            "logs": self.logs,
+        }
+
+
+T = TypeVar('T', bound=BaseTask)
+
+
+class BaseTaskService(Generic[T]):
+    """
+    基础任务服务类
+    提供通用的任务管理、日志记录和账户更新功能
+    """
+
+    def __init__(
+        self,
+        multi_account_mgr,
+        http_client,
+        user_agent: str,
+        account_failure_threshold: int,
+        rate_limit_cooldown_seconds: int,
+        session_cache_ttl_seconds: int,
+        global_stats_provider: Callable[[], dict],
+        set_multi_account_mgr: Optional[Callable[[Any], None]] = None,
+        log_prefix: str = "TASK",
+    ) -> None:
+        """
+        初始化基础任务服务
+
+        Args:
+            multi_account_mgr: 多账户管理器
+            http_client: HTTP客户端
+            user_agent: 用户代理
+            account_failure_threshold: 账户失败阈值
+            rate_limit_cooldown_seconds: 速率限制冷却秒数
+            session_cache_ttl_seconds: 会话缓存TTL秒数
+            global_stats_provider: 全局统计提供者
+            set_multi_account_mgr: 设置多账户管理器的回调
+            log_prefix: 日志前缀
+        """
+        self._executor = ThreadPoolExecutor(max_workers=1)
+        self._tasks: Dict[str, T] = {}
+        self._current_task_id: Optional[str] = None
+        self._lock = asyncio.Lock()
+        self._log_lock = threading.Lock()
+        self._log_prefix = log_prefix
+
+        self.multi_account_mgr = multi_account_mgr
+        self.http_client = http_client
+        self.user_agent = user_agent
+        self.account_failure_threshold = account_failure_threshold
+        self.rate_limit_cooldown_seconds = rate_limit_cooldown_seconds
+        self.session_cache_ttl_seconds = session_cache_ttl_seconds
+        self.global_stats_provider = global_stats_provider
+        self.set_multi_account_mgr = set_multi_account_mgr
+
+    def get_task(self, task_id: str) -> Optional[T]:
+        """获取指定任务"""
+        return self._tasks.get(task_id)
+
+    def get_current_task(self) -> Optional[T]:
+        """获取当前任务"""
+        if not self._current_task_id:
+            return None
+        return self._tasks.get(self._current_task_id)
+
+    def _append_log(self, task: T, level: str, message: str) -> None:
+        """
+        添加日志到任务
+
+        Args:
+            task: 任务对象
+            level: 日志级别 (info, warning, error)
+            message: 日志消息
+        """
+        entry = {
+            "time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
+            "level": level,
+            "message": message,
+        }
+        with self._log_lock:
+            task.logs.append(entry)
+            if len(task.logs) > 200:
+                task.logs = task.logs[-200:]
+
+        log_message = f"[{self._log_prefix}] {message}"
+        if level == "warning":
+            logger.warning(log_message)
+        elif level == "error":
+            logger.error(log_message)
+        else:
+            logger.info(log_message)
+
+    def _apply_accounts_update(self, accounts_data: list) -> None:
+        """
+        应用账户更新
+
+        Args:
+            accounts_data: 账户数据列表
+        """
+        global_stats = self.global_stats_provider() or {}
+        new_mgr = update_accounts_config(
+            accounts_data,
+            self.multi_account_mgr,
+            self.http_client,
+            self.user_agent,
+            self.account_failure_threshold,
+            self.rate_limit_cooldown_seconds,
+            self.session_cache_ttl_seconds,
+            global_stats,
+        )
+        self.multi_account_mgr = new_mgr
+        if self.set_multi_account_mgr:
+            self.set_multi_account_mgr(new_mgr)

core/config.py

@@ -2,13 +2,12 @@
 统一配置管理系统
 
 优先级规则：
-1. 环境变量（最高优先级）
-2. YAML 配置文件
-3. 默认值（最低优先级）
+1. 安全配置：仅环境变量（ADMIN_KEY, SESSION_SECRET_KEY）
+2. 业务配置：YAML 配置文件 > 默认值
 
 配置分类：
-- 安全配置：仅从环境变量读取，不可热更新（ADMIN_KEY, PATH_PREFIX, SESSION_SECRET_KEY）
-- 业务配置：环境变量 > YAML，支持热更新（API_KEY, PROXY, 重试策略等）
+- 安全配置：仅从环境变量读取，不可热更新（ADMIN_KEY, SESSION_SECRET_KEY）
+- 业务配置：仅从 YAML 读取，支持热更新（API_KEY, BASE_URL, PROXY, 重试策略等）
 """
 
 import os
@@ -24,6 +23,21 @@ from core import storage
 # 加载 .env 文件
 load_dotenv()
 
+def _parse_bool(value, default: bool) -> bool:
+    if isinstance(value, bool):
+        return value
+    if value is None:
+        return default
+    if isinstance(value, (int, float)):
+        return value != 0
+    if isinstance(value, str):
+        lowered = value.strip().lower()
+        if lowered in ("1", "true", "yes", "y", "on"):
+            return True
+        if lowered in ("0", "false", "no", "n", "off"):
+            return False
+    return default
+
 
 # ==================== 配置模型定义 ====================
 
@@ -32,6 +46,13 @@ class BasicConfig(BaseModel):
     api_key: str = Field(default="", description="API访问密钥（留空则公开访问）")
     base_url: str = Field(default="", description="服务器URL（留空则自动检测）")
     proxy: str = Field(default="", description="代理地址")
+    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="自动化浏览器无头模式")
+    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="默认注册域名（推荐）")
 
 
 class ImageGenerationConfig(BaseModel):
@@ -107,7 +128,7 @@ class ConfigManager:
 
         优先级规则：
         1. 安全配置（ADMIN_KEY, SESSION_SECRET_KEY）：仅从环境变量读取
-        2. 其他配置：YAML > 环境变量 > 默认值
+        2. 其他配置：YAML > 默认值
         """
         # 1. 加载 YAML 配置
         yaml_data = self._load_yaml()
@@ -118,12 +139,24 @@ class ConfigManager:
             session_secret_key=os.getenv("SESSION_SECRET_KEY", self._generate_secret())
         )
 
-        # 3. 加载基础配置（YAML > 环境变量 > 默认值）
+        # 3. 加载基础配置（YAML > 默认值）
         basic_data = yaml_data.get("basic", {})
+        refresh_window_raw = basic_data.get("refresh_window_hours", 1)
+        register_default_raw = basic_data.get("register_default_count", 1)
+        register_domain_raw = basic_data.get("register_domain", "")
+        duckmail_api_key_raw = basic_data.get("duckmail_api_key", "")
+
         basic_config = BasicConfig(
-            api_key=basic_data.get("api_key") or os.getenv("API_KEY", ""),
-            base_url=basic_data.get("base_url") or os.getenv("BASE_URL", ""),
-            proxy=basic_data.get("proxy") or os.getenv("PROXY", "")
+            api_key=basic_data.get("api_key") or "",
+            base_url=basic_data.get("base_url") or "",
+            proxy=basic_data.get("proxy") or "",
+            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),
+            refresh_window_hours=int(refresh_window_raw),
+            register_default_count=int(register_default_raw),
+            register_domain=str(register_domain_raw or "").strip(),
         )
 
         # 4. 加载其他配置（从 YAML）

core/duckmail_client.py

@@ -0,0 +1,272 @@
+import os
+import random
+import string
+import time
+from typing import Optional
+
+import requests
+
+from core.mail_utils import extract_verification_code
+
+
+class DuckMailClient:
+    """DuckMail客户端"""
+
+    def __init__(
+        self,
+        base_url: str = "https://api.duckmail.sbs",
+        proxy: str = "",
+        verify_ssl: bool = True,
+        api_key: str = "",
+        log_callback=None,
+    ) -> None:
+        self.base_url = base_url.rstrip("/")
+        self.verify_ssl = verify_ssl
+        self.proxies = {"http": proxy, "https": proxy} if proxy else None
+        self.api_key = api_key.strip()
+        self.log_callback = log_callback
+
+        self.email: Optional[str] = None
+        self.password: Optional[str] = None
+        self.account_id: Optional[str] = None
+        self.token: Optional[str] = None
+
+    def set_credentials(self, email: str, password: str) -> None:
+        self.email = email
+        self.password = password
+
+    def _request(self, method: str, url: str, **kwargs) -> requests.Response:
+        """发送请求并打印详细日志"""
+        headers = kwargs.pop("headers", None) or {}
+        if self.api_key and "Authorization" not in headers:
+            headers["Authorization"] = f"Bearer {self.api_key}"
+        kwargs["headers"] = headers
+        self._log("info", f"[HTTP] {method} {url}")
+        if "json" in kwargs:
+            self._log("info", f"[HTTP] Request body: {kwargs['json']}")
+
+        try:
+            res = requests.request(
+                method,
+                url,
+                proxies=self.proxies,
+                verify=self.verify_ssl,
+                timeout=kwargs.pop("timeout", 15),
+                **kwargs,
+            )
+            self._log("info", f"[HTTP] Response: {res.status_code}")
+            log_body = os.getenv("DUCKMAIL_LOG_BODY", "").strip().lower() in ("1", "true", "yes", "y", "on")
+            if res.content and (log_body or res.status_code >= 400):
+                try:
+                    self._log("info", f"[HTTP] Response body: {res.text[:500]}")
+                except Exception:
+                    pass
+            return res
+        except Exception as e:
+            self._log("error", f"[HTTP] Request failed: {e}")
+            raise
+
+    def register_account(self, domain: Optional[str] = None) -> bool:
+        """注册新邮箱账号"""
+        # 获取域名
+        if not domain:
+            domain = self._get_domain()
+        self._log("info", f"DuckMail domain: {domain}")
+
+        # 生成随机邮箱和密码
+        rand = "".join(random.choices(string.ascii_lowercase + string.digits, k=10))
+        timestamp = str(int(time.time()))[-4:]
+        self.email = f"t{timestamp}{rand}@{domain}"
+        self.password = f"Pwd{rand}{timestamp}"
+        self._log("info", f"DuckMail register email: {self.email}")
+
+        try:
+            res = self._request(
+                "POST",
+                f"{self.base_url}/accounts",
+                json={"address": self.email, "password": self.password},
+            )
+            if res.status_code in (200, 201):
+                data = res.json() if res.content else {}
+                self.account_id = data.get("id")
+                self._log("info", "DuckMail register success")
+                return True
+        except Exception as e:
+            self._log("error", f"DuckMail register failed: {e}")
+            return False
+
+        self._log("error", "DuckMail register failed")
+        return False
+
+    def login(self) -> bool:
+        """登录获取token"""
+        if not self.email or not self.password:
+            return False
+
+        try:
+            res = self._request(
+                "POST",
+                f"{self.base_url}/token",
+                json={"address": self.email, "password": self.password},
+            )
+            if res.status_code == 200:
+                data = res.json() if res.content else {}
+                token = data.get("token")
+                if token:
+                    self.token = token
+                    self._log("info", f"DuckMail login success, token: {token[:20]}...")
+                    return True
+        except Exception as e:
+            self._log("error", f"DuckMail login failed: {e}")
+            return False
+
+        self._log("error", "DuckMail login failed")
+        return False
+
+    def fetch_verification_code(self, since_time=None) -> Optional[str]:
+        """获取验证码"""
+        if not self.token:
+            if not self.login():
+                return None
+
+        try:
+            # 获取邮件列表
+            res = self._request(
+                "GET",
+                f"{self.base_url}/messages",
+                headers={"Authorization": f"Bearer {self.token}"},
+            )
+
+            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}",
+                headers={"Authorization": f"Bearer {self.token}"},
+            )
+
+            if detail.status_code != 200:
+                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}")
+            return code
+
+        except Exception as e:
+            self._log("error", f"DuckMail fetch code failed: {e}")
+            return None
+
+    def poll_for_code(
+        self,
+        timeout: int = 120,
+        interval: int = 4,
+        since_time=None,
+    ) -> Optional[str]:
+        """轮询获取验证码"""
+        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")
+        return None
+
+    def _get_domain(self) -> str:
+        """获取可用域名"""
+        try:
+            res = self._request("GET", f"{self.base_url}/domains")
+            if res.status_code == 200:
+                data = res.json() if res.content else {}
+                members = data.get("hydra:member", [])
+                if members:
+                    return members[0].get("domain") or "duck.com"
+        except Exception:
+            pass
+        return "duck.com"
+
+    def _log(self, level: str, message: str) -> None:
+        if self.log_callback:
+            try:
+                self.log_callback(level, message)
+            except Exception:
+                pass
+
+    @staticmethod
+    def _extract_code(text: str) -> Optional[str]:
+        return extract_verification_code(text)

core/gemini_automation.py

@@ -0,0 +1,400 @@
+"""
+Gemini自动化登录模块（用于新账号注册）
+"""
+import random
+import string
+import time
+from datetime import datetime, timedelta
+from typing import Optional
+from urllib.parse import quote
+
+from DrissionPage import ChromiumPage, ChromiumOptions
+
+
+# 常量
+AUTH_HOME_URL = "https://auth.business.gemini.google/"
+DEFAULT_XSRF_TOKEN = "KdLRzKwwBTD5wo8nUollAbY6cW0"
+
+
+class GeminiAutomation:
+    """Gemini自动化登录"""
+
+    def __init__(
+        self,
+        user_agent: str = "",
+        proxy: str = "",
+        headless: bool = True,
+        timeout: int = 60,
+        log_callback=None,
+    ) -> None:
+        self.user_agent = user_agent or self._get_ua()
+        self.proxy = proxy
+        self.headless = headless
+        self.timeout = timeout
+        self.log_callback = log_callback
+
+    def login_and_extract(self, email: str, mail_client) -> dict:
+        """执行登录并提取配置"""
+        page = None
+        try:
+            page = self._create_page()
+            return self._run_flow(page, email, mail_client)
+        except Exception as exc:
+            self._log("error", f"automation error: {exc}")
+            return {"success": False, "error": str(exc)}
+        finally:
+            if page:
+                try:
+                    page.quit()
+                except Exception:
+                    pass
+
+    def _create_page(self) -> ChromiumPage:
+        """创建浏览器页面"""
+        options = ChromiumOptions()
+        options.set_argument("--no-sandbox")
+        options.set_argument("--disable-setuid-sandbox")
+        options.set_argument("--disable-blink-features=AutomationControlled")
+        options.set_argument("--window-size=1280,800")
+        options.set_user_agent(self.user_agent)
+
+        if self.proxy:
+            options.set_argument(f"--proxy-server={self.proxy}")
+
+        if self.headless:
+            # 使用新版无头模式，更接近真实浏览器
+            options.set_argument("--headless=new")
+            options.set_argument("--disable-gpu")
+            options.set_argument("--disable-dev-shm-usage")
+            options.set_argument("--no-first-run")
+            options.set_argument("--disable-extensions")
+            # 反检测参数
+            options.set_argument("--disable-infobars")
+            options.set_argument("--lang=zh-CN,zh")
+            options.set_argument("--enable-features=NetworkService,NetworkServiceInProcess")
+
+        options.auto_port()
+        page = ChromiumPage(options)
+        page.set.timeouts(self.timeout)
+
+        # 反检测：注入脚本隐藏自动化特征
+        if self.headless:
+            try:
+                page.run_cdp("Page.addScriptToEvaluateOnNewDocument", source="""
+                    Object.defineProperty(navigator, 'webdriver', {get: () => undefined});
+                    Object.defineProperty(navigator, 'plugins', {get: () => [1, 2, 3, 4, 5]});
+                    Object.defineProperty(navigator, 'languages', {get: () => ['zh-CN', 'zh', 'en']});
+                    window.chrome = {runtime: {}};
+                """)
+            except Exception:
+                pass
+
+        return page
+
+    def _run_flow(self, page, email: str, mail_client) -> dict:
+        """执行登录流程"""
+
+        # Step 1: 导航到首页并设置 Cookie
+        self._log("info", f"navigating to login page for {email}")
+
+        page.get(AUTH_HOME_URL, timeout=self.timeout)
+        time.sleep(2)
+
+        # 设置两个关键 Cookie
+        try:
+            page.set.cookies({
+                "name": "__Host-AP_SignInXsrf",
+                "value": DEFAULT_XSRF_TOKEN,
+                "url": AUTH_HOME_URL,
+                "path": "/",
+                "secure": True,
+            })
+            # 添加 reCAPTCHA Cookie
+            page.set.cookies({
+                "name": "_GRECAPTCHA",
+                "value": "09ABCL...",
+                "url": "https://google.com",
+                "path": "/",
+                "secure": True,
+            })
+        except Exception as e:
+            self._log("warning", f"failed to set cookies: {e}")
+
+        login_hint = quote(email, safe="")
+        login_url = f"https://auth.business.gemini.google/login/email?continueUrl=https%3A%2F%2Fbusiness.gemini.google%2F&loginHint={login_hint}&xsrfToken={DEFAULT_XSRF_TOKEN}"
+        page.get(login_url, timeout=self.timeout)
+        time.sleep(5)
+
+        # Step 2: 检查当前页面状态
+        current_url = page.url
+        has_business_params = "business.gemini.google" in current_url and "csesidx=" in current_url and "/cid/" in current_url
+
+        if has_business_params:
+            return self._extract_config(page, email)
+
+        # Step 3: 尝试触发发送验证码
+        if not self._click_send_code_button(page):
+            self._log("error", "send code button not found")
+            self._save_screenshot(page, "send_code_button_missing")
+            return {"success": False, "error": "send code button not found"}
+
+        # Step 4: 等待验证码输入框出现
+        code_input = self._wait_for_code_input(page)
+        if not code_input:
+            self._log("error", "code input not found")
+            self._save_screenshot(page, "code_input_missing")
+            return {"success": False, "error": "code input not found"}
+
+        time.sleep(3)
+
+        # Step 5: 轮询邮件获取验证码
+        self._log("info", "polling for verification code")
+        code = mail_client.poll_for_code(timeout=40, interval=4)
+
+        if not code:
+            self._log("error", "verification code timeout")
+            self._save_screenshot(page, "code_timeout")
+            return {"success": False, "error": "verification code timeout"}
+
+        self._log("info", f"code received: {code}")
+
+        # Step 6: 输入验证码并提交
+        code_input = page.ele("css:input[jsname='ovqh0b']", timeout=3) or \
+                     page.ele("css:input[type='tel']", timeout=2)
+
+        if not code_input:
+            self._log("error", "code input expired")
+            return {"success": False, "error": "code input expired"}
+
+        code_input.input(code, clear=True)
+        time.sleep(0.5)
+
+        verify_btn = page.ele("css:button[jsname='XooR8e']", timeout=3)
+        if verify_btn:
+            verify_btn.click()
+        else:
+            verify_btn = self._find_verify_button(page)
+            if verify_btn:
+                verify_btn.click()
+            else:
+                code_input.input("\n")
+
+        time.sleep(5)
+
+        # Step 7: 处理协议页面（如果有）
+        self._handle_agreement_page(page)
+
+        # Step 8: 导航到业务页面
+        page.get("https://business.gemini.google/", timeout=self.timeout)
+        time.sleep(3)
+
+        # Step 9: 检查是否需要设置用户名
+        if "cid" not in page.url:
+            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"}
+
+    def _click_send_code_button(self, page) -> bool:
+        """点击发送验证码按钮（如果需要）"""
+        time.sleep(2)
+
+        # 方法1: 直接通过ID查找
+        direct_btn = page.ele("#sign-in-with-email", timeout=5)
+        if direct_btn:
+            try:
+                direct_btn.click()
+                return True
+            except Exception:
+                pass
+
+        # 方法2: 通过关键词查找
+        keywords = ["通过电子邮件发送验证码", "通过电子邮件发送", "email", "Email", "Send code", "Send verification", "Verification code"]
+        try:
+            buttons = page.eles("tag:button")
+            for btn in buttons:
+                text = (btn.text or "").strip()
+                if text and any(kw in text for kw in keywords):
+                    try:
+                        btn.click()
+                        return True
+                    except Exception:
+                        pass
+        except Exception:
+            pass
+
+        # 检查是否已经在验证码输入页面
+        code_input = page.ele("css:input[jsname='ovqh0b']", timeout=2) or page.ele("css:input[name='pinInput']", timeout=1)
+        if code_input:
+            return True
+
+        return False
+
+    def _wait_for_code_input(self, page, timeout: int = 30):
+        """等待验证码输入框出现"""
+        selectors = [
+            "css:input[jsname='ovqh0b']",
+            "css:input[type='tel']",
+            "css:input[name='pinInput']",
+            "css:input[autocomplete='one-time-code']",
+        ]
+        for _ in range(timeout // 2):
+            for selector in selectors:
+                try:
+                    el = page.ele(selector, timeout=1)
+                    if el:
+                        return el
+                except Exception:
+                    continue
+            time.sleep(2)
+        return None
+
+    def _find_verify_button(self, page):
+        """查找验证按钮（排除重新发送按钮）"""
+        try:
+            buttons = page.eles("tag:button")
+            for btn in buttons:
+                text = (btn.text or "").strip().lower()
+                if text and "重新" not in text and "发送" not in text and "resend" not in text and "send" not in text:
+                    return btn
+        except Exception:
+            pass
+        return None
+
+    def _handle_agreement_page(self, page) -> None:
+        """处理协��页面"""
+        if "/admin/create" in page.url:
+            agree_btn = page.ele("css:button.agree-button", timeout=5)
+            if agree_btn:
+                agree_btn.click()
+                time.sleep(2)
+
+    def _wait_for_cid(self, page, timeout: int = 10) -> bool:
+        """等待URL包含cid"""
+        for _ in range(timeout):
+            if "cid" in page.url:
+                return True
+            time.sleep(1)
+        return False
+
+    def _handle_username_setup(self, page) -> bool:
+        """处理用户名设置页面"""
+        current_url = page.url
+
+        if "auth.business.gemini.google/login" in current_url:
+            return False
+
+        selectors = [
+            "css:input[type='text']",
+            "css:input[name='displayName']",
+            "css:input[aria-label*='用户名' i]",
+            "css:input[aria-label*='display name' i]",
+        ]
+
+        username_input = None
+        for selector in selectors:
+            try:
+                username_input = page.ele(selector, timeout=2)
+                if username_input:
+                    break
+            except Exception:
+                continue
+
+        if not username_input:
+            return False
+
+        suffix = "".join(random.choices(string.ascii_letters + string.digits, k=3))
+        username = f"Test{suffix}"
+
+        try:
+            username_input.click()
+            time.sleep(0.2)
+            username_input.clear()
+            username_input.input(username)
+            time.sleep(0.3)
+
+            buttons = page.eles("tag:button")
+            submit_btn = None
+            for btn in buttons:
+                text = (btn.text or "").strip().lower()
+                if any(kw in text for kw in ["确认", "提交", "继续", "submit", "continue", "confirm", "save", "保存", "下一步", "next"]):
+                    submit_btn = btn
+                    break
+
+            if submit_btn:
+                submit_btn.click()
+            else:
+                username_input.input("\n")
+
+            time.sleep(5)
+            return True
+        except Exception:
+            return False
+
+    def _extract_config(self, page, email: str) -> dict:
+        """提取配置"""
+        try:
+            if "cid/" not in page.url:
+                page.get("https://business.gemini.google/", timeout=self.timeout)
+                time.sleep(3)
+
+            url = page.url
+            if "cid/" not in url:
+                return {"success": False, "error": "cid not found"}
+
+            config_id = url.split("cid/")[1].split("?")[0].split("/")[0]
+            csesidx = url.split("csesidx=")[1].split("&")[0] if "csesidx=" in url else ""
+
+            cookies = page.cookies()
+            ses = next((c["value"] for c in cookies if c["name"] == "__Secure-C_SES"), None)
+            host = next((c["value"] for c in cookies if c["name"] == "__Host-C_OSES"), None)
+
+            ses_obj = next((c for c in cookies if c["name"] == "__Secure-C_SES"), None)
+            if ses_obj and "expiry" in ses_obj:
+                expires_at = datetime.fromtimestamp(ses_obj["expiry"] - 43200).strftime("%Y-%m-%d %H:%M:%S")
+            else:
+                expires_at = (datetime.now() + timedelta(hours=12)).strftime("%Y-%m-%d %H:%M:%S")
+
+            config = {
+                "id": email,
+                "csesidx": csesidx,
+                "config_id": config_id,
+                "secure_c_ses": ses,
+                "host_c_oses": host,
+                "expires_at": expires_at,
+            }
+            return {"success": True, "config": config}
+        except Exception as e:
+            return {"success": False, "error": str(e)}
+
+    def _save_screenshot(self, page, name: str) -> None:
+        """保存截图"""
+        try:
+            import os
+            screenshot_dir = os.path.join("data", "automation")
+            os.makedirs(screenshot_dir, exist_ok=True)
+            path = os.path.join(screenshot_dir, f"{name}_{int(time.time())}.png")
+            page.get_screenshot(path=path)
+        except Exception:
+            pass
+
+    def _log(self, level: str, message: str) -> None:
+        """记录日志"""
+        if self.log_callback:
+            try:
+                self.log_callback(level, message)
+            except Exception:
+                pass
+
+    @staticmethod
+    def _get_ua() -> str:
+        """生成随机User-Agent"""
+        v = random.choice(["120.0.0.0", "121.0.0.0", "122.0.0.0"])
+        return f"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{v} Safari/537.36"

core/gemini_automation_uc.py

@@ -0,0 +1,390 @@
+"""
+Gemini自动化登录模块（使用 undetected-chromedriver）
+更强的反检测能力，支持无头模式
+"""
+import random
+import string
+import time
+from datetime import datetime, timedelta
+from typing import Optional
+from urllib.parse import quote
+
+import undetected_chromedriver as uc
+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, NoSuchElementException
+
+
+# 常量
+AUTH_HOME_URL = "https://auth.business.gemini.google/"
+LOGIN_URL = "https://auth.business.gemini.google/login?continueUrl=https:%2F%2Fbusiness.gemini.google%2F&wiffid=CAoSJDIwNTlhYzBjLTVlMmMtNGUxZS1hY2JkLThmOGY2ZDE0ODM1Mg"
+DEFAULT_XSRF_TOKEN = "KdLRzKwwBTD5wo8nUollAbY6cW0"
+
+
+class GeminiAutomationUC:
+    """Gemini自动化登录（使用 undetected-chromedriver）"""
+
+    def __init__(
+        self,
+        user_agent: str = "",
+        proxy: str = "",
+        headless: bool = True,
+        timeout: int = 60,
+        log_callback=None,
+    ) -> None:
+        self.user_agent = user_agent or self._get_ua()
+        self.proxy = proxy
+        self.headless = headless
+        self.timeout = timeout
+        self.log_callback = log_callback
+        self.driver = None
+
+    def login_and_extract(self, email: str, mail_client) -> dict:
+        """执行登录并提取配置"""
+        try:
+            self._create_driver()
+            return self._run_flow(email, mail_client)
+        except Exception as exc:
+            self._log("error", f"automation error: {exc}")
+            return {"success": False, "error": str(exc)}
+        finally:
+            self._cleanup()
+
+    def _create_driver(self):
+        """创建浏览器驱动"""
+        options = uc.ChromeOptions()
+
+        # 基础参数
+        options.add_argument("--no-sandbox")
+        options.add_argument("--disable-setuid-sandbox")
+        options.add_argument("--window-size=1280,800")
+
+        # 代理设置
+        if self.proxy:
+            options.add_argument(f"--proxy-server={self.proxy}")
+
+        # 无头模式
+        if self.headless:
+            options.add_argument("--headless=new")
+            options.add_argument("--disable-gpu")
+            options.add_argument("--disable-dev-shm-usage")
+
+        # User-Agent
+        if self.user_agent:
+            options.add_argument(f"--user-agent={self.user_agent}")
+
+        # 创建驱动（undetected-chromedriver 会自动处理反检测）
+        self.driver = uc.Chrome(
+            options=options,
+            version_main=None,  # 自动检测 Chrome 版本
+            use_subprocess=True,
+        )
+
+        # 设置超时
+        self.driver.set_page_load_timeout(self.timeout)
+        self.driver.implicitly_wait(10)
+
+    def _run_flow(self, email: str, mail_client) -> dict:
+        """执行登录流程"""
+
+        self._log("info", f"navigating to login page for {email}")
+
+        # 访问登录页面
+        self.driver.get(LOGIN_URL)
+        time.sleep(3)
+
+        # 检查当前页面状态
+        current_url = self.driver.current_url
+        has_business_params = "business.gemini.google" in current_url and "csesidx=" in current_url and "/cid/" in current_url
+
+        if has_business_params:
+            return self._extract_config(email)
+
+        # 输入邮箱地址
+        try:
+            self._log("info", "entering email address")
+            email_input = WebDriverWait(self.driver, 30).until(
+                EC.element_to_be_clickable((By.XPATH, "/html/body/c-wiz/div/div/div[1]/div/div/div/form/div[1]/div[1]/div/span[2]/input"))
+            )
+            email_input.click()
+            email_input.clear()
+            for char in email:
+                email_input.send_keys(char)
+                time.sleep(0.02)
+            time.sleep(0.5)
+        except Exception as e:
+            self._log("error", f"failed to enter email: {e}")
+            self._save_screenshot("email_input_failed")
+            return {"success": False, "error": f"failed to enter email: {e}"}
+
+        # 点击继续按钮
+        try:
+            continue_btn = WebDriverWait(self.driver, 10).until(
+                EC.element_to_be_clickable((By.XPATH, "/html/body/c-wiz/div/div/div[1]/div/div/div/form/div[2]/div/button"))
+            )
+            self.driver.execute_script("arguments[0].click();", continue_btn)
+            time.sleep(2)
+        except Exception as e:
+            self._log("error", f"failed to click continue: {e}")
+            self._save_screenshot("continue_button_failed")
+            return {"success": False, "error": f"failed to click continue: {e}"}
+
+        # 等待验证码输入框出现
+        code_input = self._wait_for_code_input()
+        if not code_input:
+            self._log("error", "code input not found")
+            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)
+
+        if not code:
+            self._log("error", "verification code timeout")
+            self._save_screenshot("code_timeout")
+            return {"success": False, "error": "verification code timeout"}
+
+        self._log("info", f"code received: {code}")
+
+        # 输入验证码
+        time.sleep(1)
+        try:
+            code_input = WebDriverWait(self.driver, 10).until(
+                EC.presence_of_element_located((By.CSS_SELECTOR, "input[name='pinInput']"))
+            )
+            code_input.click()
+            time.sleep(0.1)
+            for char in code:
+                code_input.send_keys(char)
+                time.sleep(0.05)
+        except Exception:
+            try:
+                span = self.driver.find_element(By.CSS_SELECTOR, "span[data-index='0']")
+                span.click()
+                time.sleep(0.2)
+                self.driver.switch_to.active_element.send_keys(code)
+            except Exception as e:
+                self._log("error", f"failed to input code: {e}")
+                self._save_screenshot("code_input_failed")
+                return {"success": False, "error": f"failed to input code: {e}"}
+
+        # 点击验证按钮
+        time.sleep(0.5)
+        try:
+            verify_btn = self.driver.find_element(By.XPATH, "/html/body/c-wiz/div/div/div[1]/div/div/div/form/div[2]/div/div[1]/span/div[1]/button")
+            self.driver.execute_script("arguments[0].click();", verify_btn)
+        except Exception:
+            try:
+                buttons = self.driver.find_elements(By.TAG_NAME, "button")
+                for btn in buttons:
+                    if "验证" in btn.text:
+                        self.driver.execute_script("arguments[0].click();", btn)
+                        break
+            except Exception as e:
+                self._log("warning", f"failed to click verify button: {e}")
+
+        time.sleep(5)
+
+        # 处理协议页面
+        self._handle_agreement_page()
+
+        # 导航到业务页面
+        self.driver.get("https://business.gemini.google/")
+        time.sleep(3)
+
+        # 处理用户名设置
+        if "cid" not in self.driver.current_url:
+            if self._handle_username_setup():
+                time.sleep(3)
+
+        # 提取配置
+        if "cid" in self.driver.current_url or self._wait_for_cid():
+            self._log("info", "login success")
+            return self._extract_config(email)
+
+        self._log("error", "login failed")
+        self._save_screenshot("login_failed")
+        return {"success": False, "error": "login failed"}
+
+    def _wait_for_code_input(self, timeout: int = 30):
+        """等待验证码输入框出现"""
+        try:
+            element = WebDriverWait(self.driver, timeout).until(
+                EC.presence_of_element_located((By.CSS_SELECTOR, "input[name='pinInput']"))
+            )
+            return element
+        except TimeoutException:
+            return None
+
+    def _find_code_input(self):
+        """查找验证码输入框"""
+        try:
+            return self.driver.find_element(By.CSS_SELECTOR, "input[name='pinInput']")
+        except NoSuchElementException:
+            return None
+
+    def _find_verify_button(self):
+        """查找验证按钮"""
+        try:
+            return self.driver.find_element(By.XPATH, "/html/body/c-wiz/div/div/div[1]/div/div/div/form/div[2]/div/div[1]/span/div[1]/button")
+        except NoSuchElementException:
+            pass
+
+        try:
+            buttons = self.driver.find_elements(By.TAG_NAME, "button")
+            for btn in buttons:
+                text = btn.text.strip()
+                if text and "验证" in text:
+                    return btn
+        except Exception:
+            pass
+
+        return None
+
+    def _handle_agreement_page(self) -> None:
+        """处理协议页面"""
+        if "/admin/create" in self.driver.current_url:
+            try:
+                agree_btn = WebDriverWait(self.driver, 5).until(
+                    EC.element_to_be_clickable((By.CSS_SELECTOR, "button.agree-button"))
+                )
+                agree_btn.click()
+                time.sleep(2)
+            except TimeoutException:
+                pass
+
+    def _wait_for_cid(self, timeout: int = 10) -> bool:
+        """等待URL包含cid"""
+        for _ in range(timeout):
+            if "cid" in self.driver.current_url:
+                return True
+            time.sleep(1)
+        return False
+
+    def _handle_username_setup(self) -> bool:
+        """处理用户名设置页面"""
+        current_url = self.driver.current_url
+
+        if "auth.business.gemini.google/login" in current_url:
+            return False
+
+        selectors = [
+            "input[formcontrolname='fullName']",
+            "input[placeholder='全名']",
+            "input[placeholder='Full name']",
+            "input#mat-input-0",
+            "input[type='text']",
+            "input[name='displayName']",
+        ]
+
+        username_input = None
+        for _ in range(30):
+            for selector in selectors:
+                try:
+                    username_input = self.driver.find_element(By.CSS_SELECTOR, selector)
+                    if username_input.is_displayed():
+                        break
+                except Exception:
+                    continue
+            if username_input and username_input.is_displayed():
+                break
+            time.sleep(1)
+
+        if not username_input or not username_input.is_displayed():
+            return False
+
+        suffix = "".join(random.choices(string.ascii_letters + string.digits, k=3))
+        username = f"Test{suffix}"
+
+        try:
+            username_input.click()
+            time.sleep(0.2)
+            username_input.clear()
+            for char in username:
+                username_input.send_keys(char)
+                time.sleep(0.02)
+            time.sleep(0.3)
+
+            from selenium.webdriver.common.keys import Keys
+            username_input.send_keys(Keys.ENTER)
+            time.sleep(1)
+
+            return True
+        except Exception:
+            return False
+
+    def _extract_config(self, email: str) -> dict:
+        """提取配置"""
+        try:
+            if "cid/" not in self.driver.current_url:
+                self.driver.get("https://business.gemini.google/")
+                time.sleep(3)
+
+            url = self.driver.current_url
+            if "cid/" not in url:
+                return {"success": False, "error": "cid not found"}
+
+            # 提取参数
+            config_id = url.split("cid/")[1].split("?")[0].split("/")[0]
+            csesidx = url.split("csesidx=")[1].split("&")[0] if "csesidx=" in url else ""
+
+            # 提取 Cookie
+            cookies = self.driver.get_cookies()
+            ses = next((c["value"] for c in cookies if c["name"] == "__Secure-C_SES"), None)
+            host = next((c["value"] for c in cookies if c["name"] == "__Host-C_OSES"), None)
+
+            # 计算过期时间
+            ses_obj = next((c for c in cookies if c["name"] == "__Secure-C_SES"), None)
+            if ses_obj and "expiry" in ses_obj:
+                expires_at = datetime.fromtimestamp(ses_obj["expiry"] - 43200).strftime("%Y-%m-%d %H:%M:%S")
+            else:
+                expires_at = (datetime.now() + timedelta(hours=12)).strftime("%Y-%m-%d %H:%M:%S")
+
+            config = {
+                "id": email,
+                "csesidx": csesidx,
+                "config_id": config_id,
+                "secure_c_ses": ses,
+                "host_c_oses": host,
+                "expires_at": expires_at,
+            }
+            return {"success": True, "config": config}
+        except Exception as e:
+            return {"success": False, "error": str(e)}
+
+    def _save_screenshot(self, name: str) -> None:
+        """保存截图"""
+        try:
+            import os
+            screenshot_dir = os.path.join("data", "automation")
+            os.makedirs(screenshot_dir, exist_ok=True)
+            path = os.path.join(screenshot_dir, f"{name}_{int(time.time())}.png")
+            self.driver.save_screenshot(path)
+        except Exception:
+            pass
+
+    def _cleanup(self) -> None:
+        """清理资源"""
+        if self.driver:
+            try:
+                self.driver.quit()
+            except Exception:
+                pass
+
+    def _log(self, level: str, message: str) -> None:
+        """记录日志"""
+        if self.log_callback:
+            try:
+                self.log_callback(level, message)
+            except Exception:
+                pass
+
+    @staticmethod
+    def _get_ua() -> str:
+        """生成随机User-Agent"""
+        v = random.choice(["120.0.0.0", "121.0.0.0", "122.0.0.0"])
+        return f"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{v} Safari/537.36"

core/login_service.py

@@ -0,0 +1,261 @@
+import asyncio
+import logging
+import os
+import time
+import uuid
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta, timezone
+from typing import Any, Callable, Dict, List, Optional
+
+from core.account import load_accounts_from_source
+from core.base_task_service import BaseTask, BaseTaskService, TaskStatus
+from core.config import config
+from core.duckmail_client import DuckMailClient
+from core.gemini_automation import GeminiAutomation
+from core.gemini_automation_uc import GeminiAutomationUC
+from core.microsoft_mail_client import MicrosoftMailClient
+
+logger = logging.getLogger("gemini.login")
+
+
+@dataclass
+class LoginTask(BaseTask):
+    """登录任务数据类"""
+    account_ids: List[str] = field(default_factory=list)
+
+    def to_dict(self) -> dict:
+        """转换为字典"""
+        base_dict = super().to_dict()
+        base_dict["account_ids"] = self.account_ids
+        return base_dict
+
+
+class LoginService(BaseTaskService[LoginTask]):
+    """登录服务类"""
+
+    def __init__(
+        self,
+        multi_account_mgr,
+        http_client,
+        user_agent: str,
+        account_failure_threshold: int,
+        rate_limit_cooldown_seconds: int,
+        session_cache_ttl_seconds: int,
+        global_stats_provider: Callable[[], dict],
+        set_multi_account_mgr: Optional[Callable[[Any], None]] = None,
+    ) -> None:
+        super().__init__(
+            multi_account_mgr,
+            http_client,
+            user_agent,
+            account_failure_threshold,
+            rate_limit_cooldown_seconds,
+            session_cache_ttl_seconds,
+            global_stats_provider,
+            set_multi_account_mgr,
+            log_prefix="REFRESH",
+        )
+        self._is_polling = False
+
+    async def start_login(self, account_ids: List[str]) -> LoginTask:
+        """启动登录任务"""
+        async with self._lock:
+            if self._current_task_id:
+                current = self._tasks.get(self._current_task_id)
+                if current and current.status == TaskStatus.RUNNING:
+                    raise ValueError("login task already running")
+
+            task = LoginTask(id=str(uuid.uuid4()), account_ids=account_ids)
+            self._tasks[task.id] = task
+            self._current_task_id = task.id
+            self._append_log(task, "info", f"login task created ({len(account_ids)} accounts)")
+            asyncio.create_task(self._run_login_async(task))
+            return task
+
+    async def _run_login_async(self, task: LoginTask) -> None:
+        """异步执行登录任务"""
+        task.status = TaskStatus.RUNNING
+        loop = asyncio.get_running_loop()
+        self._append_log(task, "info", "login task started")
+
+        for account_id in task.account_ids:
+            try:
+                result = await loop.run_in_executor(self._executor, self._refresh_one, account_id, task)
+            except Exception as exc:
+                result = {"success": False, "email": account_id, "error": str(exc)}
+            task.progress += 1
+            task.results.append(result)
+
+            if result.get("success"):
+                task.success_count += 1
+                self._append_log(task, "info", f"refresh success: {account_id}")
+            else:
+                task.fail_count += 1
+                self._append_log(task, "error", f"refresh failed: {account_id} - {result.get('error')}")
+
+        task.status = TaskStatus.SUCCESS if task.fail_count == 0 else TaskStatus.FAILED
+        task.finished_at = time.time()
+        self._current_task_id = None
+        self._append_log(task, "info", f"login task finished ({task.success_count}/{len(task.account_ids)})")
+
+    def _refresh_one(self, account_id: str, task: LoginTask) -> dict:
+        """刷新单个账户"""
+        accounts = load_accounts_from_source()
+        account = next((acc for acc in accounts if acc.get("id") == account_id), None)
+        if not account:
+            return {"success": False, "email": account_id, "error": "account not found"}
+
+        if account.get("disabled"):
+            return {"success": False, "email": account_id, "error": "account disabled"}
+
+        # 获取邮件提供商
+        mail_provider = (account.get("mail_provider") or "").lower()
+        if not mail_provider:
+            if account.get("mail_client_id") or account.get("mail_refresh_token"):
+                mail_provider = "microsoft"
+            else:
+                mail_provider = "duckmail"
+
+        # 获取邮件配置
+        mail_password = account.get("mail_password") or account.get("email_password")
+        mail_client_id = account.get("mail_client_id")
+        mail_refresh_token = account.get("mail_refresh_token")
+        mail_tenant = account.get("mail_tenant") or "consumers"
+
+        log_cb = lambda level, message: self._append_log(task, level, f"[{account_id}] {message}")
+
+        # 创建邮件客户端
+        if mail_provider == "microsoft":
+            if not mail_client_id or not mail_refresh_token:
+                return {"success": False, "email": account_id, "error": "microsoft oauth missing"}
+            mail_address = account.get("mail_address") or account_id
+            client = MicrosoftMailClient(
+                client_id=mail_client_id,
+                refresh_token=mail_refresh_token,
+                tenant=mail_tenant,
+                proxy=config.basic.proxy,
+                log_callback=log_cb,
+            )
+            client.set_credentials(mail_address)
+        elif mail_provider == "duckmail":
+            if not mail_password:
+                return {"success": False, "email": account_id, "error": "mail password missing"}
+            # DuckMail: account_id 就是邮箱地址
+            client = DuckMailClient(
+                base_url=config.basic.duckmail_base_url,
+                proxy=config.basic.proxy,
+                verify_ssl=config.basic.duckmail_verify_ssl,
+                api_key=config.basic.duckmail_api_key,
+                log_callback=log_cb,
+            )
+            client.set_credentials(account_id, mail_password)
+        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,
+        )
+        try:
+            result = automation.login_and_extract(account_id, client)
+        except Exception as exc:
+            return {"success": False, "email": account_id, "error": str(exc)}
+        if not result.get("success"):
+            return {"success": False, "email": account_id, "error": result.get("error", "automation failed")}
+
+        # 更新账户配置
+        config_data = result["config"]
+        config_data["mail_provider"] = mail_provider
+        config_data["mail_password"] = mail_password
+        if mail_provider == "microsoft":
+            config_data["mail_address"] = account.get("mail_address") or account_id
+            config_data["mail_client_id"] = mail_client_id
+            config_data["mail_refresh_token"] = mail_refresh_token
+            config_data["mail_tenant"] = mail_tenant
+        config_data["disabled"] = account.get("disabled", False)
+
+        for acc in accounts:
+            if acc.get("id") == account_id:
+                acc.update(config_data)
+                break
+
+        self._apply_accounts_update(accounts)
+        return {"success": True, "email": account_id, "config": config_data}
+
+
+    def _get_expiring_accounts(self) -> List[str]:
+        accounts = load_accounts_from_source()
+        expiring = []
+        beijing_tz = timezone(timedelta(hours=8))
+        now = datetime.now(beijing_tz)
+
+        for account in accounts:
+            if account.get("disabled"):
+                continue
+            mail_provider = (account.get("mail_provider") or "").lower()
+            if not mail_provider:
+                if account.get("mail_client_id") or account.get("mail_refresh_token"):
+                    mail_provider = "microsoft"
+                else:
+                    mail_provider = "duckmail"
+
+            mail_password = account.get("mail_password") or account.get("email_password")
+            if mail_provider == "microsoft":
+                if not account.get("mail_client_id") or not account.get("mail_refresh_token"):
+                    continue
+            else:
+                if not mail_password:
+                    continue
+            expires_at = account.get("expires_at")
+            if not expires_at:
+                continue
+
+            try:
+                expire_time = datetime.strptime(expires_at, "%Y-%m-%d %H:%M:%S")
+                expire_time = expire_time.replace(tzinfo=beijing_tz)
+                remaining = (expire_time - now).total_seconds() / 3600
+            except Exception:
+                continue
+
+            if remaining <= config.basic.refresh_window_hours:
+                expiring.append(account.get("id"))
+
+        return expiring
+
+    async def check_and_refresh(self) -> None:
+        if os.environ.get("ACCOUNTS_CONFIG"):
+            logger.info("[LOGIN] ACCOUNTS_CONFIG set, skipping refresh")
+            return
+        expiring_accounts = self._get_expiring_accounts()
+        if not expiring_accounts:
+            logger.debug("[LOGIN] no accounts need refresh")
+            return
+
+        try:
+            await self.start_login(expiring_accounts)
+        except ValueError as exc:
+            logger.warning("[LOGIN] %s", exc)
+
+    async def start_polling(self) -> None:
+        if self._is_polling:
+            logger.warning("[LOGIN] polling already running")
+            return
+
+        self._is_polling = True
+        logger.info("[LOGIN] refresh polling started (interval: 30 minutes)")
+        try:
+            while self._is_polling:
+                await self.check_and_refresh()
+                await asyncio.sleep(1800)
+        except asyncio.CancelledError:
+            logger.info("[LOGIN] polling stopped")
+        except Exception as exc:
+            logger.error("[LOGIN] polling error: %s", exc)
+        finally:
+            self._is_polling = False
+
+    def stop_polling(self) -> None:
+        self._is_polling = False
+        logger.info("[LOGIN] stopping polling")

core/mail_utils.py

@@ -0,0 +1,33 @@
+import re
+from typing import Optional
+
+
+def extract_verification_code(text: str) -> Optional[str]:
+    """提取验证码"""
+    if not text:
+        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:
+        candidate = match.group(1)
+        # 排除 CSS 单位值
+        if not re.match(r"^\d+(?:px|pt|em|rem|vh|vw|%)$", candidate, re.IGNORECASE):
+            return candidate
+
+    # 策略2: 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

@@ -0,0 +1,176 @@
+import imaplib
+import time
+from datetime import datetime, timedelta
+from email import message_from_bytes
+from email.utils import parsedate_to_datetime
+from typing import Optional
+
+import requests
+
+from core.mail_utils import extract_verification_code
+
+
+class MicrosoftMailClient:
+    def __init__(
+        self,
+        client_id: str,
+        refresh_token: str,
+        tenant: str = "consumers",
+        proxy: str = "",
+        log_callback=None,
+    ) -> None:
+        self.client_id = client_id
+        self.refresh_token = refresh_token
+        self.tenant = tenant or "consumers"
+        self.proxies = {"http": proxy, "https": proxy} if proxy else None
+        self.log_callback = log_callback
+        self.email: Optional[str] = None
+
+    def set_credentials(self, email: str, password: Optional[str] = None) -> None:
+        self.email = email
+
+    def _get_access_token(self) -> Optional[str]:
+        url = f"https://login.microsoftonline.com/{self.tenant}/oauth2/v2.0/token"
+        data = {
+            "client_id": self.client_id,
+            "grant_type": "refresh_token",
+            "refresh_token": self.refresh_token,
+        }
+        try:
+            res = requests.post(url, data=data, proxies=self.proxies, timeout=15)
+            if res.status_code != 200:
+                self._log("error", f"Microsoft token error: {res.status_code}")
+                return None
+            payload = res.json() if res.content else {}
+            token = payload.get("access_token")
+            if not token:
+                self._log("error", "Microsoft token missing")
+                return None
+            return token
+        except Exception as exc:
+            self._log("error", f"Microsoft token exception: {exc}")
+            return None
+
+    def fetch_verification_code(self, since_time: Optional[datetime] = None) -> Optional[str]:
+        if not self.email:
+            return None
+        token = self._get_access_token()
+        if not token:
+            return None
+
+        auth_string = f"user={self.email}\x01auth=Bearer {token}\x01\x01".encode()
+        client = imaplib.IMAP4_SSL("outlook.office365.com", 993)
+        try:
+            client.authenticate("XOAUTH2", lambda _: auth_string)
+        except Exception as exc:
+            self._log("error", f"Microsoft 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")
+
+        try:
+            for mailbox in ("INBOX", "Junk"):
+                try:
+                    status, _ = client.select(mailbox, readonly=True)
+                    if status != "OK":
+                        continue
+                except Exception:
+                    continue
+
+                status, data = client.search(None, "SINCE", since_str)
+                if status != "OK" or not data or not data[0]:
+                    continue
+
+                ids = data[0].split()
+                ids = ids[-20:]
+                for msg_id in reversed(ids):
+                    status, msg_data = client.fetch(msg_id, "(RFC822)")
+                    if status != "OK" or not msg_data:
+                        continue
+                    raw_bytes = None
+                    for item in msg_data:
+                        if isinstance(item, tuple) and len(item) > 1:
+                            raw_bytes = item[1]
+                            break
+                    if not raw_bytes:
+                        continue
+
+                    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:
+                        return code
+        finally:
+            try:
+                client.logout()
+            except Exception:
+                pass
+
+        return None
+
+    def poll_for_code(
+        self,
+        timeout: int = 120,
+        interval: int = 4,
+        since_time: Optional[datetime] = None,
+    ) -> Optional[str]:
+        if not self.email:
+            return None
+
+        max_retries = max(1, timeout // interval)
+        for _ in range(max_retries):
+            code = self.fetch_verification_code(since_time=since_time)
+            if code:
+                return code
+            if interval:
+                time.sleep(interval)
+        return None
+
+    @staticmethod
+    def _message_to_text(msg) -> str:
+        if msg.is_multipart():
+            parts = []
+            for part in msg.walk():
+                content_type = part.get_content_type()
+                if content_type not in ("text/plain", "text/html"):
+                    continue
+                payload = part.get_payload(decode=True)
+                if not payload:
+                    continue
+                charset = part.get_content_charset() or "utf-8"
+                parts.append(payload.decode(charset, errors="ignore"))
+            return "".join(parts)
+        payload = msg.get_payload(decode=True)
+        if isinstance(payload, bytes):
+            return payload.decode(msg.get_content_charset() or "utf-8", errors="ignore")
+        return str(payload) if payload else ""
+
+    @staticmethod
+    def _parse_message_date(value: Optional[str]) -> Optional[datetime]:
+        if not value:
+            return None
+        try:
+            parsed = parsedate_to_datetime(value)
+            if parsed is None:
+                return None
+            if parsed.tzinfo:
+                return parsed.astimezone(tz=None).replace(tzinfo=None)
+            return parsed
+        except Exception:
+            return None
+
+    def _log(self, level: str, message: str) -> None:
+        if self.log_callback:
+            try:
+                self.log_callback(level, message)
+            except Exception:
+                pass

core/register_service.py

@@ -0,0 +1,150 @@
+import asyncio
+import logging
+import os
+import time
+import uuid
+from dataclasses import dataclass, field
+from typing import Any, Callable, Dict, List, Optional
+
+from core.account import load_accounts_from_source
+from core.base_task_service import BaseTask, BaseTaskService, TaskStatus
+from core.config import config
+from core.duckmail_client import DuckMailClient
+from core.gemini_automation import GeminiAutomation
+from core.gemini_automation_uc import GeminiAutomationUC
+
+logger = logging.getLogger("gemini.register")
+
+
+@dataclass
+class RegisterTask(BaseTask):
+    """注册任务数据类"""
+    count: int = 0
+
+    def to_dict(self) -> dict:
+        """转换为字典"""
+        base_dict = super().to_dict()
+        base_dict["count"] = self.count
+        return base_dict
+
+
+class RegisterService(BaseTaskService[RegisterTask]):
+    """注册服务类"""
+
+    def __init__(
+        self,
+        multi_account_mgr,
+        http_client,
+        user_agent: str,
+        account_failure_threshold: int,
+        rate_limit_cooldown_seconds: int,
+        session_cache_ttl_seconds: int,
+        global_stats_provider: Callable[[], dict],
+        set_multi_account_mgr: Optional[Callable[[Any], None]] = None,
+    ) -> None:
+        super().__init__(
+            multi_account_mgr,
+            http_client,
+            user_agent,
+            account_failure_threshold,
+            rate_limit_cooldown_seconds,
+            session_cache_ttl_seconds,
+            global_stats_provider,
+            set_multi_account_mgr,
+            log_prefix="REGISTER",
+        )
+
+    async def start_register(self, count: Optional[int] = None, domain: Optional[str] = None) -> RegisterTask:
+        """启动注册任务"""
+        async with self._lock:
+            if os.environ.get("ACCOUNTS_CONFIG"):
+                raise ValueError("ACCOUNTS_CONFIG is set; register is disabled")
+            if self._current_task_id:
+                current = self._tasks.get(self._current_task_id)
+                if current and current.status == TaskStatus.RUNNING:
+                    raise ValueError("register task already running")
+
+            domain_value = (domain or "").strip()
+            if not domain_value:
+                domain_value = (config.basic.register_domain or "").strip() or None
+
+            register_count = count or config.basic.register_default_count
+            register_count = max(1, min(30, int(register_count)))
+            task = RegisterTask(id=str(uuid.uuid4()), count=register_count)
+            self._tasks[task.id] = task
+            self._current_task_id = task.id
+            self._append_log(task, "info", f"register task created (count={register_count})")
+            asyncio.create_task(self._run_register_async(task, domain_value))
+            return task
+
+    async def _run_register_async(self, task: RegisterTask, domain: Optional[str]) -> None:
+        """异步执行注册任务"""
+        task.status = TaskStatus.RUNNING
+        loop = asyncio.get_running_loop()
+        self._append_log(task, "info", "register task started")
+
+        for _ in range(task.count):
+            try:
+                result = await loop.run_in_executor(self._executor, self._register_one, domain, task)
+            except Exception as exc:
+                result = {"success": False, "error": str(exc)}
+            task.progress += 1
+            task.results.append(result)
+
+            if result.get("success"):
+                task.success_count += 1
+                self._append_log(task, "info", f"register success: {result.get('email')}")
+            else:
+                task.fail_count += 1
+                self._append_log(task, "error", f"register failed: {result.get('error')}")
+
+        task.status = TaskStatus.SUCCESS if task.fail_count == 0 else TaskStatus.FAILED
+        task.finished_at = time.time()
+        self._current_task_id = None
+        self._append_log(task, "info", f"register task finished ({task.success_count}/{task.count})")
+
+    def _register_one(self, domain: Optional[str], task: RegisterTask) -> dict:
+        """注册单个账户"""
+        log_cb = lambda level, message: self._append_log(task, level, message)
+        client = DuckMailClient(
+            base_url=config.basic.duckmail_base_url,
+            proxy=config.basic.proxy,
+            verify_ssl=config.basic.duckmail_verify_ssl,
+            api_key=config.basic.duckmail_api_key,
+            log_callback=log_cb,
+        )
+        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,
+        )
+
+        try:
+            result = automation.login_and_extract(client.email, client)
+        except Exception as exc:
+            return {"success": False, "error": str(exc)}
+        if not result.get("success"):
+            return {"success": False, "error": result.get("error", "automation failed")}
+
+        config_data = result["config"]
+        config_data["mail_provider"] = "duckmail"
+        config_data["mail_address"] = client.email
+        config_data["mail_password"] = client.password
+
+        accounts_data = load_accounts_from_source()
+        updated = False
+        for acc in accounts_data:
+            if acc.get("id") == config_data["id"]:
+                acc.update(config_data)
+                updated = True
+                break
+        if not updated:
+            accounts_data.append(config_data)
+
+        self._apply_accounts_update(accounts_data)
+
+        return {"success": True, "email": client.email, "config": config_data}