如何提升提取Gmail邮件标签的Python脚本运行速度？

你的脚本速度慢的核心原因是冗余的API调用，而非谷歌的硬性限制（Gmail API有每日配额，但合理优化后完全能支撑大型账户的导出需求）。以下是具体优化方案：

1. 预加载所有标签的完整路径 #

当前每次处理邮件时，都会递归调用labels().get获取标签路径，这会产生大量重复请求。可以一次性拉取所有标签，提前构建label_id -> 完整路径的映射字典：

def get_all_label_paths(service):
    label_paths = {}
    # 获取所有标签
    results = service.users().labels().list(userId='me').execute()
    labels = results.get('labels', [])
    
    # 先构建父级与名称的映射
    parent_map = {label['id']: label.get('parent') for label in labels}
    name_map = {label['id']: label['name'] for label in labels}
    
    # 为每个标签生成完整路径
    for label_id in name_map:
        path = []
        current_id = label_id
        while current_id:
            path.append(name_map[current_id])
            current_id = parent_map.get(current_id)
        # 反转路径得到从父到子的层级结构
        label_paths[label_id] = '/'.join(reversed(path))
    return label_paths

之后处理邮件时，直接从字典中取路径，无需再调用API。

2. 批量获取邮件数据 #

原来的循环中每次单独调用messages().get，可以改用批量请求，一次请求获取多封邮件的信息，大幅减少HTTP请求次数：

from googleapiclient.http import BatchHttpRequest

def batch_get_messages(service, message_ids, label_paths):
    messages_data = []
    
    def callback(request_id, response, exception):
        if exception:
            print(f"获取邮件失败: {exception}")
            return
        # 提取Message-ID
        headers = response['payload']['headers']
        msg_id = next((h['value'] for h in headers if h['name'].lower() == 'message-id'), None)
        # 获取标签路径
        label_ids = response.get('labelIds', [])
        labels = [label_paths[label_id] for label_id in label_ids]
        messages_data.append((msg_id, labels))
    
    batch = service.new_batch_http_request(callback=callback)
    for msg_id in message_ids:
        # 只请求需要的字段，减少数据传输量
        batch.add(service.users().messages().get(
            userId='me',
            id=msg_id,
            format='metadata',
            metadataHeaders=['Message-ID'],
            fields='payload/headers,labelIds'
        ))
    batch.execute()
    return messages_data

3. 精简请求字段 #

调用messages().get时，通过format='metadata'、metadataHeaders和fields参数，只获取你需要的Message-ID和labelIds，避免下载整个邮件内容，减少数据处理和传输时间。

4. 增大分页查询数量 #

messages().list默认每页返回10条邮件，最大支持500条。调用时设置maxResults=500，减少分页请求的次数：

results = service.users().messages().list(
    userId='me',
    pageToken=page_token,
    maxResults=500
).execute()

优化后的完整核心代码 #

def get_all_label_paths(service):
    label_paths = {}
    results = service.users().labels().list(userId='me').execute()
    labels = results.get('labels', [])
    
    parent_map = {label['id']: label.get('parent') for label in labels}
    name_map = {label['id']: label['name'] for label in labels}
    
    for label_id in name_map:
        path = []
        current_id = label_id
        while current_id:
            path.append(name_map[current_id])
            current_id = parent_map.get(current_id)
        label_paths[label_id] = '/'.join(reversed(path))
    return label_paths

def batch_get_messages(service, message_ids, label_paths):
    messages_data = []
    
    def callback(request_id, response, exception):
        if exception:
            print(f"获取邮件失败: {exception}")
            return
        headers = response['payload']['headers']
        msg_id = next((h['value'] for h in headers if h['name'].lower() == 'message-id'), None)
        label_ids = response.get('labelIds', [])
        labels = [label_paths[label_id] for label_id in label_ids]
        messages_data.append((msg_id, labels))
    
    batch = service.new_batch_http_request(callback=callback)
    for msg_id in message_ids:
        batch.add(service.users().messages().get(
            userId='me',
            id=msg_id,
            format='metadata',
            metadataHeaders=['Message-ID'],
            fields='payload/headers,labelIds'
        ))
    batch.execute()
    return messages_data

def get_message_ids_with_labels(service):
    # 预加载所有标签路径
    label_paths = get_all_label_paths(service)
    
    page_token = None
    with open(OUTPUT_FILE_PATH, 'w') as output_file:
        while True:
            # 每页获取500条邮件
            results = service.users().messages().list(
                userId='me',
                pageToken=page_token,
                maxResults=500
            ).execute()
            messages = results.get('messages', [])
            
            if not messages:
                break
            
            # 提取当前页的所有邮件ID
            message_ids = [msg['id'] for msg in messages]
            # 批量获取邮件数据
            messages_data = batch_get_messages(service, message_ids, label_paths)
            
            # 写入文件
            for msg_id, labels in messages_data:
                output_file.write(f"Message-ID: {msg_id} - Labels: {labels}\n")

            page_token = results.get('nextPageToken')
            if not page_token:
                break

关于谷歌API配额的说明 #

Gmail API的默认每日配额是100,000次调用，优化后的脚本调用次数会大幅降低：

• 标签只需要1次调用
• 分页调用次数 = 总邮件数 / 500
• 批量邮件调用次数 = 总邮件数 / 100（批量请求每次最多支持100个请求）

对于10万封邮件的账户，总调用次数仅约1000次，远低于配额限制。

内容的提问来源于stack exchange，提问作者x0100