## 1. 为什么你需要一个自己的邮件机器人？ 每天上班第一件事，打开邮箱，几十封未读邮件扑面而来。有系统自动发送的报告，有同事发来的表格，还有一堆需要下载、整理、汇总的附件。手动一封封点开、下载、处理，一上午的时间就这么没了。这种重复、枯燥但又不得不做的活儿，你是不是也受够了？ 我刚开始工作那会儿，就经常被这类邮件处理任务搞得焦头烂额。直到我开始用Python写脚本，把这些活儿都交给程序去干。早上到公司，泡杯咖啡的功夫，脚本已经帮我收好了所有附件，把数据整理成报告，甚至把处理结果都回复给了发件人。那种解放双手的感觉，真的太爽了。 今天我要跟你分享的，就是这套让我效率倍增的“秘密武器”——基于Python `imaplib`和`email`库的邮件自动化处理方案。别被这两个库的名字吓到，它们都是Python自带的，不需要额外安装，而且用起来比你想象的要简单得多。我会手把手带你，从一个完全不懂邮件协议的小白，到能写出一个健壮、实用的邮件处理机器人。我们不仅会讲怎么连接邮箱、下载附件，还会深入聊聊我踩过的那些坑，以及怎么让你的脚本更聪明、更稳定。 ## 2. 动手之前：环境准备与核心库扫盲 工欲善其事，必先利其器。在开始写代码之前，我们得先把“工具箱”准备好。放心，不需要复杂的配置，Python环境你肯定有，我们需要的库也基本都是“开箱即用”的。 首先，打开你的代码编辑器，新建一个Python文件，比如就叫 `email_robot.py`。然后，我们把今天要用到的几个核心“队员”请出来： ```python import imaplib # 负责跟邮件服务器“握手”和“对话”的通信专家 import email # 负责把服务器传回来的“乱码”邮件，拆解成我们能看懂的部分 from email.header import decode_header # 专门对付那些带着奇怪编码的邮件主题和发件人名字 import re # 正则表达式，用来从复杂的字符串里精准地“抠”出邮箱地址 import os # 帮我们在电脑上创建文件夹、保存附件文件 from datetime import datetime # 记录处理时间，给报告加个时间戳 ``` 你可能注意到了，我没有提 `pandas` 和 `chardet`。原始文章里提到了它们，这很好，但我想根据我的经验给你更灵活的选择。`pandas` 确实是处理Excel、CSV数据的利器，但如果你只是处理文本附件，或者想用更轻量的 `csv` 库，完全可以不装。`chardet` 用于自动检测编码，对于处理来源复杂的邮件非常有用，但它是个第三方库，需要 `pip install chardet` 来安装。我的建议是，先不用急着装，等我们遇到编码问题再请它出马也不迟。这样能让我们的环境更干净。 这里重点说说 `imaplib` 和 `email`。你可以把它们想象成邮差和翻译官。`imaplib` 就是那个邮差，它负责跑到邮件服务器（比如网易、腾讯的服务器）那里，按照IMAP这个“国际通用邮差手语”，把指令（“给我找未读邮件”）传达过去，再把服务器返回的一堆原始数据搬回来。而 `email` 库就是翻译官，它能把邮差搬回来的、像天书一样的原始邮件数据，翻译成我们可以理解的“发件人是谁”、“主题是什么”、“正文和附件在哪里”。 理解了这个分工，后面的代码你就会觉得特别顺理成章。我们指挥邮差（imaplib）去干活，然后把拿回来的东西交给翻译官（email）去解析，最后我们直接看翻译结果就行。 ## 3. 第一步：让你的程序成功“登录”邮箱 万事开头难，和邮件服务器建立连接就是第一步。这一步的核心是找到正确的“地址”和“口令”。这里有个**大坑**我当年踩过：密码不是你的邮箱登录密码！ 现在大多数邮箱服务商（比如Gmail、QQ邮箱、163邮箱等），为了安全，都要求使用“授权码”或“应用专用密码”来让第三方程序登录。你得去邮箱的设置里，找到“POP3/IMAP/SMTP服务”或“安全设置”相关选项，开启IMAP服务，并生成一个专属的授权码。这个生成的字符串，才是我们代码里要用的 `password`。 假设我们已经拿到了授权码，接下来就是连接代码。我习惯用一个函数把连接过程包装起来，这样逻辑清晰，出错也容易处理。 ```python def connect_to_email(imap_url, username, password): """ 连接到IMAP邮件服务器 :param imap_url: 邮箱的IMAP服务器地址 :param username: 你的邮箱账号 :param password: 邮箱的授权码（不是登录密码！） :return: 一个已经登录成功的IMAP连接对象，或者None（如果失败） """ try: print(f"正在尝试连接服务器: {imap_url}...") # 使用IMAP4_SSL进行加密连接，这是标准做法，保证通信安全 mail = imaplib.IMAP4_SSL(imap_url) # 登录账户 mail.login(username, password) print("连接成功！") return mail except imaplib.IMAP4.error as e: print(f"登录失败，请检查账号、授权码或服务器地址: {e}") except Exception as e: print(f"连接过程中发生未知错误: {e}") return None ``` 怎么用这个函数呢？我们先把邮箱信息准备好。**重要提示：千万不要把密码直接写在代码里然后上传到GitHub！** 这是安全大忌。我推荐两种方法：一是用环境变量，二是用本地的配置文件。 **方法一：使用环境变量（推荐）** 在运行脚本前，在终端里设置（Linux/Mac）： ```bash export EMAIL_USER='your_email@example.com' export EMAIL_PASS='your_authorization_code' ``` 或者在代码里使用 `os.getenv` 读取： ```python import os username = os.getenv('EMAIL_USER') password = os.getenv('EMAIL_PASS') imap_url = 'imap.example.com' # 替换成你的邮箱服务器 mail = connect_to_email(imap_url, username, password) ``` **方法二：使用配置文件** 创建一个 `config.json` 文件（记得加入 `.gitignore`）： ```json { "email": { "username": "your_email@example.com", "password": "your_authorization_code", "imap_server": "imap.example.com" } } ``` 然后在代码中读取： ```python import json with open('config.json', 'r') as f: config = json.load(f) email_cfg = config['email'] mail = connect_to_email(email_cfg['imap_server'], email_cfg['username'], email_cfg['password']) ``` 不同邮箱的IMAP服务器地址不一样，这里给你列个常见的，方便你替换： - **QQ邮箱/ Foxmail**: `imap.qq.com` (端口993) - **163/126网易邮箱**: `imap.163.com` 或 `imap.126.com` - **Gmail**: `imap.gmail.com` - **Outlook/Hotmail**: `outlook.office365.com` 连接成功后，你就获得了一个 `mail` 对象，后续所有操作，比如查邮件、删邮件，都通过它来指挥服务器。 ## 4. 精准定位：如何搜索和筛选你需要的邮件 成功登录后，你的邮箱里可能躺着成千上万封邮件。我们不可能全部处理，所以必须学会“搜索”。IMAP协议提供了一套强大的搜索命令，而 `imaplib` 让我们能用简单的函数调用它们。 首先，我们需要“进入”一个文件夹。默认是收件箱（‘INBOX’），但你也可以处理“已发送”、“垃圾邮件”或其他自定义文件夹。 ```python def search_emails(mail, folder='INBOX', criteria='UNSEEN'): """ 在指定邮箱文件夹中搜索符合条件的邮件 :param mail: 已连接的IMAP对象 :param folder: 邮箱文件夹，如 'INBOX', 'Sent', 'Junk' :param criteria: 搜索条件字符串 :return: 符合条件的邮件ID列表 """ try: # 1. 选择文件夹 status, _ = mail.select(folder, readonly=False) # readonly=False 表示我们可能要修改邮件（如标记已读） if status != 'OK': print(f"无法选择文件夹: {folder}") return [] # 2. 执行搜索 status, message_ids_bytes = mail.search(None, criteria) if status != 'OK': print("邮件搜索失败") return [] # 3. 处理结果：返回一个邮件ID的字符串列表 # mail.search返回的ID是字节字符串，如 b'1 2 3 10' ids = message_ids_bytes[0].split() # 按空格分割，得到 [b'1', b'2', b'3', b'10'] print(f"在文件夹【{folder}】中找到 {len(ids)} 封符合条件的邮件。") return ids except Exception as e: print(f"搜索邮件时出错: {e}") return [] ``` 这个 `criteria` 参数是搜索的灵魂。它就像你邮箱的搜索框里输入的关键词，但格式是IMAP规定的。下面这些是我最常用的几种搜索条件，你可以像搭积木一样组合它们： - **`'ALL'`**: 所有邮件。慎用，除非你想处理整个邮箱。 - **`'UNSEEN'`**: 未读邮件。这是自动化脚本最常用的，处理新邮件。 - **`'SEEN'`**: 已读邮件。 - **`'FROM "sender@example.com"`'**: 来自特定发件人。注意引号。 - **`'SUBJECT "月度报告"`'**: 主题包含“月度报告”。支持中文，但确保编码正确。 - **`'SINCE "01-APR-2024"`'**: 2024年4月1日之后（含当天）的邮件。日期格式是`日-月-年`。 - **`'BEFORE "01-MAY-2024"`'**: 2024年5月1日之前的邮件。 - **组合搜索**：`'UNSEEN FROM "boss@company.com" SUBJECT "紧急"'` 查找来自老板的、主题含“紧急”的未读邮件。 搜索返回的是一串邮件ID。注意，这些ID是邮件在当前文件夹中的**序列号**，不是全局唯一的，而且可能会变（比如你删除了前面的邮件）。所以，我们通常是在一次会话中，获取ID后立刻处理对应的邮件内容。 ## 5. 拆解邮件：从乱码到清晰信息的解析术 拿到了邮件ID，下一步就是把邮件内容“下载”下来并解析。服务器返回的原始邮件数据遵循MIME格式，是一大坨包含了头部信息、正文、附件编码数据的字节流。`email` 库的价值就在这里体现。 我们写一个解析函数，目标是提取出对我们有用的信息：发件人、主题、日期、正文文本和附件。 ```python def parse_email(mail, msg_id): """ 根据邮件ID，获取并解析一封邮件的详细内容 :param mail: IMAP连接对象 :param msg_id: 邮件ID（字节字符串） :return: 包含邮件关键信息的字典 """ try: # 1. 获取原始邮件数据。'(RFC822)' 表示获取完整的邮件源码 status, msg_data = mail.fetch(msg_id, '(RFC822)') if status != 'OK': print(f"获取邮件 {msg_id} 内容失败") return None # msg_data结构复杂，取第一个元组的第二部分才是邮件原始字节流 raw_email_bytes = msg_data[0][1] # 2. 将字节流解析为email.message.Message对象 email_message = email.message_from_bytes(raw_email_bytes) # 3. 解析发件人 (From) from_header = email_message.get('From', '') # decode_header 能处理编码问题，比如 =?UTF-8?B?...?= 这种格式 decoded_from, charset = decode_header(from_header)[0] if isinstance(decoded_from, bytes): sender_name = decoded_from.decode(charset or 'utf-8', errors='ignore') else: sender_name = str(decoded_from) # 用正则表达式从发件人字符串中提取出纯邮箱地址 sender_email_match = re.search(r'[\w\.-]+@[\w\.-]+\.\w+', from_header) sender_email = sender_email_match.group(0) if sender_email_match else '' # 4. 解析主题 (Subject) subject_header = email_message.get('Subject', '') decoded_subj, subj_charset = decode_header(subject_header)[0] if isinstance(decoded_subj, bytes): subject = decoded_subj.decode(subj_charset or 'utf-8', errors='ignore') else: subject = str(decoded_subj) # 5. 解析日期 (Date) date_str = email_message.get('Date', 'Unknown') # 6. 解析正文 (Body) # 邮件正文可能有多部分（纯文本+HTML），我们优先取纯文本 body_text = '' if email_message.is_multipart(): for part in email_message.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": # 获取正文内容，并处理编码 payload = part.get_payload(decode=True) charset = part.get_content_charset() or 'utf-8' try: body_text = payload.decode(charset, errors='ignore') except (LookupError, UnicodeDecodeError): # 如果编码检测失败，尝试通用方案 body_text = payload.decode('utf-8', errors='ignore') break # 找到纯文本正文就退出 else: # 非多部分邮件，直接解码 payload = email_message.get_payload(decode=True) charset = email_message.get_content_charset() or 'utf-8' body_text = payload.decode(charset, errors='ignore') return { 'id': msg_id.decode(), 'sender_name': sender_name.strip(), 'sender_email': sender_email, 'subject': subject.strip(), 'date': date_str, 'body_preview': (body_text[:200] + '...') if len(body_text) > 200 else body_text, # 预览前200字符 'raw_message': email_message # 保留原始对象，方便后续处理附件 } except Exception as e: print(f"解析邮件 {msg_id} 时发生错误: {e}") import traceback traceback.print_exc() return None ``` 这个函数有点长，但逻辑是清晰的。核心是 `email.message_from_bytes()`，它把原始数据变成我们可以遍历和查询的结构。`decode_header` 是处理中文或其他非ASCII字符主题/发件人名的关键，它能正确解码那些像“=?UTF-8?B?5Lit5paH?=”的字符串。 关于正文提取，我采用了“优先纯文本”的策略。一封邮件可能同时包含 `text/plain`（纯文本）和 `text/html`（HTML格式）两个版本。对于自动化处理，我们通常只关心文字内容，所以提取纯文本部分就够了。`part.walk()` 方法会递归地遍历邮件的所有部分，`get_payload(decode=True)` 则能解码被 `base64` 或 `quoted-printable` 编码的内容。 ## 6. 核心任务：自动下载与分类处理邮件附件 邮件自动化，一多半的功夫都在处理附件上。可能是Excel报表、PDF合同、图片，或是压缩包。我们的目标是自动识别它们，并保存到本地指定的文件夹，甚至根据内容做进一步处理。 首先，我们写一个通用的附件处理器： ```python def save_attachments(email_message, save_folder='./attachments'): """ 遍历邮件，找出所有附件并保存到本地文件夹 :param email_message: 解析后的email.message对象 :param save_folder: 附件保存目录 :return: 保存的附件信息列表 """ if not os.path.exists(save_folder): os.makedirs(save_folder) # 如果文件夹不存在就创建它 attachment_list = [] for part in email_message.walk(): # 获取内容处置头，用于判断是否是附件 content_disposition = str(part.get("Content-Disposition")) # 判断依据：1. 有文件名；2. 内容处置包含‘attachment’ filename = part.get_filename() is_attachment = filename is not None and ("attachment" in content_disposition.lower()) if is_attachment: # 解码可能被编码过的文件名（特别是中文文件名） decoded_filename_tuple = decode_header(filename)[0] if isinstance(decoded_filename_tuple[0], bytes): # 如果是字节类型，用指定的或默认的utf-8解码 charset = decoded_filename_tuple[1] or 'utf-8' filename = decoded_filename_tuple[0].decode(charset, errors='ignore') else: filename = decoded_filename_tuple[0] # 构建完整的保存路径 filepath = os.path.join(save_folder, filename) # 防止文件名重复，如果已存在，则在文件名后加时间戳 if os.path.exists(filepath): name, ext = os.path.splitext(filename) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"{name}_{timestamp}{ext}" filepath = os.path.join(save_folder, filename) # 将附件内容解码并写入文件 payload = part.get_payload(decode=True) # decode=True 是关键！ with open(filepath, 'wb') as f: # 以二进制写入模式打开 f.write(payload) print(f" 附件已保存: {filename}") attachment_list.append({ 'filename': filename, 'filepath': filepath, 'content_type': part.get_content_type() }) return attachment_list ``` 这个函数会遍历邮件的每个部分，通过 `get_filename()` 和 `Content-Disposition` 头信息来判断是否是附件。`part.get_payload(decode=True)` 这句非常重要，它负责把经过 `base64` 等编码的附件内容解码回原始的二进制数据，我们才能正确保存。 接下来，我们升级一下，让脚本能根据附件类型做不同的处理。这就是自动化的“智能”所在。 ```python def process_attachments_based_on_type(attachments): """ 根据附件类型进行分发处理 :param attachments: save_attachments 返回的附件信息列表 """ for att in attachments: filepath = att['filepath'] content_type = att['content_type'] print(f"处理附件: {att['filename']} (类型: {content_type})") # 根据文件扩展名或MIME类型判断 if att['filename'].lower().endswith(('.xlsx', '.xls')) or 'spreadsheet' in content_type: process_excel_attachment(filepath) elif att['filename'].lower().endswith('.csv') or 'csv' in content_type: process_csv_attachment(filepath) elif att['filename'].lower().endswith(('.pdf', '.PDF')): process_pdf_attachment(filepath) # 需要安装PyPDF2或pdfplumber elif att['filename'].lower().endswith(('.zip', '.rar', '.7z')): process_archive_attachment(filepath) # 需要安装patool或zipfile elif 'image' in content_type: process_image_attachment(filepath) # 可能需要PIL库 else: print(f" 未知类型附件，已保存至: {filepath}") ``` 这里我定义了一个处理器的“路由表”。你可以根据实际需求，为每种文件类型编写对应的 `process_xxx_attachment` 函数。比如，处理Excel的函数可以读取数据并入库；处理PDF的函数可以提取文字；处理图片的函数可以压缩或上传到图床。 ## 7. 实战演练：构建一个日报自动汇总机器人 光说不练假把式。现在我们把前面所有的模块像拼乐高一样组合起来，打造一个真实的自动化场景：**自动处理每日销售日报邮件**。 假设场景：每天上午10点，各区域的销售经理会发来主题为“销售日报”的邮件，附件是一个固定格式的Excel文件。我们需要写一个脚本，自动收取这些邮件，把附件里的数据汇总到一个总表，并生成一份简单的统计报告，最后发一封汇总邮件给老板。 **第一步：组装主流程** ```python import pandas as pd from datetime import datetime, timedelta def daily_sales_report_robot(): """日报自动汇总机器人主函数""" print(f"{'='*50}") print(f"开始执行日报汇总任务 @ {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"{'='*50}") # 0. 配置 IMAP_SERVER = 'imap.163.com' USERNAME = os.getenv('SALES_EMAIL_USER') PASSWORD = os.getenv('SALES_EMAIL_PASS') SAVE_DIR = './daily_reports' TODAY = datetime.now().strftime('%Y-%m-%d') YESTERDAY = (datetime.now() - timedelta(days=1)).strftime('%d-%b-%Y') # IMAP日期格式 # 1. 连接邮箱 mail = connect_to_email(IMAP_SERVER, USERNAME, PASSWORD) if not mail: print("邮箱连接失败，任务终止。") return try: # 2. 搜索邮件：主题包含“销售日报”，并且是昨天至今发送的（避免处理旧邮件） # 搜索条件可以组合：'SUBJECT "销售日报" SINCE "01-APR-2024"' search_criteria = f'SUBJECT "销售日报" SINCE "{YESTERDAY}"' email_ids = search_emails(mail, criteria=search_criteria) if not email_ids: print("今日未找到符合条件的销售日报邮件。") return all_sales_data = [] # 用于存放所有区域的数据 report_summary = [] # 用于存放报告摘要 # 3. 遍历处理每一封邮件 for eid in email_ids: print(f"\n--- 处理邮件ID: {eid.decode()} ---") email_info = parse_email(mail, eid) if not email_info: continue print(f" 发件人: {email_info['sender_name']} <{email_info['sender_email']}>") print(f" 主 题: {email_info['subject']}") # 4. 保存并处理附件 attachments = save_attachments(email_info['raw_message'], SAVE_DIR) for att in attachments: if att['filename'].endswith(('.xlsx', '.xls')): # 调用Excel处理函数，并获取返回的数据 region_data, region_summary = process_sales_excel(att['filepath'], email_info['sender_name']) if region_data is not None: all_sales_data.append(region_data) report_summary.append(region_summary) # 5. （可选）标记邮件为已处理或移动到特定文件夹 # mail.store(eid, '+FLAGS', '\\Seen') # 标记为已读 # mail.copy(eid, 'Processed/SalesReports') # 复制到“已处理/销售报告”文件夹 # mail.store(eid, '+FLAGS', '\\Deleted') # 标记原邮件为删除（需expunge生效） # 6. 汇总所有数据并生成总报告 if all_sales_data: generate_master_report(all_sales_data, report_summary, TODAY) print(f"\n所有日报处理完成，总报告已生成。") else: print(f"\n未找到有效的Excel附件数据。") except Exception as e: print(f"\n!!! 任务执行过程中发生错误: {e}") import traceback traceback.print_exc() finally: # 7. 务必关闭连接 mail.close() mail.logout() print("邮箱连接已关闭。") ``` **第二步：编写核心的数据处理函数** ```python def process_sales_excel(filepath, region_name): """ 处理单个销售Excel文件 :param filepath: Excel文件路径 :param region_name: 区域名称（从发件人解析） :return: (清洗后的DataFrame, 区域摘要字典) """ try: # 读取Excel，假设数据在第一个工作表，且表头在第一行 df = pd.read_excel(filepath, sheet_name=0) # 基础数据清洗（根据你的实际表格结构调整） # 例如：重命名列、删除空行、转换日期格式等 df.columns = [col.strip() for col in df.columns] # 去除列名空格 df['日期'] = pd.to_datetime(df['日期'], errors='coerce') # 转换日期列 df = df.dropna(subset=['销售员', '产品', '销售额']) # 删除关键列为空的行 # 添加区域信息 df['区域'] = region_name # 计算该区域的摘要信息 total_sales = df['销售额'].sum() avg_per_salesperson = df.groupby('销售员')['销售额'].sum().mean() top_product = df.groupby('产品')['销售额'].sum().idxmax() summary = { '区域': region_name, '邮件文件': os.path.basename(filepath), '数据行数': len(df), '总销售额': f"¥{total_sales:,.2f}", '人均销售额': f"¥{avg_per_salesperson:,.2f}", '最畅销产品': top_product, '处理时间': datetime.now().strftime('%H:%M:%S') } print(f" 区域【{region_name}】数据处理完成，共 {len(df)} 条记录。") return df, summary except Exception as e: print(f" 处理Excel文件 {filepath} 时出错: {e}") return None, None def generate_master_report(all_data_frames, summaries, date_str): """ 生成汇总报告和总数据文件 """ # 1. 合并所有区域的数据 master_df = pd.concat(all_data_frames, ignore_index=True) # 2. 保存合并后的总数据到Excel master_file = f"./汇总结果/销售总表_{date_str}.xlsx" os.makedirs(os.path.dirname(master_file), exist_ok=True) master_df.to_excel(master_file, index=False) print(f" 总数据表已保存至: {master_file}") # 3. 生成文本摘要报告 report_file = f"./汇总结果/日报摘要_{date_str}.txt" with open(report_file, 'w', encoding='utf-8') as f: f.write(f"销售日报自动汇总报告\n") f.write(f"生成日期: {date_str}\n") f.write(f"汇总时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") f.write("="*50 + "\n\n") f.write(f"今日共收到并处理 {len(summaries)} 个区域的报告。\n\n") # 写入各区域摘要 f.write("【各区域数据摘要】\n") for s in summaries: f.write(f"\n* 区域: {s['区域']}\n") f.write(f" 数据量: {s['数据行数']} 条\n") f.write(f" 总销售额: {s['总销售额']}\n") f.write(f" 人均销售额: {s['人均销售额']}\n") f.write(f" 最畅销产品: {s['最畅销产品']}\n") # 写入整体统计 f.write("\n" + "="*50 + "\n") f.write("【整体统计】\n") total_all = master_df['销售额'].sum() f.write(f"全公司今日总销售额: ¥{total_all:,.2f}\n") top_salesperson = master_df.groupby('销售员')['销售额'].sum().idxmax() f.write(f"今日销售冠军: {top_salesperson}\n") overall_top_product = master_df.groupby('产品')['销售额'].sum().idxmax() f.write(f"今日最畅销产品: {overall_top_product}\n") print(f" 文本摘要报告已保存至: {report_file}") ``` **第三步：让机器人定时运行** 脚本写好了，总不能每天手动去运行吧？我们可以用系统的计划任务（Cron on Linux/Mac, Task Scheduler on Windows）或者Python的轻量级库 `schedule` 来实现定时执行。 这里用 `schedule` 举个例了，适合在长期运行的服务器上使用： ```python # 在脚本末尾添加 if __name__ == "__main__": # 直接运行一次 # daily_sales_report_robot() # 或者使用schedule定时运行（例如每天上午11点运行） import schedule import time def job(): print("\n定时任务被触发...") daily_sales_report_robot() # 设定每天11:00执行 schedule.every().day.at("11:00").do(job) print("日报汇总机器人已启动，等待定时执行（每天11:00）...") print("按 Ctrl+C 退出。") while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次 ``` 这样，一个完整的、能够自动收取、解析、处理、汇总并生成报告的邮件机器人就诞生了。你只需要在初期配置好邮箱信息和搜索条件，它就能每天默默无闻地为你工作。 ## 8. 避坑指南与高级技巧：让你的脚本更健壮 在实际项目中跑起来，你肯定会遇到各种问题。下面这些是我用血泪教训换来的经验，能帮你避开很多坑。 **8.1 编码问题的终极解决方案** 邮件编码五花八门，`decode_header` 和指定 `charset` 能解决大部分问题，但偶尔还是会遇到“乱码”。这时候可以请出 `chardet` 这个“猜测”编码的库。 ```python import chardet def safe_decode(byte_data, default_encoding='utf-8'): """一个更健壮的字节解码函数""" if not isinstance(byte_data, bytes): return str(byte_data) # 先尝试用chardet检测 detection = chardet.detect(byte_data) guessed_encoding = detection['encoding'] confidence = detection['confidence'] # 如果置信度较高，使用检测到的编码 if guessed_encoding and confidence > 0.7: try: return byte_data.decode(guessed_encoding, errors='ignore') except (LookupError, UnicodeDecodeError): pass # 如果失败，fallback到默认 # 最后尝试默认编码 try: return byte_data.decode(default_encoding, errors='ignore') except UnicodeDecodeError: # 实在不行，用错误忽略模式解码为拉丁-1，至少不会崩溃 return byte_data.decode('latin-1', errors='ignore') ``` 在解析邮件主题或正文时，可以先用这个函数处理一下 `decoded_subj[0]` 或 `payload`。 **8.2 连接超时与重试机制** 网络不稳定时，连接或操作可能会超时。`imaplib` 有默认超时，但我们可以设置得更友好，并加入重试逻辑。 ```python import imaplib import time def robust_connect(imap_url, username, password, retries=3, timeout=30): """带重试和超时设置的连接函数""" imaplib.IMAP4_SSL.Debug = 4 # 可选：开启调试信息，生产环境关闭 for attempt in range(1, retries + 1): try: print(f"连接尝试第 {attempt} 次...") # 设置超时（秒） mail = imaplib.IMAP4_SSL(imap_url, timeout=timeout) mail.login(username, password) return mail except (imaplib.IMAP4.error, TimeoutError, ConnectionError) as e: print(f"第 {attempt} 次连接失败: {e}") if attempt == retries: raise Exception(f"连接失败，已重试 {retries} 次。") from e wait_time = attempt * 5 # 重试等待时间递增 print(f"等待 {wait_time} 秒后重试...") time.sleep(wait_time) ``` **8.3 处理大量邮件的分页与性能** 如果你需要处理成百上千封历史邮件，一次性搜索 `ALL` 可能会让服务器返回超时，或者占用大量内存。这时需要分页处理。 IMAP本身支持通过 `FETCH` 命令的范围来分页，但更常见的做法是利用搜索条件的日期范围，或者使用 `sort` 命令（如果服务器支持）。一个简单的策略是按月份分批处理： ```python def batch_process_by_month(mail, start_year=2023, start_month=1): """按月分批处理历史邮件""" from datetime import datetime current_date = datetime.now() current_year, current_month = current_date.year, current_date.month for year in range(start_year, current_year + 1): month_start = start_month if year == start_year else 1 month_end = current_month if year == current_year else 12 for month in range(month_start, month_end + 1): # 构造IMAP的日期格式，例如 "01-Jan-2023" since_date = f"01-{datetime(1900, month, 1).strftime('%b').upper()}-{year}" # 下个月的第一天作为BEFORE条件（处理当月） next_month = month + 1 if month < 12 else 1 next_year = year if month < 12 else year + 1 before_date = f"01-{datetime(1900, next_month, 1).strftime('%b').upper()}-{next_year}" criteria = f'SINCE "{since_date}" BEFORE "{before_date}"' print(f"\n处理 {year}年{month}月 的邮件...") email_ids = search_emails(mail, criteria=criteria) # ... 处理这批邮件 ... ``` **8.4 邮件状态管理与文件夹操作** 处理完邮件后，良好的“家务管理”很重要，避免重复处理。 - **标记为已读**：`mail.store(msg_id, '+FLAGS', '\\Seen')` - **标记为重要**：`mail.store(msg_id, '+FLAGS', '\\Flagged')` - **移动邮件**：先复制到目标文件夹，再将原邮件标记为删除。 ```python # 假设目标文件夹是'Processed' mail.copy(msg_id, 'Processed') mail.store(msg_id, '+FLAGS', '\\Deleted') # 注意：标记为删除后，需要执行 expunge() 才能真正删除，或者等会话结束。 # mail.expunge() # 谨慎使用，会立即删除 ``` - **删除邮件**：同上，标记为 `\\Deleted` 后，在关闭连接时选择 `expunge=True` 或在之前调用 `expunge()`。 **8.5 错误处理与日志记录** 生产环境的脚本必须有完善的错误处理和日志。 ```python import logging # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('email_robot.log', encoding='utf-8'), logging.StreamHandler() # 同时输出到控制台 ] ) logger = logging.getLogger(__name__) def main_robot(): try: # ... 你的主逻辑 ... logger.info("任务开始执行。") # ... 处理过程 ... logger.info(f"成功处理了 {count} 封邮件。") except imaplib.IMAP4.error as e: logger.error(f"IMAP协议错误: {e}", exc_info=True) except Exception as e: logger.critical(f"程序发生未预期错误: {e}", exc_info=True) # 可以考虑在这里添加邮件或短信报警 finally: logger.info("任务执行结束。") ``` 把 `print` 换成 `logger.info/warning/error`，你的脚本就拥有了记录时间、级别和上下文的能力，出问题时翻看日志文件就能快速定位。 最后，也是最关键的一点：**安全第一**。永远不要将邮箱密码或授权码写在代码里或提交到版本控制系统。使用环境变量、密钥管理服务或加密的配置文件。对于重要的生产任务，考虑使用专门的服务账户邮箱，并为其设置最小的必要权限。