yml2213
d47b8e24fa
- 通过 http://111.229.206.54:8000/read.php 读取邮件，不再依赖IMAP - 解决IMAP密码错误/授权码过期导致45%登录失败的问题 - 错误密码不再抛异常，返回None由上层处理 - 兼容原有EmailVerifier接口签名，login.py无需改动
198 lines
6.6 KiB
Python
198 lines
6.6 KiB
Python
"""邮箱验证模块 - 通过 HTTP 接口获取验证码（替代 IMAP）"""
|
||
|
||
import html
|
||
import re
|
||
import time
|
||
from typing import Optional
|
||
from loguru import logger
|
||
import requests
|
||
|
||
|
||
# HTTP 读邮件接口地址
|
||
MAIL_API_URL = "http://111.229.206.54:8000/read.php"
|
||
|
||
|
||
class EmailVerifier:
|
||
"""邮箱验证器（HTTP 方式）"""
|
||
|
||
def __init__(
|
||
self,
|
||
imap_server: str = "",
|
||
imap_port: int = 143,
|
||
username: str = "",
|
||
password: str = "",
|
||
timeout: float = 10,
|
||
mailbox: str = "INBOX",
|
||
lookback_minutes: int = 10,
|
||
max_messages: int = 20,
|
||
use_ssl: bool = False,
|
||
mail_api_url: str = MAIL_API_URL,
|
||
):
|
||
# 兼容旧参数名，HTTP 模式下 imap_* 参数仅用于日志
|
||
self.imap_server = imap_server
|
||
self.imap_port = imap_port
|
||
self.username = username # 邮箱地址
|
||
self.password = password # 邮箱密码
|
||
self.timeout = timeout
|
||
self.mail_api_url = mail_api_url
|
||
self.use_ssl = use_ssl
|
||
|
||
def connect(self) -> None:
|
||
"""兼容接口：HTTP 模式下无需持久连接"""
|
||
pass
|
||
|
||
def disconnect(self) -> None:
|
||
"""兼容接口"""
|
||
pass
|
||
|
||
def get_verification_code(
|
||
self,
|
||
max_wait: int = 60,
|
||
interval: int = 3,
|
||
after_timestamp: Optional[float] = None,
|
||
allow_old_seconds: int = 15,
|
||
) -> str:
|
||
"""
|
||
轮询 HTTP 读邮件接口获取斗鱼验证码。
|
||
|
||
Args:
|
||
max_wait: 最大等待时间（秒）
|
||
interval: 轮询间隔（秒）
|
||
after_timestamp: 发送验证码请求的时间戳，用于过滤旧邮件
|
||
allow_old_seconds: 允许的时间偏移（秒）
|
||
|
||
Returns:
|
||
6位验证码
|
||
"""
|
||
logger.info(f"等待斗鱼验证码邮件，最大等待 {max_wait} 秒...")
|
||
|
||
deadline = time.monotonic() + max_wait
|
||
last_error = ""
|
||
|
||
while time.monotonic() < deadline:
|
||
try:
|
||
code = self._fetch_code_via_http(after_timestamp, allow_old_seconds)
|
||
if code:
|
||
logger.success(f"获取到验证码: {code}")
|
||
return code
|
||
logger.debug("未找到验证码，等待中...")
|
||
except Exception as e:
|
||
last_error = str(e)
|
||
logger.warning(f"读取邮件异常: {e}")
|
||
time.sleep(interval)
|
||
|
||
raise TimeoutError(f"等待验证码超时{'（' + last_error + '）' if last_error else ''}")
|
||
|
||
def _fetch_code_via_http(
|
||
self,
|
||
after_timestamp: Optional[float] = None,
|
||
allow_old_seconds: int = 15,
|
||
) -> Optional[str]:
|
||
"""通过 HTTP 接口读取最新邮件并提取验证码"""
|
||
try:
|
||
response = requests.get(
|
||
self.mail_api_url,
|
||
params={
|
||
"yhm": self.username,
|
||
"mm": self.password,
|
||
},
|
||
timeout=self.timeout,
|
||
)
|
||
response.raise_for_status()
|
||
content = response.text
|
||
|
||
# 检查是否是斗鱼验证码邮件
|
||
if not self._is_douyu_email(content):
|
||
logger.debug("最新邮件不是斗鱼验证码邮件")
|
||
return None
|
||
|
||
# 检查邮件时间是否足够新（发送验证码请求之后）
|
||
if after_timestamp and not self._is_email_recent(content, after_timestamp, allow_old_seconds):
|
||
logger.debug("邮件时间早于验证码发送时间，跳过")
|
||
return None
|
||
|
||
# 提取验证码
|
||
code = self._extract_verification_code(content)
|
||
if code:
|
||
logger.info(f"从邮件中提取到验证码: {code}")
|
||
return code
|
||
|
||
return None
|
||
|
||
except requests.RequestException as e:
|
||
logger.warning(f"HTTP读邮件请求失败: {e}")
|
||
return None
|
||
|
||
def _is_douyu_email(self, content: str) -> bool:
|
||
"""判断邮件内容是否是斗鱼验证码邮件"""
|
||
douyu_keywords = ['斗鱼', 'douyu', 'douyutv']
|
||
verify_keywords = ['验证码', '安全验证', '登录验证', '动态码', '校验码']
|
||
combined = content.lower()
|
||
|
||
return (
|
||
any(keyword in combined for keyword in douyu_keywords)
|
||
or any(keyword in content for keyword in verify_keywords)
|
||
)
|
||
|
||
def _is_email_recent(
|
||
self,
|
||
content: str,
|
||
after_timestamp: float,
|
||
allow_old_seconds: int = 15,
|
||
) -> bool:
|
||
"""
|
||
检查邮件时间是否晚于验证码发送时间。
|
||
|
||
HTTP 接口返回的 HTML 中包含 "发信时间: Tue, 23 Jun 2026 01:06:09 +0800"
|
||
"""
|
||
# 提取发信时间
|
||
from email.utils import parsedate_to_datetime
|
||
time_match = re.search(r'发信时间:\s*(.+?)[\r\n<]', content)
|
||
if not time_match:
|
||
# 找不到时间，保守认为足够新
|
||
logger.debug("邮件中未找到发信时间，保守纳入候选")
|
||
return True
|
||
|
||
time_str = time_match.group(1).strip()
|
||
try:
|
||
from datetime import datetime
|
||
msg_time = parsedate_to_datetime(time_str)
|
||
if msg_time and msg_time.timestamp() < after_timestamp - allow_old_seconds:
|
||
return False
|
||
return True
|
||
except (TypeError, ValueError):
|
||
logger.debug(f"无法解析邮件时间: {time_str}，保守纳入候选")
|
||
return True
|
||
|
||
def _extract_verification_code(self, text: str) -> Optional[str]:
|
||
"""从文本中提取6位验证码"""
|
||
# 清理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',
|
||
]
|
||
|
||
for pattern in patterns:
|
||
match = re.search(pattern, text, re.IGNORECASE)
|
||
if match:
|
||
return match.group(1)
|
||
|
||
return None
|
||
|
||
|
||
def get_email_config_for_account(email_address: str) -> dict:
|
||
"""根据邮箱地址返回配置（HTTP 模式下仅用于兼容）"""
|
||
return {
|
||
'server': '',
|
||
'port': 143,
|
||
'ssl': False,
|
||
}
|