7

browser_automation.py

@@ -233,6 +233,55 @@ def log_url_change(page, old_url: str, action: str = None):
         log.warning(f"记录URL变化失败: {e}")
 
 
+def acquire_lock_with_keepalive(lock, page, timeout: float = 120, check_interval: float = 2.0) -> bool:
+    """获取锁的同时保持浏览器活跃，防止等待期间浏览器连接断开
+    
+    在并发模式下，多个 Worker 可能需要等待授权回调锁。
+    如果等待时间过长 (>15-30秒)，浏览器连接可能会因为空闲而断开。
+    此函数通过定期访问页面来保持浏览器活跃。
+    
+    Args:
+        lock: threading.Lock 对象
+        page: 浏览器页面对象
+        timeout: 最大等待时间 (秒)
+        check_interval: 检查间隔 (秒)
+    
+    Returns:
+        bool: 是否成功获取锁
+    """
+    import threading
+    
+    start_time = time.time()
+    acquired = False
+    
+    while time.time() - start_time < timeout:
+        # 尝试非阻塞获取锁
+        acquired = lock.acquire(blocking=False)
+        if acquired:
+            return True
+        
+        # 未获取到锁，保持浏览器活跃
+        try:
+            # 访问页面属性以保持连接活跃
+            _ = page.url
+            # 可选：执行一个轻量级操作
+            try:
+                page.run_js("1+1")  # 执行简单的 JavaScript 保持活跃
+            except Exception:
+                pass
+        except Exception as e:
+            # 浏览器连接可能已断开
+            log.warning(f"保持浏览器活跃时出错: {e}")
+            return False
+        
+        # 等待一小段时间后重试
+        time.sleep(check_interval)
+    
+    # 超时，尝试最后一次阻塞获取 (短暂)
+    acquired = lock.acquire(blocking=True, timeout=0.1)
+    return acquired
+
+
 def cleanup_chrome_processes():
     """清理残留的 Chrome 进程 (跨平台支持)"""
     try:
@@ -2154,11 +2203,16 @@ def register_and_authorize(email: str, password: str, use_api_register: bool = T
                 time.sleep(0.5)
 
                 # ========== 授权流程 - CPA/S2A 需要串行执行 (避免回调端口冲突) ==========
-                # 在调用授权函数之前获取锁，确保浏览器不会在等待锁时空闲断开
+                # 使用 keepalive 获取锁，防止等待期间浏览器连接断开
+                lock_acquired = False
                 if auth_lock:
                     log.step("等待授权回调锁...")
-                    auth_lock.acquire()
+                    lock_acquired = acquire_lock_with_keepalive(auth_lock, ctx.page, timeout=180, check_interval=2.0)
+                    if lock_acquired:
                         log.step("获取授权回调锁，开始授权...")
+                    else:
+                        log.error("获取授权回调锁超时或浏览器连接断开")
+                        continue  # 重试
 
                 try:
                     # 根据配置选择授权方式
@@ -2175,7 +2229,7 @@ def register_and_authorize(email: str, password: str, use_api_register: bool = T
                         codex_data = perform_codex_authorization(ctx.page, email, password)
                         return True, codex_data
                 finally:
-                    if auth_lock:
+                    if auth_lock and lock_acquired:
                         log.step("释放授权回调锁")
                         auth_lock.release()
 
@@ -2212,10 +2266,16 @@ def authorize_only(email: str, password: str) -> tuple[bool, dict]:
         for attempt in ctx.attempts():
             try:
                 # ========== 授权流程 - CPA/S2A 需要串行执行 (避免回调端口冲突) ==========
+                # 使用 keepalive 获取锁，防止等待期间浏览器连接断开
+                lock_acquired = False
                 if auth_lock:
                     log.step("等待授权回调锁...")
-                    auth_lock.acquire()
+                    lock_acquired = acquire_lock_with_keepalive(auth_lock, ctx.page, timeout=180, check_interval=2.0)
+                    if lock_acquired:
                         log.step("获取授权回调锁，开始授权...")
+                    else:
+                        log.error("获取授权回调锁超时或浏览器连接断开")
+                        continue  # 重试
 
                 try:
                     # 根据配置选择授权方式
@@ -2252,7 +2312,7 @@ def authorize_only(email: str, password: str) -> tuple[bool, dict]:
                                 continue
                             return False, None
                 finally:
-                    if auth_lock:
+                    if auth_lock and lock_acquired:
                         log.step("释放授权回调锁")
                         auth_lock.release()