Обробка голосових запитів в 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='uk-ua',
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
Джерело: Хабрахабр

0 коментарів

Тільки зареєстровані та авторизовані користувачі можуть залишати коментарі.