本教程详细介绍了如何利用 python 的 Boto3 库，高效地统计 AWS S3 存储桶中符合特定命名模式（例如 file_*.ts）的文件数量。文章将着重阐述 boto3.Resource 的优势，包括其自动处理分页的能力，并提供清晰的代码示例，以实现对指定虚拟文件夹及其子文件夹内文件的精确计数。
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
© 版权声明
文章版权归作者所有，未经允许请勿转载。
THE END
