Como Gerar Imagens de Produto em Massa com a API Seedream 5.0 Pro

O que “Seedream 5.0 Pro API Batch” realmente significa

Três frases são frequentemente tratadas como sinônimos. Elas não são.

A entrada com múltiplas referências melhora um resultado. Ela não cria dez resultados. Uma vista frontal do produto, um close-up do material e uma referência de iluminação podem todos orientar uma única imagem final. Para criar imagens para 100 produtos, o cliente ainda envia 100 solicitações.

Essa distinção afeta tudo a jusante. Um SKU por solicitação dá a cada saída um proprietário, nome de arquivo, registro de custo e histórico de tentativas. A concorrência faz a fila andar mais rápido; ela não transforma o modelo em um endpoint de múltiplas saídas. O comportamento subjacente do modelo está documentado na matriz de geração de imagens da BytePlus, enquanto a GPT Proto expõe o modelo por meio de suas próprias rotas /api/v3/.

O que você vai construir

Os dados passam por cinco estágios:

catálogo CSV → prompt e referências → pool de solicitações controlado → arquivos baixados → revisão humana

O script usa solicitações síncronas com enable_sync_mode: true. Essa escolha é deliberada. A página ao vivo do modelo da GPT Proto documenta um fluxo de trabalho assíncrono com result_id, mas o esquema completo de resposta de status e saída não é publicado nessa página. Um tutorial de produção não deve inventar esses campos.

O modo síncrono ainda oferece suporte à geração em massa. Cada worker processa um produto desde a solicitação até o download, enquanto o ThreadPoolExecutor permite que um pequeno número de workers seja executado ao mesmo tempo. O manifesto é escrito pela thread principal após cada worker terminar, para que execuções interrompidas possam pular arquivos que já estão presentes.

Para uma versão posterior de alto rendimento, defina enable_sync_mode como false somente depois de capturar uma resposta ao vivo e confirmar os campos atuais de task ID, status, output e error. Em seguida, armazene o result_id retornado antes de consultar /api/v3/predictions/{result_id}/result.

O que você precisa antes de começar

Instale o Python e a única dependência:

python --version
python -m pip install requests

Crie uma chave de API da GPT Proto e mantenha-a em uma variável de ambiente:

export GPTPROTO_API_KEY="your-key-here"

Os exemplos atuais da API ao vivo usam este formato de cabeçalho:

Authorization: Bearer YOUR_API_KEY

Você também precisa de referências de produto em URLs HTTPS públicas. Um caminho como /Users/me/photos/mug.jpg existe apenas no seu computador. A API remota não consegue buscá-lo. Coloque os arquivos em um CDN, bucket de armazenamento de objetos ou outro local HTTPS que a API possa acessar sem cookie de login.

Comece com uma pequena fatia do catálogo. Escolha produtos com formatos, materiais e complexidade de rótulo diferentes. Uma caneca simples pode passar enquanto uma bolsa com várias alças falha, então um teste fácil não é suficiente para aprovar um prompt compartilhado.

Estime o custo do lote antes de executá-lo

O preço ao vivo do Seedream 5.0 Pro da GPT Proto foi verificado em 17 de setembro de 2026. Use a página do modelo para preços por imagem e a página de preços da GPT Proto para informações de preços no nível da conta.

A atual API de Edição de Imagem Seedream 5.0 Pro aceita até 10 imagens de referência. A primeira está incluída no preço base. As referências de dois a dez adicionam $0.0027 cada.

Isso produz estes números de planejamento antes de novas tentativas ou gerações rejeitadas:

A conta de geração bruta é fácil de estimar. O custo por imagem aprovada não é. Se um produto precisar de várias tentativas, cada imagem gerada conta. Use esta fórmula para um orçamento mais realista:

estimated catalog cost =
  number of SKUs
  × average attempts per approved image
  × (base output price + paid reference-image cost)

Use 1K para testar o sistema de prompt em produtos representativos. Não o trate como uma prévia garantida de uma chamada posterior em 2K. A solicitação em 2K cria uma nova imagem, então a composição e os detalhes podem mudar. Revise novamente o resultado na resolução final.

Etapa 1: Crie um CSV de produtos

Salve isto como products.csv:

