first commit: DuoPlus云手机协议注册工具 - 完整实现
This commit is contained in:
huangzhenpc
463
email_manager.py
Normal file
463
email_manager.py
Normal file
@@ -0,0 +1,463 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
"""
|
||||
自建邮箱管理模块
|
||||
支持创建邮箱用户和IMAP收件
|
||||
使用 Stalwart API
|
||||
"""
|
||||
|
||||
import imaplib
|
||||
import email
|
||||
import time
|
||||
import re
|
||||
import random
|
||||
import string
|
||||
import secrets
|
||||
from typing import Optional, List, Dict, Tuple
|
||||
from datetime import datetime, timedelta
|
||||
from colorama import init, Fore
|
||||
from stalwart_client import StalwartClient
|
||||
from exceptions import StalwartError, ApiError
|
||||
|
||||
# 初始化 colorama
|
||||
init(autoreset=True)
|
||||
|
||||
class EmailManager:
|
||||
"""邮箱管理器"""
|
||||
|
||||
def __init__(self, api_base_url: str = "https://mail.evnmail.com/api",
|
||||
api_key: str = "admin:Hunter1520.",
|
||||
imap_server: str = "mail.evnmail.com",
|
||||
imap_port: int = 993):
|
||||
"""
|
||||
初始化邮箱管理器
|
||||
|
||||
Args:
|
||||
api_base_url: Stalwart API基础URL
|
||||
api_key: API认证密钥 (格式: username:password)
|
||||
imap_server: IMAP服务器地址
|
||||
imap_port: IMAP端口
|
||||
"""
|
||||
self.api_base_url = api_base_url
|
||||
self.api_key = api_key
|
||||
self.imap_server = imap_server
|
||||
self.imap_port = imap_port
|
||||
|
||||
# 初始化 Stalwart 客户端
|
||||
self.client = StalwartClient(api_base_url, api_key=api_key)
|
||||
|
||||
def generate_random_password(self, length: int = 12) -> str:
|
||||
"""
|
||||
生成随机安全密码（只包含大小写字母和数字）
|
||||
|
||||
Args:
|
||||
length: 密码长度
|
||||
|
||||
Returns:
|
||||
随机密码
|
||||
"""
|
||||
characters = string.ascii_letters + string.digits
|
||||
password = [
|
||||
secrets.choice(string.ascii_uppercase),
|
||||
secrets.choice(string.ascii_lowercase),
|
||||
secrets.choice(string.digits),
|
||||
]
|
||||
password.extend(secrets.choice(characters) for _ in range(length - 3))
|
||||
random.shuffle(password)
|
||||
return ''.join(password)
|
||||
|
||||
def generate_random_username(self, prefix: str = "user", length: int = 8) -> str:
|
||||
"""
|
||||
生成随机用户名
|
||||
|
||||
Args:
|
||||
prefix: 用户名前缀
|
||||
length: 随机部分长度
|
||||
|
||||
Returns:
|
||||
随机用户名
|
||||
"""
|
||||
random_part = ''.join(random.choices(string.ascii_lowercase + string.digits, k=length))
|
||||
return f"{prefix}{random_part}"
|
||||
|
||||
def create_email_account(self, username: Optional[str] = None,
|
||||
domain: str = "cursor.edu.kg",
|
||||
password: Optional[str] = None,
|
||||
quota_mb: int = 10) -> Dict[str, str]:
|
||||
"""
|
||||
创建邮箱账户
|
||||
|
||||
Args:
|
||||
username: 用户名，如果为None则随机生成
|
||||
domain: 邮箱域名
|
||||
password: 密码，如果为None则随机生成
|
||||
quota_mb: 邮箱配额（MB）
|
||||
|
||||
Returns:
|
||||
包含邮箱信息的字典
|
||||
"""
|
||||
# 生成用户名和密码
|
||||
if not username:
|
||||
username = self.generate_random_username()
|
||||
if not password:
|
||||
password = self.generate_random_password()
|
||||
|
||||
email_address = f"{username}@{domain}"
|
||||
quota_bytes = quota_mb * 1024 * 1024
|
||||
|
||||
print(f"{Fore.CYAN}[Email] 创建邮箱账户: {email_address}")
|
||||
|
||||
# 准备用户数据
|
||||
user_data = {
|
||||
"type": "individual",
|
||||
"name": email_address, # 使用完整邮箱地址作为name
|
||||
"description": f"{email_address} account",
|
||||
"quota": quota_bytes,
|
||||
"emails": [email_address],
|
||||
"roles": ["user"],
|
||||
"memberOf": [],
|
||||
"secrets": [password] # 直接在创建时设置密码
|
||||
}
|
||||
|
||||
try:
|
||||
# 使用 Stalwart 客户端创建用户
|
||||
print(f"{Fore.YELLOW}[Email] 提交用户创建请求...")
|
||||
response = self.client.create_principal(user_data)
|
||||
|
||||
# 获取用户ID
|
||||
user_id = response.get('data') if isinstance(response, dict) else response
|
||||
|
||||
if user_id:
|
||||
print(f"{Fore.GREEN}[Email] ✅ 邮箱创建成功！ID: {user_id}")
|
||||
|
||||
# 验证用户信息
|
||||
try:
|
||||
user_info = self.client.fetch_principal(user_id)
|
||||
print(f"{Fore.GREEN}[Email] 验证完成")
|
||||
except Exception as e:
|
||||
print(f"{Fore.YELLOW}[Email] 验证失败，但用户已创建: {str(e)}")
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"user_id": user_id,
|
||||
"email": email_address,
|
||||
"username": username,
|
||||
"password": password,
|
||||
"domain": domain,
|
||||
"imap_server": self.imap_server,
|
||||
"imap_port": self.imap_port
|
||||
}
|
||||
else:
|
||||
print(f"{Fore.RED}[Email] 创建失败: 未返回用户ID")
|
||||
return {"success": False, "message": "No user ID returned"}
|
||||
|
||||
except ApiError as e:
|
||||
print(f"{Fore.RED}[Email] API错误: {e}")
|
||||
return {"success": False, "message": f"API Error: {str(e)}"}
|
||||
except StalwartError as e:
|
||||
print(f"{Fore.RED}[Email] Stalwart错误: {e}")
|
||||
return {"success": False, "message": f"Stalwart Error: {str(e)}"}
|
||||
except Exception as e:
|
||||
print(f"{Fore.RED}[Email] 创建异常: {str(e)}")
|
||||
return {"success": False, "message": str(e)}
|
||||
|
||||
def connect_imap(self, email_address: str, password: str) -> Optional[imaplib.IMAP4_SSL]:
|
||||
"""
|
||||
连接到IMAP服务器
|
||||
|
||||
Args:
|
||||
email_address: 邮箱地址
|
||||
password: 密码
|
||||
|
||||
Returns:
|
||||
IMAP连接对象
|
||||
"""
|
||||
try:
|
||||
print(f"{Fore.YELLOW}[IMAP] 连接到 {self.imap_server}:{self.imap_port}")
|
||||
|
||||
# 连接到IMAP服务器
|
||||
imap = imaplib.IMAP4_SSL(self.imap_server, self.imap_port)
|
||||
|
||||
# 登录
|
||||
imap.login(email_address, password)
|
||||
print(f"{Fore.GREEN}[IMAP] 登录成功")
|
||||
|
||||
return imap
|
||||
|
||||
except imaplib.IMAP4.error as e:
|
||||
print(f"{Fore.RED}[IMAP] 登录失败: {str(e)}")
|
||||
return None
|
||||
except Exception as e:
|
||||
print(f"{Fore.RED}[IMAP] 连接失败: {str(e)}")
|
||||
return None
|
||||
|
||||
def fetch_verification_code(self, email_address: str, password: str,
|
||||
sender_filter: str = "@duoplus",
|
||||
timeout: int = 60,
|
||||
check_interval: int = 5) -> Optional[str]:
|
||||
"""
|
||||
从邮箱获取验证码
|
||||
|
||||
Args:
|
||||
email_address: 邮箱地址
|
||||
password: 密码
|
||||
sender_filter: 发件人过滤条件
|
||||
timeout: 超时时间（秒）
|
||||
check_interval: 检查间隔（秒）
|
||||
|
||||
Returns:
|
||||
验证码字符串
|
||||
"""
|
||||
print(f"{Fore.CYAN}[IMAP] 等待接收验证码邮件...")
|
||||
|
||||
start_time = time.time()
|
||||
|
||||
while time.time() - start_time < timeout:
|
||||
imap = self.connect_imap(email_address, password)
|
||||
if not imap:
|
||||
time.sleep(check_interval)
|
||||
continue
|
||||
|
||||
try:
|
||||
# 选择收件箱
|
||||
imap.select('INBOX')
|
||||
|
||||
# 搜索未读邮件
|
||||
search_criteria = f'(UNSEEN FROM "{sender_filter}")'
|
||||
result, data = imap.search(None, search_criteria)
|
||||
|
||||
if result == 'OK' and data[0]:
|
||||
email_ids = data[0].split()
|
||||
|
||||
# 获取最新的邮件
|
||||
for email_id in reversed(email_ids):
|
||||
result, data = imap.fetch(email_id, '(RFC822)')
|
||||
|
||||
if result == 'OK':
|
||||
raw_email = data[0][1]
|
||||
msg = email.message_from_bytes(raw_email)
|
||||
|
||||
# 获取邮件内容
|
||||
body = self._get_email_body(msg)
|
||||
|
||||
# 提取验证码（通常是6位数字）
|
||||
code_patterns = [
|
||||
r'验证码[：:]\s*(\d{6})',
|
||||
r'验证码是[：:]\s*(\d{6})',
|
||||
r'您的验证码是[：:]\s*(\d{6})',
|
||||
r'(\d{6})', # 直接匹配6位数字
|
||||
]
|
||||
|
||||
for pattern in code_patterns:
|
||||
match = re.search(pattern, body)
|
||||
if match:
|
||||
code = match.group(1)
|
||||
print(f"{Fore.GREEN}[IMAP] ✅ 获取到验证码: {code}")
|
||||
|
||||
# 标记邮件为已读
|
||||
imap.store(email_id, '+FLAGS', '\\Seen')
|
||||
|
||||
imap.close()
|
||||
imap.logout()
|
||||
return code
|
||||
|
||||
imap.close()
|
||||
imap.logout()
|
||||
|
||||
except Exception as e:
|
||||
print(f"{Fore.YELLOW}[IMAP] 检查邮件时出错: {str(e)}")
|
||||
if imap:
|
||||
try:
|
||||
imap.logout()
|
||||
except:
|
||||
pass
|
||||
|
||||
# 等待下次检查
|
||||
elapsed = int(time.time() - start_time)
|
||||
remaining = timeout - elapsed
|
||||
print(f"{Fore.YELLOW}[IMAP] 暂未收到验证码，{remaining}秒后超时...")
|
||||
time.sleep(check_interval)
|
||||
|
||||
print(f"{Fore.RED}[IMAP] 获取验证码超时")
|
||||
return None
|
||||
|
||||
def _get_email_body(self, msg) -> str:
|
||||
"""
|
||||
提取邮件正文
|
||||
|
||||
Args:
|
||||
msg: email.message对象
|
||||
|
||||
Returns:
|
||||
邮件正文文本
|
||||
"""
|
||||
body = ""
|
||||
|
||||
if msg.is_multipart():
|
||||
for part in msg.walk():
|
||||
content_type = part.get_content_type()
|
||||
content_disposition = str(part.get("Content-Disposition"))
|
||||
|
||||
if content_type == "text/plain" and "attachment" not in content_disposition:
|
||||
try:
|
||||
body = part.get_payload(decode=True).decode('utf-8', errors='ignore')
|
||||
break
|
||||
except:
|
||||
pass
|
||||
elif content_type == "text/html" and not body:
|
||||
try:
|
||||
html_body = part.get_payload(decode=True).decode('utf-8', errors='ignore')
|
||||
# 简单移除HTML标签
|
||||
body = re.sub('<[^<]+?>', '', html_body)
|
||||
except:
|
||||
pass
|
||||
else:
|
||||
try:
|
||||
body = msg.get_payload(decode=True).decode('utf-8', errors='ignore')
|
||||
except:
|
||||
body = str(msg.get_payload())
|
||||
|
||||
return body
|
||||
|
||||
def get_recent_emails(self, email_address: str, password: str,
|
||||
count: int = 5) -> List[Dict[str, str]]:
|
||||
"""
|
||||
获取最近的邮件列表
|
||||
|
||||
Args:
|
||||
email_address: 邮箱地址
|
||||
password: 密码
|
||||
count: 获取邮件数量
|
||||
|
||||
Returns:
|
||||
邮件列表
|
||||
"""
|
||||
emails = []
|
||||
|
||||
imap = self.connect_imap(email_address, password)
|
||||
if not imap:
|
||||
return emails
|
||||
|
||||
try:
|
||||
# 选择收件箱
|
||||
imap.select('INBOX')
|
||||
|
||||
# 搜索所有邮件
|
||||
result, data = imap.search(None, 'ALL')
|
||||
|
||||
if result == 'OK' and data[0]:
|
||||
email_ids = data[0].split()
|
||||
|
||||
# 获取最近的几封邮件
|
||||
for email_id in reversed(email_ids[-count:]):
|
||||
result, data = imap.fetch(email_id, '(RFC822)')
|
||||
|
||||
if result == 'OK':
|
||||
raw_email = data[0][1]
|
||||
msg = email.message_from_bytes(raw_email)
|
||||
|
||||
emails.append({
|
||||
"id": email_id.decode(),
|
||||
"from": msg.get('From', ''),
|
||||
"to": msg.get('To', ''),
|
||||
"subject": msg.get('Subject', ''),
|
||||
"date": msg.get('Date', ''),
|
||||
"body": self._get_email_body(msg)[:200] + "..."
|
||||
})
|
||||
|
||||
imap.close()
|
||||
imap.logout()
|
||||
|
||||
except Exception as e:
|
||||
print(f"{Fore.RED}[IMAP] 获取邮件列表失败: {str(e)}")
|
||||
if imap:
|
||||
try:
|
||||
imap.logout()
|
||||
except:
|
||||
pass
|
||||
|
||||
return emails
|
||||
|
||||
|
||||
class AutoEmailVerification:
|
||||
"""自动邮箱验证码获取"""
|
||||
|
||||
def __init__(self, email_manager: EmailManager):
|
||||
self.email_manager = email_manager
|
||||
|
||||
def create_temp_email(self, domain: str = "cursor.edu.kg") -> Dict[str, str]:
|
||||
"""
|
||||
创建临时邮箱
|
||||
|
||||
Args:
|
||||
domain: 邮箱域名
|
||||
|
||||
Returns:
|
||||
邮箱信息
|
||||
"""
|
||||
return self.email_manager.create_email_account(domain=domain)
|
||||
|
||||
def wait_for_code(self, email_info: Dict[str, str],
|
||||
sender_filter: str = "@duoplus",
|
||||
timeout: int = 60) -> Optional[str]:
|
||||
"""
|
||||
等待并获取验证码
|
||||
|
||||
Args:
|
||||
email_info: 邮箱信息字典
|
||||
sender_filter: 发件人过滤
|
||||
timeout: 超时时间
|
||||
|
||||
Returns:
|
||||
验证码
|
||||
"""
|
||||
if not email_info.get("success"):
|
||||
print(f"{Fore.RED}[Auto] 邮箱信息无效")
|
||||
return None
|
||||
|
||||
return self.email_manager.fetch_verification_code(
|
||||
email_info["email"],
|
||||
email_info["password"],
|
||||
sender_filter=sender_filter,
|
||||
timeout=timeout
|
||||
)
|
||||
|
||||
|
||||
# 测试函数
|
||||
def test_email_system():
|
||||
"""测试邮箱系统"""
|
||||
print(f"{Fore.CYAN}{'='*60}")
|
||||
print(f"{Fore.CYAN}测试自建邮箱系统")
|
||||
print(f"{Fore.CYAN}{'='*60}\n")
|
||||
|
||||
# 创建邮箱管理器
|
||||
manager = EmailManager()
|
||||
|
||||
# 创建测试邮箱
|
||||
email_info = manager.create_email_account(
|
||||
username=None, # 自动生成
|
||||
domain="cursor.edu.kg",
|
||||
password=None, # 自动生成
|
||||
quota_mb=10
|
||||
)
|
||||
|
||||
if email_info.get("success"):
|
||||
print(f"\n{Fore.GREEN}邮箱创建成功！")
|
||||
print(f"{Fore.GREEN}邮箱地址: {email_info['email']}")
|
||||
print(f"{Fore.GREEN}密码: {email_info['password']}")
|
||||
|
||||
# 测试IMAP连接
|
||||
print(f"\n{Fore.CYAN}测试IMAP连接...")
|
||||
imap = manager.connect_imap(email_info['email'], email_info['password'])
|
||||
if imap:
|
||||
print(f"{Fore.GREEN}IMAP连接成功！")
|
||||
imap.logout()
|
||||
else:
|
||||
print(f"{Fore.RED}IMAP连接失败")
|
||||
else:
|
||||
print(f"{Fore.RED}邮箱创建失败: {email_info.get('message')}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_email_system()
|
||||
Reference in New Issue
Block a user