Пример кода: генерация трека через API на Python


suno api удобно подключать к Python-скрипту как обычный HTTP API: авторизация передаётся через Bearer-ключ, задача запускается отдельным запросом, а результат можно получить после проверки статуса. В этой статье разберём именно такой сценарий: описание трека → запуск задачи → опрос → скачивание mp3.

Перед запуском нужен ключ из раздела API. Он показывается один раз после выдачи, поэтому его лучше сразу сохранить в переменной окружения. Стоимость операции «Создать музыку (music)» составляет 59 ₽ за один запуск, а при выпуске первого ключа на баланс начисляется 50 ₽.

Важное ограничение этой инструкции: в исходной справке не приведена точная JSON-схема тела POST /generate. Поэтому ниже HTTP-часть, обработка задач, таймауты и скачивание файла сделаны полностью конкретно, а само тело генерации вынесено в переменную. Это позволяет не придумывать названия полей, которых нет в предоставленной документации: фактическое тело нужно взять из схемы операции, доступной для вашего ключа.

1) Что делает скрипт и что нужно перед запуском

Скрипт принимает текстовое описание будущего трека, отправляет его на маршрут POST /generate, получает идентификатор задачи и затем обращается к GET /tasks/{id}. Пока задача не готова, Python делает паузу и повторяет запрос. После получения результата программа скачивает готовый файл и проверяет его размер.

Базовый адрес API — https://genius-bot.ru/wp-json/genius/v1. Авторизация выполняется заголовком Authorization: Bearer <ключ>. Для Python достаточно библиотеки requests, а ключ удобно передать через переменную окружения.

pip install requests

Например, в Linux или macOS ключ можно задать так:

export GENIUS_API_KEY="ваш_ключ"

В Windows PowerShell:

$env:GENIUS_API_KEY="ваш_ключ"

Для операции музыки цена составляет 59 ₽ за запуск. Остаток можно проверить отдельным запросом к GET /balance, а список операций и цен получить через GET /services. Это полезно сделать перед постановкой пачки задач, чтобы не запускать обработку, которая заведомо остановится из-за недостатка средств.

Операция Цена за запуск
Создать музыку (music) 59 ₽
Звук по описанию (sfx) 9 ₽
Расшифровка записи (stt) 10 ₽
Оживить фото (photo-video) 25 ₽
Видео по описанию (video) 119 ₽

Цены в этой таблице относятся к одному запуску. Для музыки это означает 59 ₽ за создание одной задачи, а не абонентскую плату. У сервиса нет абонплаты и минимального платежа.

Скрипт генерации музыки на Python: этапы работы
Скрипт генерации музыки на Python: этапы работы

2) Запуск задачи: тело запроса и что в нём обязательно

Точка входа для запуска операции — POST /generate. Из предоставленной справки точно известны адрес маршрута и способ авторизации, но не перечислены конкретные JSON-поля тела запроса. Поэтому в рабочем коде разумно не зашивать выдуманную схему вроде prompt или service, если её нет в вашей документации.

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

import json
import os
import time
from pathlib import Path

import requests

BASE_URL = "https://genius-bot.ru/wp-json/genius/v1"
API_KEY = os.environ["GENIUS_API_KEY"]

HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}

DESCRIPTION = "Инструментальный электронный трек, спокойное вступление и энергичная середина"

# Здесь должен находиться JSON тела POST /generate,
# соответствующий операции music в документации API.
# Поля намеренно не выдумываются: точная схема generate
# не приведена в исходной справке.
GENERATE_BODY = {
    # вставьте документированные поля операции music
}

response = requests.post(
    f"{BASE_URL}/generate",
    headers=HEADERS,
    json=GENERATE_BODY,
    timeout=30,
)

response.raise_for_status()
data = response.json()

print(data)

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

Тот же HTTP-вызов через curl можно проверить отдельно:

curl -X POST \
  "https://genius-bot.ru/wp-json/genius/v1/generate" \
  -H "Authorization: Bearer $GENIUS_API_KEY" \
  -H "Content-Type: application/json" \
  -d '<точное JSON-тело операции music>'

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

Читать  Бесплатный API нейросети для учёбы и пет-проектов

Если API возвращает ошибку HTTP, её нужно обрабатывать до попытки читать идентификатор задачи. Например, raise_for_status() остановит выполнение при ответах 4xx и 5xx, после чего приложение может классифицировать ошибку по коду HTTP и содержимому ответа.

3) Опрос статуса без блокировки: цикл с ограничением по времени

После запуска задача не должна считаться готовой только потому, что сервер принял запрос. Для проверки состояния используется GET /tasks/{id}. Самый простой вариант — периодический опрос с паузой, но обязательно с общим таймаутом, иначе зависшая задача может оставить процесс работающим бесконечно.

