使用Python和Boto3高效统计AWS S3存储桶中特定文件数量

import boto3  def count_specific_files_in_s3(bucket_name, prefix, file_pattern_start='file_', file_pattern_end='.ts'):     """     统计S3桶中指定前缀下，符合特定命名模式的文件数量。      Args:         bucket_name (str): S3桶的名称。         prefix (str): 要搜索的前缀路径（例如 'path/to/folder/'）。         file_pattern_start (str): 文件名必须以此字符串开头（例如 'file_'）。         file_pattern_end (str): 文件名必须以此字符串结尾（例如 '.ts'）。      Returns:         int: 符合条件的文件数量。     """     s3_resource = boto3.resource('s3')     bucket = s3_resource.Bucket(bucket_name)      count = 0     print(f"Checking folder: s3://{bucket_name}/{prefix}")      try:         # 使用filter(Prefix=prefix)获取指定路径下的所有对象         # boto3 resource会自动处理分页         for obj in bucket.objects.filter(Prefix=prefix):             # object.key 是对象的完整键，例如 'path/to/folder/file_001.ts'             # 确保我们只计算当前“目录”下的文件，而不是子目录中的文件             # 并且匹配特定的文件名模式              # 假设prefix是 'path/to/folder/'             # 那么文件键应该是 'path/to/folder/file_XXX.ts'             # 我们需要检查文件键是否以完整的prefix开头，并且在prefix之后的部分符合我们的文件名模式              # 获取对象键中，去除prefix后的相对路径/文件名             relative_key = obj.key[len(prefix):] if obj.key.startswith(prefix) else obj.key              # 排除自身作为目录的情况 (例如 prefix是'foo/'，obj.key是'foo/')             # 排除子目录 (例如 prefix是'foo/'，obj.key是'foo/bar/file.ts')             # 只统计当前层级的文件             if '/' not in relative_key and                 relative_key.startswith(file_pattern_start) and                 relative_key.endswith(file_pattern_end):                 count += 1                 # print(f"Matched file: {obj.key}") # 调试用         print(f"Actual chunks found: {count}")         return count     except Exception as e:         print(f"Error Accessing S3 bucket {bucket_name} with prefix {prefix}: {e}")         return 0 # 返回0表示发生错误或无法访问  # 示例用法 (假设你的AWS凭证已配置) # bucket_name = 'coursevideotesting' # prefix = 'Financial_Freedom_Course_Kannada/00_Course_Trailer_New_update/360p/' # actual_count = count_specific_files_in_s3(bucket_name, prefix) # print(f"Count for {prefix}: {actual_count}")

import csv import boto3 import re  # 1. 解析S3 URL的辅助函数 def parse_s3_url(s3_url):     match = re.match(r's3://([^/]+)/(.*)', s3_url)     if match:         bucket_name = match.group(1)         prefix = match.group(2)         # 确保前缀以斜杠结尾，如果它代表一个目录         if prefix and not prefix.endswith('/'):             prefix += '/'         return bucket_name, prefix     else:         raise ValueError(f"Invalid S3 URL format: {s3_url}")  # 2. 统计S3中特定文件的函数 def count_specific_files_in_s3(bucket_name, prefix, file_pattern_start='file_', file_pattern_end='.ts'):     s3_resource = boto3.resource('s3')     bucket = s3_resource.Bucket(bucket_name)      count = 0     print(f"Checking folder: s3://{bucket_name}/{prefix}")      try:         for obj in bucket.objects.filter(Prefix=prefix):             # object.key 是完整的对象键，例如 'path/to/folder/file_001.ts'             # 我们需要确保只统计当前层级的文件，而不是子目录中的文件             # 例如，如果prefix是'videos/qualityA/'             # obj.key可能是 'videos/qualityA/file_001.ts' (要统计)             # 或 'videos/qualityA/subfolder/file_001.ts' (不统计)              # 提取对象键中，去除prefix后的相对路径/文件名             # 如果obj.key不以prefix开头，则跳过（理论上不会发生，因为filter已经处理了）             if not obj.key.startswith(prefix):                 continue              relative_key = obj.key[len(prefix):]              # 检查relative_key是否包含斜杠，如果有，说明是子目录或子目录中的文件，跳过             # 并且检查文件名是否符合模式             if '/' not in relative_key and                 relative_key.startswith(file_pattern_start) and                 relative_key.endswith(file_pattern_end):                 count += 1         print(f"Actual chunks found: {count}")         return count     except Exception as e:         print(f"Error accessing S3 bucket {bucket_name} with prefix {prefix}: {e}")         return 0 # 发生错误时返回0  # --- 主脚本逻辑 --- # 输入和输出csv文件名称 input_csv_file = 'ldt_ffw_course_videos_temp.csv'  # 替换为你的输入CSV文件 output_csv_file = 'file_count_result.csv'          # 替换为你的输出CSV文件  # 确保输入CSV文件存在并包含正确的列名 # 假设输入CSV包含 'course_video_s3_url' 和 'course_video_ts_file_cnt' 列  # 读取URLs从输入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列名是否存在     if 'course_video_s3_url' not in reader.fieldnames or         'course_video_ts_file_cnt' not in reader.fieldnames:         raise ValueError(f"Input CSV must contain 'course_video_s3_url' and 'course_video_ts_file_cnt' columns. Found: {reader.fieldnames}")      fieldnames = ['URL', 'Actual Files', 'Expected Files']     writer = csv.DictWriter(outfile, fieldnames=fieldnames)     writer.writeheader()      for row in reader:         s3_url_from_csv = row['course_video_s3_url']         expected_files = int(row['course_video_ts_file_cnt'])          try:             # 解析S3 URL             bucket_name, s3_prefix = parse_s3_url(s3_url_from_csv)              # 调用函数统计文件             actual_files = count_specific_files_in_s3(bucket_name, s3_prefix)              writer.writerow({                 'URL': s3_url_from_csv,                 'Actual Files': actual_files,                 'Expected Files': expected_files             })         except ValueError as ve:             print(f"Skipping row due to invalid S3 URL: {s3_url_from_csv} - {ve}")             writer.writerow({                 'URL': s3_url_from_csv,                 'Actual Files': 'Error: Invalid URL',                 'Expected Files': expected_files             })         except Exception as e:             print(f"An unexpected error occurred for URL {s3_url_from_csv}: {e}")             writer.writerow({                 'URL': s3_url_from_csv,                 'Actual Files': 'Error: ' + str(e),                 'Expected Files': expected_files             })  print(f"File counting complete. Results written to {output_csv_file}")