Delete 收发邮件.py

收发邮件.py

@@ -1,316 +0,0 @@
-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)}")
-