Как получить видео по API: статусы, ссылка результата и скачивание MP4

HTTP 202 — начало ожидания, а не готовое видео

После POST /v1/media/generations сервер возвращает объект media_generation, ID и Location. Сохраните их вместе с заказом. Этот ответ подтверждает асинхронную работу: показывать клиенту готовый ролик или окончательную цену на этом этапе рано. Повторная платная отправка ради быстрого результата не требуется.

GET /v1/media/generations/:id обновляет существующую задачу. Опрашивайте её с разумной паузой и конечной границей локального ожидания. Если программа закончила ждать, заказ остаётся на сервере. Позже продолжайте по тому же ID; закрытие терминала не является отменой работы провайдера.

Результаты разных моделей получаются тем же GET /v1/media/generations/:id. Сохраняйте общий ID из ответа создания вместе с вашим заказом и канонической моделью. Заголовок Location содержит путь проверки; вам не нужно выбирать отдельный API поставщика по имени модели. Различия формата файлов обрабатываются после получения result и files.

Дождитесь подтверждённого завершения

Пока статус queued, submitting, pending или in_progress, продолжайте наблюдать ту же работу. awaiting_billing означает ожидание подтверждения итогового расхода. Общий API не следует считать завершённым финансовым результатом до completed. У этой стадии может быть другое время ожидания, чем у самой генерации файла.

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

В интерфейсе достаточно показывать последний подтверждённый этап и возможность вернуться позже. Вымышленный процент готовности только мешает. После completed сохраните итоговый объект, включая provider и billing.cost_rub, чтобы пользователь мог связать полученный файл с фактическим списанием конкретного заказа.

Почему нельзя полагаться только на files.content_type

Массив files помогает собрать результаты, но content_type может быть null или application/octet-stream. Это не означает отсутствие видео. Разные модели иногда возвращают video.url без MIME, строковый video_url или дополнительные файлы. Структура исходного result сверяется с output_schema выбранной модели, а получение задания по /v1/media остаётся единым.

Ищите кандидата по нескольким признакам: video/ MIME, расширению URL, file_name и явному полю video в result. Расширение URL проверяется по пути без строки query: подписанные параметры не являются частью имени файла. Если метаданных недостаточно, не объявляйте любой первый URL видеороликом.

Даже несколько согласованных признаков дают кандидата, а не доказательство содержимого. Ссылка с окончанием .mp4 может вернуть ошибку HTML или другой файл. Поэтому после скачивания полезна проверка настоящих потоков и контейнера. Наличие аудио также проверяется по файлу, а не только по запрошенному флагу generate_audio.

Обновите ссылку, если она истекла

Часть результатов Gate выдаёт по подписанной ссылке с ограниченным временем действия. Не храните только URL: ID задания позволяет повторить GET завершённой работы и получить актуальный ответ. Старая ссылка в переписке и свежая ссылка из задания могут отличаться, хотя относятся к тому же файлу.

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

Ключ Gate нужен для авторизованного GET задания на API Gate. Его нельзя автоматически прикладывать к любому URL результата. Для подписанной выдачи и внешнего CDN используйте соответствующую ссылку без передачи секрета поставщику файлов. При редиректе проверяйте каждый следующий адрес, а не слепо пересылайте заголовки.

Пример загрузчика: только готовый файл и разрешённые адреса

Сохраните свежий завершённый ответ задания в job.json. Программа ниже выбирает видеокандидата по метаданным, проверяет HTTPS и разрешённый адрес, скачивает во временный файл и проверяет медиапоток через ffprobe. Она не содержит API-ключа и не выполняет платных генераций; создание и авторизованный GET выполняются отдельно.

Для запуска нужны Python, requests и установленный ffprobe из FFmpeg. Команда: python download_video.py job.json output.mp4. Локальный предел 200 МиБ — политика учебного загрузчика, не заявленный лимит Gate. Уже существующий выходной файл или .part не перезаписывается; используйте отдельное имя и один загрузчик на файл.

Список доверенных хостов включает fal.media с поддоменами и путь выдачи видео api.celestriagate.com. Если подтверждённая схема конкретной модели вернула другой CDN, сначала проверьте его, затем явно расширьте правила. Не отключайте проверку всего адреса ради удобства. Код не конвертирует WebM или MOV в MP4 и не обещает художественное качество.

