从Azure Blob Storage下载文件的最佳实践及Python代码问题咨询

一、避免连接字符串硬编码的方案 #

• 使用环境变量/应用设置：Azure Functions V2 中直接在门户的应用设置里配置AZURE_CONNECTION_STRING，本地开发用local.settings.json存储，代码通过os.getenv读取，彻底避免硬编码到代码库。
• 使用托管标识（Managed Identity）：给 Azure Functions 分配系统或用户托管标识，为其赋予 Blob Storage 的对应权限（如Storage Blob Data Contributor），代码用DefaultAzureCredential认证，无需连接字符串：

  from azure.identity import DefaultAzureCredential
  from azure.storage.blob import ContainerClient
  
  credential = DefaultAzureCredential()
  container_client = ContainerClient(
      account_url="https://<你的存储账户名>.blob.core.windows.net",
      container_name=AZURE_CONTAINER_NAME,
      credential=credential
  )
  

二、本地存储 vs 内存存储的选择 #

• 内存存储：适合几MB以内的小文件（如小型PNG、短音频），无需磁盘IO，处理速度更快，适合临时处理后直接销毁的场景。注意控制内存占用，大文件易引发内存溢出。
• 本地存储：适合大文件，或后续处理工具（如moviepy）要求读取本地文件路径的场景。可使用临时文件（如代码中的tempfile），但务必在处理完成后清理文件，避免磁盘空间浪费。

三、Blob Storage Bindings 能否在普通Python代码中使用？ #

Blob Storage Bindings 是 Azure Functions 专属功能，仅能在 Functions 的触发器/绑定上下文内使用，非 Functions 环境的普通 Python 代码无法直接调用。普通代码需通过 Azure Storage SDK for Python（azure-storage-blob）操作 Blob。

四、PNG与MP3合并时的报错解决 #

1. AttributeError: 'PngImageFile' object has no attribute 'shape' #

该报错是误将 PIL 的PngImageFile对象当作 numpy 数组使用导致的，解决方式：

• 若仅需获取图片尺寸，直接使用 PIL 自带属性：image.size（返回(宽度, 高度)）
• 若确实需要 numpy 数组，将 PIL 对象转换为数组：

  from PIL import Image
  import numpy as np
  
  img = Image.open(image_temp_file_path)
  img_array = np.array(img)
  # 此时可调用 img_array.shape 获取维度
  

2. UnicodeDecodeError: 'utf-8' codec can't decode byte 0x89 in position 0 (moviepy相关) #

此报错是因为 moviepy 尝试将二进制的图片/音频文件当作文本读取，解决方式：

• 确保传给 moviepy 的是本地文件路径，而非内存字节流或错误的文件对象
• 若从内存处理数据，需先写入临时文件，再将文件路径传给 moviepy 方法（如ImageClip支持文件路径或 PIL 对象，不支持直接传字节流）
• 检查代码中是否误用open(..., 'r')打开二进制文件，必须使用'rb'模式

五、代码优化建议 #

针对现有代码，可做以下优化：

• 复用ContainerClient：无需重复创建BlobClient，直接调用container_client.get_blob_client(blob.name)减少冗余代码
• 自动清理临时文件：处理完成后删除临时文件，避免磁盘残留，可通过finally块实现
• 增加异常捕获：避免单个文件处理失败导致整个流程中断

优化后的代码片段：

def merge_image_audio():
    """Merge PNG with mp3 files."""
    from azure.storage.blob import ContainerClient
    import tempfile
    import os

    # 通过环境变量或托管标识获取容器客户端
    azure_connection_string = os.getenv("AZURE_CONNECTION_STRING")
    container_client = ContainerClient.from_connection_string(
        conn_str=azure_connection_string,
        container_name=AZURE_CONTAINER_NAME
    )

    blob_list = container_client.list_blobs(name_starts_with="temp/")
    image_paths = []
    audio_paths = []

    try:
        for blob in blob_list:
            blob_client = container_client.get_blob_client(blob)
            blob_data = blob_client.download_blob()

            if blob.name.endswith('.png'):
                with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as temp_file:
                    temp_file.write(blob_data.readall())
                    image_paths.append(temp_file.name)
            elif blob.name.endswith(".mp3"):
                with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as temp_file:
                    temp_file.write(blob_data.readall())
                    audio_paths.append(temp_file.name)
        
        # 此处添加合并逻辑（示例）
        # from moviepy.editor import ImageClip, AudioFileClip
        # for img_path, audio_path in zip(image_paths, audio_paths):
        #     img_clip = ImageClip(img_path).set_duration(AudioFileClip(audio_path).duration)
        #     final_clip = img_clip.set_audio(AudioFileClip(audio_path))
        #     final_clip.write_videofile(f"output_{os.path.basename(img_path)}.mp4")
    finally:
        # 清理临时文件
        for path in image_paths + audio_paths:
            if os.path.exists(path):
                os.unlink(path)

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