如何通过Gmail API或Pub/Sub在Python中标记指定邮箱的新旧邮件

核心问题分析 #

1. 权限不足：当前使用的readonly权限仅能读取邮件，无法修改标签，需添加gmail.modify权限。
2. Pub/Sub消息解析错误：你尝试用message.message_id获取Gmail邮件ID，但这是Pub/Sub自身的消息ID，正确的Gmail邮件ID存储在消息的data字段中，需解析JSON后提取。
3. Watch过滤条件不符合需求：当前请求仅监听未读邮件，若要标记所有未来邮件，需调整监听范围（比如监听收件箱所有新邮件）。
4. 缺少标签标记逻辑：拿到邮件ID后，需调用Gmail API的modify接口添加目标标签。

修正后的完整代码 #

from __future__ import print_function

from concurrent.futures import TimeoutError
from google.cloud import pubsub_v1
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
import os.path
import json

from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow

# 更新权限：添加modify权限，用于修改邮件标签
SCOPES = [
    'https://www.googleapis.com/auth/gmail.readonly',
    'https://www.googleapis.com/auth/gmail.modify'
]

# 替换成你的目标标签ID
TARGET_LABEL_ID = "your_custom_label_id"

creds = None
if os.path.exists('token.json'):
    creds = Credentials.from_authorized_user_file('token.json', SCOPES)
if not creds or not creds.valid:
    if creds and creds.expired and creds.refresh_token:
        creds.refresh(Request())
    else:
        flow = InstalledAppFlow.from_client_secrets_file(
            'credentials.json', SCOPES)
        creds = flow.run_local_server(port=0)
    with open('token.json', 'w') as token:
        token.write(creds.to_json())

# 替换成你的Google Cloud项目ID和Pub/Sub主题、订阅ID
project_id = "your_project_id"
topic_name = "your_topic_name"
subscription_id = "your_subscription_id"
timeout = 180.0

service = build('gmail', 'v1', credentials=creds)

# 配置Watch请求：监听收件箱所有新邮件（可根据需求调整过滤条件）
watch_request = {
    'labelIds': ['INBOX'],  # 监听收件箱的新邮件
    'labelFilterAction': 'include',
    'topicName': f'projects/{project_id}/topics/{topic_name}'
}
try:
    watch_response = service.users().watch(userId='me', body=watch_request).execute()
    print(f"Watch已启动，历史ID: {watch_response['historyId']}")
except HttpError as error:
    print(f"启动Watch失败: {error}")
    exit()

subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(project_id, subscription_id)

def callback(message: pubsub_v1.subscriber.message.Message) -> None:
    try:
        # 解析Pub/Sub消息中的Gmail事件数据
        data = json.loads(message.data.decode('utf-8'))
        # 获取邮件ID列表
        for history_entry in data.get('history', []):
            for message_added in history_entry.get('messagesAdded', []):
                gmail_message_id = message_added['message']['id']
                print(f"收到新邮件，ID: {gmail_message_id}")
                
                # 为邮件添加目标标签
                modify_body = {
                    'addLabelIds': [TARGET_LABEL_ID],
                    # 如果需要移除其他标签，可添加'removeLabelIds': ['UNREAD']等
                }
                service.users().messages().modify(
                    userId='me',
                    id=gmail_message_id,
                    body=modify_body
                ).execute()
                print(f"已为邮件 {gmail_message_id} 添加标签")
    
    except Exception as e:
        print(f"处理消息出错: {e}")
    finally:
        message.ack()  # 确认消息已处理，避免重复投递

streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)
print(f"正在监听Pub/Sub订阅 {subscription_path}...\n")

with subscriber:
    try:
        streaming_pull_future.result(timeout=timeout)
    except TimeoutError:
        streaming_pull_future.cancel()
        streaming_pull_future.result()

关键修改说明 #

• 权限更新：新增gmail.modify权限，确保代码具备修改邮件标签的权限。
• Watch请求调整：将labelIds设为INBOX，监听所有收件箱新邮件；若只需监听特定类型邮件，可调整为其他标签（如UNREAD）。
• 消息解析修正：通过json.loads(message.data.decode('utf-8'))解析Gmail发送的事件数据，从history.messagesAdded.message.id中提取真实的Gmail邮件ID。
• 标签标记逻辑：调用users().messages().modify接口，通过addLabelIds参数添加目标标签。
• 错误处理：增加异常捕获，避免单个消息处理失败导致整个订阅中断，并确保消息被确认（ack()），防止重复投递。

额外注意事项 #

1. 确保Google Cloud项目已启用Gmail API和Pub/Sub API，且Pub/Sub主题与订阅已正确关联。
2. TARGET_LABEL_ID需要替换为你创建的特定标签的ID，可通过Gmail API的users.labels.list接口获取。
3. 若需长期运行，建议将脚本部署为后台服务（如使用systemd、Docker等），避免超时停止。
4. 处理重复消息：Gmail的Watch可能发送重复的历史事件，可通过记录已处理的邮件ID避免重复标记。

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