diff --git a/core/douyu/email_verifier.py b/core/douyu/email_verifier.py index 3d25b7b..7b75295 100644 --- a/core/douyu/email_verifier.py +++ b/core/douyu/email_verifier.py @@ -1,381 +1,197 @@ -"""邮箱验证模块 - IMAP获取验证码""" +"""邮箱验证模块 - 通过 HTTP 接口获取验证码（替代 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 +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, - username: str, - password: str, - timeout: float = 12, + 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 = True, + 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.username = username # 邮箱地址 + self.password = password # 邮箱密码 self.timeout = timeout - self.mailbox = mailbox - self.lookback_minutes = lookback_minutes - self.max_messages = max_messages + self.mail_api_url = mail_api_url self.use_ssl = use_ssl - self._connection: Optional[imaplib.IMAP4_SSL] = None def connect(self) -> None: - """连接IMAP服务器""" - try: - logger.info(f"连接IMAP服务器: {self.imap_server}:{self.imap_port} (SSL={self.use_ssl})") - self._validate_login_text() - if self.use_ssl: - self._connection = imaplib.IMAP4_SSL( - self.imap_server, - self.imap_port, - timeout=self.timeout, - ) - else: - self._connection = imaplib.IMAP4( - 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 + """兼容接口：HTTP 模式下无需持久连接""" + pass def disconnect(self) -> None: - """断开IMAP连接""" - if self._connection: - try: - self._connection.logout() - except: - pass - self._connection = None + """兼容接口""" + pass def get_verification_code( self, max_wait: int = 60, - interval: int = 2, + interval: int = 3, after_timestamp: Optional[float] = None, allow_old_seconds: int = 15, ) -> str: """ - 获取斗鱼验证码 + 轮询 HTTP 读邮件接口获取斗鱼验证码。 Args: max_wait: 最大等待时间（秒） interval: 轮询间隔（秒） - after_timestamp: 发起发送验证码请求的时间戳，用于过滤旧邮件 - allow_old_seconds: 邮件服务器时间允许向前偏移的秒数 + after_timestamp: 发送验证码请求的时间戳，用于过滤旧邮件 + allow_old_seconds: 允许的时间偏移（秒） Returns: 6位验证码 """ logger.info(f"等待斗鱼验证码邮件，最大等待 {max_wait} 秒...") - self.connect() + deadline = time.monotonic() + max_wait + last_error = "" - 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, - ) + 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("未找到验证码，等待中...") - time.sleep(interval) + except Exception as e: + last_error = str(e) + logger.warning(f"读取邮件异常: {e}") + time.sleep(interval) - raise TimeoutError("等待验证码超时") - finally: - self.disconnect() + raise TimeoutError(f"等待验证码超时{'（' + last_error + '）' if last_error else ''}") - def _fetch_latest_code( + def _fetch_code_via_http( self, after_timestamp: Optional[float] = None, allow_old_seconds: int = 15, ) -> Optional[str]: - """从IMAP获取最新验证码""" + """通过 HTTP 接口读取最新邮件并提取验证码""" 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', '') - to_addr = msg.get('To', '') or msg.get('Delivered-To', '') or '' - body = self._get_email_body(msg) - - if not self._is_douyu_email(subject, from_addr, body, to_addr=to_addr): - 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) + response = requests.get( + self.mail_api_url, + params={ + "yhm": self.username, + "mm": self.password, + }, + timeout=self.timeout, ) - else: - since_dt = datetime.now() - timedelta(minutes=self.lookback_minutes) + response.raise_for_status() + content = response.text - return self._format_imap_date(since_dt) + # 检查是否是斗鱼验证码邮件 + if not self._is_douyu_email(content): + logger.debug("最新邮件不是斗鱼验证码邮件") + return None - 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}" + # 检查邮件时间是否足够新（发送验证码请求之后） + 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 - 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: + except requests.RequestException as e: + logger.warning(f"HTTP读邮件请求失败: {e}") 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 = "", to_addr: str = "") -> bool: - """判断是否是发给当前账号的斗鱼验证码邮件""" - # 先检查收件人：多个账号共用同一个IMAP时，必须匹配收件人地址 - if to_addr and self.username: - # to_addr 可能是 "name " 格式，提取邮箱 - to_email = re.search(r'[\w.+-]+@[\w.-]+', to_addr) - if to_email: - to_lower = to_email.group(0).lower() - my_lower = self.username.lower() - # 精确匹配，或者 catch-all 前缀匹配 - if to_lower != my_lower and not my_lower.endswith('@' + to_lower.split('@')[-1]): - logger.debug(f"跳过非当前账号邮件: 收件人={to_lower}, 当前={my_lower}") - return False - + def _is_douyu_email(self, content: str) -> bool: + """判断邮件内容是否是斗鱼验证码邮件""" douyu_keywords = ['斗鱼', 'douyu', 'douyutv'] verify_keywords = ['验证码', '安全验证', '登录验证', '动态码', '校验码'] - combined = f"{subject}\n{from_addr}\n{body[:500]}".lower() + combined = content.lower() return ( any(keyword in combined for keyword in douyu_keywords) - or any(keyword in subject for keyword in verify_keywords) + or any(keyword in content for keyword in verify_keywords) ) - def _decode_subject(self, subject: str) -> str: - """解码邮件主题""" - if not subject: - return "" + def _is_email_recent( + self, + content: str, + after_timestamp: float, + allow_old_seconds: int = 15, + ) -> bool: + """ + 检查邮件时间是否晚于验证码发送时间。 - 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) + 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 - 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: - """解码邮件片段内容。""" + time_str = time_match.group(1).strip() 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 "" + 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]: - """从文本中提取验证码""" - # 清理HTML标签和实体，方便匹配中文邮件模板。 + """从文本中提取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', # 候选邮件已经过滤为斗鱼验证码邮件，最后再兜底匹配6位数字。 + r'\b(\d{6})\b', ] for pattern in patterns: match = re.search(pattern, text, re.IGNORECASE) if match: - code = match.group(1) - return code + return match.group(1) return None def get_email_config_for_account(email_address: str) -> dict: - """根据邮箱地址自动配置IMAP服务器，默认使用统一IMAP代理""" - configs = { - 'qq.com': {'server': 'imap.qq.com', 'port': 993, 'ssl': True}, - '163.com': {'server': 'imap.163.com', 'port': 993, 'ssl': True}, - '126.com': {'server': 'imap.126.com', 'port': 993, 'ssl': True}, - 'gmail.com': {'server': 'imap.gmail.com', 'port': 993, 'ssl': True}, - 'outlook.com': {'server': 'outlook.office365.com', 'port': 993, 'ssl': True}, - 'hotmail.com': {'server': 'outlook.office365.com', 'port': 993, 'ssl': True}, + """根据邮箱地址返回配置（HTTP 模式下仅用于兼容）""" + return { + 'server': '', + 'port': 143, + 'ssl': False, } - - domain = email_address.split('@')[-1].lower() - configs = { - 'qq.com': {'server': 'imap.qq.com', 'port': 993, 'ssl': True}, - '163.com': {'server': 'imap.163.com', 'port': 993, 'ssl': True}, - '126.com': {'server': 'imap.126.com', 'port': 993, 'ssl': True}, - 'gmail.com': {'server': 'imap.gmail.com', 'port': 993, 'ssl': True}, - 'outlook.com': {'server': 'outlook.office365.com', 'port': 993, 'ssl': True}, - 'hotmail.com': {'server': 'outlook.office365.com', 'port': 993, 'ssl': True}, - 'bdhg.xyz': {'server': 'mail.bdhg.xyz', 'port': 143, 'ssl': False}, - } - - domain = email_address.split('@')[-1].lower() - return configs.get(domain, {'server': domain, 'port': 143, 'ssl': False})