500 lines
19 KiB
Python
500 lines
19 KiB
Python
"""邮箱验证模块 - 通过 Roundcube Webmail API 获取验证码。"""
|
||
|
||
import html
|
||
import os
|
||
import re
|
||
import threading
|
||
import time
|
||
from datetime import UTC, datetime, timedelta
|
||
|
||
import requests
|
||
from loguru import logger
|
||
|
||
# Roundcube Webmail 地址。可通过环境变量覆盖。空值时回退到默认。
|
||
ROUNDCUBE_URL = os.getenv("MAIL_ROUNDCUBE_URL") or "http://111.229.206.54:8000/"
|
||
EMAIL_BACKUP_PASSWORDS = [
|
||
pwd.strip()
|
||
for pwd in os.getenv("MAIL_BACKUP_PASSWORDS", "aa778899").split(";")
|
||
if pwd.strip()
|
||
]
|
||
|
||
|
||
class EmailLoginError(RuntimeError):
|
||
"""邮箱登录失败，通常是邮箱账号或密码错误。"""
|
||
|
||
|
||
class EmailVerifier:
|
||
"""邮箱验证器（Roundcube API）。"""
|
||
|
||
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,
|
||
roundcube_url: str = "",
|
||
backup_passwords: list[str] | None = None,
|
||
):
|
||
# 兼容旧参数名
|
||
self.imap_server = imap_server
|
||
self.imap_port = imap_port
|
||
self.username = username # 邮箱地址
|
||
self.password = password # 邮箱密码
|
||
self.timeout = timeout
|
||
self.roundcube_url = roundcube_url or ROUNDCUBE_URL
|
||
self.backup_passwords = (
|
||
backup_passwords if backup_passwords is not None else EMAIL_BACKUP_PASSWORDS
|
||
)
|
||
|
||
# Roundcube 会话（懒初始化）
|
||
self._rc_session: requests.Session | None = None
|
||
self._rc_token: str | None = None
|
||
self._rc_logged_in = False
|
||
self._rc_login_error = ""
|
||
|
||
def connect(self) -> None:
|
||
"""兼容接口"""
|
||
|
||
def disconnect(self) -> None:
|
||
"""关闭 Roundcube 会话"""
|
||
if self._rc_session:
|
||
try:
|
||
self._rc_session.get(
|
||
f"{self.roundcube_url}?_task=logout",
|
||
timeout=5,
|
||
)
|
||
except Exception as exc: # noqa: BLE001
|
||
logger.debug(f"Roundcube logout failed: {exc}")
|
||
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
|
||
|
||
self._rc_login_error = ""
|
||
credential_failed = False
|
||
passwords = self._candidate_passwords()
|
||
|
||
for index, password in enumerate(passwords):
|
||
is_backup = index > 0
|
||
ok, password_failed = self._try_roundcube_login(
|
||
password, is_backup=is_backup
|
||
)
|
||
if ok:
|
||
return True
|
||
credential_failed = credential_failed or password_failed
|
||
if password_failed and index == 0 and len(passwords) > 1:
|
||
logger.info("Roundcube: 主邮箱密码失败，尝试备用邮箱密码")
|
||
|
||
if credential_failed:
|
||
self._rc_login_error = "邮箱登录失败"
|
||
logger.warning(f"Roundcube: {self._rc_login_error}")
|
||
|
||
return False
|
||
|
||
def _candidate_passwords(self) -> list[str]:
|
||
"""返回去重后的邮箱密码候选列表：主密码优先，其次备用密码。"""
|
||
candidates = [self.password, *self.backup_passwords]
|
||
result = []
|
||
seen = set()
|
||
for password in candidates:
|
||
if not password or password in seen:
|
||
continue
|
||
result.append(password)
|
||
seen.add(password)
|
||
return result
|
||
|
||
def _try_roundcube_login(
|
||
self, password: str, is_backup: bool = False
|
||
) -> tuple[bool, bool]:
|
||
"""尝试一次 Roundcube 登录，返回 (是否成功, 是否明确为密码错误)。"""
|
||
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, 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": password,
|
||
},
|
||
timeout=self.timeout,
|
||
allow_redirects=True,
|
||
)
|
||
|
||
# 3. 检查登录是否成功：登录失败页也可能有 request_token，不能只靠 token 判定。
|
||
new_token_match = re.search(
|
||
r'request_token["\s:]+["\']([^"\']+)', resp.text
|
||
)
|
||
if not new_token_match:
|
||
logger.warning("Roundcube: 登录失败（未找到 request_token）")
|
||
return False, False
|
||
|
||
text_lower = resp.text.lower()
|
||
login_failed = "_err=loginfailed" in resp.url or "loginfailed" in text_lower
|
||
login_form_present = (
|
||
'name="_user"' in resp.text and 'name="_pass"' in resp.text
|
||
)
|
||
mail_page = "_task=mail" in resp.url or re.search(
|
||
r'(?:env\.task\s*=\s*|["\']task["\']\s*:\s*)["\']mail["\']', resp.text
|
||
)
|
||
|
||
if login_failed or login_form_present or not mail_page:
|
||
password_name = "备用邮箱密码" if is_backup else "主邮箱密码"
|
||
logger.warning(f"Roundcube: {password_name}登录失败")
|
||
return False, True
|
||
|
||
self._rc_session = session
|
||
self._rc_token = new_token_match.group(1)
|
||
self._rc_logged_in = True
|
||
password_name = "备用邮箱密码" if is_backup else "主邮箱密码"
|
||
logger.debug(f"Roundcube: {password_name}登录成功 ({self.username})")
|
||
return True, False
|
||
|
||
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
|
||
logger.warning(f"Roundcube: 登录异常: {e}")
|
||
return False, 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():
|
||
if self._rc_login_error:
|
||
raise EmailLoginError(self._rc_login_error)
|
||
return []
|
||
if self._rc_session is None:
|
||
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: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
|
||
logger.warning(f"Roundcube: 获取邮件列表失败: {e}")
|
||
return []
|
||
|
||
# ── Roundcube 读取邮件内容 ─────────────────────────────
|
||
|
||
def _rc_fetch_email_body(self, uid: int) -> str | None:
|
||
"""通过 Roundcube 读取指定 UID 邮件的正文"""
|
||
if not self._ensure_roundcube_session():
|
||
if self._rc_login_error:
|
||
raise EmailLoginError(self._rc_login_error)
|
||
return None
|
||
if self._rc_session is None:
|
||
return None
|
||
|
||
try:
|
||
resp = self._rc_session.get(
|
||
f"{self.roundcube_url}?_task=mail&_action=show&_mbox=INBOX&_uid={uid}",
|
||
timeout=self.timeout,
|
||
)
|
||
return resp.text
|
||
|
||
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
|
||
logger.warning(f"Roundcube: 读取邮件 {uid} 失败: {e}")
|
||
return None
|
||
|
||
# ── Roundcube 日期解析 ─────────────────────────────────
|
||
|
||
@staticmethod
|
||
def _parse_rc_date(date_str: str) -> datetime | None:
|
||
"""
|
||
解析 Roundcube 返回的日期字符串为 datetime。
|
||
|
||
格式示例:
|
||
- "今天 09:02" → 今天的 09:02
|
||
- "昨天 23:29" → 昨天的 23:29
|
||
- "星期一 23:29" → 本周星期一的 23:29
|
||
- "2026-06-20" → 直接解析
|
||
- "06-20" → 今年的该日期
|
||
"""
|
||
now = datetime.now(UTC)
|
||
|
||
# 今天
|
||
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").replace(tzinfo=UTC)
|
||
except ValueError:
|
||
pass
|
||
|
||
# 日期格式 2026-06-20 14:30
|
||
try:
|
||
return datetime.strptime(date_str.strip(), "%Y-%m-%d %H:%M").replace(
|
||
tzinfo=UTC
|
||
)
|
||
except ValueError:
|
||
pass
|
||
|
||
# 日期格式 2026-06-20 14:30:00
|
||
try:
|
||
return datetime.strptime(date_str.strip(), "%Y-%m-%d %H:%M:%S").replace(
|
||
tzinfo=UTC
|
||
)
|
||
except ValueError:
|
||
pass
|
||
|
||
# 日期格式 06-20
|
||
try:
|
||
dt = datetime.strptime(date_str.strip(), "%m-%d").replace(tzinfo=UTC)
|
||
return dt.replace(year=now.year)
|
||
except ValueError:
|
||
pass
|
||
|
||
# 日期格式 06-20 14:30
|
||
try:
|
||
dt = datetime.strptime(date_str.strip(), "%m-%d %H:%M").replace(tzinfo=UTC)
|
||
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: float | None = None,
|
||
allow_old_seconds: int = 15,
|
||
stop_event: threading.Event | None = None,
|
||
) -> str:
|
||
"""
|
||
轮询获取斗鱼验证码。
|
||
|
||
Args:
|
||
max_wait: 最大等待时间（秒）
|
||
interval: 轮询间隔（秒）
|
||
after_timestamp: 发送验证码请求的时间戳，用于过滤旧邮件
|
||
allow_old_seconds: 允许的时间偏移（秒）
|
||
stop_event: 外部停止信号，触发后尽快中断等待
|
||
|
||
Returns:
|
||
6位验证码
|
||
"""
|
||
logger.info(f"等待斗鱼验证码邮件，最大等待 {max_wait} 秒...")
|
||
|
||
deadline = time.monotonic() + max_wait
|
||
last_error = ""
|
||
|
||
while time.monotonic() < deadline:
|
||
if stop_event and stop_event.is_set():
|
||
raise InterruptedError("任务已停止")
|
||
|
||
try:
|
||
code = self._fetch_code_via_roundcube(
|
||
after_timestamp, allow_old_seconds
|
||
)
|
||
if code:
|
||
logger.success(f"获取到验证码: {code}")
|
||
return code
|
||
except EmailLoginError:
|
||
raise
|
||
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
|
||
last_error = str(e)
|
||
logger.warning(f"Roundcube 读邮件异常: {e}")
|
||
|
||
logger.debug("未找到验证码，等待中...")
|
||
sleep_deadline = time.monotonic() + interval
|
||
while time.monotonic() < sleep_deadline:
|
||
if stop_event and stop_event.is_set():
|
||
raise InterruptedError("任务已停止")
|
||
time.sleep(min(0.2, sleep_deadline - time.monotonic()))
|
||
|
||
raise TimeoutError(
|
||
f"等待验证码超时{'（' + last_error + '）' if last_error else ''}"
|
||
)
|
||
|
||
# ── Roundcube 方式获取验证码 ────────────────────────────
|
||
|
||
def _fetch_code_via_roundcube(
|
||
self,
|
||
after_timestamp: float | None = None,
|
||
allow_old_seconds: int = 15,
|
||
) -> str | None:
|
||
"""通过 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 is None:
|
||
# 日期无法解析的旧邮件（如完整时间戳），在有 after_timestamp 时跳过
|
||
logger.debug(
|
||
f"Roundcube: 邮件 UID={msg['uid']} 日期 '{msg['date']}' 无法解析，已跳过"
|
||
)
|
||
continue
|
||
# Roundcube 日期只有分钟精度（如"今天 14:20"→14:20:00），
|
||
# 而 after_timestamp 是秒级精度（如14:20:49）。
|
||
# 直接比较会把同一分钟内的邮件误判为"早于发送时间"跳过。
|
||
# 解决：将 after_timestamp 也向下取整到分钟后再比较。
|
||
after_minute_ts = after_timestamp - (after_timestamp % 60)
|
||
if msg_dt.timestamp() < after_minute_ts - 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
|
||
|
||
# ── 通用工具方法 ───────────────────────────────────────
|
||
|
||
def _extract_verification_code(self, text: str) -> str | None:
|
||
"""从文本中提取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,
|
||
}
|