系列文章目錄
python辦公自動化–數據可視化(pandas+matplotlib)–生成條形圖和餅狀圖
python辦公自動化–數據可視化(pandas+matplotlib)–生成折線圖
python辦公自動化–數據可視化(pandas讀取excel文件,matplotlib生成可視化圖表)
python辦公自動化-openpyxl學習-工資表生成工資條
python辦公自動化–使用將csv大文件分割為xlsx小文件
python辦公自動化----使用pandas和os合并多個訂單表
三種方法批量填充訂單表中的空白單元格–python,excel vba,excel
python辦自動化–批量發送帶附件的變量郵件(示例:給員工發送工資條)
文章目錄
- 系列文章目錄
- 前言
- 一、登錄郵箱
- 二、記錄日志
- 三、搜索解析郵件
- 四、設置搜索條件
- 1.前五十封
- 2.發件人白名單
- 3.郵件主題白名單
- 4.發件日期為今天
- 五、下載.xlsx附件
- 六、關閉郵箱安全退出
- 七.代碼整合(可直接使用)
- 總結
前言
今天我們來學習使用python讀取郵箱中的郵件,并且將里面的附件表格下載到我們本地的文件夾中。
我們這個系列叫做python自動化辦公,要實現真正意義上的辦公自動化,就是要解放我們的雙手,實現全過程的自動化。基本上數據分析的工作分為以下流程:收集數據----清晰整理數據-----分析數據-----發送結果
我們前面學習了使用pandas清洗整理數據(結合我的另一個專欄–python數據分析),學習了python發送帶附件的變量郵件(發送結果),至于最精彩最關鍵的分析數據這個階段,我們也略微接觸了一些,今天我們就好好學習一下python讀取郵件。
情景設置:你是公司的數據分析師,每天各個部門的負責人每天八點會將昨天的數據作為附件表格,以“order”和“kucun”作為主題發送到你的郵箱,你需要使用將每天的郵件里面的附件下載到本地的E盤的“附件下載”這個文件夾中。
一、登錄郵箱
# 導入imaplib庫用于IMAP協議郵箱操作
import imaplib
# 郵箱登錄用戶名
email_user = "youremail"
# 郵箱授權碼(非登錄密碼),用于IMAP協議認證
email_password = "yourpassword"
# IMAP服務器地址,Foxmail使用QQ郵箱的IMAP服務器
imap_server = "imap.qq.com"
mail = imaplib.IMAP4_SSL(imap_server)
# 使用用戶名和授權碼登錄郵箱
mail.login(email_user, email_password)
print("郵箱登錄成功")
首先我們要在郵箱的設置界面打開我們的smtp/imap/pop3等服務,在這個過程中,我們會獲得一個授權碼,通過這個授權碼,我們就可以將連接到我們的郵箱。我這里登錄的是QQ郵箱,所以服務器地址就是“imap.qq.com”,那么大家在嘗試登錄的時候,就要先把上面的用戶名,授權碼以及郵箱地址改為自己的地址和授權碼。
登錄成功以后,我們才可以進行下面的操作。
二、記錄日志
因為我們讀取郵件下載附件這個動作并不是只做一次,所以我們需要設置日志來記錄腳本的運行信息。
# 導入logging庫用于記錄日志
import logging
# 導入os庫用于文件和目錄操作
import os
# 附件下載目錄,使用雙反斜杠轉義
download_folder = "E:\\附件下載"# 日志系統配置
# ============
logging.basicConfig(level=logging.DEBUG, # 設置日志級別為DEBUG,記錄所有級別日志format="%(asctime)s - %(levelname)s - %(message)s", # 日志格式: 時間-級別-消息handlers=[# 文件處理器,將日志寫入到下載目錄中的email_downloader.log文件logging.FileHandler(os.path.join(download_folder, "email_downloader.log")),# 控制臺處理器,將日志輸出到標準輸出logging.StreamHandler()]
)
# 獲取當前模塊的日志記錄器實例
logger = logging.getLogger(__name__)
這里我們在E盤新建一個文件夾叫做“附件下載”,在這里面我們會新建一個log文件來保存日志
三、搜索解析郵件
# 導入email.utils用于郵件日期解析
import email.utils
# 導入email.header用于解碼郵件頭
import email.header
# 選擇收件箱文件夾
mail.select("inbox")
# 執行IMAP搜索命令
status, messages = mail.search(None, search_criteria)
if status != "OK": # 檢查搜索是否成功raise Exception("郵件搜索失敗")
# 解析郵件內容為Message對象
email_message = email.message_from_bytes(msg_data[0][1])
四、設置搜索條件
# 1. FROM 白名單中的任一發件人(使用OR連接)
# 2. SINCE 今天日期
# 3. HAS attachment 必須包含附件
from_clause = " OR ".join([f'FROM "{sender}"' for sender in sender_whitelist])
search_criteria = f'({from_clause} SINCE "{today}" HAS attachment)'
logger.info(f"嚴格模式: 只處理今天({today})收到的郵件")
logger.info(f"發件人白名單: {sender_whitelist}")
logger.info(f"主題白名單: {subject_whitelist}")
logger.info(f"搜索條件: {search_criteria}")
1.前五十封
代碼如下(示例):
# 獲取郵件ID列表并只取最后50個(最新的50封)
mail_ids = messages[0].split()[-50:]
logger.info(f"找到 {len(mail_ids)} 封匹配郵件,只處理最近50封")# 從最新到最舊處理郵件
for mail_id in reversed(mail_ids):# 獲取郵件完整內容(RFC822格式)status, msg_data = mail.fetch(mail_id, "(RFC822)")if status != "OK": # 如果獲取失敗則跳過continue
通常我們的郵箱每天不會收到太多郵件,50封這個量對于一天來說應該綽綽有余了
2.發件人白名單
# 發件人郵箱白名單列表
sender_whitelist = ["fajianren1@163.com", # 示例郵箱1"fajianren2@gmail.com", # 示例郵箱2
]
if from_header not in sender_whitelist: # 嚴格檢查白名單logger.debug(f"跳過非白名單發件人: {from_header}")continue
通常每天給我們發送郵件的人是固定的,我們只需要下載他們發送的郵件里面的附件,因此添加發件人白名單這一個限制條件。
3.郵件主題白名單
# 主題白名單列表(精確匹配)
subject_whitelist = ["order", # 示例主題1"kucun"
]
subject = email_message['Subject']
if subject not in subject_whitelist: # 嚴格檢查主題白名單logger.debug(f"跳過非白名單主題: {subject}")continue
有時候這部門負責人給我們發的郵件并不是昨天的銷售數據,所以我們要用郵件主題進行區分
這里和上面的發件人白名單是一樣的處理邏輯,這里注意兩點;1.使用英文作為郵件主題,比如上面的"order"和"kucun";2.注意大小寫,這里是嚴格匹配,所以大小寫需要保持一致
4.發件日期為今天
# 從datetime導入datetime用于日期時間處理
from datetime import datetime# 郵件日期驗證
# ===========
try:# 解析郵件日期頭mail_date = email.utils.parsedate_to_datetime(email_message['Date']) if email_message['Date'] else Noneif mail_date and mail_date.date() != datetime.now().date(): # 如果不是今天則跳過logger.debug(f"跳過非今天({mail_date.date()})收到的郵件")continue
這里我們需要設置時間為條件,確保我們每天下載的都是最新的數據,要不然會重復下載前面幾天的數據。
五、下載.xlsx附件
# 遍歷郵件各部分
# =============
for part in email_message.walk(): # 遞歸遍歷郵件所有部分if part.get_content_maintype() == "multipart": # 跳過multipart容器部分continue# 檢查是否為.xlsx附件# ==================filename = part.get_filename() # 獲取附件文件名content_type = part.get_content_type() # 獲取內容類型# 判斷條件:# 1. 文件名以.xlsx結尾,或# 2. 內容類型是Excel文件if (filename and filename.lower().endswith(".xlsx")) or \(content_type == "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"):# 處理無文件名的情況# ================if not filename: # 如果附件沒有文件名# 使用郵件ID生成唯一文件名filename = f"attachment_{mail_id.decode()}.xlsx"# 文件名處理# =========# 解碼郵件頭中的文件名decoded_name = email.header.decode_header(filename)[0][0]if isinstance(decoded_name, bytes):decoded_name = decoded_name.decode('utf-8')today_str = datetime.now().strftime("%Y%m%d") # 獲取當前日期字符串base_name, ext = os.path.splitext(decoded_name) # 拆分文件名和擴展名# 生成新文件名格式:YYYYMMDD-原文件名.xlsxnew_filename = f"{today_str}-{base_name}{ext}"# 拼接完整的文件保存路徑filepath = os.path.join(download_folder, new_filename)# 附件保存處理# ===========if not os.path.exists(filepath): # 檢查文件是否已存在# 以二進制模式保存附件with open(filepath, "wb") as f:f.write(part.get_payload(decode=True)) # 解碼并寫入附件內容logger.info(f"下載附件: {filename} -> {new_filename}")else: # 文件已存在則跳過logger.info(f"附件已存在,跳過: {new_filename}")
首先我們根據文件類型篩選出.xlsx格式的附件,然后我們將當天的日期和附件的名字結合在一起,作為新名字,最后我們將附件以新名字命名保存到E盤的‘附件下載”這個文件夾中
六、關閉郵箱安全退出
# 關閉郵箱連接# ===========mail.close() # 關閉當前郵箱文件夾mail.logout() # 退出IMAP會話logger.info("處理完成") # 記錄處理完成信息except Exception as e: # 異常處理部分# 記錄錯誤信息,包含堆棧跟蹤logger.error(f"發生錯誤: {str(e)}", exc_info=True) # 確保郵箱連接被正確關閉if 'mail' in locals() and mail.state != 'LOGOUT': # 檢查mail對象是否存在且未退出try:mail.logout() # 嘗試安全退出except:pass # 忽略退出時的任何錯誤if __name__ == "__main__":# 當腳本直接運行時執行主函數download_specific_attachments()
七.代碼整合(可直接使用)
# 導入imaplib庫用于IMAP協議郵箱操作
import imaplib
# 導入email庫用于解析郵件內容
import email
# 導入os庫用于文件和目錄操作
import os
# 導入logging庫用于記錄日志
import logging
# 從datetime導入datetime用于日期時間處理
from datetime import datetime
# 導入email.utils用于郵件日期解析
import email.utils
# 導入email.header用于解碼郵件頭
import email.headerdef download_specific_attachments():"""下載指定發件人今天發送的郵件中的.xlsx附件主要功能:1. 連接IMAP郵箱服務器2. 搜索指定發件人今天發送的郵件3. 只處理最近20封郵件4. 解析郵件內容并識別.xlsx附件5. 下載附件并按指定格式重命名6. 記錄完整操作日志"""# 郵箱服務器配置部分# ====================# 郵箱登錄用戶名email_user = "youremail"# 郵箱授權碼(非登錄密碼),用于IMAP協議認證email_password = "yourpassword"# IMAP服務器地址,Foxmail使用QQ郵箱的IMAP服務器imap_server = "imap.qq.com"# 發件人郵箱白名單列表sender_whitelist = ["fajianren1@163.com", # 示例郵箱1"fajianren2@gmail.com", # 示例郵箱2]# 主題白名單列表(精確匹配)subject_whitelist = ["order" , # 示例主題1"kucun" ]# 附件下載目錄,使用雙反斜杠轉義download_folder = "E:\\附件下載"# 日志系統配置# ============logging.basicConfig(level=logging.DEBUG, # 設置日志級別為DEBUG,記錄所有級別日志format="%(asctime)s - %(levelname)s - %(message)s", # 日志格式: 時間-級別-消息handlers=[# 文件處理器,將日志寫入到下載目錄中的email_downloader.log文件logging.FileHandler(os.path.join(download_folder, "email_downloader.log")),# 控制臺處理器,將日志輸出到標準輸出logging.StreamHandler()])# 獲取當前模塊的日志記錄器實例logger = logging.getLogger(__name__)# 創建附件下載目錄# exist_ok=True表示如果目錄已存在不會報錯os.makedirs(download_folder, exist_ok=True)try:# 郵箱連接和登錄部分# ==================logger.info("正在連接到IMAP服務器...")# 創建IMAP4_SSL安全連接對象mail = imaplib.IMAP4_SSL(imap_server)# 使用用戶名和授權碼登錄郵箱mail.login(email_user, email_password)# 記錄登錄成功狀態logger.info("登錄成功")# 選擇收件箱文件夾mail.select("inbox")# 構造IMAP搜索條件# ===============# 獲取今天的日期,格式化為IMAP要求的格式(如"21-Jul-2025")today = datetime.now().strftime("%d-%b-%Y")# 構造IMAP搜索條件:# 1. FROM 白名單中的任一發件人(使用OR連接)# 2. SINCE 今天日期# 3. HAS attachment 必須包含附件from_clause = " OR ".join([f'FROM "{sender}"' for sender in sender_whitelist])search_criteria = f'({from_clause} SINCE "{today}" HAS attachment)'logger.info(f"嚴格模式: 只處理今天({today})收到的郵件")logger.info(f"發件人白名單: {sender_whitelist}")logger.info(f"主題白名單: {subject_whitelist}")logger.info(f"搜索條件: {search_criteria}")# 執行IMAP搜索命令status, messages = mail.search(None, search_criteria)if status != "OK": # 檢查搜索是否成功raise Exception("郵件搜索失敗")# 處理搜索結果# =============# 獲取郵件ID列表并只取最后20個(最新的20封)mail_ids = messages[0].split()[-20:]logger.info(f"找到 {len(mail_ids)} 封匹配郵件,只處理最近20封")# 從最新到最舊處理郵件for mail_id in reversed(mail_ids):# 獲取郵件完整內容(RFC822格式)status, msg_data = mail.fetch(mail_id, "(RFC822)")if status != "OK": # 如果獲取失敗則跳過continue# 解析郵件內容為Message對象email_message = email.message_from_bytes(msg_data[0][1])# 驗證發件人是否在白名單中# ========================# 驗證郵件主題是否在白名單中,并處理中文漢字主題# ========================from_header = email.utils.parseaddr(email_message['From'])[1] # 解析發件人郵箱if from_header not in sender_whitelist: # 嚴格檢查白名單logger.debug(f"跳過非白名單發件人: {from_header}")continuetry:# 嘗試將郵件主題解碼為utf-8,處理可能的中文字符問題subject = email_message['Subject'].encode('latin1').decode('utf-8')except UnicodeDecodeError:# 如果解碼失敗,則記錄日志并跳過該郵件logger.warning("郵件主題編碼問題,跳過該郵件")continueif subject not in subject_whitelist: # 嚴格檢查主題白名單logger.debug(f"跳過非白名單主題: {subject}")continue# 郵件日期驗證# ===========try:# 解析郵件日期頭mail_date = email.utils.parsedate_to_datetime(email_message['Date']) if email_message['Date'] else Noneif mail_date and mail_date.date() != datetime.now().date(): # 如果不是今天則跳過logger.debug(f"跳過非今天({mail_date.date()})收到的郵件")continueelif not mail_date: # 如果郵件沒有日期信息則默認處理logger.debug("郵件無日期信息,默認處理")except Exception as e: # 日期解析錯誤處理logger.warning(f"日期解析錯誤: {str(e)},默認處理")# 遍歷郵件各部分# =============for part in email_message.walk(): # 遞歸遍歷郵件所有部分if part.get_content_maintype() == "multipart": # 跳過multipart容器部分continue# 檢查是否為.xlsx附件# ==================filename = part.get_filename() # 獲取附件文件名content_type = part.get_content_type() # 獲取內容類型# 判斷條件:# 1. 文件名以.xlsx結尾,或# 2. 內容類型是Excel文件if (filename and filename.lower().endswith(".xlsx")) or \(content_type == "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"):# 處理無文件名的情況# ================if not filename: # 如果附件沒有文件名# 使用郵件ID生成唯一文件名filename = f"attachment_{mail_id.decode()}.xlsx"# 文件名處理# =========# 解碼郵件頭中的文件名decoded_name = email.header.decode_header(filename)[0][0]if isinstance(decoded_name, bytes):decoded_name = decoded_name.decode('utf-8')today_str = datetime.now().strftime("%Y%m%d") # 獲取當前日期字符串base_name, ext = os.path.splitext(decoded_name) # 拆分文件名和擴展名# 生成新文件名格式:YYYYMMDD-原文件名.xlsxnew_filename = f"{today_str}-{base_name}{ext}"# 拼接完整的文件保存路徑filepath = os.path.join(download_folder, new_filename) # 附件保存處理# ===========if not os.path.exists(filepath): # 檢查文件是否已存在# 以二進制模式保存附件with open(filepath, "wb") as f: f.write(part.get_payload(decode=True)) # 解碼并寫入附件內容logger.info(f"下載附件: {filename} -> {new_filename}") else: # 文件已存在則跳過logger.info(f"附件已存在,跳過: {new_filename}") # 關閉郵箱連接# ===========mail.close() # 關閉當前郵箱文件夾mail.logout() # 退出IMAP會話logger.info("處理完成") # 記錄處理完成信息except Exception as e: # 異常處理部分# 記錄錯誤信息,包含堆棧跟蹤logger.error(f"發生錯誤: {str(e)}", exc_info=True) # 確保郵箱連接被正確關閉if 'mail' in locals() and mail.state != 'LOGOUT': # 檢查mail對象是否存在且未退出try:mail.logout() # 嘗試安全退出except:pass # 忽略退出時的任何錯誤if __name__ == "__main__":# 當腳本直接運行時執行主函數download_specific_attachments()
點擊運行后,在我本地的E盤里面的“附件下載”這個文件夾內容如下
可以看到,成功下載了附件以及生成了log日志
那么大家在使用上面的代碼的時候,需要修改的就是郵箱,授權碼,以及發件人白名單和郵件主題白名單了。
總結
今天我們學習了用python讀取郵箱里面的郵件,并且下載附件到本地,我們還設置了時間、發件人以及郵件主題等多個限制條件。其實這里面可以玩的東西遠不止這么多,我大家可以自行去探索。
OK,那么到今天為止,我們就講完了python發郵件以及收郵件,我們得到了數據源,那么下階段將正式開始學習python處理數據,當然主要是excel表格的數據,包括一些匹配、聚合等等,當然如果有空也會更新python處理word或者pptx等。
至于下階段的更新,我會放在python數據分析這個專欄。如果大家對上面的文章有不懂的地方,也可以隨時私信我。同樣的,如果大家發現哪里不對,也歡迎批評指正哈。
大家有興趣的可以點個關注以及免費的贊贊喲,愛你們!!