238 lines
8.0 KiB
Python
238 lines
8.0 KiB
Python
"""Cursor 服务管理类"""
|
||
|
||
import os
|
||
import json
|
||
import hashlib
|
||
from datetime import datetime, timedelta
|
||
from typing import Dict, Optional, Tuple
|
||
import urllib3
|
||
import requests
|
||
|
||
from cursor_token_refresher import CursorTokenRefresher
|
||
from machine_resetter import MachineResetter
|
||
from update_disabler import UpdateDisabler
|
||
from logger import logger
|
||
from common_utils import (
|
||
get_hardware_id,
|
||
get_user_state,
|
||
check_user_state,
|
||
refresh_user_state,
|
||
update_user_state,
|
||
activate_device
|
||
)
|
||
|
||
class CursorService:
|
||
"""Cursor 服务管理类"""
|
||
_instance = None
|
||
_initialized = False
|
||
_last_check_time = None
|
||
|
||
def __new__(cls):
|
||
if cls._instance is None:
|
||
cls._instance = super().__new__(cls)
|
||
return cls._instance
|
||
|
||
def __init__(self):
|
||
if not self._initialized:
|
||
# 基础组件初始化
|
||
self.token_refresher = CursorTokenRefresher()
|
||
self.machine_resetter = MachineResetter()
|
||
self.update_disabler = UpdateDisabler()
|
||
self.logger = logger.get_logger("CursorService")
|
||
|
||
# 配置文件路径
|
||
self.config_dir = os.path.join(os.getenv('APPDATA'), 'TingquanAssistant')
|
||
self.config_file = os.path.join(self.config_dir, 'config.json')
|
||
|
||
# 确保配置目录存在
|
||
os.makedirs(self.config_dir, exist_ok=True)
|
||
|
||
# 初始化配置
|
||
self._init_config()
|
||
|
||
# 首次初始化用户状态
|
||
self._lazy_init_user_state()
|
||
|
||
self._initialized = True
|
||
|
||
def _init_config(self):
|
||
"""初始化配置文件"""
|
||
# 默认配置
|
||
self.default_config = {
|
||
"version": "4.0.0.1",
|
||
"last_check_update": None,
|
||
"auto_check_update": True,
|
||
"theme": "light"
|
||
}
|
||
|
||
# 读取或创建配置文件
|
||
if not os.path.exists(self.config_file):
|
||
with open(self.config_file, 'w', encoding='utf-8') as f:
|
||
json.dump(self.default_config, f, indent=2, ensure_ascii=False)
|
||
self.config = self.default_config
|
||
else:
|
||
try:
|
||
with open(self.config_file, 'r', encoding='utf-8') as f:
|
||
self.config = json.load(f)
|
||
except Exception as e:
|
||
self.logger.error(f"读取配置文件失败: {e}")
|
||
self.config = self.default_config
|
||
|
||
def _lazy_init_user_state(self):
|
||
"""
|
||
懒加载方式初始化用户状态
|
||
只在首次使用时初始化，避免频繁刷新
|
||
"""
|
||
try:
|
||
if not self._last_check_time:
|
||
machine_id = get_hardware_id()
|
||
success, msg, state_data = refresh_user_state(machine_id)
|
||
if success:
|
||
self._last_check_time = datetime.now()
|
||
self.logger.info("用户状态初始化成功")
|
||
else:
|
||
self.logger.warning(f"用户状态初始化失败: {msg}")
|
||
except Exception as e:
|
||
self.logger.error(f"初始化用户状态异常: {e}")
|
||
|
||
def _should_refresh_state(self) -> bool:
|
||
"""
|
||
检查是否需要刷新状态
|
||
每5分钟检测一次设备状态
|
||
|
||
Returns:
|
||
bool: 是否需要刷新状态
|
||
"""
|
||
try:
|
||
# 如果没有上次检查时间，需要刷新
|
||
if not self._last_check_time:
|
||
return True
|
||
|
||
# 计算距离上次检查的时间
|
||
now = datetime.now()
|
||
time_diff = now - self._last_check_time
|
||
|
||
# 如果超过5分钟，需要刷新
|
||
if time_diff.total_seconds() >= 300: # 5分钟 = 300秒
|
||
self.logger.debug("距离上次状态检查已超过5分钟")
|
||
return True
|
||
|
||
return False
|
||
|
||
except Exception as e:
|
||
self.logger.error(f"检查状态刷新时间异常: {e}")
|
||
# 发生异常时返回False，避免频繁刷新
|
||
return False
|
||
|
||
def check_activation_status(self) -> Dict:
|
||
"""
|
||
检查当前激活状态
|
||
如果需要刷新则自动刷新状态
|
||
|
||
Returns:
|
||
Dict: 状态信息字典
|
||
"""
|
||
try:
|
||
# 检查是否需要刷新状态
|
||
if self._should_refresh_state():
|
||
machine_id = get_hardware_id()
|
||
success, msg, state_data = refresh_user_state(machine_id)
|
||
if success:
|
||
self._last_check_time = datetime.now()
|
||
self.logger.info("设备状态检查完成")
|
||
else:
|
||
self.logger.warning(f"设备状态检查失败: {msg}")
|
||
except Exception as e:
|
||
self.logger.error(f"设备状态检查异常: {e}")
|
||
|
||
# 无论是否刷新成功，都返回当前状态
|
||
return get_user_state()
|
||
|
||
def activate_with_code(self, activation_code: str) -> Tuple[bool, str]:
|
||
"""
|
||
使用激活码激活
|
||
|
||
Args:
|
||
activation_code: 激活码
|
||
|
||
Returns:
|
||
Tuple[bool, str]: (是否成功, 消息)
|
||
"""
|
||
try:
|
||
self.logger.info(f"开始处理激活码: {activation_code}")
|
||
|
||
# 获取机器ID
|
||
machine_id = get_hardware_id()
|
||
if not machine_id:
|
||
self.logger.error("获取机器ID失败")
|
||
return False, "获取机器ID失败"
|
||
|
||
self.logger.info(f"获取到机器ID: {machine_id}")
|
||
|
||
# 调用激活接口
|
||
self.logger.info("正在调用激活接口...")
|
||
success, msg, data = activate_device(machine_id, activation_code)
|
||
|
||
if success:
|
||
self.logger.info(f"激活成功: {msg}")
|
||
else:
|
||
self.logger.warning(f"激活失败: {msg}")
|
||
|
||
return success, msg
|
||
|
||
except Exception as e:
|
||
error_msg = f"激活失败: {str(e)}"
|
||
self.logger.error(error_msg)
|
||
return False, error_msg
|
||
|
||
def refresh_account(self, machine_id: str) -> Tuple[bool, str, Optional[Dict]]:
|
||
"""刷新账号"""
|
||
# 检查是否已激活
|
||
if not check_user_state():
|
||
return False, "请先激活软件", None
|
||
return self.token_refresher.refresh_token(machine_id)
|
||
|
||
def reset_machine(self) -> Tuple[bool, str, Dict]:
|
||
"""重置机器码"""
|
||
# 检查是否已激活
|
||
if not check_user_state():
|
||
return False, "请先激活软件", {}
|
||
|
||
try:
|
||
new_ids, new_guid = self.machine_resetter.run()
|
||
return True, "重置机器码成功请重启cursor编辑器", {
|
||
"new_ids": new_ids,
|
||
"new_guid": new_guid
|
||
}
|
||
except Exception as e:
|
||
return False, f"重置失败: {str(e)}", {}
|
||
|
||
def disable_update(self) -> Tuple[bool, str]:
|
||
"""禁用更新"""
|
||
# 检查是否已激活
|
||
if not check_user_state():
|
||
return False, "请先激活软件"
|
||
|
||
success = self.update_disabler.disable()
|
||
return success, "禁用成功" if success else "禁用失败"
|
||
|
||
def save_config(self, new_config: Dict) -> bool:
|
||
"""
|
||
保存配置
|
||
:param new_config: 新的配置信息
|
||
:return: 是否成功
|
||
"""
|
||
try:
|
||
self.config.update(new_config)
|
||
with open(self.config_file, 'w', encoding='utf-8') as f:
|
||
json.dump(self.config, f, indent=2, ensure_ascii=False)
|
||
return True
|
||
except Exception as e:
|
||
self.logger.error(f"保存配置失败: {e}")
|
||
return False
|
||
|
||
def get_config(self) -> Dict:
|
||
"""获取当前配置"""
|
||
return self.config.copy()
|
||
|