每天早上九点十分,我端着咖啡坐到工位上,打开邮箱,看着未读邮件的数字发呆。
187封。
这是上个月某个周一的真实数字。我花了四十分钟翻完它们,把审批邮件、项目通知、广告推销、系统告警一堆东西混在一起看。就在我逐封点开的时候,同事阿杰走过来拍了我肩膀:“昨天发的采购审批你看了没?财务部催了两轮了。”
我没看。那封邮件夹在187封未读里,标题是“关于2024年Q3采购流程的审批请求”,发件人是财务部的王姐。我直接把它当成普通通知跳过了。
那天下午被领导叫去谈了半小时。
回到工位我就开始想,这事不能这么干了。每天光翻邮件就要花大半个小时,还经常漏掉重要的。我得让机器帮我分拣。
为什么是Python
邮箱客户端自带的过滤规则太死了。换个发件人地址就失效,关键词匹配也不够聪明。我需要一套能按发件人、关键词、附件类型多个维度同时判断的逻辑,而且最好能自动跑。
Python的imaplib是内置库,不用装额外依赖就能连邮箱。再配合email库解析邮件内容,写个分类脚本很快。
思路
整个脚本做的事情很直接:
用IMAP协议连上邮箱服务器
拉取所有未读邮件
解析每封邮件的发件人、标题、正文、附件
按预设规则打分分类:领导邮件、审批类、项目通知、垃圾邮件等
把分类结果输出一份摘要,重要的单独提醒
分类规则用字典配置,改起来方便。
完整代码
import imaplibimport emailfrom email.header import decode_headerimport refrom datetime import datetimefrom collections import defaultdict# ========== 配置区 ==========# 邮箱设置(以QQ邮箱为例,需要开启IMAP并获取授权码)EMAIL_ACCOUNT = "your_email@qq.com"EMAIL_PASSWORD = "your_auth_code_here"# 不是登录密码,是授权码IMAP_SERVER = "imap.qq.com"IMAP_PORT = 993# 分类规则 —— 按需修改# VIP发件人:包含这些关键词的发件人地址或名称,归为"重要"VIP_SENDERS = ["boss@company.com","leader@company.com","王总","李总监",]# 审批类关键词:标题或正文包含这些词,归为"审批"APPROVAL_KEYWORDS = ["审批", "批准", "签批", "审核", "请批","报销", "采购", "付款", "合同", "用印",]# 项目通知关键词PROJECT_KEYWORDS = ["项目进度", "里程碑", "上线", "发版", "需求变更","迭代", "sprint", "排期", "联调",]# 垃圾邮件关键词SPAM_KEYWORDS = ["优惠", "促销", "限时", "免费领取", "中奖","unsubscribe", "退订",]# 重要附件类型IMPORTANT_EXTENSIONS = [".pdf", ".docx", ".xlsx", ".pptx", ".doc"]# ========== 工具函数 ==========def decode_mime_words(s):"""解码邮件标题中的编码字符串"""if not s:return""decoded_parts = decode_header(s)result = []for part, charset in decoded_parts:if isinstance(part, bytes):# 尝试用指定编码或utf-8解码try:result.append(part.decode(charsetor"utf-8", errors="replace"))except (LookupError, UnicodeDecodeError):result.append(part.decode("utf-8", errors="replace"))else:result.append(part)return"".join(result)def get_email_body(msg):"""提取邮件正文,优先取纯文本"""body = ""if msg.is_multipart():for part in msg.walk():content_type = part.get_content_type()if content_type == "text/plain":charset = part.get_content_charset() or"utf-8"try:body = part.get_payload(decode=True).decode(charset, errors="replace")except (LookupError, UnicodeDecodeError):body = part.get_payload(decode=True).decode("utf-8", errors="replace")breakelse:charset = msg.get_content_charset() or"utf-8"try:body = msg.get_payload(decode=True).decode(charset, errors="replace")except (LookupError, UnicodeDecodeError):body = msg.get_payload(decode=True).decode("utf-8", errors="replace")return bodydef get_attachments(msg):"""获取附件文件名列表"""attachments = []if msg.is_multipart():for part in msg.walk():filename = part.get_filename()if filename:decoded_name = decode_mime_words(filename)attachments.append(decoded_name)return attachmentsdef classify_email(sender, subject, body, attachments):""" 根据规则对邮件分类。 返回分类标签列表(一封邮件可能同时属于多个类别)。 """labels = []# 把所有文本统一转小写用于关键词匹配sender_lower = sender.lower()subject_lower = subject.lower()body_lower = body.lower()# 规则1:VIP发件人for vip in VIP_SENDERS:if vip.lower() in sender_lower:labels.append("VIP")break# 规则2:审批类for kw in APPROVAL_KEYWORDS:if kw in subject or kw in body[:500]: # 正文只看前500字labels.append("审批")break# 规则3:项目通知for kw in PROJECT_KEYWORDS:if kw.lower() in subject_lower or kw.lower() in body_lower[:500]:labels.append("项目")break# 规则4:有重要附件for att in attachments:for ext in IMPORTANT_EXTENSIONS:if att.lower().endswith(ext):labels.append("有附件")breakif"有附件"inlabels:break# 规则5:垃圾邮件for kw in SPAM_KEYWORDS:if kw.lower() in subject_lower or kw.lower() in body_lower[:300]:labels.append("垃圾")break# 没匹配到任何规则if not labels:labels.append("普通")returnlabels# ========== 主流程 ==========def fetch_and_classify():"""连接邮箱,拉取未读邮件并分类"""print(f"[{datetime.now().strftime('%H:%M:%S')}] 正在连接邮箱...")# 连接IMAP服务器mail = imaplib.IMAP4_SSL(IMAP_SERVER, IMAP_PORT)mail.login(EMAIL_ACCOUNT, EMAIL_PASSWORD)# 选择收件箱mail.select("INBOX")# 搜索未读邮件(也可以用 ALL 拉全部)status, messages = mail.search(None, "UNSEEN")if status!= "OK":print("搜索邮件失败")returnemail_ids = messages[0].split()total = len(email_ids)print(f"找到 {total} 封未读邮件\n")if total == 0:print("没有未读邮件,收工。")mail.logout()return# 分类统计classified = defaultdict(list)# 逐封处理for i, eid in enumerate(email_ids, 1):status, msg_data = mail.fetch(eid, "(RFC822)")if status!= "OK":continueraw_email = msg_data[0][1]msg = email.message_from_bytes(raw_email)# 解析基本信息subject = decode_mime_words(msg.get("Subject", ""))sender = decode_mime_words(msg.get("From", ""))date_str = msg.get("Date", "")body = get_email_body(msg)attachments = get_attachments(msg)# 分类labels = classify_email(sender, subject, body, attachments)# 记录结果email_info = {"subject": subject,"sender": sender,"date": date_str,"labels": labels,"attachments": attachments, }# 一封邮件可能属于多个分类,取最高优先级priority = labels[0]classified[priority].append(email_info)# 打印进度if i % 20 == 0 or i == total:print(f" 已处理 {i}/{total} 封...")mail.logout()# ========== 输出分类报告 ==========print("\n"+"="*60)print(f"邮件分类报告 - {datetime.now().strftime('%Y-%m-%d %H:%M')}")print("="*60)# 按优先级输出priority_order = ["VIP", "审批", "项目", "有附件", "普通", "垃圾"]for label in priority_order:emails = classified.get(label, [])if not emails:continueprint(f"\n【{label}】共 {len(emails)} 封")print("-"*40)for e in emails:subj = e["subject"][:50] # 截断太长的标题sender_short = e["sender"][:30]att_mark = f" {len(e['attachments'])}"if e["attachments"] else""print(f" · {subj}")print(f" 来自: {sender_short}{att_mark}")# 重点提醒vip_count = len(classified.get("VIP", []))approval_count = len(classified.get("审批", []))if vip_count or approval_count:print("\n"+"!"*40)print(f"注意:你有 {vip_count} 封VIP邮件,{approval_count} 封审批邮件需要处理")print("!"*40)# 统计汇总print(f"\n总计处理 {total} 封邮件:")for label in priority_order:count = len(classified.get(label, []))if count:print(f" {label}: {count} 封")return classifiedif__name__ == "__main__":fetch_and_classify()运行效果
脚本跑起来之后,输出大概长这样:
[09:12:03] 正在连接邮箱...找到 187 封未读邮件 已处理 20/187 封... 已处理 40/187 封... ... 已处理 187/187 封...============================================================邮件分类报告 - 2024-09-02 09:15============================================================【VIP】共 3 封---------------------------------------- · 关于Q3预算调整的确认 来自: 王总 <boss@company.com> · 下周一管理层会议议程 来自: 李总监 <leader@company.com>【审批】共 7 封---------------------------------------- · 关于2024年Q3采购流程的审批请求 来自: 财务部王姐 <wang@company.com> 2 · 差旅报销单-请审批 来自: 张三 <zhangsan@company.com> 1【项目】共 23 封---------------------------------------- · v2.3.1 发版通知 来自: devops@company.com · 需求变更评审会议纪要 来自: 产品经理小刘 <liu@company.com>【普通】共 131 封 ...【垃圾】共 23 封 ...!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!注意:你有 3 封VIP邮件,7 封审批邮件需要处理!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!总计处理 187 封邮件: VIP: 3 封 审批: 7 封 项目: 23 封 普通: 131 封 垃圾: 23 封
187封邮件,秒出结果。
我第一眼看到的是最上面的VIP和审批区。阿杰催的那封采购审批,稳稳地出现在“审批”分类里,标着“2”表示有两个附件。再也不会漏了。
几个实际踩过的坑
授权码不是登录密码。 QQ邮箱要在设置里开启IMAP服务,它会生成一个授权码。用登录密码连的话会报认证失败。网易163邮箱也是同样的套路。
中文标题乱码。 邮件标题的编码方式千奇百怪,有GBK的,有UTF-8的,还有混合编码的。decode_mime_words这个函数就是专门处理这种情况的,它会逐段解码。
正文太大别全读。 有些营销邮件正文几十KB,全读进来做关键词匹配很慢。我只取前500字判断,够用了。
配合Windows计划任务自动跑。 我把脚本存成.py文件,在计划任务里设成每天早上8:55执行。到工位的时候分类报告已经生成好了,直接看结果。
后来我又加了什么
跑了一段时间后,我把分类结果接上了企业微信的webhook。VIP和审批类邮件会直接推一条消息到我手机的企业微信上,连邮箱都不用打开。
实现也很简单,就是在脚本最后加一段:
import requestsWEBHOOK_URL = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=your_key_here"def send_alert(vip_emails, approval_emails):"""把重要邮件摘要推送到企业微信"""content_parts = []if vip_emails:content_parts.append("**VIP邮件:**")for e in vip_emails:content_parts.append(f"> {e['subject']}\n> 来自: {e['sender']}")if approval_emails:content_parts.append("**待审批:**")for e in approval_emails:content_parts.append(f"> {e['subject']}")if content_parts:payload = {"msgtype": "markdown","markdown": {"content": "\n".join(content_parts) } }requests.post(WEBHOOK_URL, json=payload)加上这段之后,我再也没漏过领导的邮件。
现在每天早上到工位,先看企业微信的推送,重要的几分钟就处理完了。剩下的普通邮件和垃圾邮件?有空再翻,不翻也无所谓。
每天省下来的不止半小时。关键是心里踏实了,不用再担心漏掉什么要紧事。
“无他,惟手熟尔”!有需要的用起来!关注微信公众号「Nicholas与Pypi」获取更多Python实战!------加入知识库与更多人一起学习------https://ima.qq.com/wiki/?shareId=f2628818f0874da17b71ffa0e5e8408114e7dbad46f1745bbd1cc1365277631c