Ниже отдельная функция принимает уже полученный task_id. В ней используется интервал 5 секунд и общий предел 10 минут. Эти числа являются параметрами клиентского скрипта, а не заявленными ограничениями API.

def wait_for_task(task_id, timeout_seconds=600, poll_seconds=5):
    deadline = time.monotonic() + timeout_seconds

    while time.monotonic() < deadline:
        response = requests.get(
            f"{BASE_URL}/tasks/{task_id}",
            headers=HEADERS,
            timeout=30,
        )

        response.raise_for_status()
        task = response.json()

        print("Статус:", task)

        # Здесь нужно использовать фактическое поле статуса
        # из ответа /tasks/{id}.
        status = task.get("status")

        if status in {"completed", "failed", "error"}:
            return task

        time.sleep(poll_seconds)

    raise TimeoutError(
        f"Задача {task_id} не завершилась за {timeout_seconds} секунд"
    )

Если в вашем ответе используются другие значения статуса, набор completed, failed и error нужно заменить на реальные значения API. То же относится к полю, в котором возвращается URL готового mp3.

Опрос не блокирует серверную задачу: между запросами Python спит, а сама операция продолжает выполняться на стороне API. Для приложения с большим количеством задач такой подход можно заменить очередью и отдельными воркерами, не меняя сам принцип работы с /tasks/{id}.

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

Обработка ошибок при запросе к API музыки
Обработка ошибок при запросе к API музыки

4) Скачивание готового файла и проверка, что он не пустой

После завершения задачи из ответа /tasks/{id} нужно получить ссылку на результат и скачать её обычным HTTP-запросом. Точная структура результата также не указана в исходной справке, поэтому URL файла в коде следует извлечь согласно фактическому ответу вашего endpoint.

Саму проверку скачанного файла можно сделать независимо от формата ответа задачи. Для mp3 недостаточно проверить только HTTP-код: сервер теоретически может вернуть успешный ответ с пустым содержимым. Поэтому файл сохраняется во временный путь, после чего проверяется размер.

def download_file(file_url, output_path):
    response = requests.get(file_url, stream=True, timeout=60)
    response.raise_for_status()

    output = Path(output_path)
    temp = output.with_suffix(output.suffix + ".part")

    total = 0

    with temp.open("wb") as f:
        for chunk in response.iter_content(chunk_size=1024 * 64):
            if not chunk:
                continue

            f.write(chunk)
            total += len(chunk)

    if total == 0:
        temp.unlink(missing_ok=True)
        raise ValueError("API вернул пустой файл")

    temp.replace(output)

    return output

Использование временного расширения .part важно для приложения: если процесс оборвётся во время скачивания, рядом не останется файла, который выглядит как готовый mp3. После полной записи временный файл переименовывается в конечный.

Если сервер вернул ошибку при скачивании, raise_for_status() остановит операцию. Для production-кода можно добавить повтор именно на сетевые ошибки, но не следует бесконечно повторять скачивание при постоянном ответе 4xx.

5) Ошибки: недостаточно средств, отказ модели, обрыв связи

Ошибки удобно разделить на три группы. Первая возникает до нормального выполнения операции: например, на балансе недостаточно средств для запуска. Вторая относится к самой задаче: API принял запрос, но операция завершилась отказом. Третья связана с сетью, когда клиент не получил ответ вовремя или соединение оборвалось.

Перед массовым запуском можно проверить GET /balance. Это не заменяет обработку ошибки непосредственно при POST /generate, потому что между проверкой баланса и запуском может произойти другой расход, но позволяет не отправлять заведомо бессмысленные запросы.

def get_balance():
    response = requests.get(
        f"{BASE_URL}/balance",
        headers=HEADERS,
        timeout=15,
    )
    response.raise_for_status()
    return response.json()

def start_generation(body):
    try:
        response = requests.post(
            f"{BASE_URL}/generate",
            headers=HEADERS,
            json=body,
            timeout=30,
        )
    except requests.RequestException as exc:
        raise RuntimeError(
            "Не удалось связаться с API при запуске задачи"
        ) from exc

    if response.status_code >= 400:
        raise RuntimeError(
            f"API отклонил запуск: HTTP {response.status_code}; "
            f"ответ: {response.text[:1000]}"
        )

    return response.json()

