get_email_code.py 5.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131
  1. import time
  2. import re
  3. from config import Config
  4. import requests
  5. class EmailVerificationHandler:
  6. def __init__(self):
  7. self.config = Config()
  8. self.session = requests.Session()
  9. self.base_url = "https://tm.ss5.xyz"
  10. self.headers = {
  11. 'accept': 'application/json, text/javascript, */*; q=0.01',
  12. 'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8',
  13. 'cache-control': 'no-cache',
  14. 'dnt': '1',
  15. 'pragma': 'no-cache',
  16. 'priority': 'u=1, i',
  17. 'referer': 'https://tm.ss5.xyz/index.html',
  18. 'sec-ch-ua': '"Google Chrome";v="131", "Chromium";v="131", "Not_A Brand";v="24"',
  19. 'sec-ch-ua-mobile': '?0',
  20. 'sec-ch-ua-platform': '"macOS"',
  21. 'sec-fetch-dest': 'empty',
  22. 'sec-fetch-mode': 'cors',
  23. 'sec-fetch-site': 'same-origin',
  24. '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',
  25. 'x-requested-with': 'XMLHttpRequest'
  26. }
  27. self.auth_token = self.config.get_auth_token()
  28. self.session.cookies.set('PHPSESSID', '967qb8iskkmkl2q56i554l5j3e')
  29. self.session.cookies.set('ysmail', self.auth_token)
  30. def get_verification_code(self, email=None):
  31. code = None
  32. try:
  33. print("正在处理...")
  34. # 使用传入的邮箱地址或从配置获取
  35. self.username = email if email else self.config.get_temp_mail()
  36. print("当前邮箱:", self.username)
  37. if self.username:
  38. # 使用邮箱地址获取验证码
  39. code = self._get_latest_mail_code()
  40. else:
  41. print("错误:未提供有效的邮箱地址")
  42. except Exception as e:
  43. print(f"获取验证码失败: {str(e)}")
  44. return code
  45. def _create_mailbox(self):
  46. url = f"{self.base_url}/user/user?a={self.auth_token}"
  47. response = self.session.get(url, headers=self.headers)
  48. if response.status_code != 200:
  49. raise Exception("创建邮箱失败")
  50. return response.json()
  51. def _get_email_list(self):
  52. url = f"{self.base_url}/user/lists?a={self.auth_token}"
  53. response = self.session.get(url, headers=self.headers)
  54. if response.status_code != 200:
  55. raise Exception("获取邮箱列表失败")
  56. data = response.json()
  57. return data.get('data', [])
  58. def _get_latest_mail_code(self):
  59. # 第一阶段:10次尝试,每次间隔3秒
  60. code = self._attempt_get_code(max_attempts=10, retry_interval=3)
  61. if code:
  62. return code
  63. print("第一阶段未获取到验证码,进入第二阶段重试...")
  64. # 第二阶段:100次尝试,每次间隔5秒
  65. return self._attempt_get_code(max_attempts=100, retry_interval=5)
  66. def _attempt_get_code(self, max_attempts, retry_interval):
  67. for attempt in range(max_attempts):
  68. url = f"{self.base_url}/mail/lists?u={self.username}&a={self.auth_token}"
  69. response = self.session.get(url, headers=self.headers)
  70. if response.status_code != 200:
  71. print(f"第 {attempt + 1} 次尝试获取邮件失败,{retry_interval} 秒后重试...")
  72. time.sleep(retry_interval)
  73. continue
  74. response_data = response.json()
  75. mail_list_data = response_data.get('data')
  76. # 检查返回的数据结构
  77. if not isinstance(mail_list_data, list) or not mail_list_data or response_data.get('code') != 0:
  78. print(f"第 {attempt + 1} 次尝试未收到邮件,{retry_interval} 秒后重试...")
  79. time.sleep(retry_interval)
  80. continue
  81. # 获取最新邮件内容
  82. try:
  83. latest_mail = mail_list_data[0]
  84. if not isinstance(latest_mail, dict):
  85. print(f"第 {attempt + 1} 次尝试邮件格式错误,{retry_interval} 秒后重试...")
  86. time.sleep(retry_interval)
  87. continue
  88. mail_text = latest_mail.get('content', '')
  89. # 移除所有 CSS 样式和 HTML 标签
  90. clean_text = re.sub(r'<style[^>]*>[^<]*</style>', '', mail_text)
  91. clean_text = re.sub(r'<[^>]+>', '', clean_text)
  92. # 移除多余的空白字符
  93. clean_text = re.sub(r'\s+', ' ', clean_text).strip()
  94. print("过滤后的邮件内容:", clean_text)
  95. # 从纯文本中提取 6 位数字验证码
  96. code_match = re.search(r'Your verification code is (\d{6})', clean_text)
  97. if code_match:
  98. return code_match.group(1)
  99. except Exception as e:
  100. print(f"处理邮件内容时出错: {str(e)}")
  101. time.sleep(retry_interval)
  102. continue
  103. print(f"第 {attempt + 1} 次尝试未找到验证码,{retry_interval} 秒后重试...")
  104. time.sleep(retry_interval)
  105. print(f"已达到最大尝试次数 ({max_attempts}),获取验证码失败")
  106. return None
  107. if __name__ == "__main__":
  108. email_handler = EmailVerificationHandler()
  109. code = email_handler.get_verification_code()
  110. print(code)