引言：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 方法时，可能会遇到以下挑战：
- 手动分页处理：list_objects_v2 每次请求最多返回 1000 个对象。如果 S3 路径下的文件数量超过此限制，需要通过检查响应中的 IsTruncated 字段和使用 NextContinuationToken 参数来手动迭代所有页面，才能获取完整的对象列表。若未正确处理分页，将导致计数不完整。
- 不精确的过滤：list_objects_v2 的 Prefix 参数用于服务器端初步过滤，但它只能匹配对象键的开头。如果需要根据对象键的中间或结尾部分进行更复杂的匹配（例如，匹配所有以 file_ 开头且以 .ts 结尾的文件），则需要在客户端（即 Python 脚本中）对返回的对象列表进行二次过滤。原始尝试中 obj['Key'].startswith('file_000.ts') 过于具体，无法统计所有 file_*.ts 文件。
- S3 路径解析：S3 中的“文件夹”实际上是对象键的一部分。例如，s3://bucket/folder/subfolder/file.txt 中，folder/subfolder/ 是 file.txt 对象键的一部分。在构建 Prefix 时，需要确保其与 S3 对象的实际键结构匹配。
推荐方案：使用 boto3.resource 进行高效计数
boto3.resource 接口提供了更高级别的抽象，相较于 boto3.client，它更面向对象，并能自动处理许多底层细节，包括分页。这使得代码更简洁、更易读。
立即学习“Python免费学习笔记（深入）”；
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)}")
# 
