使用 Python 从 gmail api 下载附件

Question

我正在尝试使用 python 从 gmail 下载附件，但我无法从我的邮件中获取附件 ID。请在下面找到我的代码

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


def get_gmail_service():
    SCOPES = ['https://www.googleapis.com/auth/gmail.readonly']
    creds = None
    # The file token.json stores the user's access and refresh tokens, and is
    # created automatically when the authorization flow completes for the first
    # time.
    if os.path.exists('token.json'):
        creds = Credentials.from_authorized_user_file('token.json', SCOPES)
        # If there are no (valid) credentials available, let the user log in.
        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)
        # Save the credentials for the next run
        with open('token.json', 'w') as token:
            token.write(creds.to_json())

        try:
            # Call the Gmail API
            service = build('gmail', 'v1', credentials=creds)
            return service

        except HttpError as error:
        # TODO(developer) - Handle errors from gmail API.
            print(f'An error occurred: {error}')

def get_email_list():
    service = get_gmail_service()
    results = service.users().messages().list(userId='me',q='from:abc@gmail.com is:read').execute()
    # print(results.get('messages',[])[0].get('id',[]))
    return results.get('messages', [])[0].get('id', [])
    # return results.get('messages',[])

def get_email_content(message_id):
    service = get_gmail_service()
    attach = service.users().messages().get(userId='me',id =message_id).execute()
    attach_id = attach.get('payloads',[]).get('parts',[]).get('body',[])
    data = service.users().messages().get(userId='me',id = message_id).execute()
    return attach_id

if __name__ == '__main__':
    # get_email_list()
    print(get_email_content(get_email_list()))

请更正我的代码，以便我可以使用 gmail api 下载附件。

Answer 1

这段代码有两个主要问题。

1. results.get() 方法返回 Message 或 MessagePart 对象。所以你只需要使用一次get() 方法就可以得到完整的对象，然后你就可以针对你想要的对象的特定部分。

   例如。 results.get('messages', [])[0]['id']

2. 电子邮件的有效负载可以是多部分的（这意味着“部分”将是 MessagePart 对象的数组）。所以我们需要迭代来获得一个包含文件的“消息部分”。在这种情况下，我们可以检查 MessagePart 对象是否有文件名。

       parts = attach.get('payload',[])['parts']
       
       for i in parts:
           if( i['filename'] ):
               return i['body']['attachmentId'] 
   

   
   所以在处理了这两个问题之后，这是新的代码：

   import os.path
   from google.auth.transport.requests import Request
   from google.oauth2.credentials import Credentials
   from google_auth_oauthlib.flow import InstalledAppFlow
   from googleapiclient.discovery import build
   from googleapiclient.errors import HttpError
   
   
   def get_gmail_service():
       SCOPES = ['https://www.googleapis.com/auth/gmail.readonly']
       creds = None
       # The file token.json stores the user's access and refresh tokens, and is
       # created automatically when the authorization flow completes for the first
       # time.
       if os.path.exists('token.json'):
           creds = Credentials.from_authorized_user_file('token.json', SCOPES)
           # If there are no (valid) credentials available, let the user log in.
           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)
           # Save the credentials for the next run
           with open('token.json', 'w') as token:
               token.write(creds.to_json())
   
           try:
               # Call the Gmail API
               service = build('gmail', 'v1', credentials=creds)
               return service
   
           except HttpError as error:
           # TODO(developer) - Handle errors from gmail API.
               print(f'An error occurred: {error}')
   
   def get_email_list():
       service = get_gmail_service()
       results = service.users().messages().list(userId='me',q='from:abc@gmail.com is:read').execute()
       # print(results.get('messages',[])[0]['id'] )
       return results.get('messages', [])[0]['id']
       # return results.get('messages',[])
   
   def get_email_content(message_id):
       print(message_id)
       service = get_gmail_service()
       data = service.users().messages().get(userId='me',id = message_id).execute()
   
       attach = service.users().messages().get(userId='me',id =message_id).execute()
       parts = attach.get('payload',[])['parts']
       
       for i in parts:
           if( i['filename'] ):
               return i['body']['attachmentId'] 
   
   if __name__ == '__main__':
       # get_email_list()
       print(get_email_content(get_email_list()))