如何正确实现GCS桶内文本文件定时归档的Cloud Function？

原代码的核心问题 #

你的代码存在多个关键错误，导致执行失败：

• storage_client.bucket()仅接受纯桶名，不能包含gs://前缀或文件夹路径
• 目标路径处理错误：归档操作在同一个存储桶内完成，无需重新创建bucket对象，只需指定目标文件的前缀路径
• 变量名大小写错误：list_blobs(bucketName)应为list_blobs(bucket_name)
• 危险操作：origin.delete()会删除整个存储桶，而非源文件
• 未过滤文本文件：原代码会复制所有类型文件，不符合需求
• 异常处理过于笼统：无法定位具体失败原因，不利于排查

修正后的完整实现 #

requirements.txt #

google-cloud-storage>=2.14.0

main.py #

from google.cloud import storage
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def copy_daily_trans_file(request):
    """
    将指定GCS桶中某文件夹内的所有文本文件，复制到同桶的archive文件夹，并删除源文件
    请求参数：
    - bucket: 存储桶名称（必填，不带gs://前缀）
    - source_prefix: 源文件夹前缀（可选，默认incr-exports/）
    """
    # 获取请求参数
    bucket_name = request.args.get('bucket')
    source_prefix = request.args.get('source_prefix', 'incr-exports/')
    dest_prefix = f"{source_prefix}archive/"

    if not bucket_name:
        return "错误：未提供存储桶名称，请在请求参数中传入bucket", 400

    try:
        storage_client = storage.Client()
        bucket = storage_client.bucket(bucket_name)

        # 列出源文件夹下的所有文本文件（过滤后缀为.txt的文件，可根据需求调整）
        blobs = bucket.list_blobs(prefix=source_prefix)
        text_blobs = [blob for blob in blobs if blob.name.endswith('.txt') and not blob.name.endswith('/')]

        if not text_blobs:
            logger.info("未找到需要归档的文本文件")
            return "未找到需要归档的文本文件", 200

        for blob in text_blobs:
            # 构造目标文件路径
            dest_blob_name = blob.name.replace(source_prefix, dest_prefix, 1)
            dest_blob = bucket.blob(dest_blob_name)

            # 复制文件
            dest_blob.copy_from(blob)
            logger.info(f"已复制文件：{blob.name} -> {dest_blob_name}")

            # 删除源文件
            blob.delete()
            logger.info(f"已删除源文件：{blob.name}")

        return f"成功归档{len(text_blobs)}个文本文件", 200

    except Exception as e:
        logger.error(f"归档失败：{str(e)}")
        return f"归档失败：{str(e)}", 500

正确调用方式 #

请求URL示例（替换为你的函数地址和桶名）：

https://us-central1-data-prod-datalake.cloudfunctions.net/func-archive-exports/?bucket=my-bucket&source_prefix=incr-exports/

注意：

• bucket参数仅传入桶名，不要带gs://前缀或路径
• source_prefix可选，若你的源文件夹不是incr-exports/，可自行修改

定时执行配置 #

方法1：使用Cloud Scheduler触发HTTP函数 #

1. 进入Google Cloud控制台的Cloud Scheduler页面
2. 创建新任务：
   • 频率：设置定时规则（例如0 0 * * *表示每天凌晨执行）
   • 目标类型：选择HTTP
   • URL：填入你的Cloud Function触发URL，带上正确的bucket和source_prefix参数
   • HTTP方法：选择GET
   • 身份验证：选择服务帐号，并确保该服务帐号有Cloud Functions调用权限和GCS读写权限

方法2：使用Cloud Functions定时触发器（仅2nd gen） #

如果使用Cloud Functions 2nd gen，可直接配置定时触发器，无需HTTP触发：

1. 修改函数为无参数形式（移除request参数）
2. 在函数部署时，设置定时触发器（例如0 0 * * *）
3. 代码中直接指定桶名和源前缀，或通过环境变量配置

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