Upload 18 files

core/config.py

@@ -49,8 +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_engine: str = Field(default="dp", description="浏览器引擎：uc (支持无头/有头) 或 dp (仅有头，更稳定)")
-    browser_headless: bool = Field(default=False, description="自动化浏览器无头模式（仅 UC 引擎支持）")
+    browser_engine: str = Field(default="dp", description="浏览器引擎：uc 或 dp")
+    browser_headless: bool = Field(default=False, 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="默认注册域名（推荐）")

core/duckmail_client.py

@@ -147,36 +147,52 @@ class DuckMailClient:
             if not messages:
                 return None
 
-            # 只获取最新一封邮件，不做时间过滤
-            msg_id = messages[0].get("id")
-            if not msg_id:
-                return None
-
-            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 {}
-
-            # 获取邮件内容
-            text_content = payload.get("text") or ""
-            html_content = payload.get("html") or ""
+            # 遍历邮件，过滤时间
+            for msg in messages:
+                msg_id = msg.get("id")
+                if not msg_id:
+                    continue
+
+                # 时间过滤
+                if since_time:
+                    created_at = msg.get("createdAt")
+                    if created_at:
+                        from datetime import datetime
+                        import re
+                        # 截断纳秒到微秒（fromisoformat 只支持6位小数）
+                        created_at = re.sub(r'(\.\d{6})\d+', r'\1', created_at)
+                        # 转换 UTC 时间到本地时区
+                        msg_time = datetime.fromisoformat(created_at.replace("Z", "+00:00")).astimezone().replace(tzinfo=None)
+                        if msg_time < since_time:
+                            continue
+
+                detail = self._request(
+                    "GET",
+                    f"{self.base_url}/messages/{msg_id}",
+                    headers={"Authorization": f"Bearer {self.token}"},
+                )
+
+                if detail.status_code != 200:
+                    continue
+
+                payload = detail.json() if detail.content else {}
+
+                # 获取邮件内容
+                text_content = payload.get("text") or ""
+                html_content = payload.get("html") or ""
+
+                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
+                code = extract_verification_code(content)
+                if code:
+                    self._log("info", f"code found: {code}")
+                    return code
 
-            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
-            code = extract_verification_code(content)
-            if code:
-                self._log("info", f"code found: {code}")
-            return code
+            return None
 
         except Exception as e:
             self._log("error", f"fetch code failed: {e}")

core/gemini_automation.py

@@ -1,10 +1,11 @@
 """
 Gemini自动化登录模块（用于新账号注册）
 """
+import os
 import random
 import string
 import time
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import Optional
 from urllib.parse import quote
 
@@ -15,6 +16,22 @@ from DrissionPage import ChromiumPage, ChromiumOptions
 AUTH_HOME_URL = "https://auth.business.gemini.google/"
 DEFAULT_XSRF_TOKEN = "KdLRzKwwBTD5wo8nUollAbY6cW0"
 
+# Linux 下常见的 Chromium 路径
+CHROMIUM_PATHS = [
+    "/usr/bin/chromium",
+    "/usr/bin/chromium-browser",
+    "/usr/bin/google-chrome",
+    "/usr/bin/google-chrome-stable",
+]
+
+
+def _find_chromium_path() -> Optional[str]:
+    """查找可用的 Chromium/Chrome 浏览器路径"""
+    for path in CHROMIUM_PATHS:
+        if os.path.isfile(path) and os.access(path, os.X_OK):
+            return path
+    return None
+
 
 class GeminiAutomation:
     """Gemini自动化登录"""
@@ -36,8 +53,10 @@ class GeminiAutomation:
     def login_and_extract(self, email: str, mail_client) -> dict:
         """执行登录并提取配置"""
         page = None
+        user_data_dir = None
         try:
             page = self._create_page()
+            user_data_dir = getattr(page, 'user_data_dir', None)
             return self._run_flow(page, email, mail_client)
         except Exception as exc:
             self._log("error", f"automation error: {exc}")
@@ -48,16 +67,29 @@ class GeminiAutomation:
                     page.quit()
                 except Exception:
                     pass
+            self._cleanup_user_data(user_data_dir)
 
     def _create_page(self) -> ChromiumPage:
         """创建浏览器页面"""
         options = ChromiumOptions()
+
+        # 自动检测 Chromium 浏览器路径（Linux/Docker 环境）
+        chromium_path = _find_chromium_path()
+        if chromium_path:
+            options.set_browser_path(chromium_path)
+            self._log("info", f"using browser: {chromium_path}")
+
+        options.set_argument("--incognito")
         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)
 
