Python+Boto3快速统计AWSS3文件

本文档详细阐述了如何利用Python和Boto3库高效地统计AWS S3存储桶中符合特定命名模式的文件数量，以应对S3文件计数挑战。针对传统boto3.client方法在处理大量文件时面临的手动分页和不精确过滤问题，文章着重介绍了boto3.resource的优势，包括自动分页和更Pythonic的接口。通过提供结合前缀过滤与客户端精确匹配的完整代码示例，展示了如何使用boto3.resource进行高效计数，并深入解析了Prefix和Key过滤机制。此外，文章还提供了一个整合到实际应用场景的示例，演示了如何批量校验S3视频切片，并结合CSV文件实现自动化文件校验与管理，助力用户提升S3文件管理的效率和准确性。

引言：S3 文件计数挑战

在 AWS S3（Simple Storage Service）中管理大量文件时，经常需要统计特定目录或符合特定命名规则的文件数量。例如，对于视频切片存储场景，可能需要验证每个视频质量版本（如 144p, 360p）下是否存在预期数量的切片文件（如 file_000.ts, file_001.ts 等）。手动检查这些文件既耗时又容易出错，因此编写自动化脚本变得尤为重要。

Python 的 boto3 库是与 AWS 服务交互的官方 SDK，提供了强大的功能来操作 S3。然而，在使用 boto3 进行文件计数时，尤其是在文件数量庞大且分散在多个“文件夹”中时，需要注意一些关键点，如分页处理和精确的文件名匹配。

传统 boto3.client 方法的局限性

在最初尝试使用 boto3.client().list_objects_v2 方法时，可能会遇到以下挑战：

1. 手动分页处理：list_objects_v2 每次请求最多返回 1000 个对象。如果 S3 路径下的文件数量超过此限制，需要通过检查响应中的 IsTruncated 字段和使用 NextContinuationToken 参数来手动迭代所有页面，才能获取完整的对象列表。若未正确处理分页，将导致计数不完整。
2. 不精确的过滤：list_objects_v2 的 Prefix 参数用于服务器端初步过滤，但它只能匹配对象键的开头。如果需要根据对象键的中间或结尾部分进行更复杂的匹配（例如，匹配所有以 file_ 开头且以 .ts 结尾的文件），则需要在客户端（即 Python 脚本中）对返回的对象列表进行二次过滤。原始尝试中 obj['Key'].startswith('file_000.ts') 过于具体，无法统计所有 file_*.ts 文件。
3. S3 路径解析：S3 中的“文件夹”实际上是对象键的一部分。例如，s3://bucket/folder/subfolder/file.txt 中，folder/subfolder/ 是 file.txt 对象键的一部分。在构建 Prefix 时，需要确保其与 S3 对象的实际键结构匹配。

推荐方案：使用 boto3.resource 进行高效计数

boto3.resource 接口提供了更高级别的抽象，相较于 boto3.client，它更面向对象，并能自动处理许多底层细节，包括分页。这使得代码更简洁、更易读。

1. boto3.resource 的优势

• 自动分页：resource 对象（如 bucket.objects.filter()）在迭代时会自动处理分页，无需手动管理 ContinuationToken。
• 更 Pythonic 的接口：提供了更直观的方法和属性来访问 S3 桶和对象。

2. 基本计数示例

以下是一个使用 boto3.resource 统计 S3 桶中特定前缀下，符合 file_*.ts 模式文件数量的基本函数：

import boto3

def count_specific_files_in_s3(bucket_name, s3_prefix, file_name_starts_with='file_', file_name_ends_with='.ts'):
    """
    统计S3桶中指定前缀下，符合特定命名模式的文件数量。

    Args:
        bucket_name (str): S3 桶的名称。
        s3_prefix (str): 要搜索的S3前缀（即“文件夹”路径）。
                         例如：'Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/144p/'
        file_name_starts_with (str): 文件名应以此字符串开头。
        file_name_ends_with (str): 文件名应以此字符串结尾。

    Returns:
        int: 符合条件的文件数量。
    """
    s3 = boto3.resource('s3')
    bucket = s3.Bucket(bucket_name)
    actual_count = 0

    print(f"正在检查 S3 路径: s3://{bucket_name}/{s3_prefix}...")

    # 使用 Prefix 进行服务器端初步过滤
    # bucket.objects.filter() 会自动处理分页
    for obj in bucket.objects.filter(Prefix=s3_prefix):
        # 进一步在客户端过滤，确保符合命名约定且不是S3模拟的文件夹对象
        # S3中没有真正的文件夹，以'/'结尾的键通常被视为文件夹
        if obj.key.startswith(f'{s3_prefix}{file_name_starts_with}') and \
           obj.key.endswith(file_name_ends_with) and \
           not obj.key.endswith('/'): # 排除S3模拟的文件夹对象
            actual_count += 1
            # print(f"  找到匹配文件: {obj.key}") # 可用于调试
    return actual_count

