mirrors/cursor-free-vip
0
0
Fork 0
mirror of https://github.com/yeongpin/cursor-free-vip.git synced 2025-04-24 08:25:23 +00:00
Code

Refactor NewTempEmail class to enhance email creation and verification processes, including improved browser setup and error handling. Add new_tempemail_api.py to .gitignore.

Browse Source
This commit is contained in:
yeongpin 2025-03-29 22:18:11 +08:00
parent e5b7e5727c
commit 816a09d4de
2 changed files with 152 additions and 260 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@ -15,6 +15,7 @@ build.sh
ENV/ ENV/
test.py test.py
new_tempemail_smail.py new_tempemail_smail.py
new_tempemail_api.py
install.bat install.bat
run.bat run.bat

411
new_tempemail.py
View File

@ -14,16 +14,8 @@ init()
class NewTempEmail: class NewTempEmail:
def __init__(self, translator=None): def __init__(self, translator=None):
self.translator = translator self.translator = translator
# Randomly choose between mail.tm and mail.gw self.page = None
self.services = [ self.setup_browser()
{"name": "mail.tm", "api_url": "https://api.mail.tm"}
]
self.selected_service = random.choice(self.services)
self.api_url = self.selected_service["api_url"]
self.token = None
self.email = None
self.password = None
self.blocked_domains = self.get_blocked_domains()
def get_blocked_domains(self): def get_blocked_domains(self):
"""Get blocked domains list""" """Get blocked domains list"""
@ -90,166 +82,116 @@ class NewTempEmail:
return filtered_domains return filtered_domains
def _generate_credentials(self):
"""generate random username and password"""
username = ''.join(random.choices(string.ascii_lowercase + string.digits, k=10))
password = ''.join(random.choices(string.ascii_letters + string.digits + string.punctuation, k=12))
return username, password
def get_extension_block(self):
"""获取插件路径"""
root_dir = os.getcwd()
extension_path = os.path.join(root_dir, "PBlock")
if hasattr(sys, "_MEIPASS"):
extension_path = os.path.join(sys._MEIPASS, "PBlock")
if not os.path.exists(extension_path):
raise FileNotFoundError(f"插件不存在: {extension_path}")
return extension_path
def setup_browser(self):
"""设置浏览器"""
try:
if self.translator:
print(f"{Fore.CYAN}ℹ️ {self.translator.get('email.starting_browser')}{Style.RESET_ALL}")
else:
print(f"{Fore.CYAN}ℹ️ 正在启动浏览器...{Style.RESET_ALL}")
# 创建浏览器选项
co = ChromiumOptions()
co.set_argument("--headless=new")
co.set_argument("--no-sandbox")
co.auto_port() # 自动设置端口
# 加载 uBlock 插件
try:
extension_path = self.get_extension_block()
co.set_argument("--allow-extensions-in-incognito")
co.add_extension(extension_path)
except Exception as e:
if self.translator:
print(f"{Fore.YELLOW}⚠️ {self.translator.get('email.extension_load_error')}: {str(e)}{Style.RESET_ALL}")
else:
print(f"{Fore.YELLOW}⚠️ 加载插件失败: {str(e)}{Style.RESET_ALL}")
self.page = ChromiumPage(co)
return True
except Exception as e:
if self.translator:
print(f"{Fore.RED}❌ {self.translator.get('email.browser_start_error')}: {str(e)}{Style.RESET_ALL}")
else:
print(f"{Fore.RED}❌ 启动浏览器失败: {str(e)}{Style.RESET_ALL}")
return False
def create_email(self): def create_email(self):
"""create temporary email""" """create temporary email"""
max_retries = 3 # Maximum number of retries try:
attempt = 0 # Current attempt count if self.translator:
print(f"{Fore.CYAN}ℹ️ {self.translator.get('email.visiting_site')}{Style.RESET_ALL}")
while attempt < max_retries: else:
attempt += 1 print(f"{Fore.CYAN}ℹ️ 正在访问 smailpro.com...{Style.RESET_ALL}")
try:
if self.translator: # load blocked domains list
print(f"{Fore.CYAN}ℹ️ {self.translator.get('email.visiting_site').replace('mail.tm', self.selected_service['name'])}{Style.RESET_ALL}") self.blocked_domains = self.get_blocked_domains()
else:
print(f"{Fore.CYAN}ℹ️ 正在访问 {self.selected_service['name']}...{Style.RESET_ALL}") # visit website
self.page.get("https://smailpro.com/")
# Get available domain list time.sleep(2)
try:
domains_response = requests.get(f"{self.api_url}/domains", timeout=10) # click create email button
if domains_response.status_code != 200: create_button = self.page.ele('xpath://button[@title="Create temporary email"]')
print(f"{Fore.RED}❌ {self.translator.get('email.domains_list_error', error=domains_response.status_code)}{Style.RESET_ALL}") if create_button:
print(f"{Fore.RED}❌ {self.translator.get('email.domains_list_error', error=domains_response.text)}{Style.RESET_ALL}") create_button.click()
raise Exception(f"{self.translator.get('email.failed_to_get_available_domains') if self.translator else 'Failed to get available domains'}") time.sleep(1)
domains = domains_response.json()["hydra:member"] # click Create button in popup
print(f"{Fore.CYAN}ℹ️ {self.translator.get('email.available_domains_loaded', count=len(domains))}{Style.RESET_ALL}") modal_create_button = self.page.ele('xpath://button[contains(text(), "Create")]')
if modal_create_button:
if not domains: modal_create_button.click()
raise Exception(f"{self.translator.get('email.no_available_domains') if self.translator else '没有可用域名'}") time.sleep(2)
except Exception as e:
print(f"{Fore.RED}❌ 获取域名列表时出错: {str(e)}{Style.RESET_ALL}") # get email address - modify selector
raise email_div = self.page.ele('xpath://div[@class="text-base sm:text-lg md:text-xl text-gray-700"]')
if email_div:
# Exclude blocked domains email = email_div.text.strip()
try: if '@' in email: # check if it's a valid email address
filtered_domains = self.exclude_blocked_domains(domains) # check if domain is blocked
if self.translator: domain = email.split('@')[1]
print(f"{Fore.CYAN}ℹ️ {self.translator.get('email.domains_filtered', count=len(filtered_domains))}{Style.RESET_ALL}") if self.blocked_domains and domain in self.blocked_domains:
else:
print(f"{Fore.CYAN}ℹ️ 过滤后剩余 {len(filtered_domains)} 个可用域名{Style.RESET_ALL}")
if not filtered_domains:
if self.translator:
print(f"{Fore.RED}❌ {self.translator.get('email.all_domains_blocked')}{Style.RESET_ALL}")
else:
print(f"{Fore.RED}❌ 所有域名都被屏蔽了，尝试切换服务{Style.RESET_ALL}")
# Switch to another service
for service in self.services:
if service["api_url"] != self.api_url:
self.selected_service = service
self.api_url = service["api_url"]
if self.translator: if self.translator:
print(f"{Fore.CYAN}ℹ️ {self.translator.get('email.switching_service', service=service['name'])}{Style.RESET_ALL}") print(f"{Fore.YELLOW}⚠️ {self.translator.get('email.domain_blocked')}: {domain}{Style.RESET_ALL}")
else: else:
print(f"{Fore.CYAN}ℹ️ 切换到 {service['name']} 服务{Style.RESET_ALL}") print(f"{Fore.YELLOW}⚠️ 域名已被屏蔽: {domain}，尝试重新创建邮箱{Style.RESET_ALL}")
return self.create_email() # Recursively call # create email again
return self.create_email()
raise Exception(f"{self.translator.get('email.no_available_domains_after_filtering') if self.translator else '过滤后没有可用域名'}")
except Exception as e: if self.translator:
print(f"{Fore.RED}❌ 过滤域名时出错: {str(e)}{Style.RESET_ALL}") print(f"{Fore.GREEN}✅ {self.translator.get('email.create_success')}: {email}{Style.RESET_ALL}")
raise else:
print(f"{Fore.GREEN}✅ 创建邮箱成功: {email}{Style.RESET_ALL}")
# Generate random username and password return email
try: if self.translator:
username, password = self._generate_credentials() print(f"{Fore.RED}❌ {self.translator.get('email.create_failed')}{Style.RESET_ALL}")
self.password = password else:
print(f"{Fore.RED}❌ 创建邮箱失败{Style.RESET_ALL}")
# Create email account return None
selected_domain = filtered_domains[0]['domain']
email = f"{username}@{selected_domain}" except Exception as e:
if self.translator:
if self.translator: print(f"{Fore.RED}❌ {self.translator.get('email.create_error')}: {str(e)}{Style.RESET_ALL}")
print(f"{Fore.CYAN}ℹ️ {self.translator.get('email.trying_to_create_email', email=email)}{Style.RESET_ALL}") else:
else: print(f"{Fore.RED}❌ 创建邮箱出错: {str(e)}{Style.RESET_ALL}")
print(f"{Fore.CYAN}ℹ️ 尝试创建邮箱: {email}{Style.RESET_ALL}") return None
account_data = {
"address": email,
"password": password
}
except Exception as e:
print(f"{Fore.RED}❌ 生成凭据时出错: {str(e)}{Style.RESET_ALL}")
raise
# Create account
try:
create_response = requests.post(f"{self.api_url}/accounts", json=account_data, timeout=15)
if create_response.status_code != 201:
if self.translator:
print(f"{Fore.RED}❌ {self.translator.get('email.failed_to_create_account', error=create_response.status_code)}{Style.RESET_ALL}")
else:
print(f"{Fore.RED}❌ 创建账户失败: 状态码 {create_response.status_code}{Style.RESET_ALL}")
if self.translator:
print(f"{Fore.RED}❌ {self.translator.get('email.failed_to_create_account', error=create_response.text)}{Style.RESET_ALL}")
else:
print(f"{Fore.RED}❌ 响应内容: {create_response.text}{Style.RESET_ALL}")
# If it's a domain problem, try the next available domain
if len(filtered_domains) > 1 and ("domain" in create_response.text.lower() or "address" in create_response.text.lower()):
print(f"{Fore.YELLOW}⚠️ 尝试使用下一个可用域名...{Style.RESET_ALL}")
# Add current domain to blocked list
if selected_domain not in self.blocked_domains:
self.blocked_domains.append(selected_domain)
# Recursively call yourself
return self.create_email()
raise Exception(f"{self.translator.get('email.failed_to_create_account') if self.translator else '创建账户失败'}")
except Exception as e:
if self.translator:
print(f"{Fore.RED}❌ {self.translator.get('email.failed_to_create_account', error=str(e))}{Style.RESET_ALL}")
else:
print(f"{Fore.RED}❌ 创建账户时出错: {str(e)}{Style.RESET_ALL}")
raise
# Get access token
try:
token_data = {
"address": email,
"password": password
}
token_response = requests.post(f"{self.api_url}/token", json=token_data, timeout=10)
if token_response.status_code != 200:
if self.translator:
print(f"{Fore.RED}❌ {self.translator.get('email.failed_to_get_access_token', error=token_response.status_code)}{Style.RESET_ALL}")
else:
print(f"{Fore.RED}❌ 获取令牌失败: 状态码 {token_response.status_code}{Style.RESET_ALL}")
if self.translator:
print(f"{Fore.RED}❌ {self.translator.get('email.failed_to_get_access_token', error=token_response.text)}{Style.RESET_ALL}")
else:
print(f"{Fore.RED}❌ 响应内容: {token_response.text}{Style.RESET_ALL}")
raise Exception(f"{self.translator.get('email.failed_to_get_access_token') if self.translator else '获取访问令牌失败'}")
self.token = token_response.json()["token"]
self.email = email
except Exception as e:
print(f"{Fore.RED}❌ 获取令牌时出错: {str(e)}{Style.RESET_ALL}")
raise
if self.translator:
print(f"{Fore.GREEN}✅ {self.translator.get('email.create_success')}: {email}{Style.RESET_ALL}")
else:
print(f"{Fore.GREEN}✅ 创建邮箱成功: {email}{Style.RESET_ALL}")
return email
except Exception as e:
if attempt < max_retries:
print(f"{Fore.YELLOW}⚠️ 尝试重新创建邮箱... (尝试 {attempt}/{max_retries}){Style.RESET_ALL}")
else:
if self.translator:
print(f"{Fore.RED}❌ {self.translator.get('email.create_error')}: {str(e)}{Style.RESET_ALL}")
else:
print(f"{Fore.RED}❌ 创建邮箱出错: {str(e)}{Style.RESET_ALL}")
return None
def close(self): def close(self):
"""close browser""" """close browser"""
if self.page: if self.page:
@ -263,11 +205,11 @@ class NewTempEmail:
else: else:
print(f"{Fore.CYAN}🔄 正在刷新邮箱...{Style.RESET_ALL}") print(f"{Fore.CYAN}🔄 正在刷新邮箱...{Style.RESET_ALL}")
# Use API to get latest email # click refresh button
headers = {"Authorization": f"Bearer {self.token}"} refresh_button = self.page.ele('xpath://button[@id="refresh"]')
response = requests.get(f"{self.api_url}/messages", headers=headers) if refresh_button:
refresh_button.click()
if response.status_code == 200: time.sleep(2) # wait for refresh to complete
if self.translator: if self.translator:
print(f"{Fore.GREEN}✅ {self.translator.get('email.refresh_success')}{Style.RESET_ALL}") print(f"{Fore.GREEN}✅ {self.translator.get('email.refresh_success')}{Style.RESET_ALL}")
else: else:
@ -275,9 +217,9 @@ class NewTempEmail:
return True return True
if self.translator: if self.translator:
print(f"{Fore.RED}❌ {self.translator.get('email.refresh_failed')}{Style.RESET_ALL}") print(f"{Fore.RED}❌ {self.translator.get('email.refresh_button_not_found')}{Style.RESET_ALL}")
else: else:
print(f"{Fore.RED}❌ 刷新邮箱失败{Style.RESET_ALL}") print(f"{Fore.RED}❌ 未找到刷新按钮{Style.RESET_ALL}")
return False return False
except Exception as e: except Exception as e:
@ -288,26 +230,19 @@ class NewTempEmail:
return False return False
def check_for_cursor_email(self): def check_for_cursor_email(self):
"""Check if there is a Cursor verification email""" """检查是否有 Cursor 的验证邮件"""
try: try:
# Use API to get email list # find verification email - use more accurate selector
headers = {"Authorization": f"Bearer {self.token}"} email_div = self.page.ele('xpath://div[contains(@class, "p-2") and contains(@class, "cursor-pointer") and contains(@class, "bg-white") and contains(@class, "shadow") and .//b[text()="no-reply@cursor.sh"] and .//span[text()="Verify your email address"]]')
response = requests.get(f"{self.api_url}/messages", headers=headers) if email_div:
if self.translator:
if response.status_code == 200: print(f"{Fore.GREEN}✅ {self.translator.get('email.verification_found')}{Style.RESET_ALL}")
messages = response.json()["hydra:member"] else:
for message in messages: print(f"{Fore.GREEN}✅ 找到验证邮件{Style.RESET_ALL}")
if message["from"]["address"] == "no-reply@cursor.sh" and "Verify your email address" in message["subject"]: # use JavaScript to click element
# Get email content self.page.run_js('arguments[0].click()', email_div)
message_id = message["id"] time.sleep(2) # wait for email content to load
message_response = requests.get(f"{self.api_url}/messages/{message_id}", headers=headers) return True
if message_response.status_code == 200:
if self.translator:
print(f"{Fore.GREEN}✅ {self.translator.get('email.verification_found')}{Style.RESET_ALL}")
else:
print(f"{Fore.GREEN}✅ 找到验证邮件{Style.RESET_ALL}")
return True
if self.translator: if self.translator:
print(f"{Fore.YELLOW}⚠️ {self.translator.get('email.verification_not_found')}{Style.RESET_ALL}") print(f"{Fore.YELLOW}⚠️ {self.translator.get('email.verification_not_found')}{Style.RESET_ALL}")
else: else:
@ -321,75 +256,31 @@ class NewTempEmail:
print(f"{Fore.RED}❌ 检查验证邮件出错: {str(e)}{Style.RESET_ALL}") print(f"{Fore.RED}❌ 检查验证邮件出错: {str(e)}{Style.RESET_ALL}")
return False return False
def get_verification_code(self, max_retries=3): def get_verification_code(self):
"""get verification code with retry mechanism""" """获取验证码"""
for attempt in range(1, max_retries + 1): try:
try: # find verification code element
# Check if token is valid code_element = self.page.ele('xpath://td//div[contains(@style, "font-size:28px") and contains(@style, "letter-spacing:2px")]')
if not self.token: if code_element:
code = code_element.text.strip()
if code.isdigit() and len(code) == 6:
if self.translator: if self.translator:
print(f"{Fore.YELLOW}⚠️ {self.translator.get('email.no_token_retry')}{Style.RESET_ALL}") print(f"{Fore.GREEN}✅ {self.translator.get('email.verification_code_found')}: {code}{Style.RESET_ALL}")
else: else:
print(f"{Fore.YELLOW}⚠️ 未获取到有效令牌，尝试重新创建邮箱... (尝试 {attempt}/{max_retries}){Style.RESET_ALL}") print(f"{Fore.GREEN}✅ 获取验证码成功: {code}{Style.RESET_ALL}")
return code
# Try to recreate email if self.translator:
self.create_email() print(f"{Fore.YELLOW}⚠️ {self.translator.get('email.verification_code_not_found')}{Style.RESET_ALL}")
if not self.token: else:
continue # Skip to next attempt if still no token print(f"{Fore.YELLOW}⚠️ 未找到有效的验证码{Style.RESET_ALL}")
return None
# Use API to get email list
headers = {"Authorization": f"Bearer {self.token}"} except Exception as e:
response = requests.get(f"{self.api_url}/messages", headers=headers) if self.translator:
print(f"{Fore.RED}❌ {self.translator.get('email.verification_code_error')}: {str(e)}{Style.RESET_ALL}")
if response.status_code == 200: else:
messages = response.json()["hydra:member"] print(f"{Fore.RED}❌ 获取验证码出错: {str(e)}{Style.RESET_ALL}")
for message in messages: return None
if message["from"]["address"] == "no-reply@cursor.sh" and "Verify your email address" in message["subject"]:
# Get email content
message_id = message["id"]
message_response = requests.get(f"{self.api_url}/messages/{message_id}", headers=headers)
if message_response.status_code == 200:
# Extract verification code from email content
email_content = message_response.json()["text"]
# Find 6-digit verification code
import re
code_match = re.search(r'\b\d{6}\b', email_content)
if code_match:
code = code_match.group(0)
if self.translator:
print(f"{Fore.GREEN}✅ {self.translator.get('email.verification_code_found')}: {code}{Style.RESET_ALL}")
else:
print(f"{Fore.GREEN}✅ 获取验证码成功: {code}{Style.RESET_ALL}")
return code
if attempt < max_retries:
wait_time = 10 * attempt # Increase wait time with each attempt
if self.translator:
print(f"{Fore.YELLOW}⚠️ {self.translator.get('email.verification_code_retry', attempt=attempt, max=max_retries, wait=wait_time)}{Style.RESET_ALL}")
else:
print(f"{Fore.YELLOW}⚠️ 未找到有效的验证码，将在 {wait_time} 秒后重试... (尝试 {attempt}/{max_retries}){Style.RESET_ALL}")
time.sleep(wait_time)
else:
if self.translator:
print(f"{Fore.RED}❌ {self.translator.get('email.verification_code_not_found')}{Style.RESET_ALL}")
else:
print(f"{Fore.RED}❌ 未找到有效的验证码{Style.RESET_ALL}")
except Exception as e:
if attempt < max_retries:
if self.translator:
print(f"{Fore.YELLOW}⚠️ {self.translator.get('email.verification_code_error_retry', error=str(e), attempt=attempt, max=max_retries)}{Style.RESET_ALL}")
else:
print(f"{Fore.YELLOW}⚠️ 获取验证码出错: {str(e)}，将重试... (尝试 {attempt}/{max_retries}){Style.RESET_ALL}")
time.sleep(get_random_wait_time(self.config, 'page_load_wait'))
else:
if self.translator:
print(f"{Fore.RED}❌ {self.translator.get('email.verification_code_error')}: {str(e)}{Style.RESET_ALL}")
else:
print(f"{Fore.RED}❌ 获取验证码出错: {str(e)}{Style.RESET_ALL}")
return None
def main(translator=None): def main(translator=None):
temp_email = NewTempEmail(translator) temp_email = NewTempEmail(translator)
@ -402,7 +293,7 @@ def main(translator=None):
else: else:
print(f"\n{Fore.CYAN}📧 临时邮箱地址: {email}{Style.RESET_ALL}") print(f"\n{Fore.CYAN}📧 临时邮箱地址: {email}{Style.RESET_ALL}")
# Test refresh function # test refresh function
while True: while True:
if translator: if translator:
choice = input(f"\n{translator.get('email.refresh_prompt')}: ").lower() choice = input(f"\n{translator.get('email.refresh_prompt')}: ").lower()
@ -417,4 +308,4 @@ def main(translator=None):
temp_email.close() temp_email.close()
if __name__ == "__main__": if __name__ == "__main__":
main() main()