Python中无法从Google Cloud Storage存储桶目录加载新图片

问题现象 #

此前通过以下代码统计Google Cloud Storage(GCS)存储桶中图片Blob数量时，输出的数组尺寸正确（文件数量×height×width）：

# connect to storage
storage_client = storage.Client(project_id)
bucket = storage_client.get_bucket(bucket_name)
blobs = bucket.list_blobs(prefix=img_dir)  # Get list of files
images = np.zeros((len(list(blobs)), height, width))

print(images.shape)

但遍历Blob尝试加载为PIL图片时，打印的filename仅显示目录（如/directory/），且触发UnidentifiedImageError：

# connect to storage
storage_client = storage.Client(project_id)
bucket = storage_client.get_bucket(bucket_name)
blobs = bucket.list_blobs(prefix=img_dir)  # Get list of files
filenames = []
i = 0
# show some images
for blob in blobs:

    filename = blob.name
    print(filename)
    img = Image.open(BytesIO(blob.download_as_bytes()))

报错信息：

---> 13 img = Image.open(BytesIO(blob.download_as_bytes()))

File ~/miniconda/envs/myenv_local/lib/python3.8/site-packages/PIL/Image.py:3186, in open(fp, mode, formats)
   3184 for message in accept_warnings:
   3185     warnings.warn(message)
-> 3186 raise UnidentifiedImageError(
   3187     "cannot identify image file %r" % (filename if filename else fp)
   3188 )

UnidentifiedImageError: cannot identify image file <_io.BytesIO object at 0x7fa9ca4b8860>

该问题仅出现在近期上传的新文件中，旧文件可正常加载。

可能的原因 #

• 目录占位Blob干扰：GCS本身没有真实目录结构，若新文件上传时通过控制台/工具创建了带/后缀的目录占位符（大小为0的Blob），list_blobs(prefix=img_dir)会将这个占位Blob也纳入结果，遍历到它时尝试加载自然会报错。旧文件上传流程可能未生成这类占位Blob。
• 新文件路径格式错误：新文件上传时的路径格式与旧文件不一致，比如路径开头多了/，或者未正确嵌套到目标目录下，导致prefix匹配到的是目录而非实际文件。
• 新文件上传不完整/损坏：上传过程中断或失败，导致Blob内容为空、不完整，PIL无法识别为有效图片。
• Content-Type标记错误：新文件上传时未正确设置Content-Type（如image/jpeg），或标记为非图片类型，可能影响PIL的识别逻辑。

调试与解决步骤 #

• 跳过目录占位Blob：遍历Blob时先过滤掉以/结尾或大小为0的占位Blob：

  for blob in blobs:
      # 跳过目录占位符和空Blob
      if blob.name.endswith('/') or blob.size == 0:
          continue
      filename = blob.name
      print(filename)
      img = Image.open(BytesIO(blob.download_as_bytes()))
  

• 打印Blob详细信息排查：输出每个Blob的名称、大小、内容类型，确认新文件是否存在且属性正常：

  for blob in blobs:
      print(f"Blob名称: {blob.name}, 大小: {blob.size}字节, 内容类型: {blob.content_type}")
  

• 单独测试新文件：直接指定单个新文件的完整路径，验证是否能正常下载加载：

  test_blob = bucket.get_blob('img_dir/新文件名.jpg')
  if test_blob:
      img = Image.open(BytesIO(test_blob.download_as_bytes()))
      img.show()
  else:
      print("指定Blob不存在")
  

• 对比新旧文件上传流程：检查新文件的上传工具/代码是否与旧文件一致，是否存在创建目录占位符、路径格式错误等差异。
• 验证本地文件完整性：将新文件下载到本地，用图片查看器打开，确认文件本身未损坏。

内容的提问来源于stack exchange，提问作者edgarbc