Как все начиналось
Этим летом я участвовал в разработке бота 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
