使用IMAP服務獲取163郵箱的未讀郵件
整體的邏輯思路如下:
- 開啟163郵箱的IMAP服務,拿到授權碼用于登錄IMAP服務
- 登錄IMAP服務,獲取郵箱的未讀郵件列表
- 遍歷未讀郵件列表,獲取郵件內容
# 導入必要的庫
import os
import imaplib
import ssl
import email
from email.header import decode_header
from email.utils import parseaddr
1. 開啟163郵箱的IMAP服務,拿到授權碼用于登錄IMAP服務
163郵箱設置IMAP服務,需要到郵箱設置頁面,選擇“郵箱設置”->“POP3/IMAP”->“開啟IMAP服務”,然后點擊“保存”,即可開啟IMAP服務。
2. 登錄IMAP服務,獲取郵箱的未讀郵件列表
# 創建一個默認的SSL上下文對象,用于服務器認證
# 參數設置為None,表示使用默認值,后續將通過代碼明確指定SSL/TLS版本范圍
ssl_context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=None, capath=None, cadata=None)
# 指定SSL/TLS的最小版本為TLS 1.2,以確保連接使用的協議不低于此版本
ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2# 指定SSL/TLS的最大版本為TLS 1.3,以確保連接不會使用超出此版本的協議
# TLS 1.3是最新的TLS版本,提供了更強的安全性和加密方法
ssl_context.maximum_version = ssl.TLSVersion.TLSv1_3
def login163(user=None, pwd=None, host="imap.163.com" ):# 嘗試建立SSL加密的IMAP連接并登錄# 異常處理增強try:imap_client = imaplib.IMAP4_SSL(host, ssl_context=ssl_context) # 確保使用TLS 1.2imap_client.login(user, pwd)except imaplib.IMAP4.error as e:print(f"登錄失敗: {e}")return# 發送ID命令給服務器,提供客戶端信息imaplib.Commands["ID"] = ('AUTH',)args = ("name", user, "contact", user, "version", "1.0.0", "vendor", "myclient")imap_client._simple_command("ID", str(args).replace(",", "").replace("'", "\""))return imap_client
3. 遍歷未讀郵件列表,獲取郵件內容
def fetch_and_mark_emails_as_read(imap_client,mailbox="INBOX"):"""從指定的郵箱中獲取未讀郵件,并將這些郵件標記為已讀。:param imap_client: IMAP客戶端對象,用于與郵件服務器交互:param mailbox: string, 指定的郵箱,默認為"INBOX"(收件箱)"""# 選擇郵箱,默認為收件箱imap_client.select(mailbox)# 搜索未讀郵件typ, dat = imap_client.search(None, "UNSEEN")# 解碼郵件編號字符串str_numbers = dat[0].decode('utf-8')# 將編號字符串分割成列表numbers_str_list = str_numbers.split()# 遍歷未讀郵件for msg in numbers_str_list:try:# 獲取郵件內容_, data = imap_client.fetch(msg, '(RFC822)')raw_email = data[0][1]email_message = email.message_from_bytes(raw_email)# 解碼郵件主題subject = decode_email_header(email_message['Subject'])# 解碼發件人信息from_info = decode_header(email_message['From'])name_parts = []for part, charset in from_info:if isinstance(part, bytes):if charset:decoded_part = part.decode(charset)else:decoded_part = part.decode('utf-8', errors='replace')else:decoded_part = partname_parts.append(decoded_part)full_name = ''.join(name_parts).strip() from_person,_=parseaddr(full_name)# 提取郵件正文all_text_body = extract_text_body(email_message)#------------------------------#你對得到的郵件的處理代碼#------------------------------# 標記郵件為已讀imap_client.store(msg, '+FLAGS', '\\Seen')except imaplib.IMAP4.error as e:print(f"處理郵件或標記已讀失敗: {e}")
郵件主題解碼函數:
def decode_email_header(header):"""解碼電子郵件頭信息。電子郵件頭信息可能包含多種編碼,這個函數旨在解析頭信息并返回解碼后的字符串。如果頭信息是ASCII碼,則直接返回;如果是非ASCII碼,會根據編碼類型進行解碼。參數:header (str): 需要解碼的電子郵件頭信息。返回:str: 解碼后的字符串。如果解碼過程中包含多種編碼,會返回一個元組,包含解碼后的字符串和對應的編碼類型。"""# 解碼頭信息,返回一個包含解碼結果和編碼類型的元組# 列表decoded_header = decode_header(header)[0]# print(decoded_header)# 判斷解碼結果是否為元組,如果是,說明存在多種編碼,需要進一步解碼if isinstance(decoded_header, tuple):# 對元組中的字符串進行解碼,并返回解碼后的結果part, charset = decoded_headerif isinstance(part, bytes): # 檢查是否為字節串if charset: # 如果有字符集,按指定字符集解碼decoded_part = part.decode(charset)else: # 沒有指定字符集時嘗試UTF-8解碼,或選擇其他策略decoded_part = part.decode('utf-8', errors='replace')else: # 如果已經是字符串,直接使用decoded_part = partreturn decoded_partelse:# 如果解碼結果不是元組,直接返回解碼后的字符串return None
正文獲取部分
def extract_text_body(email_message):"""從電子郵件消息中提取純文本正文。參數:email_message: 一個電子郵件消息對象,可以是使用Python email庫構建的或從文件中讀取的。返回:一個字符串,包含電子郵件的純文本正文。如果沒有找到純文本正文或郵件為空,則返回空字符串。"""# 初始化一個字符串,用于存儲提取的文本正文all_text_body = ""# 遍歷電子郵件的每一個部分for part in email_message.walk():# 獲取當前部分的Content-Typectype = part.get_content_type()# 獲取當前部分的Content-Dispositioncdispo = str(part.get('Content-Disposition'))# 檢查當前部分是否為純文本且不是附件if ctype == 'text/plain' and 'attachment' not in cdispo:# 獲取當前部分的payload(實際內容),并解碼body = part.get_payload(decode=True)# 獲取當前部分的內容字符集charset = part.get_content_charset()# 將解碼后的文本添加到存儲所有文本正文的字符串中all_text_body += body.decode(charset, errors='replace')# 找到純文本正文后立即終止循環,以提高效率break # 如果找到文本部分,即停止搜索,提高效率# 返回收集到的所有文本正文return all_text_body
測試代碼
def imap_mail_get(client):retry_limit = 3for i in range(retry_limit):try:if client is None:client = login163(username, password)#password為IMAP授權碼# 對兩個郵箱進行操作for mailbox in ["INBOX", "Sent"]:fetch_and_mark_emails_as_read(client, mailbox=mailbox)return client except Exception as e:pass