sku,name,color,material,angle,background,product_image_url,detail_image_url,style_reference_url,avoid
MUG-001,Stackable coffee mug,matte navy,stoneware,three-quarter,soft warm gray,https://cdn.example.com/mug-front.jpg,https://cdn.example.com/mug-glaze.jpg,,steam or hands
BAG-014,Compact crossbody bag,forest green,pebbled leather,front three-quarter,light beige,https://cdn.example.com/bag-front.jpg,https://cdn.example.com/bag-texture.jpg,https://cdn.example.com/style-neutral.jpg,model or extra straps

Substitua todas as URLs cdn.example.com antes de executar o script. Os campos do CSV têm funções diferentes:

• sku é a chave permanente do produto. Ele também se torna parte do nome do arquivo de saída.

• name, color, material, angle e background contêm fatos que mudam por produto.

• product_image_url é a referência de identidade e é obrigatória.

• detail_image_url pode mostrar material, controles, ferragens ou um close-up do rótulo.

• style_reference_url controla iluminação e tratamento de fundo, não a identidade do produto.

• avoid lista alguns erros visíveis que importam para aquele SKU.

Mantenha os valores curtos e factuais. Não coloque um briefing criativo inteiro em uma célula. Não coloque a chave de API no CSV, no código-fonte ou no manifesto.

O script abaixo rejeita SKUs duplicados, URLs de produto ausentes, referências não HTTPS e mais de dez referências antes de enviar uma solicitação paga.

Etapa 2: Dê a cada imagem de referência uma função

Uma solicitação de imagem com múltiplas referências da API Seedream 5.0 Pro funciona melhor quando o prompt explica por que cada imagem existe. Use uma ordem estável:

1. Referência de identidade do produto: silhueta, proporções, design visível e embalagem.

2. Referência de detalhe: material, ferragens, controles ou tratamento de rótulo.

3. Referência de estilo: apenas iluminação, superfície, paleta e fundo.

O prompt gerado incluirá instruções como:

Image 1 is the product identity reference. Preserve its silhouette, proportions, visible design, and packaging. Image 2 is the detail reference. Use it for material and hardware details. Image 3 is the lighting and background reference only. Do not copy objects from it.

Comece com uma a três referências úteis. Dez entradas não rotuladas custam mais e podem entrar em conflito umas com as outras. Mais imagens não são automaticamente mais precisas.

Um post de lançamento do ComfyUI descreve características estáveis de produto em variações de cena como um caso de uso de produção. Trate isso como uma afirmação de fluxo de trabalho, não como um benchmark independente. A identidade do produto ainda precisa de uma revisão lado a lado com a origem.

Etapa 3: Crie um prompt de produto reutilizável

Mantenha regras para todo o catálogo no código e fatos do SKU no CSV. Uma ordem útil é:

asset type
→ product identity and protected details
→ requested background
→ camera and composition
→ lighting and shadow
→ material behavior
→ exclusions

Para a linha da caneca, o script produz instruções próximas a isto:

Create one square e-commerce hero photograph of the Stackable coffee mug.
The product is matte navy stoneware. Preserve the product identity, silhouette,
proportions, visible design, color family, material, and packaging details from
the references. Show a three-quarter view on a soft warm gray background.
Use a large diffused key light from the upper left and a natural contact shadow.
Keep the whole product inside the frame with crop-safe space. Return one product
image, not a collage or contact sheet. Do not add steam or hands. Do not invent
logos, labels, accessories, duplicate products, handles, straps, or controls.

Nomeie detalhes observáveis. “Qualidade premium” fornece pouca informação física ao modelo. “Sombra de contato suave caindo para trás à direita” fornece. Mantenha as exclusões curtas porque a solicitação atual da GPT Proto não expõe um campo separado negative_prompt.

Para mais padrões de prompt, use o Guia de Prompt do Seedream 5.0 Pro.

Etapa 4: Teste um produto antes de iniciar a fila

Execute uma solicitação síncrona de edição de imagem primeiro:

