在 Python 中**安全、合规、稳定地保存 Outlook 最新一封邮件到本地**，**不能通过直接读取 Outlook 数据文件（如 `.pst`）或注入进程实现**（这违反 EULA、触发反病毒软件、且 Windows 10/11 上因 COM 安全策略常失败）。✅ 正确路径是：**使用 Microsoft Graph API（现代身份认证 + OAuth2）** —— 这是微软官方推荐、跨平台、支持 MFA、无需本地 Outlook 客户端、且权限粒度精细的唯一生产级方案。 > ⚠️ 重要前提： > - 你拥有该 Outlook 账户（Microsoft 365 / Outlook.com）的登录权限； > - 你有管理员权限为应用注册 Azure AD 应用（个人开发者可用「个人 Microsoft 账户」+ 应用注册）； > - **不支持“无账号凭据静默访问”——必须用户首次授权（OAuth2 流程）**，后续可刷新 token。 --- ## ✅ 一、整体流程（Graph API 方案） ```mermaid graph LR A[Python 脚本] --> B[启动本地 HTTP 服务器] B --> C[打开浏览器 → Azure 登录页] C --> D[用户登录并授权 mail.read 权限] D --> E[获取 authorization_code] E --> F[向 Microsoft Token Endpoint 换取 access_token] F --> G[调用 Graph API：GET /me/messages?$top=1&$orderby=receivedDateTime%20desc] G --> H[解析 JSON → 提取 subject/body/attachments] H --> I[保存为 .eml / .msg / PDF / TXT] ``` --- ## ✅ 二、详细步骤与完整可运行代码 ### 🔧 Step 1：在 Azure 门户注册应用（免费） 1. 访问 [https://portal.azure.com/#view/Microsoft_AAD_RegisteredApps/ApplicationsListBlade](https://portal.azure.com/#view/Microsoft_AAD_RegisteredApps/ApplicationsListBlade) 2. 点击「+ 新建注册」 - 名称：`Outlook-Mail-Saver` - 支持的账户类型：✅ “任何组织目录(任何 Azure AD 目录 - 多租户) 和 个人 Microsoft 帐户” - 重定向 URI：`http://localhost:8000/auth/callback`（开发用） 3. 注册后，记下： - `CLIENT_ID`（应用（客户端）ID） - `TENANT_ID`（目录（租户）ID，填 `common` 即可支持个人账户） 4. 在「证书和密码」→「客户端密码」→ 生成一个密码，记下 `CLIENT_SECRET`（⚠️仅显示一次！） 5. 在「API 权限」→ 「+ 添加权限」→ Microsoft Graph → 「委托的权限」→ 添加： - ✅ `Mail.Read`（读取邮箱） - ✅ `User.Read`（读取用户资料） → 点击「代表用户授予管理员同意」（若你是管理员）或「为租户中的用户授予同意」 --- ### 🐍 Step 2：安装依赖 ```bash pip install msal requests python-dotenv ``` --- ### 📜 Step 3：完整 Python 脚本（含自动 OAuth2、token 缓存、邮件保存） > ✅ 功能： > - 首次运行打开浏览器授权； > - 自动缓存 token 到 `token_cache.bin`，后续免登录； > - 获取最新一封邮件（按 `receivedDateTime` 降序）； > - 保存为标准 `.eml` 文件（可被 Outlook / Thunderbird / Apple Mail 直接打开）； > - 同时导出纯文本 `.txt` 和附件（如有）。 ```python # save_latest_outlook_email.py import os import json import time import base64 import mimetypes from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.mime.base import MIMEBase from email import encoders from email.utils import formatdate, make_msgid from urllib.parse import urlparse, parse_qs import msal import requests from dotenv import load_dotenv # === 配置区（请替换为你自己的值）=== CLIENT_ID = "your-client-id-here" # Azure 应用 ID CLIENT_SECRET = "your-client-secret-here" # Azure 应用密钥 TENANT_ID = "common" # 支持个人账户（Outlook.com）和企业账户 AUTHORITY = f"https://login.microsoftonline.com/{TENANT_ID}" SCOPE = ["https://graph.microsoft.com/Mail.Read"] # 本地回调地址（必须与 Azure 注册一致） REDIRECT_URI = "http://localhost:8000/auth/callback" # 缓存文件（自动保存/加载 token） TOKEN_CACHE_FILE = "token_cache.bin" # 输出目录 OUTPUT_DIR = "outlook_emails" os.makedirs(OUTPUT_DIR, exist_ok=True) # === 工具函数 === def save_token_cache(cache): with open(TOKEN_CACHE_FILE, "w") as f: f.write(json.dumps(cache.serialize(), indent=2)) print("✅ Token 已缓存到", TOKEN_CACHE_FILE) def load_token_cache(): if os.path.exists(TOKEN_CACHE_FILE): with open(TOKEN_CACHE_FILE, "r") as f: cache = msal.SerializableTokenCache() cache.deserialize(f.read()) return cache return msal.SerializableTokenCache() def get_access_token(): cache = load_token_cache() app = msal.ConfidentialClientApplication( CLIENT_ID, authority=AUTHORITY, client_credential=CLIENT_SECRET, token_cache=cache ) # 尝试从缓存获取未过期 token accounts = app.get_accounts() if accounts: token_response = app.acquire_token_silent(SCOPE, account=accounts[0]) if token_response: print("✅ 使用缓存 Token") return token_response["access_token"] # 否则走 OAuth2 授权码流 print("🔍 正在启动本地服务器等待授权...") from http.server import HTTPServer, BaseHTTPRequestHandler import threading auth_code = [] server_running = True class CallbackHandler(BaseHTTPRequestHandler): def do_GET(self): if self.path.startswith("/auth/callback"): query = urlparse(self.path).query params = parse_qs(query) code = params.get("code", [""])[0] if code: auth_code.append(code) self.send_response(200) self.end_headers() self.wfile.write(b"<h1>✅ 授权成功！窗口可关闭。</h1>") else: self.send_error(400, "Missing code") else: self.send_error(404) def log_message(self, format, *args): pass # 屏蔽日志 # 启动本地服务器（后台线程） server = HTTPServer(('localhost', 8000), CallbackHandler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() # 构造授权 URL auth_url = app.get_authorization_request_url( SCOPE, redirect_uri=REDIRECT_URI, state="random_state_string" ) print(f"🌐 请在浏览器中打开此链接授权：\n{auth_url}\n") # 等待用户授权 while not auth_code and server_running: time.sleep(0.1) server.shutdown() if not auth_code: raise RuntimeError("❌ 用户未完成授权") # 用 code 换取 token token_response = app.acquire_token_by_authorization_code( auth_code[0], scopes=SCOPE, redirect_uri=REDIRECT_URI ) if "error" in token_response: raise RuntimeError(f"❌ Token 获取失败：{token_response.get('error_description')}") save_token_cache(app.token_cache) print("✅ Token 获取成功（已缓存）") return token_response["access_token"] def fetch_latest_email(token: str) -> dict: """调用 Graph API 获取最新一封邮件""" headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json" } url = "https://graph.microsoft.com/v1.0/me/messages" params = { "$top": "1", "$orderby": "receivedDateTime desc", "$select": "id,subject,from,receivedDateTime,body,hasAttachments" } resp = requests.get(url, headers=headers, params=params) resp.raise_for_status() data = resp.json() if not data.get("value"): raise RuntimeError("❌ 未找到邮件") return data["value"][0] def download_attachments(token: str, message_id: str, output_dir: str) -> list: """下载邮件所有附件到指定目录""" headers = {"Authorization": f"Bearer {token}"} url = f"https://graph.microsoft.com/v1.0/me/messages/{message_id}/attachments" resp = requests.get(url, headers=headers) resp.raise_for_status() attachments = resp.json().get("value", []) saved_paths = [] for att in attachments: name = att.get("name", "unnamed") content_type = att.get("contentType", "application/octet-stream") content_bytes = base64.b64decode(att["contentBytes"]) path = os.path.join(output_dir, name) with open(path, "wb") as f: f.write(content_bytes) saved_paths.append(path) print(f"📎 已保存附件：{name}") return saved_paths def save_as_eml(message: dict, token: str, output_dir: str) -> str: """将 Graph API 返回的邮件数据构造成标准 .eml 文件""" msg = MIMEMultipart() msg["From"] = message["from"]["emailAddress"]["address"] msg["To"] = "You (saved)" # Graph 不返回 to，此处简化 msg["Subject"] = message.get("subject", "(无主题)") msg["Date"] = formatdate(time.mktime(time.strptime( message["receivedDateTime"].replace("Z", "+0000"), "%Y-%m-%dT%H:%M:%S.%f%z" ))) msg["Message-ID"] = make_msgid(domain="saved.local") # Body body_text = message.get("body", {}).get("content", "") if message.get("body", {}).get("contentType") == "html": msg.attach(MIMEText(body_text, "html", "utf-8")) else: msg.attach(MIMEText(body_text, "plain", "utf-8")) # Attachments (if any) if message.get("hasAttachments"): att_paths = download_attachments(token, message["id"], output_dir) for p in att_paths: with open(p, "rb") as f: part = MIMEBase("application", "octet-stream") part.set_payload(f.read()) encoders.encode_base64(part) part.add_header( "Content-Disposition", f'attachment; filename="{os.path.basename(p)}"', ) msg.attach(part) # 写入 .eml eml_path = os.path.join(output_dir, f"latest_{int(time.time())}.eml") with open(eml_path, "w", encoding="utf-8") as f: f.write(msg.as_string()) print(f"📧 已保存为 EML：{eml_path}") return eml_path def save_as_txt(message: dict, output_dir: str) -> str: """保存纯文本摘要""" txt_path = os.path.join(output_dir, f"latest_{int(time.time())}.txt") with open(txt_path, "w", encoding="utf-8") as f: f.write(f"发件人: {message['from']['emailAddress']['address']}\n") f.write(f"时间: {message['receivedDateTime']}\n") f.write(f"主题: {message.get('subject', '(无主题)')}\n") f.write(f"正文:\n{message.get('body', {}).get('content', '')[:1000]}...\n") print(f"📄 已保存为 TXT：{txt_path}") return txt_path # === 主程序 === if __name__ == "__main__": try: print("🚀 开始获取最新 Outlook 邮件...") token = get_access_token() latest = fetch_latest_email(token) print(f"📬 获取到最新邮件：{latest.get('subject', '(无主题)')}") # 保存为多种格式 save_as_eml(latest, token, OUTPUT_DIR) save_as_txt(latest, OUTPUT_DIR) print("🎉 完成！查看", os.path.abspath(OUTPUT_DIR)) except Exception as e: print("❌ 错误：", e) exit(1) ``` --- ## ✅ 三、运行说明 1. 将脚本中 `CLIENT_ID` 和 `CLIENT_SECRET` 替换为你在 Azure 注册的真实值； 2. 首次运行： ```bash python save_latest_outlook_email.py ``` → 自动弹出浏览器，登录你的 Outlook 账户并点击「接受」； → 回到终端，看到 `✅ Token 已缓存`； 3. 后续运行：**无需再登录**，直接秒出结果； 4. 输出示例： ``` outlook_emails/ ├── latest_1715678901.eml ← 标准 RFC 2822 邮件，双击可用 Outlook 打开 ├── latest_1715678901.txt ← 纯文本摘要 └── invoice.pdf ← 如果原邮件带附件，也会下载 ``` --- ## ✅ 四、关键优势与注意事项 | 项目 | 说明 | |------|------| | ✅ **完全合规** | 使用 Microsoft 官方 Graph API，符合《Microsoft Services Agreement》 | | ✅ **支持 MFA** | OAuth2 流程天然兼容多因素认证（无需绕过） | | ✅ **跨平台** | Windows/macOS/Linux 均可运行（无需安装 Outlook 客户端） | | ✅ **权限最小化** | 仅申请 `Mail.Read`，无法发送/删除/修改邮件 | | ⚠️ **首次需人工授权** | 无法 100% 全自动（这是安全设计，不可规避） | | ⚠️ **Token 有效期** | 默认 1 小时，但脚本自动刷新（`acquire_token_silent`） | | ⚠️ **附件大小限制** | Graph API 单附件 ≤ 4MB（大文件需用 `@odata.mediaReadLink` 分块下载，本例已简化） | --- ## ✅ 五、进阶建议 - 🔐 **生产环境加固**：将 `CLIENT_SECRET` 存入 Azure Key Vault 或系统环境变量（`os.getenv("CLIENT_SECRET")`）； - 📈 **定时任务**：用 `cron`（Linux/macOS）或 Task Scheduler（Windows）每天执行一次； - 📥 **监听新邮件**：改用 Graph Webhook（订阅 `/me/mailFolders/inbox/messages`），实时推送（需公网 HTTPS 回调地址）； - 🧩 **解析 HTML 邮件**：用 `BeautifulSoup` 清洗 `<body>` 内容，提取表格/链接/图片。 ---