GCP Bucket递归清理疑问：Python删除旧CSV与空文件夹的困惑

针对你的GCP Bucket路径结构和清理需求，不需要复杂的递归方案，利用Blob路径的字符串特性和GCP存储的虚拟文件夹机制，就能高效实现目标。以下是具体步骤和Python代码实现：

核心思路 #

1. 先清理过期CSV文件：筛选出超过4天的文件删除，同时记录这些文件所在的目录。
2. 从深到浅清理空文件夹：GCP的文件夹是虚拟存在的，只有当路径下有Blob（包括文件或文件夹占位符）时才会显示。我们按目录深度从大到小（先day层级，再month，最后year）检查，确保子目录清理完毕后再处理父目录，避免遗漏。

Python代码实现 #

1. 依赖安装 #

pip install google-cloud-storage

2. 完整代码 #

from google.cloud import storage
from datetime import datetime, timedelta
import os

def cleanup_gcp_bucket(bucket_name):
    client = storage.Client()
    bucket = client.get_bucket(bucket_name)
    cutoff_time = datetime.now() - timedelta(days=4)
    affected_dirs = set()

    # 第一步：删除过期CSV文件
    for blob in bucket.list_blobs(prefix="reporting/"):
        # 仅处理CSV文件
        if not blob.name.endswith(".csv"):
            continue
        # 用Blob的更新时间判断是否过期（也可改用路径中的日期，见下方说明）
        if blob.updated.replace(tzinfo=None) < cutoff_time:
            blob.delete()
            # 记录文件所在目录
            dir_path = os.path.dirname(blob.name)
            affected_dirs.add(dir_path)

    # 第二步：清理空文件夹
    # 按目录深度倒序排序，优先处理子目录
    sorted_dirs = sorted(affected_dirs, key=lambda x: x.count("/"), reverse=True)

    while sorted_dirs:
        dir_path = sorted_dirs.pop(0)
        # 检查目录下是否有内容（文件或子目录）
        # 使用delimiter="/"只获取当前目录的直接子项
        blobs_iter = bucket.list_blobs(prefix=f"{dir_path}/", delimiter="/")
        has_files = any(1 for _ in blobs_iter)
        has_subdirs = bool(blobs_iter.prefixes)

        if not has_files and not has_subdirs:
            # 删除文件夹占位符（如果存在）
            folder_blob = bucket.blob(f"{dir_path}/")
            if folder_blob.exists():
                folder_blob.delete()
            # 将父目录加入待检查列表（子目录清空后父目录可能变空）
            parent_dir = os.path.dirname(dir_path)
            if parent_dir and parent_dir not in affected_dirs:
                affected_dirs.add(parent_dir)
                # 重新排序目录列表
                sorted_dirs = sorted(affected_dirs, key=lambda x: x.count("/"), reverse=True)

if __name__ == "__main__":
    cleanup_gcp_bucket("your-bucket-name")

关键细节说明 #

• 过期时间判断：代码中用Blob的updated时间判断过期，如果你的文件命名/路径规范严格（比如路径中的日期就是文件生成日期），可以改用路径解析日期的方式，更精准：

  # 从路径reporting/YYYY/MM/DD/file.csv解析日期
  path_parts = blob.name.split("/")
  if len(path_parts) >= 4:
      try:
          file_date = datetime(int(path_parts[1]), int(path_parts[2]), int(path_parts[3]))
          if file_date.date() < cutoff_time.date():
              # 执行删除操作
              blob.delete()
              dir_path = os.path.dirname(blob.name)
              affected_dirs.add(dir_path)
      except ValueError:
          # 路径格式异常，跳过该文件
          continue
  

• 虚拟文件夹处理：如果你的Bucket中没有显式创建过文件夹（即没有名称以/结尾的占位符Blob），当目录下没有文件时，GCP会自动隐藏该目录，无需额外删除操作。代码中保留了删除占位符的逻辑，兼容显式创建文件夹的场景。
• 目录遍历顺序：按目录深度倒序处理，确保子目录先被清理，父目录只有在子目录清空后才会被检查，避免父目录提前被误判为空。

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