curl --request POST \
  --url "https://gptproto.com/api/v3/doubao/dola-seedream-5-0-pro-260628/image-edit" \
  --header "Authorization: Bearer $GPTPROTO_API_KEY" \
  --header "Content-Type: application/json" \
  --data '{
    "images": [
      "https://your-cdn.example.com/mug-front.jpg",
      "https://your-cdn.example.com/mug-glaze.jpg"
    ],
    "prompt": "Create one square e-commerce hero photograph of the referenced mug. Image 1 is the product identity reference. Preserve its silhouette, proportions, handle, color, and visible design. Image 2 is the glaze detail reference. Show a three-quarter view on a soft warm gray background with a natural contact shadow. Keep the entire product inside the frame. No hands, steam, extra text, duplicate products, or collage.",
    "size": "1024x1024",
    "enable_base64_output": false,
    "enable_sync_mode": true,
    "output_format": "png"
  }'

O endpoint e o cabeçalho Bearer atual correspondem ao exemplo do modelo ao vivo da GPT Proto. Imprima e retenha a primeira resposta JSON bem-sucedida antes de adicionar mais automação. Verifique se ela contém uma saída que seu cliente possa baixar.

Não continue só porque a imagem parece atraente. Compare-a com as referências:

• A silhueta é reconhecível?

• A cor e o material estão aceitavelmente próximos?

• O modelo adicionou ou removeu uma alça, correia, botão, tampa ou rótulo?

• O texto obrigatório está correto caractere por caractere?

• O produto faz contato físico com a superfície?

• Há espaço suficiente seguro para corte para o posicionamento pretendido no marketplace?

Teste um produto representativo e um produto difícil. Ambos devem passar antes que a fila completa comece.

Etapa 5: Execute uma solicitação por SKU

O script em lote envia o mesmo payload de edição de imagem usado no teste cURL. As variáveis vêm de cada linha do CSV. MAX_WORKERS controla quantas solicitações independentes podem estar ativas ao mesmo tempo.

Comece com um valor pequeno, como 2 ou 3. A documentação da BytePlus pode publicar limites para contas diretas da BytePlus, mas esses números não são automaticamente os limites da sua conta GPT Proto. Se a API retornar 429, reduza a contagem de workers e siga as orientações de nova tentativa da resposta.

O script usa nomes determinísticos:

{sku}_hero_{attempt:02d}.{extension}

Exemplos incluem MUG-001_hero_01.png e BAG-014_hero_02.png. A tentativa permanece visível, para que um revisor possa rastrear o arquivo selecionado de volta ao manifesto.

Etapa 6: Adicione novas tentativas e suporte à retomada

Nem toda falha deve disparar outra solicitação paga.

O script tenta novamente respostas explícitas 429 e 5xx com backoff exponencial e jitter. Ele também respeita um cabeçalho numérico Retry-After quando presente. Erros de autenticação, parâmetro inválido, saldo e moderação são registrados sem um reenvio inalterado.

Timeouts de leitura e conexões interrompidas são diferentes. O servidor pode ter aceitado a solicitação mesmo que o cliente não tenha recebido a resposta. Sem uma chave de idempotência documentada, um reenvio automático poderia criar uma segunda imagem e uma segunda cobrança. O script marca esses casos como manual_check. Isso é mais lento do que tentar novamente às cegas. Também é mais seguro para um orçamento de produção.

Após cada worker terminar, a thread principal atualiza manifest.json por meio de uma substituição atômica de arquivo. Ao reiniciar:

• um item com status downloaded, approved ou manual_review e um arquivo local existente é ignorado;

• um item failed_retryable continua somente se ainda houver tentativas;

• um item manual_check não é enviado novamente de forma automática; e

• um item failed mantém seu erro e é ignorado até que um operador corrija a causa, então remove essa entrada de SKU do manifesto ou altera seu status para failed_retryable.

Este design síncrono não pode salvar um ID de tarefa assíncrona. Se você adotar posteriormente o fluxo de trabalho em segundo plano documentado, adicione task_id ao mesmo manifesto e salve o result_id ao vivo imediatamente após o envio.

Script Python completo para imagens de produtos Seedream em massa

Salve o seguinte como bulk_seedream_products.py ao lado de products.csv.

from __future__ import annotations

import base64
import csv
import json
import os
import random
import re
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.parse import urlparse

import requests

API_URL = (
    "https://gptproto.com/api/v3/doubao/"
    "dola-seedream-5-0-pro-260628/image-edit"
)
CSV_PATH = Path(os.getenv("PRODUCTS_CSV", "products.csv"))
OUTPUT_DIR = Path(os.getenv("OUTPUT_DIR", "generated"))
MANIFEST_PATH = Path(os.getenv("MANIFEST_PATH", "manifest.json"))

