Дмитрий Коваль
Разработка

AI-видео готово, но файл потерян: сохраняем результат на Python, а не временную ссылку

Провайдер вернул succeeded и URL на ролик. Интерфейс показал «Готово», а позже скачать результат уже не удалось. Разберу этот сценарий без платных вызовов модели: где заканчивается генерация, зачем нужен отдельный этап сохранения и как проверить запись файла на Python при обрыве потока.

succeeded у провайдера — не ready в вашем продукте

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

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

Временная ссылка — способ забрать результат, а не сам результат. Срок её жизни зависит от конкретного API и хранилища. Например, документация Amazon S3 прямо указывает: presigned URL может перестать работать раньше заданного срока, если истекли выпустившие его временные credentials. Это пример поведения S3, а не универсальный TTL для всех видеомоделей.

Я бы разделял как минимум такие состояния:

generation_succeeded → archiving → validating → ready

Сбой загрузки оставляет задачу в archive_retry или требует разбора. Сбой проверки файла — в validation_failed. Ни то ни другое само по себе не означает, что нужно снова запускать модель.

Что сохранять вместо одной ссылки

Для задачи нужны provider_task_id, владелец результата и время завершения генерации. После получения файла добавляются внутренний storage_key, число байтов, SHA-256 и результат проверки формата.

Ссылку провайдера можно временно хранить для повторного скачивания, но отдельно от публичной карточки результата и с ограниченным доступом. У S3 подписанные ссылки фактически являются bearer tokens: тот, у кого есть ссылка, получает соответствующий доступ. Поэтому полный URL с подписью не должен попадать в обычные логи и аналитику.

Пользователю лучше выдавать доступ к вашей копии через собственную проверку прав. Внутренний storage_key не должен превращаться в публичный доступ ко всем чужим файлам.

Минимальное ядро на Python: сначала .part, потом сохранённый объект

Ниже не универсальный загрузчик URL и не валидатор MP4. Функция принимает поток bytes от отдельного сетевого адаптера и сохраняет исходные данные в закрытый staging-каталог. Каталог заранее создаёт приложение; пользователь не выбирает путь.

Задачи этого слоя узкие: ограничить объём, не оставлять частичный файл под итоговым именем и посчитать хеш. Расширение .bin намеренное: мы ещё не доказали, что получили корректное видео.

import hashlib import os import tempfile import uuid from pathlib import Path def stage_bytes(chunks, root, *, max_bytes, expected_bytes=None): root = Path(root) # Private directory created by the application. if max_bytes <= 0: raise ValueError("invalid byte limit") if expected_bytes is not None and not 0 <= expected_bytes <= max_bytes: raise ValueError("invalid expected size") part = None size = 0 digest = hashlib.sha256() try: with tempfile.NamedTemporaryFile( dir=root, prefix=".part-", delete=False ) as out: part = Path(out.name) for block in chunks: if not isinstance(block, bytes): raise TypeError("expected bytes") size += len(block) if size > max_bytes: raise ValueError("too large") out.write(block) digest.update(block) if size == 0: raise ValueError("empty body") if expected_bytes is not None and size != expected_bytes: raise ValueError("size mismatch") out.flush() os.fsync(out.fileno()) target = root / (uuid.uuid4().hex + ".bin") os.replace(part, target) return {"path": target, "bytes": size, "sha256": digest.hexdigest()} finally: if part is not None: part.unlink(missing_ok=True)

Временный и итоговый файлы находятся в одном каталоге. По документации Python, os.replace может завершиться ошибкой между разными файловыми системами; при успешном переименовании операция атомарна. Здесь атомарность касается появления итогового имени, а не всей транзакции с БД и не вечной сохранности диска.

UUID даёт каждой попытке собственное имя. Это не бизнес-идемпотентность: параллельные воркеры всё ещё могут сохранить две копии. Какую из них закрепить за задачей, решает отдельная условная запись в БД.

expected_bytes имеет смысл только для той же последовательности байтов, которую отдаёт адаптер. Нельзя бездумно сопоставлять длину декодированного потока с Content-Length сжатого HTTP-тела. Если достоверный размер неизвестен, параметр остаётся None, а обязательная проверка формата выполняется следующим этапом.