+        # 语言设置（确保使用中文界面）
+        options.set_argument("--lang=zh-CN")
+        options.set_pref("intl.accept_languages", "zh-CN,zh")
+
         if self.proxy:
             options.set_argument(f"--proxy-server={self.proxy}")
 
@@ -70,7 +102,6 @@ class GeminiAutomation:
             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()
@@ -85,6 +116,23 @@ class GeminiAutomation:
                     Object.defineProperty(navigator, 'plugins', {get: () => [1, 2, 3, 4, 5]});
                     Object.defineProperty(navigator, 'languages', {get: () => ['zh-CN', 'zh', 'en']});
                     window.chrome = {runtime: {}};
+
+                    // 额外的反检测措施
+                    Object.defineProperty(navigator, 'maxTouchPoints', {get: () => 1});
+                    Object.defineProperty(navigator, 'platform', {get: () => 'Win32'});
+                    Object.defineProperty(navigator, 'vendor', {get: () => 'Google Inc.'});
+
+                    // 隐藏 headless 特征
+                    Object.defineProperty(navigator, 'hardwareConcurrency', {get: () => 8});
+                    Object.defineProperty(navigator, 'deviceMemory', {get: () => 8});
+
+                    // 模拟真实的 permissions
+                    const originalQuery = window.navigator.permissions.query;
+                    window.navigator.permissions.query = (parameters) => (
+                        parameters.name === 'notifications' ?
+                            Promise.resolve({state: Notification.permission}) :
+                            originalQuery(parameters)
+                    );
                 """)
             except Exception:
                 pass
@@ -94,6 +142,10 @@ class GeminiAutomation:
     def _run_flow(self, page, email: str, mail_client) -> dict:
         """执行登录流程"""
 
+        # 记录开始时间，用于邮件时间过滤
+        from datetime import datetime
+        send_time = datetime.now()
+
         # Step 1: 导航到首页并设置 Cookie
         self._log("info", f"navigating to login page for {email}")
 
@@ -132,10 +184,7 @@ class GeminiAutomation:
         if has_business_params:
             return self._extract_config(page, email)
 
-        # Step 3: 记录发送验证码的时间并触发发送
-        from datetime import datetime
-        send_time = datetime.now()
-
+        # Step 3: 点击发送验证码按钮
         self._log("info", "clicking send verification code button")
         if not self._click_send_code_button(page):
             self._log("error", "send code button not found")
@@ -154,9 +203,22 @@ class GeminiAutomation:
         code = mail_client.poll_for_code(timeout=40, interval=4, since_time=send_time)
 
         if not code:
-            self._log("error", "verification code timeout")
-            self._save_screenshot(page, "code_timeout")
-            return {"success": False, "error": "verification code timeout"}
+            self._log("warning", "verification code timeout, trying to resend")
+            # 更新发送时间（在点击按钮之前记录）
+            send_time = datetime.now()
+            # 尝试点击重新发送按钮
+            if self._click_resend_code_button(page):
+                self._log("info", "resend button clicked, waiting for new code")
+                # 再次轮询验证码
+                code = mail_client.poll_for_code(timeout=40, interval=4, since_time=send_time)
+                if not code:
+                    self._log("error", "verification code timeout after resend")
+                    self._save_screenshot(page, "code_timeout_after_resend")
+                    return {"success": False, "error": "verification code timeout after resend"}
+            else:
+                self._log("error", "verification code timeout and resend button not found")
+                self._save_screenshot(page, "code_timeout")
+                return {"success": False, "error": "verification code timeout"}
 
         self._log("info", f"code received: {code}")
 
@@ -168,46 +230,76 @@ class GeminiAutomation:
             self._log("error", "code input expired")
             return {"success": False, "error": "code input expired"}
 
+        self._log("info", "inputting verification code")
         code_input.input(code, clear=True)
         time.sleep(0.5)
 
         verify_btn = page.ele("css:button[jsname='XooR8e']", timeout=3)
         if verify_btn:
+            self._log("info", "clicking verify button (method 1)")
             verify_btn.click()
         else:
             verify_btn = self._find_verify_button(page)
             if verify_btn:
+                self._log("info", "clicking verify button (method 2)")
                 verify_btn.click()
             else:
+                self._log("info", "pressing enter to submit")
                 code_input.input("\n")
 
-        time.sleep(5)
+        # Step 7: 等待页面自动重定向（提交验证码后 Google 会自动跳转）
+        self._log("info", "waiting for auto-redirect after verification")
+        time.sleep(12)  # 增加等待时间，让页面有足够时间完成重定向（如果网络慢可以继续增加）
+
+        # 记录当前 URL 状态
+        current_url = page.url
+        self._log("info", f"current URL after verification: {current_url}")
+
+        # 检查是否还停留在验证码页面（说明提交失败）
+        if "verify-oob-code" in current_url:
+            self._log("error", "verification code submission failed, still on verification page")
+            self._save_screenshot(page, "verification_submit_failed")
+            return {"success": False, "error": "verification code submission failed"}
 
-        # Step 7: 处理协议页面（如果有）
+        # Step 8: 处理协议页面（如果有）
         self._handle_agreement_page(page)
 
-        # Step 8: 导航到业务页面并等待参数生成
-        self._log("info", "navigating to business page")
-        page.get("https://business.gemini.google/", timeout=self.timeout)
-        time.sleep(3)
+        # Step 9: 检查是否已经在正确的页面
+        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:
+            # 已经在正确的页面，不需要再次导航
+            self._log("info", "already on business page with parameters")
+            return self._extract_config(page, email)
 
-        # Step 9: 检查是否需要设置用户名
+        # Step 10: 如果不在正确的页面，尝试导航
+        if "business.gemini.google" not in current_url:
+            self._log("info", "navigating to business page")
+            page.get("https://business.gemini.google/", timeout=self.timeout)
+            time.sleep(5)  # 增加等待时间
+            current_url = page.url
+            self._log("info", f"URL after navigation: {current_url}")
+
+        # Step 11: 检查是否需要设置用户名
         if "cid" not in page.url:
             if self._handle_username_setup(page):
-                time.sleep(3)
+                time.sleep(5)  # 增加等待时间
 
-        # Step 10: 等待 URL 参数生成（csesidx 和 cid）
+        # Step 12: 等待 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)
+            time.sleep(5)  # 增加等待时间
             if not self._wait_for_business_params(page):
                 self._log("error", "URL parameters generation failed")
+                current_url = page.url
+                self._log("error", f"final URL: {current_url}")
                 self._save_screenshot(page, "params_missing")
                 return {"success": False, "error": "URL parameters not found"}
 
-        # Step 11: 提取配置
+        # Step 13: 提取配置
         self._log("info", "login success")
         return self._extract_config(page, email)
 
@@ -277,6 +369,28 @@ class GeminiAutomation:
             pass
         return None
 
+    def _click_resend_code_button(self, page) -> bool:
+        """点击重新发送验证码按钮"""
+        time.sleep(2)
+
+        # 查找包含重新发送关键词的按钮（与 _find_verify_button 相反）
+        try:
+            buttons = page.eles("tag:button")
+            for btn in buttons:
+                text = (btn.text or "").strip().lower()
+                if text and ("重新" in text or "resend" in text):
+                    try:
+                        self._log("info", f"found resend button: {text}")
+                        btn.click()
+                        time.sleep(2)
+                        return True
+                    except Exception:
+                        pass
+        except Exception:
+            pass
+
+        return False
+
     def _handle_agreement_page(self, page) -> None:
         """处理协议页面"""
         if "/admin/create" in page.url:
@@ -376,10 +490,14 @@ class GeminiAutomation:
             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)
+            # 使用北京时区，确保时间计算正确（Cookie expiry 是 UTC 时间戳）
+            beijing_tz = timezone(timedelta(hours=8))
             if ses_obj and "expiry" in ses_obj:
-                expires_at = datetime.fromtimestamp(ses_obj["expiry"] - 43200).strftime("%Y-%m-%d %H:%M:%S")
+                # 将 UTC 时间戳转为北京时间，再减去12小时作为刷新窗口
+                cookie_expire_beijing = datetime.fromtimestamp(ses_obj["expiry"], tz=beijing_tz)
+                expires_at = (cookie_expire_beijing - timedelta(hours=12)).strftime("%Y-%m-%d %H:%M:%S")
             else:
-                expires_at = (datetime.now() + timedelta(hours=12)).strftime("%Y-%m-%d %H:%M:%S")
+                expires_at = (datetime.now(beijing_tz) + timedelta(hours=12)).strftime("%Y-%m-%d %H:%M:%S")
 
             config = {
                 "id": email,
@@ -412,6 +530,17 @@ class GeminiAutomation:
             except Exception:
                 pass
 
+    def _cleanup_user_data(self, user_data_dir: Optional[str]) -> None:
+        """清理浏览器用户数据目录"""
+        if not user_data_dir:
+            return
+        try:
+            import shutil
+            if os.path.exists(user_data_dir):
+                shutil.rmtree(user_data_dir, ignore_errors=True)
+        except Exception:
+            pass
+
     @staticmethod
     def _get_ua() -> str:
         """生成随机User-Agent"""

core/gemini_automation_uc.py

@@ -5,7 +5,7 @@ Gemini自动化登录模块（使用 undetected-chromedriver）
 import random
 import string
 import time
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import Optional
 from urllib.parse import quote
 
@@ -39,6 +39,7 @@ class GeminiAutomationUC:
         self.timeout = timeout
         self.log_callback = log_callback
         self.driver = None
+        self.user_data_dir = None
 
     def login_and_extract(self, email: str, mail_client) -> dict:
         """执行登录并提取配置"""
@@ -53,13 +54,25 @@ class GeminiAutomationUC:
 
     def _create_driver(self):
         """创建浏览器驱动"""
+        import tempfile
         options = uc.ChromeOptions()
 
+        # 创建临时用户数据目录
+        self.user_data_dir = tempfile.mkdtemp(prefix='uc-profile-')
+        options.add_argument(f"--user-data-dir={self.user_data_dir}")
+
         # 基础参数
+        options.add_argument("--incognito")
         options.add_argument("--no-sandbox")
         options.add_argument("--disable-setuid-sandbox")
         options.add_argument("--window-size=1280,800")
 
+        # 语言设置（确保使用中文界面）
+        options.add_argument("--lang=zh-CN")
+        options.add_experimental_option("prefs", {
+            "intl.accept_languages": "zh-CN,zh"
+        })
+
         # 代理设置
         if self.proxy:
             options.add_argument(f"--proxy-server={self.proxy}")
@@ -88,6 +101,10 @@ class GeminiAutomationUC:
     def _run_flow(self, email: str, mail_client) -> dict:
         """执行登录流程"""
 
+        # 记录开始时间，用于邮件时间过滤
+        from datetime import datetime
+        send_time = datetime.now()
+
         self._log("info", f"navigating to login page for {email}")
 
         # 访问登录页面
@@ -130,10 +147,6 @@ 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():
@@ -401,12 +414,15 @@ class GeminiAutomationUC:
             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)
+            beijing_tz = timezone(timedelta(hours=8))
             if ses_obj and "expiry" in ses_obj:
-                expires_at = datetime.fromtimestamp(ses_obj["expiry"] - 43200).strftime("%Y-%m-%d %H:%M:%S")
+                # Cookie expiry 是 UTC 时间戳，转为北京时间后减去12小时作为刷新窗口
+                cookie_expire_beijing = datetime.fromtimestamp(ses_obj["expiry"], tz=beijing_tz)
+                expires_at = (cookie_expire_beijing - timedelta(hours=12)).strftime("%Y-%m-%d %H:%M:%S")
             else:
-                expires_at = (datetime.now() + timedelta(hours=12)).strftime("%Y-%m-%d %H:%M:%S")
+                expires_at = (datetime.now(beijing_tz) + timedelta(hours=12)).strftime("%Y-%m-%d %H:%M:%S")
 
             config = {
                 "id": email,
@@ -439,6 +455,15 @@ class GeminiAutomationUC:
             except Exception:
                 pass
 
+        if self.user_data_dir:
+            try:
+                import shutil
+                import os
+                if os.path.exists(self.user_data_dir):
+                    shutil.rmtree(self.user_data_dir, ignore_errors=True)
+            except Exception:
+                pass
+
     def _log(self, level: str, message: str) -> None:
         """记录日志"""
         if self.log_callback:

core/google_api.py

@@ -2,12 +2,12 @@
 
 负责与Google Gemini Business API的所有交互操作
 """
