使用Google API转发Gmail邮件时字典类型消息编码问题求助

问题描述 #

通过Google API客户端实现邮件转发时，读取的消息为字典类型，尝试调用as_bytes()编码时触发AttributeError，无法完成修改后的邮件发送。

原代码 #

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
import os
import base64
from datetime import datetime
SCOPES = ['https://www.googleapis.com/auth/gmail.readonly','https://www.googleapis.com/auth/gmail.modify', 'https://www.googleapis.com/auth/gmail.send']

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(               
            # your creds file here. Please create json file as here https://cloud.google.com/docs/authentication/getting-started
            '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())

    
service = build('gmail', 'v1', credentials=creds)
results = service.users().messages().list(userId='me', labelIds=['INBOX']).execute()
messages = results.get('messages',[]);


message = messages[0]
msg = service.users().messages().get(userId='me', id=message['id']).execute()
print(type(msg))

msg['To'] = 'me@mydomain.com'

encoded_message = base64.urlsafe_b64encode(message.as_bytes()).decode()
create_message = {"raw": encoded_message}

message = (service.users().messages().send(userId='me', body=body).execute())

报错信息 #

<class 'dict'>
---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
Cell In[57], line 32
     28 print(type(msg))
     30 msg['To'] = 'me@mydomain.com'
---&gt; 32 encoded_message = base64.urlsafe_b64encode(message.as_bytes()).decode()
     33 create_message = {"raw": encoded_message}
     35 message = (service.users().messages().send(userId='me', body=body).execute())

AttributeError: 'dict' object has no attribute 'as_bytes'

解决方案 #

Gmail API默认返回的是结构化字典，无法直接调用as_bytes()。正确做法是获取原始RFC822格式的邮件内容，修改后重新编码发送：

修正后的代码 #

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
import os
import base64
from datetime import datetime
SCOPES = ['https://www.googleapis.com/auth/gmail.readonly','https://www.googleapis.com/auth/gmail.modify', 'https://www.googleapis.com/auth/gmail.send']

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())

service = build('gmail', 'v1', credentials=creds)
results = service.users().messages().list(userId='me', labelIds=['INBOX']).execute()
messages = results.get('messages', [])

if not messages:
    print("No messages found.")
else:
    message_id = messages[0]['id']
    # 获取原始格式的邮件内容
    msg = service.users().messages().get(userId='me', id=message_id, format='raw').execute()
    
    # 解码Base64编码的原始邮件
    raw_email = base64.urlsafe_b64decode(msg['raw']).decode('utf-8')
    
    # 拆分邮件头与正文，修改收件人字段
    email_parts = raw_email.split('\n\n', 1)
    headers = email_parts[0]
    body = email_parts[1] if len(email_parts) > 1 else ''
    
    new_headers = []
    to_updated = False
    for line in headers.split('\n'):
        if line.lower().startswith('to:'):
            # 替换原收件人
            new_headers.append('To: me@mydomain.com')
            to_updated = True
        elif line.lower().startswith('cc:') or line.lower().startswith('bcc:'):
            # 可选：移除原抄送/密送
            continue
        else:
            new_headers.append(line)
    # 如果原邮件没有To字段，添加新的To
    if not to_updated:
        new_headers.append('To: me@mydomain.com')
    
    # 重新组合邮件内容
    updated_email = '\n'.join(new_headers) + '\n\n' + body
    
    # 重新编码为URL安全的Base64格式
    encoded_message = base64.urlsafe_b64encode(updated_email.encode('utf-8')).decode('utf-8')
    create_message = {"raw": encoded_message}
    
    # 发送修改后的邮件
    sent_message = service.users().messages().send(userId='me', body=create_message).execute()
    print(f"邮件已发送，ID: {sent_message['id']}")

关键说明 #

• 获取原始邮件：调用messages().get()时添加format='raw'参数，直接获取Base64编码的完整RFC822邮件内容，而非结构化字典。
• 编码解码处理：使用base64.urlsafe_b64decode解码原始邮件，修改后再用base64.urlsafe_b64encode重新编码，确保符合Gmail API的raw字段要求。
• 邮件头修改：手动拆分并修改邮件头，避免直接修改字典导致的格式错误，保证邮件结构合规。

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