我有一个可以在本地运行的 python 脚本。它处理文本文件并将行导入到谷歌工作表中。我正在尝试使用云函数和云存储桶来运行此脚本来自动化该过程。
我可以将工作目录更改为存储桶目录吗?或者我是否必须从存储桶中以文本形式下载文件才能处理它们并运行代码?
如果我可以更改工作目录,脚本应该按原样运行,并且我不必使用 google.cloud 中的存储模块(我假设)。
如果我确实必须下载文件,我将不得不弄清楚如何使用 glob.glob() 循环遍历目录中的文件并仅处理某些文件。
有人知道解决这个问题的最佳方法吗?
我目前可以下载文件并处理它,但是,我必须显式调用文件名,并且无法使用 glob.glob 来查找与特定模式匹配的文件。
您可以在下面看到这段代码的前半部分,我可以下载文件并开始处理它。
在该函数中,如果我可以指向文件所在的目录,您可以看到它通常如何工作。
from google.cloud import storage
from cloudevents.http import CloudEvent
import functions_framework
import glob
import os
#Can I change the working directory?
#os.chdir(bucket)
#print(os.getcwd())
# Instantiates a client
client = storage.Client()
bucket_name = 'working_files'
bucket = client.get_bucket(bucket_name)
#This will show me all of the files in the directory
#blobs = client.list_blobs('working_files')
#for blob in blobs:
# print(blob.name)
##I have to call out the specific file name for this to work.
blob = bucket.blob("explicit_file_name.txt")
blob = blob.download_as_text()
#example of the work we will do in the file
Main_DF = blob[50:100]
##Splitting the data and cleaning out the = dividers
Main_DF = [row.strip().replace('=','').split(', ') for row in Main_DF]
##Getting rid of whitespace
Main_DF = [row[0].split() for row in Main_DF]
print(Main_DF)
@functions_framework.cloud_event
def hello_gcs(cloud_event: CloudEvent):
##example of how the code would typically work if I could just iterate through files in the direcotry
Daily_File = [f for f in glob.glob("*common_identifier*")]
if Daily_File == []:
print("No relevant file found")
#Daily_File
else:
print(Daily_File)
您需要做的是重新构建代码并使用Python的云存储库。你已经开始使用它了。然后您可以从那里处理文件内容。我建议使用官方文档作为起点,因为它包含许多很好的示例。