Генерация видео через API Python: сохранение заказов и восстановление после сбоя

Сбой между POST и ответом — нормальный рабочий случай

Допустим, Python backend отправил запрос, но соединение оборвалось через несколько секунд. Видео могло уже попасть в очередь провайдера. Если приложение создаст новое задание с новым ключом, получится потенциально вторая оплачиваемая работа. Для восстановления нужен сохранённый заказ, а не ещё одна случайная попытка.

В примере ниже отделим три действия: подготовить неизменяемое тело, привязать возвращённый ID и обновить статус. Программа состояния сама не выполняет платный POST. Она предназначена для сервера, где отправка подключается по общему quickstart, а постоянная запись защищает её от путаницы после перезапуска.

Для подготовки нужен Python 3.10 или новее. Стандартный sqlite3 хранит записи в локальном файле; requests требуется только команде refresh. В многопользовательском сервисе добавьте собственную привязку заказа к пользователю и организации: пример рассчитан на одно контролируемое серверное окружение, а не на публичный доступ к базе.

Что именно хранить в Python backend

Минимальная запись связывает ваш номер заказа, Idempotency-Key, точное тело, ID Gate и последний ответ. Полное тело важнее одного prompt: режим, звук, исходники и длительность тоже меняют заказ. Для сравнения JSON удобно привести к стабильной форме с отсортированными ключами; порядок элементов массивов при этом сохраняется.

Ключ API не относится к записи заказа. Его берут из окружения серверного процесса при сетевом вызове. В базе могут оказаться приватные исходные ссылки и результат: файл базы нужно хранить вне публичного каталога и не добавлять в репозиторий. Локальный пример создаёт новые файлы с ограниченными правами через umask.

Хеш помогает быстро сравнивать записи и искать изменения, но не заменяет исходное тело для повторной отправки. Восстановить только по хешу можно совпадение, а не содержимое запроса. Поэтому пример сохраняет оба значения. Не редактируйте строку вручную ради обхода конфликта: новому содержанию нужен новый заказ.

Сначала сохраните заказ, затем отправляйте

Команда prepare читает request.json и записывает его до обращения к платной генерации. Повтор той же команды ничего не меняет. Если у существующего order ID другое тело или ключ, программа остановится. Это локальная проверка приложения; сервер Gate дополнительно проверяет свою пару ключа и тела.

Для отправки используйте сохранённые model и input без изменения. Схему и оценку проверьте перед принятием заказа. POST /v1/media/generations создаёт оплачиваемую работу и требует сохранённый Idempotency-Key. Полученный ID передайте команде attach. Если ответ потерялся, тело и ключ остаются доступны для восстановления.

Не держите транзакцию SQLite открытой на время HTTP-запроса. Сначала завершите локальное сохранение, затем работайте с сетью, потом коротко запишите результат. Так медленная модель не превращает блокировку базы в многоминутное препятствие для остальных запросов. Несколько процессов требуют отдельного контроля владения отправкой.

Полная утилита состояния: prepare, attach и refresh

Сохраните код как orders.py. Команды prepare и attach работают только с локальной базой и не обращаются к генерации. refresh делает один GET уже известного задания. Это позволяет безопасно встроить утилиту в обработчик или расписание, не создавая при каждом пробуждении новый ролик.

Ваша платная отправка должна использовать тот же сохранённый JSON и ключ. Утилита не подставляет новые значения автоматически. Если ID пока неизвестен, refresh остановится с инструкцией восстановить исходную отправку. Это честнее, чем создавать новый заказ только ради получения удобного идентификатора.

Ниже нет секретных констант. CELESTRIA_API_KEY задаётся только в окружении сервера, а media_generation_… копируется из настоящего ответа HTTP 202. Учебные обозначения в командах после программы не являются работающими ID. Для проверки самой логики можно использовать локальные имитации, не выполняя платных генераций.

import argparse
from contextlib import closing
import hashlib
import json
import os
from pathlib import Path
import re
import sqlite3
from urllib.parse import quote


def canonical(value):
    return json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":"))


