如何使用 Python 和 Boto3 高效统计 AWS S3 特定文件

import boto3
from urllib.parse import urlparse

def count_matching_s3_objects(s3_path, file_start_pattern='file_', file_end_pattern='.ts'):
    """
    统计指定 S3 路径（前缀）下，符合特定命名模式的文件数量。
    该函数会遍历指定前缀下的所有子文件夹。

    Args:
        s3_path (str): 目标 S3 路径，可以是完整的 S3 URL (s3://bucket/prefix/) 或直接的 bucket/prefix。
                       例如: 's3://coursevideotesting/Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/360p/'
                       或 'coursevideotesting/Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/360p/'
        file_start_pattern (str): 文件名必须以此字符串开头 (例如 'file_')。
        file_end_pattern (str): 文件名必须以此字符串结尾 (例如 '.ts')。

    Returns:
        int: 符合条件的文件总数。
        tuple: (int, str, str) - 返回文件总数、S3桶名称和S3前缀。
               如果解析 S3 路径失败，则返回 (0, None, None)。
    """
    bucket_name = None
    prefix = ''

    # 解析 S3 路径，提取桶名称和前缀
    if s3_path.startswith('s3://'):
        parsed_url = urlparse(s3_path)
        bucket_name = parsed_url.netloc
        # path 部分通常以 '/' 开头，需要去除
        prefix = parsed_url.path.lstrip('/')
    elif '/' in s3_path:
        # 假设格式是 'bucket_name/prefix/path'
        parts = s3_path.split('/', 1)
        bucket_name = parts[0]
        prefix = parts[1] if len(parts) > 1 else ''
    else:
        # 如果只有桶名，则前缀为空
        bucket_name = s3_path
        prefix = ''

    if not bucket_name:
        print(f"错误: 无法从路径 '{s3_path}' 解析出有效的 S3 桶名称。")
        return 0, None, None

    s3 = boto3.resource('s3')
    try:
        bucket = s3.Bucket(bucket_name)
    except Exception as e:
        print(f"错误: 无法连接到 S3 桶 '{bucket_name}'。请检查桶名称和权限。详细信息: {e}")
        return 0, bucket_name, prefix

    count = 0
    print(f"正在检查 S3 桶: {bucket_name}, 前缀: {prefix}")

    try:
        # 使用 filter(Prefix=prefix) 限制从 S3 获取的对象，boto3.resource 会自动处理分页。
        for obj in bucket.objects.filter(Prefix=prefix):
            # 1. 排除“文件夹”对象：S3 中的文件夹实际上是零字节且以 '/' 结尾的对象。
            #    我们只关心实际的文件。
            if obj.key.endswith('/'):
                continue

            # 2. 确保对象键确实以指定前缀开头（尽管 filter 已经做了大部分工作，
            #    但为了逻辑严谨性，特别是当 prefix 不以 '/' 结尾时，
            #    或者需要处理更复杂的边界情况时，可以再加一层检查）。
            #    此处我们假设 prefix 已经正确，直接处理文件名。

            # 3. 提取文件名部分：从完整的对象键中获取文件名（即最后一个 '/' 后的部分）。
            #    例如，对于 'path/to/file_001.ts'，文件名是 'file_001.ts'。
            filename = obj.key.split('/')[-1]

            # 4. 应用文件名模式匹配。
            if filename.startswith(file_start_pattern) and filename.endswith(file_end_pattern):
                count += 1
        print(f"在 '{bucket_name}/{prefix}' 中找到 {count} 个匹配文件。")
        return count, bucket_name, prefix

    except Exception as e:
        print(f"在 S3 桶 '{bucket_name}'，前缀 '{prefix}' 中统计文件时发生错误: {e}")
        return 0, bucket_name, prefix

import csv
# 假设上面的 count_matching_s3_objects 函数已经定义

# --- 配置参数 ---
# 基础 S3 URL，如果 CSV 中的 URL 是相对路径，则需要拼接
base_s3_url = 's3://coursevideotesting/' 
# 输入 CSV 文件名
input_csv_file = 'ldt_ffw_course_videos_temp.csv'
# 输出 CSV 文件名
output_csv_file = 'file_count_result.csv'

# 模拟输入 CSV 文件内容 (ldt_ffw_course_videos_temp.csv)
# URL_Column_Name,Expected_Files_Column_Name
# Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/144p/,2