-import asyncio
-import json
-import logging
-import os
-import time
-import uuid
+import asyncio
+import json
+import logging
+import os
+import time
+import uuid
 from typing import TYPE_CHECKING, List
 
 import httpx
@@ -163,20 +163,20 @@ async def upload_context_file(
     )
 
     req_tag = f"[req_{request_id}] " if request_id else ""
-    if r.status_code != 200:
-        logger.error(f"[FILE] [{account_manager.config.account_id}] {req_tag}文件上传失败: {r.status_code}")
-        error_text = r.text
-        if r.status_code == 400:
-            try:
-                payload = json.loads(r.text or "{}")
-                message = payload.get("error", {}).get("message", "")
-            except Exception:
-                message = ""
-            if "Unsupported file type" in message:
-                mime_type = message.split("Unsupported file type:", 1)[-1].strip()
-                hint = f"不支持的文件类型: {mime_type}。请转换为 PDF、图片或纯文本后再上传。"
-                raise HTTPException(400, hint)
-        raise HTTPException(r.status_code, f"Upload failed: {error_text}")
+    if r.status_code != 200:
+        logger.error(f"[FILE] [{account_manager.config.account_id}] {req_tag}文件上传失败: {r.status_code}")
+        error_text = r.text
+        if r.status_code == 400:
+            try:
+                payload = json.loads(r.text or "{}")
+                message = payload.get("error", {}).get("message", "")
+            except Exception:
+                message = ""
+            if "Unsupported file type" in message:
+                mime_type = message.split("Unsupported file type:", 1)[-1].strip()
+                hint = f"不支持的文件类型: {mime_type}。请转换为 PDF、图片或纯文本后再上传。"
+                raise HTTPException(400, hint)
+        raise HTTPException(r.status_code, f"Upload failed: {error_text}")
 
     data = r.json()
     file_id = data.get("addContextFileResponse", {}).get("fileId")