SIZE = os.getenv("SEEDREAM_SIZE", "1024x1024")
OUTPUT_FORMAT = os.getenv("SEEDREAM_OUTPUT_FORMAT", "png")
MAX_WORKERS = int(os.getenv("MAX_WORKERS", "3"))
MAX_ATTEMPTS = int(os.getenv("MAX_ATTEMPTS", "4"))
REQUEST_TIMEOUT = int(os.getenv("REQUEST_TIMEOUT", "300"))
PROMPT_VERSION = os.getenv("PROMPT_VERSION", "product-hero-v1")

REFERENCE_FIELDS = (
    (
        "product_image_url",
        "the product identity reference. Preserve its silhouette, proportions, "
        "visible design, color family, material, and packaging",
    ),
    (
        "detail_image_url",
        "the detail reference. Use it for material, hardware, controls, and label details",
    ),
    (
        "style_reference_url",
        "the lighting and background reference only. Do not copy products or props from it",
    ),
)

def now_utc() -> str:
    return datetime.now(timezone.utc).isoformat()

def is_https_url(value: str) -> bool:
    parsed = urlparse(value)
    return parsed.scheme == "https" and bool(parsed.netloc)

def clean_filename(value: str) -> str:
    cleaned = re.sub(r"[^A-Za-z0-9._-]+", "-", value.strip())
    return cleaned.strip("-._") or "product"

def reference_data(row: dict[str, str]) -> tuple[list[str], list[str]]:
    urls: list[str] = []
    role_lines: list[str] = []
    for field, role in REFERENCE_FIELDS:
        url = row.get(field, "").strip()
        if not url:
            continue
        urls.append(url)
        role_lines.append(f"Image {len(urls)} is {role}.")
    return urls, role_lines

def validate_and_load_rows(path: Path) -> list[dict[str, str]]:
    required = {
        "sku",
        "name",
        "color",
        "material",
        "angle",
        "background",
        "product_image_url",
        "detail_image_url",
        "style_reference_url",
        "avoid",
    }

    with path.open("r", encoding="utf-8-sig", newline="") as handle:
        reader = csv.DictReader(handle)
        fields = set(reader.fieldnames or [])
        missing = required - fields
        if missing:
            raise ValueError(f"CSV is missing columns: {sorted(missing)}")

        rows: list[dict[str, str]] = []
        seen_skus: set[str] = set()

        for line_number, raw_row in enumerate(reader, start=2):
            row = {key: (value or "").strip() for key, value in raw_row.items()}
            sku = row["sku"]
            if not sku:
                raise ValueError(f"Line {line_number}: sku is required")
            if sku in seen_skus:
                raise ValueError(f"Line {line_number}: duplicate sku {sku!r}")
            seen_skus.add(sku)

            urls, _ = reference_data(row)
            if not row["product_image_url"]:
                raise ValueError(
                    f"Line {line_number}: product_image_url is required"
                )
            if len(urls) > 10:
                raise ValueError(
                    f"Line {line_number}: {sku} has more than 10 references"
                )
            invalid_urls = [url for url in urls if not is_https_url(url)]
            if invalid_urls:
                raise ValueError(
                    f"Line {line_number}: references must be public HTTPS URLs: "
                    f"{invalid_urls}"
                )
            rows.append(row)

    if not rows:
        raise ValueError("The CSV contains no product rows")
    return rows

def build_prompt(row: dict[str, str], role_lines: list[str]) -> str:
    avoid = row.get("avoid") or "extra products or unrelated props"
    references = " ".join(role_lines)
    return (
        f"Create one square e-commerce hero photograph of the {row['name']}. "
        f"{references} "
        f"The product is {row['color']} {row['material']}. "
        "Preserve the product identity, silhouette, proportions, visible design, "
        "color family, material, and packaging details from the references. "
        f"Show a {row['angle']} view on a {row['background']} background. "
        "Use a large diffused key light from the upper left and a natural contact "
        "shadow falling back-right. Keep the whole product inside the frame with "
        "crop-safe space. Return one product image, not a collage or contact sheet. "
        f"Do not add {avoid}. Do not invent logos, labels, accessories, duplicate "
        "products, handles, straps, buttons, caps, or controls."
    )

