收发邮件.py

import base64
import email
import imaplib
import smtplib
import requests
from email.header import decode_header, Header
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from typing import List, Dict
from unibo_auth2_get_AuthCode_RefreshToken_sucsses import OAuth2Authenticator


class EmailClient:
    def __init__(self, username: str, password: str, email_address: str):
        """
        Initialize the EmailClient with user credentials.

        Args:
            username: The username for authentication
            password: The password for authentication
            email_address: The full email address
        """
        self.username = username
        self.password = password
        self.email_address = email_address
        self.client_id = "9e5f94bc-e8a4-4e73-b8be-63364c29d753"
        self.imap_server = "outlook.office365.com"
        self.imap_port = 993
        # self.smtp_server = "submission.unipi.it"  # 使用密码登录
        self.smtp_server_XOAUTH2 = "smtp.office365.com"  # 使用OAuth2登录
        self.smtp_port = 587
        self.refresh_token = None
        self.access_token = None

    def authenticate_oauth(self) -> None:
        """Authenticate using OAuth2 and get refresh/access tokens."""
        authenticator = OAuth2Authenticator(
            username=self.username,
            password=self.password
        )

        try:
            auth_code = authenticator.execute_flow()
            print(f"成功获取授权码: {auth_code}")
            self.refresh_token, self.access_token = authenticator.get_refresh_token(auth_code)
        except Exception as e:
            raise Exception(f"认证失败: {e}")

    def _generate_auth_string(self) -> bytes:
        """Generate XOAUTH2 authentication string."""
        auth_string = f"user={self.email_address}\x01auth=Bearer {self.access_token}\x01\x01"
        return auth_string.encode('utf-8')

    def connect_imap(self) -> imaplib.IMAP4:
        """
        Connect to IMAP server using XOAUTH2 authentication.

        Returns:
            IMAP4 connection object

        Raises:
            Exception: If authentication fails
        """
        try:
            imap = imaplib.IMAP4_SSL(self.imap_server, self.imap_port)

            if 'AUTH=XOAUTH2' not in imap.capabilities:
                raise Exception("Server does not support XOAUTH2 authentication")

            auth_bytes = self._generate_auth_string()

            def oauth2_callback(response):
                return auth_bytes

            print(f"Attempting XOAUTH2 authentication as {self.email_address}...")
            typ, data = imap.authenticate('XOAUTH2', oauth2_callback)
            if typ != 'OK':
                raise Exception(f"Authentication failed: {data[0].decode('utf-8', errors='replace')}")

            print("XOAUTH2 authentication successful")
            return imap
        except Exception as e:
            raise Exception(f"IMAP XOAUTH2 authentication failed: {str(e)}")

    def _get_decoded_header(self, header: str) -> str:
        """解码邮件头信息

        Args:
            header: 需要解码的邮件头字符串

        Returns:
            解码后的字符串
        """
        try:
            decoded = decode_header(header)

            # 修复括号匹配问题
            return "".join(
                str(
                    t[0].decode(t[1] or "utf-8", errors="replace")
                    if isinstance(t[0], bytes)
                    else t[0]
                )
                for t in decoded
            )
        except Exception as e:
            print(f"邮件头解码失败: {str(e)}")
            return header

    def _parse_email_message(self, msg: email.message.Message) -> Dict:
        """Parse an email message into a dictionary."""
        email_data = {}

        # Parse headers
        email_data["from"] = self._get_decoded_header(msg["from"]) if msg["from"] else "N/A"
        email_data["subject"] = self._get_decoded_header(msg["subject"]) if msg["subject"] else "N/A"
        email_data["date"] = msg["date"] if msg["date"] else "N/A"
        email_data["to"] = self._get_decoded_header(msg.get("to", ""))

        # Parse body content
        def get_body_content(part):
            charset = part.get_content_charset() or "utf-8"
            try:
                return part.get_payload(decode=True).decode(charset, errors="replace")
            except Exception as e:
                print(f"正文解码失败: {str(e)}")
                return ""

        if msg.is_multipart():
            for part in msg.walk():
                content_type = part.get_content_type()
                content_disposition = str(part.get("Content-Disposition"))
                if "attachment" in content_disposition:
                    continue
                if content_type == "text/plain":
                    email_data["text"] = get_body_content(part)
                elif content_type == "text/html":
                    email_data["html"] = get_body_content(part)
        else:
            content_type = msg.get_content_type()
            if content_type == "text/plain":
                email_data["text"] = get_body_content(msg)
            elif content_type == "text/html":
                email_data["html"] = get_body_content(msg)

        return email_data

    def read_emails(self, mailbox: str = "INBOX", limit: int = 1, keyword: str = 'ALL') -> List[Dict]:
        """
        Read emails from the specified mailbox.

        Args:
            mailbox: Mailbox to read from (default: INBOX)
            limit: Maximum number of emails to return (default: 1)
            keyword: Search keyword (default: 'ALL')

        Returns:
            List of parsed email dictionaries
        """
        imap = self.connect_imap()
        try:
            status, _ = imap.select(mailbox, readonly=True)
            if status != 'OK':
                raise Exception(f"无法选择邮箱 {mailbox}")

            _, messages = imap.search(None, keyword)
            email_list = []
            message_ids = messages[0].split()

            if not message_ids:
                print(f"邮箱 {mailbox} 中没有邮件")
                return []

            for num in reversed(message_ids[:limit]):
                _, msg_data = imap.fetch(num, "(RFC822)")
                email_message = email.message_from_bytes(msg_data[0][1])
                email_data = self._parse_email_message(email_message)
                email_data["uid"] = num.decode()
                email_list.append(email_data)

            print(f"成功读取 {len(email_list)} 封邮件")
            return email_list
        finally:
            imap.logout()

    def send_email(self, receiver: str, subject: str, body: str = None, is_html: bool = True) -> None:
        """
        Send an email using SMTP.

        Args:
            receiver: Recipient email address
            subject: Email subject
            body: Email body content
            is_html: Whether the body is HTML (default: True)

        Raises:
            Exception: If email sending fails
        """
        if body is None:
            body = """
            <h1>Python SMTP 测试邮件</h1>
            <p>这是一封通过Python SMTP发送的测试邮件。</p>
            """
            is_html = True

        message = MIMEMultipart()
        # message['From'] = self.email_address
        message['To'] = receiver
        message['Subject'] = subject

        message.attach(MIMEText(body, 'html' if is_html else 'plain', 'utf-8'))

        try:
            if self.smtp_port == 465:
                server = smtplib.SMTP_SSL(self.smtp_server_XOAUTH2, self.smtp_port)
            else:
                server = smtplib.SMTP(self.smtp_server_XOAUTH2, self.smtp_port)
                server.starttls()

            # 建立 SMTP 连接
            server = smtplib.SMTP(self.smtp_server_XOAUTH2, self.smtp_port)
            # server.set_debuglevel(1)  # 启用调试输出
            server.ehlo()  # 识别客户端
            server.starttls()  # 启用TLS加密
            server.ehlo()  # 再次识别客户端
            # 准备OAuth2认证字符串
            auth_string = f"user={self.email_address}\1auth=Bearer {self.access_token}\1\1"
            auth_string = base64.b64encode(auth_string.encode('utf-8')).decode('utf-8')

            # 执行OAuth2认证
            code, response = server.docmd('AUTH', 'XOAUTH2 ' + auth_string)
            server.sendmail(self.email_address, [receiver], message.as_string())
            print("邮件发送成功")
        except Exception as e:
            raise Exception(f"邮件发送失败: {e}")
        finally:
            server.quit()

    def refresh_access_token(self) -> str:
        """
        Refresh the access token using the refresh token.

        Returns:
            New access token

        Raises:
            Exception: If token refresh fails
        """
        if not self.refresh_token:
            raise Exception("No refresh token available")

        url = "https://login.microsoftonline.com/common/oauth2/v2.0/token"
        headers = {
            "User-Agent": 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:137.0) Gecko/20100101 Thunderbird/137.0',
            "Content-Type": 'application/x-www-form-urlencoded;charset=UTF-8',
        }
        data = {
            "client_id": self.client_id,
            "grant_type": "refresh_token",
            "refresh_token": self.refresh_token,
            "scope": "https://outlook.office.com/IMAP.AccessAsUser.All offline_access",
        }

        response = requests.post(url, headers=headers, data=data, verify=False)
        if response.status_code != 200:
            error_text = response.text
            if "invalid_grant" in error_text and "70000" in error_text:
                print("\n[安全警告] 账户需要安全验证！")
                print("微软检测到您的账户可能存在安全风险，需要进行额外验证。")
                print("请登录 https://account.microsoft.com 或 https://account.live.com 进行安全验证")
                print("验证完成后，您需要重新获取refreshtoken并更新Excel文件\n")
            raise Exception(f"获取access token失败: {error_text}")

        response_data = response.json()
        self.access_token = response_data["access_token"]
        if "refresh_token" in response_data:
            self.refresh_token = response_data["refresh_token"]

        return self.access_token


if __name__ == "__main__":
    try:
        # Example usage
        client = EmailClient(
            username="[email protected]",
            password="123",
            email_address="[email protected]"
        )

        # Authenticate and get tokens
        client.authenticate_oauth()
        #
        # Read emails with GitHub keyword
        emails = client.read_emails(mailbox="INBOX",
                limit=5,keyword='BODY "jetbrain"')

        # Print email details
        for i, email_data in enumerate(emails, 1):
            print(f"\n邮件 {i}:")
            print(f"发件人: {email_data.get('from', 'N/A')}")
            print(f"主题: {email_data.get('subject', 'N/A')}")
            print(f"日期: {email_data.get('date', 'N/A')}")
            print(f"文本内容: {email_data.get('text', 'N/A')[:1000]}...")
            print("-" * 50)

        # Send a test email
        # client.send_email(
        #     receiver="[email protected]",
        #     subject="Test Email",
        #     body="This is a test email"
        # )

    except Exception as e:
        print(f"Error: {str(e)}")