|
|
@@ -6,81 +6,107 @@ import requests
|
|
|
|
|
|
class EmailVerificationHandler:
|
|
|
def __init__(self):
|
|
|
- self.username = Config().get_temp_mail()
|
|
|
+ self.config = Config()
|
|
|
self.session = requests.Session()
|
|
|
- self.emailExtension = "@mailto.plus"
|
|
|
+ self.base_url = "https://tm.ss5.xyz"
|
|
|
+ self.headers = {
|
|
|
+ 'accept': 'application/json, text/javascript, */*; q=0.01',
|
|
|
+ 'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8',
|
|
|
+ 'cache-control': 'no-cache',
|
|
|
+ 'dnt': '1',
|
|
|
+ 'pragma': 'no-cache',
|
|
|
+ 'priority': 'u=1, i',
|
|
|
+ 'referer': 'https://tm.ss5.xyz/index.html',
|
|
|
+ 'sec-ch-ua': '"Google Chrome";v="131", "Chromium";v="131", "Not_A Brand";v="24"',
|
|
|
+ 'sec-ch-ua-mobile': '?0',
|
|
|
+ 'sec-ch-ua-platform': '"macOS"',
|
|
|
+ 'sec-fetch-dest': 'empty',
|
|
|
+ 'sec-fetch-mode': 'cors',
|
|
|
+ 'sec-fetch-site': 'same-origin',
|
|
|
+ 'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36',
|
|
|
+ 'x-requested-with': 'XMLHttpRequest'
|
|
|
+ }
|
|
|
+ self.auth_token = self.config.get_auth_token()
|
|
|
+ self.session.cookies.set('PHPSESSID', '967qb8iskkmkl2q56i554l5j3e')
|
|
|
+ self.session.cookies.set('ysmail', self.auth_token)
|
|
|
|
|
|
- def get_verification_code(self):
|
|
|
+ def get_verification_code(self, email=None):
|
|
|
code = None
|
|
|
|
|
|
try:
|
|
|
print("正在处理...")
|
|
|
-
|
|
|
- # 等待并获取最新邮件
|
|
|
- code, first_id = self._get_latest_mail_code()
|
|
|
-
|
|
|
- # 清理邮件
|
|
|
- self._cleanup_mail(first_id)
|
|
|
+ # 使用传入的邮箱地址或从配置获取
|
|
|
+ self.username = email if email else self.config.get_temp_mail()
|
|
|
+ print("当前邮箱:", self.username)
|
|
|
+
|
|
|
+ if self.username:
|
|
|
+ # 使用邮箱地址获取验证码
|
|
|
+ code = self._get_latest_mail_code()
|
|
|
+ else:
|
|
|
+ print("错误:未提供有效的邮箱地址")
|
|
|
|
|
|
except Exception as e:
|
|
|
print(f"获取验证码失败: {str(e)}")
|
|
|
|
|
|
return code
|
|
|
|
|
|
- # 手动输入验证码
|
|
|
- def _get_latest_mail_code(self):
|
|
|
- # 获取邮件列表
|
|
|
- mail_list_url = f"https://tempmail.plus/api/mails?email={self.username}{self.emailExtension}&limit=20&epin="
|
|
|
- mail_list_response = self.session.get(mail_list_url)
|
|
|
- mail_list_data = mail_list_response.json()
|
|
|
- time.sleep(0.5)
|
|
|
- if not mail_list_data.get("result"):
|
|
|
- return None, None
|
|
|
-
|
|
|
- # 获取最新邮件的ID
|
|
|
- first_id = mail_list_data.get("first_id")
|
|
|
- if not first_id:
|
|
|
- return None, None
|
|
|
-
|
|
|
- # 获取具体邮件内容
|
|
|
- mail_detail_url = f"https://tempmail.plus/api/mails/{first_id}?email={self.username}{self.emailExtension}&epin="
|
|
|
- mail_detail_response = self.session.get(mail_detail_url)
|
|
|
- mail_detail_data = mail_detail_response.json()
|
|
|
- time.sleep(0.5)
|
|
|
- if not mail_detail_data.get("result"):
|
|
|
- return None, None
|
|
|
-
|
|
|
- # 从邮件文本中提取6位数字验证码
|
|
|
- mail_text = mail_detail_data.get("text", "")
|
|
|
- code_match = re.search(r"\b\d{6}\b", mail_text)
|
|
|
-
|
|
|
- if code_match:
|
|
|
- return code_match.group(), first_id
|
|
|
- return None, None
|
|
|
-
|
|
|
- def _cleanup_mail(self, first_id):
|
|
|
- # 构造删除请求的URL和数据
|
|
|
- delete_url = "https://tempmail.plus/api/mails/"
|
|
|
- payload = {
|
|
|
- "email": f"{self.username}{self.emailExtension}",
|
|
|
- "first_id": first_id,
|
|
|
- "epin": "",
|
|
|
- }
|
|
|
-
|
|
|
- # 最多尝试5次
|
|
|
- for _ in range(5):
|
|
|
- response = self.session.delete(delete_url, data=payload)
|
|
|
+ def _create_mailbox(self):
|
|
|
+ url = f"{self.base_url}/user/user?a={self.auth_token}"
|
|
|
+ response = self.session.get(url, headers=self.headers)
|
|
|
+ if response.status_code != 200:
|
|
|
+ raise Exception("创建邮箱失败")
|
|
|
+ return response.json()
|
|
|
+
|
|
|
+ def _get_email_list(self):
|
|
|
+ url = f"{self.base_url}/user/lists?a={self.auth_token}"
|
|
|
+ response = self.session.get(url, headers=self.headers)
|
|
|
+ if response.status_code != 200:
|
|
|
+ raise Exception("获取邮箱列表失败")
|
|
|
+ data = response.json()
|
|
|
+ return data.get('data', [])
|
|
|
+
|
|
|
+ def _get_latest_mail_code(self, max_attempts=10, retry_interval=3):
|
|
|
+ for attempt in range(max_attempts):
|
|
|
+ url = f"{self.base_url}/mail/lists?u={self.username}&a={self.auth_token}"
|
|
|
+ response = self.session.get(url, headers=self.headers)
|
|
|
+ if response.status_code != 200:
|
|
|
+ print(f"第 {attempt + 1} 次尝试获取邮件失败,{retry_interval} 秒后重试...")
|
|
|
+ time.sleep(retry_interval)
|
|
|
+ continue
|
|
|
+
|
|
|
+ response_data = response.json()
|
|
|
+ mail_list_data = response_data.get('data')
|
|
|
+
|
|
|
+ # 检查返回的数据结构
|
|
|
+ if not isinstance(mail_list_data, list) or not mail_list_data or response_data.get('code') != 0:
|
|
|
+ print(f"第 {attempt + 1} 次尝试未收到邮件,{retry_interval} 秒后重试...")
|
|
|
+ time.sleep(retry_interval)
|
|
|
+ continue
|
|
|
+
|
|
|
+ # 获取最新邮件内容
|
|
|
try:
|
|
|
- result = response.json().get("result")
|
|
|
- if result is True:
|
|
|
- return True
|
|
|
- except:
|
|
|
- pass
|
|
|
-
|
|
|
- # 如果失败,等待0.5秒后重试
|
|
|
- time.sleep(0.5)
|
|
|
-
|
|
|
- return False
|
|
|
+ latest_mail = mail_list_data[0]
|
|
|
+ if not isinstance(latest_mail, dict):
|
|
|
+ print(f"第 {attempt + 1} 次尝试邮件格式错误,{retry_interval} 秒后重试...")
|
|
|
+ time.sleep(retry_interval)
|
|
|
+ continue
|
|
|
+
|
|
|
+ mail_text = latest_mail.get('content', '')
|
|
|
+
|
|
|
+ # 从邮件文本中提取指定格式的验证码
|
|
|
+ code_match = re.search(r">(\d{6})<", mail_text)
|
|
|
+ if code_match:
|
|
|
+ return code_match.group(1)
|
|
|
+ except Exception as e:
|
|
|
+ print(f"处理邮件内容时出错: {str(e)}")
|
|
|
+ time.sleep(retry_interval)
|
|
|
+ continue
|
|
|
+
|
|
|
+ print(f"第 {attempt + 1} 次尝试未找到验证码,{retry_interval} 秒后重试...")
|
|
|
+ time.sleep(retry_interval)
|
|
|
+
|
|
|
+ print(f"已达到最大尝试次数 ({max_attempts}),获取验证码失败")
|
|
|
+ return None
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|