基于CSV的图片批量下载重命名及GCP Bucket自动上传技术问询

我来帮你拆解这个需求，分成读取CSV数据、批量下载图片、上传至GCP Bucket、自动化流程四个核心部分，每个部分都给你具体的代码实现和注意事项。

第一步：准备依赖 #

首先安装需要的Python库：

pip install requests google-cloud-storage

第二步：核心功能实现 #

1. 读取CSV中的图片ID和URL #

用Python内置的csv模块读取数据，确保只提取有效行，不依赖列的顺序：

import csv

def read_image_csv(csv_file_path):
    image_records = []
    with open(csv_file_path, mode='r', newline='', encoding='utf-8') as csv_file:
        reader = csv.DictReader(csv_file)
        for row in reader:
            image_id = row.get('image_id')
            image_url = row.get('image_url')
            # 跳过缺少ID或URL的无效行
            if image_id and image_url:
                image_records.append((image_id, image_url))
    return image_records

2. 从URL下载图片并按ID命名 #

处理下载异常、跳过重复文件，自动从URL提取文件扩展名：

import requests
import os
from urllib.parse import urlparse

def download_image(image_url, temp_dir, image_id):
    # 创建临时存储目录（不存在则自动创建）
    os.makedirs(temp_dir, exist_ok=True)
    
    # 从URL提取扩展名，无扩展名则默认用.jpg
    parsed_url = urlparse(image_url)
    file_ext = os.path.splitext(parsed_url.path)[1] or '.jpg'
    file_name = f"{image_id}{file_ext}"
    save_path = os.path.join(temp_dir, file_name)
    
    # 跳过已存在的文件，避免重复下载浪费资源
    if os.path.exists(save_path):
        print(f"✅ 已存在，跳过：{file_name}")
        return save_path
    
    try:
        # 设置超时时间防止请求卡住
        response = requests.get(image_url, timeout=15)
        response.raise_for_status()  # 主动抛出HTTP错误（如404、500）
        
        with open(save_path, 'wb') as img_file:
            img_file.write(response.content)
        print(f"✅ 下载成功：{file_name}")
        return save_path
    except Exception as e:
        print(f"❌ 下载失败 {image_url}：{str(e)}")
        return None

3. 上传图片至GCP Bucket #

先配置GCP认证：

• 从GCP控制台下载服务账号密钥文件（JSON格式）
• 设置环境变量指向密钥文件：

  # Linux/macOS
  export GOOGLE_APPLICATION_CREDENTIALS="/path/to/your/service-account-key.json"
  
  # Windows
  set GOOGLE_APPLICATION_CREDENTIALS="C:\path\to\your\service-account-key.json"
  

实现上传函数：

from google.cloud import storage

def upload_to_gcp(local_file_path, bucket_name, blob_path=None):
    if not local_file_path or not os.path.exists(local_file_path):
        print(f"❌ 本地文件不存在：{local_file_path}")
        return False
    
    try:
        storage_client = storage.Client()
        bucket = storage_client.bucket(bucket_name)
        # 默认用本地文件名作为Bucket中的路径，也可以自定义前缀（比如加个images目录）
        blob_name = blob_path or os.path.basename(local_file_path)
        blob = bucket.blob(blob_name)
        
        # 自动检测文件类型，也可以手动指定（如content_type='image/jpeg'）
        blob.upload_from_filename(local_file_path)
        print(f"✅ 上传成功至Bucket：{blob_name}")
        return True
    except Exception as e:
        print(f"❌ 上传失败 {local_file_path}：{str(e)}")
        return False

4. 整合完整流程 #

把上面的函数串起来，实现从CSV到GCP的全流程：

def run_pipeline(csv_path, temp_dir, gcp_bucket_name):
    # 读取CSV数据
    image_records = read_image_csv(csv_path)
    if not image_records:
        print("⚠️ CSV中没有有效数据")
        return
    
    for image_id, image_url in image_records:
        # 下载图片
        local_file = download_image(image_url, temp_dir, image_id)
        if not local_file:
            continue
        
        # 上传到GCP（自定义Bucket中的存储路径，比如加个images前缀）
        gcp_blob_path = f"images/{os.path.basename(local_file)}"
        upload_success = upload_to_gcp(local_file, gcp_bucket_name, gcp_blob_path)
        
        # 可选：上传成功后删除本地临时文件，节省磁盘空间
        if upload_success:
            os.remove(local_file)
            print(f"🗑️ 删除本地临时文件：{local_file}")

if __name__ == "__main__":
    # 替换成你的实际参数
    CONFIG = {
        "csv_path": "your_images.csv",
        "temp_dir": "temp_image_storage",
        "gcp_bucket": "your-gcp-bucket-name"
    }
    run_pipeline(**CONFIG)

第三步：自动化流程实现 #

如果需要自动触发这个流程，推荐两种方案：

方案1：GCP Cloud Functions（事件触发） #

当CSV文件上传到某个GCP Bucket时，自动执行下载+上传：

1. 在GCP控制台创建Cloud Function，选择Cloud Storage作为触发器，事件类型为最终创建/更新对象
2. 修改代码，读取Bucket中的CSV文件（用storage.Client获取文件内容），替代本地文件读取逻辑
3. 部署函数，后续只要CSV上传到指定Bucket，就会自动执行全流程

方案2：定时任务（Cron） #

如果需要定期处理CSV（比如每天凌晨）：

• 在Linux/macOS上编辑Cron任务：

  crontab -e
  

  添加一行定时执行命令：

  0 0 * * * /usr/bin/python3 /path/to/your/script.py >> /path/to/logs/image_upload.log 2>&1
  

• 在Windows上用任务计划程序创建定时任务，执行Python脚本

注意事项 #

• 权限控制：确保GCP服务账号拥有storage.objectCreator（上传文件）和storage.objectViewer（如果读取Bucket中的CSV）权限
• 重试机制：如果网络不稳定，可以用tenacity库添加重试逻辑，避免单次失败导致流程中断
• 批量优化：如果CSV有上万条数据，建议分批次处理，避免内存占用过高
• 日志记录：可以用Python内置的logging模块替代print，方便后续排查问题

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