Персональные письма клиентам из CRM через API

id,name,company,event,facts,manager
C22,Константин,"ООО «Сантехника Дом»",ушёл 3 месяца назад,"причина: не было интеграции с Ozon — сейчас она есть",Ольга
C29,Мария,"магазин чая «Пиала»",жалоба в поддержку не решена,"ждёт ответа 3 дня по вопросу отчёта о продажах",Игорь

Продукт: «Складка» — онлайн-учёт склада для малого
бизнеса.
Тарифы: «Старт» — 990 ₽ в месяц, один склад;
«Бизнес» — 2 490 ₽ в месяц, до пяти складов.
Пробный период — 14 дней, без привязки карты.
Интеграции: 1С, Ozon, Wildberries, МойСклад не
поддерживается.
Акция до 31 октября: при оплате за год — два месяца
в подарок.
Поддержка: чат в приложении, с 9 до 21 по Москве,
без выходных.
Демо: 30 минут по видеосвязи, менеджер показывает на
данных клиента.

{
  "id": "C22",
  "subject": "Складка теперь с Ozon",
  "send": true,
  "body": "Константин, добрый день. Появилась интеграция с Ozon — помню, вы ждали её, когда работали со Складкой. Остальные условия прежние: тариф «Старт» за 990…"
}

# write_mail.py — персональные письма клиентам по строкам CRM через Polza
import asyncio, csv, json, os
from openai import AsyncOpenAI, APIStatusError

MODEL = "deepseek/deepseek-v4-pro"
PROVIDER = "streamlake/fp8"  # самый дешёвый провайдер модели: top_provider.name в каталоге
CONCURRENCY = 10
# острые поводы решает код, а не модель: такие письма уходят менеджеру на проверку
HOLD_WORDS = ["жалоб", "возврат", "вернуть деньги", "вернула", "груб", "не писать"]
SENSITIVE_MODEL = "anthropic/claude-sonnet-5"  # черновик для менеджера пишет модель аккуратнее

ai = AsyncOpenAI(base_url="https://polza.ai/api/v1", api_key=os.environ["POLZA_API_KEY"])
SYSTEM = open("prompt.txt", encoding="utf-8").read() + "\n\nО продукте:\n" + open("offer.txt", encoding="utf-8").read()

SCHEMA = {
    "name": "client_email",
    "strict": True,
    "schema": {
        "type": "object",
        "properties": {"subject": {"type": "string"}, "body": {"type": "string"},
                       "send": {"type": "boolean"}, "reason": {"type": "string"}},
        "required": ["subject", "body", "send", "reason"],
        "additionalProperties": False,
    },
}

async def write(row, sem, stats):
    hold = any(w in (row["event"] + " " + row["facts"]).lower() for w in HOLD_WORDS)
    model = SENSITIVE_MODEL if hold else MODEL
    extra = {"reasoning": {"effort": "none"}}
    if not hold:
        extra["provider"] = {"only": [PROVIDER], "allow_fallbacks": False}
    prompt = (f"Клиент: {row['name']}, {row['company']}\nПовод: {row['event']}\n"
              f"Факты из CRM: {row['facts']}\nМенеджер: {row['manager']}")
    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=extra,
                )
                break
            except APIStatusError as e:
                if e.status_code in (429, 500, 502, 503) and attempt < 4:
                    await asyncio.sleep(2 ** attempt)
                    continue
                raise
    stats["rub"] += r.usage.model_extra.get("cost_rub", 0)
    mail = json.loads(r.choices[0].message.content)
    if hold:
        mail["send"], mail["reason"] = False, "острый повод: письмо проверяет менеджер"
    return {"id": row["id"], "model": model, **mail}

async def main():
    rows = list(csv.DictReader(open("crm.csv", encoding="utf-8")))
    sem, stats = asyncio.Semaphore(CONCURRENCY), {"rub": 0}
    results = await asyncio.gather(*(write(r, sem, stats) for r in rows), return_exceptions=True)
    ok = 0
    with open("emails.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['id']}\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())