1. 问题背景与挑战
在处理 AWS S3 中的大量数据时，经常需要统计特定类型的文件。例如，在一个存储了视频切片（chunks）的S3桶中，视频文件可能按质量（如144p/、360p/）和切片序号（如file_000.ts、file_001.ts等）组织。我们的目标是：
- 针对给定的S3路径（前缀），统计其中符合特定命名模式（例如，文件名以file_开头并以.ts结尾）的文件数量。
- 能够遍历该路径下的所有子文件夹，确保所有层级的文件都能被统计到。
- 高效处理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 的自动分页能力使其成为更简洁、更易于维护的选择。
立即学习“Python免费学习笔记（深入）”；
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