def load_manifest(path: Path) -> dict[str, Any]:
    if not path.exists():
        return {"prompt_version": PROMPT_VERSION, "items": {}}
    with path.open("r", encoding="utf-8") as handle:
        data = json.load(handle)
    if not isinstance(data.get("items"), dict):
        raise ValueError("manifest.json must contain an object named 'items'")
    return data

def save_manifest(path: Path, manifest: dict[str, Any]) -> None:
    temporary = path.with_suffix(path.suffix + ".tmp")
    with temporary.open("w", encoding="utf-8") as handle:
        json.dump(manifest, handle, indent=2, ensure_ascii=False)
        handle.write("\n")
    os.replace(temporary, path)

def output_items(payload: dict[str, Any]) -> list[Any]:
    data = payload.get("data")
    containers = [data, payload] if isinstance(data, dict) else [payload]
    for container in containers:
        for key in ("outputs", "output", "images", "image"):
            value = container.get(key)
            if value:
                return value if isinstance(value, list) else [value]
    raise ValueError(
        "No output field found. Inspect the saved response and update output_items() "
        "to match the current API response."
    )

def output_value(item: Any) -> str:
    if isinstance(item, str):
        return item
    if isinstance(item, dict):
        for key in ("url", "image_url", "b64_json", "base64"):
            value = item.get(key)
            if isinstance(value, str) and value:
                return value
    raise ValueError(f"Unsupported output item: {item!r}")

def extension_from_content_type(content_type: str) -> str:
    mime = content_type.split(";", 1)[0].strip().lower()
    return {
        "image/png": ".png",
        "image/jpeg": ".jpg",
        "image/webp": ".webp",
    }.get(mime, f".{OUTPUT_FORMAT}")

def save_output(value: str, file_stem: Path) -> tuple[Path, str | None]:
    if value.startswith("data:image/"):
        header, encoded = value.split(",", 1)
        mime = header.split(";", 1)[0].removeprefix("data:")
        destination = file_stem.with_suffix(extension_from_content_type(mime))
        destination.write_bytes(base64.b64decode(encoded))
        return destination, None

    if value.startswith("https://") or value.startswith("http://"):
        response = requests.get(value, timeout=REQUEST_TIMEOUT)
        response.raise_for_status()
        extension = extension_from_content_type(
            response.headers.get("Content-Type", "")
        )
        destination = file_stem.with_suffix(extension)
        destination.write_bytes(response.content)
        return destination, value

# Some responses return raw Base64 without a data URI.
    try:
        decoded = base64.b64decode(value, validate=True)
    except ValueError as exc:
        raise ValueError("Output is neither a URL nor valid Base64") from exc
    destination = file_stem.with_suffix(f".{OUTPUT_FORMAT}")
    destination.write_bytes(decoded)
    return destination, None

def backoff_seconds(attempt: int, response: requests.Response | None) -> float:
    if response is not None:
        retry_after = response.headers.get("Retry-After")
        if retry_after:
            try:
                return max(0.0, float(retry_after))
            except ValueError:
                pass
    return min(60.0, (2 ** (attempt - 1)) + random.uniform(0.0, 1.0))

def result_record(
    row: dict[str, str],
    references: list[str],
    status: str,
    attempt: int,
    *,
    output_url: str | None = None,
    local_path: str | None = None,
    error: str | None = None,
    created_at: str | None = None,
) -> dict[str, Any]:
    timestamp = now_utc()
    return {
        "sku": row["sku"],
        "asset_type": "hero",
        "status": status,
        "attempt": attempt,
        "prompt_version": PROMPT_VERSION,
        "reference_urls": references,
        "output_url": output_url,
        "local_path": local_path,
        "error": error,
        "requester": "",
        "reviewer": "",
        "approval_status": "pending" if status == "downloaded" else "",
        "created_at": created_at or timestamp,
        "updated_at": timestamp,
    }

