### 使用 Python 的 `imapclient` 和 `email` 库实现定时收取邮件并解析 以下是基于需求的解决方案，包括如何使用 `imapclient` 和 `email` 模块来定期获取特定发件人的电子邮件内容，并从中提取关键字段信息，最终将其写入到 Excel 文件中的方法。 #### 定义功能模块 为了完成这一目标，可以分为以下几个部分： 1. **连接 IMAP 邮箱服务器** 利用 `imapclient.IMAPClient` 连接到邮箱服务提供商（如 Gmail 或其他支持 IMAP 协议的服务商），登录账户并选择要操作的文件夹。 2. **筛选特定发件人邮件** 使用 `imapclient.search()` 方法通过条件过滤指定发件人的邮件列表。 3. **解析邮件内容** 借助 `email.message_from_bytes()` 函数加载原始邮件数据，并进一步处理其结构化内容以提取所需的关键字段。 4. **保存至 Excel 表格** 将提取的数据存储为 Pandas DataFrame 对象，再调用 pandas 的 `.to_excel()` 方法导出成 Excel 文档。 --- #### 示例代码实现 ```python import time from imapclient import IMAPClient import email from email.header import decode_header import re import pandas as pd def fetch_emails(host, username, password, folder="INBOX", sender=None): """ 获取来自特定发件人的所有未读邮件. 参数: host (str): IMAP 服务器地址. username (str): 用户名. password (str): 密码. folder (str): 要检索的文件夹，默认为 INBOX. sender (str): 特定发件人邮箱地址. 返回值: list: 符合条件的邮件对象列表. """ with IMAPClient(host) as server: server.login(username, password) server.select_folder(folder) criteria = ['UNSEEN'] if sender is not None: criteria.extend(['FROM', sender]) messages_ids = server.search(criteria)[^2] response = server.fetch(messages_ids, ["BODY[]", "FLAGS"]) emails = [] for msg_id, data in response.items(): raw_email = data[b"BODY[]"] mail = email.message_from_bytes(raw_email) subject, encoding = decode_header(mail["Subject"])[0] from_ = str(email.utils.parseaddr(mail.get("From"))[1]) content = "" if mail.is_multipart(): for part in mail.walk(): ctype = part.get_content_type() cdispo = str(part.get('Content-Disposition')) if ctype == "text/plain" and "attachment" not in cdispo: content += part.get_payload(decode=True).decode(encoding or 'utf-8') else: content = mail.get_payload(decode=True).decode() emails.append({ "id": msg_id, "subject": subject.decode() if isinstance(subject, bytes) else subject, "sender": from_, "content": content.strip(), }) return emails def extract_keywords(texts, patterns): """ 根据正则表达式模式从文本中抽取关键字. 参数: texts (list of str): 待分析的一组字符串. patterns (dict): 关键字对应的匹配规则 {key: regex}. 返回值: dict: 各项关键字及其对应的结果集合. """ results = {} for key, pattern in patterns.items(): matches = set(re.findall(pattern, "\n".join(texts))) results[key] = ", ".join(matches) if matches else "<无>" return results def save_to_excel(dataframe, output_file): """ 把数据框存入 Excel 文件. 参数: dataframe (pd.DataFrame): 数据表. output_file (str): 输出路径及文件名. """ writer = pd.ExcelWriter(output_file, engine='xlsxwriter') dataframe.to_excel(writer, index=False, sheet_name="Sheet1") writer.close() if __name__ == "__main__": # 设置参数 HOST = "imap.example.com" USERNAME = "your-email@example.com" PASSWORD = "password" SENDER_EMAIL = "specific-sender@example.com" KEYWORDS_PATTERNS = { "订单号": r"订单编号\s*:\s*(\d+)", "金额总计": r"总金额\s*:?\s*\$(\d+\.\d{2})", "客户姓名": r"(?<=客户的)\w+(?:\s+\w+)?(?=\s*)" } while True: try: fetched_emails = fetch_emails(HOST, USERNAME, PASSWORD, sender=SENDER_EMAIL) extracted_data = [] for eml in fetched_emails: keywords_result = extract_keywords([eml["content"]], KEYWORDS_PATTERNS) record = {"主题": eml["subject"], **keywords_result} extracted_data.append(record) df = pd.DataFrame(extracted_data) current_time = time.strftime("%Y%m%d-%H%M%S", time.localtime()) excel_filename = f"extracted_{current_time}.xlsx" save_to_excel(df, excel_filename) print(f"{len(fetched_emails)} 条新消息已成功处理并保存.") except Exception as e: print(f"发生错误: {e}") finally: # 设定轮询间隔时间（单位秒） time.sleep(60 * 5) # 每隔五分钟运行一次循环 ``` --- ### 功能说明 1. 上述脚本实现了每五分钟左右检查是否有新的符合条件的邮件到达的功能[^3]。 2. 如果发现有满足设定条件的新邮件，则会尝试解析这些邮件的内容，并按照预定义好的关键词模板提取相关信息。 3. 所有的结果会被整理成表格形式并通过 Pandas 工具包输出为本地磁盘上的 Excel 文件。 ---