import argparse
import json
import os
from pathlib import Path
import re
import shutil
import subprocess
from urllib.parse import urljoin, urlsplit

BASE = "https://api.celestriagate.com"
MAX_BYTES = 200 * 1024 * 1024  # Локальная политика примера, не лимит Gate.


def safe_url(value):
    if not isinstance(value, str) or len(value) > 16_384:
        return None
    try:
        parsed = urlsplit(urljoin(BASE, value))
        host = (parsed.hostname or "").lower()
        cdn = host == "fal.media" or host.endswith(".fal.media")
        gate = host == "api.celestriagate.com" and bool(
            re.fullmatch(r"/(?:api/)?v1/videos/[A-Za-z0-9_-]+/content", parsed.path)
        )
        if (parsed.scheme != "https" or parsed.username or parsed.password
                or parsed.fragment or parsed.port not in (None, 443) or not (cdn or gate)):
            return None
        return parsed.geturl()
    except ValueError:
        return None


def candidates(job):
    found = []

    def add(raw, context=False):
        item = {"url": raw} if isinstance(raw, str) else raw
        if not isinstance(item, dict):
            return
        url = safe_url(item.get("url"))
        if not url:
            return
        mime = str(item.get("content_type") or "").lower().split(";")[0]
        name = str(item.get("file_name") or "")
        extension = any(re.search(r"\.(mp4|webm|mov|m4v)
quot;, x, re.I)
                        for x in (name, urlsplit(url).path))
        if (context or mime.startswith("video/") or extension) and url not in found:
            found.append(url)
    for item in job.get("files", []):
        add(item)
    result = job.get("result")
    if isinstance(result, dict):
        add(result.get("video"), True)
        add(result.get("video_url"), True)
    return found


def download(url, destination):
    import requests
    owned_part = False
    part = destination.with_name(destination.name + ".part")
    if destination.exists() or part.exists():
        raise RuntimeError("Выходной файл или .part уже существует; выберите новое имя")
    try:
        for redirects in range(4):
            url = safe_url(url)
            if not url:
                raise RuntimeError("Адрес или хост не разрешён")
            # Подписанная ссылка/CDN: ключ Gate сюда не добавляется.
            with requests.get(url, stream=True, timeout=(10, 30), allow_redirects=False) as response:
                if response.status_code in {301, 302, 303, 307, 308}:
                    location = response.headers.get("Location")
                    if not location:
                        raise RuntimeError("Редирект без Location")
                    url = urljoin(url, location)
                    continue
                response.raise_for_status()
                mime = response.headers.get("Content-Type", "").split(";")[0].lower()
                if mime not in {"video/mp4", "application/octet-stream", ""}:
                    raise RuntimeError("Ответ не объявляет MP4; конвертация здесь не выполняется")
                total = 0
                with part.open("xb") as target:
                    owned_part = True
                    for chunk in response.iter_content(chunk_size=64 * 1024):
                        total += len(chunk)
                        if total > MAX_BYTES:
                            raise RuntimeError("Превышен локальный предел размера")
                        target.write(chunk)
                break
        else:
            raise RuntimeError("Слишком много редиректов")
        probe = subprocess.run(
            ["ffprobe", "-v", "error", "-protocol_whitelist", "file,pipe", "-show_streams", "-show_format", "-of", "json", str(part.resolve())],
            check=True, capture_output=True, text=True, timeout=20,
        )
        info = json.loads(probe.stdout)
        streams = [s for s in info.get("streams", []) if s.get("codec_type") == "video"
                   and not s.get("disposition", {}).get("attached_pic")]
        fmt = info.get("format", {})
        brand = str(fmt.get("tags", {}).get("major_brand", "")).strip()
        if not streams or "mp4" not in fmt.get("format_name", "").split(",") or brand == "qt":
            raise RuntimeError("Не подтверждено видео в MP4-совместимом контейнере")
        # Оба файла в одной папке. link атомарно откажет, если имя уже занято.
        # Существующий файл не заменяется даже при параллельном появлении.
        os.link(part, destination)
        part.unlink()
        owned_part = False
        print(json.dumps({"file": str(destination), "bytes": total,
                          "width": streams[0].get("width"), "height": streams[0].get("height"),
                          "duration": fmt.get("duration")}, ensure_ascii=False, indent=2))
    finally:
        if owned_part:
            part.unlink(missing_ok=True)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("job_file")
    parser.add_argument("output_file")
    args = parser.parse_args()
    if not shutil.which("ffprobe"):
        raise SystemExit("Для проверки файла нужен установленный ffprobe")
    job = json.loads(Path(args.job_file).read_text(encoding="utf8"))
    if job.get("status") != "completed":
        raise SystemExit("Сначала получите completed по ID задания")
    urls = candidates(job)
    if not urls:
        raise SystemExit("Видео не определено; проверьте output_schema и разрешённый CDN")
    os.umask(0o077)
    download(urls[0], Path(args.output_file))