def process_product(
    row: dict[str, str], previous: dict[str, Any] | None
) -> dict[str, Any]:
    api_key = os.environ["GPTPROTO_API_KEY"]
    references, role_lines = reference_data(row)
    prompt = build_prompt(row, role_lines)
    created_at = (previous or {}).get("created_at") or now_utc()
    first_attempt = int((previous or {}).get("attempt", 0)) + 1

    if first_attempt > MAX_ATTEMPTS:
        return result_record(
            row,
            references,
            "failed",
            first_attempt - 1,
            error="Maximum attempts already reached",
            created_at=created_at,
        )

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }
    body = {
        "images": references,
        "prompt": prompt,
        "size": SIZE,
        "enable_base64_output": False,
        "enable_sync_mode": True,
        "output_format": OUTPUT_FORMAT,
    }

    for attempt in range(first_attempt, MAX_ATTEMPTS + 1):
        response: requests.Response | None = None
        try:
            response = requests.post(
                API_URL,
                headers=headers,
                json=body,
                timeout=REQUEST_TIMEOUT,
            )

            if response.status_code == 429 or response.status_code >= 500:
                message = f"HTTP {response.status_code}: {response.text[:500]}"
                if attempt == MAX_ATTEMPTS:
                    return result_record(
                        row,
                        references,
                        "failed_retryable",
                        attempt,
                        error=message,
                        created_at=created_at,
                    )
                time.sleep(backoff_seconds(attempt, response))
                continue

            response.raise_for_status()
            payload = response.json()
            item = output_items(payload)[0]
            value = output_value(item)
            file_stem = OUTPUT_DIR / (
                f"{clean_filename(row['sku'])}_hero_{attempt:02d}"
            )
            local_file, output_url = save_output(value, file_stem)
            return result_record(
                row,
                references,
                "downloaded",
                attempt,
                output_url=output_url,
                local_path=str(local_file),
                created_at=created_at,
            )

        except requests.exceptions.ConnectTimeout as exc:

# No connection was established; a bounded retry is reasonable.
            if attempt == MAX_ATTEMPTS:
                return result_record(
                    row,
                    references,
                    "failed_retryable",
                    attempt,
                    error=str(exc),
                    created_at=created_at,
                )
            time.sleep(backoff_seconds(attempt, response))

        except (
            requests.exceptions.ReadTimeout,
            requests.exceptions.ConnectionError,
        ) as exc:

# The server may have accepted the request. Avoid a blind resubmission.
            return result_record(
                row,
                references,
                "manual_check",
                attempt,
                error=f"Ambiguous network failure; verify account history: {exc}",
                created_at=created_at,
            )

        except requests.exceptions.HTTPError as exc:
            status_code = response.status_code if response is not None else "unknown"
            details = response.text[:500] if response is not None else str(exc)
            return result_record(
                row,
                references,
                "failed",
                attempt,
                error=f"HTTP {status_code}: {details}",
                created_at=created_at,
            )

        except (ValueError, KeyError, json.JSONDecodeError) as exc:
            return result_record(
                row,
                references,
                "manual_check",
                attempt,
                error=str(exc),
                created_at=created_at,
            )

    return result_record(
        row,
        references,
        "failed",
        MAX_ATTEMPTS,
        error="Request ended without a result",
        created_at=created_at,
    )

def should_skip(entry: dict[str, Any] | None) -> bool:
    if not entry:
        return False
    if entry.get("status") in {"manual_check", "failed"}:
        return True
    if entry.get("status") not in {"downloaded", "manual_review", "approved"}:
        return False
    local_path = entry.get("local_path")
    return bool(local_path and Path(local_path).exists())

def main() -> None:
    if not os.getenv("GPTPROTO_API_KEY"):
        raise SystemExit("Set GPTPROTO_API_KEY before running the script")
    if MAX_WORKERS < 1:
        raise SystemExit("MAX_WORKERS must be at least 1")

    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
    rows = validate_and_load_rows(CSV_PATH)
    manifest = load_manifest(MANIFEST_PATH)
    items: dict[str, Any] = manifest["items"]

    pending: list[dict[str, str]] = []
    for row in rows:
        entry = items.get(row["sku"])
        if should_skip(entry):
            print(f"SKIP {row['sku']}: {entry['status']}")
        else:
            pending.append(row)

    if not pending:
        print("Nothing to submit")
        return

    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        future_to_row = {
            executor.submit(process_product, row, items.get(row["sku"])): row
            for row in pending
        }
        for future in as_completed(future_to_row):
            row = future_to_row[future]
            try:
                record = future.result()
            except Exception as exc:  # Preserve an unexpected worker failure.
                references, _ = reference_data(row)
                record = result_record(
                    row,
                    references,
                    "failed",
                    int(items.get(row["sku"], {}).get("attempt", 0)),
                    error=f"Unexpected worker error: {exc}",
                )
            items[row["sku"]] = record
            manifest["prompt_version"] = PROMPT_VERSION
            manifest["updated_at"] = now_utc()
            save_manifest(MANIFEST_PATH, manifest)
            print(f"{record['status'].upper()} {row['sku']}: {record.get('local_path') or record.get('error')}")