При недостатке средств приложение не должно автоматически повторять POST /generate: повторный запуск не исправляет состояние баланса. Для отказа самой операции нужно получить финальный ответ через /tasks/{id}\` и записать причину в журнал, если она присутствует в ответе.

Сетевой обрыв обрабатывается иначе. Если запрос к /generate завершился таймаутом, клиент не всегда знает, дошёл ли запрос до сервера. Безопасный повтор такого POST нельзя считать гарантированно идемпотентным по данным из предоставленной справки. Поэтому приложение не должно автоматически запускать вторую платную операцию только потому, что первый HTTP-запрос не получил ответ.

Для GET-запросов ситуация проще: /balance и /tasks/{id} можно повторить после временной сетевой ошибки. Для них имеет смысл использовать несколько попыток с увеличивающейся задержкой, например 1, 2, 4 и 8 секунд, при этом общий таймаут остаётся ограниченным.

def get_with_retry(url, attempts=4):
    delays = [1, 2, 4, 8]

    for attempt in range(attempts):
        try:
            response = requests.get(
                url,
                headers=HEADERS,
                timeout=30,
            )
            response.raise_for_status()
            return response

        except requests.RequestException:
            if attempt == attempts - 1:
                raise

            time.sleep(delays[attempt])

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

Сохранение готового трека в MP3
Сохранение готового трека в MP3

6) Как встроить скрипт в очередь задач приложения

В веб-приложении генерацию музыки лучше не выполнять внутри HTTP-запроса пользователя от начала до конца. Клиенту можно сразу вернуть внутренний идентификатор задания приложения, а отдельный worker выполнит платный запуск, сохранит внешний task_id, будет опрашивать /tasks/{id} и после завершения положит mp3 в хранилище приложения.

Минимальная модель данных может содержать пять полей: id, status, external_task_id, output_path и error. Например, после постановки задания статус становится queued, после успешного POST /generate — running, после получения готового файла — completed.

def process_job(job):
    job.status = "running"

    try:
        result = start_generation(job.generate_body)

        # Извлеките task_id из фактического ответа API.
        job.external_task_id = result["<поле с id задачи>"]
        save_job(job)

        task = wait_for_task(job.external_task_id)

        # Извлеките URL результата из фактического ответа /tasks/{id}.
        file_url = task["<поле с URL результата>"]

        output = download_file(
            file_url,
            f"music/{job.id}.mp3",
        )

        job.output_path = str(output)
        job.status = "completed"

    except TimeoutError as exc:
        job.status = "timeout"
        job.error = str(exc)

    except Exception as exc:
        job.status = "failed"
        job.error = str(exc)

    finally:
        save_job(job)

Здесь два обращения к структуре ответа намеренно обозначены как placeholders. Без фактической схемы POST /generate и GET /tasks/{id} нельзя корректно назвать JSON-поля для идентификатора и URL mp3. Остальная логика не зависит от их названий.

Для больших очередей полезно вынести интервал опроса и общий лимитер запросов в один объект. Например, при 10 задачах не следует запускать 10 независимых циклов, каждый из которых делает запрос каждые 5 секунд: суммарный поток запросов нужно считать для всего ключа, поскольку ограничение 60 запросов в минуту относится к ключу.

Ещё один вариант — webhook. В справке предусмотрен необязательный параметр callback_url: когда задача готова, API отправляет на этот адрес POST. Тогда приложение может не держать worker в цикле опроса, а принимать уведомление и уже после него получать или сохранять результат в соответствии с фактической структурой callback.

Для небольшого приложения polling проще отлаживать: состояние видно через GET /tasks/{id}, а общий таймаут ограничивает продолжительность обработки. Для очереди с большим числом параллельных задач webhook позволяет убрать регулярные проверки готовности, но обработчик всё равно должен проверять входящие данные и корректно обрабатывать повторные уведомления.

Таким образом, api музыки код сводится к нескольким независимым этапам: авторизация, запуск /generate, получение ID задачи, контроль состояния через /tasks/{id}, скачивание результата и фиксация ошибки. Самое важное место, которое нельзя заполнять по догадке, — JSON-схема конкретной операции music.

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

Сколько стоит один запуск генерации музыки?

Операция «Создать музыку (music)» стоит 59 ₽ за один запуск. Оплата взимается за запуск, абонентской платы и минимального платежа нет.

Где взять ключ для suno api?

Ключ выдаётся после входа в разделе API. Он показывается один раз, поэтому его следует сохранить сразу после получения.

Можно ли получить готовый mp3 без постоянного опроса?

Да, в API предусмотрен необязательный параметр callback_url. После готовности задачи на указанный адрес приходит POST. Другой вариант — периодически проверять GET /tasks/{id}.

Какой таймаут поставить для Python?

В примере HTTP-запросы имеют сетевые таймауты от 15 до 60 секунд, а ожидание конкретной задачи ограничено 600 секундами. Эти значения относятся к клиентскому коду и могут быть изменены под очередь приложения.

Что делать при обрыве связи после POST /generate?

Не следует автоматически повторять платный POST, если неизвестно, был ли первый запрос принят сервером. Иначе можно создать две операции. Для повторов безопаснее использовать GET-запросы проверки состояния после того, как идентификатор задачи уже известен.

Где проверить доступные операции и цены?

Для этого предусмотрен маршрут GET /services. Документацию и данные для работы с ключом можно открыть в разделе API.