ИИ пишет описания для 1000 товаров за ночь: Python + Claude Batch API без копирайтера

sku,name,category,material,size,purpose,brand
SKU-001,Дрель ударная ProMax 800,tools,металл/пластик,30×23×8 см,сверление бетона кирпича дерева,ProMax
SKU-002,Розетка двойная с заземлением,electrical,поликарбонат,8.5×8.5 см,скрытая установка,Schneider
SKU-003,Свитер мериносовый оверсайз,clothing,мериносовая шерсть 100%,S/M/L/XL,повседневная носка,

import anthropic
import pandas as pd
import time
import json

from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request

CLAUDE_API_KEY = "YOUR_ANTHROPIC_API_KEY"

# Системные промпты для каждого стиля категории
STYLE_PROMPTS = {
  "tools": (
      "Ты пишешь технические описания товаров для интернет-магазина. "
      "Стиль: факты и характеристики, профессиональный тон, конкретные цифры. "
      "Длина: 300–450 символов. Без восклицательных знаков. "
      "Включи: назначение, ключевые характеристики, применение."
  ),
  "electrical": (
      "Ты пишешь потребительские описания товаров. "
      "Стиль: выгода для покупателя, простой язык, что даёт дома. "
      "Длина: 280–420 символов. "
      "Включи: главную выгоду, простоту установки, надёжность."
  ),
  "clothing": (
      "Ты пишешь эмоциональные описания одежды для интернет-магазина. "
      "Стиль: образ и ощущение, покупатель видит себя в вещи. "
      "Длина: 300–450 символов. Материал и уход — ненавязчиво в конце."
  ),
}

# Дефолтный промпт для неизвестных категорий
DEFAULT_PROMPT = (
  "Ты пишешь описания товаров для интернет-магазина. "
  "Стиль: чёткий и информативный. "
  "Длина: 300–450 символов. "
  "Включи: назначение, ключевые характеристики, преимущества."
)

def make_product_prompt(row: dict) -> str:
  """Формируем промпт для конкретного товара из строки CSV."""
  parts = [f"Товар: {row['name']}"]
  if row.get('material'):
      parts.append(f"Материал: {row['material']}")
  if row.get('size'):
      parts.append(f"Размер/характеристики: {row['size']}")
  if row.get('purpose'):
      parts.append(f"Назначение: {row['purpose']}")
  if row.get('brand'):
      parts.append(f"Бренд: {row['brand']}")
  parts.append("Напиши описание товара для карточки в интернет-магазине.")
  return "\n".join(parts)


def submit_batch(df: pd.DataFrame) -> str:
  """Отправляем батч-запрос и возвращаем batch_id."""
  client = anthropic.Anthropic(api_key=CLAUDE_API_KEY)

  requests = []
  for _, row in df.iterrows():
      system_prompt = STYLE_PROMPTS.get(row.get("category", ""), DEFAULT_PROMPT)
      user_prompt = make_product_prompt(row.to_dict())

      requests.append(Request(
          custom_id=str(row["sku"]),
          params=MessageCreateParamsNonStreaming(
              model="claude-haiku-4-5",  # Haiku — оптимум для описаний
              max_tokens=600,
              system=system_prompt,
              messages=[{"role": "user", "content": user_prompt}],
          ),
      ))

  batch = client.messages.batches.create(requests=requests)
  print(f"Батч создан: {batch.id} — запросов: {len(requests)}")
  return batch.id


def poll_until_done(batch_id: str, poll_interval: int = 60) -> None:
  """Ждём пока батч обработается. Проверяем каждые N секунд."""
  client = anthropic.Anthropic(api_key=CLAUDE_API_KEY)

  while True:
      batch = client.messages.batches.retrieve(batch_id)
      status = batch.processing_status

      if status == "ended":
          counts = batch.request_counts
          print(f"Готово. Успешно: {counts.succeeded}, ошибок: {counts.errored}")
          return
      elif status == "canceling":
          raise RuntimeError("Батч отменён")

      print(f"Статус: {status} — ждём {poll_interval} с...")
      time.sleep(poll_interval)


def collect_results(batch_id: str, df: pd.DataFrame) -> pd.DataFrame:
  """Собираем результаты и добавляем описания в датафрейм."""
  client = anthropic.Anthropic(api_key=CLAUDE_API_KEY)

  results = {}
  for result in client.messages.batches.results(batch_id):
      if result.result.type == "succeeded":
          text = result.result.message.content[0].text
          results[result.custom_id] = text.strip()
      else:
          results[result.custom_id] = ""
          print(f"Ошибка для {result.custom_id}: {result.result.type}")

  df["description"] = df["sku"].astype(str).map(results).fillna("")
  return df


if __name__ == "__main__":
  # 1. Загружаем CSV
  df = pd.read_csv("products.csv")
  print(f"Загружено товаров: {len(df)}")

  # 2. Можно обрабатывать по 1000 — одним батчем
  batch_id = submit_batch(df)

  # 3. Сохраняем batch_id — на случай если скрипт упадёт
  with open("batch_id.txt", "w") as f:
      f.write(batch_id)

  # 4. Ждём завершения
  poll_until_done(batch_id)

  # 5. Собираем результаты
  df = collect_results(batch_id, df)

  # 6. Сохраняем в новый CSV
  df.to_csv("products_with_descriptions.csv", index=False)
  print(f"Готово. Сохранено: products_with_descriptions.csv")

import pandas as pd

df = pd.read_csv("products_with_descriptions.csv")

# 1. Пустые описания
empty = df[df["description"].str.len() == 0]
print(f"Пустые: {len(empty)}")

# 2. Слишком короткие (< 200 символов)
short = df[df["description"].str.len() < 200]
print(f"Слишком короткие: {len(short)}")
print(short[["sku", "name", "description"]].to_string())

# 3. Подозрительные дубли — сравниваем первые 60 символов
df["desc_start"] = df["description"].str[:60]
dupes = df[df["desc_start"].duplicated(keep=False)]
print(f"Похожих по началу: {len(dupes)}")

# 4. Базовая статистика
print(f"\nСредняя длина: {df['description'].str.len().mean():.0f} символов")
print(f"Минимум: {df['description'].str.len().min()}")
print(f"Максимум: {df['description'].str.len().max()}")