Python通过Gmail API直传CSV附件到GCP Bucket报错修复

报错本质是方法用错了：blob.upload_from_filename()的设计用途是读取本地磁盘文件上传，入参必须是合法的本地文件路径字符串。你把Gmail API返回的解码后字节流传入这个方法，方法内部会把字节对象当路径字符串去做MIME类型探测，自然触发cannot use a string pattern on a bytes-like object的类型错误。

要实现附件不落地直接上传GCS，不需要写临时文件，直接用GCS客户端原生支持的内存上传接口即可。

• 替换上传方法：将upload_from_filename替换为upload_from_string，该方法可直接接受内存中的字节/字符串内容完成上传，无需依赖本地文件。
• 清理原代码中的无效参数、未定义变量、拼写错误等问题，避免运行时触发其他异常。

import os
import base64
from typing import List
from Google import Create_Service
from google.cloud import storage

os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = r'mystorageaccont.json'

storage_client = storage.Client()
bucket_name = 'mybucketname'

def upload_to_bucket(blob_name, content, bucket_name, content_type='text/csv'):
    bucket = storage_client.get_bucket(bucket_name)
    blob = bucket.blob(blob_name)
    # 直接上传内存字节流，无需本地文件路径
    blob.upload_from_string(content, content_type=content_type)
    return blob

def search_emails(query_string: str, label_ids: List=None):
    try:
        message_list_response = service.users().messages().list(
            userId='me',
            labelIds=label_ids,
            q=query_string
        ).execute()

        message_items = message_list_response.get('messages', [])
        next_page_token = message_list_response.get('nextPageToken')
        
        while next_page_token:
            message_list_response = service.users().messages().list(
                userId='me',
                labelIds=label_ids,
                q=query_string,
                pageToken=next_page_token
            ).execute()

            message_items.extend(message_list_response.get('messages', []))
            next_page_token = message_list_response.get('nextPageToken')
        return message_items
    except Exception as e:
        raise Exception('No emails returned')
    
def get_file_data(message_id, attachment_id):
    response = service.users().messages().attachments().get(
        userId='me',
        messageId=message_id,
        id=attachment_id
    ).execute()

    file_data = base64.urlsafe_b64decode(response.get('data').encode('UTF-8'))
    return file_data
    
def get_message_detail(message_id, msg_format='metadata', metadata_headers: List=None):
    message_detail = service.users().messages().get(
        userId='me',
        id=message_id,
        format=msg_format,
        metadataHeaders=metadata_headers
    ).execute()
    return message_detail

def save_file_data(email_messages):
    for email_message in email_messages: 
        messageDetail = get_message_detail(email_message['id'], msg_format='full', metadata_headers=['parts'])  
        messageDetailPayload = messageDetail.get('payload') 
        if 'parts' in messageDetailPayload: 
            for msgPayload in messageDetailPayload['parts']: 
                file_name = msgPayload['filename'] 
                if file_name.endswith('.csv'):
                    body = msgPayload['body'] 
                    if 'attachmentId' in body: 
                        attachment_id = body['attachmentId'] 
                        attachment_content = get_file_data(email_message['id'], attachment_id) 
                        upload_to_bucket(file_name, attachment_content, bucket_name)

if __name__ == '__main__': 
    CLIENT_FILE = 'mycredentialsforconnectgmail.json' 
    API_NAME = 'gmail' 
    API_VERSION = 'v1' 
    SCOPES = ['https://mail.google.com/'] 
    service = Create_Service(CLIENT_FILE, API_NAME, API_VERSION, SCOPES) 
    query_string = 'has:attachment' 
    email_messages = search_emails(query_string) 
    save_file_data(email_messages)

• 全流程附件解码、上传都在内存中完成，不会在本地生成任何临时文件，完全符合不落地存储的需求。
• 上传时手动指定CSV的Content-Type为text/csv，避免GCS自动识别类型出错，后续下载、读取桶内文件时格式识别更准确。
• 顺手修复了原代码的几处隐性bug：query_stirng参数拼写错误、自定义异常NoEmailFound未定义、save_location变量未定义、接口返回列表取值无默认值可能触发KeyError的问题。

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