Два теста, которые полезнее демонстрации счастливого пути

Сохраните функцию как artifact_stage.py, а следующий фрагмент — как test_artifact_stage.py. Тесты используют синтетические байты, не настоящее видео и не запрос к модели.

import tempfile import unittest from pathlib import Path from artifact_stage import stage_bytes class StageTests(unittest.TestCase): def test_disconnect_leaves_no_file(self): def broken(): yield b"first chunk" raise ConnectionError("synthetic disconnect") with tempfile.TemporaryDirectory() as directory: with self.assertRaises(ConnectionError): stage_bytes(broken(), directory, max_bytes=100) self.assertEqual(list(Path(directory).iterdir()), []) def test_short_body_is_not_accepted(self): with tempfile.TemporaryDirectory() as directory: with self.assertRaises(ValueError): stage_bytes( [b"abc"], directory, max_bytes=100, expected_bytes=5 ) self.assertEqual(list(Path(directory).iterdir()), []) if __name__ == "__main__": unittest.main()

Запуск: python -m unittest -v test_artifact_stage.py

Для ядра примера проверены девять локальных сценариев: успешная запись с известным SHA-256, пустое тело, превышение лимита, несовпадение размера, исключение посреди потока, неверный тип чанка, точное попадание в лимит, недопустимый ожидаемый размер и сохранность постороннего файла при ошибке. Все девять прошли. Это проверка поведения кода на синтетических данных, не тест качества или скорости видеомодели.

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

Что остаётся за пределами этих сорока строк

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

URL нельзя произвольно брать из пользовательского ввода. Нужны правила допустимых источников и сетевой egress-контроль; редиректы тоже требуют проверки. Одна проверка строки hostname не является полной защитой от SSRF.

Затем файл проверяет отдельный обработчик с лимитами CPU, памяти и времени. Для изображения важны реальное декодирование и размеры; для видео — контейнер, потоки и возможность прочитать необходимые кадры. Красивое расширение, Content-Type и даже корректный SHA-256 не доказывают, что внутри пригодный ролик.

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

Между файлом и БД всё равно есть окно сбоя

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

Практическая схема: сохранить закрытый объект, проверить его, затем условно закрепить storage_key за нужной задачей и владельцем. Обновление проходит только из ожидаемого состояния и с актуальной версией записи либо lease. Повторный воркер не должен перезаписать уже выбранный результат.

Непривязанные объекты разбирает сверка: часть можно восстановить в манифесте, часть — удалить после безопасного периода хранения. Код выше показывает только локальную запись. Для production-хранилища отдельно нужны его гарантии загрузки, проверки целостности, резервирования и восстановления.

Если ссылка уже не работает

Сначала проверяем, есть ли собственная копия и не потерялась ли связь с задачей. Затем — можно ли через API провайдера получить новую ссылку на тот же результат. Такая возможность зависит от контракта; обещать её для любого сервиса нельзя.

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

Считать полезно не только provider_success_rate, но и долю задач, дошедших до ready, возраст очереди archiving, число истёкших ссылок и объём непривязанных объектов. Стоимость генерации, которую пользователь так и не получил, — тоже часть стоимости полезного результата.

Вывод

Модель заканчивает работу раньше, чем ваш продукт выполняет обещание пользователю. Между ними нужен небольшой, но явный этап: получить байты, проверить их, сохранить свою копию и только затем показать ready.

Начать можно с одного воркера и staging-каталога. Главное — не выдавать сохранённую ссылку за сохранённое видео.

У вас завершением AI-задачи считается ответ провайдера или уже проверенный файл в собственном хранилище? И что происходит, если пользователь приходит за ним на следующий день?

Документация

Amazon S3: сроки действия presigned URL и обращение с ними как с bearer tokens — https://docs.aws.amazon.com/AmazonS3/latest/userguide/using-presigned-url.html

Python: поведение os.replace и ограничение файловой системы — https://docs.python.org/3/library/os.html#os.replace