def get_cursor_session_token(tab, max_attempts: int = 3, retry_interval: int = 2) -> Optional[Tuple[str, str]]: """ 获取Cursor会话token Args: tab: 浏览器标签页对象 max_attempts: 最大尝试次数 retry_interval: 重试间隔(秒) Returns: Tuple[str, str] | None: 成功返回(userId, accessToken)元组,失败返回None """ logging.info("开始获取会话令牌") # 首先尝试使用UUID深度登录方式 logging.info("尝试使用深度登录方式获取token") def _generate_pkce_pair(): """生成PKCE验证对""" code_verifier = secrets.token_urlsafe(43) # 生成随机的code_verifier code_challenge_digest = hashlib.sha256(code_verifier.encode('utf-8')).digest() # 对verifier进行SHA256哈希 code_challenge = base64.urlsafe_b64encode(code_challenge_digest).decode('utf-8').rstrip('=') # Base64编码并移除填充字符 return code_verifier, code_challenge attempts = 0 while attempts < max_attempts: try: verifier, challenge = _generate_pkce_pair() # 生成PKCE验证对 id = uuid.uuid4() # 生成唯一的UUID client_login_url = f"https://www.cursor.com/cn/loginDeepControl?challenge={challenge}&uuid={id}&mode=login" # 构建深度登录URL logging.info(f"访问深度登录URL: {client_login_url}") tab.get(client_login_url) # 访问登录页面 save_screenshot(tab, f"deeplogin_attempt_{attempts}") # 保存登录页面截图 if tab.ele("xpath=//span[contains(text(), 'Yes, Log In')]", timeout=5): # 查找登录确认按钮 logging.info("点击确认登录按钮") tab.ele("xpath=//span[contains(text(), 'Yes, Log In')]").click() # 点击确认登录 time.sleep(1.5) # 等待页面响应 auth_poll_url = f"https://api2.cursor.sh/auth/poll?uuid={id}&verifier={verifier}" # 构建认证轮询URL headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Cursor/0.48.6 Chrome/132.0.6834.210 Electron/34.3.4 Safari/537.36", "Accept": "*/*" } # 设置请求头 logging.info(f"轮询认证状态: {auth_poll_url}") response = requests.get(auth_poll_url, headers=headers, timeout=5) # 发送认证状态轮询请求 if response.status_code == 200: # 请求成功 data = response.json() # 解析JSON响应 accessToken = data.get("accessToken", None) # 获取访问令牌 authId = data.get("authId", "") # 获取认证ID if accessToken: # 如果成功获取到访问令牌 userId = "" if len(authId.split("|")) > 1: # 从authId中提取userId userId = authId.split("|")[1] logging.info("成功获取账号token和userId") return userId, accessToken # 返回用户ID和访问令牌 else: logging.error(f"API请求失败,状态码: {response.status_code}") # 记录API请求失败 else: logging.warning("未找到登录确认按钮") # 记录未找到登录按钮 attempts += 1 # 增加尝试次数 if attempts < max_attempts: # 如果还有重试机会 wait_time = retry_interval * attempts # 计算等待时间,逐步增加 logging.warning(f"第 {attempts} 次尝试未获取到token,{wait_time}秒后重试...") save_screenshot(tab, f"token_attempt_{attempts}") # 保存失败截图 time.sleep(wait_time) # 等待指定时间后重试 except Exception as e: # 捕获所有异常 logging.error(f"深度登录获取token失败: {str(e)}") # 记录异常信息 attempts += 1 # 增加尝试次数 save_screenshot(tab, f"token_error_{attempts}") # 保存错误截图 if attempts < max_attempts: # 如果还有重试机会 wait_time = retry_interval * attempts # 计算等待时间 logging.warning(f"将在 {wait_time} 秒后重试...") time.sleep(wait_time) # 等待指定时间后重试