Batch API-джоба завершилась

Batch API (OpenAI, Anthropic) дешевле обычного вызова, но выполняется асинхронно — от минут до 24 часов. Всё это время джоба висит в статусе in_progress, и обновлять dashboard руками бессмысленно. Куда удобнее получить push ровно в тот момент, когда она стала completed или failed, да ещё с разбивкой «сколько запросов прошло, сколько упало».

Вариант 1: поллинг статуса → push

Маленький воркер опрашивает статус раз в минуту и шлёт алёрт при переходе в терминальное состояние:

import os, time, requests
import openai

NOTIFLY_URL   = os.environ["NOTIFLY_URL"]
NOTIFLY_TOKEN = os.environ["NOTIFLY_TOKEN"]
POLL_SEC      = 60

client = openai.OpenAI()

def notify(title, msg, prio):
    requests.post(f"{NOTIFLY_URL}/message",
                  params={"token": NOTIFLY_TOKEN},
                  json={"title": title, "message": msg, "priority": prio},
                  timeout=5)

def watch_batch(batch_id):
    """Опрашивать джобу, пока не завершится; затем push с итогами."""
    while True:
        b = client.batches.retrieve(batch_id)
        if b.status in ("completed", "failed", "expired", "cancelled"):
            counts = b.request_counts   # .completed / .failed / .total
            if b.status == "completed" and counts.failed == 0:
                notify(
                    f"✅ Batch готов: {counts.completed} запросов",
                    f"Джоба {batch_id} завершилась без ошибок.",
                    priority=5,
                )
            else:
                notify(
                    f"⚠️ Batch {b.status}: {counts.failed} ошибок",
                    f"Готово {counts.completed}/{counts.total}, "
                    f"ошибок {counts.failed}. Проверьте error_file_id.",
                    priority=8,
                )
            return b.status
        time.sleep(POLL_SEC)

# использование: отправили батч и следим за ним
batch = client.batches.create(
    input_file_id="file-abc", endpoint="/v1/chat/completions",
    completion_window="24h")
watch_batch(batch.id)

Успех без ошибок — тихий priority=5 (можно посмотреть, когда удобно); любые failed или статус failed/expired — громкий priority=8, потому что часть данных недосчиталась. request_counts избавляет от нужды скачивать результат, чтобы понять масштаб.

Вариант 2: не держать процесс — cron/YC-функция

Если не хочется держать воркер живым сутки, вынесите поллинг в scheduled-функцию на Yandex Cloud с timer-trigger. Список активных batch_id положите в файл/YDB, на каждом тике проверяйте статусы, а готовые — вычёркивайте. Готовый скелет — в рецепте Своя cloud-функция integrity-проверки: тот же handler, только вместо /health дёргается client.batches.retrieve.

Вариант 3: webhook провайдера → Webhook Router

Если провайдер умеет слать webhook о завершении батча (OpenAI это поддерживает), поллинг вообще не нужен. Направьте вебхук в Webhook Router — он превратит событие batch.completed в push мгновенно, без единого опроса статуса.

Что положить в текст алёрта

• batch_id и финальный статус (completed / failed / expired);
• разбивку completed / failed / total;
• error_file_id, если ошибки были, — чтобы сразу пойти за деталями;
• сколько джоба выполнялась (полезно для планирования следующей).