yml2213andClaude Fable 5
a996028f5b
- 移除旧的 bdhg.xyz 等默认域名映射 - 未匹配的邮箱域名统一使用 111.229.206.54 IMAP代理 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
349 lines
12 KiB
Python
349 lines
12 KiB
Python
"""邮箱验证模块 - IMAP获取验证码"""
|
||
|
||
import imaplib
|
||
import email
|
||
import html
|
||
import re
|
||
import time
|
||
from datetime import datetime, timedelta
|
||
from email.header import decode_header
|
||
from email.utils import parsedate_to_datetime
|
||
from typing import Optional
|
||
from loguru import logger
|
||
|
||
|
||
class EmailVerifier:
|
||
"""邮箱验证器"""
|
||
|
||
def __init__(
|
||
self,
|
||
imap_server: str,
|
||
imap_port: int,
|
||
username: str,
|
||
password: str,
|
||
timeout: float = 12,
|
||
mailbox: str = "INBOX",
|
||
lookback_minutes: int = 10,
|
||
max_messages: int = 20,
|
||
):
|
||
self.imap_server = imap_server
|
||
self.imap_port = imap_port
|
||
self.username = username
|
||
self.password = password
|
||
self.timeout = timeout
|
||
self.mailbox = mailbox
|
||
self.lookback_minutes = lookback_minutes
|
||
self.max_messages = max_messages
|
||
self._connection: Optional[imaplib.IMAP4_SSL] = None
|
||
|
||
def connect(self) -> None:
|
||
"""连接IMAP服务器"""
|
||
try:
|
||
logger.info(f"连接IMAP服务器: {self.imap_server}:{self.imap_port}")
|
||
self._validate_login_text()
|
||
self._connection = imaplib.IMAP4_SSL(
|
||
self.imap_server,
|
||
self.imap_port,
|
||
timeout=self.timeout,
|
||
)
|
||
self._connection.login(self.username, self.password)
|
||
logger.info("IMAP连接成功")
|
||
except Exception as e:
|
||
logger.error(f"IMAP连接失败: {e}")
|
||
raise
|
||
|
||
def _validate_login_text(self) -> None:
|
||
"""提前检查IMAP登录字段，给出比ascii编码异常更明确的提示。"""
|
||
for label, value in (("邮箱账号", self.username), ("邮箱密码/授权码", self.password)):
|
||
try:
|
||
value.encode("ascii")
|
||
except UnicodeEncodeError as exc:
|
||
raise ValueError(
|
||
f"{label}包含中文或其他非ASCII字符，IMAP无法登录；"
|
||
"请检查导入格式是否为：用户名|密码|邮箱|邮箱密码"
|
||
) from exc
|
||
|
||
def disconnect(self) -> None:
|
||
"""断开IMAP连接"""
|
||
if self._connection:
|
||
try:
|
||
self._connection.logout()
|
||
except:
|
||
pass
|
||
self._connection = None
|
||
|
||
def get_verification_code(
|
||
self,
|
||
max_wait: int = 60,
|
||
interval: int = 2,
|
||
after_timestamp: Optional[float] = None,
|
||
allow_old_seconds: int = 15,
|
||
) -> str:
|
||
"""
|
||
获取斗鱼验证码
|
||
|
||
Args:
|
||
max_wait: 最大等待时间（秒）
|
||
interval: 轮询间隔（秒）
|
||
after_timestamp: 发起发送验证码请求的时间戳，用于过滤旧邮件
|
||
allow_old_seconds: 邮件服务器时间允许向前偏移的秒数
|
||
|
||
Returns:
|
||
6位验证码
|
||
"""
|
||
logger.info(f"等待斗鱼验证码邮件，最大等待 {max_wait} 秒...")
|
||
|
||
self.connect()
|
||
|
||
try:
|
||
deadline = time.monotonic() + max_wait
|
||
while time.monotonic() < deadline:
|
||
code = self._fetch_latest_code(
|
||
after_timestamp=after_timestamp,
|
||
allow_old_seconds=allow_old_seconds,
|
||
)
|
||
if code:
|
||
logger.success(f"获取到验证码: {code}")
|
||
return code
|
||
|
||
logger.debug("未找到验证码，等待中...")
|
||
time.sleep(interval)
|
||
|
||
raise TimeoutError("等待验证码超时")
|
||
finally:
|
||
self.disconnect()
|
||
|
||
def _fetch_latest_code(
|
||
self,
|
||
after_timestamp: Optional[float] = None,
|
||
allow_old_seconds: int = 15,
|
||
) -> Optional[str]:
|
||
"""从IMAP获取最新验证码"""
|
||
try:
|
||
if not self._connection:
|
||
raise RuntimeError("IMAP未连接")
|
||
|
||
status, _ = self._connection.select(self.mailbox, readonly=True)
|
||
if status != 'OK':
|
||
logger.warning(f"选择邮箱目录失败: {self.mailbox}")
|
||
return None
|
||
|
||
since = self._build_since_date(after_timestamp, allow_old_seconds)
|
||
# IMAP命令只能稳定发送ASCII条件，中文主题/发件人改到本地解析过滤。
|
||
status, messages = self._connection.search(None, 'SINCE', since)
|
||
|
||
if status != 'OK' or not messages[0]:
|
||
return None
|
||
|
||
message_ids = messages[0].split()
|
||
recent_ids = list(reversed(message_ids[-self.max_messages:]))
|
||
logger.debug(f"扫描最近 {len(recent_ids)} 封邮件，SINCE {since}")
|
||
|
||
for message_id in recent_ids:
|
||
msg = self._fetch_message(message_id)
|
||
if not msg:
|
||
continue
|
||
|
||
if not self._is_recent_enough(msg, after_timestamp, allow_old_seconds):
|
||
continue
|
||
|
||
subject = self._decode_subject(msg.get('Subject', ''))
|
||
from_addr = msg.get('From', '')
|
||
body = self._get_email_body(msg)
|
||
|
||
if not self._is_douyu_email(subject, from_addr, body):
|
||
continue
|
||
|
||
code = self._extract_verification_code(body)
|
||
if code:
|
||
logger.info(f"从邮件中提取到验证码，主题: {subject}")
|
||
return code
|
||
|
||
return None
|
||
|
||
except imaplib.IMAP4.abort as e:
|
||
logger.warning(f"IMAP连接中断，准备下轮重连: {e}")
|
||
self.disconnect()
|
||
self.connect()
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"获取邮件失败: {e}")
|
||
return None
|
||
|
||
def _build_since_date(
|
||
self,
|
||
after_timestamp: Optional[float],
|
||
allow_old_seconds: int,
|
||
) -> str:
|
||
"""构造IMAP SINCE日期，月份固定用英文缩写。"""
|
||
if after_timestamp:
|
||
since_dt = datetime.fromtimestamp(
|
||
max(0, after_timestamp - allow_old_seconds)
|
||
)
|
||
else:
|
||
since_dt = datetime.now() - timedelta(minutes=self.lookback_minutes)
|
||
|
||
return self._format_imap_date(since_dt)
|
||
|
||
def _format_imap_date(self, value: datetime) -> str:
|
||
"""格式化IMAP日期，避免系统locale影响月份名称。"""
|
||
months = [
|
||
"Jan", "Feb", "Mar", "Apr", "May", "Jun",
|
||
"Jul", "Aug", "Sep", "Oct", "Nov", "Dec",
|
||
]
|
||
return f"{value.day:02d}-{months[value.month - 1]}-{value.year}"
|
||
|
||
def _fetch_message(self, message_id: bytes) -> Optional[email.message.Message]:
|
||
"""获取邮件完整内容，使用PEEK避免标记已读。"""
|
||
status, msg_data = self._connection.fetch(message_id, '(BODY.PEEK[])')
|
||
if status != 'OK':
|
||
return None
|
||
|
||
raw_message = self._join_fetch_payload(msg_data)
|
||
if not raw_message:
|
||
return None
|
||
|
||
return email.message_from_bytes(raw_message)
|
||
|
||
def _join_fetch_payload(self, msg_data) -> bytes:
|
||
"""合并IMAP fetch返回中的邮件字节内容。"""
|
||
chunks = []
|
||
for item in msg_data:
|
||
if isinstance(item, tuple) and len(item) >= 2 and isinstance(item[1], bytes):
|
||
chunks.append(item[1])
|
||
elif isinstance(item, bytes) and item.startswith(b'From '):
|
||
chunks.append(item)
|
||
return b''.join(chunks)
|
||
|
||
def _is_recent_enough(
|
||
self,
|
||
msg: email.message.Message,
|
||
after_timestamp: Optional[float],
|
||
allow_old_seconds: int,
|
||
) -> bool:
|
||
"""检查邮件时间是否晚于本次发送验证码请求。"""
|
||
if not after_timestamp:
|
||
return True
|
||
|
||
message_time = self._parse_message_time(msg)
|
||
if not message_time:
|
||
logger.debug("邮件缺少Date头，保守纳入候选")
|
||
return True
|
||
|
||
if message_time.timestamp() < after_timestamp - allow_old_seconds:
|
||
subject = self._decode_subject(msg.get('Subject', ''))
|
||
logger.debug(f"跳过旧邮件: {subject}")
|
||
return False
|
||
|
||
return True
|
||
|
||
def _parse_message_time(self, msg: email.message.Message) -> Optional[datetime]:
|
||
"""解析邮件Date头。"""
|
||
date_header = msg.get('Date')
|
||
if not date_header:
|
||
return None
|
||
|
||
try:
|
||
return parsedate_to_datetime(date_header)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
def _is_douyu_email(self, subject: str, from_addr: str, body: str = "") -> bool:
|
||
"""判断是否是斗鱼的邮件"""
|
||
douyu_keywords = ['斗鱼', 'douyu', 'douyutv']
|
||
verify_keywords = ['验证码', '安全验证', '登录验证', '动态码', '校验码']
|
||
combined = f"{subject}\n{from_addr}\n{body[:500]}".lower()
|
||
|
||
return (
|
||
any(keyword in combined for keyword in douyu_keywords)
|
||
or any(keyword in subject for keyword in verify_keywords)
|
||
)
|
||
|
||
def _decode_subject(self, subject: str) -> str:
|
||
"""解码邮件主题"""
|
||
if not subject:
|
||
return ""
|
||
|
||
decoded_parts = decode_header(subject)
|
||
result = []
|
||
for part, charset in decoded_parts:
|
||
if isinstance(part, bytes):
|
||
result.append(part.decode(charset or 'utf-8', errors='ignore'))
|
||
else:
|
||
result.append(part)
|
||
return ' '.join(result)
|
||
|
||
def _get_email_body(self, msg: email.message.Message) -> str:
|
||
"""获取邮件正文"""
|
||
parts = []
|
||
|
||
if msg.is_multipart():
|
||
for part in msg.walk():
|
||
content_type = part.get_content_type()
|
||
if part.get_content_disposition() == 'attachment':
|
||
continue
|
||
|
||
if content_type == 'text/plain' or content_type == 'text/html':
|
||
decoded = self._decode_part_payload(part)
|
||
if decoded:
|
||
parts.append(decoded)
|
||
else:
|
||
decoded = self._decode_part_payload(msg)
|
||
if decoded:
|
||
parts.append(decoded)
|
||
|
||
return "\n".join(parts)
|
||
|
||
def _decode_part_payload(self, part: email.message.Message) -> str:
|
||
"""解码邮件片段内容。"""
|
||
try:
|
||
payload = part.get_payload(decode=True)
|
||
if payload is None:
|
||
raw_payload = part.get_payload()
|
||
return raw_payload if isinstance(raw_payload, str) else ""
|
||
|
||
charset = part.get_content_charset() or 'utf-8'
|
||
return payload.decode(charset, errors='ignore')
|
||
except Exception:
|
||
return ""
|
||
|
||
def _extract_verification_code(self, text: str) -> Optional[str]:
|
||
"""从文本中提取验证码"""
|
||
# 清理HTML标签和实体，方便匹配中文邮件模板。
|
||
text = html.unescape(text)
|
||
text = re.sub(r'<[^>]+>', ' ', text)
|
||
text = re.sub(r'\s+', ' ', text)
|
||
|
||
# 查找6位数字验证码
|
||
# 注意：中文冒号和英文冒号需要分别处理
|
||
patterns = [
|
||
r'(?:验证码|校验码|动态码|安全码)\s*(?:是|为|:|：)?\s*(\d{6})',
|
||
r'(?:您的|你的|本次)?(?:验证码|校验码|动态码|安全码)\s*(?:是|为)?\s*(\d{6})',
|
||
r'(\d{6})\s*(?:是|为)?(?:您的|你的|本次)?(?:验证码|校验码|动态码|安全码)',
|
||
r'(?:verification\s+code|security\s+code|code)\s*(?:is|:|：)?\s*(\d{6})',
|
||
r'\b(\d{6})\b', # 候选邮件已经过滤为斗鱼验证码邮件，最后再兜底匹配6位数字。
|
||
]
|
||
|
||
for pattern in patterns:
|
||
match = re.search(pattern, text, re.IGNORECASE)
|
||
if match:
|
||
code = match.group(1)
|
||
return code
|
||
|
||
return None
|
||
|
||
|
||
def get_email_config_for_account(email_address: str) -> dict:
|
||
"""根据邮箱地址自动配置IMAP服务器，默认使用统一IMAP代理"""
|
||
configs = {
|
||
'qq.com': {'server': 'imap.qq.com', 'port': 993},
|
||
'163.com': {'server': 'imap.163.com', 'port': 993},
|
||
'126.com': {'server': 'imap.126.com', 'port': 993},
|
||
'gmail.com': {'server': 'imap.gmail.com', 'port': 993},
|
||
'outlook.com': {'server': 'outlook.office365.com', 'port': 993},
|
||
'hotmail.com': {'server': 'outlook.office365.com', 'port': 993},
|
||
}
|
||
|
||
domain = email_address.split('@')[-1].lower()
|
||
return configs.get(domain, {'server': '111.229.206.54', 'port': 993})
|