if __name__ == "__main__":
    main()

Execute com:

python bulk_seedream_products.py

Você pode alterar a contagem de workers e o tamanho de saída sem editar o arquivo:

MAX_WORKERS=2 SEEDREAM_SIZE=2048x2048 python bulk_seedream_products.py

O console informa cada SKU quando ele termina. Abra manifest.json para ver quais arquivos foram baixados, falharam ou estão retidos para verificação manual. Se a resposta ao vivo usar um formato de campo não coberto por output_items(), o script registra a incompatibilidade em vez de tratar silenciosamente o trabalho como concluído. Atualize esse pequeno parser usando o JSON capturado durante o teste de produto único.

Revise cada imagem antes que ela chegue à loja

Uma imagem baixada não é uma imagem aprovada. Mantenha o status de geração e o status de aprovação separados.

Mova um registro de pending para approved, rejected ou needs_edit somente depois que uma pessoa comparar a saída com as imagens de origem. Se o texto do rótulo precisar ser legalmente exato, mantenha o rótulo de origem na pós-produção em vez de confiar em texto gerado sem verificação.

Para um benchmark de qualidade próprio, execute uma pequena matriz antes de escalar:

• um frasco rotulado em um fundo de catálogo, cena lifestyle e layout de campanha;

• uma bolsa de couro nos mesmos três tipos de cena; e

• um abajur com geometria e detalhes de controle visíveis.

Registre a taxa de aprovação na primeira passagem, tentativas por imagem aprovada, falhas de identidade, erros de rótulo e custo de API por imagem aprovada. Essas medições descrevem seu catálogo melhor do que uma demonstração genérica de modelo.

Torne o fluxo de trabalho utilizável por uma equipe

O manifesto é o contrato de transferência entre o operador, o designer e o revisor. Mantenha estes campos mesmo que uma pessoa atualmente faça todo o trabalho:

Trate o SKU como imutável. Se o prompt compartilhado mudar, incremente PROMPT_VERSION em vez de sobrescrever o histórico. Reexecute linhas rejeitadas, não o catálogo inteiro. Armazene ativos aprovados separadamente dos rascunhos e nunca faça commit de uma chave de API no mesmo repositório.

Esta é a diferença entre um script em lote e um fluxo de trabalho de equipe. O script produz arquivos. Os registros explicam como esses arquivos foram produzidos e se eles podem ser publicados.

Problemas comuns e correções

Não adicione novas tentativas até saber se a falha ocorreu antes ou depois de o servidor aceitar a solicitação. Confiabilidade não é o mesmo que reenviar tudo.

Quando o Seedream 5.0 Pro é a escolha certa

O Seedream 5.0 Pro é adequado para cenas lifestyle de produtos, layouts de campanha, criativos localizados e variações de catálogo guiadas por imagens de referência. Seus níveis fixos por imagem também tornam o orçamento de geração bruta simples de calcular.

Ele não substitui fotografia de produto ou pós-produção quando a saída precisa preservar geometria idêntica em pixels, embalagem legalmente exata ou texto de rótulo regulamentado. Também não é adequado para um sistema que publica toda imagem gerada sem revisão, ou para um requisito de que uma solicitação retorne muitos arquivos de produto separados.

Use-o para produção criativa escalável. Mantenha uma pessoa entre a geração e a publicação.

Comece com uma fatia representativa do catálogo

Escolha um pequeno conjunto de SKUs que cubra formatos, materiais, rótulos e acessórios diferentes. Bloqueie o esquema CSV e a versão do prompt. Meça tentativas, motivos de rejeição e custo por imagem aprovada. Expanda somente depois que o fluxo de retomada e revisão funcionar.

Revise os parâmetros e preços atuais na página da API Seedream 5.0 Pro, então execute um produto antes de financiar um lote de catálogo completo. Para referências de produto em vez de geração apenas de texto, abra a API de Edição de Imagem Seedream 5.0 Pro.