Python+Boto3高效统计AWS S3指定文件方法

1. 问题背景与挑战

在处理 AWS S3 中的大量数据时，经常需要统计特定类型的文件。例如，在一个存储了视频切片（chunks）的S3桶中，视频文件可能按质量（如144p/、360p/）和切片序号（如file_000.ts、file_001.ts等）组织。我们的目标是：

1. 针对给定的S3路径（前缀），统计其中符合特定命名模式（例如，文件名以file_开头并以.ts结尾）的文件数量。
2. 能够遍历该路径下的所有子文件夹，确保所有层级的文件都能被统计到。
3. 高效处理S3可能返回的超过1000个对象的场景（即分页问题）。

传统的 boto3.client 方法，如 list_objects_v2，虽然提供了强大的控制，但在处理分页和迭代所有对象时可能需要额外的逻辑。不正确的 Prefix 和文件名匹配逻辑也可能导致计数不准确。

2. 为什么选择 boto3.resource？

Boto3 提供了两种主要方式与AWS服务交互：client 和 resource。

• boto3.client: 提供低级别、一对一的服务操作，直接映射到 AWS API 调用。对于需要精细控制每个请求（如设置分页令牌、最大返回数量等）的场景非常有用。
• boto3.resource: 提供更高级别、面向对象的抽象。它封装了许多底层 API 调用，使得代码更加简洁和“Pythonic”。对于需要迭代集合（如S3对象、EC2实例）的场景，resource 会自动处理分页，大大简化了开发。

对于S3对象计数这类需要遍历大量对象的任务，boto3.resource 的自动分页能力使其成为更简洁、更易于维护的选择。

3. 核心计数逻辑实现

我们将创建一个函数来封装统计逻辑，使其可复用。

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

4. 实际应用示例

假设您有一个CSV文件，其中包含需要检查的S3路径和期望的文件数量。我们可以将上述计数函数集成到您的CSV处理流程中。

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