def main():
    parser = argparse.ArgumentParser()
    sub = parser.add_subparsers(dest="command", required=True)
    prepare = sub.add_parser("prepare")
    prepare.add_argument("order")
    prepare.add_argument("request_file")
    prepare.add_argument("idempotency_key")
    attach = sub.add_parser("attach")
    attach.add_argument("order")
    attach.add_argument("job_id")
    refresh = sub.add_parser("refresh")
    refresh.add_argument("order")
    args = parser.parse_args()
    os.umask(0o077)
    with closing(sqlite3.connect("video-orders.sqlite")) as db:
        db.row_factory = sqlite3.Row
        db.execute("""CREATE TABLE IF NOT EXISTS orders (
            id TEXT PRIMARY KEY, idem TEXT UNIQUE NOT NULL,
            body TEXT NOT NULL, digest TEXT NOT NULL,
            gate_id TEXT, status TEXT NOT NULL, snapshot TEXT
        )""")
        if args.command == "prepare":
            if not 1 <= len(args.idempotency_key) <= 200:
                raise SystemExit("Ключ идемпотентности: 1–200 символов")
            body = canonical(json.loads(Path(args.request_file).read_text(encoding="utf8")))
            digest = hashlib.sha256(body.encode()).hexdigest()
            with db:
                old = db.execute("SELECT * FROM orders WHERE id=?", (args.order,)).fetchone()
                if old and (old["body"] != body or old["idem"] != args.idempotency_key):
                    raise SystemExit("Заказ уже сохранён с другим телом или ключом. Нужен новый заказ.")
                owner = db.execute("SELECT id FROM orders WHERE idem=?", (args.idempotency_key,)).fetchone()
                if owner and owner["id"] != args.order:
                    raise SystemExit("Ключ уже принадлежит другому заказу; задайте новый ключ.")
                if old is None:
                    db.execute("INSERT INTO orders VALUES (?,?,?,?,NULL,'prepared',NULL)",
                               (args.order, args.idempotency_key, body, digest))
            print("Сохранён неизменяемый заказ:", args.order)
            return
        row = db.execute("SELECT * FROM orders WHERE id=?", (args.order,)).fetchone()
        if row is None:
            raise SystemExit("Заказ не найден; сначала prepare")
        if args.command == "attach":
            if not re.fullmatch(r"media_generation_[A-Za-z0-9_-]+", args.job_id):
                raise SystemExit("Укажите настоящий ID общего задания Gate")
            if row["gate_id"] and row["gate_id"] != args.job_id:
                raise SystemExit("Этот заказ уже связан с другим заданием")
            with db:
                db.execute("UPDATE orders SET gate_id=? WHERE id=?", (args.job_id, args.order))
            print("Привязан ID:", args.job_id)
            return
        if not row["gate_id"]:
            raise SystemExit("Пока нет ID. Восстановите исходную отправку с сохранённым ключом.")
        import requests
        token = os.environ["CELESTRIA_API_KEY"]
        response = requests.get(
            "https://api.celestriagate.com/v1/media/generations/"
            + quote(row["gate_id"], safe=""),
            headers={"Authorization": f"Bearer {token}"}, timeout=(10, 30),
        )
        response.raise_for_status()
        job = response.json()
        if job.get("id") != row["gate_id"]:
            raise SystemExit("ID ответа не совпал с заказом")
        with db:
            db.execute("UPDATE orders SET status=?, snapshot=? WHERE id=? AND gate_id=?",
                       (job["status"], json.dumps(job, ensure_ascii=False), args.order, row["gate_id"]))
        print(json.dumps({"id": job["id"], "status": job["status"],
                          "provider": job.get("provider"), "billing": job.get("billing")},
                         ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

Порядок работы на одном заказе

Сначала подготовьте request.json по схеме нужной операции. Запустите prepare с собственным номером и ключом. Затем один раз выполните создание задания по базовой инструкции, используя сохранённое тело и тот же ключ. После ответа 202 привяжите настоящий ID. Обновление статуса делайте refresh столько раз, сколько нужно с разумными паузами.

В командах ниже первая строка пригодна для локальной подготовки. Во второй MEDIA_JOB_ID требуется заменить настоящим идентификатором вида media_generation_…. Пример намеренно не включает автоматическую платную отправку: её жизненный цикл уже разобран отдельно, а здесь проверяется надёжность хранения и восстановления.

Перед refresh установите requests в окружение проекта и задайте CELESTRIA_API_KEY. Если GET временно не ответил, сохранённый gate_id остаётся прежним. При следующем запуске снова обновляйте эту запись. Изменение сетевого состояния не является основанием сбрасывать поле и отправлять новую генерацию.

python orders.py prepare order-001 request.json order-001-v1
python orders.py attach order-001 MEDIA_JOB_ID
python orders.py refresh order-001

Как восстановиться после трёх разных сбоев

Если процесс завершился до отправки, у вас есть подготовленный заказ без ID. После проверки владения обработчик отправляет его с сохранённым ключом. Если завершился после отправки, но до сохранения ответа, исход неизвестен: повторите исходное тело с тем же ключом и сохраните возвращённое задание. Новый ключ в этом случае не нужен.

Если ID уже записан, восстановление проще: выполняйте GET. Не запускайте проверку качества через повторную генерацию, пока не получили старый результат. Если запрос с прежним ключом и изменённым телом даёт конфликт, верните исходное тело для старого заказа либо оформите новое самостоятельное намерение.

При обновлении нескольких заказов защищайте выбор задач от одновременной обработки разными процессами. Минимальная локальная таблица не реализует распределённую очередь. Добавьте короткую блокировку владения или используйте подходящую очередь вашего приложения; сетевые операции выполняйте вне длительной транзакции. Это проектное решение вашей системы, а не дополнительный контракт Gate.

Записывайте итог один раз, обновляйте снимок многократно

refresh обновляет одну существующую строку, поэтому повторное чтение completed не создаёт вторую запись расхода. Предварительную оценку и окончательное списание храните раздельно. У Gate cost_rub после подтверждения относится к конкретному заданию, reserved_rub — к его текущему резерву; их нельзя сложить как два независимых платежа.

Снимок awaiting_billing ещё не является окончательным отчётом. Сохраните статус и продолжите обновление позже. Для failed и cancelled сохраните error и billing, чтобы клиент видел наблюдаемую причину и учёт. Не заменяйте ответ универсальным сообщением о бесплатной ошибке или гарантированном возврате.

При работе с несколькими попытками у одного пользовательского заказа сделайте отдельную таблицу попыток. Каждая хранит свой ID и ключ; итог по заказу складывается из относящихся к нему расходов. Не переписывайте первую попытку вторым роликом: потеряется связь между оплатой, параметрами и полученным результатом.

Проверьте восстановление без расхода на модели

Для локальной приёмки достаточно временной базы и фиктивного ответа GET. Повтор prepare с тем же телом должен быть безвреден; изменение prompt при прежнем order ID должно остановиться. attach не должен заменять уже привязанный ID другим. Два refresh одного завершённого задания должны обновить одну строку, а не удвоить расход.

Отдельно остановите обработчик после сохранения тела и после ответа создания, но до записи ID, используя имитацию API. Убедитесь, что восстановление читает прежний ключ и тело. Реальная платная проверка нужна для художественного качества и внешнего учёта; она не должна быть единственным способом проверить локальную логику.

После успешных проверок добавьте понятную карточку заказа в существующий чат: статус, результат, фактическое списание и безопасное копирование запроса. Пользователь не обязан знать устройство SQLite. Ему важно вернуться позже и увидеть тот же заказ, а вашей системе — сохранить доказуемую связь между его действием и одной работой.

Частые вопросы

Эта утилита сама генерирует видео?

Нет. Она сохраняет заказ, привязывает ID и делает GET существующего задания. Платный POST подключается отдельно по базовой инструкции и использует сохранённое тело и ключ.

Почему недостаточно сохранить только ID Gate?

До получения ответа ID может быть неизвестен. Сохранённые тело и Idempotency-Key позволяют восстановить ту же отправку после сетевого сбоя.

Можно ли изменять request.json после prepare?

Для нового содержания создайте новый заказ и ключ. Попытка повторного prepare с другим телом для прежнего заказа отклоняется, чтобы не смешать параметры и оплату.

Подходит ли локальная SQLite для нескольких серверов?

Пример показывает постоянное состояние одного контролируемого окружения. Распределённая очередь, владение задачами и разделение пользователей требуют отдельной архитектуры приложения.

Источники

• Документация Celestria Gate
• Подтверждённые расходы fal
• Python: модуль sqlite3
• Python requests: HTTP-запросы и тайм-ауты