如何在同一GCS Bucket内实现不同文件夹之间的文件移动？

首先你需要安装Google Cloud Storage的Python官方依赖库：

pip install google-cloud-storage

实现原理 #

GCS对象存储没有原生的「移动」接口，移动同Bucket内文件的本质逻辑是：先将源文件复制到目标路径，再删除源路径的文件。

可用代码示例 #

你可以直接在你现有处理流程结束后插入以下代码段，替换对应参数即可使用：

from google.cloud import storage

def move_gcs_file(bucket_name, source_blob_name, dest_blob_name):
    """移动GCS同Bucket内的单个文件
    Args:
        bucket_name: GCS桶名称
        source_blob_name: 源文件的完整路径，比如 input/测试文件.csv
        dest_blob_name: 目标文件的完整路径，比如 archived/测试文件.csv
    """
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    source_blob = bucket.blob(source_blob_name)
    
    # 复制到目标路径
    bucket.copy_blob(source_blob, bucket, dest_blob_name)
    # 删除源文件
    source_blob.delete()
    print(f"文件已移动: {source_blob_name} -> {dest_blob_name}")

# ------------------- 调用示例 -------------------
# 替换为你的实际参数
YOUR_BUCKET_NAME = "你的GCS桶名称"
SOURCE_FOLDER_PREFIX = "input/"
DEST_FOLDER_PREFIX = "archived/"

# 假设你已经把处理完成的所有源文件路径存在了processed_files列表里
processed_files = [
    "input/文件1.csv",
    "input/文件2.json"
]

for file_path in processed_files:
    # 生成目标路径：把源前缀替换为目标前缀即可
    dest_path = file_path.replace(SOURCE_FOLDER_PREFIX, DEST_FOLDER_PREFIX, 1)
    move_gcs_file(YOUR_BUCKET_NAME, file_path, dest_path)

注意事项 #

• 上述代码默认会覆盖目标路径下的同名文件，如果需要避免覆盖，可以在复制前加逻辑判断bucket.blob(dest_blob_name).exists()，存在则跳过或者重命名
• 建议添加try-except异常捕获逻辑，避免复制环节出错后直接删除源文件，造成数据丢失
• 如果需要移动input文件夹下的所有文件，不需要单独维护processed_files列表，可以直接遍历桶下所有前缀为SOURCE_FOLDER_PREFIX的blob：

  # 遍历input文件夹下所有文件的示例
  blobs = bucket.list_blobs(prefix=SOURCE_FOLDER_PREFIX)
  for blob in blobs:
      # 跳过文件夹本身的占位blob
      if blob.name.endswith("/"):
          continue
      dest_path = blob.name.replace(SOURCE_FOLDER_PREFIX, DEST_FOLDER_PREFIX, 1)
      move_gcs_file(YOUR_BUCKET_NAME, blob.name, dest_path)
  

• 执行脚本前请确认你的服务账号/本地认证有对应GCS桶的对象创建、对象删除权限

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