Python GCP-将文件上传到GCP存储桶

Question

我从Gmail收件箱中获取CSV文件，我正在使用Gmail API，一旦获得了这些CSV文件，我不想将它们存储在本地，我想将这些文件从GCP连接到我的存储桶中，但是我无法获得我的路径从CSV文件。

 upload_to_bucket(file_name, attachment_content,bucket_name)
  File "/Users/Emails/mypython”, line 91, in upload_to_bucket
    blob.upload_from_filename(file_path)
  File "/Users/.local/share/virtualenvs/Emails-wLEQ9xGC/lib/python3.10/site-packages/google/cloud/storage/blob.py", line 2704, in upload_from_filename
    content_type = self._get_content_type(content_type, filename=filename)
  File "/Users/.local/share/virtualenvs/Emails-wLEQ9xGC/lib/python3.10/site-packages/google/cloud/storage/blob.py", line 1674, in _get_content_type
    content_type, _ = mimetypes.guess_type(filename)
  File "/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/mimetypes.py", line 307, in guess_type
    return _db.guess_type(url, strict)
  File "/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/mimetypes.py", line 123, in guess_type
    scheme, url = urllib.parse._splittype(url)
  File "/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/urllib/parse.py", line 1039, in _splittype
    match = _typeprog.match(url)
TypeError: cannot use a string pattern on a bytes-like object

我已经进行了测试以将这些CSV文件从收件箱中获取到本地，但知道我添加了“ upload_bycket”函数，但是我尝试从附件获得的file_path，但找不到它也无法找到它，

import os
import base64
from typing import List
import time
from datetime import datetime
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, file_path, bucket_name):
    bucket = storage_client.get_bucket(bucket_name)
    blob = bucket.blob(blob_name)
    blob.upload_from_filename(file_path)
    return blob

def search_emails(query_stirng: 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 NoEmailFound('No emails returned'

)
    
def get_file_data(message_id, attachment_id, file_name, new_Location):
    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'])  
        headers=messageDetail["payload"]["headers"]
        messageDetailPayload = messageDetail.get('payload') 
        if 'parts' in messageDetailPayload: 
            for msgPayload in messageDetailPayload['parts']: 
                file_name = msgPayload['filename'] 
                filetype = ".csv"
                if file_name.find(filetype) != -1:
                    body = msgPayload['body'] 
                    if 'attachmentId' in body: 
                        attachment_id = body['attachmentId'] 
                        attachment_content = get_file_data(email_message['id'], attachment_id, file_name, save_location) 
                        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)


                 

我也没有找到一个小型应用程序，它使用我的本地和我的local和它有效，但是现在路径在我的变量上从收件箱中

Answer 1

您的错误说无法在类似字节的对象上使用字符串模式

python 3将您的数据输入字节。您无法编码它。更改代码

file_data = base64.urlsafe_b64decode（response.get（'data'）。encode（'utf-8''））

to

file_data = base64.urlsafe_b64decode(response.get('data'))