批量生成电商产品图的自动化脚本，核心不在提示词本身，而是把输入、输出、失败路径先固定下来。Seedream 5.0 Pro 这类图像生成模型通常提供 HTTP 接口，脚本只需要做四件事：读取商品描述、构造请求、保存返回图片、记录失败行。这里直接给出可运行的骨架，接口地址和鉴权字段需要按实际服务替换。

适用于需要一次性生成几十到几百张产品图的运营或开发场景。具体做法：先把商品描述整理成 CSV，每个商品一行；再封装单次请求函数，确认能拿到图片；最后加循环、限速和失败记录。边界：脚本只解决生成与落盘，不能替代效果审核；接口鉴权、响应结构和限速阈值必须结合真实环境确认。

准备商品描述文件与图片保存目录

建议用 CSV 作为输入，每行一个商品，至少包含 product_id 和 product_name 两个字段。其他字段按提示词需要增加，例如场景、风格、细节要求。文件名只用 product_id 加随机短串，不要用商品名，避免特殊字符和重复覆盖。

product_id,product_name,scene,style,extra
SKU001,白色陶瓷马克杯,简约办公桌,自然光摄影,品牌标识朝上
SKU002,黑色蓝牙耳机,深色背景,科技感,配充电盒

输出目录建议分两个：output/ 存图，logs/ 存失败记录。CSV 文件放在 input/ 下。这样重跑时只需要处理失败清单。

编写基础请求函数：发送一条提示词并接收结果

不要一上来就写循环。先写一个单次生成函数，跑通后再批量。函数里用占位 API 地址；请求头需要包含鉴权信息，但不要硬编码在源码里。设置超时，并检查 HTTP 状态码。

import os
import requests

API_URL = 'https://api.example.com/v1/images/generations'
API_KEY = os.environ.get('IMAGE_API_KEY')

HEADERS = {
    'Authorization': f'Bearer {API_KEY}',
    'Content-Type': 'application/json',
}

def generate_product_image(prompt, product_id):
    payload = {
        'model': 'seedream-5.0-pro',
        'prompt': prompt,
        'n': 1,
        'response_format': 'b64_json',
    }
    try:
        resp = requests.post(API_URL, json=payload, headers=HEADERS, timeout=60)
        resp.raise_for_status()
    except requests.exceptions.RequestException as exc:
        raise RuntimeError(f'请求失败 product_id={product_id}: {exc}') from exc

    data = resp.json()
    image_data = data['data'][0].get('b64_json') or data['data'][0].get('url')
    if not image_data:
        raise RuntimeError(f'响应中没有图片数据 product_id={product_id}: {data}')
    return image_data

先单独调用一次，打印返回类型。如果是 base64 字符串，保存前需要解码；如果是 URL，则需要下载。确认响应结构后再进入批量循环。

加入循环与限速逻辑

批量循环用 csv.DictReader 逐行读取，避免一次性把大文件读进内存。每次请求之间加一个可配置的间隔，从短间隔开始，遇到限流就调大。重试用指数退避，连续失败多次后写入失败清单，不要中断整个任务。

import csv
import time

def build_prompt(row):
    return '，'.join([row['product_name'], row['scene'], row['style'], row['extra']])

def process_csv(csv_path, output_dir, failure_log_path,
                request_interval=1.0, max_retries=3):
    with open(csv_path, encoding='utf-8-sig') as f:
        reader = csv.DictReader(f)
        for row in reader:
            product_id = row['product_id']
            prompt = build_prompt(row)

            for attempt in range(max_retries):
                try:
                    image_data = generate_product_image(prompt, product_id)
                    save_image(image_data, output_dir, product_id)
                    break
                except Exception as exc:
                    if attempt == max_retries - 1:
                        log_failure(row, str(exc), failure_log_path)
                    else:
                        time.sleep(2 ** attempt)

            time.sleep(request_interval)

注意：遇到 401、403 这类鉴权错误，重试没有意义。建议先把单次请求调通，再保留重试逻辑处理偶发的超时和 5xx。

保存图片并记录失败项

保存图片时需要同时处理两种常见返回：URL 或 base64。文件名使用 product_id 加随机短串，避免上次生成的文件被覆盖。失败记录用 CSV 追加写入，保留原行字段和错误信息，方便后续只重跑失败项。

import base64
import requests
from pathlib import Path
import uuid

def save_image(image_data, output_dir, product_id):
    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)
    filename = f'{product_id}_{uuid.uuid4().hex[:8]}.png'
    output_path = output_dir / filename

    if isinstance(image_data, str) and image_data.startswith('http'):
        img_resp = requests.get(image_data, timeout=60)
        img_resp.raise_for_status()
        output_path.write_bytes(img_resp.content)
    else:
        output_path.write_bytes(base64.b64decode(image_data))
    return str(output_path)

def log_failure(row, error_message, failure_log_path):
    failure_log_path = Path(failure_log_path)
    failure_log_path.parent.mkdir(parents=True, exist_ok=True)
    file_exists = failure_log_path.exists()

    with open(failure_log_path, 'a', newline='', encoding='utf-8') as f:
        writer = csv.DictWriter(f, fieldnames=list(row.keys()) + ['error'])
        if not file_exists:
            writer.writeheader()
        writer.writerow({**row, 'error': error_message})

将这几个片段按顺序合并为一个 Python 文件，并加上调用入口：

def main():
    process_csv(
        csv_path='input/products.csv',
        output_dir='output',
        failure_log_path='logs/failures.csv',
    )

重跑失败项时，可以读取 logs/failures.csv，用其中的 product_id 和原始字段重新调用 process_csv，也可以只筛选错误行生成一个新的输入 CSV。脚本本身不做人工审核，生成结果是否符合商品主图要求，仍需要人工抽样检查。