Массовая генерация SEO-текстов через API

sku,name,specs,keywords
LM-12Q,Настольная лампа Lumo Q с беспроводной зарядкой,"12 Вт; беспроводная зарядка Qi 10 Вт; 3000–6000 K; таймер отключения 30 мин; гибкая стойка","лампа с беспроводной зарядкой; настольная лампа с зарядкой"
KT-TC,Чайник с терморегулятором Boil Control,"1,5 л; 5 температур: 40, 70, 80, 90, 100 °C; поддержание температуры 30 мин; 2000 Вт","чайник с выбором температуры; чайник с поддержанием температуры"

{
  "sku": "LM-12Q",
  "title": "Настольная лампа Lumo Q с беспроводной зарядкой Qi 10 Вт",
  "description": "Настольная лампа с зарядкой Qi 10 Вт и регулировкой цветовой температуры от 3000 до 6000 K",
  "text": "Настольная лампа с беспроводной зарядкой Lumo Q решает задачу освещения рабочего стола и одновременной подзарядки телефона. Гибкая стойка позволяет направить свет точно туда, где вы читаете или пишете вечером. Встроенная…"
}

# seo_batch.py — описания товаров из CSV через Polza
import asyncio, csv, json, os
from openai import AsyncOpenAI, APIStatusError

MODEL = "deepseek/deepseek-v4-pro"
# самый дешёвый провайдер модели: поле top_provider.name в каталоге
PROVIDER = "streamlake/fp8"
CONCURRENCY = 10  # сколько запросов идёт одновременно

ai = AsyncOpenAI(base_url="https://polza.ai/api/v1", api_key=os.environ["POLZA_API_KEY"])
SYSTEM = open("prompt.txt", encoding="utf-8").read()

SCHEMA = {
    "name": "product_page",
    "strict": True,
    "schema": {
        "type": "object",
        "properties": {
            "title": {"type": "string"},
            "description": {"type": "string"},
            "text": {"type": "string"},
        },
        "required": ["title", "description", "text"],
        "additionalProperties": False,
    },
}

async def generate(row, sem, stats):
    prompt = (f"Товар: {row['name']}\n"
              f"Характеристики: {row['specs']}\n"
              f"Ключевые фразы: {row['keywords']}")
    async with sem:
        for attempt in range(5):
            try:
                r = await ai.chat.completions.create(
                    model=MODEL,
                    messages=[{"role": "system", "content": SYSTEM},
                              {"role": "user", "content": prompt}],
                    response_format={"type": "json_schema", "json_schema": SCHEMA},
                    extra_body={
                        # без рассуждений: дешевле и быстрее на шаблонных текстах
                        "reasoning": {"effort": "none"},
                        # без закрепления часть запросов уходит к дорогим провайдерам
                        "provider": {"only": [PROVIDER], "allow_fallbacks": False},
                    },
                )
                break
            except APIStatusError as e:
                if e.status_code in (429, 500, 502, 503) and attempt < 4:
                    await asyncio.sleep(2 ** attempt)  # пауза 1, 2, 4, 8 секунд
                    continue
                raise
    stats["rub"] += r.usage.model_extra.get("cost_rub", 0)  # фактическая стоимость запроса
    return {"sku": row["sku"], **json.loads(r.choices[0].message.content)}

async def main():
    rows = list(csv.DictReader(open("products.csv", encoding="utf-8")))
    sem, stats = asyncio.Semaphore(CONCURRENCY), {"rub": 0}
    results = await asyncio.gather(*(generate(r, sem, stats) for r in rows),
                                   return_exceptions=True)
    ok = 0
    with open("results.jsonl", "w", encoding="utf-8") as out, \
         open("errors.txt", "w", encoding="utf-8") as err:
        for row, res in zip(rows, results):
            if isinstance(res, Exception):
                err.write(f"{row['sku']}\t{res}\n")  # перезапустите только эти строки
            else:
                out.write(json.dumps(res, ensure_ascii=False) + "\n")
                ok += 1
    print(f"Готово {ok} из {len(rows)}, расход {stats['rub']:.2f} ₽")

asyncio.run(main())