AIOgram3 16. Перевод голосовых сообщений в текст

Голосовые сообщения – весьма спорная тема. Кто-то без них жить не может, а кто-то ненавидит их всей душой. Если опустить прения на эту тему, то у голосовых сообщений есть одна проблема: их банально не всегда удобно слушать. Чаще всего голосовые сообщения – это "информация в моменте", и, когда появляется возможность прослушать их, они теряют свою актуальность.

У нас в чате любят слать голосовые, и, чтобы все могли оставаться "в теме", даже когда не у каждого есть возможность прослушать их, мы решили научить бота переводить голосовые в текст.

Нейронных сетей, заточенных под распознавание голоса, свыше десятка, но они требуют серьёзных вычислительных мощностей, чего мой не самый мощный сервер не может предоставить.
Готовые сервисы тоже есть, но практически все - платные. Не беря в расчёт "пробные периоды" из бесплатных (вернее с бесплатным лимитом в месяц) есть два сервиса:

• Google Cloud - думаю не стоит рассказывать о Google. Они предлагают бесплатный тариф - 60 минут в месяц.
• SpeechFlow - сервис, позиционирующий себя как "лидера рынка". Поддерживают транскрипцию с 14-ти языков. Предлагают бесплатный тариф: 30 минут онлайн-распознавания (на сайте) и пять часов по API в месяц.

Сервис от Google нам не подходит: 60 минут на целый месяц слишком мало, поэтому выбор был сделан в пользу SpeechFlow. Пять часов в месяц – тоже не то чтобы много, но с этим уже можно работать. Ну, и будет интересно посмотреть статистику длительности голосовых и их количества.

Регистрация и получение API-ключа.

Переходим на сайт сервиса: https://speechflow.io/ru/

Регистрация максимально проста, особенно если заходить через Google-аккаунт.

После регистрации попадаем в личный кабинет. Там в левой панели выбираем раздел "API".

На открывшейся странице нажимаем кнопку "Сгенерировать ключ API".

В появившемся окне будет две строки:

• KeyId - идентификатор ключа
• KeySecret - секретный ключ

ВНИМАНИЕ! Ключ отображается всего один раз! Сохраните его в удобном для вас месте.

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

Отслеживание и расшифровка голосовых сообщений.

В посте "AIOgram3 14. Фильтруем запрещённые слова" мы писали обработку для проверки сообщений на запрещённые слова. В этом посте мы дополним ее функционалом проверки голосовых сообщений.

Но сперва добавим ключи в класс Secrets.

Откроем файл settings.py и в классе Secrets добавим две строки:

audio_key_id: str = "ваш_KeyId"  
audio_key_secret: str = "ваш_KeySecret"

В значение полей вставьте полученные от SpeechFlow данные доступа.

Обработка голосовых сообщений.

Откройте файл filter_words.py.

Тут у нас уже есть функция check_message с условием if message.text. Для того, чтобы "отлавливать" голосовые сообщения, нужно другое условие - if message.voice.

Добавим его ниже, использовав elif.

Сперва нам необходимо скачать голосовое сообщение с сервера Telegram. Для этого выполним несколько шагов:

1. Создадим переменную file_id, в которую получим идентификатор голосового сообщения.
2. Создадим переменную file, в которую асинхронно получим информацию о файле голосового сообщения.
3. Создадим переменную file_path, в которую получим путь до файла.
4. Создадим переменную file_name, в которой укажем путь и имя сохраняемого файла на нашем сервере. Убедитесь, что директория, в которой вы собираетесь сохранять файл, существует.
5. Асинхронно выполним команду скачивания файла.

Получается вот такой блок кода:

file_id = message.voice.file_id  
file = await bot.get_file(file_id)  
file_path = file.file_path  
file_name = f"files/audio{file_id}.mp3"  
await bot.download_file(file_path, file_name)

Создаём четыре переменные, которые понадобятся далее в коде:

1. headers – в этой переменной создаём словарь с ключами keyId и keySecret, в которые передаём значения соответствующих полей из класса Secrets.
2. create_url – в этой переменной определяем URL-адрес, на который отправляем файл для транскрипции. Обратите внимание на query-параметр lang: в нём указываем язык голосового сообщения.
3. query_url – в этой переменной определяем URL-адрес, на который будем обращаться для получения результата транскрипции.
4. files – в этой переменной создаём словарь с ключом file, в значение которого помещаем открытый аудиофайл.

