import os import json import platform import requests import urllib3 import ssl import sys import subprocess import hashlib from cursor_auth_manager import CursorAuthManager from logger import logging from reset_machine import MachineIDResetter import patch_cursor_get_machine_id from exit_cursor import ExitCursor import go_cursor_help from logo import print_logo from typing import Tuple, Dict, Optional class CursorTokenUpdater: def __init__(self): self.auth_manager = CursorAuthManager() self._hardware_id = None # 延迟初始化硬件ID @property def hardware_id(self) -> str: """获取硬件ID（延迟初始化）""" if self._hardware_id is None: self._hardware_id = self._get_hardware_id() return self._hardware_id def _get_hardware_id(self) -> str: """获取硬件唯一标识 方案1: CPU ID + 主板序列号 + BIOS序列号 方案2: 系统盘序列号 + Windows安装时间 方案3: 计算机名(最后的备选方案) """ try: # 创建startupinfo对象来隐藏命令行窗口 startupinfo = None if sys.platform == "win32": startupinfo = subprocess.STARTUPINFO() startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW startupinfo.wShowWindow = subprocess.SW_HIDE # 方案1: 尝试获取硬件信息 try: # 获取CPU ID cpu_info = subprocess.check_output('wmic cpu get ProcessorId', startupinfo=startupinfo).decode() cpu_id = cpu_info.split('\n')[1].strip() # 获取主板序列号 board_info = subprocess.check_output('wmic baseboard get SerialNumber', startupinfo=startupinfo).decode() board_id = board_info.split('\n')[1].strip() # 获取BIOS序列号 bios_info = subprocess.check_output('wmic bios get SerialNumber', startupinfo=startupinfo).decode() bios_id = bios_info.split('\n')[1].strip() # 如果所有信息都获取成功且有效 if all([cpu_id, board_id, bios_id]) and not all(x in ['', '0', 'None', 'To be filled by O.E.M.'] for x in [cpu_id, board_id, bios_id]): combined = f"{cpu_id}:{board_id}:{bios_id}" hardware_id = hashlib.md5(combined.encode()).hexdigest() logging.info("使用硬件信息生成ID成功") return hardware_id except Exception as e: logging.warning(f"方案1失败: {str(e)}") # 方案2: 系统盘序列号 + Windows安装时间 try: backup_info = [] # 获取系统盘序列号 volume_info = subprocess.check_output('wmic logicaldisk where "DeviceID=\'C:\'" get VolumeSerialNumber', startupinfo=startupinfo).decode() volume_serial = volume_info.split('\n')[1].strip() if volume_serial and volume_serial not in ['', '0']: backup_info.append(("volume", volume_serial)) # 获取Windows安装时间 os_info = subprocess.check_output('wmic os get InstallDate', startupinfo=startupinfo).decode() install_date = os_info.split('\n')[1].strip() if install_date: backup_info.append(("install", install_date)) if backup_info: combined = "|".join(f"{k}:{v}" for k, v in sorted(backup_info)) hardware_id = hashlib.md5(combined.encode()).hexdigest() logging.info("使用系统信息生成ID成功") return hardware_id except Exception as e: logging.warning(f"方案2失败: {str(e)}") # 方案3: 使用计算机名(最后的备选方案) computer_name = platform.node() if computer_name: hardware_id = hashlib.md5(computer_name.encode()).hexdigest() logging.info("使用计算机名生成ID成功") return hardware_id raise ValueError("无法获取任何可用信息来生成硬件ID") except Exception as e: error_msg = f"生成硬件ID失败: {str(e)}" logging.error(error_msg) raise RuntimeError(error_msg) def get_unused_account(self) -> Tuple[bool, str, Optional[Dict]]: """ 从API获取未使用的账号 Returns: Tuple[bool, str, Optional[Dict]]: - 是否成功 - 错误信息 - 账号数据（如果成功） """ endpoint = "https://cursorapi.nosqli.com/admin/api.account/getUnused" data = { "machine_id": self.hardware_id } headers = { "Content-Type": "application/json" } # 禁用SSL警告 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) request_kwargs = { "json": data, "headers": headers, "timeout": 30, "verify": False } try: try: response = requests.post(endpoint, **request_kwargs) except requests.exceptions.SSLError: # SSL错误时使用自定义SSL上下文 ssl_context = ssl.create_default_context() ssl_context.check_hostname = False ssl_context.verify_mode = ssl.CERT_NONE session = requests.Session() session.verify = False response = session.post(endpoint, **request_kwargs) response_data = response.json() if response_data.get("code") == 200: account_data = response_data.get("data", {}) # 获取账号信息 email = account_data.get("email", "") access_token = account_data.get("access_token", "") refresh_token = account_data.get("refresh_token", "") expire_time = account_data.get("expire_time", "") days_left = account_data.get("days_left", 0) if not all([email, access_token, refresh_token]): return False, "获取账号信息不完整", None account_info = { "email": email, "access_token": access_token, "refresh_token": refresh_token, "expire_time": expire_time, "days_left": days_left } logging.info(f"成功获取账号信息 - 邮箱: {email}, 剩余天数: {days_left}") return True, "", account_info else: error_msg = response_data.get("msg", "未知错误") return False, f"API返回错误: {error_msg}", None except Exception as e: error_msg = f"获取账号时发生错误: {str(e)}" logging.error(error_msg) return False, error_msg, None def update_auth_info(self, email: str, access_token: str, refresh_token: str = None) -> bool: """ 更新Cursor的认证信息 Args: email: 用户邮箱 access_token: 访问令牌 refresh_token: 刷新令牌（如果没有提供，将使用access_token） Returns: bool: 更新是否成功 """ try: # 如果没有提供refresh_token，使用access_token if refresh_token is None: refresh_token = access_token # 更新认证信息 result = self.auth_manager.update_auth( email=email, access_token=access_token, refresh_token=refresh_token ) if result: logging.info(f"认证信息更新成功 - 邮箱: {email}") return True else: logging.error("认证信息更新失败") return False except Exception as e: logging.error(f"更新认证信息时发生错误: {str(e)}") return False def reset_machine_id(self, greater_than_0_45: bool = True) -> bool: """ 重置机器码 Args: greater_than_0_45: 是否大于0.45版本 Returns: bool: 重置是否成功 """ try: logging.info("开始重置机器码...") resetter = MachineIDResetter() result = resetter.reset(greater_than_0_45) if result: logging.info("机器码重置成功") # 重置后更新硬件ID缓存 self._hardware_id = None return True else: logging.error("机器码重置失败") return False except Exception as e: logging.error(f"重置机器码时发生错误: {str(e)}") return False def patch_machine_id(self) -> bool: """ 修补机器码获取方法 Returns: bool: 修补是否成功 """ try: logging.info("开始修补机器码获取方法...") patch_cursor_get_machine_id.patch() logging.info("机器码获取方法修补完成") return True except Exception as e: logging.error(f"修补机器码获取方法时发生错误: {str(e)}") return False def exit_cursor(self) -> bool: """ 退出Cursor进程 Returns: bool: 退出是否成功 """ try: logging.info("正在退出Cursor进程...") exit_handler = ExitCursor() exit_handler.exit() logging.info("Cursor进程已退出") return True except Exception as e: logging.error(f"退出Cursor进程时发生错误: {str(e)}") return False def full_update_process(self, email: str = None, access_token: str = None, refresh_token: str = None) -> bool: """ 执行完整的更新流程 Args: email: 用户邮箱（可选，如果不提供则从API获取） access_token: 访问令牌（可选，如果不提供则从API获取） refresh_token: 刷新令牌（可选，如果不提供则从API获取） Returns: bool: 更新流程是否全部成功 """ print_logo() logging.info("=== 开始完整更新流程 ===") # 1. 退出Cursor进程 if not self.exit_cursor(): return False # 2. 修补机器码获取方法 if not self.patch_machine_id(): return False # 3. 重置机器码 if not self.reset_machine_id(greater_than_0_45=True): return False # 4. 如果没有提供认证信息，从API获取 if not all([email, access_token]): success, error_msg, account_info = self.get_unused_account() if not success: logging.error(f"无法获取账号信息: {error_msg}") return False email = account_info["email"] access_token = account_info["access_token"] refresh_token = account_info["refresh_token"] # 5. 更新认证信息 if not self.update_auth_info(email, access_token, refresh_token): return False logging.info("=== 所有操作已完成 ===") return True def main(): updater = CursorTokenUpdater() # 从环境变量获取认证信息（可选） # 如果环境变量中有认证信息，使用环境变量中的信息 # 否则，将从API获取新的账号信息 success = updater.full_update_process( ) print("更新状态:", "成功" if success else "失败") if __name__ == "__main__": main()