"""邮箱验证模块 - 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._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 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服务器""" 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': f'imap.{domain}', 'port': 993})