Gmail API推送通知问题：新邮件触发两次通知及过滤需求

问题1：重复收到通知的修复 #

重复通知核心原因是多次调用watch方法创建了多份活跃订阅，同一邮箱事件会被多个订阅同时捕获，导致重复通知。

修复方案：
每次启动脚本时先停止旧的活跃订阅，再创建新订阅，确保仅存在一份有效监听：

# 停止旧的watch订阅，避免重复通知
try:
    # 尝试停止当前活跃的订阅
    service.users().stop(userId='me').execute()
except HttpError as e:
    # 无活跃订阅时会返回404，忽略该错误
    if e.resp.status != 404:
        print(f"停止旧订阅时出错: {e}")

问题2：仅过滤新邮件到达事件 #

要实现只在新邮件到达时触发通知，需从watch配置和事件解析两方面入手：

步骤1：修正watch请求配置 #

你之前的exclude配置错误（包含UNREAD标签会排除所有未读邮件相关事件，而新邮件默认是未读状态），正确配置应聚焦收件箱的变化：

request = {
  'labelIds': ['INBOX'],  # 仅监听收件箱的标签变化
  'labelFilterAction': 'include',
  'topicName': 'projects/projectgmailapi-366508/topics/mytopic'
}

步骤2：在回调中精准解析事件类型 #

Pub/Sub消息仅携带历史ID，需调用Gmail API获取详细历史记录，过滤出新增邮件到收件箱的事件：

import json

def callback(message: pubsub_v1.subscriber.message.Message) -> None:
    try:
        # 解析Pub/Sub消息内容
        msg_data = json.loads(message.data.decode('utf-8'))
        start_history_id = msg_data['historyId']
        
        # 仅获取新增邮件的历史记录
        history_response = service.users().history().list(
            userId='me',
            startHistoryId=start_history_id,
            historyTypes=['messageAdded'],
            maxResults=5
        ).execute()
        
        # 验证是否是收件箱新增邮件
        if 'history' in history_response:
            for history_item in history_response['history']:
                if 'messagesAdded' in history_item:
                    for msg_added in history_item['messagesAdded']:
                        # 检查邮件是否在收件箱中
                        msg = service.users().messages().get(
                            userId='me',
                            id=msg_added['message']['id'],
                            format='metadata',
                            metadataHeaders=['Subject']
                        ).execute()
                        subject = next((h['value'] for h in msg['payload']['headers'] if h['name'] == 'Subject'), '无主题')
                        print(f"收到新邮件: {subject}")
                        # 此处添加你的通知逻辑（如弹窗、短信等）
    except Exception as e:
        print(f"处理消息时出错: {e}")
    
    message.ack()

完整修正后的代码 #

from __future__ import print_function

from concurrent.futures import TimeoutError
from google.cloud import pubsub_v1
from googleapiclient.discovery import build

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
from googleapiclient.errors import HttpError

# 若修改权限范围，需删除token.json
SCOPES = ['https://www.googleapis.com/auth/gmail.readonly']


creds = None
# token.json存储用户的访问和刷新令牌，首次授权后自动生成
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())

# 开发者配置项
project_id = "projectgmailapi-366508"
subscription_id = "mysub"
# 订阅者监听消息的超时时间（秒）
timeout = 180.0

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

# 停止旧的watch订阅，避免重复通知
try:
    service.users().stop(userId='me').execute()
except HttpError as e:
    if e.resp.status != 404:
        print(f"停止旧订阅时出错: {e}")

# 创建新的watch订阅，仅监听收件箱变化
request = {
  'labelIds': ['INBOX'],
  'labelFilterAction': 'include',
  'topicName': 'projects/projectgmailapi-366508/topics/mytopic'
}
service.users().watch(userId='me', body=request).execute()

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消息内容
        msg_data = json.loads(message.data.decode('utf-8'))
        start_history_id = msg_data['historyId']
        
        # 获取仅包含新增邮件的历史记录
        history_response = service.users().history().list(
            userId='me',
            startHistoryId=start_history_id,
            historyTypes=['messageAdded'],
            maxResults=5
        ).execute()
        
        # 检查是否有新邮件添加到收件箱的事件
        if 'history' in history_response:
            for history_item in history_response['history']:
                if 'messagesAdded' in history_item:
                    for msg_added in history_item['messagesAdded']:
                        # 验证邮件是否在收件箱中
                        msg = service.users().messages().get(
                            userId='me',
                            id=msg_added['message']['id'],
                            format='metadata',
                            metadataHeaders=['Subject']
                        ).execute()
                        subject = next((h['value'] for h in msg['payload']['headers'] if h['name'] == 'Subject'), '无主题')
                        print(f"收到新邮件: {subject}")
                        # 此处添加自定义通知逻辑
    except Exception as e:
        print(f"处理消息时出错: {e}")
    
    message.ack()

streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)
print(f"正在 {subscription_path} 监听消息..\n")

# 使用with块自动关闭订阅者
with subscriber:
    try:
        streaming_pull_future.result(timeout=timeout)
    except TimeoutError:
        streaming_pull_future.cancel()  # 触发关闭
        streaming_pull_future.result()  # 等待关闭完成

关键说明 #

1. 单订阅监听：通过先停止旧订阅，确保同一时间仅存在一份活跃的Gmail watch订阅，从根源避免重复通知。
2. 精准事件过滤：结合historyTypes=['messageAdded']和收件箱标签验证，仅对新增到收件箱的邮件触发通知，排除删除、已读等无关操作。
3. 配置合理性：监听INBOX标签而非UNREAD，避免因邮件已读状态变化触发无效通知。

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