Genius-bot · 12 минут чтения
Разберём, как подключить api нейросети к Python-проекту без хранения секрета в исходниках и без лишней обвязки. На практике достаточно подготовить окружение, получить ключ, передать его в заголовке Authorization и отправить HTTP-запрос. Для начала ключ можно выпустить в разделе API: после входа он показывается один раз.
Ниже разберём полный цикл: переменная окружения, запрос через requests, получение ответа, ожидание долгой задачи, сохранение результата и обработка временных ошибок. Отдельно покажу структуру готового Python-модуля, который можно положить в проект и адаптировать под нужную операцию.
1) Что подготовить перед первой строкой кода
API доступен по базовому адресу https://genius-bot.ru/wp-json/genius/v1. Авторизация выполняется заголовком Authorization: Bearer <ключ>. Доступны маршруты для получения списка операций и цен, проверки баланса, загрузки файлов, запуска операции и проверки состояния задачи.
Для Python достаточно HTTP-клиента requests. Если он ещё не установлен, добавьте его в окружение проекта:
pip install requestsПеред первым запросом полезно проверить, какие операции доступны и сколько они стоят. Для этого предусмотрен GET /services. Остаток проверяется через GET /balance. Эти запросы не требуют знания параметров конкретной генерации и подходят для первоначальной проверки подключения.
curl -H "Authorization: Bearer $GENIUS_API_KEY" \
https://genius-bot.ru/wp-json/genius/v1/servicesОплата выполняется за запуск, без абонентской платы и минимального платежа. При выпуске первого ключа на баланс начисляется 50 ₽ для пробы. Для одного ключа действует ограничение не более 60 запросов в минуту, поэтому при пакетной обработке это ограничение нужно учитывать на уровне приложения.
| Операция | Стоимость |
|---|---|
| Картинка по описанию | 9 ₽ |
| Изменить фото по описанию | 35 ₽ |
| Увеличить качество фото | 50 ₽ |
| Оживить фото | 25 ₽ |
| Видео по описанию | 119 ₽ |
| Расшифровка записи | 10 ₽ |
| Озвучка текста | 18 ₽ за 1000 знаков |
Остальные операции имеют отдельные цены: музыка — 59 ₽, звук из видео — 0 ₽, говорящий аватар — 120 ₽, удаление вокала — 45 ₽, удаление шума — 36 ₽, звук по описанию — 9 ₽. Цены указаны за один запуск, кроме озвучки текста, где стоимость указана за 1000 знаков.
2) Ключ в переменной окружения, а не в коде
Ключ не стоит записывать непосредственно в Python-файл. В таком виде секрет легко попадёт в Git, архив проекта или логи. Вместо этого сохраним его в переменной окружения GENIUS_API_KEY и будем читать через стандартный модуль os.
В Linux или macOS переменную можно установить так:
export GENIUS_API_KEY="ваш_ключ"В Windows PowerShell используется другой синтаксис:
$env:GENIUS_API_KEY="ваш_ключ"В Python получение ключа выглядит так:
import os
API_KEY = os.environ["GENIUS_API_KEY"]Если переменной нет, программа завершится сразу с понятной ошибкой, а не после отправки нескольких запросов. Для приложения это удобнее, чем держать ключ в константе рядом с URL API.
Общий заголовок для запросов можно сформировать один раз:
HEADERS = {
"Authorization": f"Bearer {API_KEY}",
"Accept": "application/json",
}Такой подход подходит и когда нужно подключить нейросеть python-кодом в небольшом скрипте, и когда HTTP-клиент является отдельным модулем большого приложения.
3) Запрос и разбор ответа
Самый простой способ проверить соединение — запросить список операций. В requests для этого используется обычный GET. Таймаут лучше указывать явно: приложение не должно бесконечно ждать ответа от сети.
import os
import requests
BASE_URL = "https://genius-bot.ru/wp-json/genius/v1"
headers = {
"Authorization": f"Bearer {os.environ['GENIUS_API_KEY']}",
"Accept": "application/json",
}
response = requests.get(
f"{BASE_URL}/services",
headers=headers,
timeout=30,
)
response.raise_for_status()
data = response.json()
print(data)Это базовый пример API нейросети на Python: запрос отправляется через requests, HTTP-ошибка превращается в исключение методом raise_for_status(), а JSON разбирается через response.json(). Такой шаблон можно использовать для /services и /balance.
Для запуска конкретной операции используется POST /generate. Здесь важно не придумывать формат параметров: конкретные поля запроса зависят от выбранной операции, поэтому перед формированием payload нужно посмотреть соответствующую операцию в ответе /services и передать предусмотренные ею параметры.
Для операций с файлами предусмотрен отдельный POST /uploads. Он используется для загрузки файла и получения ссылки, которую затем можно передать в соответствующую операцию. Само наличие этого маршрута позволяет разделить загрузку исходного файла и запуск задачи.
После запуска задачи используется GET /tasks/{id}. Этот маршрут возвращает состояние задачи и результат. Поэтому приложение не должно считать HTTP-ответ самого POST /generate готовым файлом: для длительных операций предусмотрен отдельный этап проверки состояния.
Есть и другой вариант для приложения, которому не подходит постоянный опрос. В запрос можно передать необязательный параметр callback_url. Когда задача будет готова, на этот адрес придёт POST. Это позволяет вынести ожидание результата из рабочего процесса Python-программы.
4) Ожидание долгой задачи без зависания
Длительная операция не должна блокировать основной поток на неопределённое время. Практический вариант — периодически проверять GET /tasks/{id} с ограничением общего времени ожидания. Частоту опроса также нужно учитывать вместе с лимитом 60 запросов в минуту на ключ.
Поскольку справка API определяет маршрут и назначение ответа, но не фиксирует здесь конкретные имена JSON-полей идентификатора и статуса, в готовом коде эти поля лучше вынести в небольшую функцию-адаптер. После первого запуска достаточно посмотреть фактический JSON ответа /generate и настроить её под возвращаемую структуру.
import time
import requests
def wait_for_task(
task_id,
headers,
base_url,
timeout_seconds=600,
interval_seconds=5,
):
started = time.monotonic()
while time.monotonic() - started < timeout_seconds:
response = requests.get(
f"{base_url}/tasks/{task_id}",
headers=headers,
timeout=30,
)
response.raise_for_status()
data = response.json()
# Сохраняем весь ответ, чтобы адаптер можно было
# настроить под фактическую структуру API.
if task_is_ready(data):
return data
time.sleep(interval_seconds)
raise TimeoutError(
f"Задача {task_id} не завершилась за "
f"{timeout_seconds} секунд"
)Функция task_is_ready() здесь намеренно оставлена адаптером, а не заменена выдуманным названием поля. Это важная граница между общим HTTP-шаблоном и конкретной схемой ответа: если документация не задаёт имя поля, его нельзя надёжно угадать по одному URL.
Для webhook логика другая. Вместо цикла с sleep() приложение поднимает endpoint, принимает входящий POST и обрабатывает уведомление о готовности задачи. В этом сценарии параметр callback_url передаётся при запуске операции.
Для файлов результат после получения ответа можно сохранить обычным бинарным записью. Если API возвращает ссылку на готовый файл, её можно скачать отдельным HTTP-запросом:
def save_file(url, path, headers=None):
response = requests.get(
url,
headers=headers,
timeout=120,
stream=True,
)
response.raise_for_status()
with open(path, "wb") as file:
for chunk in response.iter_content(chunk_size=1024 * 1024):
if chunk:
file.write(chunk)Здесь используется потоковая запись блоками по 1 МБ, поэтому файл не нужно целиком держать в оперативной памяти. Перед использованием функции следует передать именно URL результата, который вернул статус задачи.
5) Обработка ошибок и повторы
Ошибки HTTP, проблемы соединения и превышение времени ожидания лучше разделять. requests.RequestException покрывает сетевые ошибки клиента, а raise_for_status() позволяет отдельно обработать HTTP-ответы с ошибочными статусами.
Повторять любой запрос подряд небезопасно: POST /generate запускает операцию, поэтому автоматический повтор может привести к дополнительному запуску и дополнительному списанию. Для повторов прежде всего подходят временные сетевые сбои и чтение состояния уже созданной задачи, когда приложение точно знает её идентификатор.
import time
import requests
def get_json_with_retries(url, headers, attempts=3):
last_error = None
for attempt in range(attempts):
try:
response = requests.get(
url,
headers=headers,
timeout=30,
)
response.raise_for_status()
return response.json()
except requests.RequestException as error:
last_error = error
if attempt + 1 == attempts:
raise
time.sleep(2 ** attempt)
raise last_errorЗадержка здесь составляет 1, затем 2 секунды перед последующими попытками. Это пример стратегии для GET-запроса, а не универсальное правило для всех операций API. Для запуска платной операции безопаснее сначала проверить, не была ли задача уже создана, вместо безусловной повторной отправки POST.
При обработке ошибок полезно логировать HTTP-статус, URL маршрута и техническое сообщение, но не сам заголовок Authorization. Ключ не должен попадать в консоль, текст исключения или систему централизованных логов.
Отдельно проверяйте баланс через GET /balance, если приложение запускает много операций. Это позволяет получать актуальное состояние счёта до очередной серии запросов. При этом приложение должно учитывать стоимость выбранной операции и ограничение частоты запросов.
6) Готовый модуль целиком
Ниже — каркас модуля, который можно положить в проект. Он закрывает общую HTTP-часть: авторизацию, GET-запросы, запуск операции, получение состояния, таймаут ожидания и скачивание результата. Вызов generate() принимает готовый словарь параметров, поэтому конкретный payload можно собрать по данным операции из /services, не зашивая неподтверждённую схему в универсальный клиент.
import os
import time
from pathlib import Path
import requests
class GeniusAPI:
BASE_URL = "https://genius-bot.ru/wp-json/genius/v1"
def __init__(self, api_key=None, timeout=30):
self.api_key = api_key or os.environ["GENIUS_API_KEY"]
self.timeout = timeout
self.headers = {
"Authorization": f"Bearer {self.api_key}",
"Accept": "application/json",
}
def services(self):
return self._get_json("/services")
def balance(self):
return self._get_json("/balance")
def upload(self, url, **kwargs):
response = requests.post(
url,
headers=self.headers,
timeout=self.timeout,
**kwargs,
)
response.raise_for_status()
return response.json()
def generate(self, payload):
response = requests.post(
f"{self.BASE_URL}/generate",
headers={
**self.headers,
"Content-Type": "application/json",
},
json=payload,
timeout=self.timeout,
)
response.raise_for_status()
return response.json()
def task(self, task_id):
return self._get_json(f"/tasks/{task_id}")
def _get_json(self, path):
response = requests.get(
f"{self.BASE_URL}{path}",
headers=self.headers,
timeout=self.timeout,
)
response.raise_for_status()
return response.json()
def download(self, url, destination):
destination = Path(destination)
response = requests.get(
url,
headers=self.headers,
timeout=120,
stream=True,
)
response.raise_for_status()
with destination.open("wb") as file:
for chunk in response.iter_content(
chunk_size=1024 * 1024
):
if chunk:
file.write(chunk)
return destination
def wait(self, task_id, is_ready, timeout=600, interval=5):
started = time.monotonic()
while time.monotonic() - started < timeout:
data = self.task(task_id)
if is_ready(data):
return data
time.sleep(interval)
raise TimeoutError(
f"Task {task_id} did not finish in {timeout} seconds"
)Пример использования выглядит так:
from genius_api import GeniusAPI
api = GeniusAPI()
print(api.services())
print(api.balance())
# Payload формируется согласно параметрам
# выбранной операции из /services.
payload = {
# параметры конкретной операции
}
created = api.generate(payload)
print(created)Последний пример специально не содержит выдуманных названий параметров операции. Если нужен, например, запуск обработки изображения, сначала получают описание доступных операций через /services, затем формируют JSON с теми полями, которые предусмотрены конкретной операцией.
Есть и совместимый с OpenAI формат API. Для чата предусмотрены POST /chat/completions и GET /models, причём они работают в том же формате, что и OpenAI. В клиентской библиотеке достаточно изменить base_url и ключ. Тариф составляет 40 ₽ за 1 млн токенов запроса и 400 ₽ за 1 млн токенов ответа.
Если задача — именно подключить нейросеть python-кодом для чата, отдельный HTTP-клиент может быть ещё проще. Например, через requests запрос к совместимому маршруту выглядит следующим образом:
import os
import requests
base_url = "https://genius-bot.ru/wp-json/genius/v1"
headers = {
"Authorization": f"Bearer {os.environ['GENIUS_API_KEY']}",
"Content-Type": "application/json",
}
payload = {
"model": "модель_из_ответа_/models",
"messages": [
{
"role": "user",
"content": "Коротко объясни, что делает этот API."
}
]
}
response = requests.post(
f"{base_url}/chat/completions",
headers=headers,
json=payload,
timeout=60,
)
response.raise_for_status()
print(response.json())В этом примере название модели намеренно не задано конкретным значением: список доступных моделей возвращает GET /models. Такой подход позволяет не зашивать в код значение, которого нет в предоставленной справке.
Таким образом, схема интеграции получается последовательной: ключ хранится в окружении, авторизация передаётся через Bearer-заголовок, простые запросы выполняются синхронно, длительные операции отслеживаются через /tasks/{id}, а для webhook можно передать callback_url. Если работа связана с файлами, сначала используется /uploads, после чего ссылка на файл участвует в соответствующей операции.
Когда нужен именно requests к api нейросети, отдельная библиотека-обёртка не обязательна: маршруты являются обычными HTTP endpoints, а requests закрывает авторизацию, JSON, таймауты и загрузку результата. Для чата можно использовать совместимый формат /chat/completions, а для остальных операций — маршруты /services, /uploads, /generate и /tasks/{id}.
Частые вопросы
Где получить API-ключ?
Ключ выдаётся после входа в разделе API. Он показывается один раз, поэтому его нужно сохранить и передать приложению через переменную окружения.
Можно ли хранить ключ прямо в Python-файле?
Технически HTTP-клиенту всё равно, откуда получен ключ, но для проекта безопаснее использовать переменную окружения. Тогда секрет не требуется помещать в исходный код и репозиторий.
Как понять, завершилась ли длительная операция?
После запуска используется маршрут GET /tasks/{id}, который возвращает состояние и результат. Альтернативой периодическому опросу служит необязательный callback_url: после готовности задачи на него приходит POST.
Есть ли ограничение на количество запросов?
Да. Для одного ключа разрешено не более 60 запросов в минуту. При реализации polling и пакетной обработки это ограничение нужно учитывать отдельно от времени ожидания задачи.
Как загрузить исходный файл?
Для загрузки предусмотрен POST /uploads. Он принимает файл и возвращает ссылку, которую затем можно использовать при запуске соответствующей операции.
Можно ли использовать совместимый с OpenAI формат?
Да. Для чата доступны POST /chat/completions и GET /models в том же формате. В клиентской библиотеке достаточно изменить base URL и ключ; тариф для чата составляет 40 ₽ за миллион токенов запроса и 400 ₽ за миллион токенов ответа.
Документация и выпуск ключа доступны в разделе API сервиса. Перед подключением конкретной операции проверьте её параметры через /services, чтобы не переносить в код предположения о формате запроса.
