Files
cursorregasscountken/interactive/cursor_auth_manager.py

134 lines
4.9 KiB
Python

import os
import json
import sys
import time
import logging
import sqlite3
from pathlib import Path
class CursorAuthManager:
"""Cursor认证信息管理器"""
def __init__(self):
# 判断操作系统
if sys.platform == "win32": # Windows
appdata = os.getenv("APPDATA")
if appdata is None:
raise EnvironmentError("APPDATA 环境变量未设置")
self.db_path = os.path.join(
appdata, "Cursor", "User", "globalStorage", "state.vscdb"
)
elif sys.platform == "darwin": # macOS
self.db_path = os.path.abspath(os.path.expanduser(
"~/Library/Application Support/Cursor/User/globalStorage/state.vscdb"
))
elif sys.platform == "linux" : # Linux 和其他类Unix系统
self.db_path = os.path.abspath(os.path.expanduser(
"~/.config/Cursor/User/globalStorage/state.vscdb"
))
else:
raise NotImplementedError(f"不支持的操作系统: {sys.platform}")
self.cursor_path = Path(os.path.expanduser("~")) / "AppData" / "Local" / "Programs" / "Cursor"
def update_auth(self, email=None, access_token=None, refresh_token=None):
"""
更新Cursor的认证信息
:param email: 新的邮箱地址
:param access_token: 新的访问令牌
:param refresh_token: 新的刷新令牌
:return: bool 是否成功更新
"""
updates = []
# 登录状态
updates.append(("cursorAuth/cachedSignUpType", "Auth_0"))
if email is not None:
updates.append(("cursorAuth/cachedEmail", email))
if access_token is not None:
updates.append(("cursorAuth/accessToken", access_token))
if refresh_token is not None:
updates.append(("cursorAuth/refreshToken", refresh_token))
if not updates:
logging.warning("没有提供任何要更新的值")
return False
conn = None
try:
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
for key, value in updates:
# 检查key是否存在
check_query = f"SELECT COUNT(*) FROM itemTable WHERE key = ?"
cursor.execute(check_query, (key,))
if cursor.fetchone()[0] == 0:
insert_query = "INSERT INTO itemTable (key, value) VALUES (?, ?)"
cursor.execute(insert_query, (key, value))
else:
update_query = "UPDATE itemTable SET value = ? WHERE key = ?"
cursor.execute(update_query, (value, key))
if cursor.rowcount > 0:
logging.info(f"成功更新 {key.split('/')[-1]}")
else:
logging.warning(f"未找到 {key.split('/')[-1]} 或值未变化")
conn.commit()
logging.info(f"认证信息更新成功: {email}")
return True
except sqlite3.Error as e:
logging.error(f"数据库错误: {str(e)}")
return False
except Exception as e:
logging.error(f"更新认证信息失败: {str(e)}")
return False
finally:
if conn:
conn.close()
def restart_cursor(self) -> bool:
"""重启Cursor编辑器
Returns:
bool: 是否成功重启
"""
try:
logging.info("正在重启Cursor...")
if sys.platform == "win32":
# Windows系统
# 关闭Cursor
os.system("taskkill /f /im Cursor.exe 2>nul")
time.sleep(2)
# 获取Cursor安装路径
cursor_exe = self.cursor_path / "Cursor.exe"
if cursor_exe.exists():
# 启动Cursor
os.startfile(str(cursor_exe))
logging.info("Cursor重启成功")
return True
else:
logging.error(f"未找到Cursor程序: {cursor_exe}")
return False
elif sys.platform == "darwin":
# macOS系统
os.system("killall Cursor 2>/dev/null")
time.sleep(2)
os.system("open -a Cursor")
logging.info("Cursor重启成功")
return True
elif sys.platform == "linux":
# Linux系统
os.system("pkill -f cursor")
time.sleep(2)
os.system("cursor &")
logging.info("Cursor重启成功")
return True
else:
logging.error(f"不支持的操作系统: {sys.platform}")
return False
except Exception as e:
logging.error(f"重启Cursor时发生错误: {str(e)}")
return False