如何修改Python方法实现S3文件夹最新创建文件的下载？

有两种可靠的方案可以实现需求，优先推荐基于S3文件LastModified字段的方案（更准确，不依赖文件名格式），也可以选择解析文件名日期的方案。

方案一：基于S3文件最后修改时间（推荐） #

步骤1：修改文件列表获取方法，直接返回最新文件 #

修正原get_file_folders的分页逻辑错误，同时收集文件的最后修改时间，筛选出最新文件：

session_root = boto3.Session(region_name='eu-west-3', profile_name='my_profile')
s3_client = session_root.client('s3')

def get_latest_file(s3_client, bucket_name, prefix=""):
    file_info = []
    folders = []

    default_kwargs = {
        "Bucket": bucket_name,
        "Prefix": prefix
    }
    next_token = ""

    while next_token is not None:
        updated_kwargs = default_kwargs.copy()
        if next_token != "":
            updated_kwargs["ContinuationToken"] = next_token

        # 修复分页参数传递错误，确保能获取所有分页数据
        response = s3_client.list_objects_v2(**updated_kwargs)
        contents = response.get("Contents", [])

        for result in contents:
            key = result.get("Key")
            if key[-1] == "/":
                folders.append(key)
            else:
                # 保存文件路径和最后修改时间
                file_info.append({
                    'key': key,
                    'last_modified': result['LastModified']
                })

        next_token = response.get("NextContinuationToken")

    if not file_info:
        return None, folders
    
    # 筛选出最后修改时间最晚的文件
    latest_file = max(file_info, key=lambda x: x['last_modified'])['key']
    return latest_file, folders

步骤2：修改下载方法，仅下载最新文件 #

调整download_files为针对单个文件的下载逻辑：

from pathlib import Path

def download_latest_file(s3_client, bucket_name, local_path, latest_file, folders):
    local_path = Path(local_path)

    # 创建必要的文件夹结构（若需要保留S3的层级）
    for folder in folders:
        folder_path = Path.joinpath(local_path, folder)
        folder_path.mkdir(parents=True, exist_ok=True)

    if latest_file:
        file_path = Path.joinpath(local_path, latest_file)
        file_path.parent.mkdir(parents=True, exist_ok=True)
        s3_client.download_file(
            bucket_name,
            latest_file,
            str(file_path)
        )
        print(f"已完成最新文件下载: {latest_file}")
    else:
        print("指定S3路径下未找到任何文件")

使用示例 #

# 配置S3桶和目标前缀
bucket_name = "my_Bucket"
target_prefix = "folder_1/folder_2/folder_3/folder_4/"

# 获取最新文件
latest_file, folders = get_latest_file(s3_client, bucket_name, target_prefix)

# 下载最新文件到本地路径
download_latest_file(s3_client, bucket_name, "./local_download_dir", latest_file, folders)

方案二：基于文件名中的日期解析（依赖固定文件名格式） #

如果必须通过文件名中的日期判断最新文件，可使用以下方法解析文件名中的年份和月份：

步骤1：添加文件名日期解析函数 #

from datetime import datetime

def get_latest_file_by_filename(file_names):
    latest_file = None
    latest_date = None
    for file in file_names:
        # 提取文件名部分
        filename = file.split("/")[-1]
        parts = filename.split("_")
        try:
            # 假设日期位于文件名的第3、4个下划线分割段（对应示例中的2021_03）
            year = int(parts[2])
            month = int(parts[3])
            file_date = datetime(year, month, 1)
            if not latest_date or file_date > latest_date:
                latest_date = file_date
                latest_file = file
        except (IndexError, ValueError):
            # 跳过格式不符合的文件
            continue
    return latest_file

步骤2：结合原有方法使用 #

# 使用原有方法获取所有文件列表
file_names, folders = get_file_folders(s3_client, bucket_name, target_prefix)

# 筛选出文件名日期最新的文件
latest_file = get_latest_file_by_filename(file_names)

# 修改原download_files方法，仅下载单个文件
def download_single_file(s3_client, bucket_name, local_path, file_name):
    local_path = Path(local_path)
    file_path = Path.joinpath(local_path, file_name)
    file_path.parent.mkdir(parents=True, exist_ok=True)
    s3_client.download_file(bucket_name, file_name, str(file_path))
    print(f"已下载文件: {file_name}")

# 执行下载
if latest_file:
    download_single_file(s3_client, bucket_name, "./local_download_dir", latest_file)
else:
    print("未找到符合日期格式的文件")

注意：方案二依赖文件名的固定格式，若文件名格式变更会失效，优先选择方案一。

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