胡绍康andCopilot
c70d2b97dd
- black 格式化 (1 file) - isort 排序修复 (30 files, --profile black) - 升级 actions/checkout v4 → v6 (全部 3 个 workflow) - 添加 pyproject.toml (black/isort 配置) - isort CI 命令添加 --profile black 参数 - 114 测试全部通过 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
479 lines
15 KiB
Python
479 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
Outlook 邮件读取测试工具
|
||
使用三种方式读取 Outlook 邮箱邮件：
|
||
1. 旧版 IMAP 方式 (outlook_imap_old_utils)
|
||
2. 新版 IMAP 方式 (outlook_imap_new_utils)
|
||
3. Graph API 方式 (graph_utils)
|
||
"""
|
||
|
||
import email
|
||
import imaplib
|
||
from email.header import decode_header
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
import requests
|
||
|
||
# ==================== 配置参数 ====================
|
||
# 邮箱账号
|
||
EMAIL = ""
|
||
# 邮箱密码（通常不需要，OAuth2 认证时使用 refresh_token）
|
||
PASSWORD = ""
|
||
# OAuth2 refresh_token
|
||
CLIENT_ID = ""
|
||
# OAuth2 client_id
|
||
REFRESH_TOKEN = ""
|
||
# 代理地址（可选，格式: host:port 或 http://host:port）
|
||
PROXY = None # 例如: "127.0.0.1:7890"
|
||
# ================================================
|
||
|
||
|
||
# Token 端点
|
||
TOKEN_URL_LIVE = "https://login.live.com/oauth20_token.srf"
|
||
TOKEN_URL_GRAPH = "https://login.microsoftonline.com/common/oauth2/v2.0/token"
|
||
TOKEN_URL_IMAP = "https://login.microsoftonline.com/consumers/oauth2/v2.0/token"
|
||
|
||
# IMAP 服务器配置
|
||
IMAP_SERVER_OLD = "outlook.office365.com"
|
||
IMAP_SERVER_NEW = "outlook.live.com"
|
||
IMAP_PORT = 993
|
||
|
||
|
||
def print_separator(title: str):
|
||
"""打印分隔线"""
|
||
print("\n" + "=" * 80)
|
||
print(f"【{title}】")
|
||
print("=" * 80)
|
||
|
||
|
||
def decode_header_value(header_value: str) -> str:
|
||
"""解码邮件头字段"""
|
||
if not header_value:
|
||
return ""
|
||
try:
|
||
decoded_parts = decode_header(str(header_value))
|
||
decoded_string = ""
|
||
for part, charset in decoded_parts:
|
||
if isinstance(part, bytes):
|
||
try:
|
||
decoded_string += part.decode(
|
||
charset if charset else "utf-8", "replace"
|
||
)
|
||
except (LookupError, UnicodeDecodeError):
|
||
decoded_string += part.decode("utf-8", "replace")
|
||
else:
|
||
decoded_string += str(part)
|
||
return decoded_string
|
||
except Exception:
|
||
return str(header_value) if header_value else ""
|
||
|
||
|
||
def print_email_info(emails: List[Any], method_name: str):
|
||
"""打印邮件信息"""
|
||
if not emails:
|
||
print(f"❌ {method_name}: 未获取到邮件")
|
||
return
|
||
|
||
print(f"✅ {method_name}: 成功获取 {len(emails)} 封邮件\n")
|
||
|
||
for i, msg in enumerate(emails[:5]): # 只显示前5封
|
||
print(f" 📧 邮件 {i + 1}:")
|
||
|
||
# 根据邮件类型获取信息
|
||
if isinstance(msg, dict):
|
||
# Graph API 返回的是字典
|
||
subject = msg.get("subject", "无主题")
|
||
from_info = msg.get("from", {})
|
||
sender = from_info.get("emailAddress", {}).get("address", "未知发件人")
|
||
received_time = msg.get("receivedDateTime", "未知时间")
|
||
print(f" 主题: {subject}")
|
||
print(f" 发件人: {sender}")
|
||
print(f" 时间: {received_time}")
|
||
else:
|
||
# IMAP 返回的是 email.message.EmailMessage
|
||
subject = decode_header_value(msg.get("Subject", "无主题"))
|
||
sender = decode_header_value(msg.get("From", "未知发件人"))
|
||
date = msg.get("Date", "未知时间")
|
||
print(f" 主题: {subject}")
|
||
print(f" 发件人: {sender}")
|
||
print(f" 时间: {date}")
|
||
print()
|
||
|
||
|
||
# ==================== 方式1: 旧版 IMAP 方式 ====================
|
||
|
||
|
||
def get_access_token_old(
|
||
account: str, client_id: str, refresh_token: str
|
||
) -> Optional[str]:
|
||
"""
|
||
旧版方式获取 access_token
|
||
使用 login.live.com 端点
|
||
"""
|
||
print(" 🔑 正在获取 access_token (旧版 login.live.com)...")
|
||
|
||
try:
|
||
data = {
|
||
"client_id": client_id,
|
||
"grant_type": "refresh_token",
|
||
"refresh_token": refresh_token,
|
||
}
|
||
|
||
ret = requests.post(TOKEN_URL_LIVE, data=data, timeout=30)
|
||
|
||
if ret.status_code != 200:
|
||
print(f" ❌ 获取 access_token 失败: {ret.status_code}")
|
||
print(f" 响应: {ret.text[:200]}...")
|
||
if "User account is found to be in service abuse mode" in ret.text:
|
||
print(" ⚠️ 账号被封禁!")
|
||
return None
|
||
|
||
access_token = ret.json().get("access_token")
|
||
if access_token:
|
||
print(f" ✅ 成功获取 access_token，长度: {len(access_token)}")
|
||
return access_token
|
||
|
||
except Exception as e:
|
||
print(f" ❌ 获取 access_token 异常: {e}")
|
||
return None
|
||
|
||
|
||
def read_emails_imap_old(
|
||
account: str, client_id: str, refresh_token: str, top: int = 10
|
||
) -> Optional[List]:
|
||
"""
|
||
方式1: 旧版 IMAP 方式读取邮件
|
||
使用 outlook.office365.com 服务器
|
||
"""
|
||
print_separator("方式1: 旧版 IMAP 方式 (outlook.office365.com)")
|
||
|
||
# 1. 获取 access_token
|
||
access_token = get_access_token_old(account, client_id, refresh_token)
|
||
if not access_token:
|
||
return None
|
||
|
||
# 2. 连接 IMAP 服务器
|
||
connection = None
|
||
try:
|
||
print(f" 📡 正在连接 IMAP 服务器: {IMAP_SERVER_OLD}...")
|
||
connection = imaplib.IMAP4_SSL(IMAP_SERVER_OLD, IMAP_PORT)
|
||
|
||
# 3. XOAUTH2 认证
|
||
auth_string = f"user={account}\1auth=Bearer {access_token}\1\1"
|
||
connection.authenticate("XOAUTH2", lambda x: auth_string)
|
||
print(" ✅ IMAP 认证成功")
|
||
|
||
# 4. 选择收件箱
|
||
connection.select("INBOX")
|
||
|
||
# 5. 搜索邮件
|
||
status, messages = connection.search(None, "ALL")
|
||
if status != "OK" or not messages or not messages[0]:
|
||
print(" ⚠️ 收件箱为空")
|
||
return []
|
||
|
||
message_ids = messages[0].split()
|
||
print(f" 📬 收件箱共有 {len(message_ids)} 封邮件")
|
||
|
||
# 6. 获取最近的邮件
|
||
recent_ids = message_ids[-top:][::-1] # 倒序，最新的在前
|
||
|
||
emails = []
|
||
for msg_id in recent_ids:
|
||
try:
|
||
status, msg_data = connection.fetch(msg_id, "(RFC822)")
|
||
if status == "OK" and msg_data and msg_data[0]:
|
||
raw_email = msg_data[0][1]
|
||
msg = email.message_from_bytes(raw_email)
|
||
emails.append(msg)
|
||
except Exception as e:
|
||
print(f" ⚠️ 解析邮件 {msg_id} 失败: {e}")
|
||
continue
|
||
|
||
return emails
|
||
|
||
except Exception as e:
|
||
print(f" ❌ IMAP 连接失败: {e}")
|
||
return None
|
||
finally:
|
||
if connection:
|
||
try:
|
||
connection.logout()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
# ==================== 方式2: 新版 IMAP 方式 ====================
|
||
|
||
|
||
def get_access_token_imap(client_id: str, refresh_token: str) -> Optional[str]:
|
||
"""
|
||
新版方式获取 IMAP access_token
|
||
使用 login.microsoftonline.com/consumers 端点，IMAP scope
|
||
"""
|
||
print(" 🔑 正在获取 access_token (新版 IMAP scope)...")
|
||
|
||
try:
|
||
proxies = None
|
||
if PROXY:
|
||
proxies = {
|
||
"all": f"http://{PROXY}" if not PROXY.startswith("http") else PROXY
|
||
}
|
||
|
||
res = requests.post(
|
||
TOKEN_URL_IMAP,
|
||
data={
|
||
"client_id": client_id,
|
||
"grant_type": "refresh_token",
|
||
"refresh_token": refresh_token,
|
||
"scope": "https://outlook.office.com/IMAP.AccessAsUser.All offline_access",
|
||
},
|
||
proxies=proxies,
|
||
timeout=30,
|
||
)
|
||
|
||
if res.status_code != 200:
|
||
print(f" ❌ 获取 access_token 失败: {res.status_code}")
|
||
print(f" 响应: {res.text[:200]}...")
|
||
if "User account is found to be in service abuse mode" in res.text:
|
||
print(" ⚠️ 账号被封禁!")
|
||
return None
|
||
|
||
access_token = res.json().get("access_token")
|
||
if access_token:
|
||
print(f" ✅ 成功获取 access_token，长度: {len(access_token)}")
|
||
return access_token
|
||
|
||
except Exception as e:
|
||
print(f" ❌ 获取 access_token 异常: {e}")
|
||
return None
|
||
|
||
|
||
def read_emails_imap_new(
|
||
account: str, client_id: str, refresh_token: str, top: int = 10
|
||
) -> Optional[List]:
|
||
"""
|
||
方式2: 新版 IMAP 方式读取邮件
|
||
使用 outlook.live.com 服务器
|
||
"""
|
||
print_separator("方式2: 新版 IMAP 方式 (outlook.live.com)")
|
||
|
||
# 1. 获取 access_token
|
||
access_token = get_access_token_imap(client_id, refresh_token)
|
||
if not access_token:
|
||
return None
|
||
|
||
# 2. 连接 IMAP 服务器
|
||
connection = None
|
||
try:
|
||
print(f" 📡 正在连接 IMAP 服务器: {IMAP_SERVER_NEW}...")
|
||
connection = imaplib.IMAP4_SSL(IMAP_SERVER_NEW, IMAP_PORT)
|
||
|
||
# 3. XOAUTH2 认证
|
||
auth_string = f"user={account}\1auth=Bearer {access_token}\1\1".encode("utf-8")
|
||
connection.authenticate("XOAUTH2", lambda x: auth_string)
|
||
print(" ✅ IMAP 认证成功")
|
||
|
||
# 4. 选择收件箱
|
||
connection.select('"INBOX"')
|
||
|
||
# 5. 搜索邮件
|
||
status, messages = connection.search(None, "ALL")
|
||
if status != "OK" or not messages or not messages[0]:
|
||
print(" ⚠️ 收件箱为空")
|
||
return []
|
||
|
||
message_ids = messages[0].split()
|
||
print(f" 📬 收件箱共有 {len(message_ids)} 封邮件")
|
||
|
||
# 6. 获取最近的邮件
|
||
recent_ids = message_ids[-top:][::-1]
|
||
|
||
emails = []
|
||
for msg_id in recent_ids:
|
||
try:
|
||
status, msg_data = connection.fetch(msg_id, "(RFC822)")
|
||
if status == "OK" and msg_data and msg_data[0]:
|
||
raw_email = msg_data[0][1]
|
||
msg = email.message_from_bytes(raw_email)
|
||
emails.append(msg)
|
||
except Exception as e:
|
||
print(f" ⚠️ 解析邮件 {msg_id} 失败: {e}")
|
||
continue
|
||
|
||
return emails
|
||
|
||
except Exception as e:
|
||
print(f" ❌ IMAP 连接失败: {e}")
|
||
return None
|
||
finally:
|
||
if connection:
|
||
try:
|
||
connection.logout()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
# ==================== 方式3: Graph API 方式 ====================
|
||
|
||
|
||
def get_access_token_graph(client_id: str, refresh_token: str) -> Optional[str]:
|
||
"""
|
||
Graph API 方式获取 access_token
|
||
使用 login.microsoftonline.com/common 端点，Graph scope
|
||
"""
|
||
print(" 🔑 正在获取 access_token (Graph API)...")
|
||
|
||
try:
|
||
proxies = None
|
||
if PROXY:
|
||
proxies = {
|
||
"all": f"http://{PROXY}" if not PROXY.startswith("http") else PROXY
|
||
}
|
||
|
||
res = requests.post(
|
||
TOKEN_URL_GRAPH,
|
||
data={
|
||
"client_id": client_id,
|
||
"grant_type": "refresh_token",
|
||
"refresh_token": refresh_token,
|
||
"scope": "https://graph.microsoft.com/.default",
|
||
},
|
||
proxies=proxies,
|
||
timeout=30,
|
||
)
|
||
|
||
if res.status_code != 200:
|
||
print(f" ❌ 获取 access_token 失败: {res.status_code}")
|
||
print(f" 响应: {res.text[:200]}...")
|
||
if "User account is found to be in service abuse mode" in res.text:
|
||
print(" ⚠️ 账号被封禁!")
|
||
return None
|
||
|
||
access_token = res.json().get("access_token")
|
||
if access_token:
|
||
print(f" ✅ 成功获取 access_token，长度: {len(access_token)}")
|
||
return access_token
|
||
|
||
except Exception as e:
|
||
print(f" ❌ 获取 access_token 异常: {e}")
|
||
return None
|
||
|
||
|
||
def read_emails_graph(
|
||
client_id: str, refresh_token: str, top: int = 10
|
||
) -> Optional[List[Dict]]:
|
||
"""
|
||
方式3: Graph API 方式读取邮件
|
||
使用 Microsoft Graph API
|
||
"""
|
||
print_separator("方式3: Graph API 方式")
|
||
|
||
# 1. 获取 access_token
|
||
access_token = get_access_token_graph(client_id, refresh_token)
|
||
if not access_token:
|
||
return None
|
||
|
||
# 2. 调用 Graph API 获取邮件
|
||
try:
|
||
proxies = None
|
||
if PROXY:
|
||
proxies = {"http": f"http://{PROXY}", "https": f"http://{PROXY}"}
|
||
|
||
print(" 📡 正在调用 Graph API...")
|
||
|
||
url = "https://graph.microsoft.com/v1.0/me/mailFolders/inbox/messages"
|
||
params = {
|
||
"$top": top,
|
||
"$select": "id,subject,from,receivedDateTime,isRead,hasAttachments,bodyPreview",
|
||
"$orderby": "receivedDateTime desc",
|
||
"$count": "true",
|
||
}
|
||
headers = {
|
||
"Authorization": f"Bearer {access_token}",
|
||
"Prefer": "outlook.body-content-type='text'",
|
||
}
|
||
|
||
res = requests.get(
|
||
url, headers=headers, params=params, proxies=proxies, timeout=30
|
||
)
|
||
|
||
if res.status_code != 200:
|
||
print(f" ❌ Graph API 调用失败: {res.status_code}")
|
||
print(f" 响应: {res.text[:200]}...")
|
||
return None
|
||
|
||
data = res.json()
|
||
messages = data.get("value", [])
|
||
total = data.get("@odata.count", len(messages))
|
||
print(f" 📬 收件箱共有 {total} 封邮件")
|
||
|
||
return messages
|
||
|
||
except Exception as e:
|
||
print(f" ❌ Graph API 调用异常: {e}")
|
||
return None
|
||
|
||
|
||
# ==================== 主函数 ====================
|
||
|
||
|
||
def main():
|
||
"""主函数：使用三种方式读取邮件"""
|
||
print("\n" + "🚀 Outlook 邮件读取测试工具")
|
||
print("=" * 80)
|
||
print(f"邮箱: {EMAIL}")
|
||
print(f"Client ID: {CLIENT_ID}")
|
||
print(f"Refresh Token: {REFRESH_TOKEN[:30]}..." if REFRESH_TOKEN else "未设置")
|
||
print(f"代理: {PROXY if PROXY else '无'}")
|
||
print("=" * 80)
|
||
|
||
# 检查配置
|
||
if EMAIL == "" or REFRESH_TOKEN == "":
|
||
print("\n⚠️ 请先配置邮箱信息！")
|
||
print(" 修改脚本顶部的 EMAIL, REFRESH_TOKEN, CLIENT_ID 变量")
|
||
return
|
||
|
||
results = {}
|
||
|
||
# 方式1: 旧版 IMAP
|
||
try:
|
||
emails_old = read_emails_imap_old(EMAIL, CLIENT_ID, REFRESH_TOKEN, top=10)
|
||
print_email_info(emails_old, "旧版 IMAP")
|
||
results["旧版 IMAP"] = "✅ 成功" if emails_old else "❌ 失败"
|
||
except Exception as e:
|
||
print(f"❌ 旧版 IMAP 异常: {e}")
|
||
results["旧版 IMAP"] = f"❌ 异常: {e}"
|
||
|
||
# 方式2: 新版 IMAP
|
||
try:
|
||
emails_new = read_emails_imap_new(EMAIL, CLIENT_ID, REFRESH_TOKEN, top=10)
|
||
print_email_info(emails_new, "新版 IMAP")
|
||
results["新版 IMAP"] = "✅ 成功" if emails_new else "❌ 失败"
|
||
except Exception as e:
|
||
print(f"❌ 新版 IMAP 异常: {e}")
|
||
results["新版 IMAP"] = f"❌ 异常: {e}"
|
||
|
||
# 方式3: Graph API
|
||
try:
|
||
emails_graph = read_emails_graph(CLIENT_ID, REFRESH_TOKEN, top=10)
|
||
print_email_info(emails_graph, "Graph API")
|
||
results["Graph API"] = "✅ 成功" if emails_graph else "❌ 失败"
|
||
except Exception as e:
|
||
print(f"❌ Graph API 异常: {e}")
|
||
results["Graph API"] = f"❌ 异常: {e}"
|
||
|
||
# 打印汇总
|
||
print_separator("测试结果汇总")
|
||
for method, result in results.items():
|
||
print(f" {method}: {result}")
|
||
|
||
print("\n" + "=" * 80)
|
||
print("测试完成!")
|
||
print("=" * 80 + "\n")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|