if __name__ == "__main__":
    main()

Проверьте размер кадра, длительность и звуковые потоки

Успешное сохранение показывает, что выбранный файл читается как видео в MP4-совместимом контейнере. Это ещё не доказывает выполнение творческой задачи. Сверьте ширину, высоту и длительность с запросом и допустимыми условиями модели. Разница в один кадр и отсутствие заданного строгого размера требуют разных объяснений.

Для проверки звука посмотрите список потоков ffprobe. Если вы просили сохранить оригинальную дорожку, одного audio-потока недостаточно: сравните содержание на слух. Если заказывали генерацию речи, проверьте произношение и синхронизацию. Контейнер MP4 сам по себе не гарантирует ни речь, ни музыку.

Отдельно просмотрите начало, середину и конец ролика, затем весь файл. Для замены объекта проверьте фон, движение, перекрытия и моменты поворота. Контактная картинка из нескольких кадров удобна для сравнения, но не заменяет просмотр движения. Отчёт о результате должен различать техническую пригодность и наблюдаемое качество.

ffprobe -v error -show_streams -show_format -of json output.mp4

Храните файл дольше, чем живёт подписанный URL

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

Сохраните исходный ID Gate, provider, параметры, окончательный billing и контрольные свойства файла. При повторном получении результата не добавляйте новый расход в свою таблицу. Скачивание готового файла не является новой видеогенерацией, хотя ваше собственное хранилище и передача могут иметь отдельные инфраструктурные условия.

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

Памятка по четырём частым проблемам

Есть ID, но нет files: сначала посмотрите статус. До completed это ожидаемо; новый POST не нужен. Есть files, но MIME пустой: проверьте имя, URL и result по схеме. Есть ссылка, но она истекла: получите свежий ответ существующей задачи. Есть файл, но браузер его не показывает: проверьте контейнер, кодек и способ выдачи.

Для локальной приёмки загрузчика подставьте фиктивные HTTP-ответы: безопасный адрес, редирект на посторонний хост, HTML вместо видео, превышение размера и файл без видеопотока. Проверьте, что временный файл очищается только если его создал этот запуск. Эти проверки не требуют новой платной работы модели.

После этого протестируйте чтение уже существующего реального результата. Покажите пользователю сам файл, выбранного провайдера и подтверждённое списание по ID. Такое завершение важнее сообщения «API ответил успешно»: оно подтверждает, что ролик действительно можно получить, проверить и использовать.

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

Почему после 202 нет готового видео?

202 подтверждает асинхронное принятие. Нужно сохранить ID и получать его состояние до completed, используя GET соответствующего media или fal задания.

Пустой content_type означает ошибку?

Не обязательно. В результате модели MIME может отсутствовать. Проверяйте также URL, file_name и структуру result по output_schema, затем содержимое скачанного файла. Это не требует другого API или ручного выбора провайдера.

Нужно ли передавать API-ключ при скачивании с fal.media?

Не пересылайте ключ Gate внешнему CDN. Ключ используется для GET задания на API Gate; результат скачивается по выданной разрешённой ссылке.

Что делать с истёкшей ссылкой?

Повторно получить завершённое задание по сохранённому ID и использовать актуальный ответ. Создавать новый платный ролик только ради новой ссылки не нужно.

Источники

• Документация Celestria Gate
• Каталог Gate
• ffprobe: контейнеры и потоки
• requests: потоковая загрузка тела
• Python: разбор и проверка URL