core/login_service.py

@@ -154,20 +154,31 @@ class LoginService(BaseTaskService[LoginTask]):
 
         # 根据配置选择浏览器引擎
         browser_engine = (config.basic.browser_engine or "dp").lower()
+        headless = config.basic.browser_headless
+
+        # Linux 环境强制使用 DP 无头模式（无图形界面无法运行有头模式）
+        import sys
+        is_linux = sys.platform.startswith("linux")
+        if is_linux:
+            if browser_engine != "dp" or not headless:
+                log_cb("warning", "Linux environment: forcing DP engine with headless mode")
+                browser_engine = "dp"
+                headless = True
+
         if browser_engine == "dp":
-            # DrissionPage 引擎：仅有头模式，更稳定
+            # DrissionPage 引擎：支持有头和无头模式
             automation = GeminiAutomation(
                 user_agent=self.user_agent,
                 proxy=config.basic.proxy,
-                headless=False,  # DP 不支持无头模式
+                headless=headless,
                 log_callback=log_cb,
             )
         else:
-            # undetected-chromedriver 引擎：支持有头和无头
+            # undetected-chromedriver 引擎：仅有头模式可用
             automation = GeminiAutomationUC(
                 user_agent=self.user_agent,
                 proxy=config.basic.proxy,
-                headless=config.basic.browser_headless,
+                headless=headless,
                 log_callback=log_cb,
             )
         try:

core/message.py

@@ -103,36 +103,36 @@ async def parse_last_message(messages: List['Message'], http_client: httpx.Async
                 else:
                     logger.warning(f"[FILE] [req_{request_id}] 不支持的文件格式: {url[:30]}...")
 
-    # 并行下载所有 URL 文件（支持图片、PDF、文档等）
-    if image_urls:
-        async def download_url(url: str):
-            try:
-                resp = await http_client.get(url, timeout=30, follow_redirects=True)
-                if resp.status_code == 404:
-                    logger.warning(f"[FILE] [req_{request_id}] URL文件已失效(404)，已跳过: {url[:50]}...")
-                    return None
-                resp.raise_for_status()
-                content_type = resp.headers.get("content-type", "application/octet-stream").split(";")[0]
-                # 移除图片类型限制，支持所有文件类型
-                b64 = base64.b64encode(resp.content).decode()
-                logger.info(f"[FILE] [req_{request_id}] URL文件下载成功: {url[:50]}... ({len(resp.content)} bytes, {content_type})")
-                return {"mime": content_type, "data": b64}
-            except httpx.HTTPStatusError as e:
-                status_code = e.response.status_code if e.response else "unknown"
-                logger.warning(f"[FILE] [req_{request_id}] URL文件下载失败({status_code}): {url[:50]}... - {e}")
-                return None
-            except Exception as e:
-                logger.warning(f"[FILE] [req_{request_id}] URL文件下载失败: {url[:50]}... - {e}")
-                return None
-
-        results = await asyncio.gather(*[download_url(u) for u in image_urls], return_exceptions=True)
-        safe_results = []
-        for result in results:
-            if isinstance(result, Exception):
-                logger.warning(f"[FILE] [req_{request_id}] URL文件下载异常: {type(result).__name__}: {str(result)[:120]}")
-                continue
-            safe_results.append(result)
-        images.extend([r for r in safe_results if r])
+    # 并行下载所有 URL 文件（支持图片、PDF、文档等）
+    if image_urls:
+        async def download_url(url: str):
+            try:
+                resp = await http_client.get(url, timeout=30, follow_redirects=True)
+                if resp.status_code == 404:
+                    logger.warning(f"[FILE] [req_{request_id}] URL文件已失效(404)，已跳过: {url[:50]}...")
+                    return None
+                resp.raise_for_status()
+                content_type = resp.headers.get("content-type", "application/octet-stream").split(";")[0]
+                # 移除图片类型限制，支持所有文件类型
+                b64 = base64.b64encode(resp.content).decode()
+                logger.info(f"[FILE] [req_{request_id}] URL文件下载成功: {url[:50]}... ({len(resp.content)} bytes, {content_type})")
+                return {"mime": content_type, "data": b64}
+            except httpx.HTTPStatusError as e:
+                status_code = e.response.status_code if e.response else "unknown"
+                logger.warning(f"[FILE] [req_{request_id}] URL文件下载失败({status_code}): {url[:50]}... - {e}")
+                return None
+            except Exception as e:
+                logger.warning(f"[FILE] [req_{request_id}] URL文件下载失败: {url[:50]}... - {e}")
+                return None
+
+        results = await asyncio.gather(*[download_url(u) for u in image_urls], return_exceptions=True)
+        safe_results = []
+        for result in results:
+            if isinstance(result, Exception):
+                logger.warning(f"[FILE] [req_{request_id}] URL文件下载异常: {type(result).__name__}: {str(result)[:120]}")
+                continue
+            safe_results.append(result)
+        images.extend([r for r in safe_results if r])
 
     return text_content, images
 

core/register_service.py

@@ -118,20 +118,31 @@ class RegisterService(BaseTaskService[RegisterTask]):
 
         # 根据配置选择浏览器引擎
         browser_engine = (config.basic.browser_engine or "dp").lower()
+        headless = config.basic.browser_headless
+
+        # Linux 环境强制使用 DP 无头模式（无图形界面无法运行有头模式）
+        import sys
+        is_linux = sys.platform.startswith("linux")
+        if is_linux:
+            if browser_engine != "dp" or not headless:
+                log_cb("warning", "Linux environment: forcing DP engine with headless mode")
+                browser_engine = "dp"
+                headless = True
+
         if browser_engine == "dp":
-            # DrissionPage 引擎：仅有头模式，更稳定
+            # DrissionPage 引擎：支持有头和无头模式
             automation = GeminiAutomation(
                 user_agent=self.user_agent,
                 proxy=config.basic.proxy,
-                headless=False,  # DP 不支持无头模式
+                headless=headless,
                 log_callback=log_cb,
             )
         else:
-            # undetected-chromedriver 引擎：支持有头和无头
+            # undetected-chromedriver 引擎：仅有头模式可用
             automation = GeminiAutomationUC(
                 user_agent=self.user_agent,
                 proxy=config.basic.proxy,
-                headless=config.basic.browser_headless,
+                headless=headless,
                 log_callback=log_cb,
             )
 

core/session_auth.py

@@ -46,11 +46,11 @@ def require_login(redirect_to_login: bool = True):
                     wants_html = "text/html" in accept_header or request.url.path.endswith("/html")
 
                     if wants_html:
-                        # 清理掉 URL 中可能重复的 PATH_PREFIX
-                        # 避免重定向路径出现多层前缀
+                        # 清理掉 URL 中可能重复的 PATH_PREFIX
+                        # 避免重定向路径出现多层前缀
                         path = request.url.path
 
-                        # 兼容 main 中 PATH_PREFIX 为空的情况
+                        # 兼容 main 中 PATH_PREFIX 为空的情况
                         import main
                         prefix = main.PATH_PREFIX
 

core/uptime.py

@@ -1,139 +1,139 @@
-"""
-Uptime 实时监控与心跳历史持久化。
-"""
-
-from collections import deque
-from datetime import datetime, timezone, timedelta
-from typing import Dict, List, Optional
-import json
-import os
-from threading import Lock
-
-# 北京时区 UTC+8
-BEIJING_TZ = timezone(timedelta(hours=8))
-
-# 每个服务保留最近 60 条心跳
-MAX_HEARTBEATS = 60
-SLOW_THRESHOLD_MS = 40000
-WARNING_STATUS_CODES = {429}
-
-_storage_path: Optional[str] = None
-_storage_lock = Lock()
-
-# 服务注册表
-SERVICES = {
-    "api_service": {"name": "API 服务", "heartbeats": deque(maxlen=MAX_HEARTBEATS)},
-    "account_pool": {"name": "服务资源", "heartbeats": deque(maxlen=MAX_HEARTBEATS)},
-    "gemini-2.5-flash": {"name": "Gemini 2.5 Flash", "heartbeats": deque(maxlen=MAX_HEARTBEATS)},
-    "gemini-2.5-pro": {"name": "Gemini 2.5 Pro", "heartbeats": deque(maxlen=MAX_HEARTBEATS)},
-    "gemini-3-flash-preview": {"name": "Gemini 3 Flash Preview", "heartbeats": deque(maxlen=MAX_HEARTBEATS)},
-    "gemini-3-pro-preview": {"name": "Gemini 3 Pro Preview", "heartbeats": deque(maxlen=MAX_HEARTBEATS)},
-}
-
-SUPPORTED_MODELS = ["gemini-2.5-flash", "gemini-2.5-pro", "gemini-3-flash-preview", "gemini-3-pro-preview"]
-
-
-def configure_storage(path: Optional[str]) -> None:
-    """配置心跳持久化路径。"""
-    global _storage_path
-    _storage_path = path
-
-
-def _classify_level(success: bool, status_code: Optional[int], latency_ms: Optional[int]) -> str:
-    if status_code in WARNING_STATUS_CODES:
-        return "warn"
-    if success and latency_ms is not None and latency_ms >= SLOW_THRESHOLD_MS:
-        return "warn"
-    return "up" if success else "down"
-
-
-def _save_heartbeats() -> None:
-    if not _storage_path:
-        return
-    try:
-        payload = {}
-        for service_id, service_data in SERVICES.items():
-            payload[service_id] = list(service_data["heartbeats"])
-        os.makedirs(os.path.dirname(_storage_path), exist_ok=True)
-        with _storage_lock, open(_storage_path, "w", encoding="utf-8") as f:
-            json.dump(payload, f, ensure_ascii=True, indent=2)
-    except Exception:
-        return
-
-
-def load_heartbeats() -> None:
-    if not _storage_path or not os.path.exists(_storage_path):
-        return
-    try:
-        with _storage_lock, open(_storage_path, "r", encoding="utf-8") as f:
-            payload = json.load(f)
-        for service_id, heartbeats in payload.items():
-            if service_id not in SERVICES:
-                continue
-            SERVICES[service_id]["heartbeats"].clear()
-            for beat in heartbeats[-MAX_HEARTBEATS:]:
-                SERVICES[service_id]["heartbeats"].append(beat)
-    except Exception:
-        return
-
-
-def record_request(
-    service: str,
-    success: bool,
-    latency_ms: Optional[int] = None,
-    status_code: Optional[int] = None
-):
-    """记录一次心跳。"""
-    if service not in SERVICES:
-        return
-
-    level = _classify_level(success, status_code, latency_ms)
-    heartbeat = {
-        "time": datetime.now(BEIJING_TZ).strftime("%H:%M:%S"),
-        "success": success,
-        "level": level,
-    }
-    if latency_ms is not None:
-        heartbeat["latency_ms"] = latency_ms
-    if status_code is not None:
-        heartbeat["status_code"] = status_code
-
-    SERVICES[service]["heartbeats"].append(heartbeat)
-    _save_heartbeats()
-
-
-def get_realtime_status() -> Dict:
-    """返回实时监控数据。"""
-    result = {"services": {}}
-
-    for service_id, service_data in SERVICES.items():
-        heartbeats = list(service_data["heartbeats"])
-        total = len(heartbeats)
-        success = sum(1 for h in heartbeats if h.get("success"))
-
-        uptime = (success / total * 100) if total > 0 else 100.0
-
-        last_status = "unknown"
-        if heartbeats:
-            last_level = heartbeats[-1].get("level")
-            if last_level in {"up", "down", "warn"}:
-                last_status = last_level
-            else:
-                last_status = "up" if heartbeats[-1].get("success") else "down"
-
-        result["services"][service_id] = {
-            "name": service_data["name"],
-            "status": last_status,
-            "uptime": round(uptime, 1),
-            "total": total,
-            "success": success,
-            "heartbeats": heartbeats[-MAX_HEARTBEATS:],
-        }
-
-    result["updated_at"] = datetime.now(BEIJING_TZ).strftime("%Y-%m-%d %H:%M:%S")
-    return result
-
-
-async def get_uptime_summary(days: int = 90) -> Dict:
-    """兼容旧接口。"""
-    return get_realtime_status()
+"""
+Uptime 实时监控与心跳历史持久化。
+"""
+
+from collections import deque
+from datetime import datetime, timezone, timedelta
+from typing import Dict, List, Optional
+import json
+import os
+from threading import Lock
+
+# 北京时区 UTC+8
+BEIJING_TZ = timezone(timedelta(hours=8))
+
+# 每个服务保留最近 60 条心跳
+MAX_HEARTBEATS = 60
+SLOW_THRESHOLD_MS = 40000
+WARNING_STATUS_CODES = {429}
+
+_storage_path: Optional[str] = None
+_storage_lock = Lock()
+
+# 服务注册表
+SERVICES = {
+    "api_service": {"name": "API 服务", "heartbeats": deque(maxlen=MAX_HEARTBEATS)},
+    "account_pool": {"name": "服务资源", "heartbeats": deque(maxlen=MAX_HEARTBEATS)},
+    "gemini-2.5-flash": {"name": "Gemini 2.5 Flash", "heartbeats": deque(maxlen=MAX_HEARTBEATS)},
+    "gemini-2.5-pro": {"name": "Gemini 2.5 Pro", "heartbeats": deque(maxlen=MAX_HEARTBEATS)},
+    "gemini-3-flash-preview": {"name": "Gemini 3 Flash Preview", "heartbeats": deque(maxlen=MAX_HEARTBEATS)},
+    "gemini-3-pro-preview": {"name": "Gemini 3 Pro Preview", "heartbeats": deque(maxlen=MAX_HEARTBEATS)},
+}
+
+SUPPORTED_MODELS = ["gemini-2.5-flash", "gemini-2.5-pro", "gemini-3-flash-preview", "gemini-3-pro-preview"]
+
+
+def configure_storage(path: Optional[str]) -> None:
+    """配置心跳持久化路径。"""
+    global _storage_path
+    _storage_path = path
+
+
+def _classify_level(success: bool, status_code: Optional[int], latency_ms: Optional[int]) -> str:
+    if status_code in WARNING_STATUS_CODES:
+        return "warn"
+    if success and latency_ms is not None and latency_ms >= SLOW_THRESHOLD_MS:
+        return "warn"
+    return "up" if success else "down"
+
+
+def _save_heartbeats() -> None:
+    if not _storage_path:
+        return
+    try:
+        payload = {}
+        for service_id, service_data in SERVICES.items():
+            payload[service_id] = list(service_data["heartbeats"])
+        os.makedirs(os.path.dirname(_storage_path), exist_ok=True)
+        with _storage_lock, open(_storage_path, "w", encoding="utf-8") as f:
+            json.dump(payload, f, ensure_ascii=True, indent=2)
+    except Exception:
+        return
+
+
+def load_heartbeats() -> None:
+    if not _storage_path or not os.path.exists(_storage_path):
+        return
+    try:
+        with _storage_lock, open(_storage_path, "r", encoding="utf-8") as f:
+            payload = json.load(f)
+        for service_id, heartbeats in payload.items():
+            if service_id not in SERVICES:
+                continue
+            SERVICES[service_id]["heartbeats"].clear()
+            for beat in heartbeats[-MAX_HEARTBEATS:]:
+                SERVICES[service_id]["heartbeats"].append(beat)
+    except Exception:
+        return
+
+
+def record_request(
+    service: str,
+    success: bool,
+    latency_ms: Optional[int] = None,
+    status_code: Optional[int] = None
+):
+    """记录一次心跳。"""
+    if service not in SERVICES:
+        return
+
+    level = _classify_level(success, status_code, latency_ms)
+    heartbeat = {
+        "time": datetime.now(BEIJING_TZ).strftime("%H:%M:%S"),
+        "success": success,
+        "level": level,
+    }
+    if latency_ms is not None:
+        heartbeat["latency_ms"] = latency_ms
+    if status_code is not None:
+        heartbeat["status_code"] = status_code
+
+    SERVICES[service]["heartbeats"].append(heartbeat)
+    _save_heartbeats()
+
+
+def get_realtime_status() -> Dict:
+    """返回实时监控数据。"""
+    result = {"services": {}}
+
+    for service_id, service_data in SERVICES.items():
+        heartbeats = list(service_data["heartbeats"])
+        total = len(heartbeats)
+        success = sum(1 for h in heartbeats if h.get("success"))
+
+        uptime = (success / total * 100) if total > 0 else 100.0
+
+        last_status = "unknown"
+        if heartbeats:
+            last_level = heartbeats[-1].get("level")
+            if last_level in {"up", "down", "warn"}:
+                last_status = last_level
+            else:
+                last_status = "up" if heartbeats[-1].get("success") else "down"
+
+        result["services"][service_id] = {
+            "name": service_data["name"],
+            "status": last_status,
+            "uptime": round(uptime, 1),
+            "total": total,
+            "success": success,
+            "heartbeats": heartbeats[-MAX_HEARTBEATS:],
+        }
+
+    result["updated_at"] = datetime.now(BEIJING_TZ).strftime("%Y-%m-%d %H:%M:%S")
+    return result
+
+
+async def get_uptime_summary(days: int = 90) -> Dict:
+    """兼容旧接口。"""
+    return get_realtime_status()