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

在 AWS S3 中管理大量文件时，经常需要统计特定目录（前缀）下符合某种命名规则的文件数量。例如，视频处理场景中，一个视频文件可能被切分为多个增量块，如 file_000.ts, file_001.ts 等，并存储在不同质量（如 144p, 360p）的嵌套文件夹中。本教程将指导您如何使用 Python 和 boto3 库来自动化这一过程。

Boto3 客户端与资源对象：选择与优势

boto3 提供了两种主要接口来与 AWS 服务交互：client 和 resource。

• client 接口 (低级)：提供与 AWS API 一对一的映射，功能最全面，但通常需要手动处理分页（例如，当 list_objects_v2 返回超过 1000 个对象时）。
• resource 接口 (高级)：提供更高级、更面向对象的方法，封装了许多底层细节，包括自动处理分页。对于大多数日常操作，尤其是需要遍历大量 S3 对象的场景，resource 接口通常更简洁、更易用。

在统计 S3 文件数量时，由于可能存在大量文件，boto3.resource 自动处理分页的能力使其成为更优的选择，避免了手动检查 IsTruncated 和 NextContinuationToken 的复杂性。

核心实现：统计 S3 中特定前缀下的文件

要准确统计 S3 中特定前缀（即文件夹）下符合特定命名模式的文件，我们需要：

1. 正确解析 S3 URL，提取桶名和前缀。
2. 使用 boto3.resource 获取桶对象。
3. 利用 bucket.objects.filter(Prefix=...) 过滤出指定前缀下的所有对象。
4. 遍历这些对象，并根据文件名进行模式匹配。

以下是一个核心函数的实现：

import boto3
import re
from urllib.parse import urlparse

def get_bucket_and_prefix_from_s3_url(s3_url: str) -> tuple[str, str]:
    """
    解析 S3 URL 以提取桶名和对象前缀（路径）。

    Args:
        s3_url (str): 完整的 S3 URL，例如 's3://your-bucket-name/path/to/folder/'

    Returns:
        tuple[str, str]: 包含桶名和前缀的元组。

    Raises:
        ValueError: 如果 S3 URL 格式不正确。
    """
    parsed_url = urlparse(s3_url)
    if parsed_url.scheme != 's3':
        raise ValueError("无效的 S3 URL 方案。URL 必须以 's3://' 开头。")

    bucket_name = parsed_url.netloc
    # path.lstrip('/') 移除路径开头的斜杠
    prefix = parsed_url.path.lstrip('/')

    # 确保前缀以斜杠结尾，如果它表示一个文件夹
    if prefix and not prefix.endswith('/'):
        prefix += '/'

    return bucket_name, prefix

def count_specific_files_in_s3(
    bucket_name: str, 
    s3_prefix: str, 
    file_regex_pattern: str = r'^file_\d+\.ts$'
) -> int:
    """
    统计 S3 存储桶中指定前缀下符合正则表达式模式的文件数量。

    Args:
        bucket_name (str): S3 存储桶的名称。
        s3_prefix (str): S3 前缀（文件夹路径），例如 'path/to/folder/'。
        file_regex_pattern (str): 用于匹配对象键中“文件名”部分的正则表达式模式。
                                  默认匹配 'file_NNN.ts' 格式的文件。

    Returns:
        int: 匹配文件的数量。如果发生错误，返回 0。
    """
    s3 = boto3.resource('s3')
    bucket = s3.Bucket(bucket_name)

    count = 0
    # 编译正则表达式以提高性能
    file_pattern = re.compile(file_regex_pattern)

    try:
        # 使用 filter 方法和 S3 前缀来获取指定路径下的所有对象
        # boto3.resource 会自动处理分页
        for obj in bucket.objects.filter(Prefix=s3_prefix):
            # 提取对象键（完整路径）中的文件名部分
            filename = obj.key.split('/')[-1]

            # 检查文件名是否与期望的模式匹配
            if file_pattern.match(filename):
                count += 1
    except Exception as e:
        print(f"错误：在 s3://{bucket_name}/{s3_prefix} 统计文件时发生异常: {e}")
        return 0 # 发生错误时返回 0，或根据需要抛出异常

    return count

在上述代码中：

• get_bucket_and_prefix_from_s3_url 函数负责将一个完整的 S3 URL 分解为桶名和在桶内的对象前缀。这是关键一步，因为 boto3 的 Prefix 参数期望的是桶内的路径，而不是完整的 S3 URL。
• count_specific_files_in_s3 函数接收桶名和前缀，并使用 boto3.resource('s3').Bucket(bucket_name).objects.filter(Prefix=s3_prefix) 来高效地遍历所有匹配的对象。
• 我们使用 re.compile 预编译正则表达式，并通过 file_pattern.match(filename) 来精确匹配文件名，确保只统计符合 file_NNN.ts 模式的文件。

完整示例：结合 CSV 处理批量统计

在实际应用中，您可能需要从一个 CSV 文件中读取多条 S3 路径，然后对每条路径进行文件统计，并将结果写入另一个 CSV 文件。以下是一个结合上述函数的完整示例：

import csv
import boto3
import re
from urllib.parse import urlparse

# (此处省略上面定义的 get_bucket_and_prefix_from_s3_url 和 count_specific_files_in_s3 函数)
# 请确保将上述两个函数定义复制到此处或导入

# 定义 S3 URL 的基础部分，如果您的 CSV 中只包含相对路径
base_s3_url = 's3://coursevideotesting/' 

# 输入和输出 CSV 文件名
input_csv_file = 'ldt_ffw_course_videos_temp.csv'
output_csv_file = 'file_count_result.csv'

# 定义要匹配的文件正则表达式模式
# 例如，匹配 'file_000.ts', 'file_001.ts' 等
TARGET_FILE_PATTERN = r'^file_\d+\.ts$' 

# 读取输入 CSV 并检查文件数量
with open(input_csv_file, mode='r', encoding='utf-8') as infile, \
     open(output_csv_file, mode='w', newline='', encoding='utf-8') as outfile:

    reader = csv.DictReader(infile)
    # 确保 CSV 字段名与您的输入文件匹配
    # 例如，如果您的输入 CSV 有 'course_video_s3_url' 和 'course_video_ts_file_cnt' 列
    fieldnames = ['URL', 'Actual Files', 'Expected Files', 'Status'] # 添加一个状态列
    writer = csv.DictWriter(outfile, fieldnames=fieldnames)
    writer.writeheader()

    for row in reader:
        # 构建完整的 S3 URL
        # 假设 input_csv_file 中的 'course_video_s3_url' 包含相对路径，如 'Financial_Freedom_Course_Kannada/...'
        full_s3_url = base_s3_url + row['course_video_s3_url']

        expected_files = int(row['course_video_ts_file_cnt'])
        actual_files = 0
        status = "Success"

        try:
            # 解析 S3 URL 获取桶名和前缀
            bucket_name, s3_prefix = get_bucket_and_prefix_from_s3_url(full_s3_url)

            # 调用函数统计文件
            actual_files = count_specific_files_in_s3(bucket_name, s3_prefix, TARGET_FILE_PATTERN)

            if actual_files != expected_files:
                status = "Mismatch"
            if actual_files == 0 and expected_files > 0:
                status = "Missing" # 预期有文件但实际没有

        except ValueError as ve:
            print(f"处理 URL 格式错误: {full_s3_url} - {ve}")
            status = f"URL Error: {ve}"
        except Exception as e:
            print(f"处理 URL 时发生未知错误: {full_s3_url} - {e}")
            status = f"Process Error: {e}"

        writer.writerow({
            'URL': full_s3_url, 
            'Actual Files': actual_files, 
            'Expected Files': expected_files,
            'Status': status
        })

print(f"文件统计结果已写入到 {output_csv_file}")

输入 CSV 文件 ldt_ffw_course_videos_temp.csv 示例：

course_video_s3_url,course_video_ts_file_cnt
Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/144p/,28
Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/360p/,34
Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/480p/,54
Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/720p/,57

预期输出 file_count_result.csv 示例：

URL,Actual Files,Expected Files,Status
s3://coursevideotesting/Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/144p/,28,28,Success
s3://coursevideotesting/Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/360p/,34,34,Success
s3://coursevideotesting/Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/480p/,52,54,Mismatch
s3://coursevideotesting/Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/720p/,57,57,Success

注意事项

1. 权限管理: 确保您的 AWS 凭证（通过环境变量、IAM 角色或 ~/.aws/credentials 文件配置）具有访问目标 S3 桶和执行 s3:ListBucket 操作的权限。
2. 前缀匹配: Prefix 参数是前缀匹配，它会返回所有以该字符串开头的对象。如果您的前缀是 myfolder/，它将匹配 myfolder/file.txt 和 myfolder/subfolder/another.txt。因此，在遍历结果时，仍需通过 obj.key.split('/')[-1] 提取文件名并进行精确匹配。
3. 性能优化: 对于包含数百万甚至数十亿对象的超大型桶，即使 boto3.resource 自动处理分页，遍历所有对象仍可能非常耗时。在这种情况下，可以考虑：
   • S3 Inventory: S3 Inventory 服务可以定期生成桶中所有对象的列表，并将其存储在另一个 S3 桶中，这比实时 API 调用更高效。
   • S3 Select: 如果您只需要查询对象内容的特定部分，S3 Select 允许您使用 SQL 表达式直接查询 S3 对象数据。
   • MaxKeys 参数: 在 client 接口中，可以通过 MaxKeys 限制每次 API 调用的返回对象数量，但这需要您手动管理分页。
4. 错误处理: 在实际应用中，务必添加健壮的错误处理机制，例如 try-except 块来捕获网络问题、权限不足