4 Commits

10 changed files with 1019 additions and 532 deletions

4
.gitignore vendored
View File

@@ -17,4 +17,6 @@ venv/
env/
# Local development settings
.env
.env
testversion.txt

View File

@@ -50,9 +50,41 @@ class AccountSwitcher:
self.package_json = self.app_path / "package.json"
self.auth_manager = CursorAuthManager()
self.config = Config()
self.hardware_id = get_hardware_id()
self.hardware_id = self.get_hardware_id() # 先获取硬件ID
self.registry = CursorRegistry() # 添加注册表操作工具类
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:
@@ -71,199 +103,126 @@ class AccountSwitcher:
import platform
import socket
import requests
import subprocess
# 获取操作系统信息
os_info = f"{platform.system()} {platform.version()}"
os_info = f"{platform.system()} {platform.release()}"
# 获取设备名称
device_name = platform.node()
# 获取地理位置(可选)
try:
ip_info = requests.get('https://ipapi.co/json/', timeout=5).json()
# 在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 = ""
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 10",
"os": "Windows",
"device_name": "未知设备",
"location": ""
"ip": "未知",
"location": "未知"
}
def check_activation_code(self, code: str) -> Tuple[bool, str, Optional[Dict]]:
"""验证激活码
def check_activation_code(self, code: str) -> tuple:
"""检查激活码
Args:
code: 激活码
Returns:
Tuple[bool, str, Optional[Dict]]: (是否成功, 提示消息, 账号信息)
tuple: (成功标志, 消息, 账号信息)
"""
try:
# 获取当前状态和历史记录
member_info = self.config.load_member_info()
activation_history = member_info.get("activation_records", []) if member_info else []
data = {
"machine_id": self.hardware_id,
"code": code
}
# 分割多个激活码
codes = [c.strip() for c in code.split(',')]
success_codes = []
failed_codes = []
activation_results = []
# 获取设备信息
device_info = self.get_device_info()
# 逐个验证激活码
for single_code in codes:
if not single_code:
continue
# 验证激活码
endpoint = "https://cursorapi.nosqli.com/admin/api.member/activate"
data = {
"code": single_code,
"machine_id": self.hardware_id,
"os": device_info["os"],
"device_name": device_info["device_name"],
"location": device_info["location"]
}
headers = {
"Content-Type": "application/json"
}
try:
response = requests.post(
endpoint,
json=data,
headers=headers,
timeout=10
)
response_data = response.json()
if response_data.get("code") == 200:
result_data = response_data.get("data", {})
logging.info(f"激活码 {single_code} 验证成功: {response_data.get('msg', '')}")
activation_results.append(result_data)
success_codes.append(single_code)
elif response_data.get("code") == 400:
error_msg = response_data.get("msg", "参数错误")
if "已被使用" in error_msg or "已激活" in error_msg:
logging.warning(f"激活码 {single_code} 已被使用")
failed_codes.append(f"{single_code} (已被使用)")
else:
logging.error(f"激活码 {single_code} 验证失败: {error_msg}")
failed_codes.append(f"{single_code} ({error_msg})")
elif response_data.get("code") == 500:
error_msg = response_data.get("msg", "系统错误")
logging.error(f"激活码 {single_code} 验证失败: {error_msg}")
failed_codes.append(f"{single_code} ({error_msg})")
else:
error_msg = response_data.get("msg", "未知错误")
logging.error(f"激活码 {single_code} 验证失败: {error_msg}")
failed_codes.append(f"{single_code} ({error_msg})")
except requests.RequestException as e:
logging.error(f"激活码 {single_code} 请求失败: {str(e)}")
failed_codes.append(f"{single_code} (网络请求失败)")
except Exception as e:
logging.error(f"激活码 {single_code} 处理失败: {str(e)}")
failed_codes.append(f"{single_code} (处理失败)")
if not success_codes:
failed_msg = "\n".join(failed_codes)
return False, f"激活失败:\n{failed_msg}", None
# 禁用SSL警告
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# 设置请求参数
request_kwargs = {
"json": data,
"headers": {"Content-Type": "application/json"},
"timeout": 5, # 减少超时时间从10秒到5秒
"verify": False # 禁用SSL验证
}
# 尝试发送请求
try:
# 使用最后一次激活的结果作为最终状态
final_result = activation_results[-1]
response = requests.post(
self.config.get_api_url("activate"),
**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
# 合并历史记录
new_activation_records = final_result.get("activation_records", [])
if activation_history:
# 保留旧的激活记录,避免重复
existing_codes = {record.get("code") for record in activation_history}
for record in new_activation_records:
if record.get("code") not in existing_codes:
activation_history.append(record)
else:
activation_history = new_activation_records
# 保存会员信息,包含完整的历史记录
member_info = {
"hardware_id": final_result.get("machine_id", self.hardware_id),
"expire_time": final_result.get("expire_time", ""),
"days_left": final_result.get("days_left", 0),
"total_days": final_result.get("total_days", 0),
"status": final_result.get("status", "inactive"),
"device_info": final_result.get("device_info", device_info),
"activation_time": final_result.get("activation_time", ""),
"activation_records": activation_history # 使用合并后的历史记录
# 使用自定义SSL上下文重试请求
session = requests.Session()
session.verify = False
response = session.post(
self.config.get_api_url("activate"),
**request_kwargs
)
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()
}
self.config.save_member_info(member_info)
# 生成结果消息
message = f"激活成功\n"
# 显示每个成功激活码的信息
for i, result in enumerate(activation_results, 1):
message += f"\n{i}个激活码:\n"
message += f"- 新增天数: {result.get('added_days', 0)}\n"
activation_time = result.get('activation_time', '')
if activation_time:
try:
from datetime import datetime
dt = datetime.strptime(activation_time, "%Y-%m-%d %H:%M:%S")
activation_time = dt.strftime("%Y-%m-%d %H:%M:%S")
except:
pass
message += f"- 激活时间: {activation_time}\n"
message += f"\n最终状态:"
message += f"\n- 总天数: {final_result.get('total_days', 0)}"
message += f"\n- 剩余天数: {final_result.get('days_left', 0)}"
# 格式化到期时间显示
expire_time = final_result.get('expire_time', '')
if expire_time:
try:
dt = datetime.strptime(expire_time, "%Y-%m-%d %H:%M:%S")
expire_time = dt.strftime("%Y-%m-%d %H:%M:%S")
except:
pass
message += f"\n- 到期时间: {expire_time}"
# 显示完整的激活记录历史
message += "\n\n历史激活记录:"
for record in activation_history:
activation_time = record.get('activation_time', '')
if activation_time:
try:
dt = datetime.strptime(activation_time, "%Y-%m-%d %H:%M:%S")
activation_time = dt.strftime("%Y-%m-%d %H:%M:%S")
except:
pass
message += f"\n- 激活码: {record.get('code', '')}"
message += f"\n 天数: {record.get('days', 0)}"
message += f"\n 时间: {activation_time}\n"
if failed_codes:
message += f"\n\n以下激活码验证失败:\n" + "\n".join(failed_codes)
return True, message, member_info
except Exception as e:
logging.error(f"处理激活结果时出错: {str(e)}")
return False, f"处理激活结果时出错: {str(e)}", None
return True, result["msg"], account_info
# 激活码无效或已被使用
elif result["code"] == 400:
logging.warning(f"激活码无效或已被使用: {result.get('msg', '未知错误')}")
return False, result.get("msg", "激活码无效或已被使用"), None
# 其他错误情况
else:
logging.error(f"激活失败: {result.get('msg', '未知错误')}")
return False, result["msg"], None # 返回 None 而不是空的账号信息
except Exception as e:
logging.error(f"验证激活码时出错: {str(e)}")
return False, f"验证激活码时出错: {str(e)}", None
logging.error(f"激活失败: {str(e)}")
return False, f"激活失败: {str(e)}", None # 返回 None 而不是空的账号信息
def reset_machine_id(self) -> bool:
"""重置机器码"""
@@ -356,93 +315,86 @@ class AccountSwitcher:
logging.error(f"激活过程出错: {str(e)}")
return False, f"激活失败: {str(e)}"
def get_member_status(self) -> Optional[Dict]:
def get_member_status(self) -> dict:
"""获取会员状态
Returns:
Optional[Dict]: 会员状态信息
dict: 会员状态信息,包含:
- status: 状态(active/inactive/expired)
- expire_time: 到期时间
- total_days: 总天数
- days_left: 剩余天数
- device_info: 设备信息
"""
try:
# 读取保存的会员信息
member_info = self.config.load_member_info()
# 构造状态检查请求
endpoint = "https://cursorapi.nosqli.com/admin/api.member/status"
data = {
"machine_id": self.hardware_id
}
headers = {
"Content-Type": "application/json"
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
}
try:
response = requests.post(endpoint, json=data, headers=headers, timeout=10)
response_data = response.json()
if response_data.get("code") == 200:
# 正常状态
data = response_data.get("data", {})
status = data.get("status", "inactive")
# 构造会员信息
member_info = {
"hardware_id": data.get("machine_id", self.hardware_id),
"expire_time": data.get("expire_time", ""),
"days_left": data.get("days_left", 0), # 使用days_left
"total_days": data.get("total_days", 0), # 使用total_days
"status": status,
"activation_records": data.get("activation_records", []) # 保存激活记录
}
# 打印调试信息
logging.info(f"API返回数据: {data}")
logging.info(f"处理后的会员信息: {member_info}")
self.config.save_member_info(member_info)
return member_info
elif response_data.get("code") == 401:
# 未激活或已过期
logging.warning("会员未激活或已过期")
return self._get_inactive_status()
elif response_data.get("code") == 400:
# 参数错误
error_msg = response_data.get("msg", "参数错误")
logging.error(f"获取会员状态失败: {error_msg}")
return self._get_inactive_status()
elif response_data.get("code") == 500:
# 系统错误
error_msg = response_data.get("msg", "系统错误")
logging.error(f"获取会员状态失败: {error_msg}")
return self._get_inactive_status()
else:
# 其他未知错误
error_msg = response_data.get("msg", "未知错误")
logging.error(f"获取会员状态失败: {error_msg}")
return self._get_inactive_status()
except requests.RequestException as e:
logging.error(f"请求会员状态失败: {str(e)}")
return self._get_inactive_status()
except Exception as e:
logging.error(f"获取会员状态出错: {str(e)}")
return self._get_inactive_status()
# 使用session来复用连接
session = requests.Session()
session.verify = False
def _get_inactive_status(self) -> Dict:
"""获取未激活状态的默认信息"""
return {
"hardware_id": self.hardware_id,
"expire_time": "",
"days": 0,
"total_days": 0,
"status": "inactive",
"last_activation": {},
"activation_records": []
}
# 尝试发送请求
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 restart_cursor(self) -> bool:
"""重启Cursor编辑器
@@ -509,17 +461,33 @@ class AccountSwitcher:
"Content-Type": "application/json"
}
# 禁用SSL警告
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# 设置请求参数
request_kwargs = {
"json": data,
"headers": headers,
"timeout": 30, # 增加超时时间
"verify": False # 禁用SSL验证
}
try:
# 添加SSL验证选项和超时设置
response = requests.post(
endpoint,
json=data,
headers=headers,
timeout=30, # 增加超时时间
verify=False, # 禁用SSL验证
)
# 禁用SSL警告
requests.packages.urllib3.disable_warnings()
# 尝试发送请求
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()

BIN
banbenjietu.png Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 37 KiB

View File

@@ -1,5 +1,7 @@
# -*- mode: python ; coding: utf-8 -*-
import os
import sys
from PyInstaller.utils.hooks import collect_all
def get_version():
with open('version.txt', 'r', encoding='utf-8') as f:
@@ -8,26 +10,43 @@ def get_version():
version = get_version()
# 收集所有需要的依赖
datas = [('icon', 'icon'), ('version.txt', '.')]
binaries = []
hiddenimports = [
'win32gui', 'win32con', 'win32process', 'psutil', # Windows API 相关
'PyQt5', 'PyQt5.QtCore', 'PyQt5.QtGui', 'PyQt5.QtWidgets', # GUI 相关
'PyQt5.sip', # PyQt5 必需
'PyQt5.QtNetwork', # 网络相关
'PIL', # Pillow 相关
'PIL._imaging', # Pillow 核心
'requests', 'urllib3', 'certifi', # 网络请求相关
'json', 'uuid', 'hashlib', 'logging', # 基础功能相关
'importlib', # 导入相关
'pkg_resources', # 包资源
]
# 主要的分析对象
a = Analysis(
['main.py'],
pathex=[],
binaries=[],
datas=[('icon', 'icon'), ('version.txt', '.')],
hiddenimports=[
'win32gui', 'win32con', 'win32process', 'psutil', # Windows API 相关
'PyQt5', 'PyQt5.QtCore', 'PyQt5.QtGui', 'PyQt5.QtWidgets', # GUI 相关
'requests', 'urllib3', 'certifi', # 网络请求相关
'json', 'uuid', 'hashlib', 'logging' # 基础功能相关
],
binaries=binaries,
datas=datas,
hiddenimports=hiddenimports,
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=['_tkinter', 'tkinter', 'Tkinter'], # 排除 tkinter 相关模块
excludes=['_tkinter', 'tkinter', 'Tkinter'],
win_no_prefer_redirects=False,
win_private_assemblies=False,
noarchive=False,
optimize=0,
module_collection_mode={'PyQt5': 'pyz+py'},
)
# 创建PYZ
pyz = PYZ(a.pure)
# 创建EXE
exe = EXE(
pyz,
a.scripts,

File diff suppressed because it is too large Load Diff

28
main.py
View File

@@ -5,12 +5,17 @@ import os
import atexit
import shutil
import tempfile
import urllib3
from pathlib import Path
from PyQt5.QtWidgets import QApplication, QMessageBox, QSystemTrayIcon, QMenu
from PyQt5.QtGui import QIcon
from PyQt5.QtCore import Qt
from gui.main_window import MainWindow
# 禁用所有 SSL 相关警告
urllib3.disable_warnings()
logging.getLogger('urllib3').setLevel(logging.ERROR)
def cleanup_temp():
"""清理临时文件"""
try:
@@ -52,6 +57,18 @@ def main():
# 注册退出时的清理函数
atexit.register(cleanup_temp)
# 创建QApplication实例
app = QApplication(sys.argv)
# 检查系统托盘是否可用
if not QSystemTrayIcon.isSystemTrayAvailable():
logging.error("系统托盘不可用")
QMessageBox.critical(None, "错误", "系统托盘不可用,程序无法正常运行。")
return 1
# 设置应用程序不会在最后一个窗口关闭时退出
app.setQuitOnLastWindowClosed(False)
setup_logging()
# 检查Python版本
@@ -66,7 +83,6 @@ def main():
logging.info(f" - {p}")
logging.info("正在初始化主窗口...")
app = QApplication(sys.argv)
# 设置应用程序ID (在设置图标之前)
if sys.platform == "win32":
@@ -96,15 +112,17 @@ def main():
logging.info("正在启动主窗口...")
window.show()
sys.exit(app.exec_())
return app.exec_()
except Exception as e:
error_msg = f"程序运行出错: {str(e)}\n{traceback.format_exc()}"
logging.error(error_msg)
# 使用 QMessageBox 显示错误
app = QApplication(sys.argv)
if QApplication.instance() is None:
app = QApplication(sys.argv)
QMessageBox.critical(None, "错误", error_msg)
sys.exit(1)
return 1
if __name__ == "__main__":
main()
sys.exit(main())

View File

@@ -2,8 +2,9 @@
# pip install -i https://pypi.tuna.tsinghua.edu.cn/simple -r requirements.txt
requests==2.31.0
setuptools>=68.0.0
altgraph>=0.17.4
pyinstaller==6.3.0
pillow==10.2.0 # For icon processing
setuptools==65.5.1 # Fix pkg_resources.extern issue
PyQt5==5.15.10 # GUI framework
pywin32==306 # Windows API support
pillow==10.2.0
PyQt5==5.15.10
pywin32==306

51
testbuild.bat Normal file
View File

@@ -0,0 +1,51 @@
@echo off
chcp 65001 >nul
setlocal EnableDelayedExpansion
REM 激活虚拟环境
call venv\Scripts\activate.bat
REM 确保安装了必要的包
pip install -r requirements.txt
REM 读取当前版本号
set /p VERSION=<version.txt
echo 当前正式版本: %VERSION%
REM 读取测试版本号(如果存在)
if exist testversion.txt (
set /p TEST_VERSION=<testversion.txt
) else (
set TEST_VERSION=0
)
REM 增加测试版本号
set /a TEST_VERSION+=1
echo !TEST_VERSION!>testversion.txt
echo 测试版本号: !TEST_VERSION!
REM 组合完整版本号
set FULL_VERSION=%VERSION%.!TEST_VERSION!
echo 完整版本号: !FULL_VERSION!
REM 创建测试版本输出目录
if not exist "dist\test" mkdir "dist\test"
REM 清理旧文件
if exist "dist\听泉cursor助手%VERSION%.exe" del "dist\听泉cursor助手%VERSION%.exe"
if exist "build" rmdir /s /q "build"
REM 执行打包
venv\Scripts\python.exe -m PyInstaller build_nezha.spec --clean
REM 移动并重命名文件
move "dist\听泉cursor助手%VERSION%.exe" "dist\test\听泉cursor助手v!FULL_VERSION!.exe"
echo.
echo 测试版本构建完成!
echo 版本号: v!FULL_VERSION!
echo 文件位置: dist\test\听泉cursor助手v!FULL_VERSION!.exe
REM 退出虚拟环境
deactivate
pause

View File

@@ -4,8 +4,15 @@ import logging
from pathlib import Path
class Config:
"""配置类"""
def __init__(self):
self.api_base_url = "https://cursorapi.nosqli.com/admin"
self.base_url = "https://cursorapi.nosqli.com"
self.api_endpoints = {
"activate": f"{self.base_url}/admin/api.member/activate",
"status": f"{self.base_url}/admin/api.member/status",
"get_unused": f"{self.base_url}/admin/api.account/getUnused"
}
self.config_dir = Path(os.path.expanduser("~")) / ".cursor_switcher"
self.config_file = self.config_dir / "config.json"
self.member_file = self.config_dir / "member.json"
@@ -64,4 +71,15 @@ class Config:
}
with open(self.config_file, "w", encoding="utf-8") as f:
json.dump(config, f, indent=2, ensure_ascii=False)
self.api_token = api_token
self.api_token = api_token
def get_api_url(self, endpoint_name: str) -> str:
"""获取API端点URL
Args:
endpoint_name: 端点名称
Returns:
str: 完整的API URL
"""
return self.api_endpoints.get(endpoint_name, "")

View File

@@ -1 +1 @@
3.3.4
3.4.0