# 示例用法 (请替换为您的实际桶名和前缀)
# fixed_bucket_name = 'your-s3-bucket-name'
# example_prefix = 'your-folder-path/sub-folder/'
# count = count_specific_files_in_s3(fixed_bucket_name, example_prefix)
# print(f"在 s3://{fixed_bucket_name}/{example_prefix} 路径下找到 {count} 个文件。")

3. 深入理解 Prefix 和 Key 过滤

• Prefix 参数: 这是 S3 服务端提供的过滤机制。它能有效地减少从 S3 传输到本地的数据量，提高效率。当您指定 Prefix='path/to/folder/' 时，S3 只会返回键以 path/to/folder/ 开头的对象。
• obj.key 客户端过滤: 尽管 Prefix 很有用，但它可能不足以满足所有复杂的过滤需求。例如，您可能需要匹配文件名的特定模式，或者排除某些类型的文件。这时，您需要在 Python 代码中对 obj.key 属性进行字符串操作（如 startswith(), endswith(), in 或正则表达式）来精确匹配所需的文件。在上述示例中，我们结合了 startswith(f'{s3_prefix}{file_name_starts_with}') 和 endswith(file_name_ends_with) 来实现精确匹配，并排除了以 / 结尾的“文件夹”对象。

整合到实际应用场景：批量校验 S3 视频切片

结合原始问题中从 CSV 读取 URL 并写入结果的需求，我们可以构建一个完整的脚本来自动化这一过程。

假设 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/,54
Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/720p/,34

其中 course_video_s3_url 是相对于 S3 桶根目录的路径，course_video_ts_file_cnt 是该路径下期望的文件数量。

import csv
import boto3
from urllib.parse import urlparse # 用于解析S3 URL，如果需要

# 辅助函数：解析 S3 URL 获取桶名和前缀 (如果输入CSV中是完整S3 URL)
def parse_s3_url(s3_url):
    """从完整的S3 URL中解析出桶名和前缀路径。"""
    parsed = urlparse(s3_url)
    if parsed.scheme != 's3':
        raise ValueError("URL 必须以 's3://' 开头。")
    bucket_name = parsed.netloc
    # path 包含前导斜杠，需要去除
    prefix = parsed.path.lstrip('/')
    return bucket_name, prefix

# 改进的计数函数 (与上面示例相同)
def count_specific_files_in_s3(bucket_name, s3_prefix, file_name_starts_with='file_', file_name_ends_with='.ts'):
    s3 = boto3.resource('s3')
    bucket = s3.Bucket(bucket_name)
    actual_count = 0

    for obj in bucket.objects.filter(Prefix=s3_prefix):
        if obj.key.startswith(f'{s3_prefix}{file_name_starts_with}') and \
           obj.key.endswith(file_name_ends_with) and \
           not obj.key.endswith('/'):
            actual_count += 1
    return actual_count

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

# 假设 S3 桶名是固定的。请替换为您的实际桶名。
# 如果您的CSV中包含完整的S3 URL (如 s3://bucket-name/path/...), 则可以使用 parse_s3_url 函数动态获取桶名。
fixed_bucket_name = 'coursevideotesting' # 替换为你的S3桶名

# 主处理逻辑
try:
    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文件是否包含必要的列
        required_columns = ['course_video_s3_url', 'course_video_ts_file_cnt']
        if not all(col in reader.fieldnames for col in required_columns):
            raise ValueError(f"输入CSV文件必须包含以下列: {', '.join(required_columns)}")

        #

以上就是《Python+Boto3快速统计AWSS3文件》的详细内容，更多关于的资料请关注golang学习网公众号！