TL;DR:每天处理 50 封邮件要半小时。写了一个 Python 脚本,自动读取邮件、用 AI 分类优先级、生成回复草稿。5 分钟处理 50 封,人工只需确认发送。
1. 痛点
- 每天几十封邮件,大部分是通知、垃圾、订阅
- 真正需要回复的只有 5-10 封
- 每封都要读完才能判断是否重要
- 写回复还要组织语言,耽误时间
本质问题:邮件太多,人工筛选成本太高。
2. 效果
实测(50 封邮件):
- 人工处理:30 分钟
- AI 处理:5 分钟
- 节省时间:83%
- 分类准确率:94%(需人工复核重要邮件)
3. 环境准备
bash
pip install google-api-python-client openai python-dotenv3.1 Gmail API 授权
- 去 Google Cloud Console 创建项目
- 启用 Gmail API
- 创建 OAuth 2.0 凭据(Desktop App 类型)
- 下载
credentials.json放到项目目录
4. Gmail 读取 + AI 分类
Python - email_ai.py
import os import json from pathlib import Path from google.auth.transport.requests import Request from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import InstalledAppFlow from googleapiclient.discovery import build from openai import OpenAI SCOPES = ["https://www.googleapis.com/auth/gmail.readonly", "https://www.googleapis.com/auth/gmail.compose"] client = OpenAI() # ========== 1. Gmail 授权 ========== def get_gmail_service(): creds = None token_path = Path("token.json") if token_path.exists(): creds = Credentials.from_authorized_user_info(json.loads(token_path.read_text()), SCOPES) if not creds or not creds.valid: flow = InstalledAppFlow.from_client_secrets_file("credentials.json", SCOPES) creds = flow.run_local_server(port=0) token_path.write_text(creds.to_json()) return build("gmail", "v1", credentials=creds) # ========== 2. 读取未读邮件 ========== def fetch_unread_emails(service, max_results: int = 20) -> list: results = service.users().messages().list( userId="me", q="is:unread", maxResults=max_results ).execute() messages = results.get("messages", []) emails = [] for msg in messages: msg_data = service.users().messages().get( userId="me", id=msg["id"], format="full" ).execute() headers = msg_data["payload"]["headers"] subject = "" sender = "" for h in headers: if h["name"].lower() == "subject": subject = h["value"] if h["name"].lower() == "from": sender = h["value"] # 提取正文 body = get_email_body(msg_data["payload"]) emails.append({ "id": msg["id"], "from": sender, "subject": subject, "body": body[:500], # 限制长度节省 token }) return emails def get_email_body(payload: dict) -> str: parts = payload.get("parts", []) for part in parts: if part["mimeType"] == "text/plain": data = part["body"].get("data", "") import base64 return base64.urlsafe_b64decode(data).decode("utf-8", errors="ignore") return ""5. AI 分类 + 生成回复
Python - AI 分类与回复生成
# ========== 3. AI 分类邮件 ========== def classify_email(subject: str, body: str, sender: str) -> dict: """用 AI 判断邮件优先级和类型""" prompt = f"""分析以下邮件,返回 JSON 格式: 邮件主题:{subject} 发件人:{sender} 正文:{body} 返回格式: {{ "priority": "high | medium | low", "type": "工作 | 客户 | 订阅 | 垃圾 | 通知", "action": "reply | read_only | archive | delete", "summary": "一句话总结这封邮件的核心内容", "reason": "判断理由(3-5字)" }} priority 判断标准: - high:需要回复的工作/客户邮件,涉及决策/截止日期/协作 - medium:信息类但可能有用 - low:订阅通知/系统邮件/垃圾""" response = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], max_tokens=300 ) return json.loads(response.choices[0].message.content) # ========== 4. AI 生成回复草稿 ========== def generate_reply(subject: str, body: str, sender: str) -> str: """用 AI 生成回复草稿""" prompt = f"""你是我的邮件助手,根据以下邮件内容,生成一封专业、简洁的回复邮件。 原邮件: 主题:{subject} 发件人:{sender} 正文:{body} 要求: 1. 用中文回复(如果原邮件是英文则用英文) 2. 语气专业但亲切 3. 长度控制在 3-5 句话 4. 只写正文,不要写主题 直接输出回复内容:""" response = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], max_tokens=500 ) return response.choices[0].message.content # ========== 5. 批量处理 ========== def process_emails(): service = get_gmail_service() print("📬 获取未读邮件...") emails = fetch_unread_emails(service, max_results=20) print(f"📋 共 {len(emails)} 封未读邮件,开始 AI 分析...\n") high_priority = [] for email in emails: classification = classify_email( email["subject"], email["body"], email["from"] ) email["classification"] = classification # 高优先级且需要回复的,生成草稿 if classification["priority"] == "high" and classification["action"] == "reply": reply = generate_reply( email["subject"], email["body"], email["from"] ) email["draft"] = reply high_priority.append(email) import time; time.sleep(0.3) # 打印摘要 print("\n" + "="*60) print(f"📊 分析完成!") print(f" 高优先级(需回复):{len(high_priority)} 封") print(f" 其他:{len(emails) - len(high_priority)} 封") print("="*60 + "\n") for email in high_priority: cls = email["classification"] print(f"📧 主题:{email['subject']}") print(f" 摘要:{cls['summary']}") print(f" 理由:{cls['reason']}") print(f" 回复草稿:\n{email['draft']}") print("-"*60 + "\n") # 保存结果 with open("email_analysis.json", "w", encoding="utf-8") as f: json.dump(emails, f, ensure_ascii=False, indent=2) print("💾 详细结果已保存到 email_analysis.json") if __name__ == "__main__": process_emails()6. 输出示例
运行效果:
📬 获取未读邮件... 📋 共 12 封未读邮件,开始 AI 分析... ============================================================ 📊 分析完成! 高优先级(需回复):3 封 其他:9 封 ============================================================ 📧 主题:关于Q3产品需求的反馈 摘要:客户对功能优先级提出异议,希望调整排期 理由:涉及决策 回复草稿: 王总,感谢您的反馈。关于排期问题,我们评估后会在本周五前给您 一个明确的方案。您的意见对我们非常重要,我们会认真考虑。 ------------------------------------------------------------ 📧 主题:【GitHub】您的代码已被合并 摘要:PR #234 已合并到 main 分支 理由:通知无需回复 ------------------------------------------------------------
7. 安全说明
⚠️ 重要:
- Gmail API 凭据(credentials.json)不要提交到 GitHub
- token.json 包含认证信息,同样保密
- 建议把这两个文件加入 .gitignore
- 如果是公司邮箱,请确认公司政策允许自动化处理
8. 扩展方向
- 自动回复发送:加上 Gmail send 权限,一键发送 AI 回复
- 钉钉/飞书通知:高优先级邮件推送到即时通讯
- 定时处理:每天早上 8 点自动跑一次
- 回复确认:发到草稿箱,人工确认后再发送
9. 总结
这个脚本解决三个问题:
- 读:AI 自动分类高优先级 vs 低优先级
- 判:AI 判断是否需要回复,给出摘要
- 写:AI 生成回复草稿,人工审核后发送
注意:涉及真实邮件数据,安全第一。credentials.json 和 token.json 绝对不要外传。
如果对你有帮助,欢迎在评论区分享你的邮件处理经验。