Google Cloud Function中Python移三月旧文件代码无报错但失效求助

以下是你的代码中存在的核心问题及对应的修复方案：

1. GCS前缀匹配不支持通配符（代码未执行的核心原因） #

Google Cloud Storage的list_blobs方法的prefix参数仅支持前缀匹配，不支持*这类通配符。你写的history/*/DATA_*、history/*/CTRL_*无法匹配到任何文件，导致外层循环直接跳过，后续逻辑完全没执行。

修复方式：
先获取history/下的所有子目录，再遍历每个子目录下以DATA_和CTRL_开头的文件：

# 先获取history下的所有子目录前缀
subdirs = []
for blob in source_bucket.list_blobs(prefix='history/', delimiter='/'):
    if blob.name.endswith('/') and blob.name != 'history/':
        subdirs.append(blob.name)

# 遍历每个子目录，匹配DATA_和CTRL_开头的文件
prefixes = []
for subdir in subdirs:
    prefixes.append(f"{subdir}DATA_")
    prefixes.append(f"{subdir}CTRL_")

2. 时区不一致导致时间判断失效 #

datetime.datetime.now()获取的是本地时间，而GCS中blob.updated返回的是UTC时间，两者时区不统一会导致时间比较结果错误，符合条件的文件被误判为不满足要求。

修复方式：
改用datetime.datetime.utcnow()获取UTC时间，并统一时区格式：

three_month_ago = datetime.datetime.utcnow() - datetime.timedelta(days=90)
three_month_ago = three_month_ago.replace(tzinfo=datetime.timezone.utc)

3. Blob上传方法错误 #

GCS的Blob对象没有upload_files方法，正确的文件上传方法是upload_from_filename，这个错误会导致即使前面逻辑正常，也无法将压缩文件上传到目标桶。

修复方式：
替换错误方法：

destination_blob.upload_from_filename(zip_Name)

4. 临时目录与文件路径问题 #

• 当blob.name包含多级子路径时，os.path.join(temp_dir, blob.name)会生成嵌套路径，但你只创建了temp根目录，未创建子目录，下载文件时会抛出FileNotFoundError。
• os.rmdir(temp_dir)放在循环外，若未创建过temp目录（无符合条件的文件），会直接报错；且如果temp下有子目录，os.rmdir也无法删除，需改用shutil.rmtree。

修复方式：

# 下载前先创建完整的文件目录
local_file_path = os.path.join(temp_dir, blob.name)
os.makedirs(os.path.dirname(local_file_path), exist_ok=True)
blob.download_to_filename(local_file_path)

# 最后删除临时目录改用shutil.rmtree
import shutil
if os.path.exists(temp_dir):
    shutil.rmtree(temp_dir)

5. 语法错误（多余转义符） #

代码中的f\"{blob.name}.zip\"存在多余的转义符，会导致语法错误，应改为f"{blob.name}.zip"。

修复后的完整代码 #

from google.cloud import storage
import zipfile
import datetime
import os
import shutil

def move_files(request, context):
    source_bucket_name = 'source_bucket_name'
    destination_bucket_name = 'destination_bucket_name'
    client = storage.Client()

    source_bucket = client.get_bucket(source_bucket_name)
    destination_bucket = client.get_bucket(destination_bucket_name)

    # 使用UTC时间，统一时区
    three_month_ago = datetime.datetime.utcnow() - datetime.timedelta(days=90)
    three_month_ago = three_month_ago.replace(tzinfo=datetime.timezone.utc)
    
    # 获取history下的所有子目录
    subdirs = []
    for blob in source_bucket.list_blobs(prefix='history/', delimiter='/'):
        if blob.name.endswith('/') and blob.name != 'history/':
            subdirs.append(blob.name)

    # 生成每个子目录下的目标前缀
    prefixes = []
    for subdir in subdirs:
        prefixes.append(f"{subdir}DATA_")
        prefixes.append(f"{subdir}CTRL_")

    temp_dir = 'temp'
    os.makedirs(temp_dir, exist_ok=True)

    for prefix in prefixes:
        for blob in source_bucket.list_blobs(prefix=prefix):
            blob_updated = blob.updated
            if blob_updated < three_month_ago:
                # 构建本地文件路径并创建目录
                local_file_path = os.path.join(temp_dir, blob.name)
                os.makedirs(os.path.dirname(local_file_path), exist_ok=True)

                # 下载文件
                blob.download_to_filename(local_file_path)

                # 压缩文件（添加压缩参数减小体积）
                zip_name = os.path.join(temp_dir, f"{blob.name}.zip")
                with zipfile.ZipFile(zip_name, 'w', zipfile.ZIP_DEFLATED) as zipfile_obj:
                    zipfile_obj.write(local_file_path, blob.name)

                # 上传压缩文件到目标桶
                destination_blob = destination_bucket.blob(f"{blob.name}.zip")
                destination_blob.upload_from_filename(zip_name)

                # 删除本地文件
                os.remove(local_file_path)
                os.remove(zip_name)

                # 删除源桶中的原文件
                blob.delete()

    # 清理临时目录
    if os.path.exists(temp_dir):
        shutil.rmtree(temp_dir)
    print("successful")

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