如何使用 Gmail API 在 python 中批量获取邮件标头？

Question

我已经查看了我能找到的所有答案，最接近的答案只提供了四行代码，我无法让它在我的较大脚本中工作。

我想要：

1. 从我的收件箱下载消息 ID 的完整列表
2. 对于每个消息 ID，仅在 500 条消息的批次中获取选定的标头

我一生都无法获得正确的语法，因为我找不到完整的示例，只有片段，而且 chatGPT 总是出错......

# Create a Gmail API service client - 'creds' already obtained
service = build('gmail', 'v1', credentials=creds)

# List all message IDs in blocks of 500
all_message_ids = []
results = service.users().messages().list(userId='me', labelIds=['INBOX'], maxResults=500).execute()
messages = results.get('messages', [])
all_message_ids.extend([message['id'] for message in messages])

while 'nextPageToken' in results:
    page_token = results['nextPageToken']
    results = service.users().messages().list(userId='me', labelIds=['INBOX'], maxResults=500, pageToken=page_token).execute()
    messages = results.get('messages', [])
    all_message_ids.extend([message['id'] for message in messages])

print(f'Total messages: {len(all_message_ids)}')

headers_array = []

# Retrieve headers for each message ID
for i in range(0, len(all_message_ids), 500):
    batch_message_ids = all_message_ids[i:i+500]
    batch = service.new_batch_http_request()
    for msg_id in batch_message_ids:
        batch.add(service.users().messages().get(userId = 'me', id = msg_id, format='metadata', metadataHeaders=['From', 'Reply-To', 'Sender', 'Return-Path']))
    batch.execute() # <--- BREAKS HERE WITH EXCEPTION

我怀疑批量请求格式错误，因为我看到一个添加回调参数的示例，但我还没有完全理解回调语法，并且不确定如何集成它或是否需要它。

我想在批处理响应中看到/接收的是一个 JSON 对象，其中包含批处理请求中发送的所有 500 个消息 ID 的选定标头。 然后，循环将发送另一个批处理请求，其中包含接下来的 500 个消息 ID，依此类推，直到我收到收件箱中所有消息的选定标头。

我已经阅读了 Gmail API 和 Google Apps 脚本文档（在你建议之前），坦率地说，它只是缺乏细节和全面的示例。我还阅读了 API 批处理指南，但该示例是针对搜索或表格脚本的，不太相关，而且我不知道如何将其翻译为 Gmail 上的作品。

我也用 Google Apps 脚本尝试过此操作，但有一个基本限制，即标签不能应用于消息，只能应用于线程，而似乎可以通过 Gmail API 使用 Python 将标签应用于消息。

目前，另一个不批量处理请求的工作脚本的性能非常慢 - 每秒处理 1 到 2 封电子邮件。我还有将近15万件事要做...

Answer 1

现阶段，官方文件是这样说的。 参考

单个批量请求中的调用次数限制为 100 次。如果您需要进行更多调用，请使用多个批量请求。请注意这一点。

那么，在您的展示脚本中，进行以下修改如何？

来自：

headers_array = []

# Retrieve headers for each message ID
for i in range(0, len(all_message_ids), 500):
    batch_message_ids = all_message_ids[i:i+500]
    batch = service.new_batch_http_request()
    for msg_id in batch_message_ids:
        batch.add(service.users().messages().get(userId = 'me', id = msg_id, format='metadata', metadataHeaders=['From', 'Reply-To', 'Sender', 'Return-Path']))
    batch.execute() # <--- BREAKS HERE WITH EXCEPTION

致：

global headers_array # Added
headers_array = []

# Retrieve headers for each message ID
for i in range(0, len(all_message_ids), 500):
    batch_message_ids = all_message_ids[i:i+100] # Modified
    batch = service.new_batch_http_request(callback=sample)
    for msg_id in batch_message_ids:
        batch.add(service.users().messages().get(userId='me', id=msg_id, format='metadata',metadataHeaders=['From', 'Reply-To', 'Sender', 'Return-Path']))
    batch.execute()

print(headers_array)

作为回调函数，请添加以下函数。

def sample(id, res, err):
    # print(id)
    # print(err)
    headers_array.append(res["payload"]["headers"])

• 运行此修改后的脚本时，只能从每封电子邮件中检索标题。

参考资料：

• Gmail API 的批量请求
• new_batch_http_request()