[Из песочницы] Обработка голосовых запросов в Telegram с помощью Yandex SpeechKit Cloud
Как все начиналось
Этим летом я участвовал в разработке бота Datatron, предоставляющего доступ с открытыми финансовыми данными РФ. В какой-то момент я захотел, чтобы бот мог обрабатывать голосовые запросы, и для реализации этой задачи решил использовать наработками Яндекса.
После долгих поисков хоть какой-то полезной информации на эту тему, я наконец-то встретил человека, который написал voiceru_bot и помог мне разобраться с этой темой (в источниках приведена ссылка на его репозиторий). Теперь я хочу поделиться этими знаниями с вами.
От слов в практике
Ниже будет по фрагментам приведен код полностью готовый к применению, который практически можно просто скопировать и вставить в ваш проект.
Шаг 1. С чего начать?
Заведите аккаунт на Яндексе (если у вас его нет). Затем прочтите условия использования SpeechKit Cloud API. Если вкратце, то для некоммерческих проектов при количестве запросов не более 1000 в сутки использование бесплатное. После зайдите в Кабинет разработчика и закажите ключ на требуемый сервис. Обычно его активируют в течение 3 рабочих дней (хотя один из моих ключей активировали через неделю). И наконец изучите документацию.
Шаг 2: Сохранение отправленной голосовой записи
Перед тем, как отправить запрос к API, нужно получить само голосовое сообщение. В коде ниже в несколько строчек получаем объект, в котором хранятся все данные о голосовом сообщении.
import requests
@bot.message_handler(content_types=['voice'])
def voice_processing(message):
file_info = bot.get_file(message.voice.file_id)
file = requests.get('https://api.telegram.org/file/bot{0}/{1}'.format(TELEGRAM_API_TOKEN, file_info.file_path)) Сохранив в переменную file объект, нас в первую очередь интересует поле content, в котором хранится байтовая запись отправленного голосового сообщения. Она нам и нужна для дальнейшей работы.
Шаг 3. Перекодирование
Голосовое сообщение в Telegram сохраняется в формате OGG с аудиокодеком Opus. SpeechKit умеет обрабатывать аудиоданные в формате OGG с аудиокодеком Speex. Таким образом, необходимо конвертировать файл, лучше всего в PCM 16000 Гц 16 бит, так как по документации этот формат обеспечивает наилучшее качество распознавания. Для этого отлично подойдет FFmpeg. Скачайте его и сохраните в директорию проекта, оставив только папку bin и ее содержимое. Ниже реализована функция, которая с помощью FFmpeg перекодирует поток байтов в нужный формат.
import subprocess
import tempfile
import os
def convert_to_pcm16b16000r(in_filename=None, in_bytes=None):
with tempfile.TemporaryFile() as temp_out_file:
temp_in_file = None
if in_bytes:
temp_in_file = tempfile.NamedTemporaryFile(delete=False)
temp_in_file.write(in_bytes)
in_filename = temp_in_file.name
temp_in_file.close()
if not in_filename:
raise Exception('Neither input file name nor input bytes is specified.')
# Запрос в командную строку для обращения к FFmpeg
command = [
r'Project\ffmpeg\bin\ffmpeg.exe', # путь до ffmpeg.exe
'-i', in_filename,
'-f', 's16le',
'-acodec', 'pcm_s16le',
'-ar', '16000',
'-'
]
proc = subprocess.Popen(command, stdout=temp_out_file, stderr=subprocess.DEVNULL)
proc.wait()
if temp_in_file:
os.remove(in_filename)
temp_out_file.seek(0)
return temp_out_file.read()Шаг 4. Передача аудиозаписи по частям
SpeechKit Cloud API принимает на вход файл размером до 1 Мб, при этом его размер нужно указывать отдельно (в Content-Length). Но лучше реализовать передачу файла по частям (размером не больше 1 Мб с использованием заголовка Transfer-Encoding: chunked). Так безопаснее, и распознавание текста будет происходить быстрее.
def read_chunks(chunk_size, bytes):
while True:
chunk = bytes[:chunk_size]
bytes = bytes[chunk_size:]
yield chunk
if not bytes:
breakШаг 5. Отправка запроса к Yandex API и парсинг ответа
Наконец, последний шаг — написать одну единственную функцию, которая будет служить «API» к этому модулю. То есть, сначала в ней будет происходить вызов методов, ответственных за конвертирование и считывание байтов по блокам, а затем идти запрос к SpeechKit Cloud и чтение ответа. По умолчанию, для запросов топик задан notes, а язык — русский.
import xml.etree.ElementTree as XmlElementTree
import httplib2
import uuid
from config import YANDEX_API_KEY
YANDEX_ASR_HOST = 'asr.yandex.net'
YANDEX_ASR_PATH = '/asr_xml'
CHUNK_SIZE = 1024 ** 2
def speech_to_text(filename=None, bytes=None, request_id=uuid.uuid4().hex, topic='notes', lang='ru-RU',
key=YANDEX_API_KEY):
# Если передан файл
if filename:
with open(filename, 'br') as file:
bytes = file.read()
if not bytes:
raise Exception('Neither file name nor bytes provided.')
# Конвертирование в нужный формат
bytes = convert_to_pcm16b16000r(in_bytes=bytes)
# Формирование тела запроса к Yandex API
url = YANDEX_ASR_PATH + '?uuid=%s&key=%s&topic=%s&lang=%s' % (
request_id,
key,
topic,
lang
)
# Считывание блока байтов
chunks = read_chunks(CHUNK_SIZE, bytes)
# Установление соединения и формирование запроса
connection = httplib2.HTTPConnectionWithTimeout(YANDEX_ASR_HOST)
connection.connect()
connection.putrequest('POST', url)
connection.putheader('Transfer-Encoding', 'chunked')
connection.putheader('Content-Type', 'audio/x-pcm;bit=16;rate=16000')
connection.endheaders()
# Отправка байтов блоками
for chunk in chunks:
connection.send(('%s\r\n' % hex(len(chunk))[2:]).encode())
connection.send(chunk)
connection.send('\r\n'.encode())
connection.send('0\r\n\r\n'.encode())
response = connection.getresponse()
# Обработка ответа сервера
if response.code == 200:
response_text = response.read()
xml = XmlElementTree.fromstring(response_text)
if int(xml.attrib['success']) == 1:
max_confidence = - float("inf")
text = ''
for child in xml:
if float(child.attrib['confidence']) > max_confidence:
text = child.text
max_confidence = float(child.attrib['confidence'])
if max_confidence != - float("inf"):
return text
else:
# Создавать собственные исключения для обработки бизнес-логики - правило хорошего тона
raise SpeechException('No text found.\n\nResponse:\n%s' % (response_text))
else:
raise SpeechException('No text found.\n\nResponse:\n%s' % (response_text))
else:
raise SpeechException('Unknown error.\nCode: %s\n\n%s' % (response.code, response.read()))
# Создание своего исключения
сlass SpeechException(Exception):
passШаг 6. Использование написанного модуля
Теперь дополним главный метод, из которого будем вызывать функцию speech_to_text. В ней нужно только дописать обработку того случая, когда пользователь отправляет голосовое сообщение, в котором нет звуков или распознаваемого текста. Не забудьте сделать импорт функции speech_to_text и класса SpeechException, если необходимо.
@bot.message_handler(content_types=['voice'])
def voice_processing(message):
file_info = bot.get_file(message.voice.file_id)
file = requests.get(
'https://api.telegram.org/file/bot{0}/{1}'.format(API_TOKEN, file_info.file_path))
try:
# обращение к нашему новому модулю
text = speech_to_text(bytes=file.content)
except SpeechException:
# Обработка случая, когда распознавание не удалось
else:
# Бизнес-логикаВот и все. Теперь вы можете легко реализовывать обработку голоса в ваших проектах. Причем не только в Telegram, но и на других платформах, взяв за основу эту статью!
Источники:
» @voiceru_bot: https://github.com/just806me/voiceru_bot
» Для работы с API Telegram на Python использовалась библиотека telebot