Блок кода:

headers = {"keyId": Secrets.audio_key_id, "keySecret": Secrets.audio_key_secret}  
create_url = "https://api.speechflow.io/asr/file/v1/create?lang=ru"  
query_url = "https://api.speechflow.io/asr/file/v1/query?taskId="  
files = {"file": open(file_name, "rb")}

Далее отправляем POST-запрос на адрес create_url и записываем ответ в переменную response.
Если статус-код ответа равен статус-коду 200, проваливаемся внутрь условия. Иначе завершаем выполнение обработки.

Внутри условия в переменную create_result получаем JSON ответа.
Затем к переменной query_url добавляем идентификатор задачи из create_result и query-параметр resultType.
Всего есть четыре значения resultType:

1. Тип результата по умолчанию: формат JSON для предложений и слов с указанием начального и конечного времени.
2. Формат JSON для сгенерированных субтитров с указанием начального и конечного времени.
3. Формат SRT для сгенерированных субтитров с указанием начального и конечного времени.
4. Чистый текстовый формат для результатов транскрипции без указания начального и конечного времени.

Для ответа в чате лучше всего подходит 4-й вариант.

Блок кода:

response = requests.post(create_url, headers=headers, files=files)  
if response.status_code  200:  
    create_result = response.json()  
    query_url += create_result["taskId"] + "&resultType=4"

Запускаем бесконечный цикл.
Внутри цикла отправляем GET-запрос на адрес query_url и сохраняем ответ в переменную response.
Если статус-код ответа равен 200, проваливаемся внутрь условия. Иначе прерываем цикл и завершаем выполнение обработки.

В переменную query_result получаем JSON ответа.

Далее у нас два условия:

Если в query_result по ключу code лежит значение 11000, значит, транскрипция завершена.
Проваливаемся внутрь условия.

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

В переменную result получаем значение ключа result из query_result. Для удобства можно дополнительно применить метод .replace(), чтобы заменить два переноса строки, например, на пробел.
Асинхронно отправляем ответ на голосовое сообщение.
Удаляем файл с сервера.
Прерываем цикл.

Иначе-если (elif) в query_result по ключу code лежит значение 11001, значит транскрипция ещё не завершена и стоит попробовать чуточку позже.
Выставляем сон на три секунды и переходим к следующей итерации цикла.

Если оба условия не были удовлетворены, то мы прерываем цикл.

Полный код:

import os
import requests  
import time

from botlogic.settings import bot, Secrets


elif message.voice:  
    file_id = message.voice.file_id  
    file = await bot.get_file(file_id)  
    file_path = file.file_path  
    file_name = f"/code/pressanybutton_bot/files/audio{file_id}.mp3"  
    await bot.download_file(file_path, file_name)  

    headers = {"keyId": Secrets.audio_key_id, "keySecret": Secrets.audio_key_secret}  
    create_url = "https://api.speechflow.io/asr/file/v1/create?lang=ru"  
    query_url = "https://api.speechflow.io/asr/file/v1/query?taskId="  
    files = {"file": open(file_name, "rb")}  

    response = requests.post(create_url, headers=headers, files=files)  
    if response.status_code  200:  
        create_result = response.json()  
        query_url += create_result["taskId"] + "&resultType=4"  
        while True:  
            response = requests.get(query_url, headers=headers)  

            if response.status_code  200:  
                query_result = response.json()  
                if query_result["code"]  11000:  
                    if query_result["result"]:  
                        result = query_result["result"].replace("\n\n", " ")  
                        await message.reply(f"<pre><code>{result}</code></pre>")  
                        os.remove(file_name)  
                    break  
                elif query_result["code"]  11001:  
                    time.sleep(3)  
                    continue  
                else:  
                    break  
            else:  
                break

Заключение.

Качество распознавания не всегда хорошее. Тот же WhisperAI на модели small выдавал куда лучший результат, но он требователен к железу. Интересно, как быстро кончится лимит в 5 часов.