如何使用client.json密钥实现Gmail API单次/离线身份验证

1. 是否每次运行都需要身份验证？ #

不需要。只要你首次完成身份验证后妥善保存了刷新令牌（refresh token），后续应用可以自动用刷新令牌更换过期的访问令牌，全程无需人工干预。你现在遇到运行中token过期崩溃的问题，是现有代码的身份验证逻辑存在缺陷，不是Google API的机制要求重复验证。

2. 长期缓存凭证的核心逻辑 #

Google OAuth2的凭证分为两部分：

• 访问令牌（access token）：有效期1小时，过期即失效
• 刷新令牌（refresh token）：长期有效（除非你手动撤销授权、或者应用被Google判定为测试状态超过7天过期），可以用来申请新的访问令牌
  你只要把包含刷新令牌的凭证存在本地文件，每次使用前检查是否过期，过期就自动刷新，刷新后更新本地存的凭证文件，就能实现永久复用，不需要人工操作。

3. 现有代码的问题和优化方案 #

现有代码的问题 #

你现在的代码有三个核心缺陷：

• 身份验证逻辑重复冗余，get_credentials函数写了一遍，main函数里又重复实现了一遍
• 运行过程中没有做异常捕获，如果调用API时因为token过期报错，直接就崩溃退出，不会触发刷新逻辑
• 没有考虑Google Cloud测试应用的刷新令牌7天过期的特殊规则，容易出现长期运行失效的问题

优化后的完整实现 #

核心改动点：

• 复用统一的凭证获取逻辑，删掉冗余代码
• 增加API调用的异常捕获，遇到认证错误自动重新获取/刷新凭证
• 刷新凭证后自动更新本地token.json，不需要人工操作
• 首次部署的时候你在本地跑一次完成授权，把生成的token.json和credentials.json一起传到树莓派即可，后续无需再操作设备

# GMail Downloader using the GMail API
from __future__ import print_function
import os
import base64
import email
import datetime
import time

from googleapiclient.discovery import build
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from googleapiclient.errors import HttpError

# Pi libraries
import RPi.GPIO as GPIO

Relay_Ch1 = 26
Relay_Ch2 = 20
Relay_Ch3 = 21

GPIO.setwarnings(False)
GPIO.setmode(GPIO.BCM)

GPIO.setup(Relay_Ch1, GPIO.OUT)
GPIO.setup(Relay_Ch2, GPIO.OUT)
GPIO.setup(Relay_Ch3, GPIO.OUT)

SCOPES = ['https://mail.google.com/'] # Full permissions
TOKEN_PATH = 'token.json'
CREDS_PATH = 'credentials.json'

def get_gmail_service():
    """获取有效的Gmail服务实例，自动处理凭证刷新"""
    creds = None
    # 读取本地存储的凭证
    if os.path.exists(TOKEN_PATH):
        creds = Credentials.from_authorized_user_file(TOKEN_PATH, SCOPES)
    
    # 凭证无效/过期处理
    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            try:
                creds.refresh(Request())
            except Exception as e:
                # 刷新失败说明refresh_token也失效了，这种情况只有首次部署或者授权被撤销才会出现
                raise RuntimeError("凭证刷新失败，请重新进行首次授权") from e
        else:
            # 没有本地凭证，走首次授权流程，这个步骤只需要在部署前在你自己电脑上跑一次就行
            flow = InstalledAppFlow.from_client_secrets_file(CREDS_PATH, SCOPES)
            creds = flow.run_local_server(port=0)
        
        # 保存更新后的凭证到本地
        with open(TOKEN_PATH, 'w') as token:
            token.write(creds.to_json())
    
    return build('gmail', 'v1', credentials=creds)

def ListMessagesMatchingQuery(service, user_id, query=''):
    """List all Messages of the user's mailbox matching the query."""
    try:
        response = service.users().messages().list(userId=user_id,q=query).execute()
        messages = []
        if 'messages' in response:
            messages.extend(response['messages'])

        while 'nextPageToken' in response:
            page_token = response['nextPageToken']
            response = service.users().messages().list(userId=user_id, q=query,pageToken=page_token).execute()
            messages.extend(response['messages'])
        return messages
    except HttpError as e:
        if e.resp.status in [401, 403]:
            # 认证错误，返回空让外层重新获取服务
            return None
        raise

def SaveAttachments(service, user_id, msg_id):
    """Get and store attachment from Message with given id."""
    message = service.users().messages().get(userId=user_id, id=msg_id).execute()

    local_date = datetime.datetime.fromtimestamp(float(message['internalDate'])/1000.)
    date_str = local_date.strftime("%y-%m-%d_%H-%M-%S_")

    for part in message['payload']['parts']:
        if part['filename']:
            if 'data' in part['body']:
                data=part['body']['data']
            else:
                att_id=part['body']['attachmentId']
                att=service.users().messages().attachments().get(userId=user_id, messageId=msg_id,id=att_id).execute()
                data=att['data']
            file_data = base64.urlsafe_b64decode(data.encode('UTF-8'))
            path = date_str+part['filename']

            with open(path, 'wb') as f:
                f.write(file_data)

def SaveMessageBody(service, user_id, msg_id):
    """Save the body from Message with given id."""
    message = service.users().messages().get(userId=user_id, id=msg_id, format='raw').execute()
    msg_str = base64.urlsafe_b64decode(message['raw'].encode('ASCII'))
    mime_msg = email.message_from_bytes(msg_str)
    messageMainType = mime_msg.get_content_maintype()
    file_data = ''
    if messageMainType == 'multipart':
        for part in mime_msg.get_payload():
            partType = part.get_content_maintype()
            if partType == 'multipart':
                for multipart in part.get_payload():
                    multipartType = multipart.get_content_maintype()
                    if multipartType == 'text':
                        file_data += multipart.get_payload()
                        break
            elif partType == 'text':
                file_data += part.get_payload()
    elif messageMainType == 'text':
        file_data += mime_msg.get_payload()

    local_date = datetime.datetime.fromtimestamp(float(message['internalDate'])/1000.)
    date_str = local_date.strftime("%y-%m-%d_%H-%M-%S_")
    
    subject = GetSubject(service, user_id, msg_id)
    for c in r' []/\;,&><&*:%=+@!#^()|?^':
        subject = subject.replace(c,'_')
 
    path = date_str+subject+".txt"

    with open(path, 'w') as f:
        f.write(file_data)

def GetSubject(service, user_id, msg_id):
    """Returns the subject of the message with given id."""
    subject = ''
    message = service.users().messages().get(userId=user_id, id=msg_id).execute()
    payload = message["payload"]
    headers = payload["headers"]
    for header in headers:
        if header["name"] == "Subject":
            subject = header["value"]
            break
    return subject

def MarkAsRead(service, user_id, msg_id):
    """Marks the message with given id as read."""
    service.users().messages().modify(userId=user_id, id=msg_id, body={ 'removeLabelIds': ['UNREAD']}).execute()

def main():
    # 获取服务实例，失败自动重试
    service = None
    while service is None:
        try:
            service = get_gmail_service()
        except RuntimeError as e:
            print(f"获取服务失败，10秒后重试: {e}")
            time.sleep(10)
    
    messages = ListMessagesMatchingQuery(service, 'me', 'subject:"Emergency Alert Activated: GSATMicro" is:unread')
    if messages is None:
        # 认证失效，下次循环重新获取服务
        print("凭证失效，重新获取中")
        return
    if messages:
        for message in messages:
            print('Processing: '+GetSubject(service, 'me', message["id"]))
            SaveMessageBody(service, 'me', message["id"])
            MarkAsRead(service, 'me', message["id"])
            print("Emergency Alert received from GSATMicro..")
            print("\n Help will be on its way .. Turning relay ON.. \n")
            #Control the Channel 1
            print("\n Turning Relay 1 ON.. \n")
            GPIO.output(Relay_Ch1, GPIO.HIGH)
            print("Channel 1:The Common Contact is access to the Normal Open Contact!")
            time.sleep(0.5)
            GPIO.output(Relay_Ch1, GPIO.LOW)
            print("Channel 1:The Common Contact is access to the Normal Closed Contact!\n")
            time.sleep(0.5)
            GPIO.output(Relay_Ch1, GPIO.HIGH)

if __name__ == '__main__':
    print('GMail API Downloader')
    print('Press Ctrl-C to quit')
    try:
        while True:
            main()
            for i in range(60):
                time.sleep(1)
    except KeyboardInterrupt:
        print('Ctrl-C received!')
        GPIO.cleanup()

额外注意事项 #

• 如果你在Google Cloud控制台创建的OAuth应用是测试模式，刷新令牌的有效期只有7天，需要把应用发布成正式模式（无需审核，只要点击发布按钮即可），刷新令牌就会长期有效
• 不要把token.json和credentials.json上传到公共代码仓库，避免泄露账号权限

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