如何使用 gmail api 获取特定用户的第一条消息

Question

如何使用 gmail api 获取特定用户的第一条消息（例如 facebook 给我的第一条消息）

目前我正在使用此代码来获取所有消息

from googleapiclient.discovery import build
from httplib2 import Http
from oauth2client import file, client, tools

SCOPES = 'https://www.googleapis.com/auth/gmail.readonly'

def main():

    store = file.Storage('token.json')
    creds = store.get()
if not creds or creds.invalid:
    flow = client.flow_from_clientsecrets('credentials.json', SCOPES)
    creds = tools.run_flow(flow, store)
service = build('gmail', 'v1', http=creds.authorize(Http()))

# Call the Gmail API to fetch INBOX
results = service.users().messages().list(userId='me',maxResults=20,labelIds = ['INBOX']).execute()
messages = results.get('messages', [])

if not messages:
    print("No messages found.")
else:
    print("Message snippets:")
    for message in messages:
        msg = service.users().messages().get(userId='me', id=message['id']).execute()
        print(msg['snippet'])

if __name__ == '__main__':
    main()

Answer 1

通过查看您的代码，我认为您需要替换以下代码片段

for message in messages:
        msg = service.users().messages().get(userId='me', id=message['id']).execute()
        print(msg['snippet'])

使用以下代码片段，

msg = service.users().messages().get(userId='me', id=messages[-1]['id']).execute()
print(msg['snippet'])

Answer 2

获取消息长度：

last_index = len(messages) - 1

msg = service.users().messages().get(userId='me', id=messages[last_index]['id']).execute()
print(msg['snippet'])

Answer 3

导入操作系统 导入base64 导入压缩文件 从 googleapiclient.discovery 导入构建

def download_attachments(creds): 服务=构建（'gmail'，'v1'，凭据=凭据）

# List messages
results = service.users().messages().list(userId='me').execute()
messages = results.get('messages', [])

# Create a zip file to store attachments
zip_filename = 'attachments.zip'
with zipfile.ZipFile(zip_filename, 'w') as zipf:
    for message in messages:
        # Get the full message data
        msg = service.users().messages().get(userId='me', id=message['id'], format='full').execute()
        # Check if 'payload' exists and if it has 'parts'
        if 'payload' in msg and 'parts' in msg['payload']:
            for part in msg['payload']['parts']:
                # Check if this part has a filename and is an attachment
                if 'filename' in part and part['filename'] and 'attachmentId' in part['body']:
                    attachment_id = part['body']['attachmentId']
                    attachment = service.users().messages().attachments().get(
                        userId='me',
                        messageId=message['id'],
                        id=attachment_id
                    ).execute()
                    
                    # Decode the attachment data
                    data = attachment['data']
                    file_data = base64.urlsafe_b64decode(data.encode('UTF-8'))
                    
                    # Write the file to the zip archive
                    zipf.writestr(part['filename'], file_data)
                    print(f'Added to zip: {part["filename"]}')
        else:
            # Handle cases without 'parts' (single part messages)
            if 'payload' in msg and 'body' in msg['payload'] and 'attachmentId' in msg['payload']['body']:
                attachment_id = msg['payload']['body']['attachmentId']
                attachment = service.users().messages().attachments().get(
                    userId='me',
                    messageId=message['id'],
                    id=attachment_id
                ).execute()

                data = attachment['data']
                file_data = base64.urlsafe_b64decode(data.encode('UTF-8'))
                
                # Use a unique filename based on message ID to avoid collisions
                zipf.writestr(f'attachment_{message["id"]}.bin', file_data)
                print(f'Added to zip: attachment_{message["id"]}.bin')

print(f'All attachments have been zipped into {zip_filename}')

def new_func（服务，消息）： msg = service.users().messages().get(userId='me', id=message['id'], format='full').execute() 返回消息