Как все начиналось

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