8b2fbef54a
1. 新增 CursorResetter 类，完整封装 cursor_win_id_modifier.ps1 的核心功能 2. 优化 AccountSwitcher 的重启逻辑，避免重复重启 3. 改进进程管理，移除 wmi 依赖，使用 tasklist 替代 4. 提升代码可维护性，后续只需更新 CursorResetter 即可适配脚本变更
974 lines
40 KiB
Python
974 lines
40 KiB
Python
import os
|
||
import json
|
||
import requests
|
||
import logging
|
||
import subprocess
|
||
import uuid
|
||
import hashlib
|
||
import sys
|
||
import time
|
||
import ctypes
|
||
from typing import Optional, Dict, Tuple, List
|
||
from pathlib import Path
|
||
from utils.config import Config
|
||
from utils.cursor_registry import CursorRegistry
|
||
from cursor_auth_manager import CursorAuthManager
|
||
from utils.cursor_resetter import CursorResetter # 添加导入
|
||
|
||
def is_admin() -> bool:
|
||
"""检查是否具有管理员权限
|
||
|
||
Returns:
|
||
bool: 是否具有管理员权限
|
||
"""
|
||
try:
|
||
return ctypes.windll.shell32.IsUserAnAdmin() != 0
|
||
except:
|
||
return False
|
||
|
||
def run_as_admin():
|
||
"""以管理员权限重新运行程序"""
|
||
try:
|
||
if not is_admin():
|
||
# 获取当前脚本的路径
|
||
script = sys.argv[0]
|
||
params = ' '.join(sys.argv[1:])
|
||
|
||
# 创建 startupinfo 对象来隐藏命令行窗口
|
||
startupinfo = None
|
||
if sys.platform == "win32":
|
||
startupinfo = subprocess.STARTUPINFO()
|
||
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
|
||
startupinfo.wShowWindow = subprocess.SW_HIDE
|
||
|
||
# 以管理员权限重新运行
|
||
ctypes.windll.shell32.ShellExecuteW(
|
||
None,
|
||
"runas",
|
||
sys.executable,
|
||
f'"{script}" {params}',
|
||
None,
|
||
1
|
||
)
|
||
return True
|
||
except Exception as e:
|
||
logging.error(f"以管理员权限运行失败: {str(e)}")
|
||
return False
|
||
|
||
def get_hardware_id() -> str:
|
||
"""获取硬件唯一标识"""
|
||
try:
|
||
# 创建startupinfo对象来隐藏命令行窗口
|
||
startupinfo = None
|
||
if sys.platform == "win32":
|
||
startupinfo = subprocess.STARTUPINFO()
|
||
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
|
||
startupinfo.wShowWindow = subprocess.SW_HIDE
|
||
|
||
# 获取CPU信息
|
||
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()
|
||
|
||
# 组合信息并生成哈希
|
||
combined = f"{cpu_id}:{board_id}:{bios_id}"
|
||
return hashlib.md5(combined.encode()).hexdigest()
|
||
except Exception as e:
|
||
logging.error(f"获取硬件ID失败: {str(e)}")
|
||
# 如果获取失败，使用UUID作为备选方案
|
||
return str(uuid.uuid4())
|
||
|
||
class AccountSwitcher:
|
||
def __init__(self):
|
||
# 检查管理员权限
|
||
if not is_admin():
|
||
logging.warning("当前不是管理员权限运行")
|
||
if run_as_admin():
|
||
sys.exit(0)
|
||
else:
|
||
logging.error("无法获取管理员权限")
|
||
raise PermissionError("需要管理员权限才能运行此程序")
|
||
|
||
self.cursor_path = Path(os.path.expanduser("~")) / "AppData" / "Local" / "Programs" / "Cursor"
|
||
self.app_path = self.cursor_path / "resources" / "app"
|
||
self.package_json = self.app_path / "package.json"
|
||
self.auth_manager = CursorAuthManager()
|
||
self.config = Config()
|
||
self.hardware_id = self.get_hardware_id() # 先获取硬件ID
|
||
self.registry = CursorRegistry() # 添加注册表操作工具类
|
||
self.resetter = CursorResetter() # 添加重置工具类
|
||
self.max_retries = 5
|
||
self.wait_time = 1
|
||
|
||
logging.info(f"初始化硬件ID: {self.hardware_id}")
|
||
|
||
def get_hardware_id(self) -> str:
|
||
"""获取硬件唯一标识"""
|
||
try:
|
||
# 创建startupinfo对象来隐藏命令行窗口
|
||
startupinfo = None
|
||
if sys.platform == "win32":
|
||
startupinfo = subprocess.STARTUPINFO()
|
||
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
|
||
startupinfo.wShowWindow = subprocess.SW_HIDE
|
||
|
||
# 获取CPU信息
|
||
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()
|
||
|
||
# 组合信息并生成哈希
|
||
combined = f"{cpu_id}:{board_id}:{bios_id}"
|
||
return hashlib.md5(combined.encode()).hexdigest()
|
||
except Exception as e:
|
||
logging.error(f"获取硬件ID失败: {str(e)}")
|
||
# 如果获取失败，使用UUID作为备选方案
|
||
return str(uuid.uuid4())
|
||
|
||
def get_cursor_version(self) -> str:
|
||
"""获取Cursor版本号"""
|
||
try:
|
||
if self.package_json.exists():
|
||
with open(self.package_json, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
return data.get("version", "未知")
|
||
return "未知"
|
||
except Exception as e:
|
||
logging.error(f"获取Cursor版本失败: {str(e)}")
|
||
return "未知"
|
||
|
||
def get_device_info(self) -> dict:
|
||
"""获取设备信息"""
|
||
try:
|
||
import platform
|
||
import socket
|
||
import requests
|
||
import subprocess
|
||
|
||
# 获取操作系统信息
|
||
os_info = f"{platform.system()} {platform.release()}"
|
||
|
||
# 获取设备名称
|
||
try:
|
||
# 在Windows上使用wmic获取更详细的计算机名称
|
||
if platform.system() == "Windows":
|
||
startupinfo = subprocess.STARTUPINFO()
|
||
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
|
||
computer_info = subprocess.check_output('wmic computersystem get model', startupinfo=startupinfo).decode()
|
||
device_name = computer_info.split('\n')[1].strip()
|
||
else:
|
||
device_name = platform.node()
|
||
except:
|
||
device_name = platform.node()
|
||
|
||
# 获取IP地址
|
||
try:
|
||
ip_response = requests.get('https://api.ipify.org?format=json', timeout=5)
|
||
ip_address = ip_response.json()['ip']
|
||
except:
|
||
ip_address = "未知"
|
||
|
||
# 获取地理位置
|
||
try:
|
||
ip_info = requests.get(f'https://ipapi.co/{ip_address}/json/', timeout=5).json()
|
||
location = f"{ip_info.get('country_name', '')}-{ip_info.get('region', '')}-{ip_info.get('city', '')}"
|
||
except:
|
||
location = "未知"
|
||
|
||
return {
|
||
"os": os_info,
|
||
"device_name": device_name,
|
||
"ip": ip_address,
|
||
"location": location
|
||
}
|
||
except Exception as e:
|
||
logging.error(f"获取设备信息失败: {str(e)}")
|
||
return {
|
||
"os": "Windows",
|
||
"device_name": "未知设备",
|
||
"ip": "未知",
|
||
"location": "未知"
|
||
}
|
||
|
||
def check_activation_code(self, code: str) -> tuple:
|
||
"""检查激活码
|
||
|
||
Args:
|
||
code: 激活码
|
||
|
||
Returns:
|
||
tuple: (成功标志, 消息, 账号信息)
|
||
"""
|
||
max_retries = 3 # 最大重试次数
|
||
retry_delay = 1 # 重试间隔(秒)
|
||
|
||
for attempt in range(max_retries):
|
||
try:
|
||
data = {
|
||
"machine_id": self.hardware_id,
|
||
"code": code
|
||
}
|
||
|
||
# 禁用SSL警告
|
||
import urllib3
|
||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||
|
||
# 设置请求参数
|
||
request_kwargs = {
|
||
"json": data,
|
||
"headers": {
|
||
"Content-Type": "application/json",
|
||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36",
|
||
"Accept": "*/*",
|
||
"Connection": "keep-alive"
|
||
},
|
||
"timeout": 10, # 增加超时时间
|
||
"verify": False # 禁用SSL验证
|
||
}
|
||
|
||
# 创建session
|
||
session = requests.Session()
|
||
session.verify = False
|
||
|
||
# 设置重试策略
|
||
retry_strategy = urllib3.Retry(
|
||
total=3, # 总重试次数
|
||
backoff_factor=0.5, # 重试间隔
|
||
status_forcelist=[500, 502, 503, 504] # 需要重试的HTTP状态码
|
||
)
|
||
adapter = requests.adapters.HTTPAdapter(max_retries=retry_strategy)
|
||
session.mount("http://", adapter)
|
||
session.mount("https://", adapter)
|
||
|
||
try:
|
||
# 尝试发送请求
|
||
response = session.post(
|
||
self.config.get_api_url("activate"),
|
||
**request_kwargs
|
||
)
|
||
response.raise_for_status() # 检查HTTP状态码
|
||
|
||
result = response.json()
|
||
# 激活成功
|
||
if result["code"] == 200:
|
||
api_data = result["data"]
|
||
# 构造标准的返回数据结构
|
||
account_info = {
|
||
"status": "active",
|
||
"expire_time": api_data.get("expire_time", ""),
|
||
"total_days": api_data.get("total_days", 0),
|
||
"days_left": api_data.get("days_left", 0),
|
||
"device_info": self.get_device_info()
|
||
}
|
||
return True, result["msg"], account_info
|
||
# 激活码无效或已被使用
|
||
elif result["code"] == 400:
|
||
logging.warning(f"激活码无效或已被使用: {result.get('msg', '未知错误')}")
|
||
return False, result.get("msg", "激活码无效或已被使用"), None
|
||
# 其他错误情况
|
||
else:
|
||
error_msg = result.get("msg", "未知错误")
|
||
if attempt < max_retries - 1: # 如果还有重试机会
|
||
logging.warning(f"第{attempt + 1}次尝试失败: {error_msg}, 准备重试...")
|
||
time.sleep(retry_delay)
|
||
continue
|
||
logging.error(f"激活失败: {error_msg}")
|
||
return False, error_msg, None
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
if attempt < max_retries - 1: # 如果还有重试机会
|
||
logging.warning(f"第{attempt + 1}次网络请求失败: {str(e)}, 准备重试...")
|
||
time.sleep(retry_delay)
|
||
continue
|
||
error_msg = self._get_network_error_message(e)
|
||
logging.error(f"网络请求失败: {error_msg}")
|
||
return False, f"网络连接失败: {error_msg}", None
|
||
|
||
except Exception as e:
|
||
if attempt < max_retries - 1: # 如果还有重试机会
|
||
logging.warning(f"第{attempt + 1}次请求发生错误: {str(e)}, 准备重试...")
|
||
time.sleep(retry_delay)
|
||
continue
|
||
logging.error(f"激活失败: {str(e)}")
|
||
return False, f"激活失败: {str(e)}", None
|
||
|
||
# 如果所有重试都失败了
|
||
return False, "多次尝试后激活失败，请检查网络连接或稍后重试", None
|
||
|
||
def _get_network_error_message(self, error: Exception) -> str:
|
||
"""获取网络错误的友好提示信息"""
|
||
if isinstance(error, requests.exceptions.SSLError):
|
||
return "SSL证书验证失败，请检查系统时间是否正确"
|
||
elif isinstance(error, requests.exceptions.ConnectionError):
|
||
if "10054" in str(error):
|
||
return "连接被重置，可能是防火墙拦截，请检查防火墙设置"
|
||
elif "10061" in str(error):
|
||
return "无法连接到服务器，请检查网络连接"
|
||
return "网络连接错误，请检查网络设置"
|
||
elif isinstance(error, requests.exceptions.Timeout):
|
||
return "请求超时，请检查网络连接"
|
||
elif isinstance(error, requests.exceptions.RequestException):
|
||
return "网络请求失败，请稍后重试"
|
||
return str(error)
|
||
|
||
def get_process_details(self, process_name: str) -> List[Dict]:
|
||
"""获取进程详细信息
|
||
|
||
Args:
|
||
process_name: 进程名称
|
||
|
||
Returns:
|
||
List[Dict]: 进程详细信息列表
|
||
"""
|
||
try:
|
||
# 使用 tasklist 命令替代 wmi
|
||
startupinfo = subprocess.STARTUPINFO()
|
||
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
|
||
|
||
output = subprocess.check_output(
|
||
f'tasklist /FI "IMAGENAME eq {process_name}" /FO CSV /NH',
|
||
startupinfo=startupinfo,
|
||
shell=True
|
||
).decode('gbk')
|
||
|
||
processes = []
|
||
if output.strip():
|
||
for line in output.strip().split('\n'):
|
||
if line.strip():
|
||
parts = line.strip('"').split('","')
|
||
if len(parts) >= 2:
|
||
processes.append({
|
||
'name': parts[0],
|
||
'pid': parts[1]
|
||
})
|
||
return processes
|
||
except Exception as e:
|
||
logging.error(f"获取进程信息失败: {str(e)}")
|
||
return []
|
||
|
||
def close_cursor_process(self) -> bool:
|
||
"""关闭所有Cursor进程
|
||
|
||
Returns:
|
||
bool: 是否成功关闭所有进程
|
||
"""
|
||
try:
|
||
if sys.platform == "win32":
|
||
# 创建startupinfo对象来隐藏命令行窗口
|
||
startupinfo = subprocess.STARTUPINFO()
|
||
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
|
||
startupinfo.wShowWindow = subprocess.SW_HIDE
|
||
|
||
# 获取进程详情
|
||
processes = self.get_process_details("Cursor.exe")
|
||
if processes:
|
||
logging.info(f"发现 {len(processes)} 个Cursor进程")
|
||
for p in processes:
|
||
logging.info(f"进程信息: PID={p['pid']}, 路径={p['name']}")
|
||
|
||
# 尝试关闭进程
|
||
subprocess.run(
|
||
"taskkill /f /im Cursor.exe >nul 2>&1",
|
||
startupinfo=startupinfo,
|
||
shell=True
|
||
)
|
||
|
||
# 等待进程关闭
|
||
retry_count = 0
|
||
while retry_count < self.max_retries:
|
||
if not self.get_process_details("Cursor.exe"):
|
||
logging.info("所有Cursor进程已关闭")
|
||
return True
|
||
|
||
retry_count += 1
|
||
if retry_count >= self.max_retries:
|
||
processes = self.get_process_details("Cursor.exe")
|
||
if processes:
|
||
logging.error(f"无法关闭以下进程:")
|
||
for p in processes:
|
||
logging.error(f"PID={p['pid']}, 路径={p['name']}")
|
||
return False
|
||
|
||
logging.warning(f"等待进程关闭, 尝试 {retry_count}/{self.max_retries}...")
|
||
time.sleep(self.wait_time)
|
||
|
||
return True
|
||
else:
|
||
# 其他系统的处理
|
||
if sys.platform == "darwin":
|
||
subprocess.run("killall Cursor 2>/dev/null", shell=True)
|
||
else:
|
||
subprocess.run("pkill -f cursor", shell=True)
|
||
time.sleep(2)
|
||
return True
|
||
|
||
except Exception as e:
|
||
logging.error(f"关闭进程失败: {str(e)}")
|
||
return False
|
||
|
||
def reset_machine_id(self) -> bool:
|
||
"""重置机器码"""
|
||
try:
|
||
# 1. 关闭所有Cursor进程
|
||
if not self.close_cursor_process():
|
||
logging.error("无法关闭所有Cursor进程")
|
||
return False
|
||
|
||
# 2. 使用新的重置工具类执行重置
|
||
success, message = self.resetter.reset_cursor(disable_update=True)
|
||
if not success:
|
||
logging.error(f"重置失败: {message}")
|
||
return False
|
||
|
||
# 不在这里重启Cursor，让调用者决定何时重启
|
||
logging.info("机器码重置完成")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logging.error(f"重置机器码失败: {str(e)}")
|
||
return False
|
||
|
||
def restart_cursor(self) -> bool:
|
||
"""重启Cursor编辑器
|
||
|
||
Returns:
|
||
bool: 是否成功重启
|
||
"""
|
||
try:
|
||
logging.info("正在重启Cursor...")
|
||
|
||
# 确保进程已关闭
|
||
if not self.close_cursor_process():
|
||
logging.error("无法关闭Cursor进程")
|
||
return False
|
||
|
||
# 等待进程完全关闭
|
||
time.sleep(2)
|
||
|
||
# 启动Cursor
|
||
if sys.platform == "win32":
|
||
cursor_exe = self.cursor_path / "Cursor.exe"
|
||
if cursor_exe.exists():
|
||
try:
|
||
# 使用subprocess启动
|
||
startupinfo = subprocess.STARTUPINFO()
|
||
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
|
||
|
||
subprocess.Popen(
|
||
str(cursor_exe),
|
||
startupinfo=startupinfo,
|
||
creationflags=subprocess.CREATE_NEW_CONSOLE
|
||
)
|
||
|
||
# 等待进程启动
|
||
time.sleep(3)
|
||
|
||
# 验证进程是否启动
|
||
processes = self.get_process_details("Cursor.exe")
|
||
if processes:
|
||
logging.info("Cursor启动成功")
|
||
return True
|
||
else:
|
||
logging.error("Cursor进程未找到")
|
||
# 尝试使用 os.startfile 作为备选方案
|
||
try:
|
||
os.startfile(str(cursor_exe))
|
||
time.sleep(3)
|
||
logging.info("使用备选方案启动Cursor")
|
||
return True
|
||
except Exception as e:
|
||
logging.error(f"备选启动方案失败: {str(e)}")
|
||
return False
|
||
|
||
except Exception as e:
|
||
logging.error(f"启动Cursor失败: {str(e)}")
|
||
# 尝试使用 os.startfile 作为备选方案
|
||
try:
|
||
os.startfile(str(cursor_exe))
|
||
time.sleep(3)
|
||
logging.info("使用备选方案启动Cursor")
|
||
return True
|
||
except Exception as e:
|
||
logging.error(f"备选启动方案失败: {str(e)}")
|
||
return False
|
||
else:
|
||
logging.error(f"未找到Cursor程序: {cursor_exe}")
|
||
return False
|
||
elif sys.platform == "darwin":
|
||
try:
|
||
subprocess.run("open -a Cursor", shell=True, check=True)
|
||
logging.info("Cursor启动成功")
|
||
return True
|
||
except subprocess.CalledProcessError as e:
|
||
logging.error(f"启动Cursor失败: {str(e)}")
|
||
return False
|
||
elif sys.platform == "linux":
|
||
try:
|
||
subprocess.run("cursor &", shell=True, check=True)
|
||
logging.info("Cursor启动成功")
|
||
return True
|
||
except subprocess.CalledProcessError as e:
|
||
logging.error(f"启动Cursor失败: {str(e)}")
|
||
return False
|
||
|
||
return False
|
||
|
||
except Exception as e:
|
||
logging.error(f"重启Cursor失败: {str(e)}")
|
||
# 尝试使用 os.startfile 作为最后的备选方案
|
||
try:
|
||
cursor_exe = self.cursor_path / "Cursor.exe"
|
||
if cursor_exe.exists():
|
||
os.startfile(str(cursor_exe))
|
||
time.sleep(3)
|
||
logging.info("使用最终备选方案启动Cursor")
|
||
return True
|
||
except:
|
||
pass
|
||
return False
|
||
|
||
def activate_and_switch(self, activation_code: str) -> Tuple[bool, str]:
|
||
"""激活并切换账号
|
||
|
||
Returns:
|
||
Tuple[bool, str]: (是否成功, 提示消息)
|
||
"""
|
||
try:
|
||
# 验证激活码
|
||
success, message, account_info = self.check_activation_code(activation_code)
|
||
return success, message
|
||
|
||
except Exception as e:
|
||
logging.error(f"激活过程出错: {str(e)}")
|
||
return False, f"激活失败: {str(e)}"
|
||
|
||
def get_member_status(self) -> dict:
|
||
"""获取会员状态
|
||
|
||
Returns:
|
||
dict: 会员状态信息，包含：
|
||
- status: 状态(active/inactive/expired)
|
||
- expire_time: 到期时间
|
||
- total_days: 总天数
|
||
- days_left: 剩余天数
|
||
- device_info: 设备信息
|
||
"""
|
||
try:
|
||
data = {
|
||
"machine_id": self.hardware_id
|
||
}
|
||
|
||
api_url = self.config.get_api_url("status")
|
||
logging.info(f"正在检查会员状态...")
|
||
|
||
# 设置请求参数
|
||
request_kwargs = {
|
||
"json": data,
|
||
"headers": {"Content-Type": "application/json"},
|
||
"timeout": 2, # 减少超时时间到2秒
|
||
"verify": False
|
||
}
|
||
|
||
# 使用session来复用连接
|
||
session = requests.Session()
|
||
session.verify = False
|
||
|
||
# 尝试发送请求
|
||
try:
|
||
response = session.post(api_url, **request_kwargs)
|
||
except requests.exceptions.Timeout:
|
||
# 超时后快速重试一次
|
||
logging.warning("首次请求超时，正在重试...")
|
||
response = session.post(api_url, **request_kwargs)
|
||
|
||
result = response.json()
|
||
logging.info(f"状态检查响应: {result}")
|
||
|
||
if result.get("code") in [1, 200]:
|
||
api_data = result.get("data", {})
|
||
return {
|
||
"status": api_data.get("status", "inactive"),
|
||
"expire_time": api_data.get("expire_time", ""),
|
||
"total_days": api_data.get("total_days", 0),
|
||
"days_left": api_data.get("days_left", 0),
|
||
"device_info": self.get_device_info()
|
||
}
|
||
else:
|
||
error_msg = result.get("msg", "未知错误")
|
||
logging.error(f"获取状态失败: {error_msg}")
|
||
return {
|
||
"status": "inactive",
|
||
"expire_time": "",
|
||
"total_days": 0,
|
||
"days_left": 0,
|
||
"device_info": self.get_device_info()
|
||
}
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
logging.error(f"API请求失败: {str(e)}")
|
||
return {
|
||
"status": "inactive",
|
||
"expire_time": "",
|
||
"total_days": 0,
|
||
"days_left": 0,
|
||
"device_info": self.get_device_info()
|
||
}
|
||
except Exception as e:
|
||
logging.error(f"获取会员状态失败: {str(e)}")
|
||
return {
|
||
"status": "inactive",
|
||
"expire_time": "",
|
||
"total_days": 0,
|
||
"days_left": 0,
|
||
"device_info": self.get_device_info()
|
||
}
|
||
|
||
def refresh_cursor_auth(self) -> Tuple[bool, str]:
|
||
"""刷新Cursor授权
|
||
|
||
Returns:
|
||
Tuple[bool, str]: (是否成功, 提示消息)
|
||
"""
|
||
try:
|
||
# 1. 获取未使用的账号
|
||
endpoint = "https://cursorapi.nosqli.com/admin/api.account/getUnused"
|
||
data = {
|
||
"machine_id": self.hardware_id
|
||
}
|
||
headers = {
|
||
"Content-Type": "application/json"
|
||
}
|
||
|
||
# 禁用SSL警告
|
||
import urllib3
|
||
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上下文
|
||
import ssl
|
||
ssl_context = ssl.create_default_context()
|
||
ssl_context.check_hostname = False
|
||
ssl_context.verify_mode = ssl.CERT_NONE
|
||
|
||
# 使用自定义SSL上下文重试请求
|
||
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, "获取账号信息不完整"
|
||
|
||
# 2. 先关闭Cursor进程
|
||
logging.info("正在关闭Cursor进程...")
|
||
if not self.close_cursor_process():
|
||
logging.error("无法关闭Cursor进程")
|
||
return False, "无法关闭Cursor进程，请手动关闭后重试"
|
||
|
||
# 3. 更新Cursor认证信息
|
||
logging.info("正在更新认证信息...")
|
||
if not self.auth_manager.update_auth(email, access_token, refresh_token):
|
||
return False, "更新Cursor认证信息失败"
|
||
|
||
# 4. 验证认证信息是否正确写入
|
||
logging.info("正在验证认证信息...")
|
||
if not self.auth_manager.verify_auth(email, access_token, refresh_token):
|
||
return False, "认证信息验证失败"
|
||
|
||
# 5. 保存email到package.json
|
||
try:
|
||
if self.package_json.exists():
|
||
with open(self.package_json, "r", encoding="utf-8") as f:
|
||
package_data = json.load(f)
|
||
package_data["email"] = email
|
||
with open(self.package_json, "w", encoding="utf-8", newline='\n') as f:
|
||
json.dump(package_data, f, indent=2)
|
||
logging.info(f"已保存email到package.json: {email}")
|
||
except Exception as e:
|
||
logging.warning(f"保存email到package.json失败: {str(e)}")
|
||
|
||
# 6. 重置机器码
|
||
logging.info("正在重置机器码...")
|
||
if not self.reset_machine_id():
|
||
return False, "重置机器码失败"
|
||
|
||
# 7. 重启Cursor（只在这里执行一次重启）
|
||
logging.info("正在重启Cursor...")
|
||
retry_count = 0
|
||
max_retries = 3
|
||
while retry_count < max_retries:
|
||
if self.restart_cursor():
|
||
break
|
||
retry_count += 1
|
||
if retry_count < max_retries:
|
||
logging.warning(f"重启失败，正在重试 ({retry_count}/{max_retries})...")
|
||
time.sleep(2)
|
||
|
||
if retry_count >= max_retries:
|
||
return True, f"授权刷新成功，但Cursor重启失败，请手动启动Cursor\n邮箱: {email}\n到期时间: {expire_time}\n剩余天数: {days_left}"
|
||
|
||
return True, f"授权刷新成功，Cursor已重启\n邮箱: {email}\n到期时间: {expire_time}\n剩余天数: {days_left}"
|
||
|
||
elif response_data.get("code") == 404:
|
||
return False, "没有可用的未使用账号"
|
||
else:
|
||
error_msg = response_data.get("msg", "未知错误")
|
||
logging.error(f"获取未使用账号失败: {error_msg}")
|
||
return False, f"获取账号失败: {error_msg}"
|
||
|
||
except requests.exceptions.SSLError as e:
|
||
logging.error(f"SSL验证失败: {str(e)}")
|
||
return False, "SSL验证失败,请检查网络设置"
|
||
except requests.exceptions.ConnectionError as e:
|
||
logging.error(f"网络连接错误: {str(e)}")
|
||
return False, "网络连接失败,请检查网络设置"
|
||
except requests.exceptions.Timeout as e:
|
||
logging.error(f"请求超时: {str(e)}")
|
||
return False, "请求超时,请稍后重试"
|
||
except requests.RequestException as e:
|
||
logging.error(f"请求失败: {str(e)}")
|
||
return False, f"网络请求失败: {str(e)}"
|
||
except Exception as e:
|
||
logging.error(f"未知错误: {str(e)}")
|
||
return False, f"发生未知错误: {str(e)}"
|
||
|
||
except Exception as e:
|
||
logging.error(f"刷新授权过程出错: {str(e)}")
|
||
return False, f"刷新失败: {str(e)}"
|
||
|
||
def disable_cursor_update(self) -> Tuple[bool, str]:
|
||
"""禁用Cursor更新"""
|
||
try:
|
||
# 1. 先关闭所有Cursor进程
|
||
if sys.platform == "win32":
|
||
# 创建startupinfo对象来隐藏命令行窗口
|
||
startupinfo = subprocess.STARTUPINFO()
|
||
startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
|
||
startupinfo.wShowWindow = subprocess.SW_HIDE
|
||
|
||
# 关闭Cursor
|
||
subprocess.run(
|
||
"taskkill /f /im Cursor.exe >nul 2>&1",
|
||
startupinfo=startupinfo,
|
||
shell=True
|
||
)
|
||
time.sleep(2)
|
||
|
||
# 2. 删除updater目录并创建同名文件以阻止更新
|
||
updater_path = Path(os.getenv('LOCALAPPDATA')) / "cursor-updater"
|
||
|
||
try:
|
||
# 如果是目录，则删除
|
||
if updater_path.is_dir():
|
||
import shutil
|
||
shutil.rmtree(str(updater_path))
|
||
logging.info("删除updater目录成功")
|
||
|
||
# 如果是文件，则删除
|
||
if updater_path.is_file():
|
||
os.remove(str(updater_path))
|
||
logging.info("删除updater文件成功")
|
||
|
||
# 创建同名空文件
|
||
updater_path.touch()
|
||
logging.info("创建updater空文件成功")
|
||
|
||
# 3. 重启Cursor
|
||
cursor_exe = self.cursor_path / "Cursor.exe"
|
||
if cursor_exe.exists():
|
||
os.startfile(str(cursor_exe))
|
||
logging.info("Cursor重启成功")
|
||
|
||
return True, "Cursor更新已禁用，程序已重启"
|
||
|
||
except Exception as e:
|
||
logging.error(f"操作updater文件失败: {str(e)}")
|
||
return False, f"禁用更新失败: {str(e)}"
|
||
|
||
except Exception as e:
|
||
logging.error(f"禁用Cursor更新失败: {str(e)}")
|
||
return False, f"禁用更新失败: {str(e)}"
|
||
|
||
def send_heartbeat(self) -> Tuple[bool, str]:
|
||
"""
|
||
发送心跳请求
|
||
Returns:
|
||
Tuple[bool, str]: (是否成功, 消息)
|
||
"""
|
||
max_retries = 3 # 最大重试次数
|
||
retry_delay = 1 # 重试间隔(秒)
|
||
|
||
# 获取硬件ID
|
||
hardware_id = self.get_hardware_id()
|
||
|
||
# 禁用SSL警告
|
||
import urllib3
|
||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||
|
||
# 设置请求参数
|
||
params = {
|
||
"machine_id": hardware_id
|
||
}
|
||
|
||
request_kwargs = {
|
||
"params": params, # 使用URL参数而不是JSON body
|
||
"headers": {
|
||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36",
|
||
"Accept": "*/*",
|
||
"Connection": "keep-alive"
|
||
},
|
||
"timeout": 10,
|
||
"verify": False
|
||
}
|
||
|
||
for attempt in range(max_retries):
|
||
try:
|
||
# 创建session
|
||
session = requests.Session()
|
||
session.verify = False
|
||
|
||
# 设置重试策略
|
||
retry_strategy = urllib3.Retry(
|
||
total=3,
|
||
backoff_factor=0.5,
|
||
status_forcelist=[500, 502, 503, 504]
|
||
)
|
||
adapter = requests.adapters.HTTPAdapter(max_retries=retry_strategy)
|
||
session.mount("http://", adapter)
|
||
session.mount("https://", adapter)
|
||
|
||
# 发送请求
|
||
response = session.get( # 改用GET请求
|
||
self.config.get_api_url("heartbeat"),
|
||
**request_kwargs
|
||
)
|
||
|
||
# 检查响应
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
if result.get("code") == 200: # 修改成功码为200
|
||
data = result.get("data", {})
|
||
expire_time = data.get("expire_time", "")
|
||
days_left = data.get("days_left", 0)
|
||
status = data.get("status", "")
|
||
return True, f"心跳发送成功 [到期时间: {expire_time}, 剩余天数: {days_left}, 状态: {status}]"
|
||
else:
|
||
error_msg = result.get("msg", "未知错误")
|
||
if attempt < max_retries - 1:
|
||
logging.warning(f"第{attempt + 1}次心跳失败: {error_msg}, 准备重试...")
|
||
time.sleep(retry_delay)
|
||
continue
|
||
return False, f"心跳发送失败: {error_msg}"
|
||
else:
|
||
if attempt < max_retries - 1:
|
||
logging.warning(f"第{attempt + 1}次心跳HTTP错误: {response.status_code}, 准备重试...")
|
||
time.sleep(retry_delay)
|
||
continue
|
||
return False, f"心跳请求失败: HTTP {response.status_code}"
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
if attempt < max_retries - 1:
|
||
logging.warning(f"第{attempt + 1}次网络请求失败: {str(e)}, 准备重试...")
|
||
time.sleep(retry_delay)
|
||
continue
|
||
error_message = self._get_network_error_message(e)
|
||
return False, f"心跳发送失败: {error_message}"
|
||
except Exception as e:
|
||
if attempt < max_retries - 1:
|
||
logging.warning(f"第{attempt + 1}次发生异常: {str(e)}, 准备重试...")
|
||
time.sleep(retry_delay)
|
||
continue
|
||
logging.error(f"心跳发送异常: {str(e)}")
|
||
return False, f"心跳发送异常: {str(e)}"
|
||
|
||
return False, "多次尝试后心跳发送失败"
|
||
|
||
def main():
|
||
"""主函数"""
|
||
try:
|
||
# 检查管理员权限
|
||
if not is_admin():
|
||
print("\n[错误] 请以管理员身份运行此程序")
|
||
print("请右键点击程序，选择'以管理员身份运行'")
|
||
if run_as_admin():
|
||
return
|
||
input("\n按回车键退出...")
|
||
return
|
||
|
||
switcher = AccountSwitcher()
|
||
|
||
print("\n=== Cursor账号切换工具 ===")
|
||
print("1. 激活并切换账号")
|
||
print("2. 仅重置机器码")
|
||
|
||
while True:
|
||
try:
|
||
choice = int(input("\n请选择操作 (1 或 2): ").strip())
|
||
if choice in [1, 2]:
|
||
break
|
||
else:
|
||
print("无效的选项，请重新输入")
|
||
except ValueError:
|
||
print("请输入有效的数字")
|
||
|
||
if choice == 1:
|
||
activation_code = input("请输入激活码: ").strip()
|
||
if switcher.activate_and_switch(activation_code):
|
||
print("\n账号激活成功！")
|
||
else:
|
||
print("\n账号激活失败，请查看日志了解详细信息")
|
||
else:
|
||
if switcher.reset_machine_id():
|
||
print("\n机器码重置成功！")
|
||
else:
|
||
print("\n机器码重置失败，请查看日志了解详细信息")
|
||
|
||
except PermissionError as e:
|
||
print(f"\n[错误] {str(e)}")
|
||
print("请右键点击程序，选择'以管理员身份运行'")
|
||
except Exception as e:
|
||
logging.error(f"程序执行出错: {str(e)}")
|
||
print("\n程序执行出错，请查看日志了解详细信息")
|
||
|
||
finally:
|
||
input("\n按回车键退出...")
|
||
|
||
if __name__ == "__main__":
|
||
main() |