Spaces:

webhook
/

codex

Sleeping

codex / src /services /outlook /providers /imap_old.py

Upload 137 files

89419ea verified 7 months ago

10.2 kB

"""
旧版 IMAP 提供者
使用 outlook.office365.com 服务器和 login.live.com Token 端点
"""

import email
import imaplib
import logging
from email.header import decode_header
from email.utils import parsedate_to_datetime
from typing import List, Optional

from ..base import ProviderType, EmailMessage
from ..account import OutlookAccount
from ..token_manager import TokenManager
from .base import OutlookProvider, ProviderConfig


logger = logging.getLogger(__name__)


class IMAPOldProvider(OutlookProvider):
    """
    旧版 IMAP 提供者
    使用 outlook.office365.com:993 和 login.live.com Token 端点
    """

    # IMAP 服务器配置
    IMAP_HOST = "outlook.office365.com"
    IMAP_PORT = 993

    @property
    def provider_type(self) -> ProviderType:
        return ProviderType.IMAP_OLD

    def __init__(
        self,
        account: OutlookAccount,
        config: Optional[ProviderConfig] = None,
    ):
        super().__init__(account, config)

        # IMAP 连接
        self._conn: Optional[imaplib.IMAP4_SSL] = None

        # Token 管理器
        self._token_manager: Optional[TokenManager] = None

    def connect(self) -> bool:
        """
        连接到 IMAP 服务器
        Returns:
            是否连接成功
        """
        if self._connected and self._conn:
            # 检查现有连接
            try:
                self._conn.noop()
                return True
            except Exception:
                self.disconnect()

        try:
            logger.debug(f"[{self.account.email}] 正在连接 IMAP ({self.IMAP_HOST})...")

            # 创建连接
            self._conn = imaplib.IMAP4_SSL(
                self.IMAP_HOST,
                self.IMAP_PORT,
                timeout=self.config.timeout,
            )

            # 尝试 XOAUTH2 认证
            if self.account.has_oauth():
                if self._authenticate_xoauth2():
                    self._connected = True
                    self.record_success()
                    logger.info(f"[{self.account.email}] IMAP 连接成功 (XOAUTH2)")
                    return True
                else:
                    logger.warning(f"[{self.account.email}] XOAUTH2 认证失败，尝试密码认证")

            # 密码认证
            if self.account.password:
                self._conn.login(self.account.email, self.account.password)
                self._connected = True
                self.record_success()
                logger.info(f"[{self.account.email}] IMAP 连接成功 (密码认证)")
                return True

            raise ValueError("没有可用的认证方式")

        except Exception as e:
            self.disconnect()
            self.record_failure(str(e))
            logger.error(f"[{self.account.email}] IMAP 连接失败: {e}")
            return False

    def _authenticate_xoauth2(self) -> bool:
        """
        使用 XOAUTH2 认证
        Returns:
            是否认证成功
        """
        if not self._token_manager:
            self._token_manager = TokenManager(
                self.account,
                ProviderType.IMAP_OLD,
                self.config.proxy_url,
                self.config.timeout,
            )

        # 获取 Access Token
        token = self._token_manager.get_access_token()
        if not token:
            return False

        try:
            # 构建 XOAUTH2 认证字符串
            auth_string = f"user={self.account.email}\x01auth=Bearer {token}\x01\x01"
            self._conn.authenticate("XOAUTH2", lambda _: auth_string.encode("utf-8"))
            return True
        except Exception as e:
            logger.debug(f"[{self.account.email}] XOAUTH2 认证异常: {e}")
            # 清除缓存的 Token
            self._token_manager.clear_cache()
            return False

    def disconnect(self):
        """断开 IMAP 连接"""
        if self._conn:
            try:
                self._conn.close()
            except Exception:
                pass
            try:
                self._conn.logout()
            except Exception:
                pass
            self._conn = None

        self._connected = False

    def get_recent_emails(
        self,
        count: int = 20,
        only_unseen: bool = True,
    ) -> List[EmailMessage]:
        """
        获取最近的邮件
        Args:
            count: 获取数量
            only_unseen: 是否只获取未读
        Returns:
            邮件列表
        """
        if not self._connected:
            if not self.connect():
                return []

        try:
            # 选择收件箱
            self._conn.select("INBOX", readonly=True)

            # 搜索邮件
            flag = "UNSEEN" if only_unseen else "ALL"
            status, data = self._conn.search(None, flag)

            if status != "OK" or not data or not data[0]:
                return []

            # 获取最新的邮件 ID
            ids = data[0].split()
            recent_ids = ids[-count:][::-1]  # 倒序，最新的在前

            emails = []
            for msg_id in recent_ids:
                try:
                    email_msg = self._fetch_email(msg_id)
                    if email_msg:
                        emails.append(email_msg)
                except Exception as e:
                    logger.warning(f"[{self.account.email}] 解析邮件失败 (ID: {msg_id}): {e}")

            return emails

        except Exception as e:
            self.record_failure(str(e))
            logger.error(f"[{self.account.email}] 获取邮件失败: {e}")
            return []

    def _fetch_email(self, msg_id: bytes) -> Optional[EmailMessage]:
        """
        获取并解析单封邮件
        Args:
            msg_id: 邮件 ID
        Returns:
            EmailMessage 对象，失败返回 None
        """
        status, data = self._conn.fetch(msg_id, "(RFC822)")
        if status != "OK" or not data or not data[0]:
            return None

        # 获取原始邮件内容
        raw = b""
        for part in data:
            if isinstance(part, tuple) and len(part) > 1:
                raw = part[1]
                break

        if not raw:
            return None

        return self._parse_email(raw)

    @staticmethod
    def _parse_email(raw: bytes) -> EmailMessage:
        """
        解析原始邮件
        Args:
            raw: 原始邮件数据
        Returns:
            EmailMessage 对象
        """
        # 移除 BOM
        if raw.startswith(b"\xef\xbb\xbf"):
            raw = raw[3:]

        msg = email.message_from_bytes(raw)

        # 解析邮件头
        subject = IMAPOldProvider._decode_header(msg.get("Subject", ""))
        sender = IMAPOldProvider._decode_header(msg.get("From", ""))
        to = IMAPOldProvider._decode_header(msg.get("To", ""))
        delivered_to = IMAPOldProvider._decode_header(msg.get("Delivered-To", ""))
        x_original_to = IMAPOldProvider._decode_header(msg.get("X-Original-To", ""))
        date_str = IMAPOldProvider._decode_header(msg.get("Date", ""))

        # 提取正文
        body = IMAPOldProvider._extract_body(msg)

        # 解析日期
        received_timestamp = 0
        received_at = None
        try:
            if date_str:
                received_at = parsedate_to_datetime(date_str)
                received_timestamp = int(received_at.timestamp())
        except Exception:
            pass

        # 构建收件人列表
        recipients = [r for r in [to, delivered_to, x_original_to] if r]

        return EmailMessage(
            id=msg.get("Message-ID", ""),
            subject=subject,
            sender=sender,
            recipients=recipients,
            body=body,
            received_at=received_at,
            received_timestamp=received_timestamp,
            is_read=False,  # 搜索的是未读邮件
            raw_data=raw[:500] if len(raw) > 500 else raw,
        )

    @staticmethod
    def _decode_header(header: str) -> str:
        """解码邮件头"""
        if not header:
            return ""

        parts = []
        for chunk, encoding in decode_header(header):
            if isinstance(chunk, bytes):
                try:
                    decoded = chunk.decode(encoding or "utf-8", errors="replace")
                    parts.append(decoded)
                except Exception:
                    parts.append(chunk.decode("utf-8", errors="replace"))
            else:
                parts.append(str(chunk))

        return "".join(parts).strip()

    @staticmethod
    def _extract_body(msg) -> str:
        """提取邮件正文"""
        import html as html_module
        import re

        texts = []
        parts = msg.walk() if msg.is_multipart() else [msg]

        for part in parts:
            content_type = part.get_content_type()
            if content_type not in ("text/plain", "text/html"):
                continue

            payload = part.get_payload(decode=True)
            if not payload:
                continue

            charset = part.get_content_charset() or "utf-8"
            try:
                text = payload.decode(charset, errors="replace")
            except LookupError:
                text = payload.decode("utf-8", errors="replace")

            # 如果是 HTML，移除标签
            if "<html" in text.lower():
                text = re.sub(r"<[^>]+>", " ", text)

            texts.append(text)

        # 合并并清理文本
        combined = " ".join(texts)
        combined = html_module.unescape(combined)
        combined = re.sub(r"\s+", " ", combined).strip()

        return combined

    def test_connection(self) -> bool:
        """
        测试 IMAP 连接
        Returns:
            连接是否正常
        """
        try:
            with self:
                self._conn.select("INBOX", readonly=True)
                self._conn.search(None, "ALL")
            return True
        except Exception as e:
            logger.warning(f"[{self.account.email}] IMAP 连接测试失败: {e}")
            return False

	"""
	旧版 IMAP 提供者
	使用 outlook.office365.com 服务器和 login.live.com Token 端点
	"""

	import email
	import imaplib
	import logging
	from email.header import decode_header
	from email.utils import parsedate_to_datetime
	from typing import List, Optional

	from ..base import ProviderType, EmailMessage
	from ..account import OutlookAccount
	from ..token_manager import TokenManager
	from .base import OutlookProvider, ProviderConfig


	logger = logging.getLogger(__name__)


	class IMAPOldProvider(OutlookProvider):
	"""
	旧版 IMAP 提供者
	使用 outlook.office365.com:993 和 login.live.com Token 端点
	"""

	# IMAP 服务器配置
	IMAP_HOST = "outlook.office365.com"
	IMAP_PORT = 993

	@property
	def provider_type(self) -> ProviderType:
	return ProviderType.IMAP_OLD

	def __init__(
	self,
	account: OutlookAccount,
	config: Optional[ProviderConfig] = None,
	):
	super().__init__(account, config)

	# IMAP 连接
	self._conn: Optional[imaplib.IMAP4_SSL] = None

	# Token 管理器
	self._token_manager: Optional[TokenManager] = None

	def connect(self) -> bool:
	"""
	连接到 IMAP 服务器

	Returns:
	是否连接成功
	"""
	if self._connected and self._conn:
	# 检查现有连接
	try:
	self._conn.noop()
	return True
	except Exception:
	self.disconnect()

	try:
	logger.debug(f"[{self.account.email}] 正在连接 IMAP ({self.IMAP_HOST})...")

	# 创建连接
	self._conn = imaplib.IMAP4_SSL(
	self.IMAP_HOST,
	self.IMAP_PORT,
	timeout=self.config.timeout,
	)

	# 尝试 XOAUTH2 认证
	if self.account.has_oauth():
	if self._authenticate_xoauth2():
	self._connected = True
	self.record_success()
	logger.info(f"[{self.account.email}] IMAP 连接成功 (XOAUTH2)")
	return True
	else:
	logger.warning(f"[{self.account.email}] XOAUTH2 认证失败，尝试密码认证")

	# 密码认证
	if self.account.password:
	self._conn.login(self.account.email, self.account.password)
	self._connected = True
	self.record_success()
	logger.info(f"[{self.account.email}] IMAP 连接成功 (密码认证)")
	return True

	raise ValueError("没有可用的认证方式")

	except Exception as e:
	self.disconnect()
	self.record_failure(str(e))
	logger.error(f"[{self.account.email}] IMAP 连接失败: {e}")
	return False

	def _authenticate_xoauth2(self) -> bool:
	"""
	使用 XOAUTH2 认证

	Returns:
	是否认证成功
	"""
	if not self._token_manager:
	self._token_manager = TokenManager(
	self.account,
	ProviderType.IMAP_OLD,
	self.config.proxy_url,
	self.config.timeout,
	)

	# 获取 Access Token
	token = self._token_manager.get_access_token()
	if not token:
	return False

	try:
	# 构建 XOAUTH2 认证字符串
	auth_string = f"user={self.account.email}\x01auth=Bearer {token}\x01\x01"
	self._conn.authenticate("XOAUTH2", lambda _: auth_string.encode("utf-8"))
	return True
	except Exception as e:
	logger.debug(f"[{self.account.email}] XOAUTH2 认证异常: {e}")
	# 清除缓存的 Token
	self._token_manager.clear_cache()
	return False

	def disconnect(self):
	"""断开 IMAP 连接"""
	if self._conn:
	try:
	self._conn.close()
	except Exception:
	pass
	try:
	self._conn.logout()
	except Exception:
	pass
	self._conn = None

	self._connected = False

	def get_recent_emails(
	self,
	count: int = 20,
	only_unseen: bool = True,
	) -> List[EmailMessage]:
	"""
	获取最近的邮件

	Args:
	count: 获取数量
	only_unseen: 是否只获取未读

	Returns:
	邮件列表
	"""
	if not self._connected:
	if not self.connect():
	return []

	try:
	# 选择收件箱
	self._conn.select("INBOX", readonly=True)

	# 搜索邮件
	flag = "UNSEEN" if only_unseen else "ALL"
	status, data = self._conn.search(None, flag)

	if status != "OK" or not data or not data[0]:
	return []

	# 获取最新的邮件 ID
	ids = data[0].split()
	recent_ids = ids[-count:][::-1] # 倒序，最新的在前

	emails = []
	for msg_id in recent_ids:
	try:
	email_msg = self._fetch_email(msg_id)
	if email_msg:
	emails.append(email_msg)
	except Exception as e:
	logger.warning(f"[{self.account.email}] 解析邮件失败 (ID: {msg_id}): {e}")

	return emails

	except Exception as e:
	self.record_failure(str(e))
	logger.error(f"[{self.account.email}] 获取邮件失败: {e}")
	return []

	def _fetch_email(self, msg_id: bytes) -> Optional[EmailMessage]:
	"""
	获取并解析单封邮件

	Args:
	msg_id: 邮件 ID

	Returns:
	EmailMessage 对象，失败返回 None
	"""
	status, data = self._conn.fetch(msg_id, "(RFC822)")
	if status != "OK" or not data or not data[0]:
	return None

	# 获取原始邮件内容
	raw = b""
	for part in data:
	if isinstance(part, tuple) and len(part) > 1:
	raw = part[1]
	break

	if not raw:
	return None

	return self._parse_email(raw)

	@staticmethod
	def _parse_email(raw: bytes) -> EmailMessage:
	"""
	解析原始邮件

	Args:
	raw: 原始邮件数据

	Returns:
	EmailMessage 对象
	"""
	# 移除 BOM
	if raw.startswith(b"\xef\xbb\xbf"):
	raw = raw[3:]

	msg = email.message_from_bytes(raw)

	# 解析邮件头
	subject = IMAPOldProvider._decode_header(msg.get("Subject", ""))
	sender = IMAPOldProvider._decode_header(msg.get("From", ""))
	to = IMAPOldProvider._decode_header(msg.get("To", ""))
	delivered_to = IMAPOldProvider._decode_header(msg.get("Delivered-To", ""))
	x_original_to = IMAPOldProvider._decode_header(msg.get("X-Original-To", ""))
	date_str = IMAPOldProvider._decode_header(msg.get("Date", ""))

	# 提取正文
	body = IMAPOldProvider._extract_body(msg)

	# 解析日期
	received_timestamp = 0
	received_at = None
	try:
	if date_str:
	received_at = parsedate_to_datetime(date_str)
	received_timestamp = int(received_at.timestamp())
	except Exception:
	pass

	# 构建收件人列表
	recipients = [r for r in [to, delivered_to, x_original_to] if r]

	return EmailMessage(
	id=msg.get("Message-ID", ""),
	subject=subject,
	sender=sender,
	recipients=recipients,
	body=body,
	received_at=received_at,
	received_timestamp=received_timestamp,
	is_read=False, # 搜索的是未读邮件
	raw_data=raw[:500] if len(raw) > 500 else raw,
	)

	@staticmethod
	def _decode_header(header: str) -> str:
	"""解码邮件头"""
	if not header:
	return ""

	parts = []
	for chunk, encoding in decode_header(header):
	if isinstance(chunk, bytes):
	try:
	decoded = chunk.decode(encoding or "utf-8", errors="replace")
	parts.append(decoded)
	except Exception:
	parts.append(chunk.decode("utf-8", errors="replace"))
	else:
	parts.append(str(chunk))

	return "".join(parts).strip()

	@staticmethod
	def _extract_body(msg) -> str:
	"""提取邮件正文"""
	import html as html_module
	import re

	texts = []
	parts = msg.walk() if msg.is_multipart() else [msg]

	for part in parts:
	content_type = part.get_content_type()
	if content_type not in ("text/plain", "text/html"):
	continue

	payload = part.get_payload(decode=True)
	if not payload:
	continue

	charset = part.get_content_charset() or "utf-8"
	try:
	text = payload.decode(charset, errors="replace")
	except LookupError:
	text = payload.decode("utf-8", errors="replace")

	# 如果是 HTML，移除标签
	if "<html" in text.lower():
	text = re.sub(r"<[^>]+>", " ", text)

	texts.append(text)

	# 合并并清理文本
	combined = " ".join(texts)
	combined = html_module.unescape(combined)
	combined = re.sub(r"\s+", " ", combined).strip()

	return combined

	def test_connection(self) -> bool:
	"""
	测试 IMAP 连接

	Returns:
	连接是否正常
	"""
	try:
	with self:
	self._conn.select("INBOX", readonly=True)
	self._conn.search(None, "ALL")
	return True
	except Exception as e:
	logger.warning(f"[{self.account.email}] IMAP 连接测试失败: {e}")
	return False