oadmin/auto_cursor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

保存现有功能 增加域名和添加时间关联

Browse Source
This commit is contained in:
huangzhenpc
2025-04-07 14:55:33 +08:00
parent 901c8c95e1
commit de057ae183
4 changed files with 159 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
debug_response.txt Normal file
View File

@@ -0,0 +1,2 @@
0:["$@1",["ARI0To80kOF93OzKZnFMN",null]]
1:{"payload":"Fe26.2*1*fb496fc3e44b03fba263d4904365d08ced6500c32da425de639cf4fd54b97e67*q0VzEVML3A50n2VWgjWCbw*jnmrTYQYUCSf8e1ly67O4w*1749190572271*0180c7834ee9ad01e6c2cb234416f8824ec91f4d96526d5aadbc1569e89f6245*GUdEqde3p6MD0n_JGEPb7-HF5-WEff4CtzpwyNl95k4~2","ttl":5184000000}

94
register/host_register_worker.py
View File

@@ -130,6 +130,55 @@ class HostRegisterWorker:
delay = random.uniform(*self.config.register_config.delay_range)
await asyncio.sleep(delay)
@staticmethod
async def _extract_auth_token(response_text: str) -> str | None:
"""从响应文本中提取pending_authentication_token"""
res = response_text.split('\n')
logger.debug(f"开始提取 auth_token，响应行数: {len(res)}")
# 检查邮箱是否可用
for line in res:
if '"code":"email_not_available"' in line:
logger.error("不受支持的邮箱")
raise RegisterError("Email is not available")
# 像register_worker.py中一样使用简单的路径
try:
for i, r in enumerate(res):
if r.startswith('0:'):
logger.debug(f"在第 {i+1} 行找到匹配")
data = json.loads(r.split('0:')[1])
# 使用完全相同的路径
auth_data = data[1][0][0][1]["children"][1]["children"][1]["children"][1]["children"][0]
params_str = auth_data.split('?')[1]
params_dict = json.loads(params_str)
token = params_dict['pending_authentication_token']
logger.debug(f"提取成功: {token[:10]}...")
return token
except Exception as e:
logger.error(f"提取token失败: {str(e)}")
logger.debug(f"响应内容预览: {response_text[:200]}...")
# 保存完整响应到文件以便调试
try:
with open('debug_response.txt', 'w', encoding='utf-8') as f:
f.write(response_text)
logger.debug("完整响应已保存到debug_response.txt")
except Exception:
pass
# 尝试备选方法 - 在整个响应文本中查找token
try:
import re
match = re.search(r'pending_authentication_token["\']?\s*[:=]\s*["\']?([^"\'&,\s]+)["\']?', response_text)
if match:
token = match.group(1)
logger.debug(f"使用正则表达式提取成功: {token[:10]}...")
return token
except Exception as e:
logger.error(f"正则表达式提取失败: {str(e)}")
return None
async def register(self, proxy: str, token_pair: Tuple[str, str]) -> Optional[Dict]:
"""使用自建邮箱完成注册流程"""
if not self.self_hosted_email:
@@ -218,12 +267,13 @@ class HostRegisterWorker:
boundary = "----WebKitFormBoundary2rKlvTagBEhneWi3"
headers = {
"accept": "text/x-component",
"next-action": "a67eb6646e43eddcbd0d038cbee664aac59f5a53",
"next-action": "770926d8148e29539286d20e1c1548d2aff6c0b9",
"content-type": f"multipart/form-data; boundary={boundary}",
"origin": "https://authenticator.cursor.sh",
"sec-fetch-dest": "empty",
"sec-fetch-mode": "cors",
"sec-fetch-site": "same-origin"
"sec-fetch-site": "same-origin",
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36"
}
params = {
@@ -251,30 +301,8 @@ class HostRegisterWorker:
text = response['body'].decode()
# 检查邮箱是否可用
if '"code":"email_not_available"' in text:
logger.error(f"邮箱 {email} 不受支持")
raise RegisterError("Email is not available")
# 提取pending_token
pending_token = None
res = text.split('\n')
try:
for i, r in enumerate(res):
if r.startswith('0:'):
logger.debug(f"在第 {i+1} 行找到匹配")
data = json.loads(r.split('0:')[1])
auth_data = data[1][0][0][1]["children"][1]["children"][1]["children"][1]["children"][0]
params_str = auth_data.split('?')[1]
params_dict = json.loads(params_str)
pending_token = params_dict['pending_authentication_token']
logger.debug(f"提取成功: {pending_token[:10]}...")
break
except Exception as e:
logger.error(f"提取token失败: {str(e)}")
raise RegisterError("Failed to extract auth token")
# 使用更简单的方法提取token，与register_worker.py相同的方式
pending_token = await self._extract_auth_token(text)
if not pending_token:
raise RegisterError("Failed to extract auth token")
@@ -332,7 +360,19 @@ class HostRegisterWorker:
redirect_url = response.get('headers', {}).get('x-action-redirect')
if not redirect_url:
raise RegisterError("未找到重定向URL，响应头: %s" % json.dumps(response.get('headers')))
logger.error(f"未找到重定向URL，响应头: {json.dumps(response.get('headers', {}))}")
# 尝试从响应体中提取
body = response.get('body', b'').decode()
if 'redirect' in body.lower():
logger.debug("尝试从响应体中提取重定向URL")
import re
match = re.search(r'redirect[^"\']*["\']([^"\']+)["\']', body)
if match:
redirect_url = match.group(1)
logger.debug(f"从响应体提取到重定向URL: {redirect_url}")
if not redirect_url:
raise RegisterError("未找到重定向URL，响应头: %s" % json.dumps(response.get('headers')))
return redirect_url

8
register/register_worker.py
View File

@@ -68,14 +68,13 @@ class FormBuilder:
def build_register_form(boundary: str, email: str, token: str) -> tuple[str, str]:
"""构建注册表单数据，返回(form_data, password)"""
password = FormBuilder._generate_password()
first_name, last_name = FormBuilder._generate_name()
fields = {
"1_state": "{\"returnTo\":\"/settings\"}",
"1_redirect_uri": "https://cursor.com/api/auth/callback",
"1_bot_detection_token": token,
"1_first_name": first_name,
"1_last_name": last_name,
"1_first_name": "wa",
"1_last_name": "niu",
"1_email": email,
"1_password": password,
"1_intent": "sign-up",
@@ -250,8 +249,7 @@ class RegisterWorker:
boundary = "----WebKitFormBoundary2rKlvTagBEhneWi3"
headers = {
"accept": "text/x-component",
# "next-action": "770926d8148e29539286d20e1c1548d2aff6c0b9",
"next-action": "a67eb6646e43eddcbd0d038cbee664aac59f5a53",
"next-action": "770926d8148e29539286d20e1c1548d2aff6c0b9",
"content-type": f"multipart/form-data; boundary={boundary}",
"origin": "https://authenticator.cursor.sh",
"sec-fetch-dest": "empty",

89
services/self_hosted_email.py
View File

@@ -1,4 +1,7 @@
import json
import re
import time
import asyncio
from typing import Optional, Dict, Any
from loguru import logger
@@ -15,17 +18,20 @@ class SelfHostedEmail:
Args:
fetch_manager: HTTP请求管理器
api_base_url: API基础URL，例如 "https://api.cursorpro.com.cn"
api_base_url: API基础URL，例如 "https://cursorapi3.nosqli.com/"
api_key: API密钥（可选）
"""
self.fetch_manager = fetch_manager
self.api_base_url = "https://api.cursorpro.com.cn"
self.api_base_url = "https://cursorapi3.nosqli.com/"
self.api_key = "1234567890"
# API端点
self.get_email_endpoint = "/admin/api.email/getEmail"
self.get_code_endpoint = "/admin/api.email/getVerificationCode"
# 新的邮件获取接口
self.new_email_api = "https://rnemail.nosqli.com/latest_email"
async def _make_api_request(self, endpoint: str, params: Dict[str, Any] = None) -> Dict[str, Any]:
"""向API发送请求
@@ -96,6 +102,13 @@ class SelfHostedEmail:
Returns:
验证码，失败时返回None
"""
# 首先尝试使用新接口获取
code = await self._get_code_from_email_api(email)
if code:
return code
# 如果新接口失败，尝试旧接口
logger.debug("新接口获取验证码失败，尝试使用旧接口")
try:
result = await self._make_api_request(
self.get_code_endpoint,
@@ -117,3 +130,75 @@ class SelfHostedEmail:
except Exception as e:
logger.error(f"获取验证码失败: {str(e)}")
return None
async def _get_code_from_email_api(self, email: str, max_retries: int = 5, retry_delay: int = 3) -> Optional[str]:
"""从新的邮件API获取验证码
Args:
email: 邮箱地址
max_retries: 最大重试次数
retry_delay: 重试间隔(秒)
Returns:
验证码，失败时返回None
"""
url = f"{self.new_email_api}?recipient={email}"
logger.debug(f"使用新接口获取验证码: {url}")
for attempt in range(max_retries):
try:
response = await self.fetch_manager.request("GET", url)
if 'error' in response:
logger.warning(f"获取邮件失败 (尝试 {attempt + 1}/{max_retries}): {response['error']}")
else:
email_data = json.loads(response['body'].decode())
logger.debug(f"获取到邮件数据: {str(email_data)[:200]}...")
# 解析邮件内容，提取验证码
if email_data and isinstance(email_data, dict):
body = email_data.get('body', '')
if 'code' in email_data:
# 如果API直接返回code字段
code = email_data.get('code')
if code:
logger.info(f"从邮件API直接获取到验证码: {code}")
return code
# 从邮件主体中提取验证码
if body:
# 正则表达式匹配6位数字验证码
code_match = re.search(r'code["\s:]*["\s]*(\d{6})["\s]*', body, re.IGNORECASE)
if code_match:
code = code_match.group(1)
logger.info(f"从邮件内容中提取到验证码: {code}")
return code
# 匹配"验证码"后的6位数字
code_match = re.search(r'[\u4e00-\u9fa5]*验证码[\u4e00-\u9fa5]*[：:]*\s*(\d{6})\b', body)
if code_match:
code = code_match.group(1)
logger.info(f"从邮件内容中提取到验证码: {code}")
return code
# 如果是Cursor邮件，尝试直接提取验证码格式
if "Cursor" in body and "verify" in body:
code_match = re.search(r'\b(\d{6})\b', body)
if code_match:
code = code_match.group(1)
logger.info(f"从Cursor邮件中提取到验证码: {code}")
return code
logger.warning(f"未能从邮件中提取到验证码 (尝试 {attempt + 1}/{max_retries})")
# 如果没有找到验证码，等待一段时间后重试
if attempt < max_retries - 1:
logger.debug(f"等待 {retry_delay} 秒后重试获取验证码...")
await asyncio.sleep(retry_delay)
except Exception as e:
logger.error(f"获取验证码出错 (尝试 {attempt + 1}/{max_retries}): {str(e)}")
if attempt < max_retries - 1:
await asyncio.sleep(retry_delay)
logger.error(f"在 {max_retries} 次尝试后未能获取到验证码")
return None