基于Graph API与服务主体的Microsoft邮件读取Python程序开发技术问询

嗨，针对你这个基于Graph API和服务主体的Microsoft邮件读取Python程序开发需求，我来分享一些实操思路和解决方案，应该能帮你覆盖到所有核心要求：

一、实现用户级权限控制（而非租户级） #

要确保服务主体只能访问已授权用户的数据，你得避开租户级的应用权限，改用委托权限+OAuth授权码流程：

• 在Azure AD注册应用时，只添加Mail.Read这类委托权限，不要申请Mail.Read.All这种全局应用权限；
• 让用户通过授权码流程完成单次授权，获取该用户专属的刷新令牌，后续用这个令牌调用Graph API时，只会访问该授权用户的邮箱数据，完全符合隐私要求。

二、参数化配置与一次性授权 #

你可以这样设计授权流程，实现用户只需授权一次：

1. 首次引导用户走OAuth授权码流程，拿到用户的刷新令牌、邮箱地址等信息，把这些数据存到配置文件（比如authorized_users.json）或数据库里；
2. 后续新增用户时，只需要让新用户完成一次授权，把他的信息追加到存储列表即可；
3. 程序启动时读取这个列表，逐个用对应用户的刷新令牌获取访问令牌，再调用API读取邮件。

举个简单的配置文件示例：

{
  "authorized_users": [
    {
      "email": "user1@example.com",
      "refresh_token": "xxx..."
    },
    {
      "email": "user2@example.com",
      "refresh_token": "yyy..."
    }
  ]
}

三、全量提取与增量/Delta提取实现 #

1. 全量提取（分页处理）

Graph API默认单次返回25条邮件，超过数量需要通过分页获取全部数据，你可以检查响应里的@odata.nextLink字段循环请求：

import requests

def fetch_all_emails(access_token, user_email, folder="inbox"):
    emails = []
    url = f"https://graph.microsoft.com/v1.0/users/{user_email}/mailFolders/{folder}/messages"
    while url:
        response = requests.get(url, headers={"Authorization": f"Bearer {access_token}"})
        response.raise_for_status()
        data = response.json()
        emails.extend(data.get("value", []))
        url = data.get("@odata.nextLink")
    return emails

2. 增量/Delta提取（自定义时间范围）

有两种实用的增量提取方式：

• 时间过滤法：直接用$filter参数筛选指定时间范围的邮件，比如：

  # 提取2024年1月到6月的收件箱邮件
  url = f"https://graph.microsoft.com/v1.0/users/{user_email}/mailFolders/inbox/messages?$filter=receivedDateTime ge 2024-01-01T00:00:00Z and receivedDateTime le 2024-06-01T23:59:59Z"
  

• Delta查询法：适合长期增量同步，能追踪邮件的新增、修改和删除。首次调用/messages/delta，后续用返回的@odata.deltaLink获取上次查询后的变化数据：

  def fetch_delta_emails(access_token, user_email, delta_link=None):
    url = delta_link if delta_link else f"https://graph.microsoft.com/v1.0/users/{user_email}/mailFolders/inbox/messages/delta"
    response = requests.get(url, headers={"Authorization": f"Bearer {access_token}"})
    response.raise_for_status()
    data = response.json()
    return data.get("value", []), data.get("@odata.deltaLink")
  

你可以把delta_link存下来，下次增量提取直接复用。

四、提取唯一邮箱地址 #

拿到邮件列表后，遍历邮件的发件人、收件人、抄送人、密送人字段，用集合自动去重：

def extract_unique_emails(emails):
    unique_emails = set()
    for email in emails:
        # 提取发件人
        if email.get("from") and email["from"].get("emailAddress"):
            unique_emails.add(email["from"]["emailAddress"]["address"])
        # 提取各类收件人
        for recipient_type in ["toRecipients", "ccRecipients", "bccRecipients"]:
            for recipient in email.get(recipient_type, []):
                if recipient.get("emailAddress"):
                    unique_emails.add(recipient["emailAddress"]["address"])
    return list(unique_emails)

额外提醒 #

• 刷新令牌有效期默认90天，要在程序里处理令牌过期逻辑，用旧刷新令牌获取新的访问/刷新令牌并更新存储；
• 要处理用户撤销授权、网络异常等情况，添加错误捕获和重试逻辑；
• Azure AD应用的重定向URI要和授权流程里的一致，避免授权失败。

备注：内容来源于stack exchange，提问作者user27548440