338 lines
12 KiB
Python
338 lines
12 KiB
Python
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() |