如何修改Python函数获取AWS S3多层级子文件夹最新文件路径

需要编写或调整函数，获取AWS S3存储桶多层级子文件夹内最新修改文件的路径。原函数在2层级子文件夹下可正常工作，但针对3层级子文件夹（如everest/consumos/bd_f_consumos_eess/）无法正确获取结果。

原函数代码：

def get_file_path(s3, bucket, prefix):
    """
    Getting last modified file in the desired bucket.
    :param bucket_name
    :param prefix 
    :return:
    """
    objs = s3.list_objects_v2(Bucket=bucket, Prefix=prefix)
    if 'Contents' in objs:
        objs = objs['Contents']
        print("objs", objs)
        get_last_modified = lambda obj: int(obj['LastModified'].strftime('%s'))
        last_modified_file_prefix = "/".join([obj['Key'] for obj in sorted(objs, key=get_last_modified, reverse=True)][0].split("/")[:-1])
        logger.info('Last modified file prefix is: %s', last_modified_file_prefix)
        return f"s3://{bucket}/{last_modified_file_prefix}/"
    else:
        return None

原函数正常调用示例（2层级子文件夹）：

table_nob = get_file_path(s3,BUSINESS_BUCKET_NAME, "totem/NOB_COD_DIAS/")

需要适配的3层级子文件夹调用：

table_vols = get_file_path(s3, BUSINESS_BUCKET_NAME, 'everest/consumos/bd_f_consumos_eess/')

尝试修改return语句为return f"s3://{bucket}//{last_modified_file_prefix}/"后仍无法解决问题。

问题出在路径拼接的细节和虚拟文件夹的处理上，修改以下几点即可适配任意层级子文件夹：

关键修正点 #

1. 修复参数文档错误：原函数注释写的bucket_name与实际参数bucket不符，统一后避免混淆。
2. 过滤虚拟文件夹条目：S3的文件夹是虚拟概念，list_objects_v2可能返回前缀本身的"文件夹"条目，需要跳过这类非文件对象。
3. 避免路径重复斜杠：原修改中的//会导致无效路径，保持单斜杠拼接即可。

方案1：获取最新文件的完整S3路径 #

如果需求是直接获取最新修改文件的完整路径，使用以下函数：

def get_file_path(s3, bucket, prefix):
    """
    获取指定S3存储桶前缀下最新修改的文件完整路径
    :param s3: boto3 S3客户端对象
    :param bucket: 存储桶名称
    :param prefix: 目标文件夹前缀（可带末尾斜杠）
    :return: 最新文件的S3路径，无文件则返回None
    """
    objs = s3.list_objects_v2(Bucket=bucket, Prefix=prefix)
    if 'Contents' not in objs:
        return None
    
    # 按最后修改时间倒序排序
    sorted_objs = sorted(objs['Contents'], key=lambda x: x['LastModified'], reverse=True)
    latest_obj_key = sorted_objs[0]['Key']
    
    # 跳过虚拟文件夹条目，取第一个实际文件
    if latest_obj_key.endswith('/'):
        if len(sorted_objs) > 1:
            latest_obj_key = sorted_objs[1]['Key']
        else:
            return None
    
    logger.info('最新修改文件路径: %s', latest_obj_key)
    return f"s3://{bucket}/{latest_obj_key}"

方案2：获取最新文件所在的文件夹路径 #

如果仍需要返回最新文件所在的文件夹路径，使用以下函数：

def get_file_path(s3, bucket, prefix):
    """
    获取指定S3存储桶前缀下最新修改文件所在的文件夹路径
    :param s3: boto3 S3客户端对象
    :param bucket: 存储桶名称
    :param prefix: 目标文件夹前缀（可带末尾斜杠）
    :return: 文件夹的S3路径，无文件则返回None
    """
    objs = s3.list_objects_v2(Bucket=bucket, Prefix=prefix)
    if 'Contents' not in objs:
        return None
    
    sorted_objs = sorted(objs['Contents'], key=lambda x: x['LastModified'], reverse=True)
    latest_obj_key = sorted_objs[0]['Key']
    
    # 跳过虚拟文件夹条目
    if latest_obj_key.endswith('/'):
        if len(sorted_objs) > 1:
            latest_obj_key = sorted_objs[1]['Key']
        else:
            return None
    
    # 分割出文件所在的文件夹路径
    folder_path = "/".join(latest_obj_key.split("/")[:-1])
    logger.info('最新文件所在文件夹前缀: %s', folder_path)
    return f"s3://{bucket}/{folder_path}/"

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