yml2213
3b36175bbb
- 新增Roundcube Webmail登录+JSON API读取邮件列表（含精确日期） - Roundcube优先，read.php降为备用回退 - 解析Roundcube返回的中文日期格式（今天/昨天/星期X HH:MM） - 修复read.php时间字段被MIME编码Subject污染导致'保守拒绝'误判 - 修复邮件时间UTC/本地时区8小时偏差导致正常邮件被跳过的问题 - read.php备用模式下解析不到时间不再保守拒绝，改为放行
490 lines
18 KiB
Python
490 lines
18 KiB
Python
"""邮箱验证模块 - 优先通过 Roundcube Webmail API 获取验证码，read.php 作为备用"""
|
||
|
||
import html
|
||
import re
|
||
import time
|
||
from datetime import datetime, timedelta
|
||
from typing import Optional
|
||
|
||
from loguru import logger
|
||
import requests
|
||
|
||
|
||
# 旧接口（时间解析不可靠，作为备用）
|
||
READ_PHP_URL = "http://111.229.206.54:8000/read.php"
|
||
|
||
# Roundcube Webmail 地址
|
||
ROUNDCUBE_URL = "http://111.229.206.54:8000/"
|
||
|
||
|
||
class EmailVerifier:
|
||
"""邮箱验证器（Roundcube API 优先，read.php 备用）"""
|
||
|
||
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 = READ_PHP_URL,
|
||
):
|
||
# 兼容旧参数名
|
||
self.imap_server = imap_server
|
||
self.imap_port = imap_port
|
||
self.username = username # 邮箱地址
|
||
self.password = password # 邮箱密码
|
||
self.timeout = timeout
|
||
self.read_php_url = mail_api_url or READ_PHP_URL
|
||
self.roundcube_url = ROUNDCUBE_URL
|
||
|
||
# Roundcube 会话（懒初始化）
|
||
self._rc_session: Optional[requests.Session] = None
|
||
self._rc_token: Optional[str] = None
|
||
self._rc_logged_in = False
|
||
|
||
def connect(self) -> None:
|
||
"""兼容接口"""
|
||
pass
|
||
|
||
def disconnect(self) -> None:
|
||
"""关闭 Roundcube 会话"""
|
||
if self._rc_session:
|
||
try:
|
||
self._rc_session.get(
|
||
f"{self.roundcube_url}?_task=logout",
|
||
timeout=5,
|
||
)
|
||
except Exception:
|
||
pass
|
||
self._rc_session = None
|
||
self._rc_logged_in = False
|
||
|
||
# ── Roundcube 会话管理 ──────────────────────────────────
|
||
|
||
def _ensure_roundcube_session(self) -> bool:
|
||
"""确保已登录 Roundcube，返回是否成功"""
|
||
if self._rc_logged_in and self._rc_session and self._rc_token:
|
||
return True
|
||
|
||
try:
|
||
session = requests.Session()
|
||
session.headers.update({
|
||
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36",
|
||
})
|
||
|
||
# 1. 访问首页获取 token
|
||
resp = session.get(self.roundcube_url, timeout=self.timeout)
|
||
token_match = re.search(r'name="_token"\s+value="([^"]+)"', resp.text)
|
||
if not token_match:
|
||
logger.warning("Roundcube: 未获取到登录 token")
|
||
return False
|
||
token = token_match.group(1)
|
||
|
||
# 2. 登录
|
||
resp = session.post(
|
||
f"{self.roundcube_url}?_task=login",
|
||
data={
|
||
"_token": token,
|
||
"_task": "login",
|
||
"_action": "login",
|
||
"_user": self.username,
|
||
"_pass": self.password,
|
||
},
|
||
timeout=self.timeout,
|
||
allow_redirects=True,
|
||
)
|
||
|
||
# 3. 检查登录是否成功（成功会跳到 ?_task=mail）
|
||
new_token_match = re.search(r'request_token["\s:]+["\']([^"\']+)', resp.text)
|
||
if not new_token_match:
|
||
logger.warning("Roundcube: 登录失败（未找到 request_token）")
|
||
return False
|
||
|
||
self._rc_session = session
|
||
self._rc_token = new_token_match.group(1)
|
||
self._rc_logged_in = True
|
||
logger.debug(f"Roundcube: 登录成功 ({self.username})")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.warning(f"Roundcube: 登录异常: {e}")
|
||
return False
|
||
|
||
# ── Roundcube 邮件列表（含精确日期） ────────────────────
|
||
|
||
def _rc_fetch_mail_list(self) -> list[dict]:
|
||
"""
|
||
通过 Roundcube JSON API 获取 INBOX 邮件列表。
|
||
返回 [{"uid": 2, "subject": "验证码 - 斗鱼", "date": "今天 09:02"}, ...]
|
||
"""
|
||
if not self._ensure_roundcube_session():
|
||
return []
|
||
|
||
try:
|
||
resp = self._rc_session.get(
|
||
f"{self.roundcube_url}?_task=mail&_action=list"
|
||
f"&_mbox=INBOX&_remote=1&_unlock=1",
|
||
headers={"X-Requested-With": "XMLHttpRequest"},
|
||
timeout=self.timeout,
|
||
)
|
||
# Roundcube 返回的 JSON 可能带 BOM，需用 utf-8-sig 解码
|
||
text = resp.content.decode("utf-8-sig")
|
||
data = __import__("json").loads(text)
|
||
|
||
# 从 exec 字段中解析 add_message_row 调用
|
||
exec_text = data.get("exec", "")
|
||
messages = []
|
||
|
||
# 匹配: this.add_message_row(UID, {subject:"...", fromto:"...", date:"今天 09:02", size:"5 KB"}, {...}, ...);
|
||
for m in re.finditer(
|
||
r'add_message_row\((\d+),\s*(\{[^}]+\})',
|
||
exec_text,
|
||
):
|
||
uid = int(m.group(1))
|
||
props_str = m.group(2)
|
||
# 提取各字段
|
||
subject_m = re.search(r'"subject"\s*:\s*"([^"]*)"', props_str)
|
||
date_m = re.search(r'"date"\s*:\s*"([^"]*)"', props_str)
|
||
|
||
messages.append({
|
||
"uid": uid,
|
||
"subject": subject_m.group(1) if subject_m else "",
|
||
"date": date_m.group(1) if date_m else "",
|
||
})
|
||
|
||
return messages
|
||
|
||
except Exception as e:
|
||
logger.warning(f"Roundcube: 获取邮件列表失败: {e}")
|
||
return []
|
||
|
||
# ── Roundcube 读取邮件内容 ─────────────────────────────
|
||
|
||
def _rc_fetch_email_body(self, uid: int) -> Optional[str]:
|
||
"""通过 Roundcube 读取指定 UID 邮件的正文"""
|
||
if not self._ensure_roundcube_session():
|
||
return None
|
||
|
||
try:
|
||
resp = self._rc_session.get(
|
||
f"{self.roundcube_url}?_task=mail&_action=show"
|
||
f"&_mbox=INBOX&_uid={uid}",
|
||
timeout=self.timeout,
|
||
)
|
||
return resp.text
|
||
|
||
except Exception as e:
|
||
logger.warning(f"Roundcube: 读取邮件 {uid} 失败: {e}")
|
||
return None
|
||
|
||
# ── Roundcube 日期解析 ─────────────────────────────────
|
||
|
||
@staticmethod
|
||
def _parse_rc_date(date_str: str) -> Optional[datetime]:
|
||
"""
|
||
解析 Roundcube 返回的日期字符串为 datetime。
|
||
|
||
格式示例:
|
||
- "今天 09:02" → 今天的 09:02
|
||
- "昨天 23:29" → 昨天的 23:29
|
||
- "星期一 23:29" → 本周星期一的 23:29
|
||
- "2026-06-20" → 直接解析
|
||
- "06-20" → 今年的该日期
|
||
"""
|
||
now = datetime.now()
|
||
|
||
# 今天
|
||
if date_str.startswith("今天"):
|
||
time_str = date_str.replace("今天", "").strip()
|
||
try:
|
||
h, m = time_str.split(":")
|
||
return now.replace(hour=int(h), minute=int(m), second=0, microsecond=0)
|
||
except (ValueError, AttributeError):
|
||
return None
|
||
|
||
# 昨天
|
||
if date_str.startswith("昨天"):
|
||
time_str = date_str.replace("昨天", "").strip()
|
||
try:
|
||
h, m = time_str.split(":")
|
||
dt = now - timedelta(days=1)
|
||
return dt.replace(hour=int(h), minute=int(m), second=0, microsecond=0)
|
||
except (ValueError, AttributeError):
|
||
return None
|
||
|
||
# 星期X
|
||
weekday_map = {
|
||
"星期一": 0, "星期二": 1, "星期三": 2, "星期四": 3,
|
||
"星期五": 4, "星期六": 5, "星期日": 6, "星期天": 6,
|
||
}
|
||
for prefix, wd in weekday_map.items():
|
||
if date_str.startswith(prefix):
|
||
time_str = date_str.replace(prefix, "").strip()
|
||
try:
|
||
h, m = time_str.split(":")
|
||
# 计算本周对应日期
|
||
days_ago = (now.weekday() - wd) % 7
|
||
if days_ago == 0:
|
||
# 同一天但可能是上一周
|
||
target = now
|
||
else:
|
||
target = now - timedelta(days=days_ago)
|
||
return target.replace(hour=int(h), minute=int(m), second=0, microsecond=0)
|
||
except (ValueError, AttributeError):
|
||
return None
|
||
|
||
# 日期格式 2026-06-20
|
||
try:
|
||
return datetime.strptime(date_str.strip(), "%Y-%m-%d")
|
||
except ValueError:
|
||
pass
|
||
|
||
# 日期格式 06-20
|
||
try:
|
||
dt = datetime.strptime(date_str.strip(), "%m-%d")
|
||
return dt.replace(year=now.year)
|
||
except ValueError:
|
||
pass
|
||
|
||
return None
|
||
|
||
# ── 主入口 ─────────────────────────────────────────────
|
||
|
||
def get_verification_code(
|
||
self,
|
||
max_wait: int = 60,
|
||
interval: int = 3,
|
||
after_timestamp: Optional[float] = None,
|
||
allow_old_seconds: int = 15,
|
||
) -> str:
|
||
"""
|
||
轮询获取斗鱼验证码。优先使用 Roundcube API，失败则回退到 read.php。
|
||
|
||
Args:
|
||
max_wait: 最大等待时间（秒）
|
||
interval: 轮询间隔（秒）
|
||
after_timestamp: 发送验证码请求的时间戳，用于过滤旧邮件
|
||
allow_old_seconds: 允许的时间偏移（秒）
|
||
|
||
Returns:
|
||
6位验证码
|
||
"""
|
||
logger.info(f"等待斗鱼验证码邮件，最大等待 {max_wait} 秒...")
|
||
|
||
deadline = time.monotonic() + max_wait
|
||
last_error = ""
|
||
tried_roundcube = False
|
||
|
||
while time.monotonic() < deadline:
|
||
# 优先尝试 Roundcube
|
||
if not tried_roundcube or self._rc_logged_in:
|
||
try:
|
||
code = self._fetch_code_via_roundcube(after_timestamp, allow_old_seconds)
|
||
if code:
|
||
logger.success(f"获取到验证码: {code}")
|
||
return code
|
||
except Exception as e:
|
||
last_error = str(e)
|
||
logger.warning(f"Roundcube 读邮件异常: {e}")
|
||
tried_roundcube = True
|
||
|
||
# 回退到 read.php
|
||
try:
|
||
code = self._fetch_code_via_readphp(after_timestamp, allow_old_seconds)
|
||
if code:
|
||
logger.success(f"获取到验证码: {code}")
|
||
return code
|
||
except Exception as e:
|
||
last_error = str(e)
|
||
logger.warning(f"read.php 读邮件异常: {e}")
|
||
|
||
logger.debug("未找到验证码，等待中...")
|
||
time.sleep(interval)
|
||
|
||
raise TimeoutError(f"等待验证码超时{'（' + last_error + '）' if last_error else ''}")
|
||
|
||
# ── Roundcube 方式获取验证码 ────────────────────────────
|
||
|
||
def _fetch_code_via_roundcube(
|
||
self,
|
||
after_timestamp: Optional[float] = None,
|
||
allow_old_seconds: int = 15,
|
||
) -> Optional[str]:
|
||
"""通过 Roundcube API 获取最新斗鱼验证码"""
|
||
messages = self._rc_fetch_mail_list()
|
||
|
||
# 筛选斗鱼验证码邮件（按 UID 降序，即最新的先看）
|
||
douyu_msgs = [
|
||
msg for msg in messages
|
||
if "斗鱼" in msg.get("subject", "") or "验证码" in msg.get("subject", "")
|
||
]
|
||
|
||
for msg in reversed(douyu_msgs): # UID 最大的先看
|
||
# 检查时间
|
||
if after_timestamp and msg.get("date"):
|
||
msg_dt = self._parse_rc_date(msg["date"])
|
||
if msg_dt and msg_dt.timestamp() < after_timestamp - allow_old_seconds:
|
||
logger.debug(f"Roundcube: 邮件 UID={msg['uid']} 日期 {msg['date']} 早于发送时间，跳过")
|
||
continue
|
||
|
||
# 读取邮件正文提取验证码
|
||
body_html = self._rc_fetch_email_body(msg["uid"])
|
||
if not body_html:
|
||
continue
|
||
|
||
code = self._extract_verification_code(body_html)
|
||
if code:
|
||
logger.info(f"Roundcube: 从邮件 UID={msg['uid']} 提取到验证码: {code}")
|
||
return code
|
||
|
||
return None
|
||
|
||
# ── read.php 方式获取验证码（备用） ─────────────────────
|
||
|
||
def _fetch_code_via_readphp(
|
||
self,
|
||
after_timestamp: Optional[float] = None,
|
||
allow_old_seconds: int = 15,
|
||
) -> Optional[str]:
|
||
"""通过 read.php HTTP 接口读取最新邮件并提取验证码"""
|
||
try:
|
||
response = requests.get(
|
||
self.read_php_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("read.php: 最新邮件不是斗鱼验证码邮件")
|
||
return None
|
||
|
||
# 检查邮件时间（read.php 时间不可靠，放宽判断）
|
||
if after_timestamp and not self._is_email_recent(content, after_timestamp, allow_old_seconds):
|
||
logger.debug("read.php: 邮件时间早于验证码发送时间，跳过")
|
||
return None
|
||
|
||
# 提取验证码
|
||
code = self._extract_verification_code(content)
|
||
if code:
|
||
logger.info(f"read.php: 从邮件中提取到验证码: {code}")
|
||
return code
|
||
|
||
return None
|
||
|
||
except requests.RequestException as e:
|
||
logger.warning(f"read.php: 请求失败: {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:
|
||
"""
|
||
检查 read.php 返回的邮件时间是否晚于验证码发送时间。
|
||
注意：read.php 的时间解析不可靠，此处仅作参考判断。
|
||
"""
|
||
from email.utils import parsedate_to_datetime
|
||
|
||
# 先清理 HTML 标签，再提取时间
|
||
clean = re.sub(r'<[^>]+>', ' ', content)
|
||
clean = re.sub(r'\s+', ' ', clean)
|
||
|
||
# 匹配 RFC2822 格式时间
|
||
time_match = re.search(
|
||
r'发信时间[:：]?\s*'
|
||
r'((?:Mon|Tue|Wed|Thu|Fri|Sat|Sun),\s*\d{1,2}\s+'
|
||
r'(?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)\s+'
|
||
r'\d{4}\s+\d{2}:\d{2}:\d{2}\s*[+-]\d{4})',
|
||
clean,
|
||
)
|
||
if not time_match:
|
||
# 尝试匹配不带星期的时间格式
|
||
time_match2 = re.search(
|
||
r'发信时间[:：]?\s*(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})',
|
||
clean,
|
||
)
|
||
if time_match2:
|
||
time_str = time_match2.group(1).strip()
|
||
try:
|
||
from datetime import datetime as dt
|
||
msg_time = dt.strptime(time_str, '%Y-%m-%d %H:%M:%S')
|
||
if msg_time.timestamp() < after_timestamp - allow_old_seconds:
|
||
logger.debug(f"邮件时间 {time_str} 早于发送时间，跳过旧邮件")
|
||
return False
|
||
return True
|
||
except ValueError:
|
||
pass
|
||
# read.php 时间不可靠时，不再保守拒绝，改为放行
|
||
# 让验证码提取逻辑自行判断
|
||
logger.debug("read.php: 未找到可解析的发信时间，跳过时间检查直接提取验证码")
|
||
return True
|
||
|
||
time_str = time_match.group(1).strip()
|
||
try:
|
||
msg_time = parsedate_to_datetime(time_str)
|
||
if msg_time and msg_time.timestamp() < after_timestamp - allow_old_seconds:
|
||
logger.debug(f"邮件时间 {time_str} 早于发送时间，跳过旧邮件")
|
||